Day 06 / 共 20 天 · 第 2 周 可信底座

ai-trust-toolkit 总览 & 渐进抽象

第 1 周你摸清了地图和一户人家的骨架;从今天起进入框架的灵魂——ai-trust-toolkit。今天不只鸟瞰它的 10 个模块,还会打开两处真源码:__init__.py(它为什么几乎什么都不导出)和 effect.py(一个"零侵入采集"模块怎么用 30 行做到"不 bind 就静默"),再顺着 CHANGELOG 读三次真实的抽象决策史——全仓最值钱的方法论就藏在这些故事里。

📍 你在 20 天里的位置(第 2 周:可信底座)
D06 toolkit 总览 D07 Critic D08 失败闸门 D09 Memory D10 评测
💡 用一个类比先兜住今天(延续「盖楼/物业」世界观) toolkit 就是整栋楼的公共设备机房 + 承重结构:消防、水泵、配电、电梯全在这——每一户(业务 Agent)不用自己装,接根管子就用。今天讲四件事:① 机房里有哪 10 台设备;② 打开两台设备的机箱看真接线(__init__.pyeffect.py);③ 机房内部的管线也分层(叶子设备不反向依赖别人);④ 这机房遵循"渐进抽象":第一户要什么先在户内凑合搭,等第二户也要一模一样的,才把它挪进公共机房。这句"第二次重复才抽"是全仓最值钱的一句话,而 CHANGELOG 把每次"挪进机房"的理由都记了账。
L01

toolkit 的定位:横切能力的公共底座

packages/ai-trust-toolkit(版本 0.6.0,见 __init__.py:31__version__)是本仓自研的核心库。它不是"某个 agent 的代码",而是所有 agent 都能用的横切能力——防幻觉、闸门、评测、记忆、成本治理。

回忆 Day 05 的"薄业务 + 厚 toolkit":一个业务 agent(如 sre-rca)真正自己写的代码可能就几百行,剩下的"可信/可观测/成本/记忆"全是从这个库 import 来的。所以吃透了 toolkit,你就吃透了整个平台。

权威参照文档:packages/ai-trust-toolkit/ARCHITECTURE.md(分层图 + 逐模块"为什么")和 docs/CHANGELOG.md(v0.1→v0.6 的抽象演进时间线)。本周讲解都以它俩和源码为准。
L02

10 个模块速记(本周地图)

🛡️

critic/

双层防幻觉:L1 代码规则 + L2 LLM 语义审查。Day 07

🚦

failsafe/

失败闸门:节点兜底/证据不足开关/Critic 路由/写回守门/预算硬熔断。Day 08

🧠

memory/

四层记忆:工作/情景/语义/向量底座。Day 09

📊

eval/

多维评测:5 维对话 + 6 维 coding + 在线评测。Day 10

🔧

tools/

safe_tool_result 装饰器 + MCP 客户端。Day 11

🤖

llm.py

get_llm / set_llm_factory:厂商无关的 LLM 注入点 + 成本自动降级 + prompt 缓存。Day 11

🧪

testing/

FakeLLM / SystemAwareFake / install_fakes:0 外网测试基础设施。Day 11

💰

cost/ + api/budget

成本感知路由 + per-tenant 月度预算硬上限。Day 08/11

👷

specialists/ + reporter/

专家节点工厂 + 末端渲染工厂:消除 95% 重复代码。Day 05 见过

🌐

api/

v1 API 地基(9 文件、~2190 行、占 toolkit 41%):envelope/auth/router/cases/cost/quota/ratelimit。Day 12

另外还有几个"横切基础件":config.py(Day 02 讲过的配置加载)、observability.py(可观测,Day 11)、handoff.py(Day 04 讲过)、effect.py(业务效果指标,下一讲就拆它)。

L03

__init__.py 走读:一个"几乎什么都不导出"的门面

打开 ai_trust_toolkit/__init__.py,第一眼会觉得奇怪——一个这么大的库,顶层只导出了 5 个符号__init__.py:34-49):

__version__ = "0.6.0"                                       # __init__.py:31
# handoff 协议给 agent 端 import
from ai_trust_toolkit.handoff import HandoffSignal, new_chain_id      # :34
# 领域效果采集给 agent 端 import
from ai_trust_toolkit.effect import (                                 # :37
    bind_effect, current_effect, record_effect)
__all__ = [
    "HandoffSignal", "new_chain_id",
    "bind_effect", "current_effect", "record_effect",                 # :43-48
]

