LangGraph实战：用状态图构建多步Agent工作流

📚 Agent全栈工程师：企业级知识库项目系列 · 第2/20篇

从LangChain到Kubernetes，20篇系统掌握AI Agent全栈开发

超越线性Chain，用StateGraph实现分支、循环与人工审批

上周我们说Agent全栈工程师要懂编排，这周来兑现承诺：用一个真实跑挂7次的故事，告诉你为什么线性Chain已经不够用，以及LangGraph的StateGraph是如何用一张有向图把"分支判断、循环重试、人工审批、断点续传"这四件Chain做不了的事，一次解决的。这是系列从"为什么"走向"怎么做"的关键转折。

我盯着监控告警发呆——一个跑了112秒的Chain，在第7步API调用上挂了。前6步白跑，5块钱tokens费用打水漂，用户在那头等着。我只能让它从头跑一遍。

这就是Chain的宿命：线性、无记忆、不可中断。直到我把它换成LangGraph的StateGraph。

LangChain官方2026 State of Agent Engineering报告里有个数据让我印象深刻：超过60%的Agent生产事故，都和状态管理有关。这不是模型问题，是工程问题。今天这篇文章就来啃下这块硬骨头。

📰 每日要闻速递

在进入正题前，先看看本周值得关注的几条要闻——它们都和今天文章的主题暗暗呼应。

• AI圈：Anthropic呼吁全球放缓AI开发，警告AI"自我改进"风险。这个发声背后是什么？是他们自己的Claude在代理化任务上能力实在太快。工程上的弦外之音很明显：Agent越能干，越需要可中断、可审批、可观测的编排框架。（36氪 2026-06-06）

• AI圈：豆包推出付费体系后，月活环比减少610万。免费红利结束，真金白银补吃隐性成本是个永恒话题——这也是为什么企业内部应用最终都要走向"可控成本"的架构。（36氪 2026-06-06）

• 财经：本周华尔街重要评级汇总：科尔士、TripAdvisor、AT&T、联合健康、特斯拉。特斯拉被评级调整背后是Robotaxi业务预期，这与Agent商业化路径高度同构——一旦自动化系统进入生产，可解释性和审计就是命门。

• 财经：5月非农就业报告即将发布，今年开头超预期的就业创造可能面临现实检验。对于考虑AI招聘市场热度的读者是个重要信号。（CNBC Markets）

• 业内：2026 Apple设计奖揭晓，12款App入选。背后一个趋势：获奖App多数在其重点交互路径中集成了Agent能力，"Agentic UX"已经从论文走进产品。（少数派）

这些表面看似不相干的新闻，背后是同一个叙事：Agent不再是"demo"，它是生产系统。而生产系统需要可控性。

第一章 为什么Chain不够用：线性执行的三宗罪

先讲清楚一件事：Chain没有错，是它解决的问题域被读者错配了。Chain天生适合"输入→处理→输出"的线性任务：翻译、摘要、单轮问答都是它的主场。但当你要构建企业级知识库Agent时，任务结构变了：需要看质量判断走哪里、需要重试、需要人工审批。这时Chain的三宗罪就暴露了。

罪一：无分支判断。Chain是"A→B→C"的硬编码路径。你可以用RouterChain勉强实现分支，但那是"在节点内部"的分支，不是"节点之间"的拓扑分支。二者差别在于：前者是一个function返回不同值；后者是整个执行拓扑变了。

罪二：无循环重试。Chain某一步失败，你只能用retry装饰器包住该节点。但如果失败原因是"上游输出质量不够"呢？你需要"跳回上游重新调用"，这在线性结构里是个死循环。

罪三：无中断恢复。这是最致命的。开头那个跑挂7次的故事里，Chain的任何中间状态都不被持久化。进程重启、服务重部、OOM被杀——上下文全部丢失，只能从头走。在成本问题上这是个灾难：LLM调用贵在重复，重复贵在架构错了。

看过一篇《断点续传让Agent恢复耗时降低80%+》的实践报告，里面重点讲了"LangGraph的Checkpoint机制让他们的文档处理pipeline从「重启=重跑」变成「重启=接上上次的状态」"，耗时从平均112秒降到20秒以内。这不是微优化，是架构变身。

这三宗罪连起来就是一件事：企业级Agent的本质不是"调用LLM"，是"编排一个可靠的工作流"。而工作流本质上是一个状态机。这也是Chain的名字为什么叫"Chain"——它是链，不是图。

第二章 StateGraph核心四件套

LangGraph的核心是StateGraph——一个有向图状态机。它只有四个概念，但这四个概念组合出了Chain做不到的一切。

