从单体 Agent 到 Agent 平台：一次架构迁移的技术方案和组织挑战

一、深度引言与场景痛点

2024 年初我们的 AI 应用还只有一个 Agent — 一个智能客服机器人。用了大半年反响不错，业务部门开始来提需求：法务要一个合同审查 Agent、HR 要一个简历筛选 Agent、运营要一个内容审核 Agent。起初我们很乐观——把第一个 Agent 的代码 copy-paste 一份，改改 Prompt 和 Tool 不就行了？

一个月后我们就笑不出来了。4 个 Agent 各自维护一套代码，LLM 调用逻辑、错误处理、日志采集全部重复。Prompt 模板散落在 4 个不同的 repo 里，想统一升级模型版本？每个 Agent 都要单独改配置、单独部署。更头疼的是 Tool 的复用——合同审查和简历筛选都需要"文档解析"这个能力，但各自的实现方式完全不同，一份 PDF 在两个 Agent 里解析出来的文本结构都不一样。

组织层面的挑战更隐蔽。Agent 的开发者是各业务线的工程师，他们对 Prompt 工程和 LLM 推理的理解程度参差不齐。法务团队写的 Agent 在输入异常时会直接抛 unhandled exception，运营团队写的 Agent 没有做任何 token 用量统计——月末对账单时才发现一个 Agent 一个月烧掉了 2000 美元的 API 费用。

单体 Agent 在验证 PMF 阶段没问题，但当 Agent 数量从 1 变成 N 时，共享能力必须平台化、治理规则必须标准化、开发体验必须工具化。这就是单体到平台的迁移。

二、底层机制与原理深度剖析

从单体到平台的架构演进，核心是把 Agent 的共性能力抽离为平台层，把业务差异封装为可插拔的配置和插件：

平台层的六个模块各司其职：

Agent 运行时引擎：标准化的 Agent 生命周期管理（创建→执行→监控→销毁），每个业务 Agent 只是一组配置 + Prompt + Tool 引用的组合。
Tool 注册中心：把所有 Tool 统一注册和管理，支持版本化、权限控制和限流，Agent 通过声明式引用接入 Tool。
Prompt 管理服务：集中管理 Prompt 模板，支持 A/B 测试、版本回滚和效果评估。
LLM Gateway：统一路由 LLM 调用，实现负载均衡、重试、fallback、token 计费。
可观测性中心：统一采集日志、trace、指标，按 Agent 维度聚合。
权限 & 安全：统一的认证授权、内容安全过滤、越权操作拦截。

三、生产级代码实现

import asyncio
import logging
import time
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Optional, Protocol

from pydantic import BaseModel, Field, ValidationError

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# ── Tool 注册中心 ────────────────────────────────────────

class ToolProtocol(Protocol):
"""Tool 接口协议"""
name: str
version: str

async def execute(self, **kwargs) -> dict: …

class ToolRegistry:
"""全局 Tool 注册中心"""

_tools: dict[str, dict[str, type]] = {} # {name: {version: ToolClass}}

@classmethod
def register(cls, tool_cls: type, version: str = "v1"):
tool_name = getattr(tool_cls, "name", tool_cls.__name__)
if tool_name not in cls._tools:
cls._tools[tool_name] = {}
cls._tools[tool_name][version] = tool_cls
logger.info(f"Tool 注册: {tool_name}@{version}")

@classmethod
def get(cls, name: str, version: str = "v1") -> type:
versions = cls._tools.get(name, {})
if version not in versions:
available = list(versions.keys())
raise ValueError(f"Tool {name}@{version} 不存在，可用: {available}")
return versions[version]

@classmethod
def list_tools(cls) -> list[dict]:
return [
{"name": name, "versions": list(versions.keys())}
for name, versions in cls._tools.items()
]

# ── Prompt 管理服务 ───────────────────────────────────────

class PromptTemplate(BaseModel):
"""Prompt 模板"""
name: str
version: str
system_prompt: str
user_prompt_template: str
variables: list[str] = Field(default_factory=list)
metadata: dict = Field(default_factory=dict)

def render(self, **kwargs) -> dict[str, str]:
try:
system = self.system_prompt.format(**kwargs)
user = self.user_prompt_template.format(**kwargs)
except KeyError as e:
raise ValueError(f"Prompt 变量缺失: {e}")
return {"system": system, "user": user}

class PromptManager:
"""集中式 Prompt 管理"""

_prompts: dict[str, dict[str, PromptTemplate]] = {}

@classmethod
def register(cls, prompt: PromptTemplate):
if prompt.name not in cls._prompts:
cls._prompts[prompt.name] = {}
cls._prompts[prompt.name][prompt.version] = prompt

@classmethod
def get(cls, name: str, version: str = "v1") -> PromptTemplate:
versions = cls._prompts.get(name, {})
if version not in versions:
raise ValueError(f"Prompt {name}@{version} 不存在")
return versions[version]

