第 5 章 · 状态管理与工具调用
本章目标：
- 掌握 MessagesState 的消息结构
- 学会定义自定义状态（TypedDict 扩展）
- 理解 @tool 装饰器与 ToolNode
- 实现完整的 Agent 图（LLM 节点 + ToolNode）
5.1 MessagesState 详解
MessagesState 是 LangGraph 的内置状态类型，包含 messages 字段：
python
from langgraph.graph import MessagesState
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
# messages 字段结构
messages = [
HumanMessage(content="你好"), # 用户消息
AIMessage(content="你好！有什么可以帮你的？"), # AI 回复
# 工具调用时会插入
AIMessage(
content="",
tool_calls=[{"id": "call_123", "name": "get_weather", "args": {"city": "北京"}}]
),
ToolMessage(content="北京今天晴朗，25°C", tool_call_id="call_123") # 工具结果
]5.2 自定义状态
python
from langgraph.graph import StateGraph, MessagesState, START, END
from typing import TypedDict, Annotated
import operator
class AgentState(MessagesState):
# 使用 Annotated + operator.add 实现消息追加（非覆盖）
documents: Annotated[list, operator.add] # 检索到的文档列表
step_count: int # 已执行步骤数
final_answer: str # 最终答案
error: str # 错误信息（如有）
# 初始化状态
initial_state = AgentState(
messages=[],
documents=[],
step_count=0,
final_answer="",
error=""
)5.3 @tool 装饰器
python
from langchain_core.tools import tool
from typing import Literal
@tool
def get_weather(city: str) -> str:
"""获取指定城市的天气信息"""
# 实际项目中应调用真实 API
weather_data = {
"北京": "晴朗，25°C",
"上海": "多云，22°C",
"广州": "小雨，28°C"
}
return weather_data.get(city, "未知城市")
@tool
def search_documents(query: str) -> list[str]:
"""搜索相关文档"""
# 实际项目中应调用向量数据库
return [
f"文档1: {query} 相关内容 A",
f"文档2: {query} 相关内容 B"
]
# 工具列表
tools = [get_weather, search_documents]5.4 ToolNode：自动处理工具调用
python
from langgraph.prebuilt import ToolNode
# ToolNode 会自动：
# 1. 解析 LLM 输出的 tool_calls
# 2. 调用对应工具函数
# 3. 将结果包装为 ToolMessage 放回状态
tool_node = ToolNode(tools)
# 使用示例
def llm_node(state: AgentState) -> dict:
# 调用 LLM（这里用模拟）
# 实际使用 createGateway 或 createOpenAICompatible
return {
"messages": [
AIMessage(
content="",
tool_calls=[{
"id": "call_123",
"name": "get_weather",
"args": {"city": "北京"}
}]
)
]
}5.5 完整 Agent 图
python
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode
from typing import Annotated, operator
class AgentState(MessagesState):
step_count: Annotated[int, operator.add]
def create_agent_graph():
# 节点
def llm_node(state: AgentState) -> dict:
# 模拟 LLM 调用
# 实际: from ai import createGateway, streamText
# gateway = createGateway({ apiKey: process.env.AI_GATEWAY_API_KEY })
# model = gateway('openai/gpt-4o')
last_msg = state["messages"][-1]
if last_msg.type == "human":
return {
"messages": [{
"role": "assistant",
"content": "",
"tool_calls": [{
"id": "call_1",
"name": "get_weather",
"args": {"city": "北京"}
}]
}],
"step_count": 1
}
return {"step_count": state.get("step_count", 0) + 1}
# 判断是否需要继续调用工具
def should_continue(state: AgentState) -> str:
last_msg = state["messages"][-1]
if hasattr(last_msg, 'tool_calls') and last_msg.tool_calls:
return "tools"
return "end"
# 构建图
graph = StateGraph(AgentState)
graph.add_node("llm", llm_node)
graph.add_node("tools", ToolNode([get_weather, search_documents]))
graph.add_edge(START, "llm")
graph.add_conditional_edges("llm", should_continue, {"tools": "tools", "end": END})
graph.add_edge("tools", "llm")
return graph.compile()
# 执行
agent = create_agent_graph()
result = agent.invoke({
"messages": [{"role": "user", "content": "北京天气怎么样？"}]
})
print(result["messages"][-1]["content"])本章小结
MessagesState管理对话消息历史- 自定义状态使用
TypedDict扩展MessagesState @tool装饰器定义工具，docstring 作为工具描述ToolNode自动处理工具调用和结果回写- 条件边实现工具调用的循环逻辑
🛠️ 动手实践
- 定义两个自定义工具（计算器、日期查询），组装成 Agent 图
- 实现工具调用失败时的错误处理（返回错误消息给 LLM）
- 扩展状态，添加
conversation_history字段保留多轮对话