LangGraph 实战：从状态图到企业级智能工单系统

1. 为什么 LangGraph 值得你花时间啃下来

如果你最近在折腾大模型应用，大概率已经听过 LangChain 这个名字。但很多人用着用着就会发现一个问题：简单的链式调用（Chain）能搞定的事情太有限了。一旦业务逻辑里出现分支判断、循环重试、多角色协作，或者需要人工介入审核，传统的链式结构就会变得极其臃肿，代码可读性断崖式下跌。

LangGraph 就是在这个背景下出现的。它把大模型应用的编排方式从“一条链走到底”升级成了“一张图”。你可以把每个处理节点想象成流水线上的一个工位，节点之间用边连接，数据沿着边流动，遇到条件判断就分叉，遇到循环就绕回去，遇到需要人确认的地方就暂停等待。这套思路其实并不新鲜，状态机、工作流引擎早就在传统软件工程里用了几十年，但把它和大模型的能力结合起来，就产生了很多有意思的玩法。

我最初接触 LangGraph 是因为一个多轮对话审核的需求。用户提交内容后，需要先经过自动审核，不通过的走人工复审，人工复审后还要根据结果决定是直接发布、打回修改还是加入观察名单。用传统 Chain 写，光是条件分支就能把人绕晕，而且状态管理全靠手动传递字典，稍不注意就丢字段。换成 LangGraph 之后，整个流程变成了一张清晰的图，每个节点职责单一，状态由框架统一管理，调试的时候还能把整张图可视化出来，效率提升非常明显。

这篇文章适合哪些人看？如果你已经用过 LangChain 的基础功能，想进一步构建更复杂的应用；或者你正在做智能客服、自动化审批、多智能体协作这类项目，需要一套可靠的编排框架；又或者你只是听说 LangGraph 很火，想找个机会系统了解一下它到底能干什么，那接下来的内容应该能帮到你。我会从核心概念讲起，一步步带你搭出一个可运行的企业级项目骨架，中间穿插我踩过的坑和总结的技巧。

2. LangGraph 核心概念拆解：别被术语吓到

2.1 状态图到底是个什么东西

LangGraph 的核心抽象是“状态图”（StateGraph）。你可以把它理解成一张有向图，图里的每个节点是一个函数，负责处理数据并返回更新后的状态；边则定义了节点之间的流转关系。整个图共享一个状态对象，每个节点都能读取和修改这个状态。

听起来有点抽象，举个生活化的例子。想象你在一家餐厅点餐，整个流程可以画成一张图：你坐下（初始状态）→ 服务员递菜单（节点A）→ 你点菜（节点B）→ 厨房做菜（节点C）→ 服务员上菜（节点D）→ 你用餐（节点E）→ 结账（节点F）。如果菜做错了，可能从节点C回到节点B重新点；如果你对菜品不满意，可能从节点E跳到节点D要求换菜。LangGraph 做的就是帮你把这类流程定义清楚，并且保证状态在流转过程中不会乱掉。

状态本身通常是一个字典或者 TypedDict，里面可以放消息列表、用户信息、中间结果、标记位等等。关键点在于：每个节点不需要返回完整的状态，只需要返回它修改的那部分字段，LangGraph 会自动帮你合并。这个设计非常实用，避免了手动维护大字典的痛苦。

2.2 节点、边与条件路由的分工

节点是图的基本执行单元。在 LangGraph 里，节点就是一个普通的 Python 函数，接收当前状态作为参数，返回一个字典表示要更新的字段。节点可以是调用大模型的函数，也可以是纯逻辑处理函数，甚至可以是调用外部 API 的函数。这种设计让节点的职责非常清晰，方便单独测试和复用。

边分为两种：普通边和条件边。普通边就是 A 执行完直接到 B，没什么好说的。条件边则是在 A 执行完后，根据当前状态决定下一步去 B 还是去 C。条件边需要你提供一个路由函数，这个函数读取状态，返回一个字符串，LangGraph 根据这个字符串找到对应的下一个节点。

这里有个容易混淆的地方：条件边本身不执行任何业务逻辑，它只负责“看状态、做判断、指方向”。真正的业务判断逻辑应该写在路由函数里，而不是塞在节点里。我见过有人在节点里做判断然后返回一个特殊标记，再在条件边里读这个标记，绕了一大圈，其实完全没必要。

2.3 检查点与人工介入机制

检查点（Checkpointer）是 LangGraph 里非常关键的一个特性。它会在每个节点执行完后自动保存当前状态，这样如果流程中断了，可以从上次保存的地方恢复，而不需要从头再来。对于需要人工审核的场景，这个功能简直是救星。

