Day28 LangGraph 复杂智能体编排

📖 基础知识讲解【基础原理 + 高级深挖知识点】

基础原理

1. 为什么需要 LangGraph：Day27 那个 while 循环到底撑不住什么

Day27 手写的最小 ReAct Agent，本质是这样一段代码：

python

while True:
    resp = llm.invoke(messages)
    if not resp.tool_calls:
        return resp.content
    messages.append(resp)
    for tc in resp.tool_calls:
        messages.append(tool_result(tc))

这段代码能跑通 80% 的 Demo，但一旦进生产就会在五个地方同时崩：

生产诉求	裸 while 循环的处境
执行到第 7 步时服务重启	全部状态在内存里，重启即丢，用户只能从头再问一遍
第 3 步是转账，要人工审批	循环没法"暂停在中间"，只能阻塞住整个线程等输入
同时查 RAG + 查数据库 + 查天气	要自己写线程池、自己合并结果、自己处理部分失败
产品要在前端画出"当前执行到哪一步"	循环内部没有可观测的节点概念，只能靠日志猜
回滚到第 2 步换个分支重跑	不可能，历史状态没有留存

一句话定位： LangGraph 不是"更高级的 Agent"，它是把 Agent 的控制流从 Python 代码里抽出来，变成一张可持久化、可中断、可回放、可并行的状态图。模型还是那个模型，工具还是那些工具，变的是谁来管流程。

和 LangChain 的关系也要说清楚，这是面试第一个送分题也是第一个送命题：

text

LangChain（LCEL Chain）         LangGraph
┌──────────────────────┐      ┌───────────────────────────────┐
│ prompt | llm | parser│      │  START → agent ⇄ tools → END  │
│ 单向、无环、一次走完   │      │  有向图，允许环、分支、并行      │
│ 状态 = 上一步的输出    │      │  状态 = 全局 State，节点共享     │
│ 中间状态不落盘         │      │  每个超步落 Checkpoint          │
│ 想暂停？没有这个概念    │      │  interrupt 原生支持人工介入      │
└──────────────────────┘   └───────────────────────────────┘

容易被问倒的边界：LangGraph 不依赖 LangChain 的 Chain 体系，它只是复用了 langchain-core 的 BaseMessage、BaseChatModel 这些基础抽象。你完全可以用 LangGraph + 裸 OpenAI SDK，不写一行 LCEL。所以"LangGraph 是 LangChain 的升级版"这个说法是错的，准确说法是：LCEL 解决数据流编排，LangGraph 解决控制流编排，二者正交。

2. 核心概念五件套：State、Node、Edge、Conditional Edge、Graph

text

                    ┌──────────── State（全局状态，贯穿整张图）────────────┐
                    │  messages / query / retrieved_docs / step_count ... │
                    └─────────────────────────────────────────────────────┘
                          ▲ 读            ▲ 读            ▲ 读
                              │ 写 patch      │ 写 patch      │ 写 patch
   START ──→ ┌──────────┐ ──→ ┌──────────┐ ──→ ┌──────────┐ ──→ END
             │  Node A  │     │  Node B  │     │  Node C  │
             └──────────┘     └──────────┘     └──────────┘
                   │                ▲
                   └── Conditional Edge（读 State 决定下一跳）

概念	是什么	代码形态	最容易搞错的点
State	全图共享的一份数据结构	`TypedDict` / `Pydantic BaseModel` / `dataclass`	它不是"节点间传参"，而是一块带合并规则（reducer）的共享内存
Node	一个执行单元	`def node(state) -> dict`	返回值是增量 patch，不是完整 State
Edge	固定跳转	`add_edge("a", "b")`	A 到 B 无条件执行；同一节点出多条普通边 = 并行
Conditional Edge	动态跳转	`add_conditional_edges("a", router, {...})`	路由函数只能读 State，不能改 State
Graph	图本身	`StateGraph(State)` → `.compile()`	必须 `compile()` 才能执行；编译期会校验孤立节点、不可达 END

python

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

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]   # 带 reducer：追加
    query: str                                 # 不带 reducer：覆盖
    retry: int

builder = StateGraph(AgentState)
builder.add_node("agent", agent_node)
builder.add_node("tools", tool_node)
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue, {"tools": "tools", "end": END})
builder.add_edge("tools", "agent")            # 回边，构成环 → 这就是 ReAct
graph = builder.compile()

START / END 是两个哨兵常量，不是真实节点。 set_entry_point("agent") 等价于 add_edge(START, "agent")，新代码统一用 START/END 写法，语义更清楚也更好画图。

3. State 的本质：带 reducer 的共享内存，不是"传参"

这是 LangGraph 最核心、也是新手 90% 的 bug 来源。节点返回的不是新 State，而是一份"写请求"（patch），框架拿 reducer 把它合并进全局 State。

text

节点返回 {"messages": [AIMessage(...)], "retry": 1}
                    │
                    ▼
       ┌──────── 合并阶段 ────────┐
       │ messages: 有 reducer      │ → old + new（追加）
       │           add_messages    │
       │ retry:   无 reducer       │ → 直接覆盖 old
       └──────────────────────────┘
                    ▼
            新的全局 State → 落 Checkpoint

声明方式	合并行为	适用场景
`query: str`	覆盖（last write wins）	单值字段：用户问题、当前意图、最终答案
`messages: Annotated[list, add_messages]`	按 id 追加/更新/删除	对话历史，唯一正确写法
`docs: Annotated[list, operator.add]`	列表拼接	多路检索结果汇总
`logs: Annotated[list, my_merge]`	自定义函数 `(old, new) -> merged`	去重、排序、取 TopK、RRF 融合

✅ 正确：返回增量

python

def agent_node(state: AgentState):
    resp = llm_with_tools.invoke(state["messages"])
    return {"messages": [resp]}          # 只返回新增的那一条

❌ 错误写法一：返回完整 State

python

def agent_node(state: AgentState):
    resp = llm_with_tools.invoke(state["messages"])
    state["messages"].append(resp)
    return state                          # 看起来"对"，实际埋了两个雷

雷在哪：① 原地 append 直接改了上一个 Checkpoint 的对象引用，时间旅行回放时拿到的是被污染的历史；② 整个 messages 列表又被 add_messages 走一遍合并 —— 这里侥幸不炸，是因为 add_messages 会按 message.id 去重覆盖，但换成 operator.add 的字段就会直接翻倍。

❌ 错误写法二：对无 reducer 的字段期望累加

python

class S(TypedDict):
    count: int                            # 没有 reducer

def n(state: S):
    return {"count": state["count"] + 1}  # ✅ 这样是对的（自己算好再覆盖）

def n_wrong(state: S):
    return {"count": 1}                   # ❌ 以为会累加，实际永远是 1

记住这句话：Annotated[X, reducer] 里的 reducer 是"写冲突裁决规则"。 串行执行时它决定"新旧怎么合"，并行执行时它决定"多个节点同时写同一个 key 怎么合"。没有 reducer 的 key 在并行场景下会直接抛 InvalidUpdateError —— 这是高级深挖第 1 条要展开的内容。

4. 节点函数的四条契约

python

def my_node(state: AgentState, config: RunnableConfig) -> dict:
    ...
    return {"some_key": value}

契约	说明	违反后果
入参 1 = state	框架注入当前全局状态（只读语义）	原地修改会污染 Checkpoint 历史
入参 2 = config（可选）	拿 `thread_id`、`callbacks`、自定义 `configurable`	不透传 config，LLM 的 token 流和 LangSmith 链路就断了
返回值 = dict / Command / None	dict 是 patch；`Command` 可同时改状态+跳转；`None` 表示不改状态	返回非法 key 时 `TypedDict` 不报错但静默丢弃，Pydantic State 会直接抛错
必须幂等	节点可能因为重试、中断恢复、超步失败而重复执行	扣款类节点重跑 = 重复扣款

同步节点与异步节点不能混搭：图里有 async def 节点，就必须用 await graph.ainvoke(...) / graph.astream(...)；用 graph.invoke() 会直接报错。反之全同步节点用 ainvoke 是允许的（框架丢线程池跑），但同步节点里的阻塞 IO 会卡住事件循环，这就回到 Day21 讲的 asyncio 那套结论。

5. 边的三种形态：固定边、条件边、并行边

python

# ① 固定边：无条件跳转
builder.add_edge("tools", "agent")

# ② 条件边：路由函数返回"键"，映射表决定去哪个节点
def should_continue(state: AgentState) -> str:
    last = state["messages"][-1]
    if getattr(last, "tool_calls", None):
        return "use_tool"
    return "finish"

builder.add_conditional_edges(
    "agent",
    should_continue,
    {"use_tool": "tools", "finish": END},   # ← 映射表，强烈建议显式写
)

# ③ 并行边：同一个源节点挂多条固定边 → fan-out
builder.add_edge("plan", "search_rag")
builder.add_edge("plan", "search_web")
builder.add_edge("plan", "search_db")
builder.add_edge("search_rag", "merge")      # fan-in
builder.add_edge("search_web", "merge")
builder.add_edge("search_db", "merge")

条件边两个高频坑：

❌ 坑	✅ 正确做法
路由函数返回了映射表里没有的键	要么补全映射表，要么不传映射表（此时返回值必须就是节点名本身）
在路由函数里改 State	路由函数的返回值只用于选路，对 State 的修改会被丢弃；要改状态就用 `Command`

路由函数还可以返回列表，表示同时走多个分支（动态 fan-out）：

python

def dispatch(state) -> list[str]:
    targets = []
    if state["need_rag"]:
        targets.append("search_rag")
    if state["need_web"]:
        targets.append("search_web")
    return targets or ["fallback"]

连完边一定要先自检图结构，再去调 LLM。编译后的图能把自己的拓扑打出来，这是排"为什么它不走我想的那条边"最快的手段：

python

print(graph.get_graph().draw_ascii())        # 终端直接看拓扑
print(graph.get_graph(xray=1).draw_ascii())  # xray=1 展开子图内部节点
print(graph.get_graph().nodes.keys())        # 核对节点名有没有拼错

自检要看什么	说明
条件边有没有连到 `END`	少了这条分支 = 永远循环，直到撞 `recursion_limit`
有没有不可达节点	节点名拼错时 `add_edge` 不报错，图上会出现一个孤岛
回边方向对不对	`tools → agent` 写反成 `agent → tools` 会变成两条平行边，直接并行执行
返回 `Command` 的节点连线	必须写 `-> Command[Literal["a", "b"]]` 注解，框架靠它推断连线，否则图上看不到这条边

这个项目的文档一律用 ASCII 图，所以这里推荐 draw_ascii()；本地调试想要图片可以用 draw_png()，但它依赖额外的渲染包。

6. 执行模型：超步（super-step）与 Pregel BSP

LangGraph 的执行引擎抄的是 Google Pregel 的 BSP（Bulk Synchronous Parallel）模型。理解这一点，并行、Checkpoint 时机、部分失败这三个问题就全通了。

text

超步 1                      超步 2                      超步 3
┌──────────────┐           ┌──────────────┐           ┌──────────────┐
│   plan       │           │ search_rag   │  并行      │   merge      │
│              │           │ search_web   │  同一超步  │              │
└──────────────┘           │ search_db    │           └──────────────┘
       │                   └──────────────┘                  │
       ▼                          ▼                          ▼
 ┌─ 屏障 barrier ─┐        ┌─ 屏障 barrier ─┐        ┌─ 屏障 barrier ─┐
 │ 合并所有 patch │        │ 用 reducer 合并 │        │ 合并 patch     │
 │ 落 Checkpoint  │        │ 三份 patch      │        │ 落 Checkpoint  │
 └────────────────┘        │ 落 Checkpoint   │        └────────────────┘
                           └─────────────────┘

结论	含义
同一超步内的节点并行执行	它们看到的是同一份超步开始时的 State 快照，互相看不到对方的写入
超步结束才合并、才落盘	所以 Checkpoint 的粒度是超步，不是节点
超步内任一节点抛异常	整个超步的写入不落盘，恢复时该超步的所有节点一起重跑 → 节点必须幂等
节点数 = 执行步数？不是	并行的 3 个节点只算 1 步，这直接关系到 `recursion_limit` 怎么算

面试加分表述：「LangGraph 的并行不是"开线程池跑任务"，而是 BSP 的超步并行 —— 同一超步的节点看不到彼此的中间状态，必须等屏障处用 reducer 统一合并。所以写并行节点时，绝不能假设'另一个分支已经把数据写进去了'，有依赖关系就必须拆成两个超步。」

7. 最小可运行 Graph：把 Day27 的 ReAct 图化

python

from typing import Annotated, TypedDict
from langchain_core.messages import SystemMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition

@tool
def get_weather(city: str) -> str:
    """查询指定城市的实时天气。city 传中文城市名，如 '北京'。"""
    return f"{city} 晴 26℃"

tools = [get_weather]
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools(tools)

class State(TypedDict):
    messages: Annotated[list, add_messages]

SYSTEM = SystemMessage("你是严谨的助手，信息不足时必须调用工具，不要编造。")

def agent(state: State):
    return {"messages": [llm.invoke([SYSTEM] + state["messages"])]}

builder = StateGraph(State)
builder.add_node("agent", agent)
builder.add_node("tools", ToolNode(tools))       # 预制件：自动执行 tool_calls
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", tools_condition)  # 预制件：有 tool_calls 就去 tools
builder.add_edge("tools", "agent")
graph = builder.compile()

out = graph.invoke({"messages": [("user", "北京天气怎么样")]})
print(out["messages"][-1].content)

两个预制件的隐含约定必须记住：

预制件	做了什么	隐含约定（踩坑点）
`ToolNode(tools)`	读最后一条 AIMessage 的 `tool_calls`，并发执行所有工具，把结果包成 `ToolMessage` 返回	State 里必须有名为 `messages` 的键；工具抛异常时默认把错误文本当 ToolMessage 返回给模型（`handle_tool_errors`）
`tools_condition`	有 `tool_calls` 返回 `"tools"`，否则返回 `END`	工具节点必须叫 `tools`，改名就要自己写映射表 `{"tools": "my_tool_node", END: END}`

8. 预制件 `create_react_agent` 与手写图的边界

python

from langgraph.prebuilt import create_react_agent

agent = create_react_agent(
    model=llm,
    tools=tools,
    prompt="你是严谨的助手……",          # 旧版本参数名是 state_modifier / messages_modifier
    checkpointer=MemorySaver(),
)
agent.invoke({"messages": [("user", "北京天气")]}, config={"configurable": {"thread_id": "u1"}})

场景	用 `create_react_agent`	手写 `StateGraph`
标准单 Agent + 工具循环	✅ 三行搞定	没必要
要在 LLM 前插入"消息裁剪/摘要"	可以，用 `pre_model_hook`	也行
State 里要加业务字段（订单号、权限、预算）	可以传自定义 `state_schema`	更直观
多 Agent、人工审批、map-reduce、子图	❌ 撑不住	✅ 必须手写

面试话术： 「预制件适合快速验证，但它本质就是一张固定形状的两节点图。只要流程里出现'第三个节点'，比如意图分类、审批、兜底、评测，就该直接手写 StateGraph —— 否则后面一定会为了绕开预制件的限制写出更难维护的代码。」

9. 终止条件：四道闸门，少一道都可能烧钱

Day27 讲过裸循环的四道闸门，在 LangGraph 里它们的落点变了：

闸门	LangGraph 落点	代码
模型主动收尾	条件边：没有 `tool_calls` → `END`	`tools_condition` 已内置
最大步数	`recursion_limit`（默认 25 个超步）	`graph.invoke(x, {"recursion_limit": 50})`
业务预算（token / 金额 / 时间）	自己在 State 里记账 + 条件边拦截	`if state["cost"] > BUDGET: return "force_answer"`
重复动作检测	State 里存 `tool_call` 指纹集合	命中就强制走 `force_answer` 节点

python

class State(TypedDict):
    messages: Annotated[list, add_messages]
    steps: int
    seen: Annotated[list, operator.add]      # 工具调用指纹

def should_continue(state: State):
    if state["steps"] >= 8:
        return "force_answer"              # ✅ 优雅收尾，而不是等框架抛异常
    last = state["messages"][-1]
    if not getattr(last, "tool_calls", None):
        return END
    fp = fingerprint(last.tool_calls)
    if fp in state["seen"]:
        return "force_answer"
    return "tools"

recursion_limit 是"兜底保险丝"，不是业务终止条件。 它触发时抛的是 GraphRecursionError，用户只会看到一个 500；而 force_answer 节点可以让模型基于已有信息给出"目前查到这些，建议人工核实"的降级答案。生产里一定要同时有这两层。

10. Checkpointer 与 thread_id：短期记忆到底存了什么

python

