这类工具最值得先看的不是功能列表而是能不能在普通环境里稳定跑起来以及它到底解决了智能体开发里的哪些具体痛点。LangGraph 不是一个孤立的库它解决的是当你用大模型 API 构建复杂、有状态的业务流程时如何把对话、工具调用、状态管理和条件分支这些零散环节串成一个清晰、可调试、可复用的“图”。很多人一上来就纠结 LangGraph 和 LangChain 的区别其实核心差异在于 LangGraph 把“状态”和“流程”作为一等公民特别适合需要多轮交互、有记忆、有决策循环的智能体场景。如果你正在做客服对话、工作流自动化、数据分析助手这类需要根据上下文决定下一步动作的应用LangGraph 提供的图结构能帮你把逻辑画清楚而不是把状态藏在各种回调函数里。我建议先从最小样例开始跑通一个最简单的“思考-行动-观察”循环再去看它的状态管理、持久化和复杂分支。下面按实际落地顺序拆一遍。1. 先理解 LangGraph 的核心用“图”来管理状态和流程很多人把 LangGraph 当成另一个 LangChain 来用这是第一个容易走偏的地方。LangGraph 的核心抽象是StateGraph它定义了一个共享的状态对象以及一系列能修改这个状态的节点Nodes和决定流转路径的边Edges。这听起来有点抽象我们拆开看。1.1 状态State是共享的“记忆白板”在 LangGraph 里你首先要定义一个状态State的类型。这个状态通常是一个字典Dict或者 Pydantic 模型它包含了流程中所有需要被记住和传递的信息。比如一个客服智能体的状态里可能有messages: 对话历史列表。user_query: 用户当前的问题。knowledge_base_results: 从数据库或知识库查到的信息。next_step: 下一步应该做什么如“回答”、“追问”、“转人工”。这个状态对象会在整个图的各个节点之间流动。每个节点都是一个函数它读取当前状态执行一些操作比如调用大模型、查询工具然后返回一个更新后的状态字典。这个“返回更新”的机制是它和简单函数调用最大的不同。from typing import Dict, TypedDict, List, Annotated from langgraph.graph import StateGraph, END # 1. 定义状态类型 class AgentState(TypedDict): messages: Annotated[List[str], 对话历史] user_input: str needs_clarification: bool final_answer: str # 2. 定义节点函数接收状态返回状态更新 def classify_intent(state: AgentState) - Dict: 节点判断用户意图 # 这里可以调用大模型 API 进行意图分类 # 假设我们简单判断 if 价格 in state[user_input]: state[needs_clarification] False state[next_action] query_price else: state[needs_clarification] True state[next_action] ask_for_clarification # 返回要更新的键值对 return {needs_clarification: state[needs_clarification], next_action: state[next_action]}关键点节点函数不直接修改传入的state对象除非你明确知道在做什么而是返回一个字典告诉框架要更新状态的哪些字段。这保证了状态变更的可预测性和可调试性。1.2 节点Nodes和边Edges组成可执行的工作流节点就是上面那样的函数。边则决定了流程的走向。LangGraph 提供了两种主要的边条件边Conditional Edges根据当前状态的某个值决定下一步去哪个节点。这实现了if-else分支。普通边直接连接两个节点表示无条件流转。你把节点和边添加到StateGraph实例中最后编译compile()成一个可执行的图。这个图对象有一个invoke(input_state)方法你传入初始状态它就会按照你定义的图逻辑执行下去直到到达终点END。# 3. 创建图并添加节点 graph StateGraph(AgentState) graph.add_node(classify_intent, classify_intent) graph.add_node(query_price, query_price_tool) # 假设已定义 graph.add_node(ask_clarify, ask_for_clarification) # 假设已定义 # 4. 设置入口和条件边 graph.set_entry_point(classify_intent) # 从哪个节点开始 # 条件边根据状态决定下一步 def route_by_action(state: AgentState): next_action state.get(next_action) if next_action query_price: return query_price elif next_action ask_for_clarification: return ask_clarify else: return END graph.add_conditional_edges( classify_intent, route_by_action, # 路由函数 { query_price: query_price, ask_for_clarification: ask_clarify, END: END } ) # 5. 添加普通边并编译 graph.add_edge(query_price, END) graph.add_edge(ask_clarify, END) app graph.compile()这样一个最简单的、带分支的智能体流程就定义好了。它的优势在于整个业务流程被可视化成了一张图而不再是散落在代码各处的if-else和函数调用。这对于调试和迭代复杂逻辑至关重要。2. 环境准备与最小可运行示例先让图跑起来在深入复杂功能前我建议先在本地创建一个干净的环境跑通一个“思考-行动”循环。这能帮你快速验证环境、理解数据流并建立信心。2.1 环境与依赖安装LangGraph 本身是一个 Python 库。它的运行不强制依赖特定的 AI 服务但通常你需要一个大模型 API如 OpenAI GPT、Anthropic Claude、国内大模型等来驱动智能体的“思考”节点。基础环境Python 3.8。建议使用虚拟环境venv或conda。包管理工具pip。核心依赖安装打开终端执行以下命令。这里安装的是 LangGraph 核心库和 OpenAI 的 SDK作为示例你可以替换成任何兼容的 SDK。# 创建并激活虚拟环境以 venv 为例 python -m venv langgraph-env source langgraph-env/bin/activate # Linux/macOS # langgraph-env\Scripts\activate # Windows # 安装核心库 pip install langgraph langchain-openai # 如果你需要更底层的控制也可以只安装 langgraph-core # pip install langgraph-core关于 API Key你需要准备一个 AI 服务的 API Key。以 OpenAI 为例将其设置为环境变量export OPENAI_API_KEY你的-api-key # Linux/macOS # set OPENAI_API_KEY你的-api-key # Windows重要提醒永远不要将 API Key 硬编码在代码中提交到版本控制系统如 Git。使用环境变量或安全的密钥管理服务。2.2 第一个 LangGraph 智能体复现经典的 ReAct 模式ReActReasoning Acting是智能体的经典模式模型先“思考”Reasoning下一步该做什么然后“行动”Acting去调用工具根据工具结果再进入下一轮思考。我们用 LangGraph 来实现它。步骤 1定义状态和工具from typing import TypedDict, List, Annotated from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain.tools import tool import operator # 定义状态。我们使用 LangGraph 推荐的“注解”方式来定义状态字段的合并策略。 class AgentState(TypedDict): messages: Annotated[List[str], operator.add] # 关键用 add 表示列表追加 current_step: str # 记录当前步骤如 think, act tool_result: str # 存储工具调用的结果 # 定义一个简单的工具计算器 tool def calculator(expression: str) - str: 计算一个数学表达式如 1 2 * 3。 try: # 警告实际生产环境请使用更安全的评估方法如 ast.literal_eval result eval(expression) return f计算结果: {result} except Exception as e: return f计算错误: {e} tools [calculator] llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 初始化大模型步骤 2创建“思考”和“行动”节点from langgraph.prebuilt import ToolExecutor, ToolInvocation from langchain_core.messages import HumanMessage, AIMessage, SystemMessage import json tool_executor ToolExecutor(tools) def think_node(state: AgentState) - Dict: 思考节点让模型决定下一步是回答还是调用工具。 # 构建给模型的提示 system_prompt SystemMessage(content你是一个助手可以回答问题和使用计算器工具。请根据对话历史决定下一步。如果需要计算就调用工具。) history state[messages] prompt_messages [system_prompt] history # 调用模型并告诉它有哪些工具可用 response llm.with_config({run_name: think}).invoke( prompt_messages, toolstools ) # 更新状态 new_messages state[messages] [response] next_step act if response.tool_calls else final_answer return { messages: new_messages, current_step: next_step, last_llm_response: response # 临时存储供行动节点使用 } def act_node(state: AgentState) - Dict: 行动节点执行模型选择的工具。 last_response state[last_llm_response] tool_calls last_response.tool_calls results [] for tc in tool_calls: # 执行工具调用 tool_result tool_executor.invoke(tc) results.append(tool_result) # 将工具执行结果转化为消息格式追加到历史 tool_result_messages [AIMessage(contentjson.dumps(r)) for r in results] new_messages state[messages] tool_result_messages return { messages: new_messages, tool_result: json.dumps(results), current_step: think # 执行完工具回到思考节点 }步骤 3构建图并运行# 创建图 graph StateGraph(AgentState) graph.add_node(think, think_node) graph.add_node(act, act_node) # 设置入口 graph.set_entry_point(think) # 定义条件边根据 current_step 决定下一步 def decide_next_step(state: AgentState): if state[current_step] act: return act elif state[current_step] final_answer: return END else: return think # 默认回到思考 graph.add_conditional_edges( think, decide_next_step, {act: act, END: END} ) graph.add_edge(act, think) # 行动完总是回到思考 # 编译应用 app graph.compile() # 运行智能体 initial_state { messages: [HumanMessage(content请问 15 的平方加上 20 等于多少)], current_step: think, tool_result: } final_state app.invoke(initial_state) print(最终回答:, final_state[messages][-1].content)运行这段代码你会看到智能体经历了“思考决定调用计算器- 行动执行计算- 再思考生成最终答案”的完整循环。这个例子虽然简单但它包含了 LangGraph 最核心的要素状态流转、条件分支和工具调用。3. 深入核心组件与高级模式超越简单循环跑通基础循环后你会遇到更实际的需求如何管理长对话记忆如何处理并行任务如何持久化状态实现断点续跑LangGraph 提供了一系列组件来应对这些场景。3.1 状态持久化与检查点Checkpointing对于需要长时间运行或服务重启后仍需保持状态的智能体如多轮客服会话状态持久化是必须的。LangGraph 的检查点机制可以将图执行过程中的任意状态快照保存下来之后可以从该点恢复执行。核心概念检查点存储器CheckpointSaver负责存储和加载状态快照。支持内存、文件系统、数据库如 Redis、PostgreSQL等多种后端。线程Thread一次完整的对话或任务执行过程。一个线程包含多个检查点。配置Config执行时的元数据如thread_id用于关联同一线程的检查点。示例使用内存存储实现多轮对话from langgraph.checkpoint.memory import MemorySaver # 创建内存检查点存储器 memory MemorySaver() # 在编译图时传入存储器 app graph.compile(checkpointermemory) # 第一次调用传入 config 指定 thread_id config {configurable: {thread_id: user_session_123}} initial_state {messages: [HumanMessage(content你好)], current_step: think} result1 app.invoke(initial_state, configconfig) print(第一轮结果:, result1[messages][-1].content) # 模拟一段时间后基于同一 thread_id 恢复对话 new_state_input {messages: [HumanMessage(content我们刚才说到哪了)]} result2 app.invoke(new_state_input, configconfig) # 会自动加载上次的检查点 print(第二轮结果带历史:, result2[messages][-1].content)在这个例子中第二次调用时app.invoke会先加载thread_id为”user_session_123“的最新检查点将新的用户消息追加到历史中然后从那里继续执行图。这对于构建 Web 服务或聊天机器人至关重要。3.2 子图Subgraphs与模块化复杂的智能体流程可以拆分成多个子图每个子图负责一个特定的子任务如“信息检索”、“数据验证”、“报告生成”。主图通过调用子图来组织工作流这极大地提高了代码的可维护性和复用性。创建子图子图本身也是一个StateGraph编译后可以作为节点加入主图。# 定义一个处理用户信息验证的子图 def validate_user_info(state: AgentState) - Dict: # 模拟验证逻辑 is_valid len(state.get(user_name, )) 0 return {is_user_valid: is_valid} validation_graph StateGraph(AgentState) validation_graph.add_node(validate, validate_user_info) validation_graph.set_entry_point(validate) validation_graph.add_edge(validate, END) validation_subgraph validation_graph.compile() # 在主图中将这个子图作为一个节点添加 main_graph StateGraph(AgentState) # 使用 add_node 并传入子图 main_graph.add_node(user_validation, validation_subgraph) # ... 添加其他节点和边通过子图你可以将验证逻辑、数据查询逻辑、格式化逻辑分别封装使主图结构清晰也便于单独测试每个子模块。3.3 并行与分支Parallelism有些任务可以同时进行比如同时查询天气和新闻。LangGraph 支持在图中定义并行执行的节点。使用add_node和条件边实现隐式并行这通常通过让一个节点产生多个后续节点并在状态中标记它们然后在下一轮迭代中同时处理来实现。更复杂的并行需要更精细的状态设计。使用langgraph.graph.MessageGraph处理消息流对于基于消息而非共享状态的并行处理MessageGraph是更合适的选择。它允许你将消息路由到多个节点并行处理然后聚合结果。from langgraph.graph import MessageGraph from langchain_core.messages import HumanMessage mgraph MessageGraph() def node_a(message): return fNode A processed: {message} def node_b(message): return fNode B processed: {message} mgraph.add_node(a, node_a) mgraph.add_node(b, node_b) # 设置入口并将输入同时发送给 a 和 b mgraph.set_entry_point(a) mgraph.set_entry_point(b) # 注意这需要根据具体版本和API调整这里展示概念 # 实际中你可能需要定义一个“广播”节点或者使用特定的边逻辑来实现并行。并行处理能显著提升处理吞吐量但也要注意资源竞争和结果合并的复杂性。4. 与外部生态集成API、MCP 与生产化部署一个智能体不可能孤立存在。它需要调用外部 API、访问数据库、使用专业工具。LangGraph 通过工具Tools和MCPModel Context Protocol等协议与外部世界连接。4.1 集成外部 API 与工具工具是 LangGraph 智能体与外界交互的标准方式。任何 Python 函数只要用tool装饰器包装或者遵循BaseTool接口都可以被智能体调用。创建自定义工具from langchain.tools import tool import requests tool def search_web(query: str) - str: 使用搜索引擎API搜索网络信息。 # 示例调用一个假设的搜索API # 注意实际使用时请替换为真实、合规的API并处理错误和限流 try: response requests.get( https://api.example.com/search, params{q: query, api_key: YOUR_API_KEY}, timeout10 ) response.raise_for_status() return response.text[:500] # 返回前500字符 except requests.RequestException as e: return f搜索请求失败: {e} # 将工具列表提供给智能体 tools [search_web, calculator]在“思考”节点中将tools列表传递给大模型模型就会学习在何时、如何使用这些工具。工具执行的结果会被写回状态供后续节点使用。处理 API 错误在实际生产中API 调用可能失败网络超时、鉴权失败、速率限制等。你需要在工具函数内部做好错误处理并返回结构化的错误信息以便智能体能够理解并采取补救措施如重试、降级处理。4.2 理解 MCPModel Context ProtocolMCP 是一个新兴的开放协议旨在标准化 AI 应用如智能体与各种数据源、工具和服务之间的连接方式。你可以把它想象成智能体世界的“USB 标准”。MCP 服务器MCP Server将一种数据源或工具如数据库、文件系统、CRM 系统的能力暴露成统一的“资源Resources”和“工具Tools”接口。MCP 客户端MCP Client智能体框架如 LangGraph通过 MCP 客户端连接到这些服务器从而获得访问这些资源和工具的能力。LangGraph 与 MCP虽然 LangGraph 核心不直接捆绑 MCP但其工具调用机制与 MCP 的理念高度兼容。你可以使用社区或官方提供的 MCP 服务器例如连接 PostgreSQL、Notion、Figma 的服务器。在 LangGraph 智能体中通过 MCP 客户端将这些服务器提供的工具注册为自己的工具列表。智能体即可无缝调用这些远程工具无需关心底层连接细节。这大大简化了为智能体扩展能力的过程。未来随着 MCP 生态的成熟集成外部服务可能会像安装一个插件一样简单。4.3 生产化部署考量将 LangGraph 智能体从笔记本搬到生产环境需要考虑以下几个关键点1. 状态存储后端内存存储MemorySaver只适用于开发和测试。生产环境需要选择可持久化、可扩展的后端。Redis速度快适合会话类状态使用RedisSaver。PostgreSQL可靠性高适合需要复杂查询或强一致性的场景使用PostgresSaver。文件系统简单但性能和多实例部署时有局限。2. 图的版本管理与热更新业务逻辑图结构变更时如何平滑升级可以考虑将图的定义Python 代码版本化部署新版本时逐步将流量切换到新图实例。对于检查点数据需要考虑向后兼容性。状态结构的重大变更可能导致旧的检查点无法加载。3. 监控与可观测性日志在每个节点函数中记录关键操作、输入输出和耗时。追踪Tracing利用 LangSmith 或 OpenTelemetry 等工具可视化每次invoke的完整执行路径、每个节点的输入输出以及大模型调用详情。这对于调试复杂流程和性能优化不可或缺。指标Metrics监控图的调用频率、成功率、各节点平均耗时、工具调用失败率等。4. 性能与扩展性节点函数优化避免在节点函数中执行阻塞性 I/O 操作考虑使用异步。大模型调用批处理如果流程中需要多次调用大模型看是否有可能合并提示词或使用批量 API。水平扩展由于状态存储在外部如 Redis你可以运行多个智能体服务实例通过负载均衡器分配请求。确保你的检查点存储后端支持并发访问。5. 常见问题排查与实战建议在实际开发和运行中你会遇到各种问题。以下是一些典型问题的排查思路和我个人的实战建议。5.1 状态更新不符合预期现象某个节点修改了状态但下一个节点读到的还是旧值。排查检查状态字段的合并策略这是最常见的原因。在定义TypedDict时对于列表、字典等可变类型必须使用Annotated指定合并操作符如operator.add用于列表追加。如果忘记指定默认行为可能是“替换”而非“合并”。# 正确列表会追加 messages: Annotated[List, operator.add] # 错误列表会被整个替换 messages: List检查节点返回值节点函数应返回一个字典其中只包含需要更新的字段。如果你返回了{“messages”: new_list}它会用new_list完全替换旧的messages列表。如果你想追加应该返回{“messages”: [new_message]}并依赖operator.add合并。使用 LangGraph Studio 可视化LangGraph 提供了可视化工具LangGraph Studio可以直观地看到状态在每个节点后的变化是调试状态流的利器。5.2 智能体陷入循环或无法结束现象图一直在“思考-行动”循环或者无法到达END节点。排查检查条件边的逻辑add_conditional_edges中定义的路由函数是核心。确保它在所有可能的状态下都能返回一个有效的节点名或END。添加默认分支return END是个好习惯。检查工具调用解析有时大模型返回的tool_calls格式不符合预期导致act_node执行失败或返回空结果状态没有正确更新从而触发新一轮思考。在act_node中打印tool_calls和工具执行结果。设置最大迭代次数在开发阶段可以在图中设置一个“安全阀”。例如在状态中添加一个iteration_count字段每经过一次“思考”节点就加1。在条件边逻辑中判断如果超过阈值如10次则强制返回END并返回一个“超时”消息。5.3 大模型 API 调用错误现象出现API error: 400,429(限流)或Connection lost等错误。排查与处理参数错误400仔细检查传递给大模型 API 的参数。例如某些 API 的thinking_budget参数要求是正整数。确保参数类型和值域符合 API 文档。认证失败401/403确认 API Key 正确且未过期并具有调用所需端点的权限。环境变量是否已正确加载速率限制429实现重试机制如指数退避。大多数 SDK如openai、langchain-openai内置了重试逻辑检查其配置。from langchain_openai import ChatOpenAI from tenacity import retry, stop_after_attempt, wait_exponential llm ChatOpenAI( modelgpt-3.5-turbo, temperature0, max_retries3, # 利用SDK内置重试 # 或者更精细的控制 # client_kwargs{max_retries: 3} )网络连接问题增加请求超时时间并在工具调用层捕获异常返回友好的错误信息给状态让智能体决定重试或降级。5.4 性能瓶颈分析现象智能体响应慢。排查顺序定位耗时环节使用 LangSmith 或手动打点记录每个节点尤其是大模型调用和工具调用的耗时。大模型调用通常是最大瓶颈。考虑使用更快的模型如gpt-3.5-turbo比gpt-4快。优化提示词减少不必要的上下文长度。尝试流式响应如果适用让用户感知更快。工具调用检查外部 API 或数据库查询的响应时间。考虑增加缓存、使用更高效的查询、或并行化可独立运行的工具调用。序列化/反序列化如果状态对象非常庞大例如包含很长的对话历史在检查点存储和加载时可能会慢。考虑对历史消息进行摘要或截断。5.5 我的实战建议从画图开始在写代码之前先用白板或绘图工具画出你想要的智能体工作流。明确状态包含哪些数据节点有哪些分支条件是什么。这能极大减少后续的返工。增量开发与测试不要试图一次性构建完整的复杂图。先构建一个最小可行图如只有两个节点的线性流测试状态流转。然后逐步添加节点、分支和工具。为状态设计深思熟虑的模式花时间设计好State的TypedDict。哪些字段是追加的哪些是覆盖的哪些是临时的清晰的状态模式是复杂流程可维护的基础。拥抱可视化调试尽早配置并使用LangGraph Studio。它能让你直观地看到执行路径、状态变化和错误点比看日志高效得多。生产环境优先考虑可观测性在部署前就把日志、指标和追踪Tracing方案想好并集成进去。当线上出现问题时这些数据是唯一的救命稻草。理解成本每次大模型调用、每次工具调用都有成本金钱或时间。在设计流程时思考是否每个分支都需要调用大模型能否用更简单的规则判断工具调用结果能否缓存LangGraph 提供的是一套强大的编排框架但它不负责解决所有问题。真正的挑战在于如何设计一个鲁棒、高效、可维护的智能体业务流程。把图结构理清把状态管好把工具集成稳剩下的就是根据具体业务需求进行迭代和优化了。