① State（状态）：一个共享的TypedDict，所有节点都读它、写它。是"图的血液"。

② Node（节点）：一个函数，输入State，输出State的部分更新。是"图的背肌"。

③ Edge（边）：节点之间的连接。可以是硬边（A一定到B）。

④ Conditional Edge（条件边）：一个路由函数，读当前State，决定下一个节点是谁。这是Chain跟Graph拉开差距的结构。

一场"文档入库"示例讲清四个概念。设想一个最简状态图：接收文档→LLM抽取结构化信息→检查置信度→高置信走入库，低置信走入人工审批。

# 第一步：定义状态 - 这是「血液」
from typing import TypedDict
from langgraph.graph import StateGraph, ENDclass DocState(TypedDict):
raw_text: str
extracted: dict
confidence: float
needs_review: bool
final_status: str

状态定义后，其它节点不需要"传参数"——谁需要什么自己从状态里拿，变更也是覆写状态。这个"取取放放"的模型比Chain里"函数递归传参"的模型后期可维护性高一个量级。

# 第二步：定义节点函数
def extract_node(state: DocState):
result = llm.invoke(state["raw_text"])
return {
"extracted": result.data,
"confidence": result.score
}def check_node(state: DocState):
threshold = 0.85
return {"needs_review":
state["confidence"] < threshold}

重点来了——Conditional Edge让文档走不同路径：

# 第三步：路由函数（关键！）
def route_after_check(state: DocState):
if state["needs_review"]:
return "human_review"
return "write_to_db"# 第四步：装配图
builder = StateGraph(DocState)
builder.add_node("extract", extract_node)
builder.add_node("check", check_node)
builder.add_node(
"human_review", hitl_node)
builder.add_node("write_to_db", db_node)builder.set_entry_point("extract")
builder.add_edge(
"extract", "check")
builder.add_conditional_edges(
"check", route_after_check)
builder.add_edge(
"write_to_db", END)
graph = builder.compile()

这里有个拐点要记住：add_conditional_edges接收的路由函数返回的是节点名字，不是节点本身。这让整个图的路径变成运行时决定的，而不是定义时硬编码的。为什么这个细节重要？因为它意味着LLM本身可以决定走哪里——这才是Agent，而不是工作流。

第三章 实战：构建带人工审批的文档处理工作流

把第二章的骨架填满，做一个能上生产的"文档入库 → LLM抽取 → 质量检查 → 人工审批 → 写入向量库"工作流。这是企业级知识库的核心管线。

先画清楚整个流程的拓扑（这是LangGraph相比Chain最值得展示的地方——架构图能直接对应代码）：

📊 文档处理工作流拓扑

START

↓

ingest（文档入库）

↓

llm_extract（LLM抽取）

↓

quality_check（质量检查）

↓ Conditional Edge

human_review（人工审批）

|

write_vector_db（写入向量库）

↓

END

低置信文档自动走人工审批分支，高置信直接写库

第一步：扩展状态定义。真实场景下State要承载更多元数据：

from typing import TypedDict, List
from datetime import datetimeclass DocPipelineState(TypedDict):
doc_id: str
raw_text: str
chunks: List[str]
extracted_meta: dict
confidence: float
review_decision: str
error_count: int
final_status: str
started_at: str

第二步：实现核心节点。每个节点都是纯函数，便于单元测试：

def ingest_node(
state: DocPipelineState):
# 文档读取与切片
chunks = chunker.split(
state["raw_text"])
return {
"chunks": chunks,
"started_at":
datetime.now().isoformat()
}def llm_extract_node(
state: DocPipelineState):
prompt = build_extraction_prompt(
state["chunks"])
result = llm.with_structured_output(
DocMeta).invoke(prompt)
return {
"extracted_meta": result.dict(),
"confidence": result.confidence
}

第三步：质量检查与路由。这是整个工作流的"决策中心"：

def quality_check_node(
state: DocPipelineState):
meta = state["extracted_meta"]
issues = []
if not meta.get("title"):
issues.append("missing_title")
if len(meta.get(
"keywords", [])) < 3:
issues.append("too_few_keywords")
return {"quality_issues": issues}def route_quality(
state: DocPipelineState):
if state["confidence"] >= 0.9 \
and not state.get(
"quality_issues"):
return "write_vector_db"
return "human_review"

这个分支判断的精髓在于：它不只看LLM的自评分（confidence），还结合规则化的quality_issues。LLM自评是参考，规则是底线。这就是"代理化"和"工程化"的边界——把可规则化的部分交给规则，把不可规则化的部分交给LLM。Anthropic在最近一份技术博客里也强调过类似观点：结构优于访问，技能维护是工程问题。

