使用LangChain构建AI智能体：工具集成与多轮对话

用户输入 → 智能体（推理） → 工具选择 → 工具执行 → 响应

from langchain.agents import create_openai_tools_agent
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# 初始化LLM
llm = ChatOpenAI(model="gpt-4", temperature=0)

# 定义提示词
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有工具访问权限的有用助手。"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

# 创建智能体
agent = create_openai_tools_agent(llm, tools, prompt)

from langchain_core.tools import tool
from typing import Optional

@tool
def search_knowledge_base(query: str) -> str:
    """搜索私有知识库获取信息。"""
    # 实现代码
    results = vector_store.similarity_search(query, k=3)
    return "\n".join([doc.page_content for doc in results])

@tool
def get_product_info(product_id: str) -> str:
    """通过ID获取详细产品信息。"""
    # 数据库查询
    product = db.get_product(product_id)
    return f"产品: {product.name}, 价格: {product.price}"

# 注册工具
tools = [search_knowledge_base, get_product_info]

from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain

# 初始化记忆
memory = ConversationBufferMemory(return_messages=True)

# 创建对话链
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)

# 在智能体中使用
agent = create_openai_tools_agent(
    llm, tools, prompt,
    memory=memory
)

from langchain.agents import AgentExecutor

# 创建智能体执行器
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=10,  # 防止无限循环
    handle_parsing_errors=True
)

# 带上下文的对话
def chat(user_input: str, conversation_id: str):
    # 加载对话历史
    history = load_conversation_history(conversation_id)
    
    # 运行智能体
    result = agent_executor.invoke({
        "input": user_input,
        "chat_history": history
    })
    
    # 保存到历史
    save_conversation_history(conversation_id, user_input, result["output"])
    
    return result["output"]

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class ChatRequest(BaseModel):
    message: str
    conversation_id: str

class ChatResponse(BaseModel):
    response: str
    conversation_id: str

@app.post("/chat", response_model=ChatResponse)
async def chat_endpoint(request: ChatRequest):
    try:
        response = chat(request.message, request.conversation_id)
        return ChatResponse(
            response=response,
            conversation_id=request.conversation_id
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

from fastapi.responses import StreamingResponse

@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    async def generate():
        async for event in agent_executor.astream_events(
            {"input": request.message},
            version="v1"
        ):
            if event["event"] == "on_chat_model_stream":
                content = event["data"]["chunk"].content
                if content:
                    yield f"data: {content}\n\n"
        yield "data: [DONE]\n\n"
    
    return StreamingResponse(generate(), media_type="text/event-stream")

@app.middleware("http")
async def error_handler(request, call_next):
    try:
        response = await call_next(request)
        return response
    except Exception as e:
        logger.error(f"智能体错误: {e}")
        return JSONResponse(
            status_code=500,
            content={"error": "内部服务器错误"}
        )

from langchain.agents import create_openai_tools_agent
from langchain.tools import Tool

# 定义农业工具
tools = [
    Tool(
        name="SearchKnowledgeBase",
        func=search_agricultural_knowledge,
        description="搜索私有农业手册获取作物病害、治疗和最佳实践"
    ),
    Tool(
        name="GeneralAgriculturalKnowledge",
        func=get_general_knowledge,
        description="当私有知识库没有答案时获取一般农业信息"
    )
]

# 强调私有数据优先的系统提示词
system_prompt = """你是一位农业专家助手。

重要提示：始终首先搜索私有知识库。仅在私有搜索没有返回相关结果时使用一般知识。

回答时：
1. 尽可能引用源文档
2. 具体说明作物名称、病害症状和治疗方法
3. 如果不确定，请说"建议咨询当地农业专家"
"""