企业官网建设流程全解析

一、引言：为什么"一把梭"的 Agent 走不远

过去一年，绝大多数人写 Agent 的起点都是一样的：一个 while 循环，把历史消息塞进 prompt，让模型自己决定下一步调什么工具，直到它认为任务完成。

这套 ReAct 式的写法在"查个天气"“总结一段文本"这类任务上表现尚可，但一旦任务变复杂——比如"调研某个技术方向的现状并写一份带引用的报告”——它就会以三种方式崩溃：

第一，上下文爆炸与目标漂移。所有中间结果都堆在同一个 messages 数组里，第 20 轮时模型已经看不清原始目标，工具返回的大段文本把关键指令挤到了注意力边缘。

第二，失败不可恢复。跑到第 15 步时网络抖动、API 限流、进程被 kill，一切从头再来。重跑三次的成本可能比人工做一遍还高。

第三，无中间校验点。Agent 在第 3 步就走错了方向，但因为没有任何"停下来看一眼"的机制，它会沿着错误路径一路狂奔到底。

1.1 一个真实的失败现场

把上面三条抽象结论落到具体数字上，会更触目惊心。我们曾经用 ReAct 循环跑过一个任务：“调研向量数据库在金融风控场景的落地现状，输出一份带来源引用的报告”。整个任务的真实执行轨迹大致是这样：

注意三个细节：第一，目标漂移是渐进的，第 6 步时模型已经开始丢约束，但没有任何机制能在此刻提醒它；第二，重复劳动是隐性的，第 9 步和第 3 步做的事高度重合，因为循环里没有"任务账本"这种东西；第三，崩溃点在第 15 步，但重跑要从第 1 步开始，因为所有中间产物只活在那个进程的内存里。

1.2 问题的本质

问题的本质是：我们把"控制流"完全交给了模型，而把"状态"丢给了无结构的消息列表。LangGraph 的价值就在于把这两件事重新夺回来——用图结构显式定义控制流，用带 reducer 的 State 显式定义状态，用 checkpointer 把状态持久化到外部存储。本文就从这三个支点出发，给出一个可中断、可恢复、可回溯的 Agent 实战实现。

换句话说，ReAct 循环把"该做什么"“做到哪了”"做错了怎么办"这三件事全部塞进了一个不断增长的字符串里，然后指望模型在每一轮都重新理解它。而 LangGraph 的做法是：该做什么交给图（节点 + 边），做到哪了交给 State（带 reducer 的类型化字段），做错了怎么办交给 checkpointer + interrupt（持久化 + 人工闸门）。

二、核心原理：LangGraph 的三个支点

在进入代码之前，先补一节执行模型，后面的所有设计都建立在它之上。

LangGraph 的运行时继承自 Google 的Pregel / BSP（Bulk Synchronous Parallel）模型：图的执行被切分成一个个superstep（超步）。在一个 superstep 内，所有被激活的节点并行执行；它们各自返回状态增量；框架等到本 superstep 内所有节点都完成之后，才用 reducer 把这些增量统一合并进全局状态，然后据此决定下一个 superstep 激活哪些节点。

这个"同步屏障"（barrier）语义解释了两件事：其一，为什么并行分支的结果可以通过operator.add安全汇聚——它们不会交错写入；其二，为什么 checkpoint 的粒度天然是"每个 superstep 结束后"，因为这正好是状态处于一致快照的时刻。

2.1 状态即契约：StateGraph 与 Reducer

LangGraph 里的 State 是一个TypedDict，它不只是"数据容器"，而是节点之间的契约。每个节点返回一个 dict，框架用对应的 reducer 把它合并进全局状态。

关键点在于Annotated[T, reducer]：

• 不写 reducer：默认是覆盖（后写的赢），适合plan、report这类单值字段；
• Annotated[list, operator.add]：追加，适合并行节点各自产出结果的场景；
• Annotated[list, add_messages]：按 message id去重合并，是对话类状态的标配。

Reducer 写错是新手最常见的坑：并行分支都返回findings却不加operator.add，最后只有一个分支的结果活下来，而且不报错。

# ❌ 错误：没有 reducer，两个并行分支会互相覆盖classBadState(TypedDict):findings:list[dict]# 结果：findings 长度恒为 1，但你不会收到任何警告# ✅ 正确：追加语义classGoodState(TypedDict):findings:Annotated[list[dict],operator.add]

