【langgraph 从入门到精通graphApi 篇】消息处理与工具集成

第 4 章：消息处理与工具集成

4.1 本章目标

学完本章你将能够：

1. 理解 LangChain 消息体系（HumanMessage / AIMessage / ToolMessage / SystemMessage）
2. 掌握@tool装饰器定义工具函数
3. 学会 ToolNode 的集成方式和工作原理
4. 理解工具调用回环（agent → tools → agent 循环）

4.2 核心概念

四种消息类型

工具调用完整流程

4.3 实战

实战 1：搜索工具 Agent

fromtypingimportTypedDict,Annotated,Literalfromlanggraph.graphimportStateGraph,START,END,add_messagesfromlanggraph.prebuiltimportToolNode,tools_conditionfromlangchain_core.messagesimportBaseMessage,HumanMessage,SystemMessagefromlangchain_core.toolsimporttoolfromlangchain_openaiimportChatOpenAI# ============================================# 定义工具：使用 @tool 装饰器# ============================================@tooldefsearch_knowledge(query:str)->str:""" 在知识库中搜索信息。当用户询问事实性问题时使用此工具。 Args: query: 搜索关键词 """# 模拟搜索（实际项目中替换为数据库查询或搜索引擎 API）knowledge_base={"天气":"今天北京天气晴，温度 25°C，湿度 40%","Python":"Python 是一种高级编程语言，由 Guido van Rossum 于 1991 年创建","LangGraph":"LangGraph 是 LangChain 团队开发的有状态多步骤 AI 应用编排框架",}forkey,valueinknowledge_base.items():ifkeyinquery:returnvaluereturnf"未找到关于「{query}」的信息"@tooldefcalculate(expression:str)->str:""" 执行数学计算。当用户需要进行数学运算时使用此工具。 Args: expression: 数学表达式，如 "2 + 3 * 4" """try:# 安全计算（仅允许数字和基本运算符）result=eval(expression,{"__builtins__":{}},{})returnf"计算结果:{expression}={result}"exceptExceptionase:returnf"计算错误:{str(e)}"# ============================================# 定义 State# ============================================classState(TypedDict):messages:Annotated[list[BaseMessage],add_messages]# ============================================# 定义节点# ============================================# 初始化 LLM 并绑定工具llm=ChatOpenAI(model="gpt-4o",temperature=0)tools=[search_knowledge,calculate]# 注意：llm.bind_tools() 让 LLM 知道有哪些工具可用llm_with_tools=llm.bind_tools(tools)defagent(state:State)->dict:""" Agent 节点：调用 LLM，LLM 决定是直接回复还是调用工具。 """# 添加系统提示词messages=state["messages"]ifnotmessagesornotisinstance(messages[0],SystemMessage):messages=[SystemMessage(content="你是一个有用的助手，可以使用工具来回答问题。")]+list(messages)response=llm_with_tools.invoke(messages)return{"messages":[response]}# ============================================# 构建图# ============================================builder=StateGraph(State)# 注册节点builder.add_node("agent",agent)builder.add_node("tools",ToolNode(tools))# ToolNode 自动处理工具调用# 添加边builder.add_edge(START,"agent")# 条件边：如果 agent 返回了 tool_calls，则路由到 tools 节点# 否则结束（tools_condition 是内置路由函数）builder.add_conditional_edges("agent",tools_condition,# 内置函数：检查 AIMessage 是否有 tool_calls# tools_condition 返回值："tools" 或 "__end__")# tools 执行完后回到 agent（继续生成回复）builder.add_edge("tools","agent")graph=builder.compile()# ============================================# 测试# ============================================defask(question:str):"""辅助函数：发送问题并打印结果"""print(f"\n{'='*50}")print(f"用户:{question}")result=graph.invoke({"messages":[HumanMessage(content=question)]})last_msg=result["messages"][-1]print(f"AI:{last_msg.content}")# 测试需要工具的问题ask("今天天气怎么样？")ask("Python 是什么？")ask("计算 15 * 8 + 20")ask("你好，今天心情好吗？")# 不需要工具的问题

实战 2：手动构建工具调用循环（深入理解底层）