逐行读:__version__ 是全库版本;然后只 re-export 了 handoff(Day 04 学过的交接信号)和 effect(下一讲)两组东西。critic / memory / api / failsafe……顶层一个都没导出——它们全都要从子包按需 import。

__init__.py 开头那段 docstring(:1-27)反倒成了这个库最有用的东西——一张"API 地图",把常用 import 路径全列了:

from ai_trust_toolkit.tools import safe_tool_result
from ai_trust_toolkit.failsafe import wrap_with_fallback, build_critic_router
from ai_trust_toolkit.critic import collect_valid_evidence_ids, build_critic_node
from ai_trust_toolkit.specialists import make_specialist_node, SpecialistConfig
from ai_trust_toolkit.eval import EvalSample, FiveDimScorer, run_eval
from ai_trust_toolkit.memory import VectorStore, PgVectorStore, InMemoryVectorStore
from ai_trust_toolkit.api import build_v1_router, CostTracker, check_or_raise
from ai_trust_toolkit.llm import set_llm_factory, get_llm
from ai_trust_toolkit.testing import FakeLLM, SystemAwareFake, install_fakes
💡 设计取舍①:为什么顶层几乎不 re-export?如果顶层 from ai_trust_toolkit import * 把 critic/memory/api 全拉起来,会触发一堆副作用和慢加载——有的模块要连 DB、要 import 重量级的 langchain/anthropic。让你"用哪个从哪个子包 import",是为了按需加载、彼此不牵连:写个只用 FakeLLM 的单测,不会因为 import 顶层就顺带把整个 api 子模块和数据库连接拉起来。
💡 设计取舍②:那为什么偏偏 handoff 和 effect 破例,被 re-export 到顶层?因为这两个是纯 agent-facing、零重依赖的小协议——handoff.py 只有一个 frozen dataclass + 一个生成 ID 的函数,effect.py 只有一个 ContextVar + 三个函数(下一讲拆),都不 import anthropic/langchain/DB。它们被业务代码高频、零成本地 import,放顶层方便且无副作用。所以这个"破例"恰恰印证了规则:不是"能不能放顶层",而是"放顶层会不会带来重加载"。
⚠️ 小白常误以为"顶层 __init__ 不 re-export,是没写完/偷懒"。恰恰相反,这是刻意的解耦。阅读任何子包前,先看它自己的 __init__.py 里的 __all__ 和 docstring——那就是这个模块的"能力清单"。本周每一天都会先看对应子包的 __init__.py
🍼 记忆口诀(本周地图)toolkit 核心能力五连:「批(critic防幻觉)、兜(failsafe闸门)、记(memory记忆)、评(eval评测)、算(cost成本)」——对应 Day 07→08→09→10→11。外壳治理另算一档 api/(Day 12)。
L04

effect.py 解剖:一个"零侵入采集"模块只要 30 行

🤔 先想一个问题平台想统计"每个 agent 到底真办成了多少件事"(比如 bmc-agent 升级了几个依赖)。怎么在不污染业务代码的前提下,让业务在"真办成一件事"的地方打个点?而且这个打点函数还得保证:即使在不该采集的地方(比如单测、非 invoke 路径)被误调了,也不能炸。effect.py 是这个问题的一个漂亮答案,值得逐行学。

先看它的"数据结构"——一个 task-local 的累加器(effect.py:29):

# 当前 invoke 的 effect 累加器 · ContextVar task-local
_current_effect: contextvars.ContextVar[dict[str, float] | None] = \
    contextvars.ContextVar("ai_trust_current_effect", default=None)   # effect.py:29

它是一个 dict[str, float](形如 {"upgraded": 3.0}),装在 ContextVar 里。默认是 None——即"当前没在采集"。再看两个用法函数:

@contextmanager
def bind_effect() -> Iterator[dict[str, float]]:              # effect.py:40
    acc: dict[str, float] = {}
    token = _current_effect.set(acc)     # 进入 with:把累加器塞进 ContextVar
    try:
        yield acc
    finally:
        _current_effect.reset(token)     # 退出 with:还原(防污染下一次 invoke)

def record_effect(key: str, n: float = 1) -> None:           # effect.py:50
    acc = _current_effect.get()
    if acc is None:
        return                           # AC-1.4 · 没 bind → 静默 no-op,不抛!
    if not isinstance(key, str) or not key:
        log.debug("record_effect ignored · invalid key: %r", key)
        return
    try:
        acc[key] = acc.get(key, 0.0) + float(n)   # 累加
    except (TypeError, ValueError):
        log.debug("record_effect ignored · non-numeric n: %r", n)

