Day 12 / 共 20 天 · 第 3 周 运行时与实战

gov-agents-server 运行时

前面 D06-D11 把可信底座的每块能力都拆过了。今天(D12)进入第 3 周主线"运行时"——把 Agent 变成能对外服务的 HTTP 接口。看清"一个镜像跑任意 agent 组合"的机制,以及一个 agent 只写业务、鉴权/记账/落库全自动的秘密——build_v1_router。这也是"跟一次请求走全程"的起点,为 D14 精读旗舰 agent 铺路。

📍 你在 20 天里的位置(可信底座 → 运行时 → 精读)
D09 记忆 D10 评测 D11 安全底座 D12 server 运行时 D13 CLI/MCP D14 精读 sre-rca D15 全家福+元 Agent
💡 用一个类比先兜住今天(延续「盖楼/物业」世界观) gov-agents-server 就是小区的物业管理处:它自己不装修、不做任何业务,只负责总台事务——① 手里攥一本住户名册(AGENT_REGISTRY,写死的名单,不是挨栋扫楼);② 按今天该开哪几户(ENABLED_AGENTS)去开门迎客;③ 给每户统一装一套门禁+访客登记+计费电表(build_v1_router),业主只管住,登记记账物业全包;④ 门口的安检流程(治理链:限流→配额→预算→放行→记账→归档)访客都得过一遍。今天 8 讲就是这四件事。
L01

server 的角色:只做托管,不含业务

packages/gov-agents-server 是平台主进程——一个 FastAPI 服务。它自己没有任何业务逻辑,只干四件事:

  • 维护一张"哪个 agent 名对应哪个模块"的注册表;
  • ENABLED_AGENTS 环境变量,懒加载并挂载对应 agent 的路由;
  • 提供首页、健康检查、静态资源(skills 目录、安装脚本);
  • 统一接可观测性。

主文件 gov_agents_server/main.py(约 1021 行)。它是 Day 02 "启动平台"那条命令 uvicorn gov_agents_server.main:app 背后的东西。

L02

一个镜像跑任意 agent 组合

平台的核心部署理念(写在 Dockerfile 开头):构建一个镜像,靠 ENABLED_AGENTS 环境变量决定这个实例跑哪些 agent

一个镜像build 一次
+
ENABLED_AGENTS环境变量
=
实例A:sre-rca,alert-triage
实例B:doc-checker,risk-reviewer
实例C:不设 = 挂全部

好处:同一个制品在不同环境按需组合,SRE 相关的部一台、质量相关的部一台,互不干扰又共用一份代码。这套逻辑就一个小函数 _parse_enabledmain.py:289):

# main.py:289
def _parse_enabled(raw: str | None) -> list[str]:
    if not raw or not raw.strip():
        return list(AGENT_REGISTRY)                 # 不设 ENABLED_AGENTS → 挂全部
    names = [a.strip() for a in raw.split(",") if a.strip()]
    unknown = [n for n in names if n not in AGENT_REGISTRY]
    if unknown:                                     # 写了名册上没有的名字
        raise RuntimeError(f"未知 agent: {unknown!r}。合法值: {sorted(AGENT_REGISTRY)}")
    return names
👶 大白话读环境变量 ENABLED_AGENTS:空的就返回注册表里全部 agent;否则按逗号切开、去空格,然后逐个核对名字在不在注册表里。只要有一个不认识,当场 raise RuntimeError 崩给你看,绝不静默跳过。
💡 设计取舍①:为什么拼错名字要"崩",不是忽略?这叫 fail-fast(快速失败)。想象你部署时把 sre-rca 拼成 sre_rca:如果程序"善解人意"地忽略这个不认识的名字,服务照常起来,但那个你以为部上了的 agent根本没挂——直到线上有人调它 404 你才发现,排查半天。fail-fast 的选择是:宁可启动时就当着你的面崩,也不要带着错误配置"假装正常"地跑起来。配置错误在部署那一刻就暴露,比藏到线上才炸好得多。
L03

AGENT_REGISTRY:显式注册表,不是自动扫描

server 怎么知道有哪些 agent?靠 main.py:62-81 一张写死的字典——{agent名: (模块路径, 工厂函数名)}

