第3篇-从零构建ReAct-Agent-LangGraph完整实战

从零构建 ReAct Agent——LangGraph 完整实战LangGraph 系列第 3 篇 · 用前两篇的组件知识亲手搭一个能自主调用工具的 AI Agent。前两篇讲完 State、Node、Edge 三大组件本篇直接上硬菜——从零构建一个 ReAct Agent。如果你曾经被llm.bind_tools()和ToolNode的关系绕晕过本篇会专门用一个章节帮你说清楚。读完这篇你不仅能写出能跑的 Agent还能准确说出每一行代码在干什么。一、ReAct 是什么为什么要用 Agent 模式1.1 ReAct Reasoning ActingReAct 是一种 LLM 使用范式核心是一个循环循环直到能回答问题 1. Reasoning推理LLM 分析当前信息决定需要什么 2. Acting行动调用工具获取新信息 3. Observation观察把工具返回结果加入上下文 4. 回到第 1 步举个例子用户问北京今天天气怎么样第 1 轮 Reasoning我需要调用天气查询工具 第 1 轮 Acting调用 get_weather(北京) 第 1 轮 Observation得到 北京晴天15°C 第 2 轮 Reasoning我已经有天气数据了可以回答用户 → 输出最终答案北京今天晴天气温 15°C1.2 为什么不用普通 LLM 调用普通 LLM 的知识有截止日期不知道实时天气、不会算复杂数学、不能查数据库。Agent 模式让 LLM 变成一个调度员——它不用自己知道答案只要能判断该调哪个工具就够了。1.3 前两篇知识的映射在 LangGraph 里实现 ReAct Agent本质上就是搭一张这样的图[START] → [agent: LLM推理] → [router: 有tool_calls?] ↑ │ │ yes │ no │ [tools: 执行工具] → [END] └──────────────────────┘State存对话历史messagesNodeagent_nodeLLM 推理tool_node执行工具Edge条件边根据tool_calls是否存在来决定循环还是结束二、环境准备pipinstalllanggraph langchain langchain-openai本文以 OpenAI 为例换成 DeepSeek / 千问 / 文心也一样改一下 ChatModel 就行。fromlangchain_openaiimportChatOpenAI llmChatOpenAI(modelgpt-4o,temperature0)三、第一步定义工具先准备三个工具模拟真实场景fromlangchain_core.toolsimporttooltooldefsearch_info(query:str)-str:搜索互联网信息输入查询关键词返回搜索结果摘要。# 模拟搜索fake_db{langgraph:LangGraph 是 LangChain 推出的图结构工作流框架支持分支、循环和状态管理。,react:ReAct 是一种让 LLM 交替进行推理Reasoning和行动Acting的 Agent 范式。,北京天气:北京今天晴15°C ~ 25°C北风 3-4 级。,}forkey,valueinfake_db.items():ifkeyinquery:returnvaluereturnf未找到关于 {query} 的信息。tooldefcalculator(expression:str)-str:执行数学计算输入一个数学表达式字符串返回计算结果。try:# 安全地计算表达式resulteval(expression,{__builtins__:{}},{})returnf计算结果{expression}{result}exceptExceptionase:returnf计算失败{e}tooldefget_current_time()-str:获取当前系统时间。fromdatetimeimportdatetimereturnf当前时间{datetime.now().strftime(%Y-%m-%d %H:%M:%S)}# 工具列表tools[search_info,calculator,get_current_time]⚠️ 生产环境不要用eval()这里仅作示例。实际应使用更安全的表达式解析器。四、第二步绑定工具到 LLM这是最容易被误解的一步我会讲得很细。llm_with_toolsllm.bind_tools(tools)bind_tools做了什么它不是让 LLM 去调用工具。它只是把工具的名称、描述、参数 schema 注入到 LLM 的 system prompt 中让 LLM知道有哪些工具可用以及每个工具需要什么参数。当 LLM 判断需要调用工具时它不会真的去执行工具而是输出一个tool_calls列表形如{content:,tool_calls:[{name:search_info,args:{query:北京天气}}]}你可以把tool_calls理解为 LLM 说“我需要调用search_info参数是query北京天气但我不会自己去调我只是告诉你我需要这个东西。”五、第三步定义 Statefromlanggraph.graphimportAnnotation AgentStateAnnotation.Root({messages:Annotation(reducerlambdaprev,next:prevnext,# 追加不覆盖defaultlambda:[],),})这里只定义了一个字段messages但 Reducer 是追加模式。这意味着每个节点返回的新消息都会被拼接到历史消息列表后面而不是相互覆盖。为什么只有messages因为 ReAct Agent 的所有上下文都在对话历史里——用户问题、LLM 推理、工具调用、工具结果全部是消息。不需要额外的tool_result、final_answer等字段。六、第四步定义两个节点6.1 agent_node——LLM 推理节点fromlangchain_core.messagesimportSystemMessage SYSTEM_PROMPT你是一个有用的 AI 助手。 当需要搜索实时信息时使用 search_info 工具。 当需要数学计算时使用 calculator 工具。 当用户询问时间时使用 get_current_time 工具。 如果不需要工具就能回答问题直接回答。asyncdefagent_node(state:AgentState)-dict:LLM 推理节点分析当前消息决定是直接回答还是调用工具。messages[SystemMessage(contentSYSTEM_PROMPT)]state[messages]responseawaitllm_with_tools.ainvoke(messages)return{messages:[response]}关键点输入是state[messages]所有历史消息调用llm_with_tools.ainvoke()LLM 返回的response可能带tool_calls也可能不带返回{messages: [response]}中间件会自动追加到历史6.2 tool_node——工具执行节点fromlanggraph.prebuiltimportToolNode tool_nodeToolNode(tools)仅此一行。ToolNode是 LangGraph 的预置组件它会自动从state[messages]的最后一条消息中提取tool_calls逐个执行对应的工具函数将每个工具的执行结果包装成ToolMessage返回{messages: [tool_messages...]}七、第五步路由函数 搭图7.1 路由函数defshould_continue(state:AgentState)-str:判断 agent_node 输出中是否包含 tool_calls。last_messagestate[messages][-1]# 如果最后一条消息有 tool_calls说明 LLM 想调工具ifhasattr(last_message,tool_calls)andlast_message.tool_calls:returntools# 否则 LLM 已经给出了最终回答结束returnend7.2 搭图fromlanggraph.graphimportStateGraph,START,END builderStateGraph(AgentState)# 添加两个节点builder.add_node(agent,agent_node)builder.add_node(tools,tool_node)# START → agentbuilder.add_edge(START,agent)# agent → 条件判断有 tool_calls 去 tools否则 ENDbuilder.add_conditional_edges(agent,should_continue,{tools:tools,end:END,})# tools → agent工具执行完结果送回 agent 继续推理builder.add_edge(tools,agent)# 编译graphbuilder.compile()图结构回顾[START] → [agent] ──no tool_calls──→ [END] ↑ │ │ has tool_calls │ ↓ └─── [tools]这条builder.add_edge(tools, agent)是整个 Agent 循环的关键——工具执行结果送回 agentagent 再判断还需不需要更多工具形成了推理→行动→推理的循环。八、运行 Agentfromlangchain_core.messagesimportHumanMessageasyncdefmain():# 初始化状态只有一条用户消息initial_state{messages:[HumanMessage(content北京今天天气怎么样)]}resultawaitgraph.ainvoke(initial_state)# 最后一条消息就是最终回答final_answerresult[messages][-1]print(fAgent 回答{final_answer.content})# 打印完整对话过程print(\n 完整对话过程 )fori,msginenumerate(result[messages]):roletype(msg).__name__ contentmsg.contentifhasattr(msg,content)elsestr(msg)ifhasattr(msg,tool_calls)andmsg.tool_calls:tools_info[tc[name]fortcinmsg.tool_calls]print(f[{i}]{role}: 决定调用{tools_info})else:previewcontent[:80]...iflen(str(content))80elsecontentprint(f[{i}]{role}:{preview})importasyncio asyncio.run(main())运行输出示例Agent 回答北京今天晴15°C ~ 25°C北风 3-4 级适合出行。 完整对话过程 [0] HumanMessage: 北京今天天气怎么样 [1] AIMessage: 决定调用 [search_info] [2] ToolMessage: 北京今天晴15°C ~ 25°C北风 3-4 级。 [3] AIMessage: 北京今天晴15°C ~ 25°C北风 3-4 级适合出行。九、重头戏llm.bind_tools 和 ToolNode 到底什么关系这是 LangGraph 新手最常见的困惑我专门辟一节讲清楚。9.1 一句话区分概念作用类比llm.bind_tools(tools)让 LLM知道有哪些工具告诉 LLM 每个工具的用途和参数工具说明书agent_node调用 bind_tools 后的 LLM让 LLM决策是否需要工具需要哪个工具拿着说明书的决策者ToolNode(tools)真正执行工具调用拿到实际结果动手干活的执行者9.2 一图胜千言┌─────────────────────────────────────────────────────┐ │ agent_node │ │ ┌───────────────────────────────────────────────┐ │ │ │ llm_with_tools.invoke(messages) │ │ │ │ │ │ │ │ LLM 思考用户问天气我应该调 search_info │ │ │ │ │ │ │ │ 输出AIMessage( │ │ │ │ content, │ │ │ │ tool_calls[{name: search_info, ...}] │ │ │ │ ) │ │ │ │ │ │ │ │ ↑ 只是 说 要调工具并没有真的调 │ │ │ └───────────────────────────────────────────────┘ │ │ │ │ ↓ tool_calls 不是空的 │ │ │ │ ToolNode │ │ ┌───────────────────────────────────────────────┐ │ │ │ for tool_call in tool_calls: │ │ │ │ result search_info(query北京天气) │ │ │ │ │ │ │ │ 输出ToolMessage( │ │ │ │ content北京今天晴15°C ~ 25°C, │ │ │ │ tool_call_id... │ │ │ │ ) │ │ │ │ │ │ │ │ ↑ 真正执行了工具拿到了结果 │ │ │ └───────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────┘9.3 常见错误只用 bind_tools 不用 ToolNode# ❌ 这样写的 Agent 不会真的调用工具asyncdefbad_agent_node(state):responseawaitllm.bind_tools(tools).ainvoke(state[messages])return{messages:[response]}# 没有 ToolNode 执行 tool_calls工具永远不会被调用错误表现LLM 输出里一直有tool_calls但tool_calls里的工具从没被执行过Agent 不会拿到任何工具返回结果。9.4 记忆口诀bind_tools 管知道ToolNode 管做到。十、加上流式输出Agent 执行可能比较慢LLM 推理 工具调用用 stream 可以实时看到进度asyncdefstream_agent():initial_state{messages:[HumanMessage(content3.14 × 15 的平方是多少)]}asyncforeventingraph.astream(initial_state):fornode_name,node_outputinevent.items():print(f\n--- [{node_name}] ---)ifmessagesinnode_output:formsginnode_output[messages]:roletype(msg).__name__ifhasattr(msg,tool_calls)andmsg.tool_calls:fortcinmsg.tool_calls:print(f 调用工具{tc[name]}({tc[args]}))elifmsg.content:previewmsg.content[:100]print(f {role}:{preview})asyncio.run(stream_agent())输出--- [agent] --- 调用工具calculator({expression: 3.14 * 15 ** 2}) --- [tools] --- ToolMessage: 计算结果3.14 * 15 ** 2 706.5 --- [agent] --- AIMessage: 3.14 × 15² 706.5计算过程是 3.14 × 225 706.5。十一、给 Agent 加上最大迭代次数保护目前的 Agent 没有循环上限保护。虽然 LLM 最终总会停止调用工具但加个保险总没错MAX_ITERATIONS10AgentStateWithLimitAnnotation.Root({messages:Annotation(reducerlambdaprev,next:prevnext,defaultlambda:[],),iteration_count:Annotation(reducerlambdaprev,next:prevnext,defaultlambda:0,),})asyncdefagent_node_with_limit(state)-dict:messages[SystemMessage(contentSYSTEM_PROMPT)]state[messages]responseawaitllm_with_tools.ainvoke(messages)return{messages:[response],iteration_count:1,# 每次 agent 推理计数 1}defshould_continue_with_limit(state)-str:ifstate[iteration_count]MAX_ITERATIONS:returnend# 强制结束last_messagestate[messages][-1]ifhasattr(last_message,tool_calls)andlast_message.tool_calls:returntoolsreturnend# tool_node 也返回 iteration_count不增加计数deftool_node_with_tracking(state)-dict:resulttool_node.invoke(state)return{**result,iteration_count:0}十二、完整代码整合把上面零散的代码合成一个完整可运行的文件importasynciofromlanggraph.graphimportStateGraph,START,END,Annotationfromlanggraph.prebuiltimportToolNodefromlangchain_openaiimportChatOpenAIfromlangchain_core.toolsimporttoolfromlangchain_core.messagesimportHumanMessage,SystemMessage# 1. 工具定义 tooldefsearch_info(query:str)-str:搜索互联网信息。fake_db{langgraph:LangGraph 是图结构工作流框架。,北京天气:北京今天晴15°C ~ 25°C。}fork,vinfake_db.items():ifkinquery:returnvreturnf未找到关于 {query} 的信息。tooldefcalculator(expression:str)-str:执行数学计算。try:resulteval(expression,{__builtins__:{}},{})returnf计算结果{expression}{result}exceptExceptionase:returnf计算失败{e}tools[search_info,calculator]# 2. 绑定工具 llmChatOpenAI(modelgpt-4o,temperature0)llm_with_toolsllm.bind_tools(tools)# 3. State AgentStateAnnotation.Root({messages:Annotation(reducerlambdap,n:pn,defaultlambda:[]),})# 4. 节点 SYSTEM_PROMPT你是有用的助手。搜索用 search_info计算用 calculator。asyncdefagent_node(state:AgentState)-dict:messages[SystemMessage(contentSYSTEM_PROMPT)]state[messages]responseawaitllm_with_tools.ainvoke(messages)return{messages:[response]}tool_nodeToolNode(tools)# 5. 路由 defshould_continue(state:AgentState)-str:laststate[messages][-1]ifhasattr(last,tool_calls)andlast.tool_calls:returntoolsreturnend# 6. 搭图 builderStateGraph(AgentState)builder.add_node(agent,agent_node)builder.add_node(tools,tool_node)builder.add_edge(START,agent)builder.add_conditional_edges(agent,should_continue,{tools:tools,end:END})builder.add_edge(tools,agent)graphbuilder.compile()# 7. 运行 asyncdefmain():resultawaitgraph.ainvoke({messages:[HumanMessage(content北京天气怎么样35*2等于几)]})print(最终回答,result[messages][-1].content)asyncio.run(main())十三、本篇小结ReAct Agent LLM 推理 工具调用 循环LangGraph 的图结构天然适合表达这个模式。llm.bind_tools让 LLM 知道有哪些工具输出 tool_callsToolNode真正执行工具返回 ToolMessage。Agent 图只有两个节点agent_node推理tool_node执行靠条件边循环。永远给循环加上限保护防止 LLM 死循环。开发用 stream 看过程生产用 invoke 拿结果。下一篇预告Agent 能跑了但离生产环境还有距离——状态怎么持久化避免崩溃丢失节点失败怎么自动重试多个 Agent 怎么协同工作第 4 篇将讲透 LangGraph 的生产级特性Checkpointer、RetryPolicy、Subgraph、CachePolicy。