本框架的一大好处是所有 Agent 目录结构统一——看懂一个就看懂全部。以 apps/sre-rca-agent/sre_rca/ 为例（下面每个文件今天都会读到真代码）：

回忆 Day 01 的铁律：L3 业务 Agent 只写 State / Specialists / Prompts / DAG(图) 这四类东西，护栏能力全从 toolkit import。上面这个结构正是这条铁律的体现。真实文件清单印证了这点——nodes/ 下是 triage/recall/rag/synthesizer/critic/writeback，specialists/ 下是 trace_sp/metric_sp/deploy_sp/log_sp + 一个 _factory.py（后面 L08 会揭穿这个"化石"）。

state.py:15 定义 class RcaState(TypedDict, total=False)。回忆 Day 03：这是整张图的"黑板"，每个节点读它、返回部分字典合并回去。真代码按"数据流阶段"分了组：

class RcaState(TypedDict, total=False):        # state.py:15
    # ── 输入 ──
    trace_id: str
    service_name: str | None
    user_question: str
    # ── Triage 输出 ──
    alert_type: str
    need_full_dag: bool
    # ── 4 Specialist 输出（字段独立，并行写零冲突）── state.py:29
    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 / Critic 输出 ──
    conclusion: dict[str, Any] | None
    critique_passed: bool
    critique_layer: int          # 1=代码层失败 / 2=LLM层失败 / 0=通过
    # ── Reducer：累加（防并发竞态）── state.py:48
    retry_count: Annotated[int, add]

核心函数 build_rca_graph()（builder.py:42）。它就是 Day 03/04 学的那套 StateGraph → add_node → add_edge → compile，只是节点更多。整张图长这样：

关键代码（来自真实 builder.py，行号已核对）：

builder = StateGraph(RcaState)                    # builder.py:73
builder.add_node("triage", triage_node)           # builder.py:76
builder.add_node("recall", build_recall_node(episodic))
builder.add_node("trace_sp", trace_sp_node)       # 4 个专家节点
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("critic", build_critic_for_rca())
# ── 线性入口 ── builder.py:88
builder.add_edge(START, "triage")
builder.add_edge("triage", "recall")
# ── 4 SP 并行扇出 + 扇入到 RAG ── builder.py:92
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")
# ── Critic 三态条件路由：直接用 toolkit 封好的！── builder.py:100
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) # builder.py:118

逐段翻译：（1）StateGraph(RcaState) 建一张以刚才那块黑板为状态的图；（2）10 个节点一个个 add_node；（3）并行的秘密就在那个 for 循环——4 个专家都"从 recall 出发、都汇入 rag"，LangGraph 看到这种"同源扇出、同汇扇入"就自动让它们并行，业务作者一个线程都不用管；（4）Critic 后面接的不是手写 if/else，而是从 toolkit import 的 build_critic_router——通过/打回重做/触顶三态路由全封好了（Day 07/08 细讲）；（5）compile(checkpointer=...) 收尾，checkpointer 是"工作记忆"（Day 09）。

每个节点是 async def node(state) -> dict，或返回这种函数的"工厂"（把依赖闭包进去，比如 build_recall_node(episodic)）。逐个看角色：

4 个专家文件每个都只有约 30 行，因为它们都是"填一份配置 SpecialistConfig → 交给工厂 make_specialist_node"。先把这份配置的字段看清（specialists/factory.py:37）：

@dataclass
class SpecialistConfig:                                       # factory.py:37
    state_field: str          # 结果写进 state 的哪个字段（如 'metric_finding'）
    system_prompt: str        # system prompt 内容（业务从 prompts/*.md 读）
    fetch_fn: Callable[[dict], Awaitable[dict]]   # 拉证据的 async 函数
    fallback_finding: dict = field(default_factory=lambda: {  # 抓取/执行失败时的降级结果
        "health": "unknown", "confidence": "低",
        "evidence_ids": [], "hint": "证据获取失败"})
    model_tier: ModelTier = "sonnet"
    name: str = "specialist"
    cache_system: bool = False        # 打开 prompt 缓存（Day 07/11）
    # ─── 3 个可选自定义点（不填就用默认实现）───
    user_prompt_builder: UserPromptBuilder | None = None
    response_parser: ResponseParser | None = None
    finding_post_process: FindingPostProcessor | None = None

