前面几天反复出现三句话："工具挂了自动兜底"、"业务不直接 import anthropic，走 get_llm"、"测试走 FakeLLM 不联网"。今天就把这三件底层能力讲透——它们是整个可信底座能成立的技术前提。

回忆 Day 08 闸⑤"基础设施降级"。它的实现之一就是 @safe_tool_result 装饰器（tools/safe_result.py:38）。它把任何工具函数包成"永远返回统一结构、永不抛异常"：

@safe_tool_result(fallback={"hint": "tool failed"})
async def get_metrics(service: str) -> dict:
    return await call_prometheus(service)   # 就算这里炸了……

# 调用方永远拿到统一结构，不用写 try/except：
result = await get_metrics(service="A")
if result["ok"]:
    data = result["data"]        # 成功：真实数据
else:
    data = result["data"]        # 失败：fallback 兜底值
    err  = result["error"]       # "ConnectionError: ..."（带异常类名）

返回的 ToolResult 是一个 TypedDict（safe_result.py:30），三个字段固定死：

# tools/safe_result.py:30
class ToolResult(TypedDict):
    ok: bool           # 成功 True / 失败 False
    data: Any          # 成功=真实返回；失败=fallback 兜底值
    error: str | None  # 成功=None；失败="异常类名: 消息"

装饰器的核心就一个 wrapper，把整个调用包进 try/except（safe_result.py:54）：

# tools/safe_result.py:54
@functools.wraps(fn)
async def wrapper(*args, **kwargs) -> ToolResult:
    try:
        if timeout:
            import asyncio
            data = await asyncio.wait_for(fn(*args, **kwargs), timeout=timeout)  # 超时也算失败
        else:
            data = await fn(*args, **kwargs)
        return {"ok": True, "data": data, "error": None}
    except Exception as e:
        logger.warning("Tool %s failed: %s", fn.__name__, e, exc_info=True)
        # 带上异常类型名，方便测试 / 监控按类别聚合
        err_msg = f"{type(e).__name__}: {e}" if str(e) else type(e).__name__
        return {"ok": False, "data": fallback, "error": err_msg}

wrapper.__safe_tool_result__ = True   # 打个标记，别处可探测这个函数已被包过

这是全框架最关键的一条约定：业务代码绝不直接 import anthropic 或 ChatAnthropic(...)，一律通过 get_llm() 拿模型（llm.py）。为什么？看这张对比：

机制其实很轻——就是一个模块级的可替换工厂 _factory（llm.py:29）。get_llm 每次去问"当前工厂是谁"：

# llm.py:29
_factory: Callable[[ModelTier], Any] | None = None   # None 时用默认工厂

# llm.py:311  测试时注入假工厂
def set_llm_factory(fn):
    global _factory
    _factory = fn

# llm.py:359  get_llm 取模型:有自定义工厂用它,否则用默认工厂
factory = _factory or _default_factory

默认工厂 _default_factory（llm.py:224）才是真正 import anthropic 的唯一地方——全框架就这一处碰真厂商 SDK：

# llm.py:224  唯一创建真 ChatAnthropic 的地方
def _default_factory(tier="sonnet", *, cache_system=False, cache_messages=False):
    from langchain_anthropic import ChatAnthropic       # 只有这里 import 厂商 SDK
    callbacks = _default_callbacks()                     # 自动挂计费回调(Day 11 L07)
    if tier == "haiku":
        model = os.environ.get("AI_TRUST_HAIKU_MODEL", "claude-haiku-4-5")
        llm = ChatAnthropic(model=model, temperature=0.2, max_tokens=..., callbacks=callbacks)
    else:  # sonnet 默认
        model = os.environ.get("AI_TRUST_SONNET_MODEL", "claude-sonnet-4-5")
        llm = ChatAnthropic(model=model, temperature=0.2, max_tokens=8192, callbacks=callbacks)
    return llm

get_llm(tier) 的 tier 参数就是 Day 05 讲的"省钱三档"（ModelTier, llm.py:27）：

