LangChain 代理 (Agents) 完全指南

深入讲解 LangChain 代理系统，包括 ReAct 框架、AgentExecutor、工具系统、流式执行和迁移到 LangGraph

代理 (Agents)

---

用途

使用 LLM 作为决策引擎，动态决定使用哪些工具来完成复杂任务。代理是 LangChain 的高级功能，能够进行多步推理和行动。

位置

libs/langchain/langchain/agents/

知识图谱上下文

核心类/模块

代理类型

• Agent: 代理基类
• ZeroShotAgent: 零样本代理（使用 ReAct 框架）
• ReActAgent: 推理-行动代理
• ChatAgent: 聊天代理（使用对话历史）
• OpenAIFunctionsAgent: OpenAI 函数调用代理
• OpenAIToolsAgent: OpenAI 工具调用代理
• StructuredChatAgent: 结构化聊天代理
• SelfAskWithSearchAgent: 自问自答代理

执行器

• AgentExecutor: 代理执行引擎
• AgentExecutorIterator: 流式执行迭代器

工具相关

• Tool: 工具基类
• StructuredTool: 带结构化输入的工具
• ToolInvocation: 工具调用记录
• AgentAction: 代理动作
• AgentFinish: 代理完成信号

依赖关系

内部依赖

• langchain_core.language_models: 语言模型接口
• langchain_core.tools: 工具抽象
• langchain_core.prompts: 提示模板
• langchain_core.callbacks: 回调系统
• langchain.memory: 记忆管理

外部依赖

• asyncio: 异步执行
• pydantic: 数据验证
• typing: 类型提示

代理执行循环

ReAct 框架

AgentExecutor 工作流程

# 伪代码展示执行逻辑
while not agent_done:
    # 1. 思考：让 LLM 决定下一步
    agent_output = agent.plan(
        inputs=inputs,
        intermediate_steps=steps
    )

    # 2. 检查是否完成
    if isinstance(agent_output, AgentFinish):
        return agent_output.return_values

    # 3. 执行工具
    tool_output = tool_executor.execute(
        tool=agent_output.tool,
        tool_input=agent_output.tool_input
    )

    # 4. 记录步骤
    steps.append((agent_output, tool_output))

代理类型详解

ZeroShotAgent

使用 ReAct（推理-行动）框架的零样本代理：

from langchain.agents import AgentExecutor, create_react_agent
from langchain.tools import Tool
from langchain_openai import OpenAI

# 定义工具
tools = [
    Tool(
        name="Search",
        func=search_func,
        description="Useful for searching current information"
    )
]

# 创建提示模板
from langchain import hub
prompt = hub.pull("hwchase17/react")

# 创建代理
agent = create_react_agent(
    llm=OpenAI(temperature=0),
    tools=tools,
    prompt=prompt
)

# 创建执行器
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True
)

ChatAgent

专为对话设计的代理：

OpenAIFunctionsAgent

使用 OpenAI 函数调用 API：

from langchain.agents import create_openai_functions_agent
from langchain_openai import ChatOpenAI

# 使用支持函数调用的模型
llm = ChatOpenAI(model="gpt-4", temperature=0)

agent = create_openai_functions_agent(
    llm=llm,
    tools=tools,
    prompt=prompt
)

工具系统

创建自定义工具

from langchain.tools import tool

@tool
def search_database(query: str) -> str:
    """Search the database for information."""
    # 实现搜索逻辑
    return f"Results for: {query}"

# 或使用 StructuredTool
from langchain.tools import StructuredTool
from pydantic import BaseModel, Field

class SearchInput(BaseModel):
    query: str = Field(description="Search query")

search_tool = StructuredTool.from_function(
    func=search_database,
    name="Search",
    description="Search database",
    args_schema=SearchInput
)

工具集合

from langchain.agents import load_tools

# 加载预定义工具集
tools = load_tools(
    ["serpapi", "llm-math", "python-repl"],
    llm=llm
)

常用内置工具

工具名	用途
serpapi	Google 搜索
wolfram-alpha	数学计算
python-repl	Python 代码执行
requests	HTTP 请求
shell	Shell 命令执行

