从单体 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(levellogging.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(fTool 注册: {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(fTool {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_factorylist) metadata: dict Field(default_factorydict) 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(fPrompt 变量缺失: {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(fPrompt {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_factorylist) # [{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(fPrompt 加载失败 {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(fTool 加载失败 {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(fAgent [{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(fAgent [{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(fAgent {config.agent_id} 已存在) runtime AgentRuntime(config) await runtime.initialize() self._agents[config.agent_id] runtime logger.info(fAgent [{config.agent_id}] 已部署) return runtime async def stop_agent(self, agent_id: str): if agent_id not in self._agents: raise ValueError(fAgent {agent_id} 不存在) stats self._agents[agent_id].get_stats() del self._agents[agent_id] logger.info(fAgent [{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(fAgent {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( namecustomer_service, versionv1, system_prompt你是客服助手请友好地回答用户问题。, user_prompt_template用户问题{user_input}, variables[user_input], )) PromptManager.register(PromptTemplate( namecontract_review, versionv1, system_prompt你是法务审查助手请检查合同条款的合规性。, user_prompt_template合同内容{user_input}, variables[user_input], )) # 创建平台 platform AgentPlatform() # 部署两个业务 Agent只需要配置zero 代码复制 service_agent AgentConfig( agent_idcustomer-service-v1, name智能客服, prompt_namecustomer_service, prompt_versionv1, tools[{name: web_search, version: v1}], llm_modelgpt-4o-mini, owner客服团队, ) legal_agent AgentConfig( agent_idcontract-review-v1, name合同审查, prompt_namecontract_review, prompt_versionv1, tools[{name: document_parser, version: v1}], llm_modelgpt-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(fAgent: {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 升级版本后比如DocumentParserv1→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 调用的参数了。