apps/sre-rca-agent 是平台的旗舰范本——一个 SRE 故障根因分析（Root Cause Analysis）Agent。你给它一个告警/现象（"service-A 大量超时"），它自动从 trace、指标、发布记录、日志四个视角取证，综合出 Top-3 最可能的根因，并给出带证据、带教学性的分析。

它是"满配"agent：10 个节点、4 专家并行、双层 Critic 三态重试、四层记忆全用上。读懂它，其它 agent 都是它的简化/变体。

看 RcaState 全貌（sre_rca/state.py:15-52），我按节点分组标注：

class RcaState(TypedDict, total=False):        # 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
    # ── RAG / Synthesizer 输出 ──
    sop_context: list[dict[str, Any]]
    conclusion: dict[str, Any] | None
    synth_skipped: bool
    synth_skip_reason: str | None
    # ── Critic 输出 ──
    critique_passed: bool
    critique_feedback: str
    critique_layer: int              # 1=代码层失败 / 2=LLM层失败 / 0=通过
    # ── Reducer：累加（防并发竞态）──
    retry_count: Annotated[int, add]     # ← 唯一一个非覆盖语义的字段
    # ── Writeback / 输出 ──
    writeback_skipped: bool
    summary_for_human: str

数据布局懂了，看 build_rca_graph（sre_rca/builder.py:42）怎么把节点连起来。这是整台机器的"接线图"：

builder = StateGraph(RcaState)                 # builder.py:73 · 以 RcaState 为共享内存
# 节点注册（工厂类的要先调工厂拿到闭包节点）
builder.add_node("triage", triage_node)
builder.add_node("recall", build_recall_node(episodic))       # 闭包注入 EpisodicStore
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
for sp in ("trace_sp", "metric_sp", "deploy_sp", "log_sp"):
    builder.add_edge("recall", sp)             # recall → 每个专家（扇出）
    builder.add_edge(sp, "rag")                # 每个专家 → rag（扇入）
builder.add_edge("rag", "synthesizer")
builder.add_edge("synthesizer", "critic")

synthesizer → critic 之后不是普通边，而是三态条件边（builder.py:100-116）：

critic_router = build_critic_router(           # builder.py:100 · 来自 toolkit（D07/D08）
    pass_dest="writeback",                     # 通过 → 去沉淀
    retry_dest="synthesizer",                  # 不过没触顶 → 回炉重做
    end_dest=END,                              # 不过且触顶 → 诚实收尾
    max_retries=max_critic_retries,            # 默认 2
)
builder.add_conditional_edges(
    "critic", critic_router,
    {"writeback": "writeback", "synthesizer": "synthesizer", END: END},
)
builder.add_edge("writeback", END)

入口节点。用关键词字典把 user_question 归到 5 类 alert_type（sre_rca/nodes/triage.py:8-35）：

ALERT_KEYWORDS: dict[str, list[str]] = {       # triage.py:8
    "DB_SLOW":     ["DB", "数据库", "连接池", "慢查询", "锁等待", "select", "update"],
    "JVM":         ["GC", "OOM", "JVM", "heap", "young gen", "old gen"],
    "ERROR_SPIKE": ["error", "异常", "exception", "5xx", "失败"],
    "GENERIC_SLOW":["慢", "卡", "延迟", "P99", "P95", "timeout"],
    "HEALTH_QUERY":["健康", "状态", "看一下", "check", "巡检"],
}

def _classify(text: str) -> str:               # triage.py:17
    text_lower = text.lower()
    # 优先级：DB_SLOW > JVM > ERROR_SPIKE > GENERIC_SLOW > HEALTH_QUERY
    for alert_type, keywords in ALERT_KEYWORDS.items():   # ← 靠 dict 插入顺序定优先级！
        for kw in keywords:
            if kw.lower() in text_lower:
                return alert_type
    return "GENERIC_SLOW"                       # 默认值

async def triage_node(state: dict) -> dict:    # triage.py:27
    user_question = state.get("user_question") or ""
    alert_type = _classify(user_question)
    need_full_dag = alert_type != "HEALTH_QUERY"   # 纯健康查询走简版，不必全 DAG
    return {"alert_type": alert_type, "need_full_dag": need_full_dag}

recall 要访问 EpisodicStore（D09 的记忆库），但 LangGraph 的节点签名是固定的 async def node(state)——没法多传参数。解法是工厂闭包（sre_rca/nodes/recall.py:10-35）：