from langgraph.checkpoint.memory import MemorySaver          # 开发/测试，进程内
# from langgraph.checkpoint.sqlite import SqliteSaver        # 单机持久化
# from langgraph.checkpoint.postgres import PostgresSaver    # 生产推荐

graph = builder.compile(checkpointer=MemorySaver())

config = {"configurable": {"thread_id": "user-123-session-456"}}
graph.invoke({"messages": [("user", "我叫张三")]}, config)
graph.invoke({"messages": [("user", "我叫什么")]}, config)   # ✅ 记得住，无需自己拼历史

加了 checkpointer 之后，行为发生三个本质变化：

变化	说明
每个超步结束自动落盘	存的是完整的 channel 值（整个 State），不是 diff
`thread_id` 成为必填	不传直接抛错；`thread_id` 就是"一个会话 / 一条执行流"的主键
输入变成"增量"	第二次 `invoke` 只传新消息即可，框架自动取出历史合并

Checkpoint 里到底存了什么（面试常问）：

text

checkpoint = {
  thread_id        会话 ID（你定义的，比如 user_id + session_id）
  checkpoint_id    本次快照 ID（递增，时间旅行就是按它定位）
  checkpoint_ns    命名空间，子图在这里体现层级
  channel_values   State 各个 key 的当前值（真正的数据）
  versions_seen    各节点已消费到的版本号，用于判断该不该再执行
  pending_writes   本超步内已完成节点的写入（崩溃恢复时避免重跑成功的节点）
  metadata         source(input/loop/update) / step / writes / parents
}

pending_writes 是个高分细节：超步内三个并行节点，第 2 个挂了，第 1、3 个的写入会记在 pending_writes 里。恢复时框架只重跑失败的那个节点，不会重跑已成功的。但注意这是"进程内恢复"的优化 —— 如果超步本身没能形成完整 Checkpoint，跨进程恢复仍会回到上一个完整快照。所以幂等仍然是刚需。

常用状态操作 API：

python

snap = graph.get_state(config)              # 当前快照
snap.values                                  # State 内容
snap.next                                    # 下一步要执行的节点（中断时非空）
snap.tasks                                   # 待执行任务，含 interrupts 信息

for s in graph.get_state_history(config):    # 全部历史快照（倒序）
    print(s.metadata["step"], s.next)

graph.update_state(config, {"query": "改写后的问题"}, as_node="rewrite")  # 手动写状态

11. 中断与恢复：Human-in-the-Loop 的两种写法

写法一：静态断点 interrupt_before / interrupt_after（粗粒度，节点级）

python

graph = builder.compile(checkpointer=saver, interrupt_before=["tools"])

graph.invoke({"messages": [("user", "给李四转账 5000")]}, config)
# 执行到 tools 之前停住，返回当前状态

snap = graph.get_state(config)
print(snap.next)                       # ('tools',)
print(snap.values["messages"][-1].tool_calls)   # 给人看：模型打算调什么

# 人工确认后继续：输入传 None 表示"从断点续跑"
graph.invoke(None, config)

写法二：动态中断 interrupt()（细粒度，节点内任意位置，推荐）

python

from langgraph.types import interrupt, Command

def transfer_node(state: State):
    amount = state["amount"]
    if amount > 1000:
        # 执行到这里直接抛出 GraphInterrupt，当前状态落盘，invoke 返回
        decision = interrupt({
            "type": "approval",
            "question": f"确认转账 {amount} 元？",
            "payload": state["tool_args"],
        })
        if decision["approve"] is False:
            return {"messages": [ToolMessage("用户拒绝了该操作", tool_call_id=...)]}
        amount = decision.get("amount", amount)   # 人工还能改参数

    result = do_transfer(amount)                   # ← 真正的副作用放在 interrupt 之后
    return {"messages": [ToolMessage(result, tool_call_id=...)]}

# 恢复：用 Command(resume=...) 把人工输入喂回去
graph.invoke(Command(resume={"approve": True, "amount": 800}), config)

这里有一个必须死记的语义 —— interrupt() 恢复时，整个节点函数从头重新执行：

text

第一次执行 transfer_node：
    读 state → 发短信通知风控 ❌ ← 执行了
    → interrupt() 抛出 → 挂起

Command(resume=...) 恢复：
    整个 transfer_node 从第一行重跑
    读 state → 发短信通知风控 ❌ ← 又执行了一次！
    → 走到 interrupt()，这次不抛出，直接返回 resume 值
    → 继续往下执行

❌ 反例	✅ 正例
在 `interrupt()` 之前写数据库、发消息、调支付	`interrupt()` 之前只做纯读取和组装，所有副作用放在它之后
一个节点里放多个 `interrupt()` 且顺序不固定	多个 `interrupt` 按索引匹配 resume 值，顺序必须稳定；更稳的做法是拆成多个节点

HITL 的四种标准模式（面试经常要求举例）：

模式	用途	实现
Approve / Reject	高危工具调用前审批	`interrupt` 返回布尔，拒绝就写一条 ToolMessage 告诉模型"被拒了"
Edit State	人工修正模型填错的参数	`interrupt` 返回修改后的参数；或 `graph.update_state(...)` 直接改
Review Tool Call	人工改写整个 tool_call	`update_state` 替换最后一条 AIMessage（id 相同即覆盖）
Time Travel	回到第 N 步换个分支重跑	`get_state_history` 拿到旧 `checkpoint_id`，带着它重新 invoke

12. 流式输出：五种 stream_mode 分别给谁看

前端要的"打字机效果"和"执行进度条"是两种完全不同的流，混淆是对接时最常见的事故。

python

for chunk in graph.stream(inputs, config, stream_mode="updates"):
    print(chunk)

`stream_mode`	产出内容	典型用途
`"values"`	每个超步后的完整 State	调试、把全量状态同步给前端
`"updates"`	每个节点返回的 patch，形如 `节点名 → 增量`	执行进度条：当前在哪个节点、做了什么
`"messages"`	`(token, metadata)` 二元组，LLM token 级	打字机效果；metadata 里有 `langgraph_node`，可按节点过滤
`"custom"`	节点内 `get_stream_writer()(...)` 主动推送的任意数据	工具执行进度、"正在检索第 3/5 篇文档"
`"debug"`	任务开始/结束、Checkpoint 细节	排障

python

# 多模式同时订阅：chunk 变成 (mode, data)
for mode, data in graph.stream(inputs, config, stream_mode=["updates", "messages"]):
    if mode == "messages":
        token, meta = data
        if meta["langgraph_node"] == "agent":      # 只把主 Agent 的 token 吐给用户
            yield sse("token", token.content)
    else:
        yield sse("step", list(data.keys()))

python

# custom 模式：在工具里推进度
from langgraph.config import get_stream_writer

def retrieve_node(state):
    writer = get_stream_writer()
    for i, q in enumerate(state["sub_queries"]):
        writer({"stage": "retrieve", "progress": f"{i+1}/{len(state['sub_queries'])}"})
        ...

两个必踩的坑：① Python < 3.11 的异步环境下，节点里手动创建的协程不会自动继承 config，LLM 的 token 流会"消失" —— 必须把 config 显式透传给 llm.ainvoke(msgs, config)。② stream_mode="messages" 只能捕获通过 LangChain 模型对象发出的 token；你在节点里直接用 openai SDK 裸调，框架是看不见的，这时只能用 custom 模式自己推。

还有第六种流：astream_events(..., version="v2")。它和 stream_mode 不是一回事，面试被追问"节点内部的 retriever / 子链路事件怎么拿"时就靠它：

维度	`astream`（stream_mode）	`astream_events(version="v2")`
粒度	图级：节点 patch、State、LLM token	Runnable 级：节点里每一个 LLM / retriever / tool / parser 的 start、stream、end
事件结构	由 mode 决定	统一结构：`event` / `name` / `tags` / `metadata` / `data`
典型用途	做产品 UI（进度条 + 打字机）	做深度观测、调试、给每个子步骤单独上报耗时
代价	轻	事件量大很多，高 QPS 下要按 `event`/`tags` 过滤后再处理

python

async for ev in graph.astream_events(inputs, config, version="v2"):
    kind = ev["event"]
    if kind == "on_chat_model_stream" and ev["metadata"].get("langgraph_node") == "agent":
        yield sse("token", ev["data"]["chunk"].content)
    elif kind == "on_retriever_end":
        yield sse("docs", [d.metadata["title"] for d in ev["data"]["output"]])
    elif kind == "on_tool_start":
        yield sse("tool", ev["name"])

事件名	含义
`on_chain_start` / `on_chain_end`	节点或子链路开始/结束（`name` 就是节点名）
`on_chat_model_stream`	LLM token 增量，`data.chunk` 是 `AIMessageChunk`
`on_tool_start` / `on_tool_end`	工具调用，`data.input` / `data.output`
`on_retriever_end`	检索完成，可以把命中文档标题推给前端做引用

选择口诀： 做用户看得见的 UI 用 stream_mode（结构稳定、量小）；做排障和细粒度上报用 astream_events。不要两个一起上，事件会重复、前端去重逻辑会变得非常难维护。给自定义工具打 tags 可以让事件过滤从"字符串判断节点名"升级成"按 tag 分流"。

生产流式还有三件事要补（Demo 阶段体会不到）：

问题	处理方式
客户端断线重连，重复收到已输出的 token	每个 SSE 事件带自增 `seq`，前端按 `seq` 去重；或重连时只回放 `get_state` 的完整结果而不重放 token
流到一半节点抛异常	先发一个 `error` 事件再关闭连接，别让前端一直转圈；同时用 `get_state` 拿到已完成的部分做降级展示
用户点"停止生成"	取消请求侧的任务即可，Checkpoint 已经落到上一个超步，下次 `invoke(None, config)` 能从那里续上

13. 递归限制：`recursion_limit` 算的是超步不是节点

python

graph.invoke(inputs, {"configurable": {"thread_id": "1"}, "recursion_limit": 50})

易错点	正解
"25 次是指 25 个节点"	是 25 个超步；并行的 5 个节点只算 1 步
"ReAct 能跑 25 轮工具"	一轮 ReAct = `agent` + `tools` = 2 个超步，默认只够约 12 轮
报错了就无脑调大	先排查是不是条件边写反导致死循环；调大只是把烧钱时间拉长
用它来做业务限流	它抛 `GraphRecursionError`，没有降级答案；业务限流要自己在 State 里记账

python

from langgraph.errors import GraphRecursionError

try:
    graph.invoke(inputs, config)
except GraphRecursionError:
    # 兜底：把当前已有的中间结果拼一个降级回复，而不是丢给用户 500
    snap = graph.get_state(config)
    return degrade_answer(snap.values)

14. 子图 Subgraph：把大图拆小的唯一正确姿势

一张图超过 15 个节点就该拆子图。拆的时候状态怎么对接是全部难点。

情况一：父子 State 共享同名 key（最简单）

python

# 子图
sub = StateGraph(SubState)        # SubState 里有 messages
...
sub_graph = sub.compile()

# 父图：直接把编译后的子图当节点用
parent.add_node("research", sub_graph)

同名 key 自动透传，父图的 messages 流进子图，子图的写入按父图的 reducer 合并回来。

情况二：父子 State 结构不同（必须手写包装函数）

python

def call_research(state: ParentState) -> dict:
    sub_in = {"query": state["messages"][-1].content, "top_k": 5}   # 父 → 子：显式映射
    sub_out = sub_graph.invoke(sub_in)
    return {"docs": sub_out["documents"]}                            # 子 → 父：显式映射

parent.add_node("research", call_research)

对比	直接挂编译后的图	包装成普通函数
状态映射	自动（靠同名 key）	手写，完全可控
流式/中断能否穿透	✅ 能，父图 `stream` 可用 `subgraphs=True` 看到子图事件	❌ 子图被当成一次普通调用，内部中断穿不出来
Checkpoint	子图有独立 `checkpoint_ns`，可单独时间旅行	子图执行过程不落盘
适用	需要观测/中断/恢复子流程	子流程是"黑盒工具"，不需要中间态

坑：子图里写了父图没有的 key，会被静默丢弃。 排查方式是用 graph.stream(..., subgraphs=True) 把子图的 updates 打出来对照，而不是干瞪着父图的最终结果猜。

15. Send API：真正的动态 map-reduce

条件边只能在已声明的节点之间选路，解决不了"运行时才知道要并行跑 N 份"的问题。Send 就是为此存在的。

python

from langgraph.types import Send

class State(TypedDict):
    chapters: list[str]
    summaries: Annotated[list, operator.add]   # ← fan-in 必须有 reducer
    final: str

# map：为每个 chapter 动态生成一个任务
def fan_out(state: State):
    return [Send("summarize_one", {"chapter": c}) for c in state["chapters"]]

# 被 Send 调用的节点，收到的 state 就是 Send 的第二个参数，不是全局 State！
def summarize_one(payload: dict):
    s = llm.invoke(f"总结这一章：{payload['chapter']}").content
    return {"summaries": [s]}                   # 写回全局 State，靠 reducer 合并

# reduce
def reduce_all(state: State):
    return {"final": llm.invoke("合并这些摘要：" + "\n".join(state["summaries"])).content}

builder.add_conditional_edges("split", fan_out, ["summarize_one"])
builder.add_edge("summarize_one", "reduce_all")

`Send` 的三个关键语义	说明
接收节点拿到的是自定义载荷	不是全局 State，所以别在里面写 `state["messages"]`
写回仍然走全局 State 的 reducer	汇总字段必须带 `operator.add` 之类的 reducer，否则并发写报 `InvalidUpdateError`
N 个 Send = 一个超步	不消耗 N 次 `recursion_limit`

fan-in 节点被执行多次的经典坑：当各分支步数不一样时（比如 A 分支一步出结果，B 分支要 检索 → 重排 → 总结 三步），汇总节点会在 A 完成后就被触发一次，等 B 完成又触发一次 —— 结果就是"摘要少了一半"或"汇总节点跑了 3 遍"。

python

# ✅ defer=True：等所有能到达它的分支都跑完，才执行一次
builder.add_node("reduce_all", reduce_all, defer=True)

方案	适用	代价
`defer=True`（推荐）	分支步数不等的 map-reduce、投票、多路检索融合	需要较新版本；汇总延迟等于最慢分支
把各分支补齐成相同步数	分支本来就差一步	要塞空节点，图变脏
汇总节点里自己判断"齐了没有"	老版本兜底	State 里要记 `expected` / `done` 计数，节点必须幂等

python

def reduce_all(state: State):
    if len(state["summaries"]) < state["expected"]:
        return {}                      # 没齐就什么都不做，等下一次被触发
    return {"final": merge(state["summaries"])}

16. 和 Day27 裸循环的能力对照总表

能力	Day27 while 循环	LangGraph
工具循环	自己写	条件边 + 回边
并行工具	自己开线程池	同超步多节点 / `Send` / `ToolNode` 内置并发
状态持久化	自己序列化到 Redis	`checkpointer` + `thread_id`
断点续跑	几乎做不到	`invoke(None, config)`
人工审批	要拆成两个接口自己存上下文	`interrupt()` + `Command(resume=...)`
回到第 N 步重跑	不可能	`get_state_history` + `checkpoint_id`
进度可视化	打日志	`stream_mode="updates"`
多 Agent 协作	手写路由	Supervisor / Swarm / 子图
心智成本	低，一眼看懂	高，要理解超步、reducer、通道

结论话术： 「流程固定、步数少于 3 步、不需要持久化和人工介入的场景，裸循环更合适，上 LangGraph 是过度设计。一旦出现'要能中断、要能恢复、要能并行、要有多个角色'中的任意两条，就该换 LangGraph。」

✨ 高级深挖知识点

1. reducer 的三类冲突，以及自定义 reducer 的正确写法

reducer 是 LangGraph 里最容易出事的地方，因为它的错误不报异常，只是结果不对。

冲突类型	现象	根因	修法
覆盖冲突	并行两个节点写同一个无 reducer 字段	默认 reducer 是 `last_value`，但并行时"谁最后"不确定	要么加 reducer，要么拆成两个字段
累加冲突	`list` 字段越跑越长，里面全是重复项	用了 `operator.add` 但节点非幂等、重跑时又追加一遍	reducer 里做去重，或改用 `dict` 按 key 覆盖
类型冲突	reducer 收到 `None` 直接抛 `TypeError`	节点返回了 `{"docs": None}`	reducer 必须自己处理 `None` 和首次初始化

python

from typing import Annotated, TypedDict

def merge_docs(left: list | None, right: list | None) -> list:
    """按 doc_id 去重合并，右侧覆盖左侧 —— 幂等、可重跑"""
    left = left or []
    right = right or []
    merged = {d["id"]: d for d in left}
    for d in right:
        merged[d["id"]] = d
    return list(merged.values())