AGENT_REGISTRY: dict[str, tuple[str, str]] = {
    "sre-rca":        ("sre_rca.server",        "build_router"),
    "risk-reviewer":  ("risk_reviewer.server",  "build_router"),
    "alert-triage":   ("alert_triage.server",   "build_router"),
    # ... 共 18 条
}

启动时按名字用 importlib.import_module 懒加载对应模块,取出它的 build_router 函数。看真实的 _load_routermain.py:299),它有三条调用约定

# main.py:299
def _load_router(name, *, case_store=None, pool=None, app=None) -> APIRouter:
    import inspect as _inspect
    mod_path, fn_name = AGENT_REGISTRY[name]         # 查名册拿到("sre_rca.server","build_router")
    mod = importlib.import_module(mod_path)          # 懒加载:用到才 import(启动更快)
    fn = getattr(mod, fn_name)
    if name in AGENTS_NEED_POOL:                     # 约定①:需要 DB 池的 agent
        if pool is None:
            raise RuntimeError(f"Agent '{name}' requires a database pool but none was provided.")
        return fn(pool)                              #   → 位置传 pool
    kwargs = {"case_store": case_store}              # 约定②:普通 agent → 关键字传 case_store
    try:
        sig = _inspect.signature(fn)                 # 约定③:探测 build_router 签名
        if "app" in sig.parameters and app is not None:
            kwargs["app"] = app                      #   含 app 参数就注入 FastAPI 实例
    except (ValueError, TypeError):
        pass                                         #   拿不到签名(罕见)→ 退化为不传
    return fn(**kwargs)
👶 大白话拿到 agent 名字后:先查名册翻出"它在哪个模块、工厂函数叫啥",import_module 把那个模块现用现加载;然后看它是不是"需要数据库的特殊 agent"(AGENTS_NEED_POOL 里只有 arch-compliance),是就位置传 pool;普通的就用关键字传 case_store;最后还用 inspect.signature 偷看这个 build_router 收不收 app 参数——收的话(比如 agent-creator 要用 sandbox)就把整个 FastAPI 实例也塞给它。
💡 设计取舍②:为什么用 inspect.signature 探测,而不是让所有 agent 签名统一?因为 21 个 agent 是陆续加进来的,早期的 build_router(case_store=...) 和后来需要 app 的新 agent 签名不一样。要么"一刀切"强制所有老 agent 改签名(改 21 处、容易漏),要么让加载器运行时探测每个 agent 到底收哪些参数、按需喂。框架选了后者——inspect.signature 让新能力(注入 app)平滑加入而不破坏任何旧 agent(源码注释原话:"旧 agent 不破")。这是"向后兼容优先"的典型取舍。

三条约定用一张表看更清楚:

1

需要 DB 池的 agent(AGENTS_NEED_POOL 只有 arch-compliance)→ fn(pool) 位置传参

2

普通 agent → fn(case_store=...) 关键字传参

3

inspect.signature 探测:build_routerapp 参数就额外注入 FastAPI 实例(给 agent-creator 的 sandbox 用)

⚠️ 小白常以为「把新 agent 的文件夹丢进 apps/,server 会自动发现它」。不会。这是本框架有意的"显式 > 隐式"取舍——名册上没登记的住户,物业根本不认。所以加新 agent 一定要手动在这张表加一行(D19 会亲手做)。好处是:翻开这本名册,一眼看全平台到底有哪些 agent,不用去扫目录猜。
重点:是显式注册表,不是文件系统自动发现。 所以加一个新 agent 必须手动往这张表加一行(还要加 workspace 依赖 + pyproject)。Day 19 起新 agent 时你会亲手改它。这种"显式 > 隐式"的选择,牺牲了一点便利,换来了"打开这张表就知道平台到底有哪些 agent"的可读性。
L04

create_app:启动流程

create_app()main.py:375)是平台工厂。模块底部 app = create_app()main.py:1021)就是 uvicorn 直接 import 的实例。启动做这些:

1

load_config()

模块顶层就加载配置,缺字段直接抛错终止——启动即校验(Day 02)。

2

凭据桥接 _bridge

把 yaml 里的 ANTHROPIC_API_KEY / Lark 配置在"env 未设时"补进环境变量(非空 env 优先,方便 K8s Vault 覆盖)。

3