# ── Agent 配置（声明式定义） ─────────────────────────────

class AgentConfig(BaseModel):
"""业务 Agent 的声明式配置"""
agent_id: str
name: str
description: str = ""
prompt_name: str
prompt_version: str = "v1"
tools: list[dict] = Field(default_factory=list) # [{name, version}]
llm_model: str = "gpt-4o-mini"
temperature: float = 0.0
max_iterations: int = 10
max_tokens: int = 4096
error_budget: int = 3
owner: str = ""
sla_target_ms: int = 5000 # P95 延迟目标

# ── Agent 运行时引擎 ─────────────────────────────────────

class AgentRuntime:
"""标准化的 Agent 运行时"""

def __init__(self, config: AgentConfig):
self.config = config
self._tools: dict[str, Any] = {}
self._prompt: Optional[PromptTemplate] = None
self._init_time = time.time()
self._call_count = 0
self._total_tokens = 0

async def _load_prompt(self):
"""从 PromptManager 加载 Prompt 模板"""
try:
self._prompt = PromptManager.get(
self.config.prompt_name, self.config.prompt_version
)
except ValueError as e:
logger.error(f"Prompt 加载失败 {self.config.agent_id}: {e}")
raise

async def _load_tools(self):
"""从 ToolRegistry 加载声明式引用的 Tool"""
for tool_ref in self.config.tools:
name = tool_ref["name"]
version = tool_ref.get("version", "v1")
try:
tool_cls = ToolRegistry.get(name, version)
self._tools[name] = tool_cls()
except ValueError as e:
logger.error(f"Tool 加载失败 {self.config.agent_id}/{name}: {e}")
raise

async def initialize(self):
"""初始化 Agent（加载 Prompt 和 Tool）"""
await asyncio.gather(self._load_prompt(), self._load_tools())
logger.info(f"Agent [{self.config.agent_id}] 初始化完成, tools={list(self._tools.keys())}")

async def execute(self, user_input: str, context: Optional[dict] = None) -> dict:
"""执行一次 Agent 对话"""
start_time = time.time()
self._call_count += 1

try:
if self._prompt is None:
raise RuntimeError("Agent 未初始化")

# 渲染 Prompt
render_vars = {"user_input": user_input, **(context or {})}
rendered = self._prompt.render(**render_vars)

# 构造 Tool 描述给 LLM
tool_descriptions = "\\n".join(
f"- {name}: {getattr(tool, 'description', 'no description')}"
for name, tool in self._tools.items()
)

# 模拟 LLM 调用（实际应走 LLM Gateway）
response = (
f"[{self.config.agent_id}] 收到: {user_input}\\n"
f"可用工具: {tool_descriptions}\\n"
f"系统指令: {rendered['system'][:100]}…"
)

# 模拟 token 统计
estimated_tokens = len(user_input) + len(rendered["system"])
self._total_tokens += estimated_tokens

elapsed_ms = (time.time() – start_time) * 1000
sla_breach = elapsed_ms > self.config.sla_target_ms

return {
"agent_id": self.config.agent_id,
"response": response,
"elapsed_ms": elapsed_ms,
"tokens_used": estimated_tokens,
"tool_calls": len(self._tools),
"sla_breach": sla_breach,
}
except Exception as e:
elapsed_ms = (time.time() – start_time) * 1000
logger.exception(f"Agent [{self.config.agent_id}] 执行异常: {e}")
return {
"agent_id": self.config.agent_id,
"error": str(e),
"elapsed_ms": elapsed_ms,
"sla_breach": elapsed_ms > self.config.sla_target_ms,
}

def get_stats(self) -> dict:
return {
"agent_id": self.config.agent_id,
"uptime_seconds": time.time() – self._init_time,
"call_count": self._call_count,
"total_tokens": self._total_tokens,
}

# ── 平台入口 ─────────────────────────────────────────────

class AgentPlatform:
"""Agent 平台主控"""

def __init__(self):
self._agents: dict[str, AgentRuntime] = {}

async def deploy_agent(self, config: AgentConfig) -> AgentRuntime:
"""部署一个 Agent"""
if config.agent_id in self._agents:
raise ValueError(f"Agent {config.agent_id} 已存在")

runtime = AgentRuntime(config)
await runtime.initialize()
self._agents[config.agent_id] = runtime
logger.info(f"Agent [{config.agent_id}] 已部署")
return runtime

async def stop_agent(self, agent_id: str):
if agent_id not in self._agents:
raise ValueError(f"Agent {agent_id} 不存在")
stats = self._agents[agent_id].get_stats()
del self._agents[agent_id]
logger.info(f"Agent [{agent_id}] 已停止, stats={stats}")