配置填好后交给工厂 make_specialist_node(cfg)（factory.py:84）。它返回一个 async def specialist(state) -> dict 节点，内部就是一条"取数→喂 LLM→解析→写字段"的流水线，最后再包一层兜底。逐块读它的核心 inner_node（factory.py:93）：

async def inner_node(state: dict) -> dict:
    # ① 拉数据 · factory.py:95
    tool_result = await cfg.fetch_fn(state)
    if not tool_result.get("ok"):                     # 拉失败 → 早返回 fallback
        finding = dict(cfg.fallback_finding)
        finding["error"] = tool_result.get("error", "tool failed")
        return {cfg.state_field: finding}
    raw = tool_result["data"]
    # ② 调 LLM · factory.py:104
    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)}])
    content = getattr(response, "content", str(response))
    # ③ 解析 + 后处理 · factory.py:114
    finding = response_parser(content)
    finding.setdefault("evidence_ids", [])            # 保证关键字段一定存在
    if cfg.finding_post_process:
        finding = cfg.finding_post_process(finding, raw)
    return {cfg.state_field: finding}

逐步翻译：① 先调 fetch_fn 拉证据；拉失败（ok=False）就地返回 fallback，压根不浪费钱去调 LLM（这是第一处边界处理）。② 用配置里的档位 get_llm(cfg.model_tier) 调 LLM，把 system prompt + 用户内容喂进去。③ 解析 LLM 输出成 finding，setdefault("evidence_ids", []) 保证这个关键字段一定存在（哪怕 LLM 没给），然后如果配了后处理钩子就再加工一下，最后写回 state_field 那一格。

再看第 ④ 步——工厂返回前把 inner_node 包进兜底闸门（factory.py:125）：

    return wrap_with_fallback(                         # factory.py:125
        inner_node,
        FallbackConfig(
            state_field=cfg.state_field,
            fallback_finding=dict(cfg.fallback_finding, fallback=True)))

上一讲第 ④ 步那个 wrap_with_fallback 是 toolkit failsafe 的第一道闸门（Day 08 主角之一），代码很短（node_fallback.py:33）：

def wrap_with_fallback(node_fn, config: FallbackConfig):     # node_fallback.py:33
    async def wrapped(state: dict) -> dict:
        try:
            return await node_fn(state)                       # 正常就原样返回
        except Exception as e:                                # node_fallback.py:55
            logger.log(config.log_level, "Node failed for state_field=%s: %s",
                       config.state_field, e, exc_info=True)
            return {config.state_field: dict(config.fallback_finding, error=str(e))}
    wrapped.__name__ = getattr(node_fn, "__name__", "wrapped_node")
    wrapped.__wrapped__ = node_fn
    return wrapped

大白话：把任意 async 节点函数包成"永远不抛异常"的版本。正常时原样返回；一旦内部抛任何异常，它 catch 住、打日志、返回一个带 error 的 fallback_finding。这样这个专家节点再怎么炸，也只是往黑板写一格"降级结果"，绝不会让整张图崩掉。

前面读了配置和工厂，现在看真实的一个专家文件——specialists/metric_sp.py，去掉 import 和读 prompt，实质就一个配置（这是完整文件，没有省略）：

from ai_trust_toolkit.specialists import SpecialistConfig, make_specialist_node
from ..tools import get_metrics

_PROMPT = (Path(__file__).resolve().parents[2] / "prompts"
           / "metric-specialist-v1.md").read_text(encoding="utf-8")