fromtypingimportTypedDict,Annotated,Literalfromlanggraph.graphimportStateGraph,START,END,add_messagesfromlangchain_core.messagesimportBaseMessage,HumanMessage,AIMessage,ToolMessage,SystemMessagefromlangchain_core.toolsimporttoolfromlangchain_openaiimportChatOpenAI# ============================================# 定义工具# ============================================@tooldefget_weather(city:str)->str:"""获取指定城市的天气"""returnf"{city}今天天气晴，25°C"@tooldefget_time()->str:"""获取当前时间"""fromdatetimeimportdatetimereturndatetime.now().strftime("%Y-%m-%d %H:%M:%S")# ============================================# 准备工作# ============================================tools=[get_weather,get_time]tools_by_name={t.name:tfortintools}# 工具名 → 工具对象llm=ChatOpenAI(model="gpt-4o",temperature=0)llm_with_tools=llm.bind_tools(tools)# ============================================# 定义 State# ============================================classState(TypedDict):messages:Annotated[list[BaseMessage],add_messages]# ============================================# 定义节点# ============================================defagent(state:State)->dict:"""Agent 节点：调用 LLM"""response=llm_with_tools.invoke([SystemMessage(content="你是一个有用的助手")]+list(state["messages"]))return{"messages":[response]}deftool_executor(state:State)->dict:""" 手动执行工具调用（不依赖 ToolNode）。 这样你可以完全控制工具执行逻辑。 """last_message=state["messages"][-1]# 获取最后一条 AIMessage 中的 tool_callstool_calls=last_message.tool_calls tool_messages=[]fortool_callintool_calls:tool_name=tool_call["name"]tool_args=tool_call["args"]tool_call_id=tool_call["id"]print(f" 🔧 调用工具:{tool_name}({tool_args})")# 找到对应的工具并执行iftool_nameintools_by_name:tool_func=tools_by_name[tool_name]result=tool_func.invoke(tool_args)else:result=f"未知工具:{tool_name}"# 创建 ToolMessage，必须包含 tool_call_idtool_messages.append(ToolMessage(content=str(result),tool_call_id=tool_call_id))print(f" 📋 结果:{result}")return{"messages":tool_messages}defshould_continue(state:State)->Literal["tools","END"]:"""判断是否需要继续调用工具"""last_message=state["messages"][-1]ifhasattr(last_message,"tool_calls")andlast_message.tool_calls:return"tools"return"END"# ============================================# 构建图# ============================================builder=StateGraph(State)builder.add_node("agent",agent)builder.add_node("tools",tool_executor)builder.add_edge(START,"agent")builder.add_conditional_edges("agent",should_continue)builder.add_edge("tools","agent")graph=builder.compile()# 测试result=graph.invoke({"messages":[HumanMessage(content="北京天气怎么样？现在几点了？")]})print(f"\n最终回复:{result['messages'][-1].content}")

4.4 API 速查

4.5 错误与避坑指南

坑 1：ToolMessage 缺少 tool_call_id

# ❌ 错误写法ToolMessage(content="结果",tool_call_id="")# 空 ID 或错误 ID# → LLM 无法将工具结果与调用关联，导致混淆# ✅ 正确写法：从 AIMessage.tool_calls 中获取 tool_call_idfortool_callinlast_message.tool_calls:result=execute_tool(tool_call)tool_messages.append(ToolMessage(content=result,tool_call_id=tool_call["id"]))

坑 2：工具函数返回值类型不兼容

# ❌ 错误写法@tooldefbad_tool(query:str)->dict:# 返回 dict → LLM 可能无法理解return{"status":"ok","data":[...]}# ✅ 正确写法@tooldefgood_tool(query:str)->str:# 返回 str → LLM 可以理解returnf"查询结果：找到 3 条相关记录"

坑 3：忘记 bind_tools

# ❌ 错误写法llm=ChatOpenAI(model="gpt-4o")# 直接使用 llm，LLM 不知道有哪些工具可用# → LLM 不会生成 tool_calls，工具永远不会被调用# ✅ 正确写法llm=ChatOpenAI(model="gpt-4o")llm_with_tools=llm.bind_tools(tools)# 必须绑定工具

坑 4：工具调用死循环

# ❌ 错误：tools_condition 未正确路由builder.add_conditional_edges("agent",tools_condition)# 如果 tools_condition 始终返回 "tools"，会无限循环# ✅ 正确：tools → agent 后，agent 检查是否还有 tool_callsbuilder.add_conditional_edges("agent",tools_condition)builder.add_edge("tools","agent")# tools_condition 在 agent 没有 tool_calls 时返回 END

4.6 最佳实践总结

1. 工具函数添加详细的 docstring：LLM 根据 docstring 决定何时调用哪个工具，描述越清晰，调用越准确
2. 使用handle_tool_errors=True：ToolNode(tools, handle_tool_errors=True)优雅处理工具执行异常
3. 工具返回字符串：LLM 更容易理解文本格式的工具结果
4. 复杂场景用create_react_agent，简单场景手写 StateGraph：不要过度设计
5. 工具设计遵循"单一职责"：每个工具只做一件事，多个简单工具胜过一个大而全的工具