get_llm("auto") 是 v0.6 的成本感知路由：预算充足用 Sonnet，预算剩余低于阈值（默认 10%）时自动降到 Haiku。决策在 cost/router.py 的 CostAwareRouter.resolve_tier()（router.py:108），按优先级短路：

# cost/router.py:127  resolve_tier 按优先级"短路"返回 (tier, reason)
if requested != "auto":                 return (requested, REASON_EXPLICIT)     # 明确指定就用它
if tracker is None:                     return (self.default_tier, REASON_NO_TRACKER)
tenant_id = getattr(tracker, "tenant_id", None) or ""
if not tenant_id:                       return (self.default_tier, REASON_NO_TENANT)
try:
    pct = get_pct_sync(tenant_id)       # 读预算剩余百分比(60s sync cache · 不 await)
except Exception:
    return (self.default_tier, REASON_FALLBACK_PCT_ERROR)   # 查失败也不降级(fail-safe)
if pct < self.budget_pct_degrade_threshold:  return (self.degrade_tier, REASON_BUDGET_DEGRADATION)
return (self.default_tier, REASON_DEFAULT)

把 resolve_tier 当调试器单步走一遍——4 个不同请求进来，看它命中哪条短路、最终用哪档模型：

每个 reason 都不是随手写的字符串，而是从一个 frozenset 白名单里取（router.py:63）：

# cost/router.py:63  reason 只能是这几个之一
REASON_ALLOWLIST: frozenset[str] = frozenset({
    REASON_EXPLICIT, REASON_NO_TRACKER, REASON_NO_TENANT,
    REASON_BUDGET_DEGRADATION, REASON_DEFAULT,
    REASON_FALLBACK_NO_BUDGET_MODULE, REASON_FALLBACK_PCT_ERROR,
    REASON_QUOTA_DEGRADATION, REASON_RATELIMIT_BURST,  # 预留未来信号
})

Critic、专家这些节点的 system prompt 很长且每次调用都一样。Anthropic 的 prompt caching 能让"重复的前缀"只算一次钱（后续命中约 9 折优惠）。get_llm("haiku", cache_system=True) 就是开启它。

实现上有个兼容性巧思，就藏在 get_llm 尾部这段 try/except 里（llm.py:359）：

# llm.py:359
factory = _factory or _default_factory
try:
    # 先假设工厂支持 cache 形参(默认工厂支持)
    return factory(actual_tier, cache_system=cache_system, cache_messages=cache_messages)
except TypeError:
    # 自定义工厂签名不支持 cache 形参(常见:测试 set_llm_factory(lambda tier: FakeLLM()))
    llm = factory(actual_tier)                       # 退一步:只按 tier 取 llm
    if cache_system or cache_messages:
        return _wrap_with_cache_control(llm, cache_system=cache_system, cache_messages=cache_messages)
    return llm

那"外面包一层"包的是什么？看 _wrap_with_cache_control（llm.py:265）——它还要分两种情况：

# llm.py:280
if isinstance(llm, Runnable):
    return RunnableLambda(_transform) | llm     # 真 ChatAnthropic:走 LangChain 管道
return _NonRunnableCacheWrap(llm, _transform)   # FakeLLM 等非 Runnable:套透明委托壳

_NonRunnableCacheWrap（llm.py:287）是个"透明壳"：只拦 invoke/ainvoke 塞进 transform，其它属性全用 __getattr__ 转发给里面的真对象（llm.py:306）——所以 FakeLLM 不用被迫去继承 LangChain 的 Runnable 协议。

代码 testing/fakes.py。三个东西支撑了"整套测试几秒跑完、0 外网、可复现"：

• FakeLLM(responses=[...])（fakes.py:31）：按顺序返回预置回复。适合单个节点的单测。记录调用次数便于断言。
• SystemAwareFake(rules, default)（fakes.py:62）：按 system prompt 里的关键词分发不同回复。适合一个进程里多个节点共用一个工厂——比如 sre-rca 里 4 个专家 + 综合 + Critic 各要不同假答案，靠各自 system prompt 的关键词区分。
• install_fakes()（fakes.py:119）：一键装好"假 LLM + 假 embedder + 内存向量库"，返回三件套，测试 fixture 一行搞定。