def build_recall_node(episodic: EpisodicStore, top_k: int = 3):   # recall.py:10 · 工厂
    async def recall_node(state: dict) -> dict:                    # recall.py:16 · 真节点
        query = f"trace_id={state.get('trace_id')} service={state.get('service_name')} q={state.get('user_question')}"
        try:
            cases = await episodic.recall(query, top_k=top_k)      # 向量检索相似历史 case
        except Exception as e:
            return {"similar_cases": [], "_recall_error": str(e)}  # ← 挂了不崩，返空
        return {"similar_cases": [
            {"id": r.id, "text": r.text, "score": score, "metadata": r.metadata}
            for r, score in cases
        ]}
    return recall_node                                             # 返回闭包，交给 builder

核心取证环节。4 个专家从 recall 出发并行跑、汇入 rag。每个专家文件才 ~30 行，因为都用 toolkit 的 make_specialist_node。先看 metric 专家的配置（sre_rca/specialists/metric_sp.py:13-30）：

metric_sp_node = make_specialist_node(         # metric_sp.py:13
    SpecialistConfig(
        name="metric_sp",
        state_field="metric_finding",          # ← 写进 state 的哪个字段（并行不冲突！）
        system_prompt=_PROMPT,                  # 从 prompts/metric-specialist-v1.md 读
        fetch_fn=lambda state: get_metrics(     # 先抓数据（@safe_tool_result 包过，见 observability.py:38）
            service=state.get("service_name") or "unknown", window_sec=60),
        fallback_finding={                      # 抓失败时的降级结果
            "health": "unknown", "confidence": "低",
            "evidence_ids": [], "hint": "指标获取失败"},
        model_tier="sonnet",                    # 取证是难活，用 Sonnet
    )
)

差异化配置就这些。真正的执行逻辑在 toolkit 工厂 make_specialist_node（ai_trust_toolkit/specialists/factory.py:84）里——4 个专家共用这套：

async def inner_node(state: dict) -> dict:     # factory.py:93
    tool_result = await cfg.fetch_fn(state)    # ① 拉数据
    if not tool_result.get("ok"):              # ← 边界：抓数据失败
        finding = dict(cfg.fallback_finding)
        finding["error"] = tool_result.get("error", "tool failed")
        return {cfg.state_field: finding}      # 直接返回降级 finding，不调 LLM（省钱）
    raw = tool_result["data"]
    llm = get_llm(cfg.model_tier, cache_system=cfg.cache_system)   # ② 调模型
    response = await llm.ainvoke([{"role":"system","content":cfg.system_prompt},
                                  {"role":"user","content":user_prompt_builder(state, raw)}])
    finding = response_parser(getattr(response,"content",str(response)))  # ③ 解析
    finding.setdefault("evidence_ids", [])     # 保证有 evidence_ids 字段（给 Critic 用）
    return {cfg.state_field: finding}

inner_node.__name__ = cfg.name
return wrap_with_fallback(inner_node,          # factory.py:125 · ④ 再包一层节点级兜底
    FallbackConfig(state_field=cfg.state_field,
                   fallback_finding=dict(cfg.fallback_finding, fallback=True)))  # ← 打 fallback=True

rag（sre_rca/nodes/rag.py:19-40）用 4 专家的线索拼查询、检索运维手册 SOP：

async def rag_node(state: dict) -> dict:       # rag.py:19
    query_parts = []
    for f_name in ("trace_finding", "metric_finding", "deploy_finding", "log_finding"):
        f = state.get(f_name)
        if f and isinstance(f, dict) and not f.get("fallback"):   # ← 跳过降级的假 finding
            hint = f.get("hint") or f.get("hypothesis") or ""
            if hint:
                query_parts.append(str(hint))
    if not query_parts:
        return {"sop_context": []}             # ← 边界：一个有效线索都没有，直接返空
    query = " | ".join(query_parts)
    try:
        sops = await semantic.lookup(query, record_type="sop", top_k=top_k)
    except Exception as e:
        return {"sop_context": [], "_rag_error": str(e)}    # 检索挂了也不崩
    return {"sop_context": [{"id": r.id, "text": r.text, "score": score, "metadata": r.metadata}
                            for r, score in sops]}

synthesizer（sre_rca/nodes/synthesizer.py）是全图最"贵"的一步——调 Sonnet 综合出 Top-3。它被两层保护包着：

async def _inner_synthesizer(state: dict) -> dict:     # synthesizer.py:15 · 真正调 LLM
    llm = get_llm("sonnet")
    user_msg = (f"trace_id: {state.get('trace_id')}\n... "
                f"trace_finding:  {json.dumps(state.get('trace_finding'), ensure_ascii=False)}\n"
                f"... --- Critic 上一轮反馈（如有）---\n{state.get('critique_feedback', '(无)')}\n"
                f"请按 system prompt 的 schema 输出 conclusion JSON。")
    response = await llm.ainvoke([{"role":"system","content":_PROMPT},
                                  {"role":"user","content":user_msg}])
    return {"conclusion": _parse_conclusion(getattr(response,"content",str(response)))}