metric_sp_node = make_specialist_node(                    # metric_sp.py:13
    SpecialistConfig(
        name="metric_sp",
        state_field="metric_finding",                     # 写黑板哪一格
        system_prompt=_PROMPT,                            # 从 prompts/ 读作业指导书
        fetch_fn=lambda state: get_metrics(               # 怎么拉证据
            service=state.get("service_name") or "unknown", window_sec=60),
        fallback_finding={"health": "unknown", "confidence": "低",
                          "evidence_ids": [], "hint": "指标获取失败"},
        model_tier="sonnet",
    ))

就这些。没有 try/except、没有 LLM 调用循环、没有 JSON 解析、没有并发处理——因为这些全在 L06/L07 那个工厂里。业务作者只回答 5 个问题：叫什么名、写哪格、用哪份 prompt、怎么拉数据、失败给什么降级值。这就是"专家为什么只有 30 行"的真相。其它三个专家（trace/deploy/log）结构一模一样，只是换 state_field / fetch_fn / prompt / tier。

目录里的"活化石"：specialists/_factory.py

你在 specialists/ 里会看到一个 _factory.py，打开一看头顶写着大大的 DEPRECATED（_factory.py:1）：

"""DEPRECATED —— 已上移到 ai_trust_toolkit.specialists（toolkit v0.2.0）。
本文件保留仅为兼容旧 import 路径，未来 v0.3 删除。新代码请用：
    from ai_trust_toolkit.specialists import SpecialistConfig, make_specialist_node
"""
from ai_trust_toolkit.specialists import (  # noqa: F401  re-export
    SpecialistConfig, default_response_parser,
    default_user_prompt_builder, make_specialist_node)

tools/ — 只读数据源

tools/observability.py 定义 get_trace_by_id / get_metrics / get_deployment_history / search_logs 四个工具。两个要点：每个都用 @safe_tool_result(fallback=...) 装饰——工具永不抛异常，失败返回约定的 {ok,data,error} 兜底（正好衔接 L06 工厂 ① 步读的那个 ok 字段，Day 11 细讲）；还有 USE_REAL 开关：默认走 mock，生产接 SkyWalking/Prometheus。

prompts/*.md — 版本化的工程文档

真实目录里有 7 个 *-v1.md：triage-v1 / trace/metric/deploy/log-specialist-v1 / synthesizer-v1 / critic-v1。它们不是干巴巴的 prompt，而是带教学说明的工程文档（正文 + 用户模板 + 设计说明 + Token 估算 + 变更记录）。

server.py / main.py — 两个入口

还有个宝藏端点 /agents/rca/graph（server.py:142）：把每个节点的 role/model/purpose 以 JSON 返回——相当于这个 Agent 的"自我说明书"，Day 14 会拿它当权威参照。

pyproject.toml — 依赖走 BOM，一行锁死

dependencies = [
    "ai-trust-toolkit-bom[core,supervisor,monitor]==0.6.0",
]

✋ 动手：把结构落到真实文件

# 1. 看旗舰 agent 的完整结构
find apps/sre-rca-agent/sre_rca -type f -name "*.py" | sort
ls apps/sre-rca-agent/prompts/          # 数一下真的是 7 个 *-v1.md

# 2. 精读黑板的两种 Reducer
sed -n '15,60p' apps/sre-rca-agent/sre_rca/state.py

# 3. 精读图的装配（今天的核心）
sed -n '73,118p' apps/sre-rca-agent/sre_rca/builder.py

# 4. 读专家工厂的数据结构 + 四步流水线 + 双层兜底
sed -n '37,131p' packages/ai-trust-toolkit/src/ai_trust_toolkit/specialists/factory.py
sed -n '33,67p' packages/ai-trust-toolkit/src/ai_trust_toolkit/failsafe/node_fallback.py

# 5. 看一个专家有多"薄" + 那块 DEPRECATED 化石
cat apps/sre-rca-agent/sre_rca/specialists/metric_sp.py
cat apps/sre-rca-agent/sre_rca/specialists/_factory.py