11.1 平台化：Agent 注册、版本与配额
第 10 章把单个 ResearchCopilot 加固到了「能上生产」的程度：trace 归一、成本治理、安全纵深。但生产环境里很快会出现第二个问题——当你手里不止一个 Agent 时，谁来管它们。研究助理、摘要器、引用校验器、报告生成器，各自有版本、各有配额、还要能被别的服务发现。这就从「一个服务」变成了「一个平台」。
本节把「单实例服务」推进到「多 Agent 平台」：给出一份最小可运行的 Agent 注册表（注册 / 发现 / semver 版本路由）与配额账本（调用前扣减、超限拒绝），并真跑一段「注册 → 路由 → 扣减」的完整演示。全程用假模型，零网络、零凭据。
11.1.1 从「一个服务」到「一个平台」：变了什么
卷二把 ResearchCopilot 做成了一个服务：它内部有一个 minagent 引擎、一套工具、一条编排。到了卷三，同一件事被框架重写过很多遍，于是系统里同时存在多个 Agent 实现。此时「平台」要额外承担三件事，这三件事都不是「再多写一个 Agent」能解决的：
| 维度 | 单服务（卷二） | 平台（本节） |
|---|---|---|
| Agent 从哪来 | 代码里硬编码一个 | 运行时注册、按名查找 |
| 版本怎么管 | 改代码重新部署 | 多版本共存、按约束路由 |
| 用量怎么控 | 全服务一个限额 | 按 Agent 独立配额、调用前扣减 |
| 谁能发现 | 没人需要发现 | 外部按 tag 发现「有哪些 Agent」 |
| 故障影响面 | 全站 | 单 Agent 可隔离、可降级 |
平台化不等于「上 K8s」或「接服务网格」。它的最小内核就是三样东西：注册表、版本、配额。把这三样做对，多 Agent 才谈得上治理。
11.1.2 注册表、版本路由与配额账本
先给结论式的设计：一个 AgentSpec 描述「一个 Agent 的某个版本」，一个 AgentRegistry 负责注册与解析，一个 QuotaLedger 负责扣减，一个 Platform 把三者串成一次调用。完整代码如下，可直接运行：
# platform_registry.py
from dataclasses import dataclass, field
from typing import Callable
from packaging.version import Version
from pydantic_ai import Agent
from pydantic_ai.models.function import FunctionModel
from pydantic_ai.messages import ModelResponse, TextPart
class QuotaExceeded(RuntimeError):
pass
@dataclass
class AgentSpec:
name: str
version: str
build: Callable[[], Agent]
tags: tuple[str, ...] = ()
class AgentRegistry:
def __init__(self) -> None:
self._specs: dict[str, dict[str, AgentSpec]] = {}
def register(self, spec: AgentSpec) -> None:
self._specs.setdefault(spec.name, {})[spec.version] = spec
def versions(self, name: str) -> list[str]:
return sorted(self._specs.get(name, {}), key=Version)
def resolve(self, name: str, constraint: str | None = None) -> AgentSpec:
vs = self._specs.get(name)
if not vs:
raise KeyError(f"未注册的 Agent: {name}")
keys = sorted(vs, key=Version)
if constraint is None:
return vs[keys[-1]]
if constraint.startswith("^"):
lo = Version(constraint[1:])
ok = [v for v in keys if Version(v).major == lo.major and Version(v) >= lo]
elif constraint.startswith(">="):
lo = Version(constraint[2:])
ok = [v for v in keys if Version(v) >= lo]
elif constraint.startswith("=="):
ok = [v for v in keys if v == constraint[2:]]
else:
ok = [v for v in keys if v == constraint]
if not ok:
raise LookupError(f"{name} 无匹配 {constraint!r} 的版本")
return vs[ok[-1]]
def discover(self, tag: str | None = None):
out = []
for name in sorted(self._specs):
latest = self.resolve(name)
if tag is None or tag in latest.tags:
out.append((name, self.versions(name), latest.version))
return out
@dataclass
class QuotaLedger:
limits: dict[str, int] = field(default_factory=dict)
used: dict[str, int] = field(default_factory=dict)
def grant(self, name: str, limit: int) -> None:
self.limits[name] = limit
def charge(self, name: str, n: int = 1) -> int | None:
limit = self.limits.get(name)
used = self.used.get(name, 0)
if limit is not None and used + n > limit:
raise QuotaExceeded(f"{name}: 已用 {used} + 本次 {n} > 上限 {limit}")
self.used[name] = used + n
return None if limit is None else limit - self.used[name]
class Platform:
def __init__(self, registry: AgentRegistry, ledger: QuotaLedger) -> None:
self.registry = registry
self.ledger = ledger
self.trace: list[dict] = []
def run(self, name: str, prompt: str, version: str | None = None) -> dict:
spec = self.registry.resolve(name, version)
remaining = self.ledger.charge(name)
result = spec.build().run_sync(prompt)
rec = {"agent": name, "version": spec.version, "remaining": remaining,
"tokens": result.usage.total_tokens, "output": result.output}
self.trace.append(rec)
return rec
几个设计要点值得单独说：
build是工厂不是实例。注册表存的是「怎么造出一个 Agent」，不是「一个已经造好的 Agent」。这样每个版本可以独立构造、独立替换依赖（模型、工具、prompt），也避免了共享可变状态。- 版本解析用真 semver。
packaging.version.Version能正确排序2.0.0-rc1（预发布排在正式版之前），自己写字符串排序会把10.0.0排到9.0.0前面。 charge返回剩余额度。返回值不是装饰，它让调用方能在同一次响应里带上「你还剩几次」，也方便测试直接断言。
11.1.3 版本路由：四种约束的语义
resolve 支持四种约束写法，语义如下表。注意 None（不传约束）表示「取该 Agent 的最高版本」，这和生产上「默认跑 latest」的习惯一致：
| 约束写法 | 语义 | 例子（已注册 1.0.0 / 1.1.0 / 2.0.0-rc1） |
|---|---|---|
不传（None） | 取最高版本 | 命中 2.0.0-rc1 |
^1.0 | 同主版本内、不低于 1.0 | 命中 1.1.0（不含 2.x） |
>=1.0 | 不低于 1.0，跨主版本 | 命中 2.0.0-rc1 |
==1.0.0 | 精确锁定 | 命中 1.0.0 |
一个容易踩的坑：^1.0 不会命中 2.0.0-rc1。caret 约束按主版本划分兼容区间，主版本变了就意味着破坏性变更，必须显式升级。这正是平台化要版本路由的意义——你可以让「研究助理」的调用方钉死在 ^1.0，等 2.0 稳定后再改约束，而不用改一行调用代码。
11.1.4 真跑：注册 → 路由 → 扣减
把上面这份代码接到一段驱动脚本上，注册三个版本、给 2 次配额，然后依次触发路由与超限：
def build(marker: str):
def _b() -> Agent:
fn = lambda messages, info: ModelResponse(parts=[TextPart(f"[{marker}] 研究结论")])
return Agent(FunctionModel(fn), instructions=f"research {marker}")
return _b
reg = AgentRegistry()
reg.register(AgentSpec("research", "1.0.0", build("v1.0.0"), ("stable",)))
reg.register(AgentSpec("research", "1.1.0", build("v1.1.0"), ("stable",)))
reg.register(AgentSpec("research", "2.0.0-rc1", build("v2.0.0-rc1"), ("canary",)))
ledger = QuotaLedger()
ledger.grant("research", 2)
plat = Platform(reg, ledger)
print("discover():", reg.discover())
print("discover('canary'):", reg.discover("canary"))
print("resolve ^1.0 :", plat.run("research", "q", "^1.0")["version"])
print("resolve latest:", plat.run("research", "q")["version"])
try:
plat.run("research", "q")
except QuotaExceeded as e:
print("QuotaExceeded:", e)
print("trace:", [(r["agent"], r["version"], r["remaining"]) for r in plat.trace])
在本机 venv3 里真跑（PYDANTIC_AI_NO_BANNER=1 只是为了去掉 pydantic-ai 的启动横幅），输出逐字节如下：
discover(): [('research', ['1.0.0', '1.1.0', '2.0.0-rc1'], '2.0.0-rc1')]
discover('canary'): [('research', ['1.0.0', '1.1.0', '2.0.0-rc1'], '2.0.0-rc1')]
resolve ^1.0 : 1.1.0
resolve latest: 2.0.0-rc1
QuotaExceeded: research: 已用 2 + 本次 1 > 上限 2
trace: [('research', '1.1.0', 1), ('research', '2.0.0-rc1', 0)]
四行输出各自兑现了一件事：discover 列出了版本清单与 latest；^1.0 路由到 1.1.0 而非 2.0.0-rc1；不传约束取到 2.0.0-rc1；第三次调用被 QuotaExceeded 拦住，trace 里只留下两次成功记录。配额生效的关键在于扣减发生在 run_sync 之前——先扣再跑，超限的调用根本不会触达模型。
11.1.5 注册表与 A2A：发现能力的对外延伸
本节的 discover 是进程内的发现：它回答「我这个平台里有哪些 Agent」。而 8.1 A2A 协议：Agent Card/Task/消息，起两个本地 Agent 通信
里的 Agent Card 是跨进程的发现：它回答「网络对端那个 Agent 能做什么」。两者是同一件事的两个尺度，可以对接——把注册表里的 AgentSpec 投影成一张 Agent Card，平台里的 Agent 就能被外部协作方发现：
def to_agent_card(spec: AgentSpec, endpoint: str) -> dict:
return {
"name": spec.name,
"version": spec.version,
"url": endpoint,
"capabilities": {"tags": list(spec.tags)},
}
关键取舍是版本号要不要暴露给对端。A2A 的 Agent Card 里带 version，但协作方通常只关心「能力是否兼容」，不关心你的内部补丁号。一个务实的做法是：对外只暴露主版本，对内（注册表）保留完整 semver。这样对端钉在 ^1 时，你在内部从 1.1.0 升到 1.4.0 不需要重新协商，只有升到 2.0 才触发协议层的兼容检查。
11.1.6 配额为什么必须「调用前扣减」
配额扣减的时机是个容易做错的点，值得展开：
| 扣减时机 | 后果 |
|---|---|
| 调用前扣减 | 超限请求被挡在模型之外，零成本、零副作用 |
| 调用后扣减 | 超限请求已经花掉额度，扣减变成「记账」而非「限流」 |
| 按 token 事后结算 | 需要模型返回真实 usage，但假模型给不出真实 token |
正确顺序是 resolve → charge → run。charge 抛异常时，run_sync 还没被调用，所以既没有网络开销，也不会有任何工具副作用。如果额度是按租户而非按 Agent 划分的（卷二的限流就是这样），把 name 换成 f"{tenant}:{agent}" 即可，QuotaLedger 本身不用改——它只认 key。
一个必须诚实交代的边界：本节演示的 charge 是单进程内存账本。真实平台要抗并发，charge 必须是原子的（Redis 的 INCR 加 Lua，或数据库行锁），否则两个并发请求会各自读到「还剩 1」然后都扣成功。这个原子性问题卷二第 6.3 节已经解决过，本节不重复，只标注「把 charge 换成原子实现」这一处即可。
11.1.7 这套注册表替你做了什么、你又让渡了什么
| 你得到的 | 你让渡的 |
|---|---|
| 运行时可发现、可路由，不必改代码切版本 | 引入了一层间接：调 Agent 要先过注册表 |
| 多版本共存，可灰度、可回滚 | 要维护版本号与 tag 的纪律，否则注册表会烂 |
| 调用前限额，超限零成本 | 配额状态本身要持久化、要原子，是新的运维面 |
| 统一的 trace 入口 | 注册表成了单点，它挂了全平台挂 |
一句话：注册表把「有哪些 Agent、跑哪个版本、还剩多少额度」从散落的代码收敛成一份可查询的状态，代价是你多了一个必须高可用的核心组件。这和第 10 章「可观测/成本/安全」的加固是同一种思路——把散在各处的隐式规则，变成一处显式的、可测的机制。
下一节 11.2 真实 API 冒烟：假模型 vs 真模型量化差距 会把本节这套「用假模型跑得全绿」的演示，拿去和真模型对照，量一量假模型到底漏掉了什么；想回看生产化的另一半，见 10.3 安全纵深：注入/越权/供应链审计 。
阅读导航：上一节：10.3 安全纵深：注入/越权/供应链审计 · 下一节：11.2 真实 API 冒烟：假模型 vs 真模型量化差距 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。