LangGraph 工作流：从基础调用到稳定运行

codeingg

331人浏览 · 2026-06-18 19:03:24

codeingg · 2026-06-18 19:03:24 发布

这篇我按“先跑起来、再讲取舍”的方式写《LangGraph 工作流：从基础调用到稳定运行》。概念会讲，但重点放在代码怎么组织、哪里容易踩坑。

摘要

本文概述文章目标、核心观点和实践价值。

> 摘要：过去大半年我带团队把多个基于 LLM 的脚本重构为生产级 Agent，踩过不少状态丢失、路由死循环和调试成本过高的坑。LangGraph 的价值不在于引入新语法，而是用有向图把“能跑通”变成“可追踪、可中断、可回滚”。本文结合一次客服工单自动流转的真实项目，拆解 State、Node、Edge 的设计取舍，重点讲如何加入人工审批、处理异常分支，以及这些经验该如何转化为简历里的硬通货。适合有后端基础、想摆脱 Prompt 拼凑式开发的开发者。

为什么需要图工作流

文章插图 1

早期我们接了一个内部工单处理需求，最初的做法是纯链式调用：接收用户输入 → 调大模型分类 → 匹配知识库 → 生成回复。测试阶段一切正常，但一上压力环境就暴露问题：用户中途追加描述、分类结果置信度低、工具调用超时，整个流程直接崩盘。代码里堆满了 `try-except` 和重试计数器，逻辑像一团乱麻。

这时候引入图工作流并不是为了赶时髦，而是解决两个实际问题：一是显式管理状态流转，二是把非确定性输出（LLM）和确定性执行（数据库、API）解耦。后端同学应该对状态机不陌生，比如 Spring 的 FSM 或者消息队列的消费者状态机。LangGraph 本质上就是把这套思想搬到 Agent 场景。

取舍也很明显：图结构前期设计成本高，你需要提前定义好所有可能的路径和状态字段；但它换来的是极高的可观测性和容错能力。当你能在图上看到一个节点失败时自动跳转到补偿节点，而不是让整个对话崩溃，这种架构升级就值得投入。

State 与 Node

文章插图 2

State 是整个图的唯一真相源。很多初学者喜欢在每个 Node 里直接修改全局变量或闭包字典，这在同步环境里勉强能跑，一旦涉及异步并发或断点续传，数据就会互相覆盖。正确的做法是把 State 当作不可变契约，Node 只负责返回增量更新。

from langchain_core.messages import HumanMessage, AIMessage
from langgraph.graph import StateGraph, MessagesState, START, END
import operator

class TicketState(MessagesState):
    # 自定义业务字段
    category: str = "unknown"
    confidence: float = 0.0
    approved: bool = False
    tool_calls: list = []

def classify_intent(state: TicketState):
    # 模拟调用 LLM 并提取结构化结果
    llm_output = {"category": "billing", "confidence": 0.85}
    return {
        "category": llm_output["category"],
        "confidence": llm_output["confidence"],
        "messages": [AIMessage(content=f"识别为{llm_output['category']}类问题")]
    }

builder = StateGraph(TicketState)
builder.add_node("classify", classify_intent)
builder.add_edge(START, "classify")

这里用了 `MessagesState` 作为基类，继承后追加业务字段。注意返回值必须是一个字典，LangGraph 会自动按字段类型合并（字符串覆盖、列表累加）。如果某个字段你希望保留历史记录，可以用 `operator.add` 作为 reducer 配置。踩坑提醒：不要在 Node 内部做耗时操作，图引擎默认会阻塞主线程，耗时逻辑建议拆成独立任务或异步回调。

CSDN资料领取方式

Edge 与条件分支

静态边适合确定流程，但 Agent 的本质就是动态路由。LangGraph 提供 `add_conditional_edges`，允许你根据上一个节点的输出决定下一步走向。很多人会把路由逻辑写死在 Node 内部，导致节点职责不清。更好的做法是抽一个独立的 Router 节点，只负责读 State 和返回目标节点名。