def keep_max_score(left: float | None, right: float | None) -> float:
    return max(left or 0.0, right or 0.0)

class State(TypedDict):
    docs: Annotated[list, merge_docs]      # ✅ 幂等合并
    best: Annotated[float, keep_max_score] # ✅ 并行写也不冲突
    answer: str                            # 只有单一节点写，无需 reducer

判断口诀：一个字段会不会被"两个以上节点"或"同一节点重跑两次"写入？会，就必须给它一个幂等的 reducer。

2. State Schema 分层：input / output / private 三件套

把所有字段塞进一个 State 会导致：内部中间变量泄露给调用方、调用方必须传一堆它不该知道的字段。

python

class InputState(TypedDict):
    question: str

class OutputState(TypedDict):
    answer: str

class PrivateState(TypedDict):          # 只在两个节点之间传递，不对外暴露
    raw_chunks: list[str]

class OverallState(InputState, OutputState):
    messages: Annotated[list, add_messages]
    retry: int

builder = StateGraph(
    OverallState,
    input_schema=InputState,            # invoke 只接受 question
    output_schema=OutputState,          # 返回值只有 answer
)

def node_a(state: OverallState) -> PrivateState:     # 写私有 schema
    return {"raw_chunks": [...]}

def node_b(state: PrivateState) -> OverallState:     # 读私有 schema
    return {"answer": summarize(state["raw_chunks"])}

好处	说明
接口收窄	前端不需要知道 `retry`、`raw_chunks`
防误用	`input_schema` 之外的字段传进来会被忽略，不会悄悄污染状态
中间态不落 Checkpoint 外泄	私有字段依然会进 checkpoint，但不会出现在返回值里

注意：output_schema 只裁剪返回值，不裁剪 checkpoint。 真正敏感的数据（用户身份证、原始 token）不要放 State，应该放 config["configurable"] 或外部存储，因为 State 是会被完整序列化落盘的。

3. `add_messages` 的隐藏能力：按 id 覆盖与 RemoveMessage

很多人以为 add_messages 就是 list.append，其实它是按 id 的 upsert。

python

from langchain_core.messages import RemoveMessage, HumanMessage, REMOVE_ALL_MESSAGES

# ① 按 id 覆盖：修正上一条消息而不是追加
def edit_node(state):
    last = state["messages"][-1]
    return {"messages": [HumanMessage(content="改过的问题", id=last.id)]}

# ② 删除指定消息：做上下文裁剪
def trim_node(state):
    old = state["messages"][:-10]
    return {"messages": [RemoveMessage(id=m.id) for m in old]}

