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

文章目录第 4 章消息处理与工具集成4.1 本章目标4.2 核心概念四种消息类型工具调用完整流程4.3 实战实战 1搜索工具 Agent实战 2手动构建工具调用循环深入理解底层4.4 API 速查4.5 错误与避坑指南坑 1ToolMessage 缺少 tool_call_id坑 2工具函数返回值类型不兼容坑 3忘记 bind_tools坑 4工具调用死循环4.6 最佳实践总结第 4 章消息处理与工具集成4.1 本章目标学完本章你将能够理解 LangChain 消息体系HumanMessage / AIMessage / ToolMessage / SystemMessage掌握tool装饰器定义工具函数学会 ToolNode 的集成方式和工作原理理解工具调用回环agent → tools → agent 循环4.2 核心概念四种消息类型消息类型角色使用场景示例HumanMessage用户用户输入HumanMessage(content今天天气怎么样)AIMessageAILLM 的回复AIMessage(content今天晴25°C)ToolMessage工具工具执行结果ToolMessage(content25°C, tool_call_id...)SystemMessage系统系统提示词/角色设定SystemMessage(content你是一个有用的助手)工具调用完整流程工具函数ToolNodetools_conditionAgent 节点 (LLM)用户工具函数ToolNodetools_conditionAgent 节点 (LLM)用户北京今天天气怎么样LLM 决策需要调用 get_weather 工具返回 AIMessage(含 tool_calls)检测到 tool_calls路由到 tools 节点调用 get_weather(北京)返回 25°C晴返回 ToolMessage(结果)LLM 基于工具结果生成最终回复北京今天天气晴温度 25°C4.3 实战实战 1搜索工具 AgentfromtypingimportTypedDict,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: 搜索关键词 # 模拟搜索实际项目中替换为数据库查询或搜索引擎 APIknowledge_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:# 安全计算仅允许数字和基本运算符resulteval(expression,{__builtins__:{}},{})returnf计算结果:{expression}{result}exceptExceptionase:returnf计算错误:{str(e)}# # 定义 State# classState(TypedDict):messages:Annotated[list[BaseMessage],add_messages]# # 定义节点# # 初始化 LLM 并绑定工具llmChatOpenAI(modelgpt-4o,temperature0)tools[search_knowledge,calculate]# 注意llm.bind_tools() 让 LLM 知道有哪些工具可用llm_with_toolsllm.bind_tools(tools)defagent(state:State)-dict: Agent 节点调用 LLMLLM 决定是直接回复还是调用工具。 # 添加系统提示词messagesstate[messages]ifnotmessagesornotisinstance(messages[0],SystemMessage):messages[SystemMessage(content你是一个有用的助手可以使用工具来回答问题。)]list(messages)responsellm_with_tools.invoke(messages)return{messages:[response]}# # 构建图# builderStateGraph(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)graphbuilder.compile()# # 测试# defask(question:str):辅助函数发送问题并打印结果print(f\n{*50})print(f用户:{question})resultgraph.invoke({messages:[HumanMessage(contentquestion)]})last_msgresult[messages][-1]print(fAI:{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°Ctooldefget_time()-str:获取当前时间fromdatetimeimportdatetimereturndatetime.now().strftime(%Y-%m-%d %H:%M:%S)# # 准备工作# tools[get_weather,get_time]tools_by_name{t.name:tfortintools}# 工具名 → 工具对象llmChatOpenAI(modelgpt-4o,temperature0)llm_with_toolsllm.bind_tools(tools)# # 定义 State# classState(TypedDict):messages:Annotated[list[BaseMessage],add_messages]# # 定义节点# defagent(state:State)-dict:Agent 节点调用 LLMresponsellm_with_tools.invoke([SystemMessage(content你是一个有用的助手)]list(state[messages]))return{messages:[response]}deftool_executor(state:State)-dict: 手动执行工具调用不依赖 ToolNode。 这样你可以完全控制工具执行逻辑。 last_messagestate[messages][-1]# 获取最后一条 AIMessage 中的 tool_callstool_callslast_message.tool_calls tool_messages[]fortool_callintool_calls:tool_nametool_call[name]tool_argstool_call[args]tool_call_idtool_call[id]print(f 调用工具:{tool_name}({tool_args}))# 找到对应的工具并执行iftool_nameintools_by_name:tool_functools_by_name[tool_name]resulttool_func.invoke(tool_args)else:resultf未知工具:{tool_name}# 创建 ToolMessage必须包含 tool_call_idtool_messages.append(ToolMessage(contentstr(result),tool_call_idtool_call_id))print(f 结果:{result})return{messages:tool_messages}defshould_continue(state:State)-Literal[tools,END]:判断是否需要继续调用工具last_messagestate[messages][-1]ifhasattr(last_message,tool_calls)andlast_message.tool_calls:returntoolsreturnEND# # 构建图# builderStateGraph(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)graphbuilder.compile()# 测试resultgraph.invoke({messages:[HumanMessage(content北京天气怎么样现在几点了)]})print(f\n最终回复:{result[messages][-1].content})4.4 API 速查API完整签名入参说明返回值说明HumanMessage(content)HumanMessage(content: str)content: 文本内容HumanMessage用户消息AIMessage(content)AIMessage(content: str | None)content: 文本内容AIMessageAI 回复ToolMessage(content, tool_call_id)ToolMessage(content: str, tool_call_id: str)content: 工具结果;tool_call_id: 调用 IDToolMessage工具执行结果SystemMessage(content)SystemMessage(content: str)content: 系统提示词SystemMessage系统提示tool装饰器tool函数必须包含 docstring带 tool 元数据的函数将函数标记为工具ToolNode(tools)ToolNode(tools: list)tools: 工具列表可调用节点自动处理工具调用tools_condition内置函数无路由函数判断 AIMessage 是否有 tool_callsllm.bind_tools(tools)bind_tools(tools: list)tools: 工具列表Runnable绑定工具的 LLM让 LLM 知道可用工具4.5 错误与避坑指南坑 1ToolMessage 缺少 tool_call_id# ❌ 错误写法ToolMessage(content结果,tool_call_id)# 空 ID 或错误 ID# → LLM 无法将工具结果与调用关联导致混淆# ✅ 正确写法从 AIMessage.tool_calls 中获取 tool_call_idfortool_callinlast_message.tool_calls:resultexecute_tool(tool_call)tool_messages.append(ToolMessage(contentresult,tool_call_idtool_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# ❌ 错误写法llmChatOpenAI(modelgpt-4o)# 直接使用 llmLLM 不知道有哪些工具可用# → LLM 不会生成 tool_calls工具永远不会被调用# ✅ 正确写法llmChatOpenAI(modelgpt-4o)llm_with_toolsllm.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 时返回 END4.6 最佳实践总结工具函数添加详细的 docstringLLM 根据 docstring 决定何时调用哪个工具描述越清晰调用越准确使用handle_tool_errorsTrueToolNode(tools, handle_tool_errorsTrue)优雅处理工具执行异常工具返回字符串LLM 更容易理解文本格式的工具结果复杂场景用create_react_agent简单场景手写 StateGraph不要过度设计工具设计遵循单一职责每个工具只做一件事多个简单工具胜过一个大而全的工具