人工介入的实现方式通常是：在某个节点后设置一个中断点，流程执行到这里就暂停，把当前状态暴露给外部系统。人工审核完成后，通过 API 把审核结果写回状态，然后继续执行。整个过程不需要保持长连接，也不需要自己实现复杂的状态持久化逻辑，LangGraph 都帮你处理好了。

我实测下来，检查点用 SQLite 做本地开发、用 PostgreSQL 做生产环境是比较稳妥的选择。内存检查点虽然快，但进程一重启就没了，只适合临时测试。

3. 环境搭建与第一个可运行图

3.1 依赖安装与版本选择

LangGraph 的安装本身不复杂，但版本兼容性是个大坑。我的建议是：先确定你用的 LangChain 版本，然后去查 LangGraph 的官方兼容性说明，不要盲目装最新版。我踩过一次坑，LangChain 用的是 0.1.x，LangGraph 装了 0.2.x，结果状态合并的逻辑变了，调试了大半天才发现是版本不匹配。

基础安装命令如下：

pip install langgraph langchain langchain-openai

如果你要用检查点功能，还需要额外安装：

pip install langgraph-checkpoint-sqlite

Python 版本建议 3.10 以上，3.9 虽然也能跑，但有些类型提示的写法会报错。另外，如果你打算用 OpenAI 的模型，记得把 API Key 配好，建议用环境变量管理，不要硬编码在代码里。

3.2 定义状态与节点函数

先来看一个最简单的例子，帮你建立直观感受。假设我们要做一个文本处理流程：接收一段文字，先判断长度，短文本直接总结，长文本先分段再总结。

from typing import TypedDict, List from langgraph.graph import StateGraph, END class TextState(TypedDict): content: str length: int segments: List[str] summary: str def check_length(state: TextState) -> dict: return {"length": len(state["content"])} def split_text(state: TextState) -> dict: text = state["content"] mid = len(text) // 2 return {"segments": [text[:mid], text[mid:]]} def summarize_short(state: TextState) -> dict: return {"summary": f"短文本总结：{state['content'][:20]}..."} def summarize_long(state: TextState) -> dict: return {"summary": f"长文本总结：共{len(state['segments'])}段"}

这里的状态定义用了 TypedDict，好处是类型清晰，IDE 能给你提示。每个节点函数只返回自己修改的字段，不需要把整个状态都返回。注意check_length只更新了length，其他字段保持不变，LangGraph 会自动合并。

3.3 构建图并跑通第一个流程

定义完节点后，就可以组装图了：

def route_by_length(state: TextState) -> str: if state["length"] < 100: return "short" return "long" builder = StateGraph(TextState) builder.add_node("check", check_length) builder.add_node("split", split_text) builder.add_node("sum_short", summarize_short) builder.add_node("sum_long", summarize_long) builder.set_entry_point("check") builder.add_conditional_edges( "check", route_by_length, {"short": "sum_short", "long": "split"} ) builder.add_edge("split", "sum_long") builder.add_edge("sum_short", END) builder.add_edge("sum_long", END) graph = builder.compile() result = graph.invoke({"content": "这是一段测试文本" * 20}) print(result["summary"])

跑通这个例子后，你应该能感受到 LangGraph 的基本套路：定义状态 → 写节点函数 → 加节点 → 连边 → 编译 → 调用。条件边负责分流，普通边负责串联，END 表示流程结束。

注意：set_entry_point在新版本里也可以用add_edge(START, "check")替代，两种写法都行，但建议统一风格，避免混用。

4. 企业级实战项目：智能工单处理系统

4.1 业务场景与流程设计

前面那个例子只是热身，现在我们来做一个真正有实用价值的项目：智能工单处理系统。这个系统的业务逻辑是这样的：用户提交工单后，先由 AI 进行自动分类和优先级判断；如果是简单问题，AI 直接生成回复；如果是复杂问题，转人工处理；人工处理完后，系统自动发送通知并归档。

整个流程涉及多个节点：接收工单、AI 分类、优先级判断、自动回复、转人工、人工处理、发送通知、归档。其中优先级判断后需要条件路由，人工处理需要中断等待，归档后流程结束。这个场景足够典型，涵盖了 LangGraph 的大部分核心功能。

我选择这个场景的原因是它在实际工作中非常常见，很多公司的客服系统、运维系统、审批系统都有类似需求。而且这个场景天然适合用图来表达，每个环节的职责边界清晰，状态流转明确。

4.2 状态字段设计与数据流转