def build_synthesizer_node():                          # synthesizer.py:49
    return build_synth_safeguard(                      # ← D08 闸②：包一层"证据不足开关"
        _inner_synthesizer,
        min_real_specialists=1,                        # 至少 1 个 SP 有真实数据才调 LLM
    )

_parse_conclusion（synthesizer.py:60）是三级容错解析——和 D07 Critic 的解析器同款套路：先裸 JSON → 再 ```json 代码块 → 最后兜底一个 info_sufficient:False 的 dict。

critic（sre_rca/nodes/critic.py:12-30）几乎不写业务——直接复用 D07 的 build_critic_node，只填配置：

def build_critic_for_rca():                    # critic.py:12
    prompt = (Path(__file__).resolve().parents[2] / "prompts" / "critic-v1.md").read_text("utf-8")
    return build_critic_node(CriticConfig(     # ← 来自 toolkit（D07 双层 Critic）
        system_prompt=prompt,
        llm_factory=lambda: get_llm("haiku", cache_system=True),  # ← Haiku 省80% + prompt缓存再省
        conclusion_field="conclusion",
        soft_fail=True,                        # LLM 挂了当"不通过"而非崩图
    ))

writeback（sre_rca/nodes/writeback.py:20-59）先守门再沉淀：

async def writeback_node(state: dict) -> dict:         # writeback.py:20
    if not should_writeback(state):            # ← D08 闸④：仅"质检过 && 信息充分"才写
        return {"writeback_skipped": True, "summary_for_human": _format_summary(state)}
    conclusion = state.get("conclusion") or {}
    case_id = state.get("trace_id") or "case-unknown"
    text = json.dumps({...hypotheses..., "summary": conclusion.get("summary_for_human")}, ensure_ascii=False)
    try:
        await episodic.writeback(case_id=case_id, text=text, metadata={...})   # 沉淀进记忆库
    except Exception as e:
        return {"writeback_skipped": True, "_writeback_error": str(e),
                "summary_for_human": _format_summary(state)}     # 写失败也不崩
    return {"writeback_skipped": False, "summary_for_human": _format_summary(state)}

别忽略 prompts/*.md——它们是本仓最好的 prompt 工程教材。以 prompts/critic-v1.md 为例（前面 critic.py 就是读它），它不只是一段 prompt，而是包含：

• 明确的模型指定：文件开头写"用 Claude Haiku 4.5 跑（spec 模型分层策略）"——prompt 和模型档位绑定记录在案。
• 6 项检查（critic-v1.md "6 项检查"小节）：证据 ID 真实性 / 置信度校准 / 语言风格铁律 / 矛盾处理 / 动作安全性 / 教学性，每项都规定了判定标准。
• 代码层预检说明：明确写"JSON 合法/必填字段/类型这些硬错误走代码不走 LLM"（呼应 D07 的 L1），Critic 只做"模型才能判断的事"。
• 三态决策：PASS / PASS_WITH_NOTES / FAIL，各自的判定条件与反馈格式。
• few-shot 示例：给出 FAIL / PASS 的完整样例，稳住 LLM 输出格式。

synthesizer-v1.md 更狠：多步思考 + 语言铁律（禁绝对措辞、必引 ≥2 专家证据、教学性、失败开关）+ 一个完整的分析范例。

✋ 动手：逐文件读一遍（今天的核心作业）

# 数据结构 + 图骨架
cat apps/sre-rca-agent/sre_rca/state.py
sed -n '42,118p' apps/sre-rca-agent/sre_rca/builder.py
# 6 个 nodes
cat apps/sre-rca-agent/sre_rca/nodes/triage.py
cat apps/sre-rca-agent/sre_rca/nodes/recall.py
cat apps/sre-rca-agent/sre_rca/nodes/rag.py
cat apps/sre-rca-agent/sre_rca/nodes/synthesizer.py
cat apps/sre-rca-agent/sre_rca/nodes/critic.py
cat apps/sre-rca-agent/sre_rca/nodes/writeback.py
# 一个专家（配置）+ toolkit 工厂（内部）
cat apps/sre-rca-agent/sre_rca/specialists/metric_sp.py
sed -n '84,131p' packages/ai-trust-toolkit/src/ai_trust_toolkit/specialists/factory.py
# 数据源工具（@safe_tool_result）+ prompt 工程文档
sed -n '1,55p' apps/sre-rca-agent/sre_rca/tools/observability.py
sed -n '1,60p' apps/sre-rca-agent/prompts/critic-v1.md
# 跑测试 + 跑评测
uv run pytest apps/sre-rca-agent -q
uv run python apps/sre-rca-agent/scripts/run_eval.py