一个稍复杂的 Agent 不是"调一次大模型就完事"，它有很多步骤：先分诊、再并行取证、再综合、再质检、质检不过还要打回重做……这些步骤有先后、有分支、有循环。

LangGraph 的思路：把这些步骤画成一张"流程图"（graph）来跑。 图里每个方框是一个"节点（Node）"= 一个步骤；方框之间的箭头是"边（Edge）"= 执行顺序。所有节点共享一块叫 State 的黑板来传数据。

下面这张图会自己动起来——高亮会依次点亮每个节点，模拟一次真实执行（这正是 Day 14 要精读的 sre-rca Agent 的简化版；L08 你会看到它的真源码）：

LangGraph 你只要先掌握这 6 个概念，就够读懂本框架 90% 的图代码了：

仓库里 docs/learn-langgraph/ 提供了 4 个能直接跑的最小 demo（默认用 MockLLM，不配 key 也能跑）。下面 L03–L07 就一个个读它们的真源码。

State 定义成一个 TypedDict。别看抽象的例子——直接看 demo1 里真实的 State 定义（它刻意和生产 sre_rca/state.py 同款 schema，方便你 L08 对照）：

class RcaState(TypedDict):
    # 输入
    trace_id: str
    user_question: str | None
    # 4 SP 各占独立字段（零冲突）—— 本 demo 只填 trace_finding
    trace_finding: dict | None
    metric_finding: dict | None
    deploy_finding: dict | None
    log_finding: dict | None
    # 累加 Reducer（解决并发，本 demo 还用不上，放着 demo 03 用）
    retry_count: Annotated[int, add]
    # 最终输出
    final_summary: str | None

• trace_id / user_question：输入字段，进图时由调用方塞进去。
• 4 个 *_finding 字段：注释点破了本框架最重要的设计——"4 SP 各占独立字段（零冲突）"。4 个专家并行时，各写各的字段，天生不打架（L06 展开）。
• retry_count: Annotated[int, add]：这是唯一一个带 Reducer 的字段。Annotated[int, add] 意思是"类型是 int，但多方写入时用 operator.add（相加）合并"。demo1 用不上，故意先摆在这，demo3 才启用。

节点函数的约定极简——读整个 state，只返回"我改了哪些字段"的 dict，剩下的 LangGraph 帮你合并。看 demo1 第一个节点：

async def fetch_trace_node(state: RcaState) -> dict:
    """节点 A：根据 trace_id 拉 trace 数据（这里 mock 一份）。"""
    trace_id = state["trace_id"]
    # 真实项目中这里会 call SkyWalking GraphQL
    raw = {"trace_id": trace_id, "spans": 42, "max_p99_ms": 8200}
    return {"trace_finding": {"raw": raw}}   # ← 只返回自己负责的字段

有了 State 和节点函数，就用 StateGraph 把它们摆好、连起来。看 demo1 的建图函数（真源码，不是伪代码）：

def build_graph():
    builder = StateGraph(RcaState)                 # 1. 以 State 类型建图

    builder.add_node("fetch_trace", fetch_trace_node)   # 2. 注册 3 个节点
    builder.add_node("analyze", analyze_node)
    builder.add_node("summarize", summarize_node)

    builder.add_edge(START, "fetch_trace")         # 3. 连边：START→A→B→C→END
    builder.add_edge("fetch_trace", "analyze")
    builder.add_edge("analyze", "summarize")
    builder.add_edge("summarize", END)

    return builder.compile()                       # 4. 编译成可执行图

• StateGraph(RcaState)：以 L03 那个 State 类型开一张空图。图知道"公告板长什么样"。
• add_node("名字", 函数)：把函数登记成一个工位，起个名字。
• add_edge(A, B)：连普通边"A 跑完一定走 B"。START / END 是内置的入口/出口特殊节点。
• compile()：把"蓝图"编译成能跑的图对象。之后 await graph.ainvoke(初始state) 就从 START 一路跑到 END，返回最终 state（demo1 用的是 ainvoke，异步版，本框架全异步）。

把它当调试器单步走一遍——传入 {"trace_id":"abc123"}，看公告板每步变成什么：

普通边是"死路"（A 之后一定走 B）。条件边让图能根据 state 的值"走不同的路"。这正是 Critic 质检的核心模式——PASS / FAIL 重试 / 触顶结束三态。看 demo2 的路由函数真源码：