def route_by_confidence(state: TicketState):
    if state["confidence"] >= 0.9:
        return "resolve_auto"
    elif state["confidence"] >= 0.7:
        return "human_review"
    else:
        return "escalate_manual"

builder.add_conditional_edges(
    "classify",
    route_by_confidence,
    {
        "resolve_auto": "resolve_auto",
        "human_review": "human_review",
        "escalate_manual": "escalate_manual"
    }
)

这段代码看起来简单，但背后有个关键判断标准：路由条件必须来自可验证的信号（如置信度阈值、关键词命中、外部 API 返回值），而不是让 LLM 自己决定跳去哪里。LLM 擅长生成内容，不适合做精确的路由决策。把路由规则收口到代码层，后续改策略只需要改阈值，不需要重新微调模型。

人工审批节点

生产环境和 Demo 的分水岭往往在这里。模型会幻觉，工具会报错，合规要求敏感操作必须有人签字。LangGraph 提供了 `interrupt_before` 机制，可以在指定节点执行前暂停流程，等待外部信号恢复。

builder.add_node("approve", lambda s: {"approved": True})
builder.add_conditional_edges("human_review",
    lambda s: "approve" if s.get("manual_flag") == "pass" else "reject",
    {"approve": "approve", "reject": END})

# 配置检查点与中断
app = builder.compile(checkpointer=SqliteSaver(), interrupt_before=["human_review"])

实际使用时，我会把 `interrupt_before` 配合持久化存储（Postgres 或 Redis）一起用。Checkpoint 会快照当前 State，服务重启或进程崩溃都不会丢上下文。踩过的坑是：中断恢复后，UI 端经常拿不到最新的消息历史。解决办法是在恢复调用时显式传入 `{"configurable": {"thread_id": "...", "checkpoint_id": "..."}}`，确保图引擎加载正确的历史快照。

审批节点的 Payload 设计也很讲究。别直接把原始 JSON 扔给前端，Layman 看不懂。应该在 Node 里做一次格式化：提炼关键参数、展示风险等级、给出推荐操作。这步省下的沟通成本远大于多写几行模板代码的时间。

工程化落地

图跑通只是第一步，要进生产还得过这几道关：

1. **状态持久化选型**：开发期用 `SqliteSaver` 或内存缓存没问题，上线后必须切到关系型或 KV 存储。Checkpoint 体积会随对话轮次膨胀，建议定期清理旧版本或使用 TTL 策略。
2. **可观测性埋点**：不要只依赖 LangSmith。在关键 Node 入口/出口打日志，记录 `input_schema`、`output_delta`、`execution_duration`。遇到慢查询或幻觉高发区，直接看日志定位是哪个 Branch 出了问题。
3. **超时与降级**：图里每个节点都应该有独立的 Timeout 设置。大模型调用超时时，不要直接抛异常中断整张图，而是走 fallback Node 返回兜底话术，并在 State 里标记 `error_handling=True` 供下游判断。
4. **部署形态**：图本身是无状态的计算单元，真正的状态存在 Checkpointer 里。对外暴露 REST/gRPC 接口即可，横向扩展只要保证共享同一个存储后端。容器化时注意保留会话 ThreadId，避免负载均衡打散同一次对话。

写简历或做项目汇报时，别只写“使用了 LangGraph”。改成：“针对高频幻觉导致的工单误判，设计基于置信度的动态路由图，引入 SqliteSaver 实现断点续传与人工审批拦截，将线上人工干预率从 32% 降至 9%，单次请求平均耗时稳定在 1.2s 以内。” 数字和因果关系比技术名词更有说服力。

学习顺序也建议按这个节奏来：先跑通 `StateGraph` + 静态边 → 掌握 Reducer 和增量更新 → 接入 `add_conditional_edges` → 配置 Checkpointer 做断点保存 → 最后加 Human-in-the-loop。跳过前两步直接搞分布式图或复杂权限控制，调试成本会指数级上升。