现在做 AI 应用开发最大的痛点往往不是“模型不够聪明”而是“怎么把模型、工具、业务逻辑可靠地串起来”。尤其是当流程里出现分支判断、循环执行、需要记忆上下文、还要人工介入确认这些复杂状态时传统的while True if/else写法会很快变得不可维护。网上资料虽然多但要么是零散的 API 介绍要么是只讲 demo 不讲设计思路真正能落地到项目里的实战教程反而很少。这篇文章会把 LangGraph 从环境安装、核心概念、代码实战到工程优化完整串一遍重点覆盖条件路由、子图、并行执行、Checkpoint 记忆机制以及和 LangChain、MCP 生态的联动方式。不管你是刚接触智能体开发的新手还是已经在用 LangChain 但想进一步提升编排能力的开发者都能从中得到一套可复用的实现思路。我尽量不贴“半截代码”所有示例都给出完整文件结构和运行方式你可以直接照着复制、改造。遇到常见报错也会单独开一节说明排查思路减少你踩坑的时间。1. LangGraph 是什么为什么要关注它1.1 从“提示词调用”到“流程编排”早期的 AI 应用开发大多数是把 Prompt 拼好然后调用一次大模型接口拿回结果就结束。这种模式实现简单但很难满足复杂业务场景。比如一个客服机器人需要先判断用户意图再决定是否调用订单查询接口查询完还要根据结果生成不同的回复。如果只靠单次模型调用逻辑会全部塞进 Prompt模型输出的稳定性和可维护性都会很差。LangGraph 正好解决这个阶段的编排问题。它是建立在 LangChain 生态之上的一个框架核心思想是把智能体执行过程建模成一个“图”。图中的每个节点是一段具体的处理逻辑比如“调用大模型”“调用数据库”“执行代码”节点之间通过边连接定义执行顺序和依赖关系。你可以像画流程图一样设计智能体的工作流系统会按照图的结构去执行。1.2 LangGraph 和 LangChain 的区别很多刚开始接触的同学会把 LangGraph 和 LangChain 混为一谈。简单来说LangChain 偏重“组件层”提供了模型封装、Prompt 模板、文档加载、向量存储、工具调用等基础能力。LangGraph 偏重“编排层”负责把 LangChain 的能力组织成一个完整、可控、有状态的工作流。换句话说LangChain 给的是零件LangGraph 给的是装配图和生产线。你可以只用 LangGraph 不用 LangChain但大多数场景下两者配合使用体验最好。1.3 哪些场景更适合使用 LangGraph根据项目实践LangGraph 比较适合以下几类场景场景类型传统实现痛点LangGraph 带来的价值多步骤任务代码嵌套深、状态管理混乱节点化编排状态显式传递条件分支if/else 散落在业务代码中条件边集中管理流程可视化循环执行用 while 循环控制容易死循环图结构天然支持递归和循环人工介入需要暂停等待输入实现复杂支持中断和恢复便于人工审核长期记忆上下文丢失每次重头开始Checkpoint 机制保存状态快照并行执行多线程/异步代码复杂图结构支持并行分支1.4 LangGraph 的官方定位与主流智能体框架对比当前智能体开发框架有不少选择比如 Dify 这类可视化平台以及 AutoGen、CrewAI 等项目。相比之下LangGraph 的特点在于将“工作流”作为一等公民执行过程可控、可观测。与 LangChain 生态深度集成工具调用、检索增强、记忆管理等组件开箱即用。适合开发者通过代码精细控制而不是完全依赖可视化拖拽。Dify 这类平台更适合快速搭建业务应用而 LangGraph 更适合需要深度定制、逻辑复杂的工程化项目。你需要根据团队的技术栈和需求来决定使用哪一类的方案。2. 环境准备与项目结构2.1 环境依赖LangGraph 基于 Python 开发因此第一步确认 Python 环境。建议使用 Python 3.10 及以上版本避免一些版本兼容问题。官方文档对 Python 版本有要求但不同版本也可能有差异这里给你一个保守的配置建议Python 3.10推荐 3.10 或 3.11 稳定版操作系统Windows / macOS / Linux 均可开发工具VS Code 或 PyCharm接下来创建虚拟环境并安装依赖python -m venv langgraph-env source langgraph-env/bin/activate # Windows 使用 langgraph-env\Scripts\activate pip install --upgrade pip pip install langgraph langchain langchain-openai如果你的项目需要连接外部工具或走 MCP 协议可能还需要安装额外的依赖包。比如pip install mcp langchain-community注意依赖版本更新很快不同版本的 API 可能略有变化。如果你的代码运行报错优先检查当前安装的版本和官方文档是否对应。2.2 验证安装安装完成后在 Python 环境里执行以下代码确认核心库可以正常导入import langgraph import langchain print(langgraph version:, langgraph.__version__) print(langchain version:, langchain.__version__)正常情况下会输出两个版本号。如果导入报错说明依赖安装不完整或存在冲突建议重新创建虚拟环境安装。2.3 准备大模型 APILangGraph 本身不直接调用大模型它需要依赖 LangChain 的模型封装接口。你可以选择 OpenAI、DeepSeek、通义千问等任意兼容 OpenAI 接口的服务商。这里以设置环境变量的方式为例export OPENAI_API_KEY你的API Key export OPENAI_BASE_URL你的服务商Base URL # 可选兼容 OpenAI 接口的服务商可能需要如果不想使用环境变量也可以在代码中直接指定api_key参数但工程化建议还是用环境变量或密钥管理服务不要把明文 Key 提交到代码仓库。3. LangGraph 核心机制拆解要真正用好 LangGraph必须先理解它的几个核心概念。这些概念是 LangGraph 工作流的最小组成单位。3.1 State全局状态LangGraph 中的State是整个图执行期间共享的数据结构可以理解为工作流的“存档”。每经过一个节点状态都可能被更新。官方示例中通常使用TypedDict来定义状态结构from typing import TypedDict, Annotated class WorkflowState(TypedDict): input: str result: str messages: list实际项目中状态里会放多种数据比如用户输入、中间结果、错误信息、重试次数等。你可以把State理解成一个字典但它在内部有一套更新机制后面会说明。3.2 Node处理节点节点是最小执行单元。一个节点可以是一个函数内部既可以调用大模型也可以执行普通 Python 代码。函数的输入是上一轮的状态输出是一个字典字典中的键会更新到全局状态里。def process_node(state: WorkflowState) - dict: # 模拟处理逻辑 return {result: state[input].upper()}3.3 Edge边与控制流边定义了节点之间的连接关系。LangGraph 支持多种边普通边一个节点执行完后进入下一个节点。条件边根据当前状态或节点返回值动态决定下一个节点。起始边从特殊节点START出发。结束边进入特殊节点END表示流程结束。条件边是 LangGraph 一个很有价值的设计它把“决策逻辑”从业务代码中拆出来统一放在图定义中。例如下面这段代码就实现了如果结果包含某个关键词进入“审核”节点否则直接输出from langgraph.graph import StateGraph, START, END def should_review(state: WorkflowState) - str: if 危险 in state.get(result, ): return review return output graph.add_conditional_edges(process, should_review, { review: review_node, output: output_node, })3.4 Checkpoint状态持久化与记忆默认情况下每次运行图的调用都是独立的状态在调用结束后就消失了。如果需要让智能体拥有“记忆”就需要用到 Checkpoint。Checkpoint 会在每个节点执行后保存一份状态快照。当流程中断或需要恢复时可以从最近一次快照继续执行。这在很多场景下很有用例如多轮对话的长期记忆。需要人工审核、暂停后恢复的流程。异常恢复避免从头重跑。使用检查点需要传入一个MemorySaver或自定义的持久化存储对象from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() graph builder.compile(checkpointermemory) config {configurable: {thread_id: user-session-1}}注意thread_id是区分不同会话的关键同一个会话共享历史状态不同会话之间隔离。3.5 Graph 的编译与执行构建完StateGraph后需要调用compile()生成可执行的图对象。执行时使用invoke或streamapp graph.compile() result app.invoke( {input: 你好}, config{configurable: {thread_id: session-1}} ) print(result)stream方法适合需要实时输出中间过程、类似流式对话的场景。我们可以按节点粒度观察状态变化for event in app.stream({input: 你好}, configconfig): print(event)4. 完整实战搭建一个可扩展的 LangGraph 智能体这一节我们从零开始搭建一个相对完整的智能体工作流。示例需求如下用户输入一段文本智能体先判断文本情感如果判断为积极直接回复鼓励语如果判断为消极先调用一个“安慰策略”节点生成关怀内容再由人工确认是否发送。整个过程记录到状态中并支持多轮会话记忆。这是一个典型的“条件路由 人工介入 记忆”的综合场景。4.1 创建项目结构建议按下面结构组织文件方便后续扩展langgraph-demo/ ├── .env # 环境变量 ├── requirements.txt # 依赖 ├── main.py # 入口运行工作流 ├── workflow.py # 定义图和节点逻辑 └── models.py # 状态定义4.2 定义状态结构models.py内容如下from typing import TypedDict, Annotated, List class AgentState(TypedDict): # 用户原始输入 user_input: str # 情感判断结果positive / negative sentiment: str # 生成的回复内容 reply: str # 是否进入人工审核节点 need_review: bool # 对话历史用于多轮记忆 messages: Annotated[List[dict], message_history]这里使用Annotated标注messages字段只是说明用途实际是否自动合并取决于你使用的更新器。如果你希望 messages 是追加而不是覆盖可以自定义一个 reducer 函数。LangGraph 的add_messages函数就是这种用途from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages]4.3 编写核心节点在workflow.py中定义节点函数。第一个节点负责调用模型做情感判断# workflow.py from langchain_openai import ChatOpenAI from models import AgentState llm ChatOpenAI(modelgpt-4o-mini, temperature0) def analyze_sentiment(state: AgentState) - dict: 判断用户输入的情感倾向 prompt f 请判断下面用户输入的情感倾向只能输出一个词positive 或 negative。 用户输入{state[user_input]} response llm.invoke(prompt) sentiment response.content.strip().lower() return {sentiment: sentiment}第二个节点在消极情绪下生成关怀回复def generate_comfort_reply(state: AgentState) - dict: 为消极情绪生成关怀内容 prompt f 用户表达了消极情绪请生成一段温暖、不敷衍、有同理心的回复。 用户原话{state[user_input]} 要求100字以内语气自然。 response llm.invoke(prompt) return { reply: response.content.strip(), need_review: True, }第三个节点在积极情绪下生成正面回复def generate_positive_reply(state: AgentState) - dict: 为积极情绪生成鼓励回复 prompt f 用户表达了积极情绪请生成一段简短、正能量的回应。 用户原话{state[user_input]} 要求50字以内。 response llm.invoke(prompt) return { reply: response.content.strip(), need_review: False, }第四个节点模拟人工审核def human_review(state: AgentState) - dict: 模拟人工审核节点实际项目中可以由用户确认后继续 print(【人工审核】生成内容, state[reply]) # 这里可以接入实际的人工确认流程 return {reply: state[reply] 已通过人工审核}4.4 构建图与条件路由现在把节点连接起来# workflow.py from langgraph.graph import StateGraph, START, END def build_graph(): builder StateGraph(AgentState) # 注册节点 builder.add_node(analyze, analyze_sentiment) builder.add_node(positive, generate_positive_reply) builder.add_node(comfort, generate_comfort_reply) builder.add_node(review, human_review) # 起始 - 情感分析 builder.add_edge(START, analyze) # 条件路由根据情感判断走向不同分支 builder.add_conditional_edges( analyze, lambda state: state[sentiment], { positive: positive, negative: comfort, }, ) # 积极分支直接到 END builder.add_edge(positive, END) # 消极分支先生成关怀内容再进入人工审核 builder.add_edge(comfort, review) builder.add_edge(review, END) return builder这里有几个点需要注意add_conditional_edges的第二个参数是路由函数函数的入参是当前状态返回值需要匹配第三个参数字典中的某个键。如果状态中sentiment的取值是positive或negative路由函数返回对应字符串图会进入对应的节点。条件路由必须覆盖所有可能的分支否则运行时会报错。4.5 添加 Checkpoint 记忆能力为了让工作流支持多轮记忆在编译时传入检查点# main.py from langgraph.checkpoint.memory import MemorySaver from workflow import build_graph builder build_graph() memory MemorySaver() app builder.compile(checkpointermemory) config {configurable: {thread_id: user-001}} user_input 我今天心情很不好感觉项目快做不完了。 result app.invoke( {user_input: user_input}, configconfig, ) print(情感判断, result[sentiment]) print(最终回复, result[reply])第一次运行后再次传入新输入可以在messages字段中看到历史消息被保留下来。这就是 LangGraph 记忆机制带来的能力。4.6 运行与验证执行入口文件python main.py预期输出类似于【人工审核】生成内容 辛苦啦项目确实会有让人压力很大的时候…… 情感判断 negative 最终回复 辛苦啦项目确实会有让人压力很大的时候……已通过人工审核不要担心输出和你跑出来的不完全一样因为大模型本身有随机性。你需要关注的是流程是否正确是否走进了comfort分支是否触发了review节点以及状态是否正确更新。5. 进阶并行分支与子图当你深入了解 LangGraph 后会接触到两个对工程非常有用的特性并行分支和子图。5.1 并行执行提升效率的利器在一些场景中多个节点之间没有依赖关系可以并行执行。例如智能客服需要同时查询订单状态、商品信息和用户历史记录三者互不依赖就可以设计成并行分支。LangGraph 支持这种并行结构。示例from langgraph.graph import StateGraph, START, END def query_order(state): return {order_info: 订单信息} def query_product(state): return {product_info: 商品信息} def merge_result(state): return {final_result: f{state.get(order_info)} {state.get(product_info)}} builder StateGraph(AgentState) builder.add_node(query_order, query_order) builder.add_node(query_product, query_product) builder.add_node(merge, merge_result) builder.add_edge(START, query_order) builder.add_edge(START, query_product) builder.add_edge(query_order, merge) builder.add_edge(query_product, merge) builder.add_edge(merge, END)当START同时指向两个节点时LangGraph 会并行调度它们。需要注意的是并行分支的结果最终必须在某个节点合并否则状态会处于分离状态难以统一处理。5.2 子图模块化复用的关键子图允许你把一个完整的StateGraph作为另一个图的节点。这非常利于团队协作和模块复用。比如你可以把一个“订单售后处理流程”单独做成一个图然后在主图的某个节点里调用它。# 定义子图 child_builder StateGraph(ChildState) child_builder.add_node(child_step, child_step) child_builder.add_edge(START, child_step) child_builder.add_edge(child_step, END) child_graph child_builder.compile() # 在主图中作为节点使用 def call_child(state): return child_graph.invoke({some_key: state[some_key]}) main_builder StateGraph(MainState) main_builder.add_node(main_process, call_child)子图的优势在于复杂流程可以拆分成多个小图各自独立测试。多个功能模块可复用同一个子图。主图逻辑保持简洁可读性更高。5.3 循环与递归智能体的“自我反思”LangGraph 也支持循环结构这也是它和传统 DAG 工作流引擎不一样的地方。你可以设计一个“生成 - 评估 - 再生成”的循环让模型不断改进答案直到满足某个条件。例如先让模型写代码再用一个评判节点检查代码是否包含语法错误如果包含则重新生成最多重试 N 次MAX_RETRY 3 def should_continue(state): if state.get(retry_count, 0) MAX_RETRY: return end return generate builder.add_conditional_edges( evaluate, should_continue, {generate: generate, end: END}, )实现循环时务必要在状态中加入计数器并在每次循环后更新避免死循环。这是 LangGraph 循环设计里的最常见坑点。6. 与 LangChain 工具调用和 MCP 生态的联动6.1 在 LangGraph 节点中调用 LangChain 工具LangGraph 的一个显著优势是能复用 LangChain 的工具生态。你可以把一个自定义函数包装成工具然后在节点内部调用from langchain_core.tools import tool tool def get_weather(city: str) - str: 查询指定城市的天气信息 # 模拟查询 return f{city} 今天晴气温 22 度。 def weather_node(state): if state.get(need_weather): tool_result get_weather.invoke({city: state[city]}) return {weather_info: tool_result} return {weather_info: 不需要查询天气}在 LangGraph 中你可以决定工具调用发生在哪个节点以及工具结果如何影响下一个节点的执行。相比 LangChain 早期版本的 Agent 自动决策LangGraph 给了你更强的控制力。6.2 MCP 协议带来的新变化MCPModel Context Protocol是近期智能体生态中关注度很高的一个协议。它做的事情类似于“AI 应用时代的 USB-C 接口”通过标准化协议让大模型应用可以方便地接入各种外部数据源和工具服务。在 LangGraph 项目中MCP 主要应用在以下方面通过 MCP Server 将企业内部工具标准化暴露给智能体调用。统一接入多个外部服务平台避免为每个平台写单独的适配代码。让智能体具备动态发现工具的能力而不是把所有工具硬编码在代码中。在代码层面你仍然可以在 LangGraph 的某个节点里通过 MCP 客户端去调用远端工具再把结果写回状态。这样既保留了 LangGraph 的编排能力又获得了 MCP 带来的标准化接入能力。一个简化的流程如下用户输入 - LangGraph 节点 A意图识别 - LangGraph 节点 B调用 MCP Client - MCP Server执行外部工具 - LangGraph 节点 C组装回复如果你刚开始接触 MCP建议先从一个 Demo Server 开始熟悉工具注册、调用、返回结果的基本流程再逐步扩展到企业内部服务。7. 常见问题与排查思路在开发 LangGraph 项目的过程中几乎每个人都会遇到下面这些问题。我把常见现象、原因和解决方案整理成一张表方便你对照排查。问题现象常见原因解决思路ImportError: cannot import name StateGraph版本过旧或安装不完整升级langgraph到最新稳定版重新安装依赖运行时报错Invalid node name节点名不存在或拼写错误检查add_node注册的节点名与边引用的节点名是否一致条件路由报错提示返回值和映射不匹配路由函数返回的键没有在映射字典中定义检查路由函数的返回值确保覆盖所有可能分支状态不更新节点返回的字段没生效返回字典中的键不在状态结构中检查TypedDict是否包含该字段状态字段需要显式声明多轮对话历史丢失编译时没有传入checkpointer编译时传入checkpointermemory并在调用时指定thread_id流程进入死循环循环条件中计数器未更新在循环分支中显式更新计数并在条件边中设置最大重试次数调用大模型接口超时网络问题或 API Key 配置错误检查 API Key、Base URL 配置尝试减少单次请求长度并行分支数据不同步没有合并节点状态分裂增加一个合并节点汇总各分支结果7.1 调试技巧LangGraph 提供了一个比较实用的调试方式使用stream方法按节点输出状态变化。这样你就能清楚看到每一步状态是什么从而定位问题。for event in app.stream({user_input: 你好}, configconfig): print(event)输出会包含每个节点的名称和状态更新比如{analyze: {sentiment: positive}} {positive: {reply: 很高兴听到这个消息, need_review: False}}建议在开发阶段全程使用stream进行观察上线后再切换为invoke提高性能。8. 工程化最佳实践把 LangGraph 项目从 demo 做到可以上生产需要关注的就不只是功能实现了。下面这些建议来自实际项目中的经验总结。8.1 状态设计要克制状态是 LangGraph 的核心但并不是状态越多越好。过于庞大的状态结构会让图变得难以理解和调试。建议只保留跨节点共享的数据。节点内部临时数据不要塞进全局状态。对状态字段进行清晰命名避免data1、data2这种无意义命名。8.2 节点函数保持单一职责一个节点只做一件事。不要在一个节点里又调模型又查数据库又写日志。这样做的目的不是增加代码量而是让每个节点都容易被单独测试和替换。后续排查问题时也能快速定位到具体环节。8.3 使用中断机制处理人工审核在生产环境中人工审核通常不是简单打印一句话而是要暂停流程等待人工操作后再恢复。LangGraph 提供了中断机制来实现这一点相比自己维护外部状态要优雅得多。基本思路是在图中配置一个interrupt_before或interrupt_after让流程在指定节点暂停等待外部恢复指令。app builder.compile( checkpointermemory, interrupt_before[review], )运行时流程会在进入review前暂停应用可以保存当前状态给用户展示审核通过后再用相同的thread_id继续执行。这样既保证了安全合规又避免了自己维护复杂的暂停恢复逻辑。8.4 日志与可观测性智能体流程比普通程序更难调试因为其中涉及大模型的不确定性。建议在每个节点中至少记录节点名称。进入节点时的关键状态。离开节点时的关键状态。调用大模型的耗时和 token 消耗。如果条件允许可以将这些日志输出到结构化日志系统便于后期分析和优化。8.5 安全与权限当工作流需要调用外部服务或执行敏感操作时务必遵循最小权限原则不要在工作流节点中硬编码密钥。API Key 通过环境变量或密钥管理服务注入。对需要人工确认的高风险操作必须经过确认节点。外部工具调用前校验输入防止注入攻击。8.6 测试策略LangGraph 的图可以拆成独立节点函数因此测试策略也比较清晰对每一个节点函数做单元测试传入固定状态断言返回结果。对条件路由函数单独测试覆盖所有可能的分支取值。对整条工作流做集成测试使用 Mock 模型接口保证流程稳定。def test_analyze_sentiment(): state {user_input: 我今天很开心} result analyze_sentiment(state) assert result[sentiment] in [positive, negative]9. 学习路线从 LangGraph 入门到项目落地如果你是从零开始学 LangGraph可以参考下面的路线第一步先掌握 LangChain 的基础用法包括模型调用、Prompt 模板、工具封装、输出解析。这些是 LangGraph 节点的基本素材。第二步理解 LangGraph 的五个核心概念即 State、Node、Edge、Conditional Edge、Checkpoint。动手写一个只有一个节点的图再逐步增加节点和边。第三步实现一个包含条件路由的完整小项目比如“客服意图分流”。这一步重点体会条件边带来的流程可读性提升。第四步学习中断机制、子图、并行分支。这三个能力能让你应对 80% 以上的真实业务需求。第五步把项目接入外部系统和 MCP 服务让智能体真正操作业务工具同时考虑日志、安全、性能等生产因素。第六步在社区和官方文档中持续跟进新特性。LangGraph 迭代非常快不要试图一次学会所有功能先掌握核心主路径再按需扩展。如果你遇到问题优先查看官方文档中对应版本的 API 说明其次可以在开发者社区搜索特定报错信息。很多问题都是版本差异导致的注意确认你的安装版本和参考资料的适用范围。整体来看LangGraph 这套基于图结构的智能体编排思路比传统硬编码流程的方式更适合 AI 应用的高速迭代。它并没有规定你必须怎么写业务逻辑只是提供了一套清晰、可控、可扩展的组织方式。上手的时候建议从一个最简图开始感受状态流转的过程再逐步叠加条件、并行、子图这些高级能力。项目中优先做好状态设计、条件边界和日志记录基本就能避免很多常见的稳定性问题。希望这篇文章对你有所帮助你可以先从第一节的安装环境开始动手跑一遍再对照实战案例改造出自己的工作流。
现在做 AI 应用开发最大的痛点往往不是“模型不够聪明”而是“怎么把模型、工具、业务逻辑可靠地串起来”。尤其是当流程里出现分支判断、循环执行、需要记忆上下文、还要人工介入确认这些复杂状态时传统的while True if/else写法会很快变得不可维护。网上资料虽然多但要么是零散的 API 介绍要么是只讲 demo 不讲设计思路真正能落地到项目里的实战教程反而很少。这篇文章会把 LangGraph 从环境安装、核心概念、代码实战到工程优化完整串一遍重点覆盖条件路由、子图、并行执行、Checkpoint 记忆机制以及和 LangChain、MCP 生态的联动方式。不管你是刚接触智能体开发的新手还是已经在用 LangChain 但想进一步提升编排能力的开发者都能从中得到一套可复用的实现思路。我尽量不贴“半截代码”所有示例都给出完整文件结构和运行方式你可以直接照着复制、改造。遇到常见报错也会单独开一节说明排查思路减少你踩坑的时间。1. LangGraph 是什么为什么要关注它1.1 从“提示词调用”到“流程编排”早期的 AI 应用开发大多数是把 Prompt 拼好然后调用一次大模型接口拿回结果就结束。这种模式实现简单但很难满足复杂业务场景。比如一个客服机器人需要先判断用户意图再决定是否调用订单查询接口查询完还要根据结果生成不同的回复。如果只靠单次模型调用逻辑会全部塞进 Prompt模型输出的稳定性和可维护性都会很差。LangGraph 正好解决这个阶段的编排问题。它是建立在 LangChain 生态之上的一个框架核心思想是把智能体执行过程建模成一个“图”。图中的每个节点是一段具体的处理逻辑比如“调用大模型”“调用数据库”“执行代码”节点之间通过边连接定义执行顺序和依赖关系。你可以像画流程图一样设计智能体的工作流系统会按照图的结构去执行。1.2 LangGraph 和 LangChain 的区别很多刚开始接触的同学会把 LangGraph 和 LangChain 混为一谈。简单来说LangChain 偏重“组件层”提供了模型封装、Prompt 模板、文档加载、向量存储、工具调用等基础能力。LangGraph 偏重“编排层”负责把 LangChain 的能力组织成一个完整、可控、有状态的工作流。换句话说LangChain 给的是零件LangGraph 给的是装配图和生产线。你可以只用 LangGraph 不用 LangChain但大多数场景下两者配合使用体验最好。1.3 哪些场景更适合使用 LangGraph根据项目实践LangGraph 比较适合以下几类场景场景类型传统实现痛点LangGraph 带来的价值多步骤任务代码嵌套深、状态管理混乱节点化编排状态显式传递条件分支if/else 散落在业务代码中条件边集中管理流程可视化循环执行用 while 循环控制容易死循环图结构天然支持递归和循环人工介入需要暂停等待输入实现复杂支持中断和恢复便于人工审核长期记忆上下文丢失每次重头开始Checkpoint 机制保存状态快照并行执行多线程/异步代码复杂图结构支持并行分支1.4 LangGraph 的官方定位与主流智能体框架对比当前智能体开发框架有不少选择比如 Dify 这类可视化平台以及 AutoGen、CrewAI 等项目。相比之下LangGraph 的特点在于将“工作流”作为一等公民执行过程可控、可观测。与 LangChain 生态深度集成工具调用、检索增强、记忆管理等组件开箱即用。适合开发者通过代码精细控制而不是完全依赖可视化拖拽。Dify 这类平台更适合快速搭建业务应用而 LangGraph 更适合需要深度定制、逻辑复杂的工程化项目。你需要根据团队的技术栈和需求来决定使用哪一类的方案。2. 环境准备与项目结构2.1 环境依赖LangGraph 基于 Python 开发因此第一步确认 Python 环境。建议使用 Python 3.10 及以上版本避免一些版本兼容问题。官方文档对 Python 版本有要求但不同版本也可能有差异这里给你一个保守的配置建议Python 3.10推荐 3.10 或 3.11 稳定版操作系统Windows / macOS / Linux 均可开发工具VS Code 或 PyCharm接下来创建虚拟环境并安装依赖python -m venv langgraph-env source langgraph-env/bin/activate # Windows 使用 langgraph-env\Scripts\activate pip install --upgrade pip pip install langgraph langchain langchain-openai如果你的项目需要连接外部工具或走 MCP 协议可能还需要安装额外的依赖包。比如pip install mcp langchain-community注意依赖版本更新很快不同版本的 API 可能略有变化。如果你的代码运行报错优先检查当前安装的版本和官方文档是否对应。2.2 验证安装安装完成后在 Python 环境里执行以下代码确认核心库可以正常导入import langgraph import langchain print(langgraph version:, langgraph.__version__) print(langchain version:, langchain.__version__)正常情况下会输出两个版本号。如果导入报错说明依赖安装不完整或存在冲突建议重新创建虚拟环境安装。2.3 准备大模型 APILangGraph 本身不直接调用大模型它需要依赖 LangChain 的模型封装接口。你可以选择 OpenAI、DeepSeek、通义千问等任意兼容 OpenAI 接口的服务商。这里以设置环境变量的方式为例export OPENAI_API_KEY你的API Key export OPENAI_BASE_URL你的服务商Base URL # 可选兼容 OpenAI 接口的服务商可能需要如果不想使用环境变量也可以在代码中直接指定api_key参数但工程化建议还是用环境变量或密钥管理服务不要把明文 Key 提交到代码仓库。3. LangGraph 核心机制拆解要真正用好 LangGraph必须先理解它的几个核心概念。这些概念是 LangGraph 工作流的最小组成单位。3.1 State全局状态LangGraph 中的State是整个图执行期间共享的数据结构可以理解为工作流的“存档”。每经过一个节点状态都可能被更新。官方示例中通常使用TypedDict来定义状态结构from typing import TypedDict, Annotated class WorkflowState(TypedDict): input: str result: str messages: list实际项目中状态里会放多种数据比如用户输入、中间结果、错误信息、重试次数等。你可以把State理解成一个字典但它在内部有一套更新机制后面会说明。3.2 Node处理节点节点是最小执行单元。一个节点可以是一个函数内部既可以调用大模型也可以执行普通 Python 代码。函数的输入是上一轮的状态输出是一个字典字典中的键会更新到全局状态里。def process_node(state: WorkflowState) - dict: # 模拟处理逻辑 return {result: state[input].upper()}3.3 Edge边与控制流边定义了节点之间的连接关系。LangGraph 支持多种边普通边一个节点执行完后进入下一个节点。条件边根据当前状态或节点返回值动态决定下一个节点。起始边从特殊节点START出发。结束边进入特殊节点END表示流程结束。条件边是 LangGraph 一个很有价值的设计它把“决策逻辑”从业务代码中拆出来统一放在图定义中。例如下面这段代码就实现了如果结果包含某个关键词进入“审核”节点否则直接输出from langgraph.graph import StateGraph, START, END def should_review(state: WorkflowState) - str: if 危险 in state.get(result, ): return review return output graph.add_conditional_edges(process, should_review, { review: review_node, output: output_node, })3.4 Checkpoint状态持久化与记忆默认情况下每次运行图的调用都是独立的状态在调用结束后就消失了。如果需要让智能体拥有“记忆”就需要用到 Checkpoint。Checkpoint 会在每个节点执行后保存一份状态快照。当流程中断或需要恢复时可以从最近一次快照继续执行。这在很多场景下很有用例如多轮对话的长期记忆。需要人工审核、暂停后恢复的流程。异常恢复避免从头重跑。使用检查点需要传入一个MemorySaver或自定义的持久化存储对象from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() graph builder.compile(checkpointermemory) config {configurable: {thread_id: user-session-1}}注意thread_id是区分不同会话的关键同一个会话共享历史状态不同会话之间隔离。3.5 Graph 的编译与执行构建完StateGraph后需要调用compile()生成可执行的图对象。执行时使用invoke或streamapp graph.compile() result app.invoke( {input: 你好}, config{configurable: {thread_id: session-1}} ) print(result)stream方法适合需要实时输出中间过程、类似流式对话的场景。我们可以按节点粒度观察状态变化for event in app.stream({input: 你好}, configconfig): print(event)4. 完整实战搭建一个可扩展的 LangGraph 智能体这一节我们从零开始搭建一个相对完整的智能体工作流。示例需求如下用户输入一段文本智能体先判断文本情感如果判断为积极直接回复鼓励语如果判断为消极先调用一个“安慰策略”节点生成关怀内容再由人工确认是否发送。整个过程记录到状态中并支持多轮会话记忆。这是一个典型的“条件路由 人工介入 记忆”的综合场景。4.1 创建项目结构建议按下面结构组织文件方便后续扩展langgraph-demo/ ├── .env # 环境变量 ├── requirements.txt # 依赖 ├── main.py # 入口运行工作流 ├── workflow.py # 定义图和节点逻辑 └── models.py # 状态定义4.2 定义状态结构models.py内容如下from typing import TypedDict, Annotated, List class AgentState(TypedDict): # 用户原始输入 user_input: str # 情感判断结果positive / negative sentiment: str # 生成的回复内容 reply: str # 是否进入人工审核节点 need_review: bool # 对话历史用于多轮记忆 messages: Annotated[List[dict], message_history]这里使用Annotated标注messages字段只是说明用途实际是否自动合并取决于你使用的更新器。如果你希望 messages 是追加而不是覆盖可以自定义一个 reducer 函数。LangGraph 的add_messages函数就是这种用途from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages]4.3 编写核心节点在workflow.py中定义节点函数。第一个节点负责调用模型做情感判断# workflow.py from langchain_openai import ChatOpenAI from models import AgentState llm ChatOpenAI(modelgpt-4o-mini, temperature0) def analyze_sentiment(state: AgentState) - dict: 判断用户输入的情感倾向 prompt f 请判断下面用户输入的情感倾向只能输出一个词positive 或 negative。 用户输入{state[user_input]} response llm.invoke(prompt) sentiment response.content.strip().lower() return {sentiment: sentiment}第二个节点在消极情绪下生成关怀回复def generate_comfort_reply(state: AgentState) - dict: 为消极情绪生成关怀内容 prompt f 用户表达了消极情绪请生成一段温暖、不敷衍、有同理心的回复。 用户原话{state[user_input]} 要求100字以内语气自然。 response llm.invoke(prompt) return { reply: response.content.strip(), need_review: True, }第三个节点在积极情绪下生成正面回复def generate_positive_reply(state: AgentState) - dict: 为积极情绪生成鼓励回复 prompt f 用户表达了积极情绪请生成一段简短、正能量的回应。 用户原话{state[user_input]} 要求50字以内。 response llm.invoke(prompt) return { reply: response.content.strip(), need_review: False, }第四个节点模拟人工审核def human_review(state: AgentState) - dict: 模拟人工审核节点实际项目中可以由用户确认后继续 print(【人工审核】生成内容, state[reply]) # 这里可以接入实际的人工确认流程 return {reply: state[reply] 已通过人工审核}4.4 构建图与条件路由现在把节点连接起来# workflow.py from langgraph.graph import StateGraph, START, END def build_graph(): builder StateGraph(AgentState) # 注册节点 builder.add_node(analyze, analyze_sentiment) builder.add_node(positive, generate_positive_reply) builder.add_node(comfort, generate_comfort_reply) builder.add_node(review, human_review) # 起始 - 情感分析 builder.add_edge(START, analyze) # 条件路由根据情感判断走向不同分支 builder.add_conditional_edges( analyze, lambda state: state[sentiment], { positive: positive, negative: comfort, }, ) # 积极分支直接到 END builder.add_edge(positive, END) # 消极分支先生成关怀内容再进入人工审核 builder.add_edge(comfort, review) builder.add_edge(review, END) return builder这里有几个点需要注意add_conditional_edges的第二个参数是路由函数函数的入参是当前状态返回值需要匹配第三个参数字典中的某个键。如果状态中sentiment的取值是positive或negative路由函数返回对应字符串图会进入对应的节点。条件路由必须覆盖所有可能的分支否则运行时会报错。4.5 添加 Checkpoint 记忆能力为了让工作流支持多轮记忆在编译时传入检查点# main.py from langgraph.checkpoint.memory import MemorySaver from workflow import build_graph builder build_graph() memory MemorySaver() app builder.compile(checkpointermemory) config {configurable: {thread_id: user-001}} user_input 我今天心情很不好感觉项目快做不完了。 result app.invoke( {user_input: user_input}, configconfig, ) print(情感判断, result[sentiment]) print(最终回复, result[reply])第一次运行后再次传入新输入可以在messages字段中看到历史消息被保留下来。这就是 LangGraph 记忆机制带来的能力。4.6 运行与验证执行入口文件python main.py预期输出类似于【人工审核】生成内容 辛苦啦项目确实会有让人压力很大的时候…… 情感判断 negative 最终回复 辛苦啦项目确实会有让人压力很大的时候……已通过人工审核不要担心输出和你跑出来的不完全一样因为大模型本身有随机性。你需要关注的是流程是否正确是否走进了comfort分支是否触发了review节点以及状态是否正确更新。5. 进阶并行分支与子图当你深入了解 LangGraph 后会接触到两个对工程非常有用的特性并行分支和子图。5.1 并行执行提升效率的利器在一些场景中多个节点之间没有依赖关系可以并行执行。例如智能客服需要同时查询订单状态、商品信息和用户历史记录三者互不依赖就可以设计成并行分支。LangGraph 支持这种并行结构。示例from langgraph.graph import StateGraph, START, END def query_order(state): return {order_info: 订单信息} def query_product(state): return {product_info: 商品信息} def merge_result(state): return {final_result: f{state.get(order_info)} {state.get(product_info)}} builder StateGraph(AgentState) builder.add_node(query_order, query_order) builder.add_node(query_product, query_product) builder.add_node(merge, merge_result) builder.add_edge(START, query_order) builder.add_edge(START, query_product) builder.add_edge(query_order, merge) builder.add_edge(query_product, merge) builder.add_edge(merge, END)当START同时指向两个节点时LangGraph 会并行调度它们。需要注意的是并行分支的结果最终必须在某个节点合并否则状态会处于分离状态难以统一处理。5.2 子图模块化复用的关键子图允许你把一个完整的StateGraph作为另一个图的节点。这非常利于团队协作和模块复用。比如你可以把一个“订单售后处理流程”单独做成一个图然后在主图的某个节点里调用它。# 定义子图 child_builder StateGraph(ChildState) child_builder.add_node(child_step, child_step) child_builder.add_edge(START, child_step) child_builder.add_edge(child_step, END) child_graph child_builder.compile() # 在主图中作为节点使用 def call_child(state): return child_graph.invoke({some_key: state[some_key]}) main_builder StateGraph(MainState) main_builder.add_node(main_process, call_child)子图的优势在于复杂流程可以拆分成多个小图各自独立测试。多个功能模块可复用同一个子图。主图逻辑保持简洁可读性更高。5.3 循环与递归智能体的“自我反思”LangGraph 也支持循环结构这也是它和传统 DAG 工作流引擎不一样的地方。你可以设计一个“生成 - 评估 - 再生成”的循环让模型不断改进答案直到满足某个条件。例如先让模型写代码再用一个评判节点检查代码是否包含语法错误如果包含则重新生成最多重试 N 次MAX_RETRY 3 def should_continue(state): if state.get(retry_count, 0) MAX_RETRY: return end return generate builder.add_conditional_edges( evaluate, should_continue, {generate: generate, end: END}, )实现循环时务必要在状态中加入计数器并在每次循环后更新避免死循环。这是 LangGraph 循环设计里的最常见坑点。6. 与 LangChain 工具调用和 MCP 生态的联动6.1 在 LangGraph 节点中调用 LangChain 工具LangGraph 的一个显著优势是能复用 LangChain 的工具生态。你可以把一个自定义函数包装成工具然后在节点内部调用from langchain_core.tools import tool tool def get_weather(city: str) - str: 查询指定城市的天气信息 # 模拟查询 return f{city} 今天晴气温 22 度。 def weather_node(state): if state.get(need_weather): tool_result get_weather.invoke({city: state[city]}) return {weather_info: tool_result} return {weather_info: 不需要查询天气}在 LangGraph 中你可以决定工具调用发生在哪个节点以及工具结果如何影响下一个节点的执行。相比 LangChain 早期版本的 Agent 自动决策LangGraph 给了你更强的控制力。6.2 MCP 协议带来的新变化MCPModel Context Protocol是近期智能体生态中关注度很高的一个协议。它做的事情类似于“AI 应用时代的 USB-C 接口”通过标准化协议让大模型应用可以方便地接入各种外部数据源和工具服务。在 LangGraph 项目中MCP 主要应用在以下方面通过 MCP Server 将企业内部工具标准化暴露给智能体调用。统一接入多个外部服务平台避免为每个平台写单独的适配代码。让智能体具备动态发现工具的能力而不是把所有工具硬编码在代码中。在代码层面你仍然可以在 LangGraph 的某个节点里通过 MCP 客户端去调用远端工具再把结果写回状态。这样既保留了 LangGraph 的编排能力又获得了 MCP 带来的标准化接入能力。一个简化的流程如下用户输入 - LangGraph 节点 A意图识别 - LangGraph 节点 B调用 MCP Client - MCP Server执行外部工具 - LangGraph 节点 C组装回复如果你刚开始接触 MCP建议先从一个 Demo Server 开始熟悉工具注册、调用、返回结果的基本流程再逐步扩展到企业内部服务。7. 常见问题与排查思路在开发 LangGraph 项目的过程中几乎每个人都会遇到下面这些问题。我把常见现象、原因和解决方案整理成一张表方便你对照排查。问题现象常见原因解决思路ImportError: cannot import name StateGraph版本过旧或安装不完整升级langgraph到最新稳定版重新安装依赖运行时报错Invalid node name节点名不存在或拼写错误检查add_node注册的节点名与边引用的节点名是否一致条件路由报错提示返回值和映射不匹配路由函数返回的键没有在映射字典中定义检查路由函数的返回值确保覆盖所有可能分支状态不更新节点返回的字段没生效返回字典中的键不在状态结构中检查TypedDict是否包含该字段状态字段需要显式声明多轮对话历史丢失编译时没有传入checkpointer编译时传入checkpointermemory并在调用时指定thread_id流程进入死循环循环条件中计数器未更新在循环分支中显式更新计数并在条件边中设置最大重试次数调用大模型接口超时网络问题或 API Key 配置错误检查 API Key、Base URL 配置尝试减少单次请求长度并行分支数据不同步没有合并节点状态分裂增加一个合并节点汇总各分支结果7.1 调试技巧LangGraph 提供了一个比较实用的调试方式使用stream方法按节点输出状态变化。这样你就能清楚看到每一步状态是什么从而定位问题。for event in app.stream({user_input: 你好}, configconfig): print(event)输出会包含每个节点的名称和状态更新比如{analyze: {sentiment: positive}} {positive: {reply: 很高兴听到这个消息, need_review: False}}建议在开发阶段全程使用stream进行观察上线后再切换为invoke提高性能。8. 工程化最佳实践把 LangGraph 项目从 demo 做到可以上生产需要关注的就不只是功能实现了。下面这些建议来自实际项目中的经验总结。8.1 状态设计要克制状态是 LangGraph 的核心但并不是状态越多越好。过于庞大的状态结构会让图变得难以理解和调试。建议只保留跨节点共享的数据。节点内部临时数据不要塞进全局状态。对状态字段进行清晰命名避免data1、data2这种无意义命名。8.2 节点函数保持单一职责一个节点只做一件事。不要在一个节点里又调模型又查数据库又写日志。这样做的目的不是增加代码量而是让每个节点都容易被单独测试和替换。后续排查问题时也能快速定位到具体环节。8.3 使用中断机制处理人工审核在生产环境中人工审核通常不是简单打印一句话而是要暂停流程等待人工操作后再恢复。LangGraph 提供了中断机制来实现这一点相比自己维护外部状态要优雅得多。基本思路是在图中配置一个interrupt_before或interrupt_after让流程在指定节点暂停等待外部恢复指令。app builder.compile( checkpointermemory, interrupt_before[review], )运行时流程会在进入review前暂停应用可以保存当前状态给用户展示审核通过后再用相同的thread_id继续执行。这样既保证了安全合规又避免了自己维护复杂的暂停恢复逻辑。8.4 日志与可观测性智能体流程比普通程序更难调试因为其中涉及大模型的不确定性。建议在每个节点中至少记录节点名称。进入节点时的关键状态。离开节点时的关键状态。调用大模型的耗时和 token 消耗。如果条件允许可以将这些日志输出到结构化日志系统便于后期分析和优化。8.5 安全与权限当工作流需要调用外部服务或执行敏感操作时务必遵循最小权限原则不要在工作流节点中硬编码密钥。API Key 通过环境变量或密钥管理服务注入。对需要人工确认的高风险操作必须经过确认节点。外部工具调用前校验输入防止注入攻击。8.6 测试策略LangGraph 的图可以拆成独立节点函数因此测试策略也比较清晰对每一个节点函数做单元测试传入固定状态断言返回结果。对条件路由函数单独测试覆盖所有可能的分支取值。对整条工作流做集成测试使用 Mock 模型接口保证流程稳定。def test_analyze_sentiment(): state {user_input: 我今天很开心} result analyze_sentiment(state) assert result[sentiment] in [positive, negative]9. 学习路线从 LangGraph 入门到项目落地如果你是从零开始学 LangGraph可以参考下面的路线第一步先掌握 LangChain 的基础用法包括模型调用、Prompt 模板、工具封装、输出解析。这些是 LangGraph 节点的基本素材。第二步理解 LangGraph 的五个核心概念即 State、Node、Edge、Conditional Edge、Checkpoint。动手写一个只有一个节点的图再逐步增加节点和边。第三步实现一个包含条件路由的完整小项目比如“客服意图分流”。这一步重点体会条件边带来的流程可读性提升。第四步学习中断机制、子图、并行分支。这三个能力能让你应对 80% 以上的真实业务需求。第五步把项目接入外部系统和 MCP 服务让智能体真正操作业务工具同时考虑日志、安全、性能等生产因素。第六步在社区和官方文档中持续跟进新特性。LangGraph 迭代非常快不要试图一次学会所有功能先掌握核心主路径再按需扩展。如果你遇到问题优先查看官方文档中对应版本的 API 说明其次可以在开发者社区搜索特定报错信息。很多问题都是版本差异导致的注意确认你的安装版本和参考资料的适用范围。整体来看LangGraph 这套基于图结构的智能体编排思路比传统硬编码流程的方式更适合 AI 应用的高速迭代。它并没有规定你必须怎么写业务逻辑只是提供了一套清晰、可控、可扩展的组织方式。上手的时候建议从一个最简图开始感受状态流转的过程再逐步叠加条件、并行、子图这些高级能力。项目中优先做好状态设计、条件边界和日志记录基本就能避免很多常见的稳定性问题。希望这篇文章对你有所帮助你可以先从第一节的安装环境开始动手跑一遍再对照实战案例改造出自己的工作流。