建单例 CaseStore

make_case_store_from_env()——跨 agent 共享的"调用历史存储",首页据此列最近 case。

4

lifespan 生命周期

启动时初始化 SqliteCaseStore、给需要的 agent 建 DB 池(不可达就 warning 降级、跳过该 agent,其余照常);退出时清理。

5

挂路由 + 静态资源 + 可观测

遍历 enabled 逐个 _load_router 并 include;挂 /skills/installsetup_observability/metrics

第 4 步的"韧性"最值得看真代码。lifespan 里给 pool 类 agent(arch-compliance)建 MySQL 连接池,注意它被 try/except 整个包住main.py:400):

# main.py:400  lifespan 内建 aiomysql 连接池
pool: aiomysql.Pool | None = None
if needs_pool:
    db = _cfg.database.arch_compliance
    if db is None:
        log.error("configs/<APP_ENV>.yaml 缺 database.arch_compliance 配置 · 跳过 pool agents")
    else:
        try:
            pool = await aiomysql.create_pool(host=db.host, port=db.port, ..., minsize=2, maxsize=10)
        except Exception as e:
            log.warning("aiomysql pool 创建失败(%s: %s)· 跳过 pool agents · 平台其他 agent 正常",
                        type(e).__name__, e)          # ← 降级:不 raise!
            pool = None
    if pool is not None:
        for name in needs_pool:
            app.include_router(_load_router(name, pool=pool))   # 有池才挂业务路由
👶 大白话只有 arch-compliance 这种 agent 需要数据库(MySQL,用 aiomysql 建池,最少 2 条最多 10 条连接)。建池要么成功、要么except 接住只打一条 warning 就 pool = None——注意这里没有 raise。池是 None 时就不挂这个 agent 的业务路由,但整个平台照常启动,其它 17 个 agent 完全不受影响。
平台启动 建 MySQL 池 (仅 arch-compliance) 成功 连不上 挂 arch-compliance 业务路由 warning + 跳过它,其它 agent 照常 ← 关键:不 raise,平台不整体崩
图注:数据库连不上时的降级分叉——单个 agent 掉线,平台整体存活。
设计取舍③·韧性设计:某个 agent 需要的数据库连不上时,平台不会整体启动失败,而是打个 warning、跳过这个 agent,让其它 agent 正常服务。这种"部分降级而非全盘崩溃"的思路,和 Day 08 的失败闸门、Day 11 的路由 fail-safe 一脉相承——但要跟上一讲的 fail-fast 区分开:"配置写错"(拼错 agent 名)要 fail-fast 当场崩,"外部依赖临时不可达"(数据库抖动)要降级存活。人为错误早暴露、环境问题别放大,是两种相反但都正确的选择。
L05

build_v1_router:一行装齐所有治理

🤔 痛点:如果没有 build_v1_router,21 个 agent 会怎样?鉴权、限流、配额、预算、生成 case_id、成本追踪、构造统一 envelope、脱敏落库、推 Prometheus 指标……这一整套每个 agent 都要、且必须一模一样。没有它,21 个 agent 就得各抄一遍这坨代码——抄错一处(比如漏了脱敏)就是事故,改一处(比如加个新指标)要改 21 遍。这正是"每户自己装门禁、装出 21 种不一样的门禁"的噩梦。

这是全平台覆盖率最高、最该理解的抽象。每个 agent 的 server.py 里的 build_router,核心就是调 toolkit 的 build_v1_router()api/router.py:261)。业务只写一个 _invoke 函数,其余全自动:

# apps/sre-rca-agent/sre_rca/server.py:66
def build_router(case_store=None):
    router = APIRouter(tags=["sre-rca"])
    graph = build_rca_graph()                    # 你的 LangGraph 图
    store = case_store or make_case_store_from_env()

    async def _invoke(body, ctx, tracker) -> AgentOutcome:
        # ↑ 你只写这个:校验入参 → 跑图 → 包成结果
        with anthropic_key_override(ctx.anthropic_api_key):
            result = await graph.ainvoke(initial, config)
        return AgentOutcome(...)

    router.include_router(build_v1_router(         # ← 一行装齐下面全部
        agent_name="rca", invoke_fn=_invoke,
        stream_fn=_stream, case_store=store))
    return router

