LangGraph实战：构建可控可扩展的智能体工作流

现在做 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 比较适合以下几类场景：

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 准备大模型 API

LangGraph 本身不直接调用大模型，它需要依赖 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(checkpointer=memory) config = {"configurable": {"thread_id": "user-session-1"}}

注意thread_id是区分不同会话的关键，同一个会话共享历史状态，不同会话之间隔离。

3.5 Graph 的编译与执行

构建完StateGraph后，需要调用compile()生成可执行的图对象。执行时使用invoke或stream：

app = graph.compile() result = app.invoke( {"input": "你好"}, config={"configurable": {"thread_id": "session-1"}} ) print(result)

stream方法适合需要实时输出中间过程、类似流式对话的场景。我们可以按节点粒度观察状态变化：

for event in app.stream({"input": "你好"}, config=config): 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(model="gpt-4o-mini", temperature=0) 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(checkpointer=memory) config = {"configurable": {"thread_id": "user-001"}} user_input = "我今天心情很不好，感觉项目快做不完了。" result = app.invoke( {"user_input": user_input}, config=config, ) 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 协议带来的新变化

MCP（Model 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 项目的过程中，几乎每个人都会遇到下面这些问题。我把常见现象、原因和解决方案整理成一张表，方便你对照排查。

7.1 调试技巧

LangGraph 提供了一个比较实用的调试方式：使用stream方法按节点输出状态变化。这样你就能清楚看到每一步状态是什么，从而定位问题。

for event in app.stream({"user_input": "你好"}, config=config): 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( checkpointer=memory, 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 应用的高速迭代。它并没有规定你必须怎么写业务逻辑，只是提供了一套清晰、可控、可扩展的组织方式。上手的时候，建议从一个最简图开始，感受状态流转的过程，再逐步叠加条件、并行、子图这些高级能力。项目中优先做好状态设计、条件边界和日志记录，基本就能避免很多常见的稳定性问题。希望这篇文章对你有所帮助，你可以先从第一节的安装环境开始动手跑一遍，再对照实战案例改造出自己的工作流。