async def execute(self, agent_id: str, user_input: str, context: Optional[dict] = None) -> dict:
if agent_id not in self._agents:
raise ValueError(f"Agent {agent_id} 未部署")
return await self._agents[agent_id].execute(user_input, context)

def list_agents(self) -> list[dict]:
return [
{"agent_id": aid, **agent.get_stats()}
for aid, agent in self._agents.items()
]

# ── 使用示例 ─────────────────────────────────────────────

async def main():
# 注册共享 Tool
class DocumentParser:
name = "document_parser"
version = "v1"
description = "解析 PDF/Word 文档为结构化文本"
async def execute(self, **kwargs) -> dict:
return {"text": "模拟文档解析结果…"}

class WebSearch:
name = "web_search"
version = "v1"
description = "搜索互联网信息"
async def execute(self, **kwargs) -> dict:
return {"results": "模拟搜索结果…"}

ToolRegistry.register(DocumentParser, "v1")
ToolRegistry.register(WebSearch, "v1")

# 注册 Prompt 模板
PromptManager.register(PromptTemplate(
name="customer_service",
version="v1",
system_prompt="你是客服助手，请友好地回答用户问题。",
user_prompt_template="用户问题：{user_input}",
variables=["user_input"],
))
PromptManager.register(PromptTemplate(
name="contract_review",
version="v1",
system_prompt="你是法务审查助手，请检查合同条款的合规性。",
user_prompt_template="合同内容：{user_input}",
variables=["user_input"],
))

# 创建平台
platform = AgentPlatform()

# 部署两个业务 Agent（只需要配置，zero 代码复制）
service_agent = AgentConfig(
agent_id="customer-service-v1",
name="智能客服",
prompt_name="customer_service",
prompt_version="v1",
tools=[{"name": "web_search", "version": "v1"}],
llm_model="gpt-4o-mini",
owner="客服团队",
)

legal_agent = AgentConfig(
agent_id="contract-review-v1",
name="合同审查",
prompt_name="contract_review",
prompt_version="v1",
tools=[{"name": "document_parser", "version": "v1"}],
llm_model="gpt-4o",
owner="法务团队",
)

try:
await platform.deploy_agent(service_agent)
await platform.deploy_agent(legal_agent)

# 执行
result = await platform.execute("customer-service-v1", "我的订单怎么还没发货？")
logger.info(f"客服回复: {result['response'][:100]}")

result = await platform.execute("contract-review-v1", "请审查这份采购合同的付款条款")
logger.info(f"法务回复: {result['response'][:100]}")

# 查看平台状态
agents = platform.list_agents()
for a in agents:
logger.info(f"Agent: {a['agent_id']}, 调用: {a['call_count']}, tokens: {a['total_tokens']}")

except ValueError as e:
logger.error(f"部署失败: {e}")
except Exception as e:
logger.exception(f"未预期错误: {e}")

if __name__ == "__main__":
asyncio.run(main())

四、边界分析与架构权衡

平台化 vs 灵活性：平台化之后，Agent 开发者只能通过声明式配置来定制行为，不能像以前那样在代码里随意插 Hook。这对 80% 的简单 Agent（客服、FAQ、字段提取）来说是足够的，对 20% 的复杂 Agent（多步推理、自定义编排）来说不够。解决方案是提供"平台基类 + 自定义扩展点"——保留一个 on_pre_execute 和 on_post_execute 的钩子方法，允许业务方注入自己的逻辑。

Tool 注册中心的版本管理：Tool 升级版本后（比如 DocumentParser v1→v2），已有的 Agent 是继续用 v1 还是自动切到 v2？建议默认保持 v1，允许 Agent 配置中显式声明版本，同时平台侧提供灰度迁移工具——先切 10% 流量到 v2 观察一周，确认无异常再全量。

业务团队的技术能力差异：法务工程师写的 Prompt 和 AI 工程师写的 Prompt 质量差距巨大。平台侧应该提供 Prompt Playground 和评估工具，让非 AI 背景的工程师也能可视化地调试 Prompt 效果，而不是在 .yaml 文件里盲写。

成本控制的挑战：平台化之后，Token 消费变成了各业务团队独立的行为，平台层必须提供预算控制。最简单的做法是按 Agent 设置月度 Token 配额，达到 80% 时发预警、100% 时自动限流。这种"给你自由但给你预算"的模式比中央管控更容易被业务团队接受。

五、总结

从单体到平台的迁移，代码层面的工作量其实只占 30%，剩下 70% 是组织协调——说服业务团队接受共享能力而非各自造轮子、建立 Prompt 和 Tool 的治理规范、提供够好用的开发者工具降低使用门槛。技术上就一句话：把 Agent 从代码变成配置，让新建一个 Agent 的时间从天级降到分钟级。跑起来之后最大的感受是：终于不用在 4 个 repo 之间来回切改同一行 LLM 调用的参数了。

从单体 Agent 到 Agent 平台：一次架构迁移的技术方案和组织挑战