这一行 build_v1_router 自动给这个 agent 挂上标准端点:

POST /v1/agent/{name}/invoke同步调用
POST /v1/agent/{name}/streamSSE 流式(逐节点推)
GET /v1/agent/{name}/cases历史调用列表 + 单条
POST /v1/agent/{name}/feedback人工反馈打分
📝 举个例子:业主实际写多少、白拿多少 业主(sre-rca 作者)写的:1 个 _invoke 函数(校验入参 → graph.ainvoke 跑图 → 包成 AgentOutcome)+ 1 行 build_v1_router(agent_name="rca", invoke_fn=_invoke, ...)
白拿的:POST /invoke/stream/cases/feedback 四个标准端点 + 鉴权 + 限流 + 配额 + 预算闸 + case_id 生成 + 成本记账 + 统一 envelope + 脱敏落库 + Prometheus 指标。约 20 行业务,换来一整套生产级治理。
为什么这个抽象这么重要? 因为"鉴权、限流、配额、预算、生成 case_id、成本追踪、构造统一 envelope、脱敏落库、推 Prometheus 指标、老路径 alias"这一整套每个 agent 都需要、且必须一致的东西,全被 build_v1_router 收了。业务作者只写业务推理(_invoke),治理白拿——这正是整个框架"业务代码 + 调 API,可信能力开箱即用"承诺的最终兑现点。
L06

一次 invoke 的完整治理链

把前面所有天串起来——一次 POST /v1/agent/rca/invoke 进来,_do_invokerouter.py:318)按顺序过这条链。开头这几行就是治理三闸 + 建账(前面的日子学的东西全在这里汇合):

# api/router.py:325  _do_invoke 开头
await get_rate_limiter().check_or_raise(ctx.user_id)   # ① 限频:超额抛 429(不计 case/cost)
await check_quota(user_id=ctx.user_id)                 # ② token 配额:查上次累计(不含本次)
await check_budget(tenant_id=ctx.tenant_id)            # ③ 租户月度 USD cap:超 → 429
case_id = new_case_id()                                # ④ 生成 26 字符 ULID
...
tracker = CostTracker(agent=agent_name, user_id=ctx.user_id, tenant_id=ctx.tenant_id)
try:
    with bind_tracker(tracker):                        # ⑤ 绑记账器(Day 11 ContextVar)
        outcome = await invoke_fn(body, ctx, tracker)  # ⑥ 执行你写的 _invoke(跑图)
    metrics = tracker.finalize(status=outcome.status, retry_count=outcome.retry_count)  # 抄表
    envelope = _build_envelope(agent=agent_name, case_id=case_id, outcome=outcome, ...)  # ⑦ 装信封
    await _persist_case(..., envelope=envelope, sensitive_fields=sensitive)              # ⑧ 脱敏落库
    await record_quota_usage(...); await record_budget_cost(...)                          # 回写用量
    return envelope
👶 大白话三道闸 check_* 都是"不通过就直接抛 429 请你回去",通过了才发一个 case_id(唯一登记号)。然后 with bind_tracker 装上电表,你写的 _invoke 只在第 ⑥ 步被调用——业务只占这一格。跑完 finalize 抄电表得到 metrics,装进统一 envelope,脱敏后落库,最后把这次用了多少 token/多少钱回写进配额和预算账本(供下一次请求的②③闸判断)。
1

限频 ratelimit:这个 user 每分钟没超 60 次?(Day 08 四件套)

2

token 配额 quota:日/月 token 没超?

3

租户预算 budget:本月 USD 没超上限?超了 429(Day 08)

4

生成 case_id(ULID)+ 透传父 case / 调用深度

5

bind CostTracker:绑定记账器(Day 11 ContextVar)

6

执行你的 _invoke:跑 LangGraph 图(Critic/闸门/记忆全在图里,Day 05-09)

7

构造 envelope:统一返回外壳

8

持久化 case:字段脱敏 → PII scrub → 落库(Day 08)