这里还有一个容易被忽视的点：reducer 是按字段生效的，不是按 State 整体生效的。同一个 State 里完全可以混用覆盖字段和追加字段，而字段之间的语义区分正是设计的核心——哪些是"当前决策"（覆盖），哪些是"历史积累"（追加），想清楚这一点，状态设计就完成了一半。

对于自定义合并逻辑（比如按id去重、保留最新版本），reducer 就是一个普通函数，签名是(left, right) -> merged：

defmerge_findings(left:list[dict],right:list[dict])->list[dict]:"""按 id 去重，重复时以新值覆盖旧值"""merged={f["id"]:fforfinleft}forfinright:merged[f["id"]]=freturnlist(merged.values())findings:Annotated[list[dict],merge_findings]

这个自定义 reducer 在"重试某一步"的场景里非常有用：重跑researcher时不会在列表里留下两份同 id 的结果。

2.2 任务拆解：Planner 与动态 Send

复杂任务的拆解有两种范式：

• 静态拆解（Plan-and-Execute）：先让 LLM 输出一个结构化的子任务列表，再按列表顺序或并行执行。优点是路径可预测、可人工审核、易于缓存；缺点是遇到"执行中才发现需要补充调研"时不够灵活。
• 动态拆解（ReAct / 图内路由）：根据当前状态用条件边决定下一跳，必要时回环。

两者的差异可以概括成一张表：

LangGraph 的真正威力在于把两者结合。SendAPI 允许条件边动态生成一批节点实例（map 阶段），每个实例带自己的 payload，结果再通过 reducer 汇聚（reduce 阶段）。这就是所谓的 map-reduce 模式，也是并行调研、批量代码审查等场景的标准解法。

deffan_out(state:ResearchState):# 返回 N 个 Send，框架就会实例化 N 个 researcher 节点# 每个实例只拿到自己的 task，而不是全量 statereturn[Send("researcher",{"topic":state["topic"],"task":t})fortinstate["plan"]]

需要特别强调的是Send与普通条件边的区别：普通条件边返回的是节点名（字符串），语义是"下一跳去哪"；Send返回的是**（节点名, 输入）的二元组**，语义是"以这份输入，再起一个该节点的实例"。前者控制路由，后者控制扇出。当你的计划列表有 5 项时，Send会创建 5 个独立的researcher调用，它们进入同一个 superstep 并行执行，完成后统一通过 reducer 合并——整个过程中你不需要写任何asyncio.gather或者线程池。

2.3 断点恢复：Checkpointer 与 thread_id

这是 LangGraph 相比手写循环最大的工程优势。编译图时传入checkpointer，框架会在每个 superstep 结束后序列化整个状态并落盘，键是(thread_id, checkpoint_id)。

• thread_id是你自己指定的会话/任务 ID，通常直接绑业务 ID（如order-8891）；
• checkpoint_id由框架生成，形成一条不可变的历史链。

于是"恢复"变成了一个极简操作：用同一个thread_id再次 invoke 即可，框架自动从最新的 checkpoint 续跑。进程重启、机器迁移、隔天再跑，都不影响。

常见的 checkpointer 实现有四种，选型建议如下：

关于"到底存了什么"，也需要有明确预期：checkpoint 序列化的是整个 State 字典，包括所有Annotated字段的累积值。这意味着如果findings里堆了几十篇长文摘要，checkpoint 的体积会随之膨胀。生产上常见的做法是：State 里只存摘要与引用 ID，原始长文本放到对象存储或向量库里，需要时按 ID 取回。这既控制了 checkpoint 体积，也避免了序列化大对象带来的延迟。

2.4 人机协同：interrupt 与 Command

真正的生产 Agent 必须有暂停点。interrupt(payload)在节点内部抛出中断信号，把当前状态冻结，控制权交还给调用方；调用方拿到__interrupt__里的人类可读信息后，用Command(resume=value)恢复，interrupt()的返回值就是value。

注意一个必须理解的行为：节点在恢复时会从函数开头重新执行，直到再次遇到interrupt()。因此interrupt之前的副作用（写库、发邮件、扣费）必须幂等，或者干脆挪到interrupt之后。