第四章 Checkpointing：让Agent"断点续跑"

现在回到开头那个"跑挂7次"的故事——LangGraph的Checkpoint机制是怎么把它解掉的。

原理很简单：每次节点执行结束，自动把State快照写到持久化存储。流程中断后，从最近一个checkpoint恢复，跳过已执行的节点。这就把Agent的"重启=重跑"变成了"重启=接上"。

从开发到生产，Checkpoint有三档配置：

▸ 开发期：InMemorySaver — 内存里存，进程退出就丢，单元测试用。

▸ 单机生产：SqliteSaver — 本地SQLite文件，进程重启不丢数据。

▸ 集群生产：PostgresSaver — 多进程共享，支持高并发，企业首选。

# 生产配置：Postgres持久化
from langgraph.checkpoint.postgres \
import PostgresSaverDB_URI = "postgresql://user:pwd@" \
"localhost:5432/agent_state"checkpointer = PostgresSaver.from_conn_string(
DB_URI)
checkpointer.setup()  # 创建schemagraph = builder.compile(
checkpointer=checkpointer,
interrupt_before=["human_review"]
)# 用thread_id标识一次会话
config = {"configurable": {
"thread_id": "doc-2026-001"}}# 第一次运行：跑到human_review暂停
result = graph.invoke(
{"raw_text": doc}, config)# 进程崩溃也没关系，下次：
# 用同一个thread_id继续
result = graph.invoke(None, config)
# LangGraph自动从checkpoint恢复

这里的interrupt_before=["human_review"]就是HITL（Human-in-the-Loop）的入口：图执行到human_review节点前会暂停，把当前State持久化，等待人工接入。这是企业Agent上线的硬门槛——没有HITL就别谈生产环境。

还有一个被低估的特性：时间旅行调试。你可以遍历某个thread_id的所有checkpoint，回到任意历史状态重跑。这在排查"为什么这次执行走错了分支"时简直是神器：

# 列出所有历史checkpoint
for snapshot in graph.get_state_history(
config):
print(snapshot.config[
"configurable"][
"checkpoint_id"])
print(snapshot.next)
print(snapshot.values)# 回到某个历史点重跑
old_config = {"configurable": {
"thread_id": "doc-2026-001",
"checkpoint_id": "abc123"
}}
graph.invoke(None, old_config)

实测下来，加上Postgres Checkpoint后，最常见的两类问题——"进程被k8s重启导致执行中断"和"用户中途修改输入导致需要回滚"——都从架构层面消失了。这就是为什么KM那篇实践报告说恢复耗时降低80%+。

第五章 Human-in-the-Loop：AI暂停等人

第四章的interrupt_before是HITL的"入门款"。LangGraph还有更细粒度的interrupt()函数，让节点内部主动暂停。这一点对企业级场景至关重要——很多审批不是"前置审批"，而是"过程中决策点"。

from langgraph.types import interrupt
from langgraph.types import Commanddef human_review_node(
state: DocPipelineState):
# 把决策上下文交给人
decision = interrupt({
"doc_id": state["doc_id"],
"meta": state["extracted_meta"],
"confidence": state["confidence"],
"prompt":
"approve / reject / edit"
})
return {"review_decision": decision}# 前端拿到interrupt后渲染审批面板
# 用户决定后用Command恢复
graph.invoke(
Command(resume="approved"),
config)

interrupt()抛出的不是异常，是一个"暂停信号"：当前State自动持久化，前端拿到interrupt数据，渲染审批UI；用户决定后通过Command(resume=...)把结果送回，节点函数继续执行——就好像那个interrupt调用刚返回一样。

四种HITL经典模式，企业级Agent里基本都见过：

模式一 · 审批确认：低置信度结果暂停等审批，approve则继续，reject则回炉。

模式二 · 内容编辑：把LLM抽取结果交给人编辑，编辑后的内容回写到State继续往下走。

模式三 · 多选决策：LLM给出几个候选方案，让人选一个。最常见于工具选择、动作规划。

模式四 · 异常上报：遇到无法处理的情况主动停下来，附上上下文给人，避免Agent硬跑出错。

第六章 SubGraph与选型对比

SubGraph是LangGraph的"模块化机制"——当工作流变大时，把"文档解析""质量校验""索引写入"等子任务封装成可复用的子图，主图只关心编排逻辑。

# 子图：文档解析
parse_subgraph = build_parse_graph()# 子图：质量校验
validate_subgraph = build_validate_graph()# 主图：直接把子图作为节点
main = StateGraph(MainState)
main.add_node("parse", parse_subgraph)
main.add_node(
"validate", validate_subgraph)
main.add_edge("parse", "validate")