🚶 第一人称走一遍:现在你就是那个 HTTP 请求 你带着 {"user_question":"service-A 超时"} 撞进物业大门。 门卫先查你这个 user 这分钟按了几次门铃,没超 60 次,放行; 查你今天/本月的 token 额度没花光; 查你租户这个月的钱没超上限(超了当场给你 429 请你回去); 前台给你贴一个唯一登记号 case_id(ULID); 给你挂上电表(CostTracker)开始计费; 这才放你进屋干活——跑 LangGraph 图(分诊→取证→综合→Critic); 干完把结果装进统一信封; 前台把你这趟脱敏后归档进案卷。你一路过的每道关,都是前面某一天学的东西。
请求 POST 限频 配额 预算 case_id+电表④⑤ 跑图 _invoke 分诊→取证→Critic ⑥ envelope 脱敏落库 一次 /v1/agent/rca/invoke 从进门到归档 ← 治理闸(toolkit) →|← 业务图(你写的) →|← 治理收尾(toolkit) →
图注:业务代码只负责中间那一格(⑥),两头的安检与归档全是 build_v1_router 自动加的。
这就是"跟一次请求走全程"的主线。你现在应该能看懂:一个请求先过一堆治理闸(限频/配额/预算),再进业务图(分诊→取证→综合→Critic),出来包成统一信封、脱敏落库、推指标。前 11 天的知识点在这一条链上全部各就各位。
⚠️ 设计取舍④:请求失败时(业务抛异常、或 BudgetExceeded 硬熔断),源码仍然落库 + 仍然回写配额用量router.py:378-400 的 BudgetExceeded 分支、router.py:428-451 的通用异常分支)。为什么失败了还要记账?源码注释写得直白:"失败 invoke 仍 record quota · 防恶意刷 fail-fast 仍烧 token"——一次调用哪怕最后失败,只要它已经烧了大模型的 token/钱,就必须记进账本,否则有人靠"故意触发失败"来白嫖额度。失败也是一次真实消耗。
⚠️ 边界/易错点:并不是所有异常都落库。业务 _invoke 主动抛的 HTTPException(如入参校验失败 422、找不到 404)会被原样 re-raise、不持久化router.py:404-406)——因为那是"请求本身不合法",还没真正开跑、没烧钱,不该占一条 case 记录。只有"跑起来之后才炸"的才落库。
L07

统一 envelope & 鉴权契约

统一 envelope(api/envelope.py:63)

所有 agent 返回的外壳完全一致(Day 02 见过),这是真实定义:

# api/envelope.py:22
InvokeStatus = Literal["success", "critic_end", "failed"]

# api/envelope.py:63
class InvokeResponse(BaseModel):
    model_config = ConfigDict(extra="forbid")   # 多一个字段都报错(严格)
    case_id: str                                 # 26 字符 ULID(带校验)
    agent: str
    status: InvokeStatus                         # success / critic_end / failed
    result: dict[str, Any]                       # agent 的业务字段装这
    metrics: Metrics                             # 平台维度量度(见下)
    trace_id: str
    deprecated_path: str | None = None           # 走老路径时的提示

    @field_validator("case_id")                  # api/envelope.py:81
    @classmethod
    def _check_case_id_format(cls, v: str) -> str:
        if len(v) != 26:                         # ULID 必须 26 字符,否则拒绝
            raise ValueError(f"case_id 必须 26 char ULID,当前 {len(v)} char: {v!r}")
        return v

metrics 里的字段名要记准(envelope.py:40),别写成简写:

# api/envelope.py:40
class Metrics(BaseModel):
    elapsed_ms: int = 0            # 耗时毫秒
    llm_tokens_in: int = 0        # 输入 token(不是 tokens_in!)
    llm_tokens_out: int = 0       # 输出 token
    cache_creation_tokens: int = 0; cache_read_tokens: int = 0   # v0.6+ prompt caching
    estimated_cost_usd: float = 0.0                              # 预估花费 USD
    retry_count: int = 0
    budget_exceeded_scope: str | None = None                    # 硬熔断时填 per_invoke/per_node
