ByteFisher AI 工程化深度实战（七）：Agent 框架源码剖析——LangGraph

从本篇开始进入编排层。第 01 篇定下的选型决策树里，LangGraph 对应”复杂状态流、需持久化”的场景。本篇不满足于”会用 API”，而是把 LangGraph 的核心机制拆到源码层——理解一个框架的边界，比会用它的接口重要得多。

先看它解决的本质问题：一个多步骤 AI 工作流（比如”读文章→提取数据→写文件→校验”）如果只是写成一串 await，那它是一段程序；如果这个流程需要中途重启、断点续跑、分支回退、并发子任务、可视化追踪，你就需要一个把”状态”当成一等公民的引擎。LangGraph 就是这样一个基于图结构的状态机运行时。

一、核心抽象：StateGraph 是怎么工作的

1.1 数据模型

LangGraph 里，状态（State）就是一个 TypedDict，图的所有节点共享这份状态，节点函数输入整个 state、输出要更新的字段。关键在于状态字段可以声明归约器（reducer）——比如 messages 字段用 add_messages，那么每个节点往这个字段里添加的每条消息都会被追加，而不是覆盖。

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END, add_messages

class BlogState(TypedDict):
    article: str            # 文章原文
    fishing_data: dict      # 提取出的鱼获数据
    dashboard_updates: Annotated[list, operator.add]  # 累加，不覆盖
    messages: Annotated[list, add_messages]           # 消息流，追加

这个设计的精髓在 Annotated[list, operator.add]：reducer 定义了”状态如何更新”。没有 reducer 的字段默认是”覆盖”，有 reducer 的字段是”归约”。这正是状态机和普通变量的分水岭——状态更新是有规则、可并发、可回放的。

1.2 图 = 节点 + 边

graph = StateGraph(BlogState)

graph.add_node("extract", extract_fishing_data)   # 节点
graph.add_node("validate", validate_data)
graph.add_node("write", write_to_dashboard)

graph.add_edge(START, "extract")                   # 静态边
graph.add_edge("extract", "validate")
graph.add_conditional_edges("validate",            # 条件边
    lambda s: "write" if s["fishing_data"] else "extract")
graph.add_edge("write", END)

编译后 graph.compile() 生成一个可执行图。运行时按拓扑顺序执行节点：START → extract → validate →（条件）→ write → END。

二、运行时核心：Pregel 与超步（Superstep）

LangGraph 的运行时叫 Pregel，这是从 Google 分布式图计算系统 Pregel 借鉴的批量同步并行（BSP）模型。理解它，就理解了 LangGraph 一半的精髓。

超步 1：START 触发，活跃节点 = {extract}
超步 2：extract 完成，产出边激活 validate
超步 3：validate 完成，条件边判定 → write（或回到 extract）
超步 4：write 完成 → END，无活跃节点，图结束

关键特性：

通道（Channels）：节点间的通信不走共享内存，而是通过”通道”传递消息——这正是 Pregel 模型，好处是节点天然可并行、可分布式（不同超步的节点无共享可写状态）。
超步是同步屏障：每个超步内所有活跃节点并行执行，全部完成后统一提交状态，进入下一个超步。这保证了状态更新的一致性——不会出现”节点 A 读到节点 B 更新一半的状态”。
这就是为什么 reducer 必须可交换可结合：并行节点的状态更新通过 reducer 归约，顺序无关，结果一致。

对工程的意义：Pregel 模型让 LangGraph 天然支持并行节点（同一超步里多个节点并行跑 LLM 调用）和可恢复执行（每个超步是原子提交点）。这两点正是普通 async 串行代码做不到的。

三、检查点（Checkpointer）：断点续跑的秘密

3.1 为什么需要检查点

Agent 工作流可能跑几分钟、涉及多次 LLM 调用。一旦中途崩溃、进程重启、或用户手动中断，如果状态丢了，整个工作流重来——成本和体验都不可接受。检查点机制让图每一步执行后都持久化状态，可以从任意步骤恢复。

from langgraph.checkpoint.memory import InMemorySaver

saver = InMemorySaver()
graph = graph.compile(checkpointer=saver)

# 执行，thread_id 标识一次会话
config = {"configurable": {"thread_id": "diary-2026-08-02"}}
result = graph.invoke({"article": article_text}, config)

# 中途中断后，用同一个 thread_id 从断点继续
resumed = graph.invoke(None, config)  # 从上次停止处继续

3.2 源码层原理

BaseCheckpointSaver 定义了检查点读写接口：

1
2
3

put(config, checkpoint, metadata, new_versions)   # 写检查点
get_tuple(config)                                 # 读检查点
list(config, ...)                                 # 列出历史检查点

每个检查点保存三样东西：通道值（channel values）、节点状态（node state）、版本号（versions）。版本号是增量更新的关键——恢复时只重放”自上次检查点以来变化的部分”，而不是全量重跑。