逐块讲:bind_effect() 是个上下文管理器,平台在一次 invoke 的边界with bind_effect(): 包住——进入时把一个空 dict 塞进 ContextVar,退出时 reset 还原。record_effect("upgraded", 1) 是业务在"真办成一件事"的那一行调用——它从 ContextVar 拿累加器,累加计数。invoke 结束时平台自动把这个 dict 收进响应的 metrics。业务只需一行 record_effect(...),完全不用管"怎么上报、往哪存"。

effect:一次 invoke 的边界内累加,边界外静默 with bind_effect(): ← ContextVar 被 set 成 {} record_effect("upgraded") record_effect("upgraded") 累加器 = {"upgraded": 2.0} → invoke 结束进 metrics 边界外调用 get()=None 静默 no-op
图注:ContextVar 让"在 with 内累加、在 with 外静默"成为自动行为——业务打点无需判断自己在不在 invoke 里。
💡 设计取舍③:为什么用 ContextVar 而不是一个全局 dict?源码注释点破了(effect.py:27-31):"LangGraph 内部 asyncio.create_task 会 copy 上下文 · 并发节点累加正确"。如果用普通全局 dict,多个 invoke 并发时会互相串数据(A 的计数记到了 B 头上)。ContextVar 是"任务本地"的——每个异步任务看到自己的那份,并发天然隔离。这跟 api/cost.py 采集成本用的是同一套范式。
💡 设计取舍④(边界/易错点):为什么没 bind 时是"静默 no-op"而不是抛异常?record_effectif acc is None: returneffect.py:57,注释标着 AC-1.4)。业务代码里散布着 record_effect(...) 调用,但这些代码不一定总在 invoke 路径上跑——可能被单测直接调、被老 agent 在没 bind 的地方调。如果没 bind 就抛异常,等于"打点代码反过来炸了业务",本末倒置。所以它选择静默返回:采集是"锦上添花",绝不能因为它让业务挂掉。连 key 非法、n 不是数字,都是 log.debug 记一笔就跳过,绝不抛。这是"可观测性代码永远不能拖垮业务"的铁律。
L05

内部依赖铁律:叶子不依赖任何人

Day 01 讲了"层与层之间"的依赖铁律。toolkit 内部模块之间也有一套依赖规则,保证它自己不会变成一团乱麻:

eval/ — 站最高,可以依赖下面所有模块(评测要串起全链路)
failsafe/ · specialists/ — 中层,依赖叶子模块,不依赖 critic/eval
llm.py · tools/ · testing/ · memory/ · effect.py · handoff.py — 叶子模块,不 import 任何 toolkit 内部模块

刚读的 effect.py 就是标准的"叶子"——它只 import 标准库 contextvars/logging,一个 toolkit 内部模块都不碰。这就是为什么它能被顶层 __init__ 安全 re-export(L03 设计取舍②)。另外还有几条"红线":

  • sensitivity/ 敏感度分类模块绝不 import anthropic / langchain——它必须极轻、无外部依赖(纯规则)。
  • 许多函数刻意"没有依赖时静默降级":record_effect 没 bind 静默 no-op(刚读过)、budget_gate 没 tracker 静默放行、get_llm 遇到不支持缓存参数的 FakeLLM 自动退化——都是为了让测试和独立调用不崩
为什么这么较真? 因为 toolkit 是被 20+ 个 agent 共同依赖的地基。地基里模块乱依赖,会导致改一个模块牵连一片、循环 import、测试要拉起一堆无关组件。这套"分层 + 叶子 + 红线"的自律,是它能被这么多 agent 放心依赖的原因。
L06

渐进抽象:全仓最值得学的方法论

这是整个项目最核心的工程思想CHANGELOG 的"关键工程经验"章节(CHANGELOG.md §关键工程经验)开门见山:

🌱 "1 个 Agent 时不抽象,第 2 个 Agent 出现重复了才抽。"

CHANGELOG 里把它写成了三行阶梯:

1 个 Agent 就抽 toolkit  → 接口不准(无第二样本)
2 个 Agent 出现重复才抽  → 接口刚好准
3+ 个 Agent 持续验证      → 抽象趋于稳定

意思是:不要在只有一个用例时就急着设计"通用框架"。等到第二个、第三个 agent 出现真实重复,你才真正知道哪些该抽、抽成什么样。过早抽象往往抽错,反而成为负担。