def critic_router(state: RcaState) -> str:
    """与 app/graph/builder.py 的 critic_router 同款三态。"""
    if state.get("critique_passed"):
        return "writeback"                    # ① 过 → 去沉淀
    if state.get("retry_count", 0) >= 2:
        return "end"                          # ② 没过但触顶 → 结束（防死循环）
    return "synthesizer"                      # ③ 没过且没触顶 → 回去重做

路由函数只做一件事：看 state，返回一个字符串（下一步节点的名字）。再把它挂到图上：

builder.add_conditional_edges(
    "critic",                    # 从 critic 节点出发
    critic_router,               # 用这个函数决定去哪
    {
        "writeback": "writeback",
        "synthesizer": "synthesizer",   # ← 指回前面的节点 = 形成循环
        "end": END,
    },
)

• add_conditional_edges(源节点, 路由函数, 映射表)：从 critic 出来后，先跑 critic_router 拿到一个字符串，再用第三个参数的映射表把字符串翻译成真正的目标节点。
• "synthesizer": "synthesizer" 这条把边指回更早的节点 → 就形成了循环（retry loop）：质检不过 → 回去重做 → 再质检。

当多个节点从同一上游出发、又都指向同一下游时，LangGraph 自动并行执行它们。看 demo3 怎么用一个 for 循环把 4 个专家扇出、再扇入到 rag：

    # 一对多扇出：START 同时触发 4 个 SP
    for sp in ("trace_sp", "metric_sp", "deploy_sp", "log_sp"):
        builder.add_edge(START, sp)
        # 多对一扇入：4 个 SP 都指向 rag → LangGraph 自动等 4 个都完成才执行 rag
        builder.add_edge(sp, "rag")

    builder.add_edge("rag", "synthesizer")

问题：4 个专家同时往 state 写。如果写同一个字段就会打架。解决办法就是 Reducer（归约器）——看 demo3 真实的 State 定义：

class RcaState(TypedDict):
    trace_id: str
    # 4 SP 字段隔离（默认覆盖 Reducer，因每个字段只有 1 个写者）
    trace_finding: dict | None
    metric_finding: dict | None
    deploy_finding: dict | None
    log_finding: dict | None
    # ⑤ 累加 Reducer：解决并发计数
    retry_count: Annotated[int, add]

• 4 个 *_finding 字段：每个字段只有一个写者（trace_sp 只写 trace_finding），天然不冲突，用默认的"覆盖"语义就行——不需要 Reducer。
• retry_count: Annotated[int, add]：这是唯一需要 Reducer 的字段。add 就是 operator.add。多个节点各返回 {"retry_count": 1}，Reducer 把它们相加而不是互相覆盖。

Checkpointer（存档器）把每跑完一个节点的 state 存下来。看 demo4 怎么挂它、怎么用 thread_id 隔离会话：

from langgraph.checkpoint.memory import InMemorySaver

def build_graph(checkpointer):
    builder = StateGraph(RcaState)
    ...
    # 测试用 InMemorySaver；生产用 RedisSaver.from_conn_string(REDIS_URL)
    return builder.compile(checkpointer=checkpointer)   # ← 挂存档器

# 跑：
checkpointer = InMemorySaver()
graph = build_graph(checkpointer)
config1 = {"configurable": {"thread_id": "session-001"}}   # ← thread_id 决定"哪次会话"
final1 = await graph.ainvoke(initial, config=config1)

• compile(checkpointer=...)：编译时挂上存档器，图每跑完一个节点就把 state 落一次盘。
• config={"configurable":{"thread_id": ...}}：thread_id 就像"会话 ID"。同一 thread_id 的多次调用共享同一份存档；换个 thread_id 就是全新一份。
• demo4 用两个 thread（session-001 / session-002）证明"两份 state 互不干扰"，还用 graph.get_state(config1)（04:105）取回会话 1 的存档。

带来两个能力：断点续跑（跑一半崩了，用同一 thread_id 能从断点接着跑）+ 多会话隔离（一个 agent 服务同时处理很多用户，各自 state 不串）。

4 个 demo 的抽象，在旗舰 Agent sre-rca 里长成什么样？先看它真实的 State——你会发现字段比 demo 多，但结构一模一样：