这条规则背后是 LangGraph 的恢复机制：它不去做"函数中间某一行"的存档，而是通过重放（replay）已经完成过的节点来重建执行位置。对于已完成的历史 superstep，框架直接使用 checkpoint 里记录的结果，不会真正重新执行节点；但对于当前正在中断的那个节点，它是从头重跑的。所以更精确的表述是：整个节点函数体会重跑，你要保证的是这段重跑是安全的。

一个实用的规避模式是把副作用外置：

defplanner(state):plan=llm_decompose(state["topic"])# 第一段：只做纯计算，可以安全重放decision=interrupt({"type":"plan_review","plan":plan})ifdecision.get("action")=="revise":plan=llm_replan(state["topic"],plan,decision["comment"])# 第二段：所有副作用放在 interrupt 之后audit_log.write({"topic":state["topic"],"final_plan":plan})return{"plan":plan,"revision":0}

另外，interrupt的 payload 必须是可 JSON 序列化的，因为它会随 checkpoint 一起落盘。把 ORM 对象、datetime、自定义类直接塞进去，会在序列化阶段直接报错——这是个非常高频的踩坑点，用str()/isoformat()提前转好即可。

三、实战：一个可中断、可恢复的深度调研 Agent

我们构建一个完整例子：给定主题 → 规划子任务 → 人工审核计划 → 并行调研 → 撰写报告 → 自评审 → 必要时回炉。

3.1 状态定义与节点

importoperatorfromtypingimportAnnotated,TypedDictfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.graph.messageimportadd_messagesfromlanggraph.typesimportSend,interruptfromlanggraph.checkpoint.sqliteimportSqliteSaverclassResearchState(TypedDict):topic:str# 原始目标，全程不变plan:list[dict]# [{"id": 1, "question": "..."}]findings:Annotated[list[dict],operator.add]# 并行结果，必须追加合并report:str# 单值，覆盖语义feedback:strrevision:intdefplanner(state:ResearchState):"""任务拆解 + 人工审核闸门"""plan=llm_decompose(state["topic"])# 返回 3~5 个可独立执行的子问题ifnotplan:raiseValueError("拆解失败，需人工介入")decision=interrupt({"type":"plan_review","topic":state["topic"],"plan":plan})ifdecision.get("action")=="revise":plan=llm_replan(state["topic"],plan,decision["comment"])# 注意：此处之后不要放非幂等副作用return{"plan":plan,"revision":0}deffan_out(state:ResearchState):"""动态生成并行调研节点，每个 Send 携带独立 payload"""return[Send("researcher",{"topic":state["topic"],"task":t})fortinstate["plan"]]defresearcher(state:dict):"""单个子任务的执行体：检索 + 摘要"""docs=web_search(state["task"]["question"])summary=llm_summarize(state["task"]["question"],docs)# 只返回增量，靠 reducer 合并；带上 id 便于后续排序与去重return{"findings":[{"id":state["task"]["id"],"summary":summary}]}defwriter(state:ResearchState):ordered=sorted(state["findings"],key=lambdax:x["id"])return{"report":llm_write(state["topic"],ordered,state.get("feedback",""))}defcritic(state:ResearchState):verdict=llm_critique(state["topic"],state["report"])ifverdict["pass"]:return{"feedback":""}return{"feedback":verdict["comment"],"revision":state["revision"]+1}defroute_after_critic(state:ResearchState)->str:ifstate["feedback"]andstate["revision"]<3:return"writer"returnEND

这段代码里有三个设计决策值得强调：

1. findings用operator.add，因为researcher会被并行实例化多次，覆盖语义会丢数据；
2. Send的 payload 是节点输入，只带该子任务需要的最小状态，避免每个并行分支都拿到全量历史；
3. route_after_critic显式设置回环上限（revision < 3），防止"永远改不好"的无限循环。

还可以补充第四个决策：revision用覆盖语义而不是追加语义。如果这里误写成Annotated[int, operator.add]，那么在并行分支合并时会出现难以预期的累加行为，而回环上限的判断就会失效——这个 bug 非常隐蔽，因为它在串行路径上表现完全正常。

3.2 组图、编译与断点恢复