# ③ 清空重开（0.3+）
def reset_node(state):
    return {"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES)]}

行为	触发条件
追加	新消息没有 `id` 或 `id` 不存在
覆盖	新消息的 `id` 与已有消息相同
删除	传 `RemoveMessage(id=...)`

坑：删除 AIMessage(tool_calls=[...]) 时必须连带删掉对应的 ToolMessage，否则下一轮请求会因为"有 tool 响应但找不到对应 tool_call"被 OpenAI 直接 400。裁剪上下文时要以完整的 tool 回合为单位切，不能按条数硬切。

4. 上下文工程：三种裁剪策略挂在图的什么位置

Agent 跑长了必然爆上下文。LangGraph 的做法是把裁剪本身做成一个节点，而不是在调用 LLM 前临时处理。

START → agent ⇄ tools
          ↑
       manage_ctx   ← 每轮进 agent 前先过一遍裁剪节点

策略	实现	适用
滑动窗口	`trim_messages(max_tokens=..., strategy="last")`	客服、闲聊，旧信息无价值
摘要压缩	超过阈值时调 LLM 把前 N 条压成一条 `SystemMessage`，再 `RemoveMessage` 掉原文	长任务、需要记住早期结论
外置检索	全量消息写入 `Store`，每轮只检索相关片段塞进 prompt	超长对话、跨会话记忆

python

from langchain_core.messages import trim_messages

def manage_ctx(state):
    msgs = state["messages"]
    if count_tokens(msgs) < 60_000:
        return None                              # ← 返回 None = 不改状态，省一次写盘
    summary = llm.invoke(SUMMARY_PROMPT + render(msgs[:-6])).content
    return {
        "messages": [RemoveMessage(id=m.id) for m in msgs[:-6]]
                    + [SystemMessage(content=f"【历史摘要】{summary}")],
    }

面试加分点： 「裁剪必须做成独立节点而不是在 agent 节点里顺手做，原因是 ① 裁剪结果要落 Checkpoint，否则断点恢复后又得重新压一遍、白花一次 LLM 调用；② 摘要本身是一次 LLM 调用，做成节点才能被 stream 观测到和被重试策略覆盖。」

5. Store：跨 thread 的长期记忆

Checkpointer 存的是单个 thread 的短期状态，换个 thread_id 就什么都不记得了。跨会话记忆要用 Store。

python

from langgraph.store.memory import InMemoryStore
# 生产：from langgraph.store.postgres import PostgresStore

store = InMemoryStore(index={"embed": embeddings, "dims": 1536})

def remember(state, *, store):                      # 节点签名注入 store
    uid = state["user_id"]
    store.put(("memories", uid), key=str(uuid4()), value={"text": state["fact"]})
    return None

def recall(state, *, store):
    uid = state["user_id"]
    hits = store.search(("memories", uid), query=state["question"], limit=3)
    return {"context": [h.value["text"] for h in hits]}

graph = builder.compile(checkpointer=checkpointer, store=store)

维度	Checkpointer	Store
作用域	单 `thread_id`	按 `namespace`（通常是 `(类型, user_id)`）跨 thread
内容	完整 State 快照 + 执行进度	自定义 KV 文档
典型用途	断点续跑、时间旅行	用户画像、偏好、长期事实
检索能力	只能按 checkpoint_id 取	支持前缀列举 + 向量语义检索

架构话术： 「短期记忆靠 Checkpointer 自动落盘，长期记忆靠 Store 显式读写。写长期记忆有两种时机 —— 热路径（对话中实时抽取，延迟高但即时生效）和后台（会话结束后异步总结，不影响响应速度）。我们选后台，因为用户不会在同一句话里既告诉你偏好又要求你立刻用上。」

6. 并行节点的事务性：超步失败到底回滚了什么

超步 N 开始
  ├─ node_a ✅ 完成 → 写入 pending_writes
  ├─ node_b ✅ 完成 → 写入 pending_writes
  └─ node_c ❌ 抛异常
       ↓
  整个超步不提交 → State 保持超步 N-1 的样子
  但 a/b 的 pending_writes 已落盘
       ↓
  恢复时：a、b 不再重跑（复用 pending_writes），只重跑 c

关键结论	工程含义
State 以超步为事务单位提交	超步内任一节点失败，这一步的状态更新全部不可见
成功节点的结果存在 `pending_writes`	恢复后不会重复调 LLM，这是省钱的关键机制
但副作用无法回滚	`node_a` 已经扣过款、已经发过邮件，LangGraph 管不了

所以生产里的铁律：有外部副作用的节点必须自带幂等键。 比如下单节点用 f"{thread_id}:{checkpoint_id}:order" 作为幂等 token 传给下游服务，让下游去重，而不是指望框架帮你回滚。

7. `InvalidUpdateError` 的四种触发场景速查

报错场景	原因	修法
并行两节点写同一无 reducer 字段	默认 reducer 不接受并发	加 `Annotated[..., reducer]`
节点返回了非 dict（比如字符串）	返回值必须是 patch dict / `Command` / `None`	包成 `{"field": value}`
patch 里有 State 未声明的 key	schema 校验	补进 schema，或走私有 schema
`Send` 的目标节点名拼错	找不到节点	条件边第三个参数把可能的目标声明清楚，让编译期就能查出来

8. `Command`：一个返回值同时干"改状态"和"跳转"两件事

条件边的问题是：路由决策和状态更新被硬拆成两个地方，路由函数只能读不能写，节点只能写不能决定去哪。Command 把两者合一。

python

from langgraph.types import Command
from typing import Literal

def router_node(state) -> Command[Literal["billing", "tech", "human"]]:
    intent = classify(state["messages"][-1].content)
    return Command(
        update={"intent": intent, "route_log": [intent]},   # 改状态
        goto=intent if intent in ("billing", "tech") else "human",  # 同时决定去哪
    )

builder.add_node("router", router_node)
# 注意：用了 Command 的节点不需要 add_conditional_edges，
# 但要靠返回类型注解 Command[Literal[...]] 让 LangGraph 画出图的连线

用法	写法	场景
状态 + 跳转	`Command(update=..., goto=...)`	路由节点、Agent handoff
跳到父图	`Command(goto="x", graph=Command.PARENT)`	子图里决定退出到父图的某个节点
恢复中断	`Command(resume=人工输入)`	HITL 续跑
并行 fan-out	`Command(goto=[Send(...), Send(...)])`	动态 map

对比话术： 「条件边适合纯路由（决策不依赖写状态）；Command 适合决策和状态耦合的场景，典型就是多 Agent 的 handoff —— 交接的同时要把任务描述写进共享状态。另外子图想跳回父图的兄弟节点，只有 Command(graph=Command.PARENT) 能做到，条件边做不到。」

9. 多智能体四种拓扑，面试必须能画出来

① Network（全连接）           ② Supervisor（中心调度）
    A ⇄ B                          ┌──→ A ──┐
    ⇅ ╳ ⇅                     Sup ─┼──→ B ──┼─→ 回 Sup
    C ⇄ D                          └──→ C ──┘

③ Hierarchical（分层）        ④ Swarm（去中心交接）
    TopSup                         A ──handoff──→ B
     ├─ SubSup1 ─ A, B                 ↖─────────↙
     └─ SubSup2 ─ C, D             （谁接手就由谁继续对话，直到再交接）

拓扑	实现要点	优点	缺点
Network	每个 Agent 都能 `goto` 任意 Agent	最灵活	token 爆炸、容易绕圈，几乎不用于生产
Supervisor	Supervisor 是唯一的路由节点，子 Agent 做完回边指向它	可控、好调试、好限流	Supervisor 是单点，prompt 会越来越长
Hierarchical	Supervisor 的下级也是 Supervisor（子图）	能撑几十个 Agent	层级深 → 延迟高
Swarm	用 handoff 工具 + `Command` 把控制权交出去	对话连贯、无中心开销	全局可观测性差，谁在负责要靠状态记录

python

# Supervisor 的最小实现：结构化输出决定下一个 Agent
class Route(BaseModel):
    next: Literal["researcher", "coder", "FINISH"]

def supervisor(state) -> Command[Literal["researcher", "coder", "__end__"]]:
    r = llm.with_structured_output(Route).invoke(SUP_PROMPT + state["messages"])
    return Command(goto=END if r.next == "FINISH" else r.next)

选型话术： 「两三个角色、流程线性 → 直接子图串起来；角色多但职责清晰、需要集中限流和审计 → Supervisor；强调对话体验、角色之间是"转接客服"关系 → Swarm。我们选 Supervisor，因为它能在一个地方统一做预算控制和越权拦截。」

10. Handoff 的标准实现：让工具返回 `Command`

多 Agent 交接的工业写法不是在 prompt 里写"请把问题转给 X"，而是把交接变成一个工具。

python

from langchain_core.tools import tool, InjectedToolCallId
from langgraph.prebuilt import InjectedState

def make_handoff(agent_name: str):
    @tool(f"transfer_to_{agent_name}", description=f"把任务交接给 {agent_name}")
    def handoff(
        task: str,
        state: Annotated[dict, InjectedState],
        tool_call_id: Annotated[str, InjectedToolCallId],
    ) -> Command:
        tool_msg = ToolMessage(
            content=f"已交接给 {agent_name}", tool_call_id=tool_call_id
        )
        return Command(
            goto=agent_name,
            graph=Command.PARENT,                       # 跳到父图的兄弟 Agent
            update={"messages": state["messages"] + [tool_msg],
                    "task_desc": task},                 # 把任务描述一起带过去
        )
    return handoff

细节	为什么不能省
必须返回 `ToolMessage`	不返回就留下一个"悬空的 tool_call"，下一轮 LLM 请求直接 400
必须带 `graph=Command.PARENT`	Agent 自己是子图，目标 Agent 是父图里的兄弟节点，不加就会在子图里找不到目标
必须显式传 `task`	不要让下一个 Agent 从几十条历史里猜任务，显式交接任务描述能明显提升成功率

11. Checkpoint 膨胀：被忽略的存储炸弹

Checkpoint 是每个超步存一份全量 State，不是存 diff。一个 20 轮对话的 Agent 很容易写出几十 MB。

膨胀来源	治理手段
每超步一份全量快照	生产用 `PostgresSaver`（大字段走 blob 表、按 channel 版本去重），别用 `SqliteSaver` 扛并发
State 里塞了检索到的原文	只存 `doc_id` + 摘要，原文留在向量库，节点内按需取
`messages` 无限增长	配裁剪节点（深挖 4），并用 `RemoveMessage` 真正删掉
历史 thread 永不清理	按 `thread_id` 建定时任务删 N 天前的 checkpoint；或业务结束时主动 `delete_thread`
调试期开了 `stream_mode="debug"` 全量落盘	生产关掉

python

# 0.6+ 的 durability 三档：用写盘时机换性能
graph.invoke(inputs, config, durability="async")

`durability`	行为	取舍
`"exit"`	只在整图结束时写一次	最快，中途崩了全丢 —— 只适合纯计算、无 HITL
`"async"`	超步结束后异步写，不阻塞下一步	默认推荐，极端情况下可能丢最后一个超步
`"sync"`	超步结束后同步写完才继续	最慢但最可靠，HITL / 付费类流程用这个

12. 节点级可靠性：retry_policy 与 cache_policy

重试不该写在节点函数里，否则重试次数不会被 Checkpoint 记录，恢复后计数归零。

python

from langgraph.types import RetryPolicy, CachePolicy

builder.add_node(
    "call_llm", call_llm,
    retry=RetryPolicy(
        max_attempts=3,
        initial_interval=0.5,
        backoff_factor=2.0,
        jitter=True,
        retry_on=lambda e: isinstance(e, (RateLimitError, APITimeoutError)),  # ← 只重试可恢复错误
    ),
    cache_policy=CachePolicy(ttl=300),        # 相同输入 5 分钟内直接命中缓存
)
graph = builder.compile(cache=InMemoryCache())

配置	不配会怎样
`retry_on` 白名单	不配就连 `ValidationError`、`AuthError` 这种重试 100 次也不会好的错误一起重试，白烧钱
`jitter`	不开就会出现"多个并行节点同时被限流、同时退避、同时重试"的雷群效应
`cache_policy`	调试时每改一行 prompt 都要把前面所有节点重跑一遍，LLM 账单飞涨

缓存的坑：缓存键由节点输入计算，含时间戳/随机 id 的输入永远不会命中。 想让缓存生效，就别把 datetime.now() 塞进 State。

13. 可观测性：一个 trace 要能回答"钱花在哪、卡在哪"

python

graph.invoke(inputs, {
    "configurable": {"thread_id": tid, "user_id": uid},
    "run_name": "support-agent",
    "tags": ["prod", "v2"],
    "metadata": {"tenant": "jd", "exp_group": "B"},    # ← 落到 LangSmith，可按维度聚合
    "callbacks": [tracer],
})

必须打的指标	为什么
每个节点的耗时 P95	定位是 LLM 慢还是工具慢
每轮的 prompt / completion token	做成本归因，找出"上下文没裁剪"的 thread
超步数分布	长尾超步数 = 绕圈的 Agent，是最大的烧钱源
工具调用成功率 / 参数校验失败率	工具 description 写得好不好，直接看这个
中断率与人工通过率	HITL 闸门设得太松还是太紧

注意：LangGraph 的 trace 是树状的，子图会嵌套。 排障时先看超步时间线定位到哪个节点，再钻进该节点看 LLM 的真实 prompt —— 不要一上来就盯最终输出猜原因。

14. 测试策略：三层金字塔

层级	测什么	怎么做
节点单测	节点是纯函数 → 给定 state 断言返回的 patch	直接 `node_fn({"messages": [...]})`，不编译图、不连网
路由单测	条件边的判定逻辑	直接调路由函数，断言返回的键
图集成测	完整路径、中断恢复、递归限制	用 `FakeListChatModel` / `FakeMessagesListChatModel` 按脚本返回固定 tool_call 序列

python

from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel

fake = FakeMessagesListChatModel(responses=[
    AIMessage(content="", tool_calls=[{"name": "search", "args": {"q": "x"}, "id": "1"}]),
    AIMessage(content="最终答案"),
])

def test_react_loop(monkeypatch):
    graph = build_graph(llm=fake, checkpointer=MemorySaver())
    out = graph.invoke({"messages": [HumanMessage("x")]},
                       {"configurable": {"thread_id": "t1"}})
    assert out["messages"][-1].content == "最终答案"
    # 断言确实走了 tools 节点
    assert any(isinstance(m, ToolMessage) for m in out["messages"])

一条经验：图的集成测试一定要带 checkpointer=MemorySaver() 并显式传 thread_id，否则你测不出任何和持久化/恢复相关的 bug，而线上出的恰恰都是这类 bug。

15. 安全：把图当权限边界用

Agent 的安全问题不是"模型会不会说坏话"，而是它手里的工具权限太大。LangGraph 的图结构正好可以当权限边界。

用户输入
   ↓
[输入校验节点]  ← prompt 注入检测、敏感词、租户归属校验
   ↓
[agent]  ⇄  [只读工具]        ← 低风险，自动执行
   ↓
[风险分级节点] → 高风险？
   ↓ 是                ↓ 否
[interrupt 人工审批]   [写操作工具]

风险	图层面的解法
Prompt 注入让 Agent 调危险工具	把写操作工具放在审批节点之后，模型根本拿不到它的调用权
工具返回的内容里带指令（间接注入）	在 `tools` 节点后加一个清洗节点，把工具结果标记为"数据而非指令"
越权访问他人数据	`user_id` 从 `config["configurable"]` 取，绝不从 State / 模型输出取（模型可被诱导改 State）
工具参数被诱导成 `rm -rf`	工具内部做白名单校验，失败就返回错误让模型重试，而不是抛异常中断整图
无限循环烧钱	四道闸门（基础 9）+ 按 `user_id` 的日预算

面试金句： 「身份与权限必须走 config，不能走 State。因为 State 是模型可以写的，而 config 是调用方注入、模型完全看不到也改不了的 —— 这是 LangGraph 里唯一可信的通道。」

16. Functional API：不想画图时的另一条路

0.3 之后 LangGraph 提供了 @entrypoint / @task，用普通控制流写，但同样拿到持久化、中断、流式。

python

from langgraph.func import entrypoint, task

@task
def search(q: str) -> str:
    return tavily.invoke(q)

@entrypoint(checkpointer=MemorySaver())
def agent(q: str) -> str:
    docs = search(q).result()                  # @task 的结果会被 Checkpoint 记住
    approved = interrupt({"draft": draft(docs)})   # 一样能中断
    return finalize(approved)

对比	Graph API	Functional API
控制流	显式图结构	普通 if / for / while
状态	共享 State + reducer	普通局部变量
可视化	✅ 自动画图	❌ 没有图
恢复粒度	超步	`@task` 调用
适合	多 Agent、要给人看流程、要时间旅行	流程是线性/树状的脚本，只想要持久化和中断

选型话术： 「流程需要被产品和运维看懂、需要可视化调试 → Graph API；只是一段复杂的业务脚本想白捡持久化和人工审批 → Functional API。两者可以混用，@task 里可以 invoke 一个 Graph。」

17. 性能：延迟都花在哪，以及怎么压

延迟来源	优化手段
串行的 ReAct 轮次（最大头）	① 支持并行 tool_calls，让一轮调多个工具；② 给模型更好的工具描述减少试错轮次
每超步的 Checkpoint 写盘	`durability="async"`；State 瘦身
首 token 等待	`stream_mode="messages"` 直接把 token 透给前端；路由/分类节点换小模型
子图层级过深	层级每多一层就多一次完整的状态序列化，控制在 2 层内
工具自身慢	工具内部加缓存 + 超时；用 `cache_policy` 挡住重复调用

一条反直觉的结论：把并行节点开得越多不一定越快，因为超步有屏障 —— 整个超步的耗时等于最慢那个节点。所以并行分支的耗时要尽量均衡，一个 10s 的节点和四个 1s 的节点并行，总耗时还是 10s。

18. 版本坑：0.2 → 0.6 的关键变更

变更	旧写法	新写法
包拆分	`from langgraph.checkpoint import SqliteSaver`	`pip install langgraph-checkpoint-sqlite` 后 `from langgraph.checkpoint.sqlite import SqliteSaver`
中断	编译时 `interrupt_before=["tools"]`	节点内 `interrupt(payload)` + `Command(resume=...)`
恢复	`invoke(None, config)`	仍可用，但动态中断推荐 `Command(resume=...)`
schema 参数	`StateGraph(S, input=I, output=O)`	`input_schema=` / `output_schema=`（旧名已弃用）
预制件	`langgraph.prebuilt` 里的多 Agent	独立包 `langgraph-supervisor` / `langgraph-swarm`
Saver 上下文	直接 `SqliteSaver(conn)`	推荐 `with SqliteSaver.from_conn_string(...) as cp:`，否则连接不释放
持久化时机	固定同步写	新增 `durability` 三档：`exit` / `async` / `sync`

升级经验：先锁 langgraph 与 langchain-core 的版本对，这两个包的 Command / interrupt / 消息类型是强耦合的，单独升一个极容易出现 ImportError 或"interrupt 不生效但也不报错"。

19. 部署形态：从 `compile()` 到可对外提供服务

写完图只是第一步。graph.compile() 得到的只是一个进程内对象，要对外提供服务有三条路，面试问"你们怎么上线的"就答这张表：

形态	做法	适合	代价
自己包一层 Web 框架	FastAPI 里 `await graph.ainvoke(...)`，SSE 转发 `astream`	已有后端体系、要复用现成的鉴权/网关/日志	线程管理、流式协议、并发控制、thread 生命周期全要自己写
LangGraph Server（自托管）	写 `langgraph.json` 声明图入口，`langgraph dev` 本地起、镜像化上线	想白捡 HTTP API、流式、HITL、定时任务	多一个服务要运维，依赖 Postgres + Redis
托管平台	直接部署，平台管扩容与持久化	快速验证、团队没有运维资源	数据出域与成本问题要先过合规

json

// langgraph.json
{
  "dependencies": ["."],
  "graphs": { "agent": "./src/agent/graph.py:graph" },
  "env": ".env"
}

bash

langgraph dev       # 本地起服务 + 可视化调试面板，热重载
langgraph build     # 打镜像

Server 形态额外送你三个能力，自己包 FastAPI 就得自己实现：

能力	说明
Assistant	同一张图 + 不同配置（模型、prompt、工具白名单）= 不同 assistant，还能带版本号，相当于"图的配置化多租户"
Thread 管理	thread 的增删查、历史消息、状态回看都是现成 HTTP 接口
Cron / 定时任务	定时往某个 thread 投一条输入，用来做日报、巡检类 Agent

不管走哪条路，持久化这几件事都躲不开：

python

from langgraph.checkpoint.postgres import PostgresSaver
from psycopg_pool import ConnectionPool

pool = ConnectionPool(DB_URI, max_size=20, kwargs={"autocommit": True})
cp = PostgresSaver(pool)
cp.setup()                 # ← 首次必须跑，建表 + 跑迁移；忘了就报表不存在
graph = builder.compile(checkpointer=cp)

上生产要确认的点	说明
`.setup()` 执行过了吗	建表与版本迁移；放到应用启动时跑一次，或单独做成迁移脚本
连接池大小	每个超步都要读写 checkpoint，池子太小会在高并发时排队，表现为"延迟忽然翻倍"
异步用对包	异步服务要用 `AsyncPostgresSaver`，混用同步 Saver 会阻塞事件循环
多副本能否水平扩	状态全在 Postgres，所以无状态副本可以随意扩；但同一 `thread_id` 的并发请求要做串行化（见下一条）
旧 thread 清理	checkpoint 是全量快照，必须有 TTL 或定时清理任务

20. Double-texting：用户没等回答完就又发了一条

这是真实流量里一定会遇到、Demo 里永远遇不到的问题：上一轮还在跑工具，用户又发来一条消息，同一个 thread_id 出现并发写。

text

t0  用户: "帮我查下 A 的库存"     → 图开始跑，正在调工具
t1  用户: "算了，查 B 的"         → 第二个请求进来，同一个 thread_id
                                   ↓
                    两个 run 同时往同一份 checkpoint 写 → 状态错乱

四种处理策略，对应到产品体验上是四种不同的交互：

策略	行为	适合场景
`reject`	直接拒掉第二条，提示"上一条还在处理"	金融、工单类严肃场景，宁可不响应也不能乱
`enqueue`	第二条排队，等第一个 run 结束再跑	用户在补充信息（"还有一点..."）
`rollback`	取消第一个 run 并回滚它写过的状态，只跑第二条	用户明显在改主意（"算了"）
`interrupt`	中断第一个 run 但保留已完成的部分，再跑第二条	通用默认值，体验最自然

python

# LangGraph Server 下只是一个参数
client.runs.create(thread_id, "agent", input=..., multitask_strategy="rollback")

自己包 FastAPI 的话要手动实现，核心是两件事：

python

# ① 同 thread 串行化：Redis 锁，粒度是 thread_id 而不是用户
async def run_locked(thread_id, payload):
    lock = redis.lock(f"lg:{thread_id}", timeout=120)
    if not await lock.acquire(blocking=False):
        raise Busy("上一条还在处理")          # ← reject 策略
    try:
        return await graph.ainvoke(payload, cfg(thread_id))
    finally:
        await lock.release()

# ② 可取消：把 run 的 task 存起来，新请求进来时 cancel 掉旧的（interrupt 策略）
task = asyncio.create_task(graph.ainvoke(payload, cfg(thread_id)))
RUNNING[thread_id] = task

实现要点	说明
锁的粒度是 `thread_id`	按用户加锁会导致他开两个会话都被互相阻塞
`cancel` 之后状态是干净的	取消点落在超步边界，上一个超步已提交、当前超步整体不提交
rollback 要能定位回滚点	用 `get_state_history` 找到该 run 开始前的 `checkpoint_id`，带着它重新 invoke
必须给前端明确信号	被 reject / 被 interrupt 都要有独立事件类型，否则前端只能看到"消息发出去没反应"

面试话术： 「我们默认用 interrupt 语义：新消息进来就取消正在跑的 run，但保留它已经完成的超步，所以用户追问时之前检索到的文档不会白跑。写操作类节点我们单独改成 reject，因为那种场景下中途取消的歧义比等一会儿更难解释。」

21. 一句话串起全链路（给面试官的收束）

「LangGraph 的本质是把 Agent 的控制流变成一张可持久化的状态机。State 是带 reducer 的共享内存，Node 是幂等的纯函数，Edge 决定控制流，超步是事务与 Checkpoint 的最小单位。有了 Checkpoint 就有了断点续跑、人工审批和时间旅行；有了条件边和 Send 就有了循环与动态并行；有了 Command 和子图就有了多 Agent 协作。它换来的是可恢复、可观测、可人工介入，代价是心智成本上升 —— 所以简单流程不该上它。」

❓ 高频面试题【基础简答 + 连环追问链 + 精简口述答案 + 项目关联话术】

题 1：LangGraph 是什么？它和 LangChain 是什么关系，为什么不用 LCEL 就够了

精简口述答案（90 秒版）

LangGraph 是 LangChain 团队做的有状态、可循环的 Agent 编排框架，底层是 Pregel 的 BSP 模型。它和 LCEL 不是替代关系而是正交的：LCEL 管的是数据流，把 prompt、model、parser 用管道串成一个 DAG，特点是单向、无环、无状态；LangGraph 管的是控制流，支持循环、分支、并行、中断、恢复，节点内部通常还是用 LCEL 写的。为什么 LCEL 不够？因为 Agent 的本质是循环——"想→做→观察→再想"，DAG 画不出环。那我自己写 while 循环行不行？能跑但撑不住生产，缺五样东西：状态持久化、断点续跑、人工审批、执行过程可观测、多角色协作。LangGraph 正好把这五样做成了基础设施：State 是带 reducer 的共享内存，Checkpointer 每个超步自动落盘，interrupt 能在任意节点暂停等人工，stream 能把每一步的进度推给前端。

追问	答题要点
底层为什么选 Pregel？	图计算的 BSP 模型天然适配"一批节点并行执行 → 屏障同步 → 合并状态 → 下一轮"，和 Agent 的多步推理同构
节点里还能用 LCEL 吗？	能，而且应该。节点内部是 `prompt \| llm \| parser` 这样的链，节点之间才是图
什么场景不该上 LangGraph？	流程固定、步数 ≤ 3、不需要持久化和人工介入 —— 这时候裸循环或 LCEL 更好维护
和 Dify / Coze 这类低代码编排的区别？	低代码是产品，适合固定流程、非开发者配置；LangGraph 是库，适合需要精细控制状态、重试、权限的自研系统

✅ 项目关联话术：我们第一版知识库助手是 Day27 那种裸 while 循环，上线两周踩了两个问题。第一是链路长的任务一旦中途失败就得全量重跑，一次检索 + 两次 LLM 的钱白花；第二是有写操作的工具必须人工确认，但 HTTP 一次请求内没法等人，我们只能把流程硬拆成两个接口、自己把中间上下文塞 Redis，代码很脏。换成 LangGraph 后，前者靠 Checkpoint 的 pending_writes 直接复用成功节点的结果，后者靠 interrupt 一行搞定，我们删掉了大约 200 行自己写的状态管理代码。

题 2：State 和 reducer 讲一下，为什么并行节点写同一个字段会报错

精简口述答案（90 秒版）

State 不是"函数传参"，而是一块所有节点共享的内存，每个字段是一个带 reducer 的通道。节点的返回值不是新 State，而是一个 patch——只写你想改的字段，框架按该字段的 reducer 把 patch 合并进全局 State。reducer 的签名是 (旧值, 新值) -> 合并后的值。不加 Annotated 的字段，默认 reducer 是"直接覆盖"；加了 operator.add 就是追加；messages 一般用 add_messages，它不是简单 append，而是按消息 id 做 upsert，还支持 RemoveMessage 删除。

并行写报 InvalidUpdateError 的原因是：同一个超步里两个节点都给同一字段写了值，默认的覆盖型 reducer 没法决定"谁赢"——如果让它随机覆盖，结果就不确定了，而不确定的状态会让 Checkpoint 恢复也不可复现。所以框架选择直接报错，强迫你显式声明合并语义。修法就是给这个字段加一个幂等的 reducer，或者让两个节点写不同字段。

追问	答题要点
节点返回完整 State 会怎样？	没写 reducer 的字段被整体覆盖，带 reducer 的字段（如 `messages`）会被重复追加一遍，上下文翻倍
返回 `None` 和返回 `{}` 的区别？	效果都是不改状态，但 `None` 语义更清晰：表示"这一步什么都不用更新"，可以省一次写盘
自定义 reducer 要注意什么？	① 必须处理 `None`（首次初始化）；② 必须幂等（超步会重跑）；③ 不能有副作用；④ 不要在里面调 LLM
`add_messages` 删消息要注意什么？	删 `AIMessage(tool_calls=[...])` 必须连带删对应的 `ToolMessage`，否则下一轮请求 400
State 能放大对象吗？	不建议。State 每个超步全量序列化落盘，放检索原文会让 Checkpoint 爆炸，应该只存 `doc_id`

✅ 项目关联话术：我们做并行检索时踩过一次。两个节点一个查向量库一个查 ES，都往 docs 字段写，本地单测没问题（因为 mock 太快、执行顺序固定），上线后偶发 InvalidUpdateError。排查出来是 docs 没加 reducer。修的时候我们没直接用 operator.add，而是写了个按 doc_id 去重合并的 reducer，因为两个检索源会命中同一篇文档，直接相加会把重复文档塞进上下文，既浪费 token 又让模型答案偏向重复项。

题 3：什么是超步（super-step）？Checkpoint 以超步为粒度会带来什么后果

精简口述答案（90 秒版）

超步是 LangGraph 的执行单位，一个超步 = 一批节点并行执行 + 一次屏障同步 + 一次状态合并提交。流程是：先确定这一轮要激活哪些节点，这些节点并行跑、各自返回 patch，全部跑完才在屏障处按 reducer 合并，然后写一个 Checkpoint，再进入下一个超步。

它带来三个直接后果。第一，recursion_limit 算的是超步不是节点，默认 25，而一轮 ReAct 是 agent + tools 两个超步，所以默认只够跑大约 12 轮工具。第二，超步内任一节点失败，整个超步的状态更新都不提交，恢复时会从这个超步的开头重来——所以节点必须幂等，不能"加一次计数"或"发一次邮件"就留下痕迹。第三，成功节点的结果会存进 pending_writes，恢复时不会重复调 LLM，只重跑失败的那个节点，这是省钱的关键机制，但它救不了副作用，已经扣的款框架回滚不了。

追问	答题要点
并行 5 个节点算几步？	1 步。所以靠并行能省 `recursion_limit` 配额
并行开越多越快吗？	不是。超步耗时 = 最慢节点的耗时，并行分支耗时要均衡
怎么让节点幂等？	不依赖"执行了几次"的隐含状态；计数类改成由 reducer 计算；有外部副作用的带幂等键交给下游去重
`durability` 三档怎么选？	`exit` 最快但崩了全丢；`async` 默认推荐；`sync` 最可靠，HITL 和付费流程用它
怎么确认某个超步真的落盘了？	`graph.get_state_history(config)` 列出所有 checkpoint，看 `next` 字段和 `metadata.step`

✅ 项目关联话术：我们吃过一次幂等的亏。有个节点里写了 state["retry"] + 1 然后同时调了外部通知接口，结果上游限流让这个超步失败重跑，用户收到了三条重复的"任务开始"推送。修法是把通知拆成单独节点，并且用 thread_id + checkpoint_id 生成幂等 key 交给推送服务去重。这件事之后我们定了条规范：凡是有外部副作用的操作，必须单独成节点，且必须带幂等键。

题 4：Checkpointer 和 thread_id 是什么？短期记忆和长期记忆怎么分

精简口述答案（90 秒版）

Checkpointer 是 LangGraph 的持久化层，编译时传进去，之后每个超步自动把完整 State 快照 + 执行进度写一份，按 thread_id 组织。thread_id 就是一次会话的主键：同一个 thread_id 再 invoke，框架会自动加载上次的 State，所谓"多轮记忆"不需要你手动拼历史。Checkpoint 里存的不只是 State，还有 next（下一步要跑哪些节点）、pending_writes（本超步已完成节点的结果）、versions_seen（通道版本），正是这些让断点续跑和时间旅行成为可能。

但 Checkpointer 是单 thread 的短期记忆，换 thread_id 就什么都不记得。跨会话的长期记忆要用 Store：它按 namespace（通常是 ("memories", user_id)）存自定义 KV 文档，支持前缀列举和向量语义检索，节点里通过签名注入 store 参数来读写。一句话区分：Checkpointer 存"这次对话到哪了"，Store 存"这个用户是谁"。

追问	答题要点
生产用哪个 Saver？	`PostgresSaver`（需 `.setup()` 建表）；`SqliteSaver` 只适合单机/本地；`MemorySaver` 只能测试
不传 `thread_id` 会怎样？	配了 checkpointer 却不传会直接报错；没配 checkpointer 则每次都是全新状态
Checkpoint 会不会越存越大？	会，因为是全量快照不是 diff。要靠 State 瘦身 + `messages` 裁剪 + 定期清理旧 thread
长期记忆什么时候写？	热路径（对话中实时抽取，即时但加延迟）或后台（会话结束异步总结，不影响响应）。多数场景选后台
怎么做"多用户隔离"？	`thread_id` 带上 `user_id` 前缀，且 `user_id` 必须从 `config["configurable"]` 取，绝不从 State 取

✅ 项目关联话术：我们一开始把 user_id 放在 State 里，后来做安全评审时被指出这是个越权漏洞——State 是模型能写的，理论上可以通过 prompt 注入诱导 Agent 改掉 user_id 去读别人的数据。现在所有身份和权限信息都走 config["configurable"]，模型完全看不见也改不了。这条我现在会当成红线讲：可信信息走 config，不可信信息走 State。

题 5：Human-in-the-Loop 怎么实现？`interrupt` 恢复的时候有什么坑

精简口述答案（90 秒版）

两种写法。老的是静态断点：编译时 interrupt_before=["tools"]，图跑到 tools 之前停住返回当前状态，人工看完后用 invoke(None, config) 续跑——优点是不改节点代码，缺点是断点位置固定、拿不到人工的输入内容。新的是动态中断：在节点里调 value = interrupt({"要确认的内容": ...})，它会抛出一个特殊异常把图挂起，恢复时用 invoke(Command(resume=人工输入), config)，interrupt() 就返回这个输入，执行继续。动态中断能带数据、能有条件触发（比如只有金额超过 1000 才中断），是现在的推荐写法。

最大的坑是：恢复时不是从 interrupt 那一行继续，而是整个节点从头重跑。因为 Checkpoint 粒度是超步，框架没有保存函数内部的执行位置。所以 interrupt 之前的代码会再跑一遍——如果你在 interrupt 之前发了短信、扣了款、写了库，就会执行两次。正确做法是把所有副作用放在 interrupt 之后，interrupt 之前只做纯计算。

追问	答题要点
一个节点里能有两个 `interrupt` 吗？	能但强烈不建议，重跑时顺序和返回值匹配极易出错，应该拆成两个节点
HITL 有哪几种模式？	Approve/Reject（批或驳）、Edit State（人工改状态再续）、Review Tool Call（改模型生成的工具参数）、Time Travel（回到第 N 步换分支）
必须配 checkpointer 吗？	必须。中断的本质是"把状态存下来然后退出"，没有持久化层就没法恢复
中断信息怎么给前端？	`invoke` 返回值里会有 `__interrupt__` 字段，里面是你传给 `interrupt()` 的 payload
用户一直不审批怎么办？	中断态会一直留在 checkpoint 里。需要自己加超时清理任务，并在恢复时校验业务数据是否已过期

✅ 项目关联话术：我们的退款 Agent 要求金额超过阈值必须人工确认。第一版就踩了重跑的坑——我们在 interrupt 之前写了一条"待审批"的业务流水，审批通过恢复后流水又插了一条，对账时发现重复。定位过程挺折磨的，因为本地测试是一路跑通的，只有真正中断再恢复才复现。现在我们的规范是：interrupt 必须是节点的第一行副作用边界，它上面只允许有纯函数和读操作。

题 6：条件边和 `Command` 都能决定下一步，什么时候用哪个

精简口述答案（90 秒版）

条件边是"节点写状态、路由函数读状态决定去哪"，职责分离：路由函数只能读 State 返回一个键，映射表把键翻译成目标节点。Command 是"节点自己同时干两件事"：Command(update={...}, goto="x") 一个返回值既改状态又指定跳转。

选择标准是决策和状态更新是否耦合。纯路由场景用条件边更清晰，比如"看 messages 最后一条有没有 tool_calls 决定去 tools 还是 END"，这个判断不需要写状态。但多 Agent 的 handoff 必须用 Command——交接的时候要把任务描述一起写进共享状态，拆成"先写状态再路由"会多一个超步且容易漏。另外有两件事只有 Command 能做：一是 Command(graph=Command.PARENT) 从子图跳到父图的兄弟节点，条件边做不到跨图；二是 Command(resume=...) 恢复中断。

追问	答题要点
用了 `Command` 还要 `add_conditional_edges` 吗？	不要。但要写返回类型注解 `Command[Literal["a","b"]]`，框架靠它画出连线做可视化和校验
条件边为什么要写第三个参数（目标列表/映射）？	让编译期就知道可能的目标，能画图、能在节点名拼错时提前报错；不写则运行时才炸
路由函数能返回列表吗？	能，返回节点名列表就是动态 fan-out；返回 `Send` 列表就是 map-reduce
路由函数里能调 LLM 吗？	技术上能，但不该。路由函数不落 Checkpoint、不被 `stream` 观测、不走重试策略，有 LLM 调用就该做成节点
条件边最常见的两个坑？	① 路由函数返回的键不在映射表里 → `KeyError`；② 忘了把 `END` 作为一个可能分支 → 死循环

✅ 项目关联话术：我们的客服主图里两种都在用。入口的意图分类用 Command，因为分类结果本身要写进 State 给后面的节点看，同时决定去账单/技术/人工三个分支，一个节点一个超步就搞定。而 ReAct 子图里"要不要继续调工具"用条件边，因为这个判断纯粹是读 messages 最后一条，不产生任何状态变更，用 Command 反而要多写一层包装。我面试时就是这么讲的：看决策要不要落状态。

题 7：运行时才知道要并行跑 N 个任务，怎么做？fan-in 怎么汇总

精简口述答案（90 秒版）

条件边只能在编译期已声明的节点之间选路，解决不了"运行时才知道要跑 10 份还是 100 份"。这时要用 Send：路由函数返回一个 Send 列表，每个 Send("节点名", 自定义载荷) 动态生成一个任务实例，框架把它们放在同一个超步里并行执行。

两个关键语义容易答错。第一，被 Send 调用的节点收到的参数是载荷本身，不是全局 State，所以别在里面写 state["messages"]。第二，这些节点的返回值仍然走全局 State 的 reducer 合并，所以汇总字段必须带 reducer（比如 Annotated[list, operator.add]），否则 N 个并发写同一字段直接 InvalidUpdateError。fan-in 就是在 Send 的目标节点后面挂一条普通边指向汇总节点——因为超步有屏障，所有并行任务跑完才会进汇总节点，不需要你手动等。

追问	答题要点
N 个 `Send` 消耗几次 `recursion_limit`？	1 次，它们属于同一个超步
其中一个 `Send` 任务失败会怎样？	整个超步不提交，但成功的任务结果进 `pending_writes`，恢复时只重跑失败的那个
怎么限制并发数？	`invoke(..., {"max_concurrency": 8})`，或者自己分批 `Send`，否则几百个任务同时打 LLM 会被限流
和"同一节点挂多条固定边"的区别？	固定边的分支数在编译期固定；`Send` 的任务数运行时决定，且能携带不同载荷
汇总节点怎么保证顺序？	并行结果的到达顺序不保证。要有序就在载荷里带 `index`，汇总时按 index 排序

✅ 项目关联话术：我们做长文档总结用的就是这个结构：切章节 → Send 到每章的总结节点 → 汇总节点合并。踩过两个坑。一是最初 summaries 字段忘了加 reducer，本地测两章没事，线上一篇 30 章的文档直接 InvalidUpdateError。二是没限并发，30 个 Send 同时打模型被限流，一半失败。现在固定 max_concurrency=8，并且给每个载荷带上章节序号，汇总时按序号排序，避免摘要顺序乱掉。

题 8：Agent 无限循环怎么防？`recursion_limit` 设大点不就行了

精简口述答案（90 秒版）

不行，recursion_limit 只是最后一道保险丝，而且它抛 GraphRecursionError、没有降级答案，直接扔给用户就是 500。要四道闸门一起上。

第一道是模型自然收尾，模型不再返回 tool_calls 就走向 END，这是正常路径。第二道是 recursion_limit，注意它算的是超步不是节点，默认 25，一轮 ReAct 两个超步，所以默认只够约 12 轮。第三道是业务预算，在 State 里记 token、费用、调用次数、耗时，超了就路由到一个"强制收尾"节点——用 tool_choice="none" 再请求一次，让模型用已有信息作答，这样用户拿到的是降级答案而不是报错。第四道是重复动作检测，把"工具名 + 规范化后的参数"做成指纹存进 State，发现重复就把这次调用拦掉、回一条"这个查询你已经做过了，结果是 X，请换个思路"的 ToolMessage——大多数绕圈不是 bug 而是模型拿着同一个参数反复查。

追问	答题要点
为什么说调大 `recursion_limit` 是错的？	它不治病，只是把烧钱时间拉长；真正的病因通常是条件边写反或工具描述太差导致模型反复试错
怎么捕获并降级？	`try/except GraphRecursionError` 里用 `get_state(config)` 拿到当前状态，拼一个降级回复
重复检测的指纹怎么算？	工具名 + 参数 JSON 按 key 排序后序列化再 hash；时间戳之类的易变字段要剔除
怎么区分"合理重试"和"绕圈"？	合理重试通常参数会变（换关键词、换页码）；参数完全相同的重复就是绕圈
预算怎么记才不出错？	记在 State 里并给个幂等的 reducer（取 max 或由节点算绝对值），不要写 `+1`，因为超步会重跑

✅ 项目关联话术：我们灰度期真出过一次。模型拿着同一个报销单号连查了十几次审批流，每次都觉得结果"不够详细"要再查一次，一个会话烧了几千 token。加了重复指纹检测之后，第二次重复调用直接被拦下并回提示，模型立刻换了思路去查制度文档。现在我们的 force_answer 降级节点也是标配——宁可给用户一个"基于已查到的信息，我判断是 X，但建议人工复核"，也不要给 500。

题 9：图太大怎么拆子图？父子 State 怎么对接

精简口述答案（90 秒版）

超过 15 个节点就该拆。拆子图有两种挂法，区别全在状态怎么对接。

第一种是直接把编译后的子图当节点挂上去：parent.add_node("research", sub_graph)。前提是父子 State 有同名 key，同名字段自动透传进去、写回时按父图的 reducer 合并。好处是子图有独立的 checkpoint_ns，父图 stream(..., subgraphs=True) 能看到子图内部每一步，子图里的 interrupt 也能穿透到最外层。第二种是包装成普通函数：自己写 sub_graph.invoke(映射后的输入) 再把输出映射回父 State。好处是映射完全可控、父子 schema 可以完全不同；代价是子图被当成一次黑盒调用，内部的中断和流式事件穿不出来，中间态也不落父图的 Checkpoint。

选择标准：子流程需不需要被观测、中断、恢复。需要就直接挂图，不需要就包成函数。

追问	答题要点
子图写了父图没有的 key 会怎样？	被静默丢弃，不报错 —— 这是最难查的一类 bug，要用 `stream(..., subgraphs=True)` 对照子图 updates
子图能单独做时间旅行吗？	能，直接挂图的方式下子图有自己的 `checkpoint_ns`，可以定位到子图内部的 checkpoint
子图层级能嵌多深？	技术上无限，但每多一层就多一次完整状态序列化，建议控制在 2 层内
子图要不要单独传 checkpointer？	不要。编译子图时不传，它会复用父图的 checkpointer；传了反而容易出现两套持久化
子图和"把流程做成工具"有什么区别？	工具是模型决定调不调；子图是你用图结构硬编码的流程，确定性更高

✅ 项目关联话术：我们主图里的检索流程（改写 query → 多路检索 → 重排 → 压缩）拆成了子图，用的是直接挂图的方式。关键收益是排障：以前检索效果差只能看最终答案猜，现在 stream(subgraphs=True) 能看到改写后的 query 长什么样、每一路召回了几篇、重排后留了哪几篇。另外踩过一次静默丢弃的坑——子图里新加了个 rerank_score 字段忘了同步到父 State，父图拿到的结果里一直没有分数，也不报错，查了半天。

题 10：多智能体有哪几种拓扑？你们怎么选的，handoff 怎么实现

精简口述答案（90 秒版）

四种。Network 是全连接，每个 Agent 都能跳到任意 Agent，最灵活但 token 爆炸、容易绕圈，基本不用于生产。Supervisor 是中心调度，一个 Supervisor 用结构化输出决定下一个 Agent，子 Agent 做完回边指向它——可控、好调试、能在一个地方统一做预算和权限控制，缺点是 Supervisor 是单点、prompt 会越来越长。Hierarchical 是分层，Supervisor 的下级也是 Supervisor，能撑几十个 Agent，代价是层级深、延迟高。Swarm 是去中心交接，谁接手就由谁一直对话直到再交接，对话体验最连贯、没有中心开销，但全局可观测性差。

handoff 的工业写法不是在 prompt 里写"请转给 X"，而是把交接做成一个工具，工具返回 Command：Command(goto="目标Agent", graph=Command.PARENT, update={...})。三个细节不能省：必须返回一条 ToolMessage 否则留下悬空 tool_call 导致下一轮 400；必须带 graph=Command.PARENT 因为 Agent 自己是子图、目标是父图的兄弟节点；必须显式传任务描述，不要让下一个 Agent 从几十条历史里猜任务。

追问	答题要点
Supervisor 怎么保证输出合法的目标名？	`with_structured_output` 配 `Literal` 枚举，而不是让它自由输出字符串再做匹配
多 Agent 的历史消息要不要全共享？	不建议全共享。常见做法是只共享"结论类"消息，各 Agent 的中间推理留在自己的子图 State 里，否则上下文爆炸
多 Agent 比单 Agent 一定好吗？	不一定。角色一多、交接一多，错误传播和延迟都放大。工具超过 15 个、职责明显互斥时才值得拆
怎么防止两个 Agent 来回踢皮球？	State 里记交接链，同一对 Agent 交接超过 N 次就强制路由到人工或收尾节点
Supervisor 的 prompt 越来越长怎么办？	升级成 Hierarchical，按业务域分组，每个 Supervisor 只认自己域里的三五个 Agent

✅ 项目关联话术：我们选的是 Supervisor。原因很具体：一是要集中做预算控制和越权拦截，中心节点最容易埋这些检查；二是运维要看"当前在哪个 Agent"，Supervisor 结构下这个信息天然在 State 里。踩过的坑是来回踢皮球——账单 Agent 觉得是技术问题、技术 Agent 觉得是账单问题，互相交接了五六轮。现在 State 里记了 handoff_chain，同一对 Agent 交接超过两次就直接转人工。

题 11：LangGraph 的流式输出怎么对接前端？几种 stream_mode 分别干什么

精简口述答案（90 秒版）

前端要的"打字机效果"和"执行进度条"是两种完全不同的流，混淆是对接时最常见的事故。stream_mode 有五种：values 吐每个超步后的完整 State，适合调试；updates 吐每个节点返回的 patch，形如"节点名 → 增量"，这是做执行进度条用的；messages 吐 (token, metadata) 二元组，是 LLM token 级的，metadata 里有 langgraph_node 可以按节点过滤，这是做打字机用的；custom 是节点里用 get_stream_writer() 主动推的任意数据，适合推工具执行进度比如"正在检索第 3/5 篇"；debug 吐任务开始结束和 Checkpoint 细节，只用于排障。

实际对接一般同时订阅多个：stream_mode=["updates", "messages"]，这时 chunk 变成 (mode, data) 二元组，按 mode 分发成不同的 SSE 事件类型。

追问	答题要点
`messages` 模式捕获不到 token，为什么？	① 节点里绕过 LangChain 直接用厂商 SDK 裸调，框架看不见；② Python < 3.11 异步环境下 `config` 没透传给 `ainvoke`
怎么只把主 Agent 的 token 推给用户？	用 metadata 里的 `langgraph_node` 过滤，把路由/摘要等内部节点的 token 丢掉
子图内部的流式事件怎么拿？	`stream(..., subgraphs=True)`，chunk 会带上 namespace 路径
中断发生时流会怎样？	流正常结束，最后能拿到 `__interrupt__` 的 payload，前端据此渲染审批弹窗
工具执行很慢，前端一直空白怎么办？	用 `custom` 模式在工具内部推进度，而不是让用户干等

✅ 项目关联话术：我们的 SSE 接口定了三种事件：step（来自 updates，渲染"正在检索 → 正在生成"的步骤条）、token（来自 messages，只放行 langgraph_node == "agent" 的）、progress（来自 custom，工具里推的细粒度进度）。踩过一个坑是最初没按节点过滤 token，结果意图分类节点和摘要节点的输出也被当成答案打到了屏幕上，用户看到一串莫名其妙的 JSON。

题 12：为什么说 LangGraph 的节点必须幂等？有副作用的操作该怎么处理

精简口述答案（90 秒版）

因为 LangGraph 的事务单位是超步，不是节点。超步内任一节点失败，这一步的状态更新全部不提交，恢复时从超步开头重来；interrupt 恢复时也不是从 interrupt 那一行继续，而是整个节点从头重跑。这两件事合起来意味着：任何节点都可能被执行多次，而框架只能回滚 State，回滚不了你发出去的短信和扣掉的钱。

处理副作用有三条规矩。第一，有副作用的操作必须单独成节点，不要和纯计算混在一起，否则重跑时计算和副作用一起重来。第二，必须带幂等键交给下游去重，比如用 thread_id + checkpoint_id + 业务动作 生成 token，让支付/推送服务自己判重——这是唯一可靠的办法，因为分布式下你没法做真正的回滚。第三，副作用必须放在 interrupt 之后，interrupt 之前只允许纯函数和读操作。另外计数类状态不要在节点里写 x + 1，要么让 reducer 去算，要么由节点算出绝对值，否则重跑就多加一次。

追问	答题要点
`pending_writes` 是干什么的？	存本超步已成功节点的结果，恢复时直接复用、不重复调 LLM —— 这是省钱的关键，但它救不了副作用
节点里重试和 `retry_policy` 有什么区别？	自己写的重试次数不进 Checkpoint，恢复后计数归零；`retry_policy` 由框架管，还能配退避和 `retry_on` 白名单
`cache_policy` 为什么经常不命中？	缓存键由节点输入算，输入里有 `datetime.now()` 或随机 id 就永远不会命中
怎么测幂等性？	集成测试里显式制造一次中断/异常再恢复，断言外部调用只发生一次（用 mock 记调用次数）
reducer 也要幂等吗？	要。超步重跑会重新合并，`operator.add` 配非幂等节点就会出现重复项，所以列表类 reducer 最好带去重

✅ 项目关联话术：这条是我们血的教训攒出来的规范。前面说的重复推送和重复流水两次事故，根因都是同一个：把副作用和纯计算写在了一个节点里，还假设节点只会执行一次。现在我们 code review 有一条硬性检查项——节点里有没有外部写操作，有的话有没有幂等键、是不是单独成节点、是不是在 interrupt 之后。我面试讲这段的时候，面试官追问过"为什么不用事务回滚"，我的回答是：调用的是第三方 HTTP 接口，根本没有分布式事务可用，幂等键 + 下游去重是成本最低且真正可靠的方案。

题 13：用户上一条还在生成就又发了一条，你们怎么处理（double-texting）

精简口述答案（90 秒版）

这是真实流量里一定会遇到、Demo 里永远遇不到的问题：同一个 thread_id 出现两个并发 run，两个 run 同时往同一份 checkpoint 写，状态必然错乱。处理方式有四种语义：reject 直接拒掉第二条并提示"上一条还在处理"；enqueue 让第二条排队，等第一个 run 跑完再接着跑；rollback 取消第一个 run 并回滚它写过的状态，只认第二条；interrupt 中断第一个 run 但保留它已完成的超步，再跑第二条。

选型看产品语义：用户在补充信息就用 enqueue，用户明显在改主意（"算了，换成 B"）就用 rollback 或 interrupt，严肃的写操作场景宁可 reject。实现上，如果用 LangGraph Server 这就是一个 multitask_strategy 参数；自己包 FastAPI 的话要手动做两件事：按 thread_id 加分布式锁做串行化，以及把运行中的 task 存起来以便取消。关键是取消点一定落在超步边界上，所以取消后的状态是干净的——上一个超步已提交，当前超步整体不提交。

追问	答题要点
锁为什么按 `thread_id` 而不是按用户？	按用户加锁会导致同一个人开两个会话互相阻塞
`rollback` 怎么定位回滚点？	`get_state_history` 找到该 run 开始前的 `checkpoint_id`，带着它重新 invoke
取消一个跑到一半的 run 会不会留脏状态？	State 不会脏（超步未提交），但已经发出去的副作用回滚不了，所以写操作节点要么幂等要么走 `reject`
用户点"停止生成"是同一套机制吗？	是。取消任务即可，checkpoint 停在上一个超步，下次 `invoke(None, config)` 能续上
前端怎么感知被拒/被中断？	必须是独立的 SSE 事件类型，否则用户只看到"发出去没反应"

✅ 项目关联话术：我们默认用 interrupt 语义——新消息进来取消正在跑的 run，但保留已完成的超步，用户追问时之前检索到的文档不会白跑。写操作类的流程单独改成 reject，因为那种场景下"中途取消了但钱扣了没扣"的歧义比让用户等一会儿更难解释。最早我们完全没管这事，线上出现过用户连点三次发送、三个 run 抢同一个 thread，最后 messages 里工具消息和 AI 消息交错错位，直接 400。

题 14：LangGraph 写完的图怎么上线？多副本部署要注意什么

精简口述答案（90 秒版）

compile() 出来的只是一个进程内对象，上线有三条路。第一条是自己包一层 Web 框架，FastAPI 里 await graph.ainvoke，流式就转发 astream——好处是复用现有网关、鉴权、日志体系，代价是 thread 生命周期、并发控制、SSE 协议全要自己写。第二条是自托管 LangGraph Server，写一个 langgraph.json 声明图入口，langgraph dev 本地起、打镜像上线，白捡 HTTP API、流式、HITL 和定时任务，还有 Assistant 这种"同一张图 + 不同配置 = 不同实例"的配置化能力，代价是多一个服务要运维，依赖 Postgres 和 Redis。第三条是托管平台，适合快速验证，但数据出域要先过合规。

多副本部署的关键是：状态全在 Postgres，所以副本本身是无状态的，可以随意水平扩。要注意四件事：PostgresSaver 首次必须跑 .setup() 建表和迁移；连接池要给够，因为每个超步都要读写 checkpoint，池子小了表现就是"延迟莫名翻倍"；异步服务必须用 AsyncPostgresSaver，混用同步 Saver 会阻塞事件循环；同一个 thread_id 的并发请求要做串行化，这就回到 double-texting 那套锁。

追问	答题要点
为什么副本可以无状态？	所有状态落在 checkpointer 后端，副本只负责执行超步，不持有会话内存
需要做 thread 粘性（session affinity）吗？	不需要粘性，但需要同 thread 串行化；两者是不同的问题
存储会不会膨胀？	会。checkpoint 是全量快照不是 diff，必须有 TTL 或定时清理，State 里也只存引用不存原文
灰度发布怎么做？	Assistant 带版本号可以按配置灰度；自建的话把图版本写进 `configurable`，按流量比例分发
本地怎么调试一张复杂图？	`langgraph dev` 带调试面板；纯代码侧用 `get_graph(xray=1).draw_ascii()` 加 `stream(..., subgraphs=True)`

✅ 项目关联话术：我们走的是第一条路，因为公司的鉴权和网关体系是统一的，没法为一个服务单开一套。代价确实付了——thread 清理、同 thread 加锁、SSE 事件协议都是自己实现的，大约两百行胶水代码。换来的好处是整条链路和现有监控体系打通了。踩过最蠢的坑是上线时忘了跑 PostgresSaver.setup()，第一个请求直接报表不存在，后来把它做成了启动时的迁移步骤。

✍️ 手撕代码【核心实现 + 边界 Case 自测清单 + 变种拓展题】

题 1：用 LangGraph 手写带四道闸门的 ReAct 图

要求

不用 create_react_agent，手写 agent / tools 两个节点和回边。
条件边必须覆盖三条出路：继续调工具、强制收尾、自然结束。
State 里记账 token 与迭代次数，预算超了路由到 force_answer 降级节点。
工具抛异常不能让图崩，要把错误作为 ToolMessage 回传给模型。
配 checkpointer，支持同一 thread_id 的多轮对话。

python

import json, operator
from typing import Annotated, TypedDict, Literal
from langchain_core.messages import AIMessage, ToolMessage, SystemMessage
from langgraph.graph import StateGraph, START, END, add_messages
from langgraph.checkpoint.memory import MemorySaver


class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    iters: Annotated[int, operator.add]      # reducer 累加，节点只返回增量 1
    used_tokens: Annotated[int, operator.add]
    fingerprints: Annotated[list, operator.add]


MAX_ITERS, MAX_TOKENS = 6, 40_000


def build_agent(llm, tool_map, system_prompt):
    bound = llm.bind_tools(list(tool_map.values()))

    # ---------- 节点 1：模型决策 ----------
    def agent_node(state: AgentState):
        msgs = [SystemMessage(content=system_prompt)] + state["messages"]
        ai: AIMessage = bound.invoke(msgs)
        usage = (ai.usage_metadata or {}).get("total_tokens", 0)
        return {"messages": [ai], "iters": 1, "used_tokens": usage}

    # ---------- 节点 2：执行工具（并行 + 异常隔离 + 重复检测）----------
    def tools_node(state: AgentState):
        ai = state["messages"][-1]
        seen = set(state["fingerprints"])
        out, new_fp = [], []
        for call in ai.tool_calls:
            fp = call["name"] + "|" + json.dumps(call["args"], sort_keys=True,
                                                 ensure_ascii=False)
            if fp in seen:                              # ← 闸门 4：重复动作
                out.append(ToolMessage(
                    content="该查询你已执行过，结果未变，请换思路或直接作答",
                    tool_call_id=call["id"], name=call["name"]))
                continue
            new_fp.append(fp)
            fn = tool_map.get(call["name"])
            if fn is None:
                text = f"工具 {call['name']} 不存在，可用工具：{list(tool_map)}"
            else:
                try:
                    text = str(fn.invoke(call["args"]))
                except Exception as e:                   # ← 单点失败不炸整图
                    text = f"工具执行失败：{type(e).__name__}: {e}"
            out.append(ToolMessage(content=text[:2000],
                                   tool_call_id=call["id"], name=call["name"]))
        return {"messages": out, "fingerprints": new_fp}

    # ---------- 节点 3：强制收尾（降级答案，不抛异常）----------
    def force_answer(state: AgentState):
        msgs = ([SystemMessage(content="已达上限，请基于已有信息直接作答，"
                                       "并说明哪些部分未能核实")]
                + state["messages"])
        return {"messages": [llm.invoke(msgs)]}          # 不 bind_tools = 不可能再调工具

    # ---------- 条件边：三条出路 ----------
    def route(state: AgentState) -> Literal["tools", "force_answer", "__end__"]:
        if state["iters"] >= MAX_ITERS or state["used_tokens"] >= MAX_TOKENS:
            return "force_answer"                        # ← 闸门 3：业务预算
        last = state["messages"][-1]
        if getattr(last, "tool_calls", None):
            return "tools"
        return END                                       # ← 闸门 1：模型自然收尾

    b = StateGraph(AgentState)
    b.add_node("agent", agent_node)
    b.add_node("tools", tools_node)
    b.add_node("force_answer", force_answer)
    b.add_edge(START, "agent")
    b.add_conditional_edges("agent", route, ["tools", "force_answer", END])
    b.add_edge("tools", "agent")                         # 回边 → 形成循环
    b.add_edge("force_answer", END)
    return b.compile(checkpointer=MemorySaver())


# 使用：recursion_limit 是闸门 2
graph = build_agent(llm, tool_map, "你是企业知识库助手")
cfg = {"configurable": {"thread_id": "t1"}, "recursion_limit": 20}
print(graph.invoke({"messages": [("user", "我上月报销卡在哪了")]}, cfg))

边界 Case 自测清单

场景	预期结果
问"你好"（无需工具）	`agent` 一次即走 END，`iters == 1`，无 `ToolMessage`
一轮返回 2 个 tool_calls	产出 2 条 `ToolMessage`，`tool_call_id` 与 `tool_calls` 一一对应
工具内 `raise ValueError`	图不崩，错误文本进 `ToolMessage`，模型下一轮换思路
模型编造不存在的工具名	回传"工具不存在 + 可用列表"，不抛 `KeyError`
同参数重复调用同一工具	第二次被拦，返回提示文案且不真正执行
构造需 8 轮的任务	第 6 轮后路由到 `force_answer`，返回降级答案而非 `GraphRecursionError`
同一 `thread_id` 追问"那第二步呢"	自动带上历史 `messages`，无需手动拼接
换一个 `thread_id`	历史清空，等于新会话
节点内把 `iters` 写成 `state["iters"] + 1`	超步重跑时会多加一次 → 必须返回增量 `1` 交给 reducer

🚀 变种拓展题

把 tools_node 改成 async，同一轮的多个 tool_calls 用 asyncio.gather 并发，并为每个工具加 5s 超时。
给 tools_node 加 RetryPolicy，只对限流和超时重试，带指数退避和 jitter。
把重复指纹从 list 改成带去重的自定义 reducer，避免 fingerprints 无限膨胀。
增加上下文裁剪节点插在 tools → agent 之间，超过 60k token 就摘要压缩。
把 force_answer 的触发原因写进 State，并在最终答案里向用户说明降级原因。

题 2：手写一个支持并行写入的幂等 reducer 集合

要求

实现 merge_docs：按 id 去重合并，右侧覆盖左侧，必须处理 None。
实现 keep_best：并行多路打分只保留最高分及其来源。
实现 dedup_append：列表追加但自动去重，保持首次出现顺序。
实现 counter_max：计数类字段用"取最大值"语义，保证超步重跑不虚增。
全部 reducer 必须满足：幂等（同一 patch 合并两次结果不变）、无副作用、可交换顺序不敏感（或明确说明不满足）。

python

from typing import Annotated, TypedDict, Any


def merge_docs(left: list | None, right: list | None) -> list:
    """按 doc id 去重合并；右侧同 id 覆盖左侧。幂等。"""
    merged: dict[str, dict] = {d["id"]: d for d in (left or [])}
    for d in (right or []):
        merged[d["id"]] = d
    return list(merged.values())


def keep_best(left: dict | None, right: dict | None) -> dict:
    """保留 score 最高的那条；并行写入顺序无关。"""
    cands = [x for x in (left, right) if x]
    if not cands:
        return {}
    return max(cands, key=lambda x: x.get("score", float("-inf")))


def dedup_append(left: list | None, right: list | None) -> list:
    """追加去重，保持首次出现顺序。幂等。"""
    seen, out = set(), []
    for item in (left or []) + (right or []):
        key = item if isinstance(item, (str, int)) else repr(item)
        if key not in seen:
            seen.add(key)
            out.append(item)
    return out


def counter_max(left: int | None, right: int | None) -> int:
    """计数取最大值：节点返回绝对值，超步重跑不会虚增。"""
    return max(left or 0, right or 0)


class State(TypedDict):
    docs: Annotated[list, merge_docs]
    best: Annotated[dict, keep_best]
    visited: Annotated[list, dedup_append]
    depth: Annotated[int, counter_max]
    answer: str                        # 单一节点写，无需 reducer

边界 Case 自测清单

场景	预期结果
首次写入（`left is None`）	不抛 `TypeError`，返回右侧内容
节点返回 `None` 字段值	reducer 内 `or` 兜底，等价于空集
同一 patch 合并两次	结果与合并一次完全相同（幂等验证）
交换左右顺序	`merge_docs` 结果的 id 集合相同、`keep_best`/`counter_max` 完全相同
两路检索命中同一 `doc_id`	只保留一份，不会重复占用上下文
`keep_best` 两侧 score 相等	行为确定（`max` 取先出现的），不随机
`dedup_append` 元素是 dict	用 `repr` 兜底做 key，不抛 unhashable 错误
节点写 `{"depth": state["depth"] + 1}`	配 `counter_max` 后重跑不虚增；若配 `operator.add` 则会错

🚀 变种拓展题

给 merge_docs 加容量上限：超过 50 篇时按 score 降序裁剪，避免 Checkpoint 膨胀。
写一个 reducer_factory(key_fn, pick_fn) 工厂，把上面四个 reducer 统一成一行配置。
实现 ttl_append：每个元素带写入时间，合并时自动丢弃超过 N 分钟的旧元素。
写一个 pytest 参数化测试，对所有 reducer 自动校验「幂等 + None 安全 + 顺序无关」三条性质。
实现一个会故意报错的 reducer：检测到并行写入语义冲突时抛出带字段名的明确异常，替代框架的笼统报错。

题 3：实现写操作的人工审批闸门（含幂等与重跑安全）

要求

只有"写类工具"且金额超过阈值才触发中断，读类工具自动执行。
用动态 interrupt()，把待确认的工具名、参数、风险等级传给前端。
支持三种人工决策：approve（批准）、edit（改参数后执行）、reject（驳回并告知模型）。
节点重跑安全：interrupt 之前只允许纯计算，所有副作用在之后。
真正执行写操作时带幂等键，防止恢复重跑导致重复下单。

python

from langgraph.types import interrupt, Command
from langchain_core.messages import ToolMessage

WRITE_TOOLS = {"refund", "create_order", "send_sms"}
AMOUNT_THRESHOLD = 1000


def approval_tools_node(state, config):
    ai = state["messages"][-1]
    out = []
    for call in ai.tool_calls:
        name, args = call["name"], dict(call["args"])

        # ===== interrupt 之前：只做纯计算，重跑无副作用 =====
        need_review = name in WRITE_TOOLS and args.get("amount", 0) >= AMOUNT_THRESHOLD

        if need_review:
            decision = interrupt({                      # ← 图在此挂起并落盘
                "type": "tool_approval",
                "tool": name,
                "args": args,
                "risk": "high",
                "hint": f"将执行 {name}，金额 {args.get('amount')} 元，请确认",
            })
            action = decision.get("action")
            if action == "reject":
                out.append(ToolMessage(
                    content=f"人工驳回：{decision.get('reason', '未说明原因')}。"
                            f"请不要重试该操作，改为向用户解释。",
                    tool_call_id=call["id"], name=name))
                continue
            if action == "edit":
                args = decision.get("args", args)       # 人工修正后的参数

        # ===== interrupt 之后：唯一允许出现副作用的位置 =====
        idem_key = f"{config['configurable']['thread_id']}:{call['id']}:{name}"
        try:
            result = TOOLS[name].invoke({**args, "idempotency_key": idem_key})
            text = str(result)
        except Exception as e:
            text = f"执行失败：{type(e).__name__}: {e}"
        out.append(ToolMessage(content=text, tool_call_id=call["id"], name=name))

    return {"messages": out}


# ---------- 调用侧：两段式 ----------
cfg = {"configurable": {"thread_id": "t1"}}
res = graph.invoke({"messages": [("user", "给订单 A123 退款 2000")]}, cfg)

if "__interrupt__" in res:
    payload = res["__interrupt__"][0].value            # 推给前端渲染审批弹窗
    # 人工点了"修改金额后批准"
    res = graph.invoke(
        Command(resume={"action": "edit", "args": {"order": "A123", "amount": 1500}}),
        cfg,
    )

边界 Case 自测清单

场景	预期结果
只读工具（如 `search`）	不触发中断，直接执行
写工具但金额 500	低于阈值，不触发中断
写工具金额 2000	返回值含 `__interrupt__`，图停在该节点，状态已落盘
恢复时传 `approve`	按原参数执行，`ToolMessage` 回传成功结果
恢复时传 `edit`	按人工修正后的参数执行
恢复时传 `reject`	不执行任何写操作，回传驳回原因且提示模型不要重试
一轮里同时有 1 个读 + 1 个写工具	两条 `ToolMessage` 都必须产出，缺一条下一轮请求 400
在 `interrupt` 之前插一条业务流水再测恢复	流水会插两条 —— 验证"重跑从节点开头开始"这个结论
恢复后再查 `get_state(cfg)`	`next` 已推进，中断态清除
未配 `checkpointer`	直接报错，中断必须依赖持久化
用户迟迟不审批	中断态永久留在 checkpoint，需要额外的超时清理任务

🚀 变种拓展题

支持"批量审批"：一轮里多个写操作合并成一次 interrupt，人工一次性勾选批准哪些。
加审批超时：恢复时校验 checkpoint 的创建时间，超过 30 分钟则强制驳回并要求重新发起。
把审批决策与操作者 ID 写入审计日志节点，满足合规留痕要求。
实现 Time Travel 版撤销：用 get_state_history 找到审批前的 checkpoint，带 checkpoint_id 重跑换一个决策。
把"风险分级"抽成独立节点，用小模型 + 规则双重判定，输出 low/medium/high 决定是否中断。

题 4：用 `Send` 实现长文档的 map-reduce 摘要

要求

运行时根据章节数动态 fan-out，不能在编译期固定分支数。
每个 map 任务只拿自己那一章（自定义载荷），不拿全局 State。
fan-in 字段必须有幂等 reducer，且结果要能按原章节顺序还原。
限制并发，避免几十个任务同时打模型被限流。
单个章节失败不能让整篇失败，要降级为占位摘要。

python

import operator
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send


def merge_by_index(left: list | None, right: list | None) -> list:
    """按 index 去重合并，天然幂等；汇总时再排序。"""
    m = {x["index"]: x for x in (left or [])}
    for x in (right or []):
        m[x["index"]] = x
    return list(m.values())


class State(TypedDict):
    doc: str
    chapters: list[str]
    parts: Annotated[list, merge_by_index]
    final: str


def split_node(state: State):
    chapters = [c.strip() for c in state["doc"].split("\n## ") if c.strip()]
    return {"chapters": chapters}


# ---------- map：动态 fan-out ----------
def fan_out(state: State):
    return [
        Send("summarize_one", {"index": i, "chapter": c})
        for i, c in enumerate(state["chapters"])
    ]


# ---------- 注意：payload 不是全局 State ----------
def summarize_one(payload: dict):
    try:
        text = llm.invoke(f"用 3 句话总结：\n{payload['chapter'][:4000]}").content
    except Exception as e:
        text = f"[该章摘要生成失败：{type(e).__name__}]"      # 降级占位，不炸整篇
    return {"parts": [{"index": payload["index"], "text": text}]}


# ---------- reduce：屏障保证所有 map 跑完才进来 ----------
def reduce_all(state: State):
    ordered = sorted(state["parts"], key=lambda x: x["index"])
    joined = "\n".join(f"{i+1}. {p['text']}" for i, p in enumerate(ordered))
    return {"final": llm.invoke(f"把以下分章摘要合并成一篇连贯综述：\n{joined}").content}


b = StateGraph(State)
b.add_node("split", split_node)
b.add_node("summarize_one", summarize_one)
b.add_node("reduce", reduce_all)
b.add_edge(START, "split")
b.add_conditional_edges("split", fan_out, ["summarize_one"])   # ← 声明可能目标
b.add_edge("summarize_one", "reduce")
b.add_edge("reduce", END)
graph = b.compile()

graph.invoke({"doc": long_text}, {"max_concurrency": 8})        # ← 限并发

边界 Case 自测清单

场景	预期结果
文档只有 1 章	正常走通，`parts` 长度为 1
文档有 0 章（空串）	`fan_out` 返回空列表，应直接走 `reduce` 或提前走 END，不报错
30 章文档	只消耗 1 次 `recursion_limit`，不是 30 次
`parts` 去掉 reducer	并发写抛 `InvalidUpdateError`（用来验证 reducer 的必要性）
第 5 章 LLM 抛异常	该章为占位文本，其余章节正常，整篇不失败
不设 `max_concurrency`	大文档下大量并发请求触发限流
map 任务里误写 `payload["chapters"]`	`KeyError` —— 验证"载荷不是全局 State"
乱序到达	`reduce` 按 `index` 排序后输出顺序稳定
同一任务被重跑	`merge_by_index` 按 index 覆盖，不产生重复条目

🚀 变种拓展题

改成两级 map-reduce：章节过多时先按 10 章一组做中间摘要，再合并，避免 reduce 的 prompt 超长。
给每个 map 任务加独立的 RetryPolicy 和超时，失败重试 2 次后才降级。
用 custom 流式模式推进度："已完成 7/30 章"。
把 reduce 拆成"合并 + 质检"两个节点，质检不通过则重新合并（带最大重试次数）。
让 fan_out 支持优先级：先处理前 5 章拿到快速预览返回给用户，其余章节后台继续。

题 5：手写 Supervisor 多智能体（含 handoff 与踢皮球防护）

要求

Supervisor 用结构化输出决定下一个 Agent，目标名必须是 Literal 枚举而非自由字符串。
子 Agent 用子图实现，做完回边指向 Supervisor。
支持 Agent 之间直接 handoff（工具返回 Command），跳过 Supervisor。
记录交接链，同一对 Agent 来回超过 2 次强制转人工。
全局预算：总超步、总 token 任一超限就路由到收尾节点。

python

import operator
from typing import Annotated, TypedDict, Literal
from pydantic import BaseModel, Field
from langchain_core.messages import ToolMessage, AIMessage
from langchain_core.tools import tool, InjectedToolCallId
from langgraph.graph import StateGraph, START, END, add_messages
from langgraph.types import Command
from langgraph.prebuilt import InjectedState

AGENTS = ("researcher", "coder", "writer")


class SupState(TypedDict):
    messages: Annotated[list, add_messages]
    task_desc: str
    handoffs: Annotated[list, operator.add]     # 交接链，形如 ["sup>coder", "coder>writer"]
    tokens: Annotated[int, operator.add]


class Route(BaseModel):
    next: Literal["researcher", "coder", "writer", "FINISH"] = Field(
        description="下一个处理人；信息已足够时返回 FINISH")
    task: str = Field(description="交给该 Agent 的具体任务描述")


# ---------- Supervisor：用 Command 同时写状态 + 跳转 ----------
def supervisor(state: SupState) -> Command:
    if state["tokens"] > 60_000 or len(state["handoffs"]) > 12:
        return Command(goto="finalize")                 # 全局预算闸

    if _ping_pong(state["handoffs"], limit=2):          # 踢皮球防护
        return Command(goto="human", update={"handoffs": ["sup>human"]})

    r = llm.with_structured_output(Route).invoke(SUP_PROMPT + state["messages"])
    if r.next == "FINISH":
        return Command(goto="finalize")
    return Command(
        goto=r.next,
        update={"task_desc": r.task, "handoffs": [f"sup>{r.next}"]},
    )


def _ping_pong(chain: list[str], limit: int) -> bool:
    """检测 A>B、B>A 反复出现"""
    pairs = [tuple(sorted(h.split(">"))) for h in chain if ">" in h]
    return any(pairs.count(p) > limit * 2 for p in set(pairs))


# ---------- handoff 工具：让 Agent 直接交接 ----------
def make_handoff(target: str, source: str):
    @tool(f"transfer_to_{target}", description=f"把任务交接给 {target}")
    def _handoff(
        task: str,
        state: Annotated[dict, InjectedState],
        tool_call_id: Annotated[str, InjectedToolCallId],
    ) -> Command:
        return Command(
            goto=target,
            graph=Command.PARENT,                       # ← 跳到父图的兄弟 Agent
            update={
                # 必须补 ToolMessage，否则悬空 tool_call 导致下一轮 400
                "messages": [ToolMessage(content=f"已交接给 {target}",
                                         tool_call_id=tool_call_id)],
                "task_desc": task,                  # 显式交接任务，别让它猜
                "handoffs": [f"{source}>{target}"],  # source 要传进来，否则链路记不准
            },
        )
    return _handoff


# ---------- 子 Agent：子图，做完回 Supervisor ----------
def make_agent(name: str, tools: list):
    sub = build_react_subgraph(llm, tools + [make_handoff(t, name) for t in AGENTS
                                             if t != name])

    def node(state: SupState) -> Command:
        out = sub.invoke({"messages": state["messages"],
                          "task_desc": state["task_desc"]})
        return Command(
            goto="supervisor",                          # 回边
            update={"messages": [AIMessage(content=out["messages"][-1].content,
                                           name=name)],
                    "tokens": out.get("tokens", 0)},
        )
    return node


b = StateGraph(SupState)
b.add_node("supervisor", supervisor)
for a, ts in [("researcher", RESEARCH_TOOLS), ("coder", CODE_TOOLS),
              ("writer", [])]:
    b.add_node(a, make_agent(a, ts))
b.add_node("human", human_takeover_node)
b.add_node("finalize", finalize_node)
b.add_edge(START, "supervisor")
b.add_edge("finalize", END)
b.add_edge("human", END)
graph = b.compile(checkpointer=checkpointer)

边界 Case 自测清单

场景	预期结果
简单问题（无需任何 Agent）	Supervisor 第一次就返回 `FINISH`，走 `finalize`
Supervisor 输出非法目标名	用 `Literal` 约束后不可能出现；去掉约束时应有兜底分支而非 `KeyError`
researcher 直接 handoff 给 writer	不经过 Supervisor，`handoffs` 多一条 `researcher>writer`
handoff 工具漏返回 `ToolMessage`	下一轮 LLM 请求 400（可手动删掉验证）
handoff 漏写 `graph=Command.PARENT`	在子图内找不到目标节点，报错
coder 与 writer 来回交接 5 次	`_ping_pong` 命中，强制路由到 `human`
累计 token 超 60k	走 `finalize` 返回降级总结，不抛 `GraphRecursionError`
同一 `thread_id` 二次追问	自动带上历史，且 `handoffs` 继续累加（注意是否需要按轮次清空）
子 Agent 内部抛异常	整个超步不提交；恢复后只重跑该 Agent，其他 Agent 结果复用
`tokens` 写成 `state["tokens"] + n`	超步重跑会虚增 → 必须返回增量交给 reducer

🚀 变种拓展题

升级成 Hierarchical：把 researcher / coder 归到一个 dev_sup 子 Supervisor 下，验证两层调度的延迟变化。
改造成 Swarm：去掉 Supervisor，只靠 handoff 工具流转，并在 State 里记录"当前负责人"。
实现消息隔离：Agent 之间只共享"结论类"消息，各自的中间推理留在子图 State，对比上下文 token 的下降幅度。
给每个 Agent 配独立的工具白名单与预算，并在 Supervisor 处做越权拦截。
用 stream(..., subgraphs=True) 做一个终端可视化面板，实时显示"当前 Agent + 当前节点 + 已花 token"。

⚠️ 当日踩坑实录

❌ 坑	✅ 解决方案
节点里 `return state`（返回完整 State）	只返回要改的字段。返回完整 State 会让带 `add_messages` 的字段重复追加一遍，上下文直接翻倍
并行两个节点写同一字段，偶发 `InvalidUpdateError`	给该字段加 `Annotated[..., reducer]`；本地测不出来是因为 mock 太快、执行顺序恰好固定
`operator.add` 累加列表，结果里全是重复项	改成按业务 key 去重的自定义 reducer；`operator.add` 配非幂等节点必然出重复
`recursion_limit=25` 却只跑了 12 轮工具就报错	它算的是超步，一轮 ReAct = `agent` + `tools` = 2 个超步
报 `GraphRecursionError` 直接 500 给用户	`try/except` 后用 `get_state` 拿当前状态拼降级答案；同时排查是不是条件边写反导致死循环
条件边 `KeyError`	路由函数返回的键不在映射表里；且必须把 `END` 作为一个显式分支，否则死循环
忘了传 `thread_id`	配了 checkpointer 就必须传，否则报错；没配则每次都是新会话、多轮记忆失效
`interrupt` 恢复后业务流水插了两条	恢复是整个节点从头重跑。所有副作用必须放在 `interrupt` 之后，之前只能有纯计算和读操作
同一节点里写了两个 `interrupt`，恢复后返回值错位	拆成两个节点。一个节点一个中断点
裁剪上下文按条数硬切，下一轮 400	删 `AIMessage(tool_calls=[...])` 必须连带删对应 `ToolMessage`，要以完整 tool 回合为单位切
子图加了新字段，父图始终拿不到且不报错	父图 State 没声明该 key 会被静默丢弃；用 `stream(..., subgraphs=True)` 对照子图 updates 排查
`Send` 的目标节点里写 `state["messages"]` 报 `KeyError`	`Send` 传的是自定义载荷，不是全局 State
30 个 `Send` 并发打爆限流	`invoke(..., {"max_concurrency": 8})` 或自己分批
`stream_mode="messages"` 拿不到 token	① 节点里绕过 LangChain 直接用厂商 SDK；② Python < 3.11 异步下 `config` 没透传给 `ainvoke`
内部节点（意图分类、摘要）的 token 被打到用户屏幕上	按 metadata 里的 `langgraph_node` 过滤，只放行主 Agent
handoff 后下一轮请求 400	handoff 工具必须返回 `ToolMessage`，不能留悬空 tool_call
子图 handoff 跳不到兄弟 Agent	要带 `graph=Command.PARENT`
两个 Agent 来回踢皮球五六轮	State 记交接链，同一对交接超过 N 次强制转人工或收尾
`user_id` 放在 State 里被安全评审打回	身份与权限必须走 `config["configurable"]` —— State 是模型可写的，存在注入越权风险
Checkpoint 几十 MB，Postgres 磁盘告警	它是全量快照不是 diff：State 只存 `doc_id` 不存原文、配裁剪节点、定期清理旧 thread
`cache_policy` 永远不命中	节点输入里有 `datetime.now()` 或随机 id，缓存键每次都不同
节点内自己写 `for i in range(3)` 重试，恢复后计数归零	用 `RetryPolicy`，由框架管理并落 Checkpoint，还能配 `retry_on` 白名单和 jitter
配了重试但 `AuthError` 也重试了 3 次	必须配 `retry_on` 只重试限流/超时这类可恢复错误
并行开了 8 个分支但没变快	超步耗时 = 最慢节点的耗时，分支耗时要均衡
升级后 `interrupt` 不生效也不报错	锁 `langgraph` 与 `langchain-core` 的版本对，二者的 `Command`/`interrupt`/消息类型强耦合
`from langgraph.checkpoint import SqliteSaver` 导入失败	已拆包：`pip install langgraph-checkpoint-sqlite`，并用 `with SqliteSaver.from_conn_string(...) as cp:` 管理连接
集成测试全绿但线上恢复相关 bug 频出	测试必须带 `checkpointer=MemorySaver()` 和显式 `thread_id`，并主动制造一次中断再恢复
map-reduce 汇总节点跑了 3 遍、结果少一半	各分支步数不等时汇总节点会被多次触发，给它加 `defer=True`；老版本就在节点里判断"齐了没有"
节点名拼错，图上多了个孤岛却不报错	连完边先 `print(graph.get_graph().draw_ascii())` 自检；子图用 `get_graph(xray=1)` 展开
返回 `Command` 的节点在图上看不到连线	必须写 `-> Command[Literal["a","b"]]` 返回注解，框架靠它推断边
`astream_events` 拿不到事件或字段对不上	必须传 `version="v2"`；v1 结构不同且已不推荐
同时上了 `stream_mode` 和 `astream_events`	事件重复、前端去重逻辑爆炸。UI 用 `stream_mode`，排障用 `astream_events`，别混
用户连点三次发送，messages 里消息错位报 400	同 `thread_id` 必须串行化（Redis 锁）或用 `multitask_strategy` 语义，锁粒度是 thread 不是用户
上线第一个请求报"表不存在"	`PostgresSaver` 首次必须 `.setup()`，把它做成启动时的迁移步骤
异步服务里用了同步 `PostgresSaver`，延迟飙高	换 `AsyncPostgresSaver`；同步 Saver 会阻塞事件循环
QPS 一上来延迟就翻倍，但 LLM 没变慢	每个超步都要读写 checkpoint，检查数据库连接池大小是不是太小

📌 当日复盘任务

概念理解（必须能脱口而出）

[ ] 用一句话说清 LCEL 和 LangGraph 的分工（数据流 vs 控制流，二者正交）
[ ] 画出 State / Node / Edge / Conditional Edge / Graph 五件套的关系
[ ] 解释"超步"：一批并行 + 屏障同步 + 一次合并提交 + 一个 Checkpoint
[ ] 说清 reducer 的签名、默认行为，以及"什么字段必须加 reducer"的判断口诀
[ ] 区分 Checkpointer（短期、单 thread）与 Store（长期、跨 thread、可语义检索）
[ ] 说清条件边与 Command 的选择标准（决策要不要落状态）

动手实现（当天必须跑通）

[ ] 从零搭一个 agent ⇄ tools 的 ReAct 图，补齐四道闸门和 force_answer 降级节点
[ ] 故意制造并行写冲突，复现 InvalidUpdateError，再用自定义 reducer 修掉
[ ] 用 interrupt() 做一个写操作审批，并在 interrupt 之前打一行日志，亲眼确认恢复时重跑
[ ] 用 Send 实现一次动态 map-reduce，确认 N 个任务只消耗 1 次 recursion_limit
[ ] 用 get_state_history 列出所有 checkpoint，带 checkpoint_id 做一次 Time Travel 重跑
[ ] 同时订阅 ["updates", "messages"]，把两种流分发成不同的 SSE 事件
[ ] 拆一个子图，用 stream(..., subgraphs=True) 看到子图内部每一步
[ ] 连完边先 print(graph.get_graph(xray=1).draw_ascii()) 自检拓扑，故意拼错一个节点名看看孤岛长什么样
[ ] 用 astream_events(..., version="v2") 抓一次 on_tool_start / on_retriever_end，对比它和 stream_mode 的事件量差异
[ ] 造一个分支步数不等的 map-reduce，复现汇总节点被触发多次，再用 defer=True 修掉

工程化加固（挑两项落地）

[ ] 给关键节点配 RetryPolicy（带 retry_on 白名单 + jitter）和 cache_policy
[ ] 把一个有副作用的操作改造成"单独节点 + 幂等键 + 放在 interrupt 之后"
[ ] 加一个上下文裁剪节点，验证 RemoveMessage 删除 tool 回合时要成对删
[ ] 把身份信息从 State 迁到 config["configurable"]，并在节点里改用 config 读取
[ ] 写一个 pytest，用 FakeMessagesListChatModel 覆盖"完整路径 + 中断恢复 + 递归超限"三条
[ ] 把 MemorySaver 换成 PostgresSaver，跑一次 .setup()，确认连接池参数与异步 Saver 选型
[ ] 实现 double-texting 防护：按 thread_id 加锁 + 可取消的 run，手动连发两条验证状态不错乱

面试输出（写成可背的稿子）

[ ] 准备"为什么从裸循环换到 LangGraph"的项目故事，要有具体的两个痛点和删掉多少代码
[ ] 准备"节点为什么必须幂等"的事故复盘：重复推送 / 重复流水，根因与规范
[ ] 准备多 Agent 选型话术：四种拓扑对比 + 你为什么选 Supervisor + 踢皮球怎么防
[ ] 准备一段 90 秒的 State/reducer 讲解，能当场手写一个幂等 reducer
[ ] 整理 Day27 ↔ Day28 的对照总表，面试时用它来回答"LangGraph 到底解决了什么"
[ ] 准备部署话术：三种上线形态的取舍、为什么选这一种、多副本为什么能无状态扩容
[ ] 准备 double-texting 话术：四种策略各自的产品语义，以及你们默认选哪种、为什么写操作要例外

Day28 LangGraph 复杂智能体编排 ​

📖 基础知识讲解【基础原理 + 高级深挖知识点】 ​

基础原理 ​

1. 为什么需要 LangGraph：Day27 那个 while 循环到底撑不住什么 ​

2. 核心概念五件套：State、Node、Edge、Conditional Edge、Graph ​

3. State 的本质：带 reducer 的共享内存，不是"传参" ​

4. 节点函数的四条契约 ​

5. 边的三种形态：固定边、条件边、并行边 ​

6. 执行模型：超步（super-step）与 Pregel BSP ​

7. 最小可运行 Graph：把 Day27 的 ReAct 图化 ​

8. 预制件 create_react_agent 与手写图的边界 ​

9. 终止条件：四道闸门，少一道都可能烧钱 ​

10. Checkpointer 与 thread_id：短期记忆到底存了什么 ​

11. 中断与恢复：Human-in-the-Loop 的两种写法 ​

12. 流式输出：五种 stream_mode 分别给谁看 ​

13. 递归限制：recursion_limit 算的是超步不是节点 ​

14. 子图 Subgraph：把大图拆小的唯一正确姿势 ​

15. Send API：真正的动态 map-reduce ​

16. 和 Day27 裸循环的能力对照总表 ​

✨ 高级深挖知识点 ​

1. reducer 的三类冲突，以及自定义 reducer 的正确写法 ​

2. State Schema 分层：input / output / private 三件套 ​

3. add_messages 的隐藏能力：按 id 覆盖与 RemoveMessage ​

4. 上下文工程：三种裁剪策略挂在图的什么位置 ​

5. Store：跨 thread 的长期记忆 ​

6. 并行节点的事务性：超步失败到底回滚了什么 ​

7. InvalidUpdateError 的四种触发场景速查 ​

8. Command：一个返回值同时干"改状态"和"跳转"两件事 ​

9. 多智能体四种拓扑，面试必须能画出来 ​

10. Handoff 的标准实现：让工具返回 Command ​

11. Checkpoint 膨胀：被忽略的存储炸弹 ​

12. 节点级可靠性：retry_policy 与 cache_policy ​

13. 可观测性：一个 trace 要能回答"钱花在哪、卡在哪" ​

14. 测试策略：三层金字塔 ​

15. 安全：把图当权限边界用 ​

16. Functional API：不想画图时的另一条路 ​

17. 性能：延迟都花在哪，以及怎么压 ​

18. 版本坑：0.2 → 0.6 的关键变更 ​

19. 部署形态：从 compile() 到可对外提供服务 ​

20. Double-texting：用户没等回答完就又发了一条 ​

21. 一句话串起全链路（给面试官的收束） ​

❓ 高频面试题【基础简答 + 连环追问链 + 精简口述答案 + 项目关联话术】 ​

题 1：LangGraph 是什么？它和 LangChain 是什么关系，为什么不用 LCEL 就够了 ​

题 2：State 和 reducer 讲一下，为什么并行节点写同一个字段会报错 ​

题 3：什么是超步（super-step）？Checkpoint 以超步为粒度会带来什么后果 ​

题 4：Checkpointer 和 thread_id 是什么？短期记忆和长期记忆怎么分 ​

题 5：Human-in-the-Loop 怎么实现？interrupt 恢复的时候有什么坑 ​

题 6：条件边和 Command 都能决定下一步，什么时候用哪个 ​

题 7：运行时才知道要并行跑 N 个任务，怎么做？fan-in 怎么汇总 ​

题 8：Agent 无限循环怎么防？recursion_limit 设大点不就行了 ​

题 9：图太大怎么拆子图？父子 State 怎么对接 ​

题 10：多智能体有哪几种拓扑？你们怎么选的，handoff 怎么实现 ​

题 11：LangGraph 的流式输出怎么对接前端？几种 stream_mode 分别干什么 ​

题 12：为什么说 LangGraph 的节点必须幂等？有副作用的操作该怎么处理 ​

题 13：用户上一条还在生成就又发了一条，你们怎么处理（double-texting） ​

题 14：LangGraph 写完的图怎么上线？多副本部署要注意什么 ​

✍️ 手撕代码【核心实现 + 边界 Case 自测清单 + 变种拓展题】 ​

题 1：用 LangGraph 手写带四道闸门的 ReAct 图 ​

题 2：手写一个支持并行写入的幂等 reducer 集合 ​

题 3：实现写操作的人工审批闸门（含幂等与重跑安全） ​

题 4：用 Send 实现长文档的 map-reduce 摘要 ​

题 5：手写 Supervisor 多智能体（含 handoff 与踢皮球防护） ​

⚠️ 当日踩坑实录 ​

📌 当日复盘任务 ​

Day28 LangGraph 复杂智能体编排

📖 基础知识讲解【基础原理 + 高级深挖知识点】

基础原理

1. 为什么需要 LangGraph：Day27 那个 while 循环到底撑不住什么

2. 核心概念五件套：State、Node、Edge、Conditional Edge、Graph

3. State 的本质：带 reducer 的共享内存，不是"传参"

4. 节点函数的四条契约

5. 边的三种形态：固定边、条件边、并行边

6. 执行模型：超步（super-step）与 Pregel BSP

7. 最小可运行 Graph：把 Day27 的 ReAct 图化

8. 预制件 `create_react_agent` 与手写图的边界

9. 终止条件：四道闸门，少一道都可能烧钱

10. Checkpointer 与 thread_id：短期记忆到底存了什么

11. 中断与恢复：Human-in-the-Loop 的两种写法

12. 流式输出：五种 stream_mode 分别给谁看

13. 递归限制：`recursion_limit` 算的是超步不是节点

14. 子图 Subgraph：把大图拆小的唯一正确姿势

15. Send API：真正的动态 map-reduce

16. 和 Day27 裸循环的能力对照总表

✨ 高级深挖知识点

1. reducer 的三类冲突，以及自定义 reducer 的正确写法

2. State Schema 分层：input / output / private 三件套

3. `add_messages` 的隐藏能力：按 id 覆盖与 RemoveMessage

4. 上下文工程：三种裁剪策略挂在图的什么位置

5. Store：跨 thread 的长期记忆

6. 并行节点的事务性：超步失败到底回滚了什么

7. `InvalidUpdateError` 的四种触发场景速查

8. `Command`：一个返回值同时干"改状态"和"跳转"两件事

9. 多智能体四种拓扑，面试必须能画出来

10. Handoff 的标准实现：让工具返回 `Command`

11. Checkpoint 膨胀：被忽略的存储炸弹

12. 节点级可靠性：retry_policy 与 cache_policy

13. 可观测性：一个 trace 要能回答"钱花在哪、卡在哪"

14. 测试策略：三层金字塔

15. 安全：把图当权限边界用

16. Functional API：不想画图时的另一条路

17. 性能：延迟都花在哪，以及怎么压

18. 版本坑：0.2 → 0.6 的关键变更

19. 部署形态：从 `compile()` 到可对外提供服务

20. Double-texting：用户没等回答完就又发了一条

21. 一句话串起全链路（给面试官的收束）

❓ 高频面试题【基础简答 + 连环追问链 + 精简口述答案 + 项目关联话术】

题 1：LangGraph 是什么？它和 LangChain 是什么关系，为什么不用 LCEL 就够了

题 2：State 和 reducer 讲一下，为什么并行节点写同一个字段会报错

题 3：什么是超步（super-step）？Checkpoint 以超步为粒度会带来什么后果

题 4：Checkpointer 和 thread_id 是什么？短期记忆和长期记忆怎么分

题 5：Human-in-the-Loop 怎么实现？`interrupt` 恢复的时候有什么坑

题 6：条件边和 `Command` 都能决定下一步，什么时候用哪个

题 7：运行时才知道要并行跑 N 个任务，怎么做？fan-in 怎么汇总

题 8：Agent 无限循环怎么防？`recursion_limit` 设大点不就行了

题 9：图太大怎么拆子图？父子 State 怎么对接

题 10：多智能体有哪几种拓扑？你们怎么选的，handoff 怎么实现

题 11：LangGraph 的流式输出怎么对接前端？几种 stream_mode 分别干什么

题 12：为什么说 LangGraph 的节点必须幂等？有副作用的操作该怎么处理

题 13：用户上一条还在生成就又发了一条，你们怎么处理（double-texting）

题 14：LangGraph 写完的图怎么上线？多副本部署要注意什么

✍️ 手撕代码【核心实现 + 边界 Case 自测清单 + 变种拓展题】

题 1：用 LangGraph 手写带四道闸门的 ReAct 图

题 2：手写一个支持并行写入的幂等 reducer 集合

题 3：实现写操作的人工审批闸门（含幂等与重跑安全）

题 4：用 `Send` 实现长文档的 map-reduce 摘要

题 5：手写 Supervisor 多智能体（含 handoff 与踢皮球防护）

⚠️ 当日踩坑实录

📌 当日复盘任务