gov-agents-server 运行时
前面 D06-D11 把可信底座的每块能力都拆过了。今天(D12)进入第 3 周主线"运行时"——把 Agent 变成能对外服务的 HTTP 接口。看清"一个镜像跑任意 agent 组合"的机制,以及一个 agent 只写业务、鉴权/记账/落库全自动的秘密——build_v1_router。这也是"跟一次请求走全程"的起点,为 D14 精读旗舰 agent 铺路。
gov-agents-server 就是小区的物业管理处:它自己不装修、不做任何业务,只负责总台事务——① 手里攥一本住户名册(AGENT_REGISTRY,写死的名单,不是挨栋扫楼);② 按今天该开哪几户(ENABLED_AGENTS)去开门迎客;③ 给每户统一装一套门禁+访客登记+计费电表(build_v1_router),业主只管住,登记记账物业全包;④ 门口的安检流程(治理链:限流→配额→预算→放行→记账→归档)访客都得过一遍。今天 8 讲就是这四件事。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 背后的东西。
一个镜像跑任意 agent 组合
平台的核心部署理念(写在 Dockerfile 开头):构建一个镜像,靠 ENABLED_AGENTS 环境变量决定这个实例跑哪些 agent。
好处:同一个制品在不同环境按需组合,SRE 相关的部一台、质量相关的部一台,互不干扰又共用一份代码。这套逻辑就一个小函数 _parse_enabled(main.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 崩给你看,绝不静默跳过。sre-rca 拼成 sre_rca:如果程序"善解人意"地忽略这个不认识的名字,服务照常起来,但那个你以为部上了的 agent根本没挂——直到线上有人调它 404 你才发现,排查半天。fail-fast 的选择是:宁可启动时就当着你的面崩,也不要带着错误配置"假装正常"地跑起来。配置错误在部署那一刻就暴露,比藏到线上才炸好得多。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_router(main.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)
import_module 把那个模块现用现加载;然后看它是不是"需要数据库的特殊 agent"(AGENTS_NEED_POOL 里只有 arch-compliance),是就位置传 pool;普通的就用关键字传 case_store;最后还用 inspect.signature 偷看这个 build_router 收不收 app 参数——收的话(比如 agent-creator 要用 sandbox)就把整个 FastAPI 实例也塞给它。build_router(case_store=...) 和后来需要 app 的新 agent 签名不一样。要么"一刀切"强制所有老 agent 改签名(改 21 处、容易漏),要么让加载器运行时探测每个 agent 到底收哪些参数、按需喂。框架选了后者——inspect.signature 让新能力(注入 app)平滑加入而不破坏任何旧 agent(源码注释原话:"旧 agent 不破")。这是"向后兼容优先"的典型取舍。三条约定用一张表看更清楚:
需要 DB 池的 agent(AGENTS_NEED_POOL 只有 arch-compliance)→ fn(pool) 位置传参
普通 agent → fn(case_store=...) 关键字传参
用 inspect.signature 探测:build_router 含 app 参数就额外注入 FastAPI 实例(给 agent-creator 的 sandbox 用)
create_app:启动流程
create_app()(main.py:375)是平台工厂。模块底部 app = create_app()(main.py:1021)就是 uvicorn 直接 import 的实例。启动做这些:
load_config()
模块顶层就加载配置,缺字段直接抛错终止——启动即校验(Day 02)。
凭据桥接 _bridge
把 yaml 里的 ANTHROPIC_API_KEY / Lark 配置在"env 未设时"补进环境变量(非空 env 优先,方便 K8s Vault 覆盖)。
建单例 CaseStore
make_case_store_from_env()——跨 agent 共享的"调用历史存储",首页据此列最近 case。
lifespan 生命周期
启动时初始化 SqliteCaseStore、给需要的 agent 建 DB 池(不可达就 warning 降级、跳过该 agent,其余照常);退出时清理。
挂路由 + 静态资源 + 可观测
遍历 enabled 逐个 _load_router 并 include;挂 /skills、/install;setup_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 完全不受影响。build_v1_router:一行装齐所有治理
这是全平台覆盖率最高、最该理解的抽象。每个 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人工反馈打分_invoke 函数(校验入参 → graph.ainvoke 跑图 → 包成 AgentOutcome)+ 1 行 build_v1_router(agent_name="rca", invoke_fn=_invoke, ...)。白拿的:
POST /invoke、/stream、/cases、/feedback 四个标准端点 + 鉴权 + 限流 + 配额 + 预算闸 + case_id 生成 + 成本记账 + 统一 envelope + 脱敏落库 + Prometheus 指标。约 20 行业务,换来一整套生产级治理。build_v1_router 收了。业务作者只写业务推理(_invoke),治理白拿——这正是整个框架"业务代码 + 调 API,可信能力开箱即用"承诺的最终兑现点。一次 invoke 的完整治理链
把前面所有天串起来——一次 POST /v1/agent/rca/invoke 进来,_do_invoke(router.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/多少钱回写进配额和预算账本(供下一次请求的②③闸判断)。限频 ratelimit:这个 user 每分钟没超 60 次?(Day 08 四件套)
token 配额 quota:日/月 token 没超?
租户预算 budget:本月 USD 没超上限?超了 429(Day 08)
生成 case_id(ULID)+ 透传父 case / 调用深度
bind CostTracker:绑定记账器(Day 11 ContextVar)
执行你的 _invoke:跑 LangGraph 图(Critic/闸门/记忆全在图里,Day 05-09)
构造 envelope:统一返回外壳
持久化 case:字段脱敏 → PII scrub → 落库(Day 08)
{"user_question":"service-A 超时"} 撞进物业大门。① 门卫先查你这个 user 这分钟按了几次门铃,没超 60 次,放行;② 查你今天/本月的 token 额度没花光;③ 查你租户这个月的钱没超上限(超了当场给你 429 请你回去);④ 前台给你贴一个唯一登记号 case_id(ULID);⑤ 给你挂上电表(CostTracker)开始计费;⑥ 这才放你进屋干活——跑 LangGraph 图(分诊→取证→综合→Critic);⑦ 干完把结果装进统一信封;⑧ 前台把你这趟脱敏后归档进案卷。你一路过的每道关,都是前面某一天学的东西。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 记录。只有"跑起来之后才炸"的才落库。统一 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
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 传身份,被封装成 CallContext(auth.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_MODE(auth.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 里也搜不到它。这是"最小信任"的设计:能不存的敏感信息就一秒都不存。router.py:40)。为什么不返"无权限"?因为 403 等于告诉攻击者"这个 case_id 真实存在,只是你没权限"——他就能靠枚举 case_id 探测哪些存在。统一返 404"查无此 case",连"存不存在"都不泄漏。sre-rca)≠ 路由名(rca,由各 agent 的 build_router 传给 build_v1_router 决定)≠ CLI/MCP 认的 canonical 名。同一个 agent 在不同层可能叫不同名字,看代码时留意 agent_name= 参数到底传的啥。今日小结 + 动手
🧠 今天你应该能回答
- 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
agentctl 命令行(给 CI/运维用)和 MCP(把 agent 暴露给 Cursor/Claude Code 等 IDE,"说人话就调 agent")。还会厘清"两个 MCP、两个 supervisor"的易错点。