LangGraph Agent 详细解读
网站使用说明
📚 概述
本文档详细解读 LangGraph 中的 Agent（智能体） 架构。这是构建智能 AI 应用的核心模式，通过让 AI 模型自主决策、调用工具、并基于结果继续推理，实现复杂任务的自动化处理。
关于 Agent，我们在后面还会介绍：
- Module 12，Multi-agent Graph
- Module 13，Agentic RAG
- Module 14，Deep Agent
📚 术语表
| 术语名称 | LangGraph 定义和解读 | Python 定义和说明 | 重要程度 |
|---|---|---|---|
| Agent | 能自主决策和行动的 AI 系统，支持多步推理和工具调用 | 通过循环边实现工具结果反馈到 LLM 的图结构 | ⭐⭐⭐⭐⭐ |
| ReAct | Reasoning + Acting 架构，推理-行动-观察循环模式 | Agent 的经典实现模式，LLM 基于观察结果持续决策 | ⭐⭐⭐⭐⭐ |
| create_agent() | LangChain 提供的生产级 Agent 工厂函数 | 基于 LangGraph 构建图运行时，快速创建标准 Agent | ⭐⭐⭐⭐⭐ |
| @tool | LangChain 的工具装饰器 | 将普通函数转换为 LangChain Tool 对象，自动解析类型和文档 | ⭐⭐⭐⭐⭐ |
| 循环边（Loop Edge） | 从工具节点返回到 LLM 节点的边，形成推理循环 | add_edge("tools", "assistant") 创建循环结构 | ⭐⭐⭐⭐⭐ |
| ToolNode | 自动执行工具调用并返回结果的预构建节点 | 解析 tool_calls，调用函数，包装为 ToolMessage | ⭐⭐⭐⭐⭐ |
| tools_condition | 判断是否继续调用工具的条件函数 | 检查 AIMessage.tool_calls，决定路由到 tools 或 END | ⭐⭐⭐⭐⭐ |
| parallel_tool_calls | 控制工具并行/顺序执行的参数 | bind_tools 参数，False 强制顺序执行（如数学计算） | ⭐⭐⭐⭐ |
| SystemMessage | 定义 AI 角色和行为的系统提示消息 | LangChain 消息类，指导 LLM 如何使用工具和推理 | ⭐⭐⭐⭐⭐ |
| recursion_limit | 限制 Agent 循环次数的参数，防止无限循环 | compile() 参数，设置最大推理步数 | ⭐⭐⭐⭐ |
| tool_call_id | 工具调用的唯一标识符，关联请求和结果 | ToolMessage 的必需属性，匹配 AIMessage.tool_calls 中的 id | ⭐⭐⭐⭐⭐ |
| stream() | 流式执行图并逐步返回结果的方法 | 异步生成器，yield 每个节点的输出，实时观察执行过程 | ⭐⭐⭐⭐ |
| Middleware | Agent 中间件，用于扩展和定制 Agent 行为 | @wrap_model_call 和 @wrap_tool_call 装饰器 | ⭐⭐⭐⭐ |
| ToolStrategy | 结构化输出策略，通过工具调用返回结构化数据 | 兼容所有支持工具调用的模型 | ⭐⭐⭐ |
🎯 核心概念
什么是 Agent？
Agent（智能体）是一种能够自主决策和行动的 AI 系统。在 LangGraph 中，Agent 具有以下特点：
自主决策
- AI 模型自己决定是否需要调用工具
- 根据工具返回的结果决定下一步行动
- 可以连续调用多个工具完成复杂任务
工具调用
- 调用外部函数获取信息或执行操作
- 将工具结果传回模型继续处理
- 支持多轮工具调用
循环推理
- 模型可以基于工具结果继续思考
- 决定是继续调用工具还是直接回答
- 形成 "思考-行动-观察" 的循环
ReAct 架构
ReAct（Reasoning + Acting）是一种经典的 Agent 架构模式：
- Reason（推理）：模型分析问题，决定需要调用什么工具
- Act（行动）：调用选定的工具
- Observe（观察）：获取工具返回结果
- 循环：基于观察结果，决定继续行动还是给出最终答案
与 Router 的区别
在之前的 Router 模式中：
- 模型调用工具后，直接返回
ToolMessage给用户 - 这是一次性的工具调用，没有循环
在 Agent 模式中：
- 模型调用工具后，
ToolMessage被传回模型 - 模型可以基于结果决定：
- 继续调用其他工具
- 或者直接回答用户
关键差异：工具结果的流向
Router: 用户 → 模型 → 工具 → 用户
Agent: 用户 → 模型 → 工具 → 模型 → ... → 用户
↑____________↓
循环反馈🎭 实战案例：数学计算 Agent
我们将构建一个数学计算助手，演示 Agent 的完整工作流程。
需求： 给定一个多步骤数学问题，Agent 需要：
- 理解问题需要哪些计算步骤
- 依次调用相应的工具（加法、乘法、除法）
- 基于每一步结果，决定下一步操作
- 最终给出答案
系统架构图
用户输入: "Add 3 and 4. Multiply the output by 2. Divide the output by 5"
↓
[assistant] 分析：需要先做加法
↓
(tools_condition 判断：需要调用工具)
↓
[tools] 调用 add(3, 4) → 返回 7
↓
[assistant] 收到结果 7，分析：需要乘以 2
↓
(tools_condition 判断：需要调用工具)
↓
[tools] 调用 multiply(7, 2) → 返回 14
↓
[assistant] 收到结果 14，分析：需要除以 5
↓
(tools_condition 判断：需要调用工具)
↓
[tools] 调用 divide(14, 5) → 返回 2.8
↓
[assistant] 收到结果 2.8，生成最终答案
↓
(tools_condition 判断：不需要工具)
↓
END → 返回结果给用户🔧 代码实现详解
1. 定义工具函数
python
def multiply(a: int, b: int) -> int:
"""Multiply a and b.
Args:
a: first int
b: second int
"""
return a * b
def add(a: int, b: int) -> int:
"""Adds a and b.
Args:
a: first int
b: second int
"""
return a + b
def divide(a: int, b: int) -> float:
"""Divide a and b.
Args:
a: first int
b: second int
"""
return a / b
tools = [add, multiply, divide]Python 知识点：函数文档字符串（Docstring）
函数的文档字符串不仅仅是注释，它是 Python 的一等公民：
python
def add(a: int, b: int) -> int:
"""Adds a and b. # ← 简短描述
Args: # ← 参数说明
a: first int
b: second int
"""
return a + b
# 可以通过 __doc__ 属性访问
print(add.__doc__)LangChain 的魔法： LangChain 会解析这些文档字符串，并将它们作为工具描述传递给 LLM：
python
# LLM 会看到类似这样的工具描述：
# Tool: add
# Description: Adds a and b.
# Parameters:
# - a: first int
# - b: second int这就是为什么 LLM 能够"理解"工具的作用！
2. 绑定工具到模型
python
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-5-nano")
llm_with_tools = llm.bind_tools(tools, parallel_tool_calls=False)关键参数：parallel_tool_calls=False
这个参数控制模型是否可以同时调用多个工具：
python
# parallel_tool_calls=True（默认）
# 模型可能同时调用：add(3, 4) 和 multiply(7, 2)
# 适合：并行、独立的任务
# parallel_tool_calls=False
# 模型必须依次调用：先 add(3, 4)，等结果，再调用 multiply
# 适合：数学计算等需要顺序执行的任务为什么数学计算要设置为 False？
数学计算通常有依赖关系：
第一步：3 + 4 = 7
第二步：7 × 2 = 14 ← 依赖第一步的结果！
第三步：14 ÷ 5 = 2.8 ← 依赖第二步的结果！如果允许并行调用，模型可能在不知道第一步结果的情况下就尝试执行第二步，导致错误。
3. 定义状态
python
from langgraph.graph import MessagesState
from langchain_core.messages import HumanMessage, SystemMessage
# System message
sys_msg = SystemMessage(content="You are a helpful assistant tasked with performing arithmetic on a set of inputs.")LangGraph 知识点：MessagesState
MessagesState 是 LangGraph 内置的状态类，专门用于聊天场景：
python
class MessagesState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]核心特性：
- 自动消息管理：维护对话历史
- 消息追加：使用
add_messagesreducer 自动追加新消息 - 类型安全：支持不同类型的消息（HumanMessage、AIMessage、ToolMessage 等）
消息类型：
python
from langchain_core.messages import (
HumanMessage, # 用户消息
AIMessage, # AI 回复
SystemMessage, # 系统提示
ToolMessage # 工具返回结果
)4. 定义 Assistant 节点
python
def assistant(state: MessagesState):
return {"messages": [llm_with_tools.invoke([sys_msg] + state["messages"])]}代码解析：
python
[sys_msg] + state["messages"]
# 构建完整的消息列表：
# [
# SystemMessage("You are a helpful assistant..."), # 系统提示
# HumanMessage("Add 3 and 4..."), # 用户问题
# AIMessage(tool_calls=[...]), # AI 的工具调用
# ToolMessage(content="7"), # 工具返回结果
# ...
# ]为什么每次都加上 sys_msg？
每次调用 LLM 时，都需要提供完整的上下文：
- 系统提示定义 AI 的角色和行为
- 历史消息提供对话上下文
- 这样 AI 才能理解当前处于对话的哪个阶段
返回格式：
python
return {"messages": [new_message]}
# 返回的消息会被追加到 state["messages"] 中
# 因为 MessagesState 使用了 add_messages reducer5. 构建 Agent 图
python
from langgraph.graph import START, StateGraph
from langgraph.prebuilt import tools_condition
from langgraph.prebuilt import ToolNode
# 创建图
builder = StateGraph(MessagesState)
# 添加节点
builder.add_node("assistant", assistant)
builder.add_node("tools", ToolNode(tools))
# 添加边
builder.add_edge(START, "assistant")
# 条件边：根据 assistant 的输出决定路由
builder.add_conditional_edges(
"assistant",
tools_condition, # 条件函数
)
# 关键：从 tools 回到 assistant 形成循环
builder.add_edge("tools", "assistant")
# 编译
react_graph = builder.compile()
# 🎨 可视化图结构
from IPython.display import Image, display
display(Image(react_graph.get_graph().draw_mermaid_png()))完整案例代码（可直接运行）
以下是完整的可运行代码，包含所有必要的导入和定义：
python
## 完整案例代码
from IPython.display import Image, display
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import START, StateGraph, MessagesState
from langgraph.prebuilt import tools_condition, ToolNode
# 1. 定义工具函数
def multiply(a: int, b: int) -> int:
"""Multiply a and b.
Args:
a: first int
b: second int
"""
return a * b
def add(a: int, b: int) -> int:
"""Adds a and b.
Args:
a: first int
b: second int
"""
return a + b
def divide(a: int, b: int) -> float:
"""Divide a and b.
Args:
a: first int
b: second int
"""
return a / b
tools = [add, multiply, divide]
# 2. 初始化模型并绑定工具（需要设置 OPENAI_API_KEY）
llm = ChatOpenAI(model="gpt-4o-mini")
llm_with_tools = llm.bind_tools(tools, parallel_tool_calls=False)
# 3. 定义系统消息
sys_msg = SystemMessage(content="You are a helpful assistant tasked with performing arithmetic on a set of inputs.")
# 4. 定义 assistant 节点
def assistant(state: MessagesState):
return {"messages": [llm_with_tools.invoke([sys_msg] + state["messages"])]}
# 5. 构建图
builder = StateGraph(MessagesState)
builder.add_node("assistant", assistant)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "assistant")
builder.add_conditional_edges("assistant", tools_condition)
builder.add_edge("tools", "assistant")
react_graph = builder.compile()
# 6. 可视化
display(Image(react_graph.get_graph().draw_mermaid_png()))
# 7. 测试执行
print("=== 测试：多步数学计算 ===")
messages = [HumanMessage(content="Add 3 and 4. Multiply the output by 2. Divide the output by 5")]
result = react_graph.invoke({"messages": messages})
for m in result['messages']:
m.pretty_print()生成的流程图：
LangGraph 知识点：ToolNode
ToolNode 是 LangGraph 提供的预构建节点，用于执行工具调用：
python
ToolNode(tools)
# 自动处理：
# 1. 解析 AIMessage 中的 tool_calls
# 2. 调用对应的工具函数
# 3. 将结果包装成 ToolMessage
# 4. 返回给下一个节点等价的手动实现：
python
def tools_node(state: MessagesState):
# 获取最后一条消息（AI 的工具调用）
last_message = state["messages"][-1]
# 执行所有工具调用
tool_messages = []
for tool_call in last_message.tool_calls:
# 找到对应的工具
tool = next(t for t in tools if t.name == tool_call["name"])
# 调用工具
result = tool.invoke(tool_call["args"])
# 包装成 ToolMessage
tool_messages.append(ToolMessage(
content=str(result),
tool_call_id=tool_call["id"]
))
return {"messages": tool_messages}LangGraph 知识点：tools_condition
tools_condition 是预构建的条件函数，用于判断是否需要调用工具：
python
def tools_condition(state: MessagesState) -> str:
# 获取最后一条消息
last_message = state["messages"][-1]
# 如果是 AIMessage 且包含 tool_calls
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools" # 路由到 tools 节点
else:
return END # 结束图执行条件边的工作流程：
python
builder.add_conditional_edges("assistant", tools_condition)
# 完整流程：
# assistant 节点执行完毕
# ↓
# tools_condition 检查输出
# ↓
# 如果有 tool_calls → 路由到 "tools"
# 如果没有 tool_calls → 路由到 ENDAgent 循环的关键：
python
builder.add_edge("tools", "assistant") # ← 这条边创建了循环！
# 完整循环：
# assistant → (有工具调用) → tools → assistant → ...
# → (无工具调用) → END6. 执行 Agent
python
messages = [HumanMessage(content="Add 3 and 4. Multiply the output by 2. Divide the output by 5")]
messages = react_graph.invoke({"messages": messages})执行过程详解：
步骤 1：assistant 节点
输入：[HumanMessage("Add 3 and 4. Multiply the output by 2. Divide the output by 5")]
输出：AIMessage(tool_calls=[ToolCall(name="add", args={"a": 3, "b": 4})])
tools_condition：检测到 tool_calls → 路由到 "tools"
步骤 2：tools 节点
输入：[HumanMessage(...), AIMessage(tool_calls=[...])]
执行：add(3, 4) = 7
输出：ToolMessage(content="7", tool_call_id="...")
自动路由：→ "assistant"
步骤 3：assistant 节点
输入：[HumanMessage(...), AIMessage(...), ToolMessage("7")]
分析：看到结果是 7，需要乘以 2
输出：AIMessage(tool_calls=[ToolCall(name="multiply", args={"a": 7, "b": 2})])
tools_condition：检测到 tool_calls → 路由到 "tools"
步骤 4：tools 节点
执行：multiply(7, 2) = 14
输出：ToolMessage(content="14")
自动路由：→ "assistant"
步骤 5：assistant 节点
输入：包含 ToolMessage("14") 的完整对话历史
分析：看到结果是 14，需要除以 5
输出：AIMessage(tool_calls=[ToolCall(name="divide", args={"a": 14, "b": 5})])
tools_condition：检测到 tool_calls → 路由到 "tools"
步骤 6：tools 节点
执行：divide(14, 5) = 2.8
输出：ToolMessage(content="2.8")
自动路由：→ "assistant"
步骤 7：assistant 节点
输入：包含所有历史消息和最终结果 2.8
分析：所有计算完成，可以给出最终答案
输出：AIMessage(content="The final result is 2.8") # 注意：没有 tool_calls
tools_condition：没有 tool_calls → 路由到 END
图执行结束7. 查看输出
python
for m in messages['messages']:
m.pretty_print()输出解析：
================================ Human Message =================================
Add 3 and 4. Multiply the output by 2. Divide the output by 5
================================== Ai Message ==================================
Tool Calls:
add (call_i8zDfMTdvmIG34w4VBA3m93Z)
Call ID: call_i8zDfMTdvmIG34w4VBA3m93Z
Args:
a: 3
b: 4解释：
- AI 理解了任务，决定先调用
add工具 Call ID是唯一标识符，用于追踪工具调用和结果的对应关系Args是传递给工具的参数
================================= Tool Message =================================
Name: add
7解释：
- 工具执行结果：3 + 4 = 7
- 这个结果会被传回 assistant 节点
================================== Ai Message ==================================
Tool Calls:
multiply (call_nE62D40lrGQC7b67nVOzqGYY)
Call ID: call_nE62D40lrGQC7b67nVOzqGYY
Args:
a: 7
b: 2解释：
- AI 收到结果 7，决定下一步调用
multiply - 注意参数
a: 7来自上一步的结果！
================================= Tool Message =================================
Name: multiply
14解释： 7 × 2 = 14
================================== Ai Message ==================================
Tool Calls:
divide (call_6Q9SjxD2VnYJqEBXFt7O1moe)
Call ID: call_6Q9SjxD2VnYJqEBXFt7O1moe
Args:
a: 14
b: 5解释：
- AI 继续推理，调用
divide - 参数
a: 14来自上一步结果
================================= Tool Message =================================
Name: divide
2.8解释： 14 ÷ 5 = 2.8
================================== Ai Message ==================================
The final result after performing the operations (3 + 4) × 2 ÷ 5 is 2.8.解释：
- 所有计算完成，AI 给出最终答案
- 这条消息没有
tool_calls，所以图执行结束
关键观察点：
- AI 自主决定了每一步的工具调用
- 每次都基于前一步的结果进行推理
- 最终给出了自然语言的回答
- 整个过程是自动化的，无需人工干预
🎓 核心知识点总结
LangGraph 特有概念
1. Agent vs Router
| 特性 | Router | Agent |
|---|---|---|
| 工具调用次数 | 一次 | 多次（循环） |
| 工具结果流向 | 直接返回用户 | 返回模型继续处理 |
| 决策次数 | 一次 | 多次（每次循环都决策） |
| 适用场景 | 简单分类、单次查询 | 复杂任务、多步骤推理 |
2. ReAct 循环
python
while True:
# Reason: AI 分析当前状态，决定行动
ai_message = assistant(state)
# 判断：需要工具吗？
if no_tool_calls(ai_message):
break # 不需要，返回答案
# Act: 调用工具
tool_result = tools(state)
# Observe: 更新状态，将结果传回 AI
state = update_state(tool_result)3. 条件边与循环
python
# 条件边：根据输出动态路由
builder.add_conditional_edges("assistant", tools_condition)
# 返回边：创建循环
builder.add_edge("tools", "assistant")
# 组合效果：
# assistant → [有工具调用] → tools → assistant → ...
# → [无工具调用] → END4. MessagesState
python
class MessagesState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
# 特点：
# 1. 自动维护对话历史
# 2. 支持多种消息类型
# 3. 使用 add_messages reducer 追加消息5. ToolNode
python
ToolNode(tools)
# 自动处理：
# - 解析 tool_calls
# - 调用函数
# - 包装结果为 ToolMessage
# - 追踪 tool_call_idPython 特有知识点
1. 函数类型注解
python
def add(a: int, b: int) -> int:
"""Adds a and b."""
return a + b
# a: int ← 参数类型注解
# -> int ← 返回类型注解
# """...""" ← 文档字符串作用：
- IDE 提供智能提示
- 类型检查工具（mypy）验证代码
- LangChain 解析用于工具描述
2. Docstring 规范（Google 风格）
python
def multiply(a: int, b: int) -> int:
"""Multiply a and b. # 简短描述（必须）
Args: # 参数列表（可选）
a: first int
b: second int
Returns: # 返回值（可选）
The product of a and b
Raises: # 异常（可选）
ValueError: If a or b is negative
"""
return a * b3. 列表拼接
python
[sys_msg] + state["messages"]
# 等价于：
result = []
result.append(sys_msg)
result.extend(state["messages"])4. 对象属性检查
python
hasattr(obj, "attribute_name")
# 示例：
hasattr(last_message, "tool_calls")
# 检查 last_message 对象是否有 tool_calls 属性💡 最佳实践
1. 何时使用 Agent？
✅ 适用场景：
- 需要多步骤推理的任务（如数学计算、数据分析）
- 需要根据中间结果动态调整策略
- 工具调用顺序不固定，需要 AI 自主决策
- 任务复杂度高，需要灵活应对
❌ 不适用场景：
- 简单的单次工具调用（用 Router 即可）
- 固定流程的任务（用普通图即可）
- 不需要循环推理的场景
2. 并行工具调用的选择
python
# 场景 1：数学计算 → parallel_tool_calls=False
llm_with_tools = llm.bind_tools([add, multiply, divide], parallel_tool_calls=False)
# 原因：步骤有依赖关系，必须顺序执行
# 场景 2：信息收集 → parallel_tool_calls=True（默认）
llm_with_tools = llm.bind_tools([search_web, query_db, call_api])
# 原因：三个工具可以同时调用，提高效率3. 系统提示的重要性
python
# ✅ 好的系统提示
sys_msg = SystemMessage(content="""
You are a helpful assistant tasked with performing arithmetic on a set of inputs.
- Break down complex calculations into steps
- Use the appropriate tool for each operation
- Show your reasoning
""")
# ❌ 不好的系统提示
sys_msg = SystemMessage(content="You are helpful.")
# 太模糊，AI 可能不知道如何使用工具4. 防止无限循环
Agent 有可能陷入无限循环，需要设置限制：
python
# 方法 1：使用 recursion_limit
react_graph = builder.compile(recursion_limit=10)
# 最多执行 10 次循环
# 方法 2：在状态中计数
class MessagesStateWithCount(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
iterations: int
def assistant(state: MessagesStateWithCount):
if state.get("iterations", 0) > 10:
return {"messages": [AIMessage(content="Reached maximum iterations")]}
# ... 正常逻辑
return {
"messages": [llm_with_tools.invoke(...)],
"iterations": state.get("iterations", 0) + 1
}5. 工具函数设计原则
python
# ✅ 好的工具设计
def search_database(query: str, limit: int = 10) -> list[dict]:
"""Search the database for matching records.
Args:
query: Search query string
limit: Maximum number of results (default: 10)
Returns:
List of matching records
"""
# 实现...
return results
# 特点：
# - 清晰的文档字符串
# - 类型注解
# - 默认参数值
# - 返回结构化数据python
# ❌ 不好的工具设计
def search(q): # 没有类型注解
"""搜索""" # 文档不清楚
# AI 不知道如何使用这个工具🚀 进阶技巧
1. 带记忆的 Agent
python
class MessagesStateWithMemory(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
user_preferences: dict # 存储用户偏好
conversation_summary: str # 对话摘要
def assistant(state: MessagesStateWithMemory):
# 可以访问历史偏好
preferences = state.get("user_preferences", {})
# 构建带上下文的提示
context = f"User preferences: {preferences}\n"
prompt = context + "\n".join([m.content for m in state["messages"]])
response = llm_with_tools.invoke(prompt)
return {"messages": [response]}2. 动态工具选择
python
def assistant(state: MessagesState):
# 根据对话内容动态选择可用工具
last_message = state["messages"][-1].content
if "weather" in last_message:
tools = [get_weather, get_forecast]
elif "database" in last_message:
tools = [query_db, insert_db, update_db]
else:
tools = all_tools
llm_with_selected_tools = llm.bind_tools(tools)
response = llm_with_selected_tools.invoke(state["messages"])
return {"messages": [response]}3. 错误处理
python
def tools_node(state: MessagesState):
last_message = state["messages"][-1]
tool_messages = []
for tool_call in last_message.tool_calls:
try:
tool = next(t for t in tools if t.name == tool_call["name"])
result = tool.invoke(tool_call["args"])
tool_messages.append(ToolMessage(
content=str(result),
tool_call_id=tool_call["id"]
))
except Exception as e:
# 返回错误信息给 AI
tool_messages.append(ToolMessage(
content=f"Error: {str(e)}",
tool_call_id=tool_call["id"],
status="error"
))
return {"messages": tool_messages}4. 流式输出
python
# 使用 stream 方法查看实时执行过程
for chunk in react_graph.stream({"messages": [HumanMessage("Calculate...")]}):
print(chunk)
# 输出：
# {'assistant': {'messages': [AIMessage(tool_calls=[...])]}}
# {'tools': {'messages': [ToolMessage(content="7")]}}
# {'assistant': {'messages': [AIMessage(tool_calls=[...])]}}
# ...📊 Agent 变体对比
1. 基础 Agent（本教程）
python
builder.add_edge("tools", "assistant") # 工具结果总是回到 assistant特点： 简单、直接，适合大多数场景
2. 条件性返回 Agent
python
def should_continue(state: MessagesState):
last_message = state["messages"][-1]
# 可以根据工具结果决定是否继续
if "error" in last_message.content:
return END
return "assistant"
builder.add_conditional_edges("tools", should_continue, ["assistant"])特点： 可以提前终止，处理错误情况
3. 多路径 Agent
python
def route_after_tool(state: MessagesState):
last_message = state["messages"][-1]
if "final" in last_message.content:
return "summarize" # 去总结节点
elif "error" in last_message.content:
return "error_handler" # 去错误处理节点
else:
return "assistant" # 继续推理
builder.add_conditional_edges("tools", route_after_tool,
["assistant", "summarize", "error_handler"])特点： 更复杂的控制流，适合大型应用
🎯 实际应用案例
案例 1：研究助手
python
tools = [
search_web, # 搜索网络
read_paper, # 读取论文
summarize_text, # 总结文本
save_notes # 保存笔记
]
# 用户：Research recent advances in quantum computing
# Agent 工作流：
# 1. search_web("quantum computing 2024")
# 2. read_paper(paper_url_1)
# 3. summarize_text(paper_content_1)
# 4. read_paper(paper_url_2)
# 5. summarize_text(paper_content_2)
# 6. save_notes(combined_summary)
# 7. 返回研究总结案例 2：数据分析助手
python
tools = [
query_database, # 查询数据库
calculate_stats, # 计算统计
create_chart, # 生成图表
export_report # 导出报告
]
# 用户：Analyze sales data for Q4 and create a report
# Agent 工作流：
# 1. query_database("SELECT * FROM sales WHERE quarter = 'Q4'")
# 2. calculate_stats(sales_data)
# 3. create_chart(stats, type="bar")
# 4. create_chart(stats, type="pie")
# 5. export_report(charts, stats)
# 6. 返回报告链接案例 3：客户支持助手
python
tools = [
search_knowledge_base, # 搜索知识库
check_order_status, # 查询订单
create_ticket, # 创建工单
send_email # 发送邮件
]
# 用户：I haven't received my order #12345
# Agent 工作流：
# 1. check_order_status("12345")
# 2. search_knowledge_base("delayed shipping")
# 3. 发现问题：物流延误
# 4. create_ticket(order_id="12345", issue="shipping_delay")
# 5. send_email(user, "Your order is delayed, ticket created")
# 6. 返回解决方案和工单号🔍 常见问题
Q1: Agent 什么时候会停止循环？
答： 当 assistant 节点返回的 AIMessage 不包含 tool_calls 时：
python
# 会继续循环
AIMessage(tool_calls=[ToolCall(name="add", ...)])
# 会停止（进入 END）
AIMessage(content="The answer is 42")Q2: 如何让 Agent 调用特定工具？
答： 通过系统提示引导：
python
sys_msg = SystemMessage(content="""
You are a helpful assistant.
When the user asks about weather, ALWAYS use the get_weather tool.
When the user asks about time, ALWAYS use the get_time tool.
""")或者使用 tool_choice 参数：
python
# 强制使用特定工具
llm_with_tools = llm.bind_tools(tools, tool_choice="get_weather")
# 强制调用任意工具（不能直接回答）
llm_with_tools = llm.bind_tools(tools, tool_choice="any")
# 禁止调用工具
llm_with_tools = llm.bind_tools(tools, tool_choice="none")Q3: 如何查看 Agent 的执行过程？
方法 1：使用 stream
python
for chunk in graph.stream(input_data):
print(chunk)方法 2：使用 LangSmith
python
import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_PROJECT"] = "my-agent"
# 在 LangSmith 网站查看详细 trace方法 3：自定义日志
python
def assistant(state: MessagesState):
print(f"[Assistant] Processing {len(state['messages'])} messages")
response = llm_with_tools.invoke(state["messages"])
print(f"[Assistant] Response: {response}")
return {"messages": [response]}Q4: 工具调用失败怎么办？
答： 在 ToolNode 中处理异常：
python
def custom_tools_node(state: MessagesState):
last_message = state["messages"][-1]
tool_messages = []
for tool_call in last_message.tool_calls:
try:
result = execute_tool(tool_call)
tool_messages.append(ToolMessage(
content=str(result),
tool_call_id=tool_call["id"]
))
except Exception as e:
# 将错误返回给 AI，让它重试或换方法
tool_messages.append(ToolMessage(
content=f"Error: {str(e)}. Please try a different approach.",
tool_call_id=tool_call["id"]
))
return {"messages": tool_messages}Q5: 如何限制 Agent 的执行成本？
方法 1：限制循环次数
python
graph = builder.compile(recursion_limit=5)方法 2：限制 token 使用
python
class TokenTrackingState(MessagesState):
total_tokens: int
def assistant(state: TokenTrackingState):
if state.get("total_tokens", 0) > 10000:
return {"messages": [AIMessage(content="Token limit reached")]}
# 调用 LLM 并记录 token
response = llm_with_tools.invoke(state["messages"])
tokens_used = response.response_metadata.get("token_usage", {}).get("total_tokens", 0)
return {
"messages": [response],
"total_tokens": state.get("total_tokens", 0) + tokens_used
}🏭 生产级 Agent：create_agent() 函数
LangChain 提供了 create_agent() 函数，这是一个生产级的 Agent 实现，基于 LangGraph 构建图运行时，包含节点（执行步骤）和边（连接）。
基础用法
python
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
# 定义工具
@tool
def search(query: str) -> str:
"""Search for information."""
return f"Results for: {query}"
@tool
def get_weather(city: str) -> str:
"""Get weather for a city."""
return f"Weather in {city}: Sunny, 25°C"
# 创建模型
model = ChatOpenAI(model="gpt-4o-mini", temperature=0.1, max_tokens=1000)
# 创建 Agent
agent = create_agent(
model,
tools=[search, get_weather],
system_prompt="You are a helpful assistant. Be concise and accurate."
)
# 调用 Agent
result = agent.invoke(
{"messages": [{"role": "user", "content": "What's the weather in San Francisco?"}]}
)@tool 装饰器
@tool 装饰器是 LangChain 推荐的工具定义方式：
python
from langchain_core.tools import tool
@tool
def search(query: str) -> str:
"""Search for information.
Args:
query: The search query string
"""
return f"Results for: {query}"
@tool
def calculate(expression: str) -> float:
"""Calculate a mathematical expression.
Args:
expression: A math expression like '2 + 2 * 3'
"""
return eval(expression) # 注意：生产环境需要更安全的实现与普通函数的对比：
python
# 普通函数方式（之前教程使用的）
def add(a: int, b: int) -> int:
"""Adds a and b."""
return a + b
# @tool 装饰器方式（推荐）
@tool
def add(a: int, b: int) -> int:
"""Adds a and b."""
return a + b
# 区别：
# - @tool 自动将函数转换为 LangChain Tool 对象
# - 自动解析类型注解和文档字符串
# - 支持更多高级功能（如验证、错误处理等）动态模型选择
使用 @wrap_model_call 装饰器实现运行时模型选择：
python
from langchain.agents.middleware import wrap_model_call
@wrap_model_call
def dynamic_model_selection(request, handler):
"""根据对话长度动态选择模型"""
message_count = len(request.state["messages"])
# 对话较长时使用更强大的模型
if message_count > 10:
model = ChatOpenAI(model="gpt-4o") # 高级模型
else:
model = ChatOpenAI(model="gpt-4o-mini") # 基础模型
return handler(request.override(model=model))
# 应用中间件
agent = create_agent(
model,
tools=tools,
middleware=[dynamic_model_selection]
)使用场景：
- 根据任务复杂度选择不同模型
- 根据用户等级使用不同模型
- 成本优化：简单问题用便宜模型，复杂问题用强力模型
工具错误处理
使用 @wrap_tool_call 中间件处理工具调用错误：
python
from langchain.agents.middleware import wrap_tool_call
from langchain_core.messages import ToolMessage
@wrap_tool_call
def handle_tool_errors(request, handler):
"""自定义工具错误处理"""
try:
return handler(request)
except Exception as e:
# 返回友好的错误消息给 Agent
return ToolMessage(
content=f"Tool error: {str(e)}. Please try a different approach.",
tool_call_id=request.tool_call["id"]
)
agent = create_agent(
model,
tools=tools,
middleware=[handle_tool_errors]
)结构化输出
Agent 可以返回结构化数据而不是纯文本：
ToolStrategy（推荐，兼容性好）
python
from langchain.agents.structured_output import ToolStrategy
from pydantic import BaseModel
class ContactInfo(BaseModel):
name: str
email: str
phone: str
agent = create_agent(
model="gpt-4o-mini",
tools=[search_tool],
response_format=ToolStrategy(ContactInfo)
)
# 返回结构化的 ContactInfo 对象
result = agent.invoke({"messages": [{"role": "user", "content": "Find contact info for John Doe"}]})ProviderStrategy（需要模型原生支持）
python
from langchain.agents.structured_output import ProviderStrategy
agent = create_agent(
model="gpt-4o-mini",
tools=[search_tool],
response_format=ProviderStrategy(ContactInfo)
)两种策略的区别：
| 特性 | ToolStrategy | ProviderStrategy |
|---|---|---|
| 兼容性 | 支持所有工具调用模型 | 需要模型原生支持 |
| 实现方式 | 通过工具调用返回结构 | 使用模型内置功能 |
| 推荐程度 | ✅ 推荐 | 特定场景使用 |
自定义状态管理
扩展 AgentState 以添加自定义状态：
python
from typing import TypedDict
from langgraph.graph import MessagesState
class CustomState(MessagesState):
"""扩展的 Agent 状态"""
user_preferences: dict # 用户偏好
conversation_context: str # 对话上下文
tool_usage_count: int # 工具调用计数
agent = create_agent(
model,
tools=[tool1, tool2],
state_schema=CustomState
)流式输出
实时获取 Agent 执行进度：
python
# 流式执行
for chunk in agent.stream(
{"messages": [{"role": "user", "content": "Search for AI news"}]},
stream_mode="values"
):
latest_message = chunk["messages"][-1]
print(f"Agent: {latest_message.content}")
# 输出示例：
# Agent: I'll search for AI news...
# Agent: [Tool call: search("AI news 2024")]
# Agent: Here are the latest AI news...stream_mode 选项：
| 模式 | 说明 |
|---|---|
"values" | 返回完整状态值 |
"updates" | 只返回状态更新 |
"messages" | 只返回消息流 |
中间件（Middleware）
中间件提供了强大的扩展能力：
python
from langchain.agents.middleware import wrap_model_call
# 日志中间件
@wrap_model_call
def logging_middleware(request, handler):
"""记录所有模型调用"""
print(f"[LOG] Calling model with {len(request.state['messages'])} messages")
result = handler(request)
print(f"[LOG] Model response: {result.content[:100]}...")
return result
# 验证中间件
@wrap_model_call
def validation_middleware(request, handler):
"""验证响应内容"""
result = handler(request)
# 检查是否包含敏感信息
if "password" in result.content.lower():
result.content = "[Redacted for security]"
return result
# 组合多个中间件
agent = create_agent(
model,
tools=tools,
middleware=[logging_middleware, validation_middleware]
)完整的 create_agent 示例
python
## 完整 create_agent 示例代码
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from pydantic import BaseModel
# 1. 定义工具
@tool
def search_web(query: str) -> str:
"""Search the web for information.
Args:
query: Search query string
"""
# 模拟搜索结果
return f"Search results for '{query}': Found 10 relevant articles."
@tool
def get_weather(city: str) -> str:
"""Get current weather for a city.
Args:
city: Name of the city
"""
# 模拟天气数据
return f"Weather in {city}: Sunny, 22°C, Humidity 45%"
@tool
def calculate(expression: str) -> str:
"""Calculate a mathematical expression.
Args:
expression: Math expression to evaluate
"""
try:
result = eval(expression)
return f"Result: {result}"
except Exception as e:
return f"Error: {str(e)}"
# 2. 创建模型
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# 3. 创建 Agent
agent = create_agent(
model,
tools=[search_web, get_weather, calculate],
system_prompt="""You are a helpful assistant with access to search, weather, and calculator tools.
- Use search_web for information queries
- Use get_weather for weather questions
- Use calculate for math problems
Always explain your reasoning."""
)
# 4. 执行查询
result = agent.invoke({
"messages": [{"role": "user", "content": "What's the weather in Tokyo?"}]
})
print(result["messages"][-1].content)
# 5. 流式执行
print("\n=== 流式输出 ===")
for chunk in agent.stream(
{"messages": [{"role": "user", "content": "Calculate 15 * 8 + 42"}]},
stream_mode="values"
):
print(chunk["messages"][-1].content)create_agent vs 手动构建图
| 特性 | create_agent() | 手动构建 StateGraph |
|---|---|---|
| 代码量 | 少 | 多 |
| 灵活性 | 中等 | 高 |
| 学习曲线 | 低 | 高 |
| 生产就绪 | ✅ 是 | 需要额外工作 |
| 自定义程度 | 通过中间件扩展 | 完全自定义 |
| 适用场景 | 标准 Agent 场景 | 复杂自定义流程 |
选择建议：
- 使用 create_agent()：快速开发、标准场景、生产部署
- 使用手动构建：需要完全自定义流程、特殊的路由逻辑、学习目的
📖 扩展阅读
总结：Agent 是 LangGraph 的核心模式，通过 "推理-行动-观察" 的循环，让 AI 能够自主完成复杂任务。本教程涵盖了两种构建方式：
- 手动构建 StateGraph：完全控制图结构，适合学习和高度自定义场景
- 使用 create_agent()：生产级封装，快速开发和部署
掌握这两种方法，是开发强大 AI 应用的基础！