🤔 如果让你自己来,你会怎么做?假设你在写第一个 agent sre-rca,需要 4 个专家节点。朴素做法:直接复制粘贴 4 段"取数→喂 LLM→解析→兜底"的相似代码,会不会立刻抽个 make_specialist 函数?——渐进抽象说先别:你只有这一个 agent,猜不准通用形态,抽早了大概率抽错。等 risk-reviewer 也要几乎一样的专家节点(真实重复出现),你才把它上移到 toolkit 成 make_specialist_node——也就是 Day 05 那块 _factory.py 化石记录的那次上移。
渐进抽象决策:第 2 次重复才「下沉」到公共机房 v0.1 · 只有 sre-rca 专家代码写在户内 1 个用例 允许"写死/重复" 第2户也要 v0.2 · risk-reviewer 出现同款重复 → 下沉 toolkit.make_specialist_node sre-rca import risk-reviewer import
图注:一份代码,第二次重复出现时才提取到公共机房,两户再各自 import——CHANGELOG 会记下这次"触发原因"。

🚫 过早抽象的坑

  • 只有 1 个用例,猜不准通用形态
  • 为"可能的未来"过度设计
  • 抽象错了,改起来比不抽还贵

✅ 渐进抽象的做法

  • 第 1 个 agent 允许"写死、重复"
  • 第 2 个出现同样代码 → 才提取到 toolkit
  • 每次抽象在 CHANGELOG 写清"触发原因"
L07

CHANGELOG 三次真实抽象走读(含"为什么之前不抽")

docs/CHANGELOG.md 完整记录了每一次抽象的"何时抽、为什么抽、为什么之前不抽"。我们读它记载的三次真实决策——注意它们不是照搬一条死规则,而是每次都在算"抽的收益 vs 抽错的代价"。

v0.2

抽出 Specialist 工厂(第 2 次重复,标准案例)

触发:risk-reviewer 的 _factory.py 跟 sre-rca 95% 重复(约 180 行) → 提取 make_specialist_node。收益写在账上:sre-rca _factory.py 113 行→17 行 re-export、risk-reviewer 70 行→17 行、重复代码 ~180 行 → 0 行

v0.3

抽出 business_critic_l1(只有 1 个用户也抽了——反例)

触发:第 3 个 agent(alert-triage)要纯规则的轻量 Critic。CHANGELOG 诚实记了一句"潜在过度抽象信号:仅 1 个用户",但仍决定抽,理由是"接口足够通用(5 个 helper 可组合)+ 抽错的代价小(154 行 + 12 测试)· 认 OK"。

v0.5

抽出 reporter 工厂(严守"第 3 次才抽")

触发:末端渲染节点在 risk-reviewer + alert-triage + arch-compliance 三次重复后才上移成 make_reporter_node——注释直接写"守渐进抽象铁律"。

💡 设计取舍⑤:v0.3 明明只有 1 个用户,为什么破例抽了?规则不是"第 2 次才抽"吗?关键在 CHANGELOG 那句权衡:"抽错的代价小"。渐进抽象的底层逻辑不是"数到 2",而是"抽错的期望损失"——第 2 次重复才抽,是因为那时你对通用形态最有把握、抽错概率低。但如果某个东西接口已经很清晰、代码量又小(154 行),哪怕只有 1 个用户,抽错了重写也不亏,那就可以提前抽。规则是为目标服务的,不是反过来。能看懂"什么时候可以破例",才算真懂了这套方法论。
⚠️ 反面信号 CHANGELOG 也照记不误:api/ 子模块占了 toolkit 41% 的 LOC,文档里写"可能将来要拆出 governance/ 顶层模块,但现在不是时候(还在生产验证期)"。这是"渐进"的另一面——不仅忍住"过早抽象",也忍住"过早拆分",等生产验证充分了再动。
怎么用这份 CHANGELOG? 当你想理解"这个抽象为什么长这样",去 CHANGELOG 找它被引入的那一版,读它的触发故事——比读代码本身更能理解设计意图。这也是读任何成熟开源项目的通用技巧。
L08

另外三条真实工程经验(都在 CHANGELOG 里)

除了"渐进抽象",CHANGELOG 的"关键工程经验"还记了三条,条条有真实数据支撑:

① 可选 > 强加

CHANGELOG 有一张大表统计 toolkit 各模块在 9 个 agent 里的复用情况,结论是"60% 模块可选——业务按需取用,不强加结构"。比如 make_specialist_node 在 RCA(N=4)/risk-reviewer(N=2)/alert-triage(N=3) 用了,但 doc-checker(双 LLM 节点)、bmc-agent(无 SP 概念)就没用,框架也不逼它们用。唯一近 100% 覆盖的是 v0.5 的 build_v1_router(api 地基,人人要)。

② 声明式 > 命令式