InvokeResponse(统一信封 · 21 个 agent 都长这样) case_id · 26 ULID agent · trace_id status: success/critic_end/failed result: dict ← 业务字段只能装这里 (hypotheses / conclusion / confidence … 每个 agent 自己定) metrics: Metrics ← 平台自动填(业务白拿) elapsed_ms / llm_tokens_in|out / estimated_cost_usd / retry_count
图注:平台字段(紫)神圣不可污染,业务字段一律装进 result(绿),度量由平台自动填入 metrics(橙)。
💡 设计取舍⑤:为什么 envelope 用 extra="forbid" + case_id 强校验?因为这个信封是全平台 21 个 agent 对外的统一契约,下游(portal 前端、CLI、其它 agent)都按这个结构解析。extra="forbid"envelope.py:71)意味着"业务想往信封顶层塞私货字段?直接报错"——业务字段只能装进 result,平台字段(case_id/status/metrics…)神圣不可污染。case_id 强制 26 字符则保证它永远是合法 ULID,下游拿去当主键、做枚举检查都不会踩到脏值。用一点"严格"换整条调用链的"可信"。

鉴权 header 契约(api/auth.py)

请求通过几个 header 传身份,被封装成 CallContextauth.py:46)传给业务:

# api/auth.py:46
@dataclass
class CallContext:
    user_id: str
    tenant_id: str
    trace_id: str
    trust_level: TrustLevel               # "high" | "low" | "anonymous"
    anthropic_api_key: str | None = None  # 来自 X-Anthropic-Api-Key · 用完即弃 · 不落库

三档鉴权模式 PLATFORM_AUTH_MODEauth.py:32):open(默认,缺 X-User-Id 自动匿名)/ header(灰度过渡)/ strict(缺 X-User-Id 直接 401)。

⚠️ 设计取舍⑥·安全:anthropic_api_key 来自 per-request 的 X-Anthropic-Api-Key header,用完即弃、绝不落库(源码注释明确)。这样每个租户可以带自己的 key 调用,而平台永远不持有你的 key——落库的 case 里也搜不到它。这是"最小信任"的设计:能不存的敏感信息就一秒都不存。
⚠️ 边界/易错点:跨 agent / 跨 user 访问别人的 case 时,源码返回的是 404 而不是 403router.py:40)。为什么不返"无权限"?因为 403 等于告诉攻击者"这个 case_id 真实存在,只是你没权限"——他就能靠枚举 case_id 探测哪些存在。统一返 404"查无此 case",连"存不存在"都不泄漏
命名的三层别混淆(Day 13 还会遇到):注册表名(sre-rca)≠ 路由名(rca,由各 agent 的 build_router 传给 build_v1_router 决定)≠ CLI/MCP 认的 canonical 名。同一个 agent 在不同层可能叫不同名字,看代码时留意 agent_name= 参数到底传的啥。
L08

今日小结 + 动手

🧠 今天你应该能回答

  • server 的四个职责是什么?(注册表/按 ENABLED_AGENTS 挂载/首页健康检查/可观测)
  • "一个镜像跑任意组合"怎么实现?(ENABLED_AGENTS 环境变量)
  • agent 是怎么被发现的?(显式 AGENT_REGISTRY + importlib 懒加载,非自动扫描)
  • build_v1_router 帮业务免写了哪些东西?(鉴权/限流/配额/预算/envelope/落库/指标)
  • 一次 invoke 要过哪 8 步治理链?

✋ 动手

# 1. 读注册表和三条调用约定
sed -n '62,81p' packages/gov-agents-server/src/gov_agents_server/main.py
sed -n '299,331p' packages/gov-agents-server/src/gov_agents_server/main.py

# 2. 读一个 agent 的 build_router 怎么用 build_v1_router
sed -n '66,125p' apps/sre-rca-agent/sre_rca/server.py

# 3. 读 v1 路由工厂 + 一次 invoke 的治理链
sed -n '261,340p' packages/ai-trust-toolkit/src/ai_trust_toolkit/api/router.py

# 4. 读统一 envelope
sed -n '40,95p' packages/ai-trust-toolkit/src/ai_trust_toolkit/api/envelope.py

# 5. 起服务看首页和自我说明书
# curl localhost:8080/api ; curl localhost:8080/agents/rca/graph
明天预告 · Day 13:HTTP 只是一种入口。Day 13 讲另外两种——agentctl 命令行(给 CI/运维用)和 MCP(把 agent 暴露给 Cursor/Claude Code 等 IDE,"说人话就调 agent")。还会厘清"两个 MCP、两个 supervisor"的易错点。
← Day 11 安全工具 Day 13 · CLI 与 MCP 接入 IDE →