builder=StateGraph(ResearchState)builder.add_node("planner",planner)builder.add_node("researcher",researcher)builder.add_node("writer",writer)builder.add_node("critic",critic)builder.add_edge(START,"planner")builder.add_conditional_edges("planner",fan_out,["researcher"])builder.add_edge("researcher","writer")builder.add_edge("writer","critic")builder.add_conditional_edges("critic",route_after_critic,{"writer":"writer",END:END})checkpointer=SqliteSaver.from_conn_string("research.db")graph=builder.compile(checkpointer=checkpointer)

这里有两处细节需要点明。

第一，add_conditional_edges("planner", fan_out, ["researcher"])的第三个参数是可能的目标节点列表。当我们用Send扇出时，所有实例的目标节点都是researcher，所以这里写["researcher"]。这个参数主要用于图的静态校验和可视化，写漏了不会立刻报错，但会让graph.get_graph().draw_mermaid()画出来的图缺边。

第二，researcher到writer用的是普通边。看起来是"多个 researcher 要触发多个 writer"，但实际上因为同步屏障的存在，所有并行 researcher 会在同一个 superstep 内跑完，合并状态后才进入下一个 superstep，所以writer只会被执行一次。这正是我们想要的：先 map 再 reduce。

下面是完整的调用侧，包含中断处理和恢复逻辑：

fromlanggraph.typesimportCommand config={"configurable":{"thread_id":"topic-rag-finrisk-001"}}inputs={"topic":"向量数据库在金融风控场景的落地现状","revision":0,"findings":[],"feedback":"","plan":[],"report":""}result=graph.invoke(inputs,config=config)# 循环处理人机协同：每次中断都拿给人类审核，恢复后继续跑while"__interrupt__"inresult:payload=result["__interrupt__"][0].valueprint(f"[需要人工审核]{payload['type']}")foriteminpayload["plan"]:print(f"{item['id']}.{item['question']}")decision={"action":"approve"}# 实际系统里来自前端/IM 卡片回调result=graph.invoke(Command(resume=decision),config=config)print(result["report"])

而"断点恢复"这个能力，用起来只有一个动作——什么都不做。假设进程在第 3 步被kill -9，第二天你只需要重新写一次同样的代码：

# 进程重启后，无需任何额外逻辑result=graph.invoke(None,config=config)

传None作为输入表示"不注入新输入，从断点继续"。框架会用thread_id="topic-rag-finrisk-001"去research.db里查出最新的 checkpoint，恢复到那个 superstep 的边界，然后继续往下执行。已经完成过的planner、已经跑完的 4 个researcher都不会重跑。

3.3 一次完整的运行时序

把上面这套代码跑起来，superstep 的推进过程大致是这样：

1. superstep 0：START → planner。LLM 拆出 4 个子问题，interrupt触发，状态落盘，控制权返回调用方。
2. 人工审核：调用方拿到 payload，人类点"approve"（或带上 comment 点"revise"）。
3. superstep 1：Command(resume=...)恢复，planner从函数开头重放（很便宜，因为llm_decompose的结果可以从 checkpoint 里复用，或者你也可以把结果缓存在 State 里避免重调），返回plan。checkpoint 写入。
4. superstep 2：fan_out生成 4 个Send，4 个researcher并行执行。每个各自检索、摘要，返回{"findings": [...]}。屏障等待全部完成后，operator.add把 4 条结果合并成 4 元素列表。checkpoint 写入。
5. superstep 3：writer按 id 排序，生成初稿报告。checkpoint 写入。
6. superstep 4：critic评审，返回feedback和自增的revision。
7. superstep 5：route_after_critic判断——若有反馈且revision < 3，回到writer；否则结束。

注意第 4 步里"屏障等待"的重要性：如果框架不做同步，writer可能在只有 2 条 findings 落地时就被触发，报告就会缺一半内容。这种"部分数据"的 bug 在异步代码里极难排查，而 LangGraph 用执行模型从根上规避了它。

3.4 状态回溯与时间旅行

checkpointer 带来的另一个能力常被低估：历史状态可枚举、可回放、可分叉。

# 列出这条 thread 的所有历史快照（从最新到最旧）forsnapingraph.get_state_history(config):print(snap.config["configurable"]["checkpoint_id"],snap.next,# 该快照之后将要执行的节点len(snap.values["findings"]))# 状态内容的可观测指标

