原文参考: LangChain 官方文档 - Agents 模块
适用版本: LangChain 1.0+

核心摘要:LangChain Agents 将大语言模型(LLM)与外部工具(Tools)相结合,创造出能够推理任务、决策使用何种工具并迭代解决问题的系统。create_agent 提供了基于 LangGraph 的生产就绪型智能体实现。智能体遵循 ReAct(推理+行动)模式,在模型推理与工具调用之间循环,直到满足停止条件(输出最终结果或达到迭代限制)。

1. 智能体架构概览

智能体本质上是一个基于 图(Graph) 的运行时结构,包含节点(步骤)和边(连接)。其核心流程如下:

  • 输入 (Input):用户发起的查询。
  • 模型节点 (Model):智能体的“大脑”,负责推理。
  • 工具节点 (Tools):智能体的“手脚”,负责执行具体动作。
  • 输出 (Output):最终的答复。

Action

Observation

Finish

输入

模型

工具

输出


2. 核心组件详解

一个智能体由 模型 (Model)工具 (Tools)提示词 (Prompt)状态 (State) 四大核心部分组成。

2.1 模型 (Model):智能体的“大脑”

模型负责推理和决策。LangChain 支持静态配置和动态选择两种模式。

静态模型

最常见的配置方式,直接指定模型标识符或实例。

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI

# 方式1:使用模型标识符字符串 (支持自动推断)
agent = create_agent("openai:gpt-4o", tools=tools)

# 方式2:使用模型实例 (更精细的控制)
model = ChatOpenAI(
    model="gpt-4o", 
    temperature=0.1, 
    max_tokens=1000,
    timeout=30
)
agent = create_agent(model, tools=tools)
动态模型 (Middleware)

基于运行时的状态(如对话轮次)动态选择模型,实现成本优化或性能分级。

from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse

basic_model = ChatOpenAI(model="gpt-4o-mini")
advanced_model = ChatOpenAI(model="gpt-4o")

@wrap_model_call
def dynamic_model_selection(request: ModelRequest, handler) -> ModelResponse:
    """根据对话复杂度选择模型。"""
    message_count = len(request.state["messages"])
    if message_count > 10:
        model = advanced_model # 长对话使用高级模型
    else:
        model = basic_model
    return handler(request.override(model=model))

agent = create_agent(
    model=basic_model,
    tools=tools,
    middleware=[dynamic_model_selection]
)

2.2 工具 (Tools):智能体的“手脚”

工具赋予智能体执行动作的能力。LangChain 支持多工具序列调用、并行调用及动态工具选择。

静态工具

创建时定义,运行期间不变。

from langchain.tools import tool

@tool
def search(query: str) -> str:
    """搜索信息"""
    return f"搜索结果: {query}"

agent = create_agent(model, tools=[search])
动态工具选择

根据状态、权限或上下文动态修改可用工具集。

场景:基于用户角色的工具过滤

from dataclasses import dataclass
from langchain.agents.middleware import wrap_model_call

@dataclass
class Context:
    user_role: str

@wrap_model_call
def context_based_tools(request: ModelRequest, handler) -> ModelResponse:
    """根据运行时上下文权限过滤工具。"""
    user_role = request.runtime.context.user_role
    
    if user_role == "admin":
        pass # 管理员拥有所有工具
    elif user_role == "editor":
        # 编辑者不能删除数据
        tools = [t for t in request.tools if t.name != "delete_data"]
        request = request.override(tools=tools)
    else:
        # 查看者仅拥有读取工具
        tools = [t for t in request.tools if t.name.startswith("read_")]
        request = request.override(tools=tools)
    return handler(request)

agent = create_agent(
    model="gpt-4o", 
    tools=[read_data, write_data, delete_data], 
    middleware=[context_based_tools],
    context_schema=Context 
)
运行时工具注册 (Runtime Tool Registration)

当工具在运行时才被发现(如从 MCP 服务器加载)时,需要同时注册工具并处理执行。

from langchain.tools import tool
from langchain.agents.middleware import AgentMiddleware, ToolCallRequest

# 定义一个动态工具
@dynamic_tool = tool("calculate_tip")
def calculate_tip(bill_amount: float) -> str:
    return f"Tip: ${bill_amount * 0.2}"