子图与主图共享父State的部分字段，其它字段独立。这种"局部状态隔离"在团队协作时特别值钱——子图作者不用关心主图所有字段，主图作者也不用关心子图实现细节。

那LangGraph在Agent生态里到底站哪儿？这是2026年企业选型的"灵魂拷问"。我整理了一张选型决策表，给读者带走：

一句话总结选型逻辑：LangGraph适合"我知道流程长什么样，需要可靠地把它跑出来"；CrewAI/AutoGen适合"我不知道流程长什么样，让Agent自己涌现"。企业知识库属于前者——所以这个系列后面18篇都会以LangGraph为底座。

一个值得引用的商业证据：Anthropic在《AI西经东译》EP109里讲他们的"代理化分析栈"，用Claude+结构化工作流让数据查询的95%自动化、95%准确率。他们的核心方法论是"分层架构（数据基础→事实来源→技能→验证）"——这与LangGraph强调的"有向图编排+持久化+评估"理念完全同构。这不是巧合，是企业级Agent正确路径的趋同。

结尾：从"AI能用"到"AI可用"

回顾这一篇，其实我们走过的是Agent工程的三层抽象：

第一层 · 调用：Chain时代，"我能让LLM跑起来"。

第二层 · 编排：StateGraph时代，"我能让多个LLM按图协作"。

第三层 · 治理：Checkpoint+HITL时代，"我能让生产Agent可中断、可审批、可观测"。

本周财经版块说5月非农就业要面临现实检验，AI圈Anthropic在呼吁放缓——这两件事都在提醒我们：技术正在快速进入"现实检验"阶段。能跑demo的人很多，能跑生产的人很少。LangGraph就是把这两者拉开差距的工具。

下周第3篇预告——《Vercel AI SDK前端接入：流式响应与Conversation UI实现》。我们已经把后端的状态图搭好了，下一步是让用户感觉它"快"——流式输出、思考过程可视化、中途取消，都是前端体验的硬指标。

📝 留一道思考题：

如果你给自家公司做知识库Agent，你会用LangGraph还是CrewAI？为什么？欢迎在留言区聊聊你的选型理由——上一篇有读者留言说"用CrewAI做产品演示，用LangGraph做生产部署"，这是我见过最务实的双轨策略。

📚 Agent全栈工程师：企业级知识库项目系列 · 第2/20篇

从LangChain到Kubernetes，20篇系统掌握AI Agent全栈开发

✅ 第1篇：LangChain入门：Chain/Agent/Tool三件套搭起第一个RAG管线

⏳ 第2篇：LangGraph实战：用状态图构建多步Agent工作流

⏳ 第3篇：Vercel AI SDK前端接入：流式响应与Conversation UI实现

⏳ 第4篇：Milvus向量数据库：从安装到百万级向量检索的工程化实践

⏳ 第5篇：文档解析与Chunking策略：如何把PDF/Word/网页切成高质量片段

⏳ 第6篇：Embedding模型选型：OpenAI/BGE/M3E对比与私有化部署

⏳ 第7篇：ElasticSearch全文检索：倒排索引+IK分词+BM25调优

⏳ 第8篇：混合检索RAG：多路召回+Reranker重排模型实战

⏳ 第9篇：Neo4j知识图谱：Graph RAG让大模型理解实体关系

⏳ 第10篇：RAG评估体系：从Ragas到自建量化评估Pipeline

⏳ 第11篇：LangSmith全链路观测：Agent调试与生产监控

⏳ 第12篇：Context压缩与Token优化：让长文档对话不爆上下文

⏳ 第13篇：Docker Compose本地开发全栈：前端+后端+向量库一键启动

⏳ 第14篇：多Agent协作架构：Supervisor/Router/Handoff模式设计

⏳ 第15篇：Tool Use高级模式：动态工具注册与权限控制

⏳ 第16篇：知识库权限与多租户：企业级数据隔离方案

⏳ 第17篇：对话记忆管理：短期记忆/长期记忆/会话摘要的工程实现

⏳ 第18篇：生产部署：Kubernetes+HPA弹性伸缩与灰度发布

⏳ 第19篇：RAG进阶：Self-RAG/Adaptive-RAG/Corrective-RAG

⏳ 第20篇：全栈整合：从0到1搭建企业级知识库Agent完整复盘

—— 本文是 Agent全栈工程师：企业级知识库项目 系列第2/20篇 ——

关注我，每周日更新，20篇带你打通Agent全栈