拿到某个历史checkpoint_id之后，你可以做两件事：

# 1) 从历史某一点重新向前跑（回放）history=list(graph.get_state_history(config))old_config=history[-3].config# 往前数第 3 个快照replayed=graph.invoke(None,config=old_config)# 2) 修完状态后另起一条新分支（分叉，不影响原 thread）graph.update_state(old_config,{"plan":corrected_plan})forked=graph.invoke(None,config={"configurable":{"thread_id":"topic-rag-finrisk-001","checkpoint_id":new_checkpoint_id,}})

这套能力在生产上至少有三种用法：事故复盘（把出错那一步的完整 State 拉出来看，而不是靠日志猜）、A/B 调试（同一份历史状态，换一版 prompt 重跑，对比结果）、人工纠偏（发现第 2 个子问题拆错了，直接update_state改掉，从该点续跑，省掉前面全部成本）。

四、常见问题 FAQ

Q1：Send和普通条件边到底怎么选？

看你要的是"换一条路"还是"开多条路"。返回字符串是路由（一个节点执行一次），返回Send列表是扇出（同一节点执行 N 次）。两者可以在同一个条件边里混用，比如"若问题简单就直接写报告，若复杂就扇出调研"。

Q2：并行分支里能不能写同一个字段？

可以，但必须配 reducer，否则后写的会覆盖先写的。而且要注意：同一个 superstep 内对同一字段的并发写入，合并顺序是不保证的。所以如果你的 reducer 不满足交换律（比如"取第一个非空值"），结果就不确定。安全做法是让 reducer 具备交换律和结合律，比如追加、求和、按 key 合并。

Q3：State 能存自定义对象吗？

能，但有前提：checkpointer 要能序列化它。SqliteSaver/PostgresSaver走的是 JSON 或 msgpack，自定义类需要注册序列化器，否则会报错。实践中更推荐在 State 里只放基础类型和 dict/list，复杂对象放到外部存储，State 里存 ID。

Q4：图跑起来"卡住"了，可能是什么原因？

三种常见情况：一、有节点内部在等一个永远不来的interrupt恢复（去查graph.get_state(config).next是否符合预期）；二、条件边返回了不在目标列表里的节点名；三、某个并行分支抛异常但被吞掉了，导致屏障永远等不齐。第三种最隐蔽，建议给所有节点统一加日志埋点。

Q5：thread_id应该怎么设计？

直接绑业务主键，比如report-2024-q3-001、order-8891-v2。不要用 UUID 每次新建，否则每次重跑都是新会话，恢复能力形同虚设。如果需要"同一次调研的多轮对话"，可以在业务 ID 后加轮次后缀，但要注意这会切断跨轮次的自动恢复。

Q6：能不能中途修改 prompt 然后续跑？

可以，但要明白语义：图定义（节点函数）改了之后，用同一个thread_id恢复，已经完成的节点不会被重新执行，只有尚未执行的节点会用新逻辑。如果你希望前面几步也用新 prompt 重跑，需要用get_state_history找一个足够早的 checkpoint 分叉。

Q7：人工审核能不能做多个闸门？

可以，interrupt可以出现在任意多个节点里。但要注意每次interrupt都要能被调用方区分——建议在 payload 里带上type字段（如plan_review、report_approval），调用方据此决定走哪套审核 UI。

Q8：并行researcher里某个失败了怎么办？

默认行为是整个 superstep 失败、整图抛异常，但状态已落盘。恢复时可以在失败节点里加节点级重试策略（RetryPolicy），或者在researcher内部把单个子任务的失败包装成"降级 finding"（比如返回{"id": x, "summary": "", "error": "..."}），让流程带着部分结果继续走。选择哪种取决于业务：报告可以接受缺一节，扣款流程不行。

五、踩坑清单与优化建议

踩坑一：Reducer 缺失且不报错。再强调一遍。所有会被多个节点或并行分支写入的字段，问自己一句"这里是覆盖还是追加"。建议在 State 定义处写一行注释说明每个字段的合并语义，这是团队协作中最省事的防御。

**踩坑二：interrupt之前有副作用。

更多硬核网安与AI工具包，请扫码获取完整源码！
** 恢复时节点从头重跑，audit_log.write、send_email会被执行两次。要么把副作用移到interrupt