先看最简单的 FakeLLM.ainvoke（fakes.py:47）——它就是个"按次数发牌"的机器：

# testing/fakes.py:47
async def ainvoke(self, messages, **kwargs) -> FakeResponse:
    self.calls.append(messages)                      # 记下这次被谁调了(方便断言)
    if self.call_count >= len(self.responses):
        self.call_count += 1
        return FakeResponse(content="(no more fake responses)")   # 预置回复用完的哨兵
    resp = self.responses[self.call_count]
    self.call_count += 1
    return resp

而 SystemAwareFake 更聪明，靠 system prompt 里的关键词路由到对应回复（fakes.py:91）：

# testing/fakes.py:91
async def ainvoke(self, messages, **kwargs) -> FakeResponse:
    system_text = self._extract_system(messages)     # 抠出 system prompt 文本(fakes.py:106)
    for key, resp in self.rules.items():
        if key in system_text:                        # 第一个命中关键词的规则胜出
            self.calls.append((key, messages))
            return resp
    self.calls.append(("default", messages))
    return self.default                               # 都不命中 → 兜底回复

# 典型的 conftest.py fixture
@pytest.fixture(autouse=True)
def fake_llm():
    set_llm_factory(lambda tier: SystemAwareFake({
        "trace 专家": '{"trace_finding": {...}}',
        "综合":       '{"conclusion": {...}}',
    }))
    yield
    reset_llm_factory()      # 测试结束还原

最后一块底座，也是最巧的一块。问题：一次调用里，成本要按节点归因、业务效果指标要统计——难道每个节点都手动传一个"记账本"进去？太侵入了。

框架用 Python 的 ContextVar（上下文变量，task-local）解决。在一次 invoke 的入口 bind_tracker(tracker) 绑一个记账器，之后任何深处的代码都能通过 current_tracker() 拿到它，不用层层传参：

# api/cost.py：一次 invoke 的自动记账
with bind_tracker(tracker):          # ContextVar 绑定
    await graph.ainvoke(...)         # 图里每次真 LLM 调用，
                                     # UsageMetadataCallbackHandler 在 on_llm_end
                                     # 自动 current_tracker().record(tokens...)
metrics = tracker.finalize(status)   # 推 3 个 Prometheus 指标

三个东西共用这套 ContextVar 范式：成本追踪（api/cost.py）、业务效果指标 record_effect()（effect.py，在"真完成一次业务动作"处记一行）、预算闸门（Day 08）。而且 LangGraph 用 asyncio task 并行跑节点时会自动 copy 上下文，所以并发累加也正确。

顺带提一句可观测性一键接入：observability.py 的 setup_observability(app, service_name) 给业务 FastAPI 一行挂上 /metrics（Prometheus）+ 可选 OTel。没装 OTel 就返回一个空操作的 tracer，业务无脑写 with tracer.start_as_current_span(...) 也不会报错。

✋ 动手

# 1. 读 safe_tool_result（就 72 行，很好读）
cat packages/ai-trust-toolkit/src/ai_trust_toolkit/tools/safe_result.py

# 2. 读 get_llm / set_llm_factory
sed -n '311,370p' packages/ai-trust-toolkit/src/ai_trust_toolkit/llm.py

# 3. 读 FakeLLM 三件套
cat packages/ai-trust-toolkit/src/ai_trust_toolkit/testing/fakes.py

# 4. 读成本感知路由的优先级短路
sed -n '108,150p' packages/ai-trust-toolkit/src/ai_trust_toolkit/cost/router.py

# 5. 跑相关单测
uv run pytest packages/ai-trust-toolkit/tests/test_llm_factory.py \
              packages/ai-trust-toolkit/tests/test_safe_result.py -q