ai-trust-toolkit 总览 & 渐进抽象
第 1 周你摸清了地图和一户人家的骨架;从今天起进入框架的灵魂——ai-trust-toolkit。今天不只鸟瞰它的 10 个模块,还会打开两处真源码:__init__.py(它为什么几乎什么都不导出)和 effect.py(一个"零侵入采集"模块怎么用 30 行做到"不 bind 就静默"),再顺着 CHANGELOG 读三次真实的抽象决策史——全仓最值钱的方法论就藏在这些故事里。
__init__.py 和 effect.py);③ 机房内部的管线也分层(叶子设备不反向依赖别人);④ 这机房遵循"渐进抽象":第一户要什么先在户内凑合搭,等第二户也要一模一样的,才把它挪进公共机房。这句"第二次重复才抽"是全仓最值钱的一句话,而 CHANGELOG 把每次"挪进机房"的理由都记了账。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 的抽象演进时间线)。本周讲解都以它俩和源码为准。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(业务效果指标,下一讲就拆它)。
__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
from ai_trust_toolkit import * 把 critic/memory/api 全拉起来,会触发一堆副作用和慢加载——有的模块要连 DB、要 import 重量级的 langchain/anthropic。让你"用哪个从哪个子包 import",是为了按需加载、彼此不牵连:写个只用 FakeLLM 的单测,不会因为 import 顶层就顺带把整个 api 子模块和数据库连接拉起来。handoff.py 只有一个 frozen dataclass + 一个生成 ID 的函数,effect.py 只有一个 ContextVar + 三个函数(下一讲拆),都不 import anthropic/langchain/DB。它们被业务代码高频、零成本地 import,放顶层方便且无副作用。所以这个"破例"恰恰印证了规则:不是"能不能放顶层",而是"放顶层会不会带来重加载"。__init__.py 里的 __all__ 和 docstring——那就是这个模块的"能力清单"。本周每一天都会先看对应子包的 __init__.py。effect.py 解剖:一个"零侵入采集"模块只要 30 行
先看它的"数据结构"——一个 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.py:27-31):"LangGraph 内部 asyncio.create_task 会 copy 上下文 · 并发节点累加正确"。如果用普通全局 dict,多个 invoke 并发时会互相串数据(A 的计数记到了 B 头上)。ContextVar 是"任务本地"的——每个异步任务看到自己的那份,并发天然隔离。这跟 api/cost.py 采集成本用的是同一套范式。record_effect 的 if acc is None: return(effect.py:57,注释标着 AC-1.4)。业务代码里散布着 record_effect(...) 调用,但这些代码不一定总在 invoke 路径上跑——可能被单测直接调、被老 agent 在没 bind 的地方调。如果没 bind 就抛异常,等于"打点代码反过来炸了业务",本末倒置。所以它选择静默返回:采集是"锦上添花",绝不能因为它让业务挂掉。连 key 非法、n 不是数字,都是 log.debug 记一笔就跳过,绝不抛。这是"可观测性代码永远不能拖垮业务"的铁律。内部依赖铁律:叶子不依赖任何人
Day 01 讲了"层与层之间"的依赖铁律。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 自动退化——都是为了让测试和独立调用不崩。
渐进抽象:全仓最值得学的方法论
这是整个项目最核心的工程思想,CHANGELOG 的"关键工程经验"章节(CHANGELOG.md §关键工程经验)开门见山:
🌱 "1 个 Agent 时不抽象,第 2 个 Agent 出现重复了才抽。"
CHANGELOG 里把它写成了三行阶梯:
1 个 Agent 就抽 toolkit → 接口不准(无第二样本)
2 个 Agent 出现重复才抽 → 接口刚好准
3+ 个 Agent 持续验证 → 抽象趋于稳定
意思是:不要在只有一个用例时就急着设计"通用框架"。等到第二个、第三个 agent 出现真实重复,你才真正知道哪些该抽、抽成什么样。过早抽象往往抽错,反而成为负担。
make_specialist 函数?——渐进抽象说先别:你只有这一个 agent,猜不准通用形态,抽早了大概率抽错。等 risk-reviewer 也要几乎一样的专家节点(真实重复出现),你才把它上移到 toolkit 成 make_specialist_node——也就是 Day 05 那块 _factory.py 化石记录的那次上移。🚫 过早抽象的坑
- 只有 1 个用例,猜不准通用形态
- 为"可能的未来"过度设计
- 抽象错了,改起来比不抽还贵
✅ 渐进抽象的做法
- 第 1 个 agent 允许"写死、重复"
- 第 2 个出现同样代码 → 才提取到 toolkit
- 每次抽象在 CHANGELOG 写清"触发原因"
CHANGELOG 三次真实抽象走读(含"为什么之前不抽")
docs/CHANGELOG.md 完整记录了每一次抽象的"何时抽、为什么抽、为什么之前不抽"。我们读它记载的三次真实决策——注意它们不是照搬一条死规则,而是每次都在算"抽的收益 vs 抽错的代价"。
抽出 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 行。
抽出 business_critic_l1(只有 1 个用户也抽了——反例)
触发:第 3 个 agent(alert-triage)要纯规则的轻量 Critic。CHANGELOG 诚实记了一句"潜在过度抽象信号:仅 1 个用户",但仍决定抽,理由是"接口足够通用(5 个 helper 可组合)+ 抽错的代价小(154 行 + 12 测试)· 认 OK"。
抽出 reporter 工厂(严守"第 3 次才抽")
触发:末端渲染节点在 risk-reviewer + alert-triage + arch-compliance 三次重复后才上移成 make_reporter_node——注释直接写"守渐进抽象铁律"。
api/ 子模块占了 toolkit 41% 的 LOC,文档里写"可能将来要拆出 governance/ 顶层模块,但现在不是时候(还在生产验证期)"。这是"渐进"的另一面——不仅忍住"过早抽象",也忍住"过早拆分",等生产验证充分了再动。另外三条真实工程经验(都在 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:
_factory.py 改成 re-export。"这反而是好事"——业务侧 import 路径可以渐进迁移(旧的 from ._factory import 还能用),而不是"大爆炸式"一次性全改。这跟"渐进抽象"是一体两面:抽象要渐进,迁移也要渐进。两类使用者: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 是右边。
今日小结 + 动手
🧠 今天你应该能回答
- 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 # 关键工程经验四条
build_critic_router)。