状态设计是整张图的地基，设计不好后面会非常痛苦。我的经验是：把所有可能用到的字段都提前想清楚，宁可多留几个字段，也不要中途频繁改状态结构。

from typing import TypedDict, Optional, List from datetime import datetime class TicketState(TypedDict): ticket_id: str content: str category: Optional[str] priority: Optional[str] auto_reply: Optional[str] need_human: bool human_result: Optional[str] notification: Optional[str] archived: bool history: List[str]

这里我加了一个history字段，用来记录每个节点的执行轨迹。这个字段在调试的时候特别有用，出问题了一看就知道卡在哪个环节。need_human是个布尔标记，用来控制是否走人工分支。archived标记工单是否已归档，避免重复处理。

数据流转的原则是：每个节点只修改自己负责的字段，不要越界。比如分类节点只改category，优先级节点只改priority，这样职责清晰，出问题容易定位。

4.3 核心节点实现与条件路由

先实现几个关键节点：

def receive_ticket(state: TicketState) -> dict: return { "history": state.get("history", []) + ["receive"], "ticket_id": f"TK-{datetime.now().strftime('%Y%m%d%H%M%S')}" } def classify_ticket(state: TicketState) -> dict: content = state["content"] if "退款" in content or "退货" in content: category = "售后" elif "故障" in content or "报错" in content: category = "技术" else: category = "咨询" return { "category": category, "history": state["history"] + ["classify"] } def judge_priority(state: TicketState) -> dict: category = state["category"] if category == "技术": priority = "高" elif category == "售后": priority = "中" else: priority = "低" return { "priority": priority, "history": state["history"] + ["priority"] }

路由函数负责决定下一步走向：

def route_by_priority(state: TicketState) -> str: if state["priority"] == "高": return "human" return "auto" def route_after_human(state: TicketState) -> str: if state.get("human_result") == "resolved": return "notify" return "archive"

这里有个细节：route_after_human里用了state.get("human_result")，而不是直接state["human_result"]。因为人工处理的结果可能还没写回状态，用 get 更安全，避免 KeyError。

4.4 人工介入与检查点配置

人工介入是这套系统的核心难点。我的实现方式是：在转人工节点后设置中断，流程暂停，等外部系统写入human_result后再恢复。

from langgraph.checkpoint.sqlite import SqliteSaver memory = SqliteSaver.from_conn_string("tickets.db") builder = StateGraph(TicketState) builder.add_node("receive", receive_ticket) builder.add_node("classify", classify_ticket) builder.add_node("priority", judge_priority) builder.add_node("auto_reply", auto_reply_node) builder.add_node("human", human_node) builder.add_node("notify", notify_node) builder.add_node("archive", archive_node) builder.set_entry_point("receive") builder.add_edge("receive", "classify") builder.add_edge("classify", "priority") builder.add_conditional_edges( "priority", route_by_priority, {"human": "human", "auto": "auto_reply"} ) builder.add_edge("auto_reply", "notify") builder.add_conditional_edges( "human", route_after_human, {"notify": "notify", "archive": "archive"} ) builder.add_edge("notify", "archive") builder.add_edge("archive", END) graph = builder.compile( checkpointer=memory, interrupt_before=["human"] )

interrupt_before=["human"]表示在执行 human 节点之前暂停。这时候你可以通过 API 查询当前状态，人工处理后把结果写回去，再调用graph.invoke(None, config)继续执行。

提示：中断点的位置很关键。放在节点前还是节点后，取决于你的业务需求。如果人工处理需要参考 AI 的中间结果，就放在节点前；如果人工只是确认 AI 的结果，可以放在节点后。

5. 调试、优化与常见问题排查

5.1 可视化整张图

LangGraph 提供了图的可视化功能，调试的时候非常有用：

from IPython.display import Image, display try: display(Image(graph.get_graph().draw_mermaid_png())) except Exception: print("可视化需要额外依赖，可跳过")

如果跑不通，也可以直接打印图的节点和边信息：

print(graph.get_graph().nodes) print(graph.get_graph().edges)

我习惯在开发阶段把图打印出来，对照业务流程图检查有没有漏掉的边或者多余的分支。有时候条件路由的映射写错了，图上看不出来，但一跑就出问题。

5.2 状态丢失与字段覆盖问题

这是新手最容易踩的坑。LangGraph 默认的状态合并策略是“后写覆盖”，如果两个节点都修改了同一个字段，后面的会覆盖前面的。如果你希望某个字段是累加的（比如消息列表），需要自定义 reducer。

