1. 项目概述从LangChain到LangGraph的思维跃迁如果你昨天跟着一起搭建了第一个基于LangChain的简单AI Agent可能会觉得嗯这玩意儿思路挺清晰但总感觉哪里有点“笨”。任务来了调用工具返回结果结束。整个过程像一条笔直的流水线缺乏“思考”和“回溯”的灵活性。这正是我们今天要解决的核心问题也是AI Agent从“脚本执行器”迈向“自主思考者”的关键一步——引入ReActReasoning Acting模式并用LangGraph这个更强大的工具来实现它。LangGraph不是LangChain的替代品你可以把它理解为LangChain生态系统中的一个“超级增强模块”。如果说LangChain提供的是构建链条Chain的积木那么LangGraph提供的就是设计复杂工作流Workflow和状态机State Machine的蓝图。它核心引入了“循环”和“有状态”的概念这让Agent能够根据中间结果决定下一步做什么甚至回到上一步重新思考这正是实现ReAct模式所必需的。我们今天的任务就是亲手用LangGraph搭建一个具备ReAct推理能力的AI Agent让它能像人一样“边想边做”。这个项目适合所有对AI应用开发感兴趣的开发者无论你是想为自己的产品增加智能助手还是希望深入理解大模型如何与外部工具协同工作。通过今天的内容你将掌握LangGraph的核心概念——StateGraph、Nodes、Edges并理解如何用代码实现“思考-行动-观察”的循环。你会发现一个能自主使用搜索引擎查资料、并判断信息是否足够的Agent其代码结构可以如此优雅和强大。2. 核心架构解析LangGraph如何为Agent注入“思考”能力在直接动手写代码之前我们必须先吃透LangGraph的设计哲学和ReAct模式的理论基础。这能让你在后续调试和扩展时清楚地知道每一行代码的意图而不是盲目复制粘贴。2.1 ReAct模式让LLM学会“三思而后行”ReAct模式的精髓在于将大语言模型LLM的推理Reasoning能力与行动Acting能力在同一个循环中结合起来。传统的简单链式调用是“行动 - 结果”而ReAct是“思考 - 行动 - 观察 - 再思考...”。一个典型ReAct循环的伪代码逻辑如下推理ReasonLLM根据当前任务和已知信息包括历史记录分析现状规划下一步应该做什么。例如“用户想知道爱因斯坦的生日。我目前不知道这个信息。我应该使用搜索工具来查找。”行动Act根据推理步骤的决策执行具体的动作。通常是调用一个预定义的工具Tool比如调用搜索引擎API查询“阿尔伯特·爱因斯坦 出生日期”。观察Observe获取行动执行后的结果。例如搜索引擎返回“阿尔伯特·爱因斯坦出生于1879年3月14日。”循环判断将观察到的结果纳入上下文再次触发推理步骤判断任务是否完成。例如“我已经获得了爱因斯坦的生日是1879年3月14日。用户的问题已得到解答任务完成。”这个循环会一直持续直到LLM在推理步骤中认为任务已经完成并生成最终答案。这种方式极大地提升了Agent处理复杂、多步骤任务的能力和可靠性。2.2 LangGraph核心三要素图、状态与节点LangGraph将上述循环抽象为一个有向图Graph模型其中包含三个核心概念State状态这是一个贯穿整个工作流的共享数据容器。通常是一个Pydantic模型或简单的字典定义了Agent运行过程中需要跟踪和更新的所有信息。对于ReAct Agent状态至少应该包含input: 用户的原始问题。messages: 对话历史或中间思考记录通常是一个消息列表。agent_outcome: 最近一次LLM推理的输出包含下一步指令。tool_results: 最近一次或历次工具调用的结果。Node节点图中的一个功能单元。每个节点是一个函数它接收当前的State作为输入执行某些操作如调用LLM、调用工具然后返回一个更新后的State字典。这个字典中更新的部分会被合并到全局状态中。在我们的ReAct Agent中主要会有两种节点agent_node: 负责“推理”的节点在这里调用LLM。tool_node: 负责“行动”的节点在这里调用具体的工具函数。Edge边定义了节点之间的流转条件。边决定了在一个节点执行完毕后下一个该执行哪个节点。LangGraph中的边可以是固定的set_entry_point,add_edge也可以是条件式的add_conditional_edges。ReAct模式的核心就在于agent_node和tool_node之间通过条件边形成的循环。2.3 与纯LangChain实现的关键差异你可能在LangChain中也见过AgentExecutor和ReAct的示例。它们的主要区别在于抽象层次和控制粒度LangChain AgentExecutor更像一个黑盒它内部封装了ReAct循环。你定义好工具和LLM它来管理循环。优点是快速上手缺点是对循环内部的精细控制比较困难比如你想在特定条件下保存中间状态到数据库。LangGraph将整个工作流白盒化、可视化。你需要显式地定义状态、节点和边亲手搭建这个循环。这带来了巨大的灵活性轻松实现复杂分支除了ReAct主循环你可以很容易地加入错误处理节点、验证节点、人工审核节点等。状态管理更清晰所有中间数据都在State对象里便于持久化、监控和调试。支持长期运行其设计天然支持“暂停-恢复”适合处理耗时很长的任务。理解了这些我们就知道用LangGraph写ReAct Agent本质上是在用代码“画”出一张明确的工作流蓝图。3. 环境准备与基础组件定义接下来我们进入实战环节。请确保你的Python环境建议3.10以上已经准备好。3.1 安装依赖库我们将使用LangChain和LangGraph的最新稳定版并搭配OpenAI的模型你也可以替换为其他兼容的模型如Ollama本地模型。pip install langchain langgraph openai如果你需要用到网络搜索工具我们以Tavily搜索引擎为例它提供了免费的API额度非常适合学习和原型开发pip install tavily-python然后去 Tavily官网 注册一个账号获取免费的API Key。3.2 定义Agent的共享状态State这是构建LangGraph应用的第一步也是最重要的一步。我们需要仔细设计State里要放什么。from typing import TypedDict, List, Annotated, Union from langchain_core.messages import BaseMessage import operator # 使用TypedDict来定义状态的结构这样类型提示更清晰 class AgentState(TypedDict): # 用户的原始输入 input: str # 完整的对话/推理历史。我们将LLM的思考、工具结果都作为消息存入。 # 使用Annotated和operator.add是LangGraph的约定表示这个字段在节点间传递时是追加append的。 messages: Annotated[List[BaseMessage], operator.add] # 可选一个独立的字段专门存放最近一次工具调用的结果便于节点访问。 # 我们这里选择将所有信息都通过messages传递更符合LangGraph常见模式。关键点解释Annotated[List[BaseMessage], operator.add]这是LangGraph的“语法糖”。它告诉框架messages这个字段在各个节点之间传递时不是替换而是将当前节点产生的消息追加到已有的消息列表末尾。这完美契合了对话或思考历史需要不断累积的场景。为什么把所有东西都放进messages因为LLM特别是ChatModel的API通常接收一个消息列表作为上下文。我们把思考过程、工具结果都格式化成HumanMessage、AIMessage、ToolMessage等能让LLM更好地理解整个任务进程。3.3 初始化核心“大脑”LLM与工具我们需要一个强大的LLM作为Agent的推理引擎以及一些工具作为它的“手脚”。from langchain_openai import ChatOpenAI from langchain_community.tools.tavily_search import TavilySearchResults from langchain.tools import Tool import os # 1. 初始化LLM。使用gpt-3.5-turbo性价比很高对于ReAct任务足够。 # 记得设置你的OpenAI API Key可以通过环境变量OPENAI_API_KEY设置。 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 2. 初始化工具。这里我们创建一个搜索工具。 # 设置环境变量 TAVILY_API_KEY os.environ[TAVILY_API_KEY] your_tavily_api_key_here search_tool TavilySearchResults(max_results3) # 限制每次搜索返回3条结果 # 3. 将工具包装成LangChain可识别的格式并绑定到LLM。 # 为了让LLM知道它能用什么工具我们需要将工具描述“灌输”给它。 tools [search_tool] llm_with_tools llm.bind_tools(tools) # 我们也可以创建一些简单的自定义工具比如一个计算器 def multiplier(a: float, b: float) - float: Multiply two numbers. return a * b custom_tool Tool( nameMultiplier, funcmultiplier, descriptionUseful for multiplying two numbers together., ) # 将自定义工具也加入列表 tools.append(custom_tool) llm_with_tools llm.bind_tools(tools) # 重新绑定让LLM知道所有工具注意bind_tools()是一个关键操作。它不会立即调用工具而是修改了LLM的调用方式使其输出内容中包含符合特定格式的“工具调用”请求。LangChain的后端会解析这个请求再去实际执行对应的工具函数。4. 构建LangGraph工作流定义节点与边现在我们开始用LangGraph的API将State、LLM和工具组装成一个可循环的工作流。4.1 构建Agent推理节点agent_node这个节点的职责是接收当前状态包含历史消息调用LLM进行推理决定下一步是回答问题还是使用工具。from langgraph.graph import StateGraph, END from langchain_core.messages import AIMessage, ToolMessage # 定义agent节点函数 def agent_node(state: AgentState): 调用LLM根据当前对话历史决定下一步行动。 输入AgentState 输出一个字典更新messages字段追加LLM的响应消息 print(f\n[Agent Node] 正在思考... 历史消息数{len(state[messages])}) # 调用绑定了工具的LLM。我们将当前所有消息作为上下文传入。 response llm_with_tools.invoke(state[messages]) # LLM的响应可能是一个普通的AIMessage也可能是一个包含工具调用请求的AIMessage。 # 无论是哪种我们都将其追加到历史消息中。 new_messages [response] # 返回更新后的状态部分。LangGraph会自动将其合并到全局State中。 return {messages: new_messages}这里有个至关重要的细节LLM的输出response是一个AIMessage对象。如果LLM决定要调用工具这个AIMessage会有一个.tool_calls属性里面包含了它想调用的工具名称和参数。LangChain的bind_tools机制已经帮我们处理好了这种结构化输出。4.2 构建工具执行节点tools_node这个节点的职责是执行Agent节点所请求的工具调用并将结果格式化后返回。def tools_node(state: AgentState): 执行Agent在上一步中请求的所有工具调用。 输入AgentState (其中最新的消息是包含tool_calls的AIMessage) 输出更新messages字段追加工具执行结果的ToolMessage。 print(f\n[Tools Node] 正在执行工具...) messages state[messages] last_message messages[-1] # 获取最新的消息即Agent的决策 tool_messages [] if hasattr(last_message, tool_calls) and last_message.tool_calls: # 遍历LLM请求调用的每一个工具 for tool_call in last_message.tool_calls: tool_name tool_call[name] tool_args tool_call[args] print(f 调用工具: {tool_name}, 参数: {tool_args}) # 根据工具名找到我们之前定义的tool对象 tool_to_use next((tool for tool in tools if tool.name tool_name), None) if tool_to_use: try: # 执行工具 result tool_to_use.invoke(tool_args) # 将结果封装成ToolMessage并关联到对应的tool_call_id tool_messages.append(ToolMessage(contentstr(result), tool_call_idtool_call[id])) except Exception as e: # 工具执行出错返回错误信息 error_msg f调用工具{tool_name}时出错: {str(e)} print(f ! 错误: {error_msg}) tool_messages.append(ToolMessage(contenterror_msg, tool_call_idtool_call[id])) else: error_msg f未知工具: {tool_name} print(f ! {error_msg}) tool_messages.append(ToolMessage(contenterror_msg, tool_call_idtool_call[id])) else: # 如果最新的消息没有请求工具调用理论上不应该进入这个节点。 # 这里我们返回一个空消息列表或者一个提示。 print( ! 警告最新消息未包含工具调用请求。) # 将工具执行结果的消息返回它们会被追加到历史中。 return {messages: tool_messages}4.3 组装工作流图创建循环逻辑这是LangGraph最精彩的部分我们将节点连接起来并设定流转规则。# 1. 创建一个StateGraph并指定我们定义的State类型 workflow StateGraph(AgentState) # 2. 添加节点 workflow.add_node(agent, agent_node) # 推理节点 workflow.add_node(tools, tools_node) # 工具执行节点 # 3. 设置入口点工作流从agent节点开始先进行推理 workflow.set_entry_point(agent) # 4. 添加固定边从tools节点执行完毕后无论结果如何都应该回到agent节点进行下一轮思考。 workflow.add_edge(tools, agent) # 5. 添加条件边这是实现ReAct循环的关键 # 从agent节点出来后我们需要判断LLM是直接给出了最终答案还是请求调用工具 # 这个判断逻辑我们写成一个路由函数。 def route_after_agent(state: AgentState) - str: 根据agent节点的输出决定下一步是去执行工具还是结束工作流。 返回下一个节点的名称。 messages state[messages] last_message messages[-1] # 判断最新消息是否包含工具调用请求 if hasattr(last_message, tool_calls) and last_message.tool_calls: # 有工具调用下一步去tools节点 print(f[Router] 检测到工具调用请求路由至 tools 节点。) return tools else: # 没有工具调用说明LLM认为任务完成给出了最终答案。工作流结束。 print(f[Router] Agent已给出最终答案工作流结束。) return END # 将条件边添加到图中从agent节点出来根据route_after_agent函数的返回值决定去向。 workflow.add_conditional_edges( agent, # 源节点 route_after_agent, # 路由判断函数 { tools: tools, # 如果函数返回tools则跳转到tools节点 END: END # 如果函数返回END则结束工作流 } ) # 6. 编译图生成可执行的应用 app workflow.compile()代码逻辑梳理从agent入口开始。agent节点调用LLM思考。根据LLM输出有无tool_calls路由函数route_after_agent决定下一步有工具调用 - 前往tools节点。无工具调用直接回答- 前往END流程结束。如果到了tools节点执行工具后通过固定边add_edge(tools, agent)无条件地回到agent节点进行下一轮“观察-思考”。如此循环直至agent节点输出最终答案。这个app对象就是我们构建好的、具备ReAct能力的AI Agent大脑。你可以把它保存下来或者集成到Web服务中。5. 运行与调试观察Agent的思考过程现在让我们用几个问题来测试这个Agent并观察它的内部思考循环。# 定义一个辅助函数来漂亮地打印对话历史 def print_messages(messages): for msg in messages: if msg.type human: print(f\n[用户]: {msg.content}) elif msg.type ai: # 检查是否是工具调用 if hasattr(msg, tool_calls) and msg.tool_calls: print(f\n[AI-思考]: 决定使用工具。) for tc in msg.tool_calls: print(f - 调用 {tc[name]}参数: {tc[args]}) else: print(f\n[AI-回答]: {msg.content}) elif msg.type tool: print(f\n[工具-结果]: {msg.content[:200]}...) # 只打印前200字符避免过长 # 测试1一个需要搜索的问题 print(*50) print(测试1需要搜索的问题) print(*50) inputs {input: 爱因斯坦的生日是哪一天他现在多少岁了, messages: []} config {configurable: {thread_id: test1}} # 使用app.stream可以逐步查看每个节点的输出非常适合调试。 for event in app.stream(inputs, configconfig, stream_modevalues): event[messages][-1].pretty_print() # 打印每一步的最新消息 print(\n *50) print(完整对话历史回顾) print(*50) final_state app.invoke(inputs, configconfig) print_messages(final_state[messages])运行上述代码你会在控制台看到类似以下的输出具体内容因模型和搜索返回结果而异 测试1需要搜索的问题 [Agent Node] 正在思考... 历史消息数1 [Router] 检测到工具调用请求路由至 tools 节点。 [Tools Node] 正在执行工具... 调用工具: tavily_search_results_json, 参数: {query: 爱因斯坦 生日} [Agent Node] 正在思考... 历史消息数3 [Router] 检测到工具调用请求路由至 tools 节点。 [Tools Node] 正在执行工具... 调用工具: tavily_search_results_json, 参数: {query: 爱因斯坦 现在多少岁 如果活着} [Agent Node] 正在思考... 历史消息数5 [Router] Agent已给出最终答案工作流结束。 完整对话历史回顾 [用户]: 爱因斯坦的生日是哪一天他现在多少岁了 [AI-思考]: 决定使用工具。 - 调用 tavily_search_results_json参数: {query: 爱因斯坦 生日} [工具-结果]: [{title: 阿尔伯特·爱因斯坦 - 维基百科自由的百科全书, url: https://zh.wikipedia.org/wiki/%E7%88%B1%E5%9B%A0%E6%96%AF%E5%9D%A6, content: 阿尔伯特·爱因斯坦德语Albert Einstein1879年3月14日—1955年4月18日犹太裔理论物理学家...}, ...] [AI-思考]: 决定使用工具。 - 调用 tavily_search_results_json参数: {query: 爱因斯坦 现在多少岁 如果活着} [工具-结果]: [{title: 爱因斯坦如果还活着今年多大了 - 百度知道, url: https://zhidao.baidu.com/question/..., content: 爱因斯坦出生于1879年如果他还活着到2024年应该是145岁。}, ...] [AI-回答]: 阿尔伯特·爱因斯坦出生于1879年3月14日。他于1955年4月18日逝世享年76岁。因此他现在已经不在世了。如果他还活着到2024年将会是145岁。太棒了Agent成功地执行了ReAct循环它先思考需要搜索“爱因斯坦 生日”得到结果后结合新信息知道他已经去世再次思考判断还需要知道“如果活着多少岁”来回答问题的后半部分再次搜索后最终综合所有信息给出了完整且准确的回答。6. 高级技巧与实战避坑指南构建一个能跑的ReAct Agent只是第一步。要让它在实际项目中稳定、可靠、高效还需要注意以下关键点。6.1 控制循环与避免无限循环最大的风险是Agent陷入“思考-调用-再思考”的死循环。我们必须设置安全阀。方案一在状态State中增加步数计数器。class AgentState(TypedDict): input: str messages: Annotated[List[BaseMessage], operator.add] # 新增步数计数器 step_count: int # 在agent_node或路由函数中检查 def route_after_agent(state: AgentState) - str: messages state[messages] last_message messages[-1] # 检查步数是否超限例如10步 if state.get(step_count, 0) 10: print([Router] 步数超限强制结束。) return END if hasattr(last_message, tool_calls) and last_message.tool_calls: # 在前往tools节点前更新步数 # 注意更新状态需要在节点返回的字典里操作路由函数只做判断。 # 更好的做法是在agent_node返回时增加step_count。 return tools else: return END # 修改agent_node使其返回时增加步数 def agent_node(state: AgentState): # ... 原有逻辑 ... new_step_count state.get(step_count, 0) 1 return {messages: [response], step_count: new_step_count}方案二利用LangGraph的interrupt机制更优雅。LangGraph支持在特定条件下中断流程将控制权交还给外部系统例如等待用户确认。这更适合复杂的人机协作场景。对于简单的步数限制方案一更直接。6.2 优化提示工程Prompt EngineeringLLM在ReAct循环中的表现极大程度上依赖于你给它的“指令”System Message。默认的绑定工具提示可能不够强。最佳实践是在初始消息中注入强引导from langchain_core.messages import SystemMessage, HumanMessage def get_initial_messages(user_input: str): 构造包含强系统指令的初始消息列表 system_prompt 你是一个有帮助的AI助手可以使用工具。请严格遵循以下流程 1. 思考用户的问题判断是否需要使用工具获取信息。 2. 如果需要一次只调用一个最相关的工具。清晰说明调用理由。 3. 观察工具返回的结果。 4. 如果结果足以回答问题请直接给出友好、准确的最终答案。 5. 如果结果不足或引发新问题继续思考并决定是否再次使用工具。 6. 如果你认为通过现有工具无法解决问题请诚实地告知用户。 请确保你的思考过程简洁。 return [ SystemMessage(contentsystem_prompt), HumanMessage(contentuser_input) ] # 在调用app时使用 inputs { input: 用户问题, messages: get_initial_messages(用户问题) # 使用强化后的初始消息 }6.3 工具选择与结果处理工具描述要精准工具函数的description参数至关重要。LLM完全依赖这个描述来决定是否以及如何调用它。描述应清晰说明功能、输入格式和适用场景。例如“用于搜索互联网最新信息回答实时性问题。”比“一个搜索工具”要好得多。处理复杂工具结果像搜索引擎返回的可能是JSON列表。直接扔给LLM可能信息过载。可以考虑在tools_node中增加一个结果预处理步骤提取最关键的一两段文本再封装成ToolMessage。工具错误处理我们的示例代码中包含了基础的try...except。在生产环境中你需要更细致的错误分类网络超时、API限额、无效输入等并返回结构化的错误信息帮助LLM理解发生了什么例如“搜索服务暂时不可用请稍后再试或换一种问法。”。6.4 可视化与调试LangGraph提供了一个非常强大的功能将你的工作流图可视化。# 将图导出为PNG图片 from IPython.display import Image, display try: display(Image(app.get_graph().draw_mermaid_png())) except: # 如果环境不支持可以打印Mermaid文本到Mermaid Live Editor查看 print(app.get_graph().draw_mermaid())这张图能让你一目了然地看清agent和tools节点之间的循环关系以及条件路由的逻辑对于理解复杂工作流和向团队解释设计非常有帮助。7. 常见问题排查与性能优化在实际运行中你可能会遇到以下典型问题问题1Agent总是忽略工具直接胡编乱造答案。排查首先检查llm.bind_tools(tools)是否成功执行。检查工具的描述是否清晰。使用app.get_graph().draw_mermaid()查看图结构确认条件边是否正确连接。解决强化系统提示词见6.2节。在agent_node中打印出response.tool_calls确认LLM是否输出了正确的工具调用结构。确保你使用的模型如gpt-3.5-turbo支持工具调用功能Function Calling。问题2Agent陷入无限循环反复调用同一个工具。排查观察每次agent_node的输入消息。工具执行结果ToolMessage是否被正确追加到历史中LLM是否每次都在“观察”到新结果后做出了不同的思考解决实现步数限制器见6.1节。检查工具返回的结果是否具有确定性。有时搜索工具每次返回结果顺序不同可能导致LLM认为信息不足而重复搜索。可以在tools_node中对结果进行排序或去重。在系统提示词中强调“避免重复相同操作”。问题3工作流运行速度慢。分析延迟主要来自两部分LLM API调用和工具API调用如网络搜索。优化LLM层面使用流式响应stream虽然对调试友好但可能增加整体耗时。对于纯后端任务使用invoke。考虑使用更快的模型如gpt-3.5-turbovsgpt-4或设置合理的超时时间。工具层面为网络请求工具设置超时如requests库的timeout参数。考虑使用异步调用如果LangGraph支持你所用工具的异步版本。对于复杂工作流可以分析是否某些工具节点可以并行执行LangGraph支持分支与并行。缓存对内容变化不频繁的查询如“爱因斯坦生日”可以在工具层或LLM调用层引入缓存避免重复计算。问题4如何保存和恢复Agent的会话状态方案LangGraph的State本身是可序列化的字典。你可以在工作流执行到任何节点后将state字典用json.dumps()保存到数据库或文件。恢复时重新构造一个包含相同messages历史等数据的state然后调用app.invoke(state)即可从断点继续执行。config参数中的thread_id就是用来区分不同会话线程的。通过第二天的学习你已经从LangChain的基础链式思维升级到了用LangGraph设计和实现具备自主推理-行动循环的ReAct Agent。这不仅仅是换了一个库更是思维模式的转变——从“线性流程”到“图状工作流”。掌握这个范式你就有能力去构建那些能够处理复杂决策、支持多轮工具交互、甚至需要人工介入审核的智能体应用了。接下来你可以尝试为你的Agent添加更多工具如数据库查询、代码执行、文件读写或者设计更复杂的条件分支比如在答案置信度低时自动转入人工审核节点。 LangGraph实战：构建具备ReAct推理能力的AI Agent工作流
1. 项目概述从LangChain到LangGraph的思维跃迁如果你昨天跟着一起搭建了第一个基于LangChain的简单AI Agent可能会觉得嗯这玩意儿思路挺清晰但总感觉哪里有点“笨”。任务来了调用工具返回结果结束。整个过程像一条笔直的流水线缺乏“思考”和“回溯”的灵活性。这正是我们今天要解决的核心问题也是AI Agent从“脚本执行器”迈向“自主思考者”的关键一步——引入ReActReasoning Acting模式并用LangGraph这个更强大的工具来实现它。LangGraph不是LangChain的替代品你可以把它理解为LangChain生态系统中的一个“超级增强模块”。如果说LangChain提供的是构建链条Chain的积木那么LangGraph提供的就是设计复杂工作流Workflow和状态机State Machine的蓝图。它核心引入了“循环”和“有状态”的概念这让Agent能够根据中间结果决定下一步做什么甚至回到上一步重新思考这正是实现ReAct模式所必需的。我们今天的任务就是亲手用LangGraph搭建一个具备ReAct推理能力的AI Agent让它能像人一样“边想边做”。这个项目适合所有对AI应用开发感兴趣的开发者无论你是想为自己的产品增加智能助手还是希望深入理解大模型如何与外部工具协同工作。通过今天的内容你将掌握LangGraph的核心概念——StateGraph、Nodes、Edges并理解如何用代码实现“思考-行动-观察”的循环。你会发现一个能自主使用搜索引擎查资料、并判断信息是否足够的Agent其代码结构可以如此优雅和强大。2. 核心架构解析LangGraph如何为Agent注入“思考”能力在直接动手写代码之前我们必须先吃透LangGraph的设计哲学和ReAct模式的理论基础。这能让你在后续调试和扩展时清楚地知道每一行代码的意图而不是盲目复制粘贴。2.1 ReAct模式让LLM学会“三思而后行”ReAct模式的精髓在于将大语言模型LLM的推理Reasoning能力与行动Acting能力在同一个循环中结合起来。传统的简单链式调用是“行动 - 结果”而ReAct是“思考 - 行动 - 观察 - 再思考...”。一个典型ReAct循环的伪代码逻辑如下推理ReasonLLM根据当前任务和已知信息包括历史记录分析现状规划下一步应该做什么。例如“用户想知道爱因斯坦的生日。我目前不知道这个信息。我应该使用搜索工具来查找。”行动Act根据推理步骤的决策执行具体的动作。通常是调用一个预定义的工具Tool比如调用搜索引擎API查询“阿尔伯特·爱因斯坦 出生日期”。观察Observe获取行动执行后的结果。例如搜索引擎返回“阿尔伯特·爱因斯坦出生于1879年3月14日。”循环判断将观察到的结果纳入上下文再次触发推理步骤判断任务是否完成。例如“我已经获得了爱因斯坦的生日是1879年3月14日。用户的问题已得到解答任务完成。”这个循环会一直持续直到LLM在推理步骤中认为任务已经完成并生成最终答案。这种方式极大地提升了Agent处理复杂、多步骤任务的能力和可靠性。2.2 LangGraph核心三要素图、状态与节点LangGraph将上述循环抽象为一个有向图Graph模型其中包含三个核心概念State状态这是一个贯穿整个工作流的共享数据容器。通常是一个Pydantic模型或简单的字典定义了Agent运行过程中需要跟踪和更新的所有信息。对于ReAct Agent状态至少应该包含input: 用户的原始问题。messages: 对话历史或中间思考记录通常是一个消息列表。agent_outcome: 最近一次LLM推理的输出包含下一步指令。tool_results: 最近一次或历次工具调用的结果。Node节点图中的一个功能单元。每个节点是一个函数它接收当前的State作为输入执行某些操作如调用LLM、调用工具然后返回一个更新后的State字典。这个字典中更新的部分会被合并到全局状态中。在我们的ReAct Agent中主要会有两种节点agent_node: 负责“推理”的节点在这里调用LLM。tool_node: 负责“行动”的节点在这里调用具体的工具函数。Edge边定义了节点之间的流转条件。边决定了在一个节点执行完毕后下一个该执行哪个节点。LangGraph中的边可以是固定的set_entry_point,add_edge也可以是条件式的add_conditional_edges。ReAct模式的核心就在于agent_node和tool_node之间通过条件边形成的循环。2.3 与纯LangChain实现的关键差异你可能在LangChain中也见过AgentExecutor和ReAct的示例。它们的主要区别在于抽象层次和控制粒度LangChain AgentExecutor更像一个黑盒它内部封装了ReAct循环。你定义好工具和LLM它来管理循环。优点是快速上手缺点是对循环内部的精细控制比较困难比如你想在特定条件下保存中间状态到数据库。LangGraph将整个工作流白盒化、可视化。你需要显式地定义状态、节点和边亲手搭建这个循环。这带来了巨大的灵活性轻松实现复杂分支除了ReAct主循环你可以很容易地加入错误处理节点、验证节点、人工审核节点等。状态管理更清晰所有中间数据都在State对象里便于持久化、监控和调试。支持长期运行其设计天然支持“暂停-恢复”适合处理耗时很长的任务。理解了这些我们就知道用LangGraph写ReAct Agent本质上是在用代码“画”出一张明确的工作流蓝图。3. 环境准备与基础组件定义接下来我们进入实战环节。请确保你的Python环境建议3.10以上已经准备好。3.1 安装依赖库我们将使用LangChain和LangGraph的最新稳定版并搭配OpenAI的模型你也可以替换为其他兼容的模型如Ollama本地模型。pip install langchain langgraph openai如果你需要用到网络搜索工具我们以Tavily搜索引擎为例它提供了免费的API额度非常适合学习和原型开发pip install tavily-python然后去 Tavily官网 注册一个账号获取免费的API Key。3.2 定义Agent的共享状态State这是构建LangGraph应用的第一步也是最重要的一步。我们需要仔细设计State里要放什么。from typing import TypedDict, List, Annotated, Union from langchain_core.messages import BaseMessage import operator # 使用TypedDict来定义状态的结构这样类型提示更清晰 class AgentState(TypedDict): # 用户的原始输入 input: str # 完整的对话/推理历史。我们将LLM的思考、工具结果都作为消息存入。 # 使用Annotated和operator.add是LangGraph的约定表示这个字段在节点间传递时是追加append的。 messages: Annotated[List[BaseMessage], operator.add] # 可选一个独立的字段专门存放最近一次工具调用的结果便于节点访问。 # 我们这里选择将所有信息都通过messages传递更符合LangGraph常见模式。关键点解释Annotated[List[BaseMessage], operator.add]这是LangGraph的“语法糖”。它告诉框架messages这个字段在各个节点之间传递时不是替换而是将当前节点产生的消息追加到已有的消息列表末尾。这完美契合了对话或思考历史需要不断累积的场景。为什么把所有东西都放进messages因为LLM特别是ChatModel的API通常接收一个消息列表作为上下文。我们把思考过程、工具结果都格式化成HumanMessage、AIMessage、ToolMessage等能让LLM更好地理解整个任务进程。3.3 初始化核心“大脑”LLM与工具我们需要一个强大的LLM作为Agent的推理引擎以及一些工具作为它的“手脚”。from langchain_openai import ChatOpenAI from langchain_community.tools.tavily_search import TavilySearchResults from langchain.tools import Tool import os # 1. 初始化LLM。使用gpt-3.5-turbo性价比很高对于ReAct任务足够。 # 记得设置你的OpenAI API Key可以通过环境变量OPENAI_API_KEY设置。 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 2. 初始化工具。这里我们创建一个搜索工具。 # 设置环境变量 TAVILY_API_KEY os.environ[TAVILY_API_KEY] your_tavily_api_key_here search_tool TavilySearchResults(max_results3) # 限制每次搜索返回3条结果 # 3. 将工具包装成LangChain可识别的格式并绑定到LLM。 # 为了让LLM知道它能用什么工具我们需要将工具描述“灌输”给它。 tools [search_tool] llm_with_tools llm.bind_tools(tools) # 我们也可以创建一些简单的自定义工具比如一个计算器 def multiplier(a: float, b: float) - float: Multiply two numbers. return a * b custom_tool Tool( nameMultiplier, funcmultiplier, descriptionUseful for multiplying two numbers together., ) # 将自定义工具也加入列表 tools.append(custom_tool) llm_with_tools llm.bind_tools(tools) # 重新绑定让LLM知道所有工具注意bind_tools()是一个关键操作。它不会立即调用工具而是修改了LLM的调用方式使其输出内容中包含符合特定格式的“工具调用”请求。LangChain的后端会解析这个请求再去实际执行对应的工具函数。4. 构建LangGraph工作流定义节点与边现在我们开始用LangGraph的API将State、LLM和工具组装成一个可循环的工作流。4.1 构建Agent推理节点agent_node这个节点的职责是接收当前状态包含历史消息调用LLM进行推理决定下一步是回答问题还是使用工具。from langgraph.graph import StateGraph, END from langchain_core.messages import AIMessage, ToolMessage # 定义agent节点函数 def agent_node(state: AgentState): 调用LLM根据当前对话历史决定下一步行动。 输入AgentState 输出一个字典更新messages字段追加LLM的响应消息 print(f\n[Agent Node] 正在思考... 历史消息数{len(state[messages])}) # 调用绑定了工具的LLM。我们将当前所有消息作为上下文传入。 response llm_with_tools.invoke(state[messages]) # LLM的响应可能是一个普通的AIMessage也可能是一个包含工具调用请求的AIMessage。 # 无论是哪种我们都将其追加到历史消息中。 new_messages [response] # 返回更新后的状态部分。LangGraph会自动将其合并到全局State中。 return {messages: new_messages}这里有个至关重要的细节LLM的输出response是一个AIMessage对象。如果LLM决定要调用工具这个AIMessage会有一个.tool_calls属性里面包含了它想调用的工具名称和参数。LangChain的bind_tools机制已经帮我们处理好了这种结构化输出。4.2 构建工具执行节点tools_node这个节点的职责是执行Agent节点所请求的工具调用并将结果格式化后返回。def tools_node(state: AgentState): 执行Agent在上一步中请求的所有工具调用。 输入AgentState (其中最新的消息是包含tool_calls的AIMessage) 输出更新messages字段追加工具执行结果的ToolMessage。 print(f\n[Tools Node] 正在执行工具...) messages state[messages] last_message messages[-1] # 获取最新的消息即Agent的决策 tool_messages [] if hasattr(last_message, tool_calls) and last_message.tool_calls: # 遍历LLM请求调用的每一个工具 for tool_call in last_message.tool_calls: tool_name tool_call[name] tool_args tool_call[args] print(f 调用工具: {tool_name}, 参数: {tool_args}) # 根据工具名找到我们之前定义的tool对象 tool_to_use next((tool for tool in tools if tool.name tool_name), None) if tool_to_use: try: # 执行工具 result tool_to_use.invoke(tool_args) # 将结果封装成ToolMessage并关联到对应的tool_call_id tool_messages.append(ToolMessage(contentstr(result), tool_call_idtool_call[id])) except Exception as e: # 工具执行出错返回错误信息 error_msg f调用工具{tool_name}时出错: {str(e)} print(f ! 错误: {error_msg}) tool_messages.append(ToolMessage(contenterror_msg, tool_call_idtool_call[id])) else: error_msg f未知工具: {tool_name} print(f ! {error_msg}) tool_messages.append(ToolMessage(contenterror_msg, tool_call_idtool_call[id])) else: # 如果最新的消息没有请求工具调用理论上不应该进入这个节点。 # 这里我们返回一个空消息列表或者一个提示。 print( ! 警告最新消息未包含工具调用请求。) # 将工具执行结果的消息返回它们会被追加到历史中。 return {messages: tool_messages}4.3 组装工作流图创建循环逻辑这是LangGraph最精彩的部分我们将节点连接起来并设定流转规则。# 1. 创建一个StateGraph并指定我们定义的State类型 workflow StateGraph(AgentState) # 2. 添加节点 workflow.add_node(agent, agent_node) # 推理节点 workflow.add_node(tools, tools_node) # 工具执行节点 # 3. 设置入口点工作流从agent节点开始先进行推理 workflow.set_entry_point(agent) # 4. 添加固定边从tools节点执行完毕后无论结果如何都应该回到agent节点进行下一轮思考。 workflow.add_edge(tools, agent) # 5. 添加条件边这是实现ReAct循环的关键 # 从agent节点出来后我们需要判断LLM是直接给出了最终答案还是请求调用工具 # 这个判断逻辑我们写成一个路由函数。 def route_after_agent(state: AgentState) - str: 根据agent节点的输出决定下一步是去执行工具还是结束工作流。 返回下一个节点的名称。 messages state[messages] last_message messages[-1] # 判断最新消息是否包含工具调用请求 if hasattr(last_message, tool_calls) and last_message.tool_calls: # 有工具调用下一步去tools节点 print(f[Router] 检测到工具调用请求路由至 tools 节点。) return tools else: # 没有工具调用说明LLM认为任务完成给出了最终答案。工作流结束。 print(f[Router] Agent已给出最终答案工作流结束。) return END # 将条件边添加到图中从agent节点出来根据route_after_agent函数的返回值决定去向。 workflow.add_conditional_edges( agent, # 源节点 route_after_agent, # 路由判断函数 { tools: tools, # 如果函数返回tools则跳转到tools节点 END: END # 如果函数返回END则结束工作流 } ) # 6. 编译图生成可执行的应用 app workflow.compile()代码逻辑梳理从agent入口开始。agent节点调用LLM思考。根据LLM输出有无tool_calls路由函数route_after_agent决定下一步有工具调用 - 前往tools节点。无工具调用直接回答- 前往END流程结束。如果到了tools节点执行工具后通过固定边add_edge(tools, agent)无条件地回到agent节点进行下一轮“观察-思考”。如此循环直至agent节点输出最终答案。这个app对象就是我们构建好的、具备ReAct能力的AI Agent大脑。你可以把它保存下来或者集成到Web服务中。5. 运行与调试观察Agent的思考过程现在让我们用几个问题来测试这个Agent并观察它的内部思考循环。# 定义一个辅助函数来漂亮地打印对话历史 def print_messages(messages): for msg in messages: if msg.type human: print(f\n[用户]: {msg.content}) elif msg.type ai: # 检查是否是工具调用 if hasattr(msg, tool_calls) and msg.tool_calls: print(f\n[AI-思考]: 决定使用工具。) for tc in msg.tool_calls: print(f - 调用 {tc[name]}参数: {tc[args]}) else: print(f\n[AI-回答]: {msg.content}) elif msg.type tool: print(f\n[工具-结果]: {msg.content[:200]}...) # 只打印前200字符避免过长 # 测试1一个需要搜索的问题 print(*50) print(测试1需要搜索的问题) print(*50) inputs {input: 爱因斯坦的生日是哪一天他现在多少岁了, messages: []} config {configurable: {thread_id: test1}} # 使用app.stream可以逐步查看每个节点的输出非常适合调试。 for event in app.stream(inputs, configconfig, stream_modevalues): event[messages][-1].pretty_print() # 打印每一步的最新消息 print(\n *50) print(完整对话历史回顾) print(*50) final_state app.invoke(inputs, configconfig) print_messages(final_state[messages])运行上述代码你会在控制台看到类似以下的输出具体内容因模型和搜索返回结果而异 测试1需要搜索的问题 [Agent Node] 正在思考... 历史消息数1 [Router] 检测到工具调用请求路由至 tools 节点。 [Tools Node] 正在执行工具... 调用工具: tavily_search_results_json, 参数: {query: 爱因斯坦 生日} [Agent Node] 正在思考... 历史消息数3 [Router] 检测到工具调用请求路由至 tools 节点。 [Tools Node] 正在执行工具... 调用工具: tavily_search_results_json, 参数: {query: 爱因斯坦 现在多少岁 如果活着} [Agent Node] 正在思考... 历史消息数5 [Router] Agent已给出最终答案工作流结束。 完整对话历史回顾 [用户]: 爱因斯坦的生日是哪一天他现在多少岁了 [AI-思考]: 决定使用工具。 - 调用 tavily_search_results_json参数: {query: 爱因斯坦 生日} [工具-结果]: [{title: 阿尔伯特·爱因斯坦 - 维基百科自由的百科全书, url: https://zh.wikipedia.org/wiki/%E7%88%B1%E5%9B%A0%E6%96%AF%E5%9D%A6, content: 阿尔伯特·爱因斯坦德语Albert Einstein1879年3月14日—1955年4月18日犹太裔理论物理学家...}, ...] [AI-思考]: 决定使用工具。 - 调用 tavily_search_results_json参数: {query: 爱因斯坦 现在多少岁 如果活着} [工具-结果]: [{title: 爱因斯坦如果还活着今年多大了 - 百度知道, url: https://zhidao.baidu.com/question/..., content: 爱因斯坦出生于1879年如果他还活着到2024年应该是145岁。}, ...] [AI-回答]: 阿尔伯特·爱因斯坦出生于1879年3月14日。他于1955年4月18日逝世享年76岁。因此他现在已经不在世了。如果他还活着到2024年将会是145岁。太棒了Agent成功地执行了ReAct循环它先思考需要搜索“爱因斯坦 生日”得到结果后结合新信息知道他已经去世再次思考判断还需要知道“如果活着多少岁”来回答问题的后半部分再次搜索后最终综合所有信息给出了完整且准确的回答。6. 高级技巧与实战避坑指南构建一个能跑的ReAct Agent只是第一步。要让它在实际项目中稳定、可靠、高效还需要注意以下关键点。6.1 控制循环与避免无限循环最大的风险是Agent陷入“思考-调用-再思考”的死循环。我们必须设置安全阀。方案一在状态State中增加步数计数器。class AgentState(TypedDict): input: str messages: Annotated[List[BaseMessage], operator.add] # 新增步数计数器 step_count: int # 在agent_node或路由函数中检查 def route_after_agent(state: AgentState) - str: messages state[messages] last_message messages[-1] # 检查步数是否超限例如10步 if state.get(step_count, 0) 10: print([Router] 步数超限强制结束。) return END if hasattr(last_message, tool_calls) and last_message.tool_calls: # 在前往tools节点前更新步数 # 注意更新状态需要在节点返回的字典里操作路由函数只做判断。 # 更好的做法是在agent_node返回时增加step_count。 return tools else: return END # 修改agent_node使其返回时增加步数 def agent_node(state: AgentState): # ... 原有逻辑 ... new_step_count state.get(step_count, 0) 1 return {messages: [response], step_count: new_step_count}方案二利用LangGraph的interrupt机制更优雅。LangGraph支持在特定条件下中断流程将控制权交还给外部系统例如等待用户确认。这更适合复杂的人机协作场景。对于简单的步数限制方案一更直接。6.2 优化提示工程Prompt EngineeringLLM在ReAct循环中的表现极大程度上依赖于你给它的“指令”System Message。默认的绑定工具提示可能不够强。最佳实践是在初始消息中注入强引导from langchain_core.messages import SystemMessage, HumanMessage def get_initial_messages(user_input: str): 构造包含强系统指令的初始消息列表 system_prompt 你是一个有帮助的AI助手可以使用工具。请严格遵循以下流程 1. 思考用户的问题判断是否需要使用工具获取信息。 2. 如果需要一次只调用一个最相关的工具。清晰说明调用理由。 3. 观察工具返回的结果。 4. 如果结果足以回答问题请直接给出友好、准确的最终答案。 5. 如果结果不足或引发新问题继续思考并决定是否再次使用工具。 6. 如果你认为通过现有工具无法解决问题请诚实地告知用户。 请确保你的思考过程简洁。 return [ SystemMessage(contentsystem_prompt), HumanMessage(contentuser_input) ] # 在调用app时使用 inputs { input: 用户问题, messages: get_initial_messages(用户问题) # 使用强化后的初始消息 }6.3 工具选择与结果处理工具描述要精准工具函数的description参数至关重要。LLM完全依赖这个描述来决定是否以及如何调用它。描述应清晰说明功能、输入格式和适用场景。例如“用于搜索互联网最新信息回答实时性问题。”比“一个搜索工具”要好得多。处理复杂工具结果像搜索引擎返回的可能是JSON列表。直接扔给LLM可能信息过载。可以考虑在tools_node中增加一个结果预处理步骤提取最关键的一两段文本再封装成ToolMessage。工具错误处理我们的示例代码中包含了基础的try...except。在生产环境中你需要更细致的错误分类网络超时、API限额、无效输入等并返回结构化的错误信息帮助LLM理解发生了什么例如“搜索服务暂时不可用请稍后再试或换一种问法。”。6.4 可视化与调试LangGraph提供了一个非常强大的功能将你的工作流图可视化。# 将图导出为PNG图片 from IPython.display import Image, display try: display(Image(app.get_graph().draw_mermaid_png())) except: # 如果环境不支持可以打印Mermaid文本到Mermaid Live Editor查看 print(app.get_graph().draw_mermaid())这张图能让你一目了然地看清agent和tools节点之间的循环关系以及条件路由的逻辑对于理解复杂工作流和向团队解释设计非常有帮助。7. 常见问题排查与性能优化在实际运行中你可能会遇到以下典型问题问题1Agent总是忽略工具直接胡编乱造答案。排查首先检查llm.bind_tools(tools)是否成功执行。检查工具的描述是否清晰。使用app.get_graph().draw_mermaid()查看图结构确认条件边是否正确连接。解决强化系统提示词见6.2节。在agent_node中打印出response.tool_calls确认LLM是否输出了正确的工具调用结构。确保你使用的模型如gpt-3.5-turbo支持工具调用功能Function Calling。问题2Agent陷入无限循环反复调用同一个工具。排查观察每次agent_node的输入消息。工具执行结果ToolMessage是否被正确追加到历史中LLM是否每次都在“观察”到新结果后做出了不同的思考解决实现步数限制器见6.1节。检查工具返回的结果是否具有确定性。有时搜索工具每次返回结果顺序不同可能导致LLM认为信息不足而重复搜索。可以在tools_node中对结果进行排序或去重。在系统提示词中强调“避免重复相同操作”。问题3工作流运行速度慢。分析延迟主要来自两部分LLM API调用和工具API调用如网络搜索。优化LLM层面使用流式响应stream虽然对调试友好但可能增加整体耗时。对于纯后端任务使用invoke。考虑使用更快的模型如gpt-3.5-turbovsgpt-4或设置合理的超时时间。工具层面为网络请求工具设置超时如requests库的timeout参数。考虑使用异步调用如果LangGraph支持你所用工具的异步版本。对于复杂工作流可以分析是否某些工具节点可以并行执行LangGraph支持分支与并行。缓存对内容变化不频繁的查询如“爱因斯坦生日”可以在工具层或LLM调用层引入缓存避免重复计算。问题4如何保存和恢复Agent的会话状态方案LangGraph的State本身是可序列化的字典。你可以在工作流执行到任何节点后将state字典用json.dumps()保存到数据库或文件。恢复时重新构造一个包含相同messages历史等数据的state然后调用app.invoke(state)即可从断点继续执行。config参数中的thread_id就是用来区分不同会话线程的。通过第二天的学习你已经从LangChain的基础链式思维升级到了用LangGraph设计和实现具备自主推理-行动循环的ReAct Agent。这不仅仅是换了一个库更是思维模式的转变——从“线性流程”到“图状工作流”。掌握这个范式你就有能力去构建那些能够处理复杂决策、支持多轮工具交互、甚至需要人工介入审核的智能体应用了。接下来你可以尝试为你的Agent添加更多工具如数据库查询、代码执行、文件读写或者设计更复杂的条件分支比如在答案置信度低时自动转入人工审核节点。