02.LangChain Agents 完整指南:从核心组件到高级模式
原文参考: LangChain 官方文档 - Agents 模块
适用版本: LangChain 1.0+
核心摘要:LangChain Agents 将大语言模型(LLM)与外部工具(Tools)相结合,创造出能够推理任务、决策使用何种工具并迭代解决问题的系统。
create_agent提供了基于 LangGraph 的生产就绪型智能体实现。智能体遵循 ReAct(推理+行动)模式,在模型推理与工具调用之间循环,直到满足停止条件(输出最终结果或达到迭代限制)。
1. 智能体架构概览
智能体本质上是一个基于 图(Graph) 的运行时结构,包含节点(步骤)和边(连接)。其核心流程如下:
- 输入 (Input):用户发起的查询。
- 模型节点 (Model):智能体的“大脑”,负责推理。
- 工具节点 (Tools):智能体的“手脚”,负责执行具体动作。
- 输出 (Output):最终的答复。
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 模式,交替进行推理与行动。
示例流程:
- Prompt: “找出目前最流行的无线耳机并验证库存。”
- Reasoning: “流行度随时间变化,我需要使用搜索工具。”
- Acting: 调用
search_products("wireless headphones") - Observation: “找到了 WH-1000XM5 等产品。”
- Reasoning: “我需要确认排名第一的产品是否有货。”
- Acting: 调用
check_inventory("WH-1000XM5") - Reasoning: “我有了最流行型号及其库存状态,可以回答用户了。”
- 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 开发模式。
更多推荐




所有评论(0)