class RcaState(TypedDict, total=False):
    # ── 输入 ──
    trace_id: str
    service_name: str | None
    user_question: str
    # ── Triage 输出 ──
    alert_type: str
    need_full_dag: bool
    # ── Recall 输出（历史 case）──
    similar_cases: list[dict[str, Any]]
    # ── 4 Specialist 输出（字段独立，并行）──
    trace_finding: dict[str, Any] | None
    metric_finding: dict[str, Any] | None
    deploy_finding: dict[str, Any] | None
    log_finding: dict[str, Any] | None
    # ── Synthesizer 输出 ──
    conclusion: dict[str, Any] | None
    # ── Critic 输出 ──
    critique_passed: bool
    critique_feedback: str
    critique_layer: int   # 1 = 代码层失败 / 2 = LLM 层失败 / 0 = 通过
    # ── Reducer：累加（防并发竞态）──
    retry_count: Annotated[int, add]
    # ── Writeback / 输出 ──
    summary_for_human: str

• total=False：所有字段可选。因为每个节点只填自己产出的那几个字段，不用一次性填满。
• 4 个 *_finding：和 demo3 完全一致的"字段隔离"设计，4 个专家并行写零冲突。
• critique_layer: int（state.py:45）：注释直接剧透了 Day 07 的 Critic 双层——1=代码层失败 / 2=LLM 层失败 / 0=通过。State 字段本身就是能力的说明书。
• retry_count: Annotated[int, add]（state.py:48）：全 State 唯一带 Reducer 的字段，和 demo 一致——用累加防竞态。

再看它真实的建图 builder.py——这才是"读一个 Agent 的正确入口"。整张图的骨架就这几十行：

    builder = StateGraph(RcaState)
    # 节点注册（10 个）
    builder.add_node("triage", triage_node)
    builder.add_node("recall", build_recall_node(episodic))
    builder.add_node("trace_sp", trace_sp_node)
    builder.add_node("metric_sp", metric_sp_node)
    builder.add_node("deploy_sp", deploy_sp_node)
    builder.add_node("log_sp", log_sp_node)
    builder.add_node("rag", build_rag_node(semantic))
    builder.add_node("synthesizer", build_synthesizer_node())
    builder.add_node("critic", build_critic_for_rca())
    builder.add_node("writeback", build_writeback_node(episodic))

    # 边：线性入口
    builder.add_edge(START, "triage")
    builder.add_edge("triage", "recall")
    # 4 SP 并行扇出 + 扇入到 RAG（和 demo3 一模一样的写法）
    for sp in ("trace_sp", "metric_sp", "deploy_sp", "log_sp"):
        builder.add_edge("recall", sp)
        builder.add_edge(sp, "rag")
    builder.add_edge("rag", "synthesizer")
    builder.add_edge("synthesizer", "critic")

    # Critic 三态路由（用 toolkit 封好的，不手写）
    critic_router = build_critic_router(
        pass_dest="writeback", retry_dest="synthesizer",
        end_dest=END, max_retries=max_critic_retries,
    )
    builder.add_conditional_edges("critic", critic_router,
        {"writeback": "writeback", "synthesizer": "synthesizer", END: END})
    builder.add_edge("writeback", END)
    return builder.compile(checkpointer=checkpointer)

• 10 个 add_node：triage 分诊 → recall 召回历史 → 4 个 SP 专家 → rag 汇合 → synthesizer 综合 → critic 质检 → writeback 沉淀。名字即职责。
• for sp in (...) 那两行：和你在 demo3（03:96-100）读到的扇出扇入一字不差。demo 不是玩具，就是生产写法的最小复刻。
• build_critic_router(...)（builder.py:100-105）：注意——生产没有手写 demo2 那个 critic_router，而是从 toolkit import build_critic_router，传 pass_dest / retry_dest / end_dest / max_retries 四个参数生成。
• compile(checkpointer=checkpointer)：挂上 L07 讲的存档器收尾。

✋ 动手：跑 4 个官方 demo + 读真 Agent（不配 key 也能跑）

# 这 4 个 demo 默认走 MockLLM，无需 API key
uv run python docs/learn-langgraph/01_state_node_edge.py
uv run python docs/learn-langgraph/02_conditional_edge.py
uv run python docs/learn-langgraph/03_reducer_parallel.py
uv run python docs/learn-langgraph/04_checkpointer.py

# 对照生产：先看数据结构，再看建图
sed -n '15,53p'  apps/sre-rca-agent/sre_rca/state.py
sed -n '73,118p' apps/sre-rca-agent/sre_rca/builder.py

# 想切真 Claude？（需要 key）
USE_REAL_LLM=True uv run python docs/learn-langgraph/01_state_node_edge.py

跑的时候留意：demo2 里条件边怎么触发不同分支；demo3 里 4 个并行节点的输出怎么被 reducer 合并；demo4 里换 thread_id 会不会看到之前的 state。然后回头看 sre-rca 的 builder.py——你会发现它就是这几个 demo 的拼装。