生产部署用持久化存储（langgraph-checkpoint-sqlite / langgraph-checkpoint-postgres / Redis）替代 InMemorySaver，配合 thread_id 就能实现跨进程、跨机器恢复。对 ByteFisher 的钓鱼流水线来说：如果文章分析任务跑了一半，服务器重启后能接着跑，而不是从头再来。

四、流式执行：两种粒度的流

Agent 工作流最怕”黑盒转圈”。LangGraph 提供两层流：

方法	粒度	用途
`graph.stream(..., stream_mode="values")`	每步节点完成后的完整状态	展示”正在做什么”
`graph.stream(..., stream_mode="messages")`	LLM 逐 token	打字机效果

ByteBot 前端已经实现了 SSE 逐 token 渲染（handleStream），如果未来 ByteBot 升级成 LangGraph 编排，流式设计可以无缝对接：

1
2
3

for chunk in graph.stream(input, config, stream_mode="messages"):
    if chunk[0].content:            # token 级
        sse_write(chunk[0].content)

工程要点：流式必须和检查点结合——逐 token 推给用户的同时，每个超步写检查点，这样”用户看到一半刷新页面”后还能从检查点恢复，而不是整段重来。

五、实战：钓鱼日记鱼获数据提取 Agent

落地到博客的 tools/douyin-blog 流水线。现有流程里 finalize.py 负责把钓鱼日记发布后把数据嵌入博客，需要人工确认鱼获数据。我们用 LangGraph 做一个”提取 + 校验 + 写入”的 Agent，自动完成：

def extract_fishing_data(state):
    """从日记原文提取鱼获数据（LLM + 结构化输出）"""
    prompt = f"""
从这篇钓鱼日记中提取结构化数据：
{state['article'][:3000]}

输出 JSON：{{"spots":[], "species":[], "total_catch":0, "weather":""}}
"""
    data = call_llm_structured(prompt)   # 第03篇的 JSON Schema 方式
    return {"fishing_data": data}

def validate_data(state):
    """校验：鱼种是否在已知鱼种表里、数字是否合理"""
    data = state["fishing_data"]
    known = load_fish_species()          # 从 fish-dashboard.yml 读取
    bad = [s for s in data["species"] if s not in known]
    data["_invalid_species"] = bad
    return {"fishing_data": data}

def write_to_dashboard(state):
    """写入 fish-dashboard 数据文件（模拟）"""
    append_to_dashboard_yaml(state["fishing_data"])
    return {"status": "written"}

graph = StateGraph(BlogState)
graph.add_node("extract", extract_fishing_data)
graph.add_node("validate", validate_data)
graph.add_node("write", write_to_dashboard)
graph.add_edge(START, "extract")
graph.add_edge("extract", "validate")
# 校验不过就回到 extract 重新提取（最多重试 N 次）
graph.add_conditional_edges("validate",
    lambda s: "write" if not s["fishing_data"].get("_invalid_species")
               else ("extract" if s.get("retries", 0) < 3 else "write"),
    {"write": "write", "extract": "extract"})
graph.add_edge("write", END)

这个案例展示了 LangGraph 三个真实价值：

条件边让”提取失败→重试”成为图的一部分，而不是散落在代码里的循环。
检查点 + thread_id：提取到一半崩了，用同一个 thread_id 续跑。
状态可视化：graph.get_state(config) 随时能看当前跑到哪一步，对接第 11 篇的观测。

六、LangGraph 的边界与坑

诚实的结论——LangGraph 不是银弹，它有明确边界：

优势	代价
显式状态、可恢复、可并行	概念陡峭，Pregel/reducer/检查点学习成本高
复杂分支清晰	简单顺序流程用它是过度设计
可观测性好	每步都有序列化开销

坑：reducer 必须可交换可结合；messages 字段别手写 add_messages 的替代实现；invoke 和 stream 别混用同一 config 的检查点语义。多数”LangGraph 卡住”的问题，本质是把 async 串行代码的思路硬套到图上——先画图，再写代码。

七、落地清单与小结

LangGraph = 图结构状态机运行时：状态（TypedDict + reducer）+ 节点 + 边（静态/条件）。
运行时是 Pregel 超步模型：每步原子提交，天然支持并行与恢复。
检查点是断点续跑的基础，生产用 SQLite/Postgres 存储 + thread_id。
流式分两层：values（步级）和 messages（token 级），与 ByteBot 的 SSE 可对接。
适用边界：复杂状态流、需持久化；简单流程别硬上。
实战：钓鱼日记鱼获提取 Agent 展示了条件重试、检查点、状态可视化。

下一篇剖析 CrewAI：角色协作框架的源码结构，与 LangGraph 用同一个”文章分析”场景对比，看它在多角色协作上的取舍。

本系列持续更新中，每周一篇。