from typing import Annotated from operator import add class State(TypedDict): messages: Annotated[list, add] count: int

Annotated[list, add]表示这个字段用 add 函数合并，也就是列表拼接。不加这个注解的话，后一个节点的列表会直接替换前一个，导致历史消息丢失。我当初就是因为没注意这个，调试了半天才发现消息列表总是只剩最后一条。

5.3 循环与死锁的预防

LangGraph 允许图里有循环，这很强大，但也很危险。如果条件路由写错了，流程可能在两个节点之间无限循环，把 token 烧光。预防措施有三个：一是设置最大迭代次数，二是加日志监控，三是在路由函数里加兜底逻辑。

def safe_route(state: State) -> str: if state.get("retry_count", 0) > 5: return "fallback" return "normal"

另外，LangGraph 的recursion_limit参数可以限制整个图的执行步数，默认是 25。如果你的流程比较复杂，可以适当调大，但不要调得太大，否则出问题的时候很难发现。

5.4 常见问题速查表

这张表是我在实际项目中遇到过的典型问题，基本上覆盖了 80% 的调试场景。遇到问题先对照这张表排查，能省不少时间。

6. 从能跑到好用：生产环境优化建议

6.1 状态持久化的选型对比

开发阶段用 SQLite 足够了，但生产环境建议换 PostgreSQL。SQLite 在并发写入的时候容易锁表，工单系统这种多用户场景下会出问题。PostgreSQL 的检查点方案更稳定，而且支持更复杂的查询，方便你做数据分析和监控。

选型的时候不要一味追求性能，稳定性和可维护性更重要。我见过有人为了追求速度用 Redis 做检查点，结果没配持久化，服务重启后所有进行中的工单全丢了，损失惨重。

6.2 节点粒度的取舍

节点拆得太细，图会变得很复杂，调试困难；拆得太粗，单个节点逻辑太重，复用性差。我的经验是：一个节点只做一件事，但这件事的粒度要适中。比如“调用大模型生成回复”可以是一个节点，但“拼接 prompt、调用模型、解析结果”就不应该拆成三个节点，因为它们是紧密耦合的。

判断标准很简单：如果两个逻辑总是一起出现，不会单独使用，那就合并成一个节点；如果某个逻辑可能被其他流程复用，那就拆出来。

6.3 日志与监控的接入

生产环境一定要加日志。LangGraph 本身不提供日志功能，需要你自己在节点里埋点。我的做法是封装一个装饰器，自动记录每个节点的进入时间、退出时间、状态变化。

import functools import logging logger = logging.getLogger(__name__) def log_node(func): @functools.wraps(func) def wrapper(state): logger.info(f"进入节点：{func.__name__}") result = func(state) logger.info(f"退出节点：{func.__name__}，更新字段：{list(result.keys())}") return result return wrapper

这个装饰器用起来很简单，加在节点函数上就行。日志输出到文件或者日志系统，出问题的时候一查就知道哪个节点耗时过长、哪个字段被意外修改。

7. 我踩过的坑与实操心得

第一个坑是状态字段的默认值。TypedDict 不会自动给字段设默认值，如果某个节点读取了一个还没被赋值的字段，会直接 KeyError。我的解决办法是在图的入口节点里把所有字段都初始化一遍，哪怕赋个空值也行。这样后续节点就不用担心字段不存在的问题。

第二个坑是条件路由的返回值。路由函数必须返回映射字典里存在的 key，否则 LangGraph 会报错。我建议在路由函数里加一个兜底分支，返回一个默认的 key，避免因为意外情况导致整个流程崩溃。

第三个坑是检查点的清理。检查点会一直累积，时间长了数据库会变得很大。我建议定期清理过期的检查点，比如只保留最近 30 天的记录。LangGraph 本身没有提供清理工具，需要自己写 SQL 或者脚本处理。

第四个坑是中断恢复时的状态一致性。人工处理期间，如果外部系统修改了状态，恢复执行的时候可能会冲突。我的做法是在中断点前把关键状态快照保存下来，恢复的时候对比一下，如果有变化就重新走一遍流程，而不是直接恢复。

这些经验都是我在实际项目中一点点积累的，文档里不会写，但真正做项目的时候一定会遇到。希望这些分享能帮你少走一些弯路。

最后再分享一个小技巧：LangGraph 的图定义和节点实现最好分开文件管理。图定义放在一个文件里，节点函数放在另一个文件里，这样修改节点逻辑的时候不会影响图的结构，测试的时候也方便单独导入节点函数。项目大了之后，这个习惯能帮你省很多事。