class DynamicToolMiddleware(AgentMiddleware):
    def wrap_model_call(self, request: ModelRequest, handler):
        # 向请求中添加动态工具
        updated = request.override(tools=[*request.tools, calculate_tip])
        return handler(updated)
        
    def wrap_tool_call(self, request: ToolCallRequest, handler):
        # 处理动态工具的执行
        if request.tool_call["name"] == "calculate_tip":
            return handler(request.override(tool=calculate_tip))
        return handler(request)

agent = create_agent(
    model="gpt-4o",
    tools=[static_tool], 
    middleware=[DynamicToolMiddleware()]
)

2.3 ReAct 循环:推理与行动

智能体遵循 ReAct 模式,交替进行推理与行动。

示例流程:

  1. Prompt: “找出目前最流行的无线耳机并验证库存。”
  2. Reasoning: “流行度随时间变化,我需要使用搜索工具。”
  3. Acting: 调用 search_products("wireless headphones")
  4. Observation: “找到了 WH-1000XM5 等产品。”
  5. Reasoning: “我需要确认排名第一的产品是否有货。”
  6. Acting: 调用 check_inventory("WH-1000XM5")
  7. Reasoning: “我有了最流行型号及其库存状态,可以回答用户了。”
  8. Acting: 输出最终答案。

2.4 系统提示词 (System Prompt)

你可以通过 system_prompt 参数塑造智能体的行为方式。

  • 静态提示词:直接传入字符串或 SystemMessage

    agent = create_agent(
        model, 
        tools, 
        system_prompt="你是一个乐于助人的助手。请简洁准确地回答。"
    )
    
  • 动态提示词:基于运行时上下文(如用户角色)动态生成。

    @dynamic_prompt
    def user_role_prompt(request: ModelRequest) -> str:
        user_role = request.runtime.context.get("user_role", "user")
        if user_role == "expert":
            return "你是专家助手。请提供详细的技术回复。"
        elif user_role == "beginner":
            return "你是初级助手。请简单解释概念,避免行话。"
        return "你是乐于助人的助手。"
    

3. 高级特性

3.1 结构化输出 (Structured Output)

强制智能体返回特定格式的数据(如 Pydantic 模型)。

策略类型 说明 适用场景
ToolStrategy 利用人工工具调用生成结构化输出 任何支持工具调用的模型
ProviderStrategy 利用模型提供商原生的结构化输出功能 支持原生结构化输出的提供商
from pydantic import BaseModel
from langchain.agents.structured_output import ToolStrategy

class ContactInfo(BaseModel):
    name: str
    email: str

# 使用 ToolStrategy
agent = create_agent(
    model="gpt-4o-mini",
    tools=[search_tool],
    response_format=ToolStrategy(ContactInfo)
)

3.2 记忆 (Memory) 与状态管理

智能体通过 State 自动维护对话历史。你可以扩展 AgentState 来存储自定义数据(短时记忆)。

from langchain.agents import AgentState
from typing import TypedDict

class CustomState(AgentState):
    user_preferences: dict

# 通过 Middleware 定义(推荐)
class CustomMiddleware(AgentMiddleware):
    state_schema = CustomState
    
    def before_model(self, state: CustomState, runtime):
        # 在这里可以访问自定义状态
        pass

agent = create_agent(
    model, 
    tools=tools, 
    middleware=[CustomMiddleware()]
)

3.3 流式传输 (Streaming)

为了在智能体执行多步操作时展示中间进度,可以使用流式传输。

for chunk in agent.stream({
    "messages": [{"role": "user", "content": "搜索 AI 新闻并总结"}]
}, stream_mode="values"):
    latest_message = chunk["messages"][-1]
    if latest_message.tool_calls:
        print(f"正在调用工具: {[tc['name'] for tc in latest_message.tool_calls]}")

3.4 中间件 (Middleware)

中间件提供了强大的扩展能力,允许你在执行的关键节点拦截和修改数据流。

  • @wrap_model_call:在模型调用前后修改请求/响应。
  • @wrap_tool_call:自定义工具执行逻辑(如错误处理)。
  • @before_model / @after_model:在模型调用前后执行特定逻辑。

4. 调用与运行

4.1 调用方式

你可以通过传递 State 的更新来调用智能体。

result = agent.invoke({
    "messages": [{"role": "user", "content": "旧金山天气如何?"}]
})

4.2 调试与追踪

建议使用 LangSmith 来追踪、调试和评估你的智能体,这对于处理复杂的 ReAct 循环非常有帮助。


本文档基于 LangChain 官方文档整理,涵盖了从基础到高级的 Agents 开发模式。

Logo

汇聚全球AI编程工具,助力开发者即刻编程。

更多推荐