命令式(v0.2 之前)

自己写 async 函数节点 / 自己包 wrap_with_fallback / 自己写 try/except

声明式(v0.3 起)

列一个 Rule 集合 / toolkit 自动包兜底 / @safe_tool_result 装饰

真实收益:alert-triage 的 critic_l1.py 从 ~50 行命令式代码缩到 15 行声明式规则集

③ 删除老代码要谨慎——留空壳 re-export

Day 05 见过的那个 DEPRECATED 的 _factory.py,CHANGELOG 专门解释了为什么不删、改成 17 行 re-export

当时的开发环境(cowork 沙盒)下文件不能直接 unlink,于是把老 _factory.py 改成 re-export。"这反而是好事"——业务侧 import 路径可以渐进迁移(旧的 from ._factory import 还能用),而不是"大爆炸式"一次性全改。这跟"渐进抽象"是一体两面:抽象要渐进,迁移也要渐进。
💡 设计取舍⑥:为什么迁移要留空壳而不是一刀切换 import 路径?如果抽象上移的同时强制所有业务立刻改 import,就等于一次改动同时动"toolkit 新增"和"N 个 agent 修改"两件事,出问题很难定位是哪边坏的。留个 re-export 空壳,让"上移"和"业务改 import"解耦成两步、各自可回滚。这是大型重构的通用安全姿势——永远保留一条向后兼容的退路。
L09

两类使用者:agent-facing 与 api

toolkit 内部按"谁来用"分成两大类,理解这个划分能帮你快速定位代码——也解释了 L03 为什么 handoff/effect(agent-facing 小件)能上顶层,而 api(重外壳)留在子包:

agent-facing 八件套

LangGraph 图里的节点用的:critic / failsafe / eval / memory / specialists / tools / llm / testing,加上小协议 handoff / effect。你在 agent 的 nodes/builder.py 里 import 的就是这些。

wrapper-facing 子包 api/

HTTP 入口(server/mcp/cli/门户)用的:build_v1_router / envelope / auth / cases / cost / quota / ratelimit / budget。它把一次调用的"外壳治理"全包了。Day 12 精讲。

一句话:图内节点用左边,HTTP 外壳用右边。 分清这个,你就知道一段代码大概在解决"业务推理"还是"服务治理"的问题。Day 05 读的 make_specialist_node 是左边,build_v1_router 是右边。

L10

今日小结 + 动手

🧠 今天你应该能回答

  • toolkit 有哪 10 个模块?(critic/failsafe/memory/eval/tools/llm/testing/cost/specialists+reporter/api)
  • 为什么顶层 __init__ 几乎不 re-export,却偏偏破例导出 handoff/effect?(避免重加载副作用;这两个是零重依赖的叶子小协议)
  • effect.py 怎么做到"零侵入 + 误调不炸"?(ContextVar task-local 累加 + 没 bind 静默 no-op)
  • 为什么用 ContextVar 而不是全局 dict?(并发 invoke 隔离,不串数据)
  • "渐进抽象"的底层逻辑是什么?为什么 v0.3 只有 1 用户也能抽?(看"抽错的期望损失",不是死数到 2)
  • 迁移为什么留空壳 re-export?(上移与改 import 解耦成两步、各自可回滚)

✋ 动手

# 1. 看 toolkit 模块全貌
find packages/ai-trust-toolkit/src/ai_trust_toolkit -maxdepth 1 | sort

# 2. 读顶层门面:只导出 5 个符号 + 一张 API 地图 docstring
sed -n '1,49p' packages/ai-trust-toolkit/src/ai_trust_toolkit/__init__.py

# 3. 逐行读"零侵入采集"模块(今天的核心真源码)
cat packages/ai-trust-toolkit/src/ai_trust_toolkit/effect.py

# 4. 读三次真实抽象决策史(含"为什么之前不抽 / 为什么破例抽")
sed -n '379,405p' docs/CHANGELOG.md      # v0.2 specialist 工厂
sed -n '352,376p' docs/CHANGELOG.md      # v0.3 只有 1 用户也抽
sed -n '587,637p' docs/CHANGELOG.md      # 关键工程经验四条
明天预告 · Day 07:进入第一个核心能力——Critic 双层防幻觉。大模型爱编造证据、过度自信,Critic 怎么用"先 0 成本代码规则、再便宜 LLM 语义审查"两层把它拦下来?还有 PASS/FAIL/触顶三态怎么形成重试循环(就是 Day 05 builder 里那个 build_critic_router)。
← Day 05 Agent 解剖 Day 07 · Critic 双层防幻觉 →