LangGraph Agent 详细解读
📚 概述
本文档详细解读 LangGraph 中的 Agent(智能体) 架构。这是构建智能 AI 应用的核心模式,通过让 AI 模型自主决策、调用工具、并基于结果继续推理,实现复杂任务的自动化处理。
🎯 核心概念
什么是 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. 定义工具函数
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 的一等公民:
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:
# LLM 会看到类似这样的工具描述:
# Tool: add
# Description: Adds a and b.
# Parameters:
# - a: first int
# - b: second int
这就是为什么 LLM 能够"理解"工具的作用!
2. 绑定工具到模型
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o")
llm_with_tools = llm.bind_tools(tools, parallel_tool_calls=False)
关键参数:parallel_tool_calls=False
这个参数控制模型是否可以同时调用多个工具:
# 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. 定义状态
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 内置的状态类,专门用于聊天场景:
class MessagesState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
核心特性:
- 自动消息管理:维护对话历史
- 消息追加:使用
add_messages
reducer 自动追加新消息 - 类型安全:支持不同类型的消息(HumanMessage、AIMessage、ToolMessage 等)
消息类型:
from langchain_core.messages import (
HumanMessage, # 用户消息
AIMessage, # AI 回复
SystemMessage, # 系统提示
ToolMessage # 工具返回结果
)
4. 定义 Assistant 节点
def assistant(state: MessagesState):
return {"messages": [llm_with_tools.invoke([sys_msg] + state["messages"])]}
代码解析:
[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 才能理解当前处于对话的哪个阶段
返回格式:
return {"messages": [new_message]}
# 返回的消息会被追加到 state["messages"] 中
# 因为 MessagesState 使用了 add_messages reducer
5. 构建 Agent 图
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()
LangGraph 知识点:ToolNode
ToolNode
是 LangGraph 提供的预构建节点,用于执行工具调用:
ToolNode(tools)
# 自动处理:
# 1. 解析 AIMessage 中的 tool_calls
# 2. 调用对应的工具函数
# 3. 将结果包装成 ToolMessage
# 4. 返回给下一个节点
等价的手动实现:
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
是预构建的条件函数,用于判断是否需要调用工具:
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 # 结束图执行
条件边的工作流程:
builder.add_conditional_edges("assistant", tools_condition)
# 完整流程:
# assistant 节点执行完毕
# ↓
# tools_condition 检查输出
# ↓
# 如果有 tool_calls → 路由到 "tools"
# 如果没有 tool_calls → 路由到 END
Agent 循环的关键:
builder.add_edge("tools", "assistant") # ← 这条边创建了循环!
# 完整循环:
# assistant → (有工具调用) → tools → assistant → ...
# → (无工具调用) → END
6. 执行 Agent
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. 查看输出
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 循环
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. 条件边与循环
# 条件边:根据输出动态路由
builder.add_conditional_edges("assistant", tools_condition)
# 返回边:创建循环
builder.add_edge("tools", "assistant")
# 组合效果:
# assistant → [有工具调用] → tools → assistant → ...
# → [无工具调用] → END
4. MessagesState
class MessagesState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
# 特点:
# 1. 自动维护对话历史
# 2. 支持多种消息类型
# 3. 使用 add_messages reducer 追加消息
5. ToolNode
ToolNode(tools)
# 自动处理:
# - 解析 tool_calls
# - 调用函数
# - 包装结果为 ToolMessage
# - 追踪 tool_call_id
Python 特有知识点
1. 函数类型注解
def add(a: int, b: int) -> int:
"""Adds a and b."""
return a + b
# a: int ← 参数类型注解
# -> int ← 返回类型注解
# """...""" ← 文档字符串
作用:
- IDE 提供智能提示
- 类型检查工具(mypy)验证代码
- LangChain 解析用于工具描述
2. Docstring 规范(Google 风格)
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 * b
3. 列表拼接
[sys_msg] + state["messages"]
# 等价于:
result = []
result.append(sys_msg)
result.extend(state["messages"])
4. 对象属性检查
hasattr(obj, "attribute_name")
# 示例:
hasattr(last_message, "tool_calls")
# 检查 last_message 对象是否有 tool_calls 属性
💡 最佳实践
1. 何时使用 Agent?
✅ 适用场景:
- 需要多步骤推理的任务(如数学计算、数据分析)
- 需要根据中间结果动态调整策略
- 工具调用顺序不固定,需要 AI 自主决策
- 任务复杂度高,需要灵活应对
❌ 不适用场景:
- 简单的单次工具调用(用 Router 即可)
- 固定流程的任务(用普通图即可)
- 不需要循环推理的场景
2. 并行工具调用的选择
# 场景 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. 系统提示的重要性
# ✅ 好的系统提示
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 有可能陷入无限循环,需要设置限制:
# 方法 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. 工具函数设计原则
# ✅ 好的工具设计
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
# 特点:
# - 清晰的文档字符串
# - 类型注解
# - 默认参数值
# - 返回结构化数据
# ❌ 不好的工具设计
def search(q): # 没有类型注解
"""搜索""" # 文档不清楚
# AI 不知道如何使用这个工具
🚀 进阶技巧
1. 带记忆的 Agent
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. 动态工具选择
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. 错误处理
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. 流式输出
# 使用 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(本教程)
builder.add_edge("tools", "assistant") # 工具结果总是回到 assistant
特点: 简单、直接,适合大多数场景
2. 条件性返回 Agent
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
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:研究助手
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:数据分析助手
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:客户支持助手
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
时:
# 会继续循环
AIMessage(tool_calls=[ToolCall(name="add", ...)])
# 会停止(进入 END)
AIMessage(content="The answer is 42")
Q2: 如何让 Agent 调用特定工具?
答: 通过系统提示引导:
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
参数:
# 强制使用特定工具
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
for chunk in graph.stream(input_data):
print(chunk)
方法 2:使用 LangSmith
import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_PROJECT"] = "my-agent"
# 在 LangSmith 网站查看详细 trace
方法 3:自定义日志
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 中处理异常:
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:限制循环次数
graph = builder.compile(recursion_limit=5)
方法 2:限制 token 使用
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 是 LangGraph 的核心模式,通过 "推理-行动-观察" 的循环,让 AI 能够自主完成复杂任务。掌握 Agent 的构建方法,是开发强大 AI 应用的基础!