LangGraph生产实战：Checkpoint持久化与条件分支编排Agent工作流

为什么生产级Agent需要图编排？

顺序调用LLM的"线性Agent"只能应付简单任务。一旦涉及多步骤决策、人工审批、并行子任务，代码就会变成一团乱麻。LangGraph把工作流建模成一张有向图：节点是"做什么"，边是"下一步做什么"，状态对象在节点间流转。它解决三个生产级痛点：流程可观测、执行可恢复（Checkpoint）、分支可控制。

一、最小可运行示例：从链式调用到图

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

class AgentState(TypedDict):
    question: str
    draft: str
    reviewed: bool

def draft_node(state: AgentState) -> dict:
    # 模拟LLM生成初稿
    return {"draft": f"关于「{state['question']}」的初稿内容"}

def review_node(state: AgentState) -> dict:
    # 模拟人工/LLM审核
    return {"reviewed": True}

# 构建图
g = StateGraph(AgentState)
g.add_node("draft", draft_node)
g.add_node("review", review_node)
g.add_edge("draft", "review")
g.add_edge("review", END)
g.set_entry_point("draft")

app = g.compile()
result = app.invoke({"question": "如何做RAG评测"})
print(result["draft"])

二、条件分支：让Agent自己决定下一步

审核不通过要打回重写，这是最常见的业务分支。用 add_conditional_edges 实现：

def route_after_review(state: AgentState) -> str:
    # 模拟审核结论：最多重写2次
    if state.get("rewrite_count", 0) < 2:
        return "draft"   # 回到草稿节点
    return END

g.add_conditional_edges(
    "review",
    route_after_review,
    {"draft": "draft", END: END}
)

关键设计：路由函数只返回下一个节点的名字，逻辑必须纯函数化（不调LLM、不写外部状态），否则无法调试。

三、Checkpoint持久化：崩溃恢复与人工审批

Agent跑一半进程挂了怎么办？Checkpoint把每一步的状态快照存进数据库，重启后从断点继续：

from langgraph.checkpoint.sqlite import SqliteSaver

with SqliteSaver.from_conn_string("checkpoints.db") as saver:
    app = g.compile(checkpointer=saver)

    # 第一次执行：到review节点停下等人工审批
    config = {"configurable": {"thread_id": "order-001"}}
    app.invoke({"question": "生成报价单"}, config)

    # 人工审批通过后，从断点继续执行
    app.invoke(None, config)  # 传入None表示沿用已有状态

四、并行节点：Fan-out/Fan-in模式

批量处理场景（如同时检索多个数据源）用并行节点：

from langgraph.graph import add_messages

g.add_node("search_web", search_web_node)
g.add_node("search_db", search_db_node)
g.add_node("merge", merge_node)

# web和db两个检索节点并行执行
g.add_edge("draft", "search_web")
g.add_edge("draft", "search_db")
g.add_edge("search_web", "merge")
g.add_edge("search_db", "merge")

并行节点共享状态时，用 Annotated[list, add_messages] 声明累加语义，避免互相覆盖。

五、生产落地建议

2. 状态只放必要字段：大对象（如长文本）放外部存储，状态里只放引用ID

3. 路由函数写单测：条件分支是最容易出bug的地方，把路由函数抽出来单独测试

4. 配好Tracing：LangSmith或自建日志，记录每个节点的输入输出与耗时

总结：LangGraph不是银弹，但当你需要"可恢复、可分支、可并行"的复杂工作流时，图编排比手写状态机更省心，也比裸调LLM更可控。