在实际 AI 应用开发中构建一个能够自主规划、执行任务并管理状态的智能体Agent是迈向复杂 AI 系统的关键一步。LangGraph 作为 LangChain 生态中用于构建有状态、多步骤应用的工作流编排框架正逐渐成为实现这类智能体的核心工具。它通过将执行步骤建模为图中的节点并用边来控制流程使得构建具备循环、分支和持久化能力的智能体变得直观。对于希望从简单的大模型调用进阶到开发具备“思考-行动-观察”循环的智能应用的开发者而言理解 LangGraph 的架构、核心组件并通过代码实战掌握其精髓是当前一项极具价值的能力。本文将带你从零开始深入剖析 LangGraph 智能体的构建过程涵盖架构设计、核心组件详解、一个可运行的代码实战案例以及开发中常见的陷阱与解决方案。1. 理解 LangGraph从状态机视角看智能体工作流在深入代码之前我们需要建立一个正确的认知模型。LangGraph 的核心思想是将一个复杂的、多步骤的 AI 应用流程抽象为一个有向图。这个图由节点Nodes和边Edges组成节点代表一个具体的执行单元如调用大模型、执行工具、更新状态边则定义了节点之间的流转条件。1.1 为什么需要 LangGraph超越简单的链式调用传统的 LangChain Chain 或简单的函数调用适用于线性流程例如“问答 - 检索 - 回答”。但当流程中需要根据上一步的结果决定下一步走向、或者需要循环执行某个步骤直到满足条件时例如一个需要反复使用搜索工具来收集信息的智能体简单的链式结构就显得力不从心。LangGraph 引入了**状态State**的概念。整个图共享一个状态字典每个节点都可以读取和修改这个状态。边的条件判断Conditional Edge基于当前状态的值来决定下一个要执行的节点从而实现了分支和循环。这使得构建像 ReActReasoning and Acting、AutoGPT 这类具备自主规划能力的智能体成为可能。1.2 核心概念状态、节点与边状态State一个类似字典Dict的对象贯穿整个图的执行过程。它定义了智能体工作流中需要跟踪的所有信息例如用户的输入、大模型的回复、工具执行的结果、历史对话等。在 LangGraph 中我们通常使用 TypedDict 来明确定义状态的模式。节点Node一个可调用的函数它接收当前状态作为输入执行某些操作如调用 LLM、运行工具并返回一个包含对状态更新内容的字典。节点是工作流中的原子操作单元。边Edge连接节点的路径。分为两种普通边Edge无条件地从一个节点指向下一个节点。条件边Conditional Edge根据状态中的某个值通常是上一个节点输出的结果来决定下一步走向哪个节点。这实现了图的分支逻辑。通过组合这些元素你可以构建出从简单的线性链到复杂的、具备循环和自省能力的智能体工作流。2. 环境准备与项目初始化在开始构建智能体之前我们需要搭建一个基础的 Python 开发环境并安装必要的依赖。本教程将以一个相对稳定的环境为例。2.1 环境与依赖配置建议使用 Python 3.9 或更高版本。首先创建一个新的虚拟环境然后安装核心包。# 创建并激活虚拟环境以 conda 为例 conda create -n langgraph-agent python3.10 conda activate langgraph-agent # 安装 LangGraph 及其相关依赖 pip install langgraph langchain langchain-openai关键依赖说明langgraph: 核心工作流编排框架。langchain: 提供了与大模型、工具、记忆等交互的高级抽象。langchain-openai: 用于方便地集成 OpenAI 系列的模型如 GPT-3.5/4。如果你使用其他模型如 Anthropic Claude 国内大模型API需要安装对应的 LangChain 集成包。注意模型 API 调用会产生费用。对于学习和测试可以使用 OpenAI 的免费额度或性价比高的模型。确保你已设置好相应的 API 密钥环境变量。2.2 设置 API 密钥为了调用大模型你需要配置 API 密钥。强烈建议通过环境变量管理避免将密钥硬编码在代码中。# 在 Linux/macOS 的终端中 export OPENAI_API_KEYyour-openai-api-key-here # 在 Windows PowerShell 中 $env:OPENAI_API_KEYyour-openai-api-key-here在你的 Python 脚本中可以通过os.getenv来读取。import os from langchain_openai import ChatOpenAI api_key os.getenv(“OPENAI_API_KEY”) if not api_key: raise ValueError(“请设置 OPENAI_API_KEY 环境变量。”) llm ChatOpenAI(model“gpt-3.5-turbo”, api_keyapi_key)3. 构建你的第一个 LangGraph 智能体ReAct 助手我们将构建一个经典的 ReActReason Act智能体。这个智能体能够理解问题决定是否需要使用工具如网络搜索、计算器来获取信息然后根据工具结果进行推理并给出最终答案。3.1 定义智能体的状态首先我们需要明确智能体在工作过程中需要记住什么。我们使用TypedDict来定义状态的结构。from typing import TypedDict, List, Annotated import operator from langchain_core.messages import BaseMessage, HumanMessage from langgraph.graph.message import add_messages # 定义状态结构 class AgentState(TypedDict): # 消息历史记录用户、助手、工具之间的对话 messages: Annotated[List[BaseMessage], add_messages] # 用户提出的原始问题 user_input: str # 智能体“思考”的步骤用于调试和观察 reasoning_steps: List[str]messages: 使用Annotated和add_messages修饰这是一个 LangGraph 的语法糖意味着当节点返回{“messages”: [new_message]}时这个新的消息列表会自动追加到现有的state[‘messages’]中而不是覆盖它。这对于维护对话历史至关重要。user_input: 存储初始问题。reasoning_steps: 记录智能体的内部推理过程方便我们理解其工作逻辑。3.2 创建工具并绑定给 LLM智能体需要“手”来执行具体操作。我们创建两个简单的工具一个模拟搜索一个模拟计算。from langchain.tools import tool from langchain.tools.render import render_text_description # 工具1模拟网络搜索 tool def search_web(query: str) - str: “”“模拟根据查询词返回搜索结果。在实际项目中这里应接入真实的搜索API如Serper、Tavily。”“” print(f“[搜索工具] 正在搜索: {query}”) # 模拟返回结果 simulated_results { “Python是什么”: “Python是一种高级、解释型的通用编程语言以其清晰的语法和代码可读性而闻名。”, “今天的天气”: “北京晴15-25摄氏度上海多云18-28摄氏度。”, “2的10次方是多少”: “2的10次方是1024。” } return simulated_results.get(query, f“未找到关于 ‘{query}’ 的明确信息。”) # 工具2模拟计算器 tool def calculate(expression: str) - str: “”“计算一个简单的数学表达式。警告此示例使用eval生产环境请使用安全库如ast.literal_eval。”“” print(f“[计算工具] 正在计算: {expression}”) try: # 注意在生产环境中直接使用eval是危险的此处仅用于演示。 result eval(expression) return str(result) except Exception as e: return f“计算错误: {e}” # 将工具打包成列表 tools [search_web, calculate] # 为LLM绑定工具使其知道有哪些工具可用 llm_with_tools llm.bind_tools(tools)关键点在于bind_tools方法它告诉大模型这些工具的存在、功能描述和调用格式。模型在思考时就能生成符合格式的工具调用请求。3.3 构建图节点思考、执行与判断一个典型的 ReAct 循环包含三个核心节点agent思考/规划、tools执行、should_continue判断是否继续。节点1智能体agent这个节点负责分析当前状态主要是对话历史决定下一步是直接回答还是调用某个工具。from langgraph.prebuilt import ToolExecutor from langchain_core.messages import ToolMessage # 工具执行器用于实际调用工具函数 tool_executor ToolExecutor(tools) def agent_node(state: AgentState): print(“\n--- [Agent节点] 开始思考 ---”) # 从状态中获取最新的消息历史 messages state[‘messages’] # 调用绑定了工具的LLM传入历史消息让其决定下一步 response llm_with_tools.invoke(messages) # 记录推理步骤可选用于调试 reasoning f“模型响应: {response}” if hasattr(response, ‘response_metadata’) and ‘tool_calls’ in response.response_metadata: reasoning f“, 决定调用工具: {response.response_metadata[‘tool_calls’]}” state[‘reasoning_steps’].append(reasoning) # 将模型的响应可能是普通消息也可能是工具调用请求添加到消息历史中 return {“messages”: [response]}节点2工具tools如果agent节点决定调用工具流程会走到这里。该节点解析工具调用请求执行对应的工具并将结果包装成ToolMessage返回。def tools_node(state: AgentState): print(“\n--- [Tools节点] 执行工具 ---”) messages state[‘messages’] # 取最后一条消息它应该是一个包含工具调用请求的 AIMessage last_message messages[-1] tool_calls last_message.tool_calls if not tool_calls: raise ValueError(f“最后一条消息没有工具调用: {last_message}”) tool_messages [] for tool_call in tool_calls: # 根据工具名找到对应的工具函数 tool_name tool_call[‘name’].lower() tool_to_use next((t for t in tools if t.name.lower() tool_name), None) if not tool_to_use: tool_messages.append(ToolMessage( contentf“错误工具 ‘{tool_name}’ 未找到。”, tool_call_idtool_call[‘id’] )) continue # 执行工具 result tool_executor.invoke(tool_to_use, tool_call[‘args’]) # 将工具执行结果封装成 ToolMessage tool_messages.append(ToolMessage( contentstr(result), tool_call_idtool_call[‘id’], nametool_name )) print(f“工具 ‘{tool_name}’ 执行结果: {result}”) # 将工具执行结果返回添加到消息历史 return {“messages”: tool_messages}节点3路由判断should_continue这是一个条件边函数。它检查上一步agent或tools的结果决定下一步是回到agent继续思考还是结束流程。def should_continue(state: AgentState) - str: messages state[‘messages’] last_message messages[-1] # 如果上一步是工具执行结果那么需要让agent再次思考 if isinstance(last_message, ToolMessage): return “continue” # 回到 agent 节点 # 否则上一步是agent的直接回答流程可以结束 else: return “end” # 结束图执行3.4 组装工作流图现在我们将节点和边组合起来形成完整的工作流。from langgraph.graph import StateGraph, END # 1. 创建一个图并指定状态的结构 workflow StateGraph(AgentState) # 2. 添加节点 workflow.add_node(“agent”, agent_node) workflow.add_node(“tools”, tools_node) # 3. 设置入口点 workflow.set_entry_point(“agent”) # 4. 添加边包括条件边 workflow.add_conditional_edges( “agent”, # 源节点 should_continue, # 条件判断函数 { “continue”: “tools”, # 如果返回 “continue”则前往 “tools” 节点 “end”: END # 如果返回 “end”则结束 } ) # 5. 从 tools 节点无条件返回 agent 节点进行下一轮思考 workflow.add_edge(“tools”, “agent”) # 6. 编译图生成可执行对象 app workflow.compile()至此一个具备 ReAct 能力的智能体工作流就构建完成了。图的结构是agent- (判断) - 如果需工具则到tools- 回到agent- (判断) - 直到无需工具结束。3.5 运行与验证智能体让我们用几个问题来测试这个智能体。# 测试1不需要工具的直接问答 def run_agent(question: str): print(f“\n 用户提问: {question} ”) # 初始化状态 initial_state: AgentState { “messages”: [HumanMessage(contentquestion)], “user_input”: question, “reasoning_steps”: [] } # 执行图 final_state app.invoke(initial_state) # 输出最终答案 final_messages final_state[‘messages’] for msg in final_messages: if msg.type ‘ai’ and not hasattr(msg, ‘tool_calls’): print(f“\n[智能体最终答案]: {msg.content}”) print(“\n[推理步骤回顾]:”, final_state[‘reasoning_steps’]) # 测试 if __name__ “__main__”: run_agent(“你好请介绍一下你自己。”) run_agent(“北京今天的天气怎么样”) run_agent(“计算一下 15 * 24 100 等于多少”)运行上述代码你将看到类似以下的输出清晰地展示了智能体的“思考-行动”过程 用户提问: 北京今天的天气怎么样 --- [Agent节点] 开始思考 --- [搜索工具] 正在搜索: 北京今天的天气 --- [Tools节点] 执行工具 --- 工具 ‘search_web’ 执行结果: 北京晴15-25摄氏度 --- [Agent节点] 开始思考 --- [智能体最终答案]根据搜索结果北京今天的天气是晴天气温在15到25摄氏度之间。4. 核心组件深度解析与高级用法掌握了基础构建后我们来深入理解 LangGraph 的一些高级特性和核心组件这对于构建生产级智能体至关重要。4.1 状态State的管理与设计状态是 LangGraph 的灵魂。设计良好的状态结构是构建复杂智能体的前提。状态合并策略我们之前用Annotated[List[BaseMessage], add_messages]来修饰messages字段这指定了该字段的合并策略是“追加”。LangGraph 支持其他策略如operator.add数值相加、覆盖等。你需要根据字段语义来选择合适的策略。复杂状态状态不仅可以包含对话历史还可以包含会话ID、用户偏好、长期记忆的索引、已执行步骤的计数器等。例如可以为智能体添加一个max_turns字段来限制最大对话轮次防止无限循环。class AdvancedAgentState(TypedDict): messages: Annotated[List[BaseMessage], add_messages] session_id: str # 会话标识 turn_count: Annotated[int, operator.add] # 轮次计数每次1 user_profile: dict # 用户画像 knowledge_base_ref: str # 外部知识库引用4.2 条件边Conditional Edge与流程控制add_conditional_edges是实现智能体自主决策的关键。条件函数 (should_continue) 的返回值决定了图的走向。多路分支你可以返回多个字符串映射到不同的下游节点实现复杂的分支逻辑。例如根据工具执行的成功与否决定是继续、重试还是转人工。基于内容的判断条件判断可以非常精细例如分析上一条消息的情感、提取实体、或检查工具返回结果是否为空。def advanced_router(state: AdvancedAgentState) - str: last_message state[‘messages’][-1] turn_count state[‘turn_count’] # 规则1超过最大轮次强制结束 if turn_count 10: return “force_end” # 规则2工具执行失败 if isinstance(last_message, ToolMessage) and “错误” in last_message.content: return “handle_error” # 规则3用户表达了感谢或告别 if isinstance(last_message, HumanMessage) and any(word in last_message.content for word in [“谢谢”, “再见”]): return “end” # 规则4模型决定调用工具 if hasattr(last_message, ‘tool_calls’) and last_message.tool_calls: return “continue” # 默认结束 return “end”4.3 持久化Persistence与长期记忆LangGraph 的一个强大特性是支持将图的状态持久化到数据库如 SQLite, PostgreSQL, Redis从而实现跨会话的长期记忆和智能体的暂停/恢复。from langgraph.checkpoint.sqlite import SqliteSaver # 创建一个 SQLite 检查点存储器 memory SqliteSaver.from_conn_string(“:memory:”) # 内存数据库重启后丢失。生产环境用文件路径。 # 在编译图时传入检查点存储器 app_with_memory workflow.compile(checkpointermemory) # 使用 config 配置运行包含线程IDthread_id来标识会话 config {“configurable”: {“thread_id”: “user-123-session-1”}} initial_state {“messages”: [HumanMessage(content“上次我们聊到哪了”)], …} # 第一次调用会创建检查点 result1 app_with_memory.invoke(initial_state, config) # 第二次调用可以从上次中断的状态继续如果图支持 result2 app_with_memory.invoke({“messages”: [HumanMessage(content“继续”)]}, config)这对于构建需要记住多轮对话上下文、或执行长时间运行任务的智能体如自动编写代码、研究报告非常有用。4.4 与 LangChain 生态集成LangGraph 与 LangChain 生态无缝集成。你可以轻松使用 LangChain 提供的数百种工具、文档加载器、向量数据库检索器以及各种链Chain。使用 LangChain 工具如前所示tool装饰器创建的工具可以直接使用。集成检索链Retrieval Chain你可以创建一个节点专门用于从向量数据库检索相关文档并将结果放入状态。from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings from langchain.chains import RetrievalQA # 假设已有一个加载了文档的向量库 vectorstore Chroma(persist_directory“./db”, embedding_functionOpenAIEmbeddings()) retriever vectorstore.as_retriever() qa_chain RetrievalQA.from_chain_type(llmllm, retrieverretriever) def retrieval_node(state: AgentState): question state[‘user_input’] # 使用检索链获取相关上下文 context qa_chain.run(question) # 注意生产环境应用 invoke 并处理异步 # 将检索到的上下文放入状态供后续节点使用 return {“retrieved_context”: context}5. 生产环境部署与常见问题排查将 LangGraph 智能体从本地脚本部署到生产环境需要考虑稳定性、监控、成本和安全。5.1 部署架构建议API 服务化使用 FastAPI 或 Flask 将编译好的app包装成 RESTful API 或 WebSocket 服务。异步支持LangGraph 支持异步执行 (ainvoke)。在生产 API 中应使用异步端点以提高并发能力。状态存储使用外部数据库如 PostgreSQL, Redis进行状态持久化而不是内存或 SQLite 文件。配置管理将模型 API 密钥、数据库连接串、工具参数等通过环境变量或配置中心管理。日志与监控为每个节点添加详细的日志记录并集成监控系统如 Prometheus来跟踪图的执行耗时、节点调用次数和错误率。5.2 常见问题与排查清单在开发 LangGraph 智能体时你可能会遇到以下典型问题问题现象可能原因检查与解决方案图编译或执行时报KeyError状态State的键在某个节点返回的字典中不存在或类型不匹配。1. 检查TypedDict定义是否与所有节点返回的字典键完全一致。2. 确保每个节点函数都返回一个字典且其键是状态的子集。智能体陷入无限循环条件边 (should_continue) 的逻辑有误导致在agent和tools节点间死循环。1. 在should_continue函数中添加日志打印判断依据。2. 在状态中添加turn_count字段并在条件函数中判断是否超过最大轮次强制跳出。工具调用未被识别或执行1. LLM 未正确绑定工具 (bind_tools)。2. 工具函数签名或描述不符合模型预期。3.tools_node中工具名匹配逻辑错误。1. 打印llm_with_tools.invoke的响应检查tool_calls字段是否存在。2. 确保工具函数的tool装饰器包含清晰的描述。3. 在tools_node中打印tool_call[‘name’]并与工具列表对比。错误api error: 400 the thinking_budget parameter must be a positive integer此错误通常与特定模型提供商如 Anthropic Claude的 API 参数有关与 LangGraph 本身无关。检查调用 LLM 时传入的参数。某些模型需要max_tokens或thinking_budget参数且必须为正整数。确保你使用的 LangChain 集成包版本与模型 API 兼容。错误transport failure for /api/agentpreset.list: http 403这看起来像是调用某个外部 AI 服务平台如 Dify的 API 时出现的权限错误。1. 确认 API 密钥是否正确且有权限。2. 确认请求的 URL 和端点是否正确。3. 检查网络环境确保能访问该服务。状态没有按预期更新状态的字段合并策略 (Annotated) 设置错误。例如本该追加的列表被覆盖了。回顾状态字段的定义。对于列表类需要累积的数据使用add_messages或自定义的追加函数。对于需要覆盖的配置项则不需要特殊修饰。性能瓶颈1. LLM 调用耗时过长。2. 工具执行如网络请求慢。3. 图结构复杂节点过多。1. 考虑使用更快的模型或设置合理的超时。2. 对工具调用实现缓存或异步执行。3. 优化图逻辑合并不必要的节点或引入“超时”和“短路”条件边。5.3 安全与成本优化最佳实践工具执行安全永远不要在生产环境中使用eval()。示例中的计算器工具应替换为安全的数学表达式解析库如ast.literal_eval用于简单表达式或numexpr。输入验证与清理对用户输入和工具返回的内容进行清洗和验证防止提示词注入或非预期输出导致图状态混乱。限制与防护轮次限制在状态和条件边中实现最大循环次数限制。超时控制为整个图的执行或单个 LLM 调用设置超时。费用监控在调用 LLM 和付费工具 API 的节点记录 Token 使用量并设置告警阈值。模块化与测试将复杂的图分解成子图Subgraph每个子图负责一个明确的功能。这有助于单独测试和维护。备降策略当主要工具或模型调用失败时应有备用的节点或流程来处理例如返回一个友好的错误信息或切换到一个更简单的回答模式。通过遵循这些架构原则和规避常见陷阱你可以构建出既强大又稳健的 LangGraph 智能体并将其成功应用于客服、数据分析、内容生成、自动化流程等实际场景中。从理解状态机模型开始逐步实践节点和边的构建最终结合生产级考量你便能驾驭这一强大的智能体编排框架。