代理配置

最大迭代次数

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_iterations=5,
    early_stopping_method="generate"  # 或 "force"
)

错误处理

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    handle_parsing_errors=True,
    max_execution_time=60  # 最大执行时间（秒）
)

内存集成

from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory,
    verbose=True
)

代理提示工程

ReAct 提示模板

from langchain.agents import AgentExecutor, create_react_agent
from langchain import hub

# 使用 Hub 上的提示模板
prompt = hub.pull("hwchase17/react")

prompt.template = """
Answer the following questions as best you can. You have access to the following tools:

{tools}

Use the following format:

Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question

Begin!

Question: {input}
Thought:{agent_scratchpad}
"""

自定义提示

from langchain_core.prompts import PromptTemplate

template = """
You are a helpful assistant. You have access to the following tools:
{tools}

To use a tool, use the following format:
Tool: <tool name>
Input: <tool input>

Question: {input}
{agent_scratchpad}
"""

prompt = PromptTemplate(
    template=template,
    input_variables=["input", "agent_scratchpad", "tools"]
)

流式执行

流式输出最终答案

async for chunk in agent_executor.astream(
    {"input": "What is the weather in Tokyo?"}
):
    print(chunk)

流式中间步骤

async for event in agent_executor.astream_events(
    {"input": "What is the weather in Tokyo?"},
    version="v1"
):
    kind = event["event"]
    if kind == "on_llm_stream":
        content = event["data"]["chunk"].content
        if content:
            print(content, end="", flush=True)

代理评估

使用 LangSmith 评估

from langsmith import Client

client = Client()

# 创建评估数据集
dataset_name = "Agent Evaluation Dataset"
dataset = client.create_dataset(dataset_name)

# 运行评估
from langchain.evaluation import LangSmithChainEvaluator

evaluator = LangSmithChainEvaluator()
eval_results = evaluator.evaluate(
    agent_executor,
    dataset_name,
    inputs_key="input",
    output_key="output"
)

自定义评估指标

from langchain.evaluation import EvaluatorType
from langchain.smith import RunEvalConfig

eval_config = RunEvalConfig(
    evaluators=[
        # 是否使用了正确工具
        "trajectory",  # 轨迹评估
        # 最终答案质量
        "qa",  # 问答评估
        # 自定义评估器
        RunEvalConfig.LabeledCriteria("helpfulness"),
    ]
)

迁移到 LangGraph

LangGraph 提供了更强大和可控的代理框架：

LangGraph 代理示例

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode

# 定义状态
class AgentState(TypedDict):
    messages: List[BaseMessage]

# 创建图
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("agent", call_model)
workflow.add_node("tools", ToolNode(tools))

# 添加边
workflow.set_entry_point("agent")
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {"continue": "tools", "end": END}
)
workflow.add_edge("tools", "agent")

最佳实践

1. 明确工具描述

# 好的描述
@tool
def search_api(query: str) -> str:
    """Search the API for current information about movies."""
    pass

# 不好的描述
@tool
def search_api(query: str) -> str:
    """Search."""
    pass

2. 使用适当的代理类型

• 简单任务：ZeroShotAgent
• 对话任务：ChatAgent
• 函数调用：OpenAIFunctionsAgent
• 复杂推理：ReActAgent

3. 限制工具数量

# 不要一次给代理太多工具
# 推荐：3-10 个相关工具
# 太多工具会降低决策质量

4. 添加错误处理

from langchain_core.tools import StructuredTool
from langchain_core.exceptions import ToolException

@tool
def my_tool(x: str) -> str:
    try:
        return process(x)
    except Exception as e:
        raise ToolException(f"Error: {str(e)}")

5. 使用回调监控

from langchain_core.callbacks import BaseCallbackHandler

class AgentCallbackHandler(BaseCallbackHandler):
    def on_agent_action(self, action, **kwargs):
        print(f"Agent action: {action.tool}")

    def on_tool_end(self, output, **kwargs):
        print(f"Tool output: {output}")

result = agent_executor.invoke(
    {"input": "Hello"},
    config={"callbacks": [AgentCallbackHandler()]}
)

---