Agent 平台化思考:从定制化开发到通用 Agent 平台的架构演进
Agent 平台化思考从定制化开发到通用 Agent 平台的架构演进一、从一个 Agent 一个项目到Agent 平台工程化必经之路2026 年初某 SaaS 公司面临一个困境过去一年他们为不同客户需求开发了 15 个独立的 Agent 项目。这些项目技术栈各异、代码重复率高达 60%、维护成本巨大。更糟糕的是当某个客户需要一个新功能其实是其他 Agent 已有的功能时他们需要重新开发无法复用。这不是个案。随着 Agent 技术从实验走向生产平台化成为必然趋势。本文将讨论 Agent 平台化的架构演进路径。二、阶段一定制化开发反模式典型问题场景客户 A 需要一个客服 Agent客户 B 需要一个销售 Agent。反模式实现# 项目 1: 客服 Agent独立代码库 # customer_service_agent/ # ├── main.py # ├── tools/ # │ ├── query_order.py # │ ├── cancel_order.py # │ └── search_kb.py # └── prompts/ # └── system_prompt.txt # 项目 2: 销售 Agent另一个独立代码库 # sales_agent/ # ├── main.py # 跟客服 Agent 的 main.py 有 70% 相同 # ├── tools/ # │ ├── search_products.py # 新工具 # │ ├── query_order.py # 重复跟客服 Agent 一样 # │ └── create_order.py # └── prompts/ # └── system_prompt.txt # 问题 # 1. query_order 工具重复实现维护困难 # 2. main.py 逻辑重复代码冗余 # 3. 两个 Agent 无法共享上下文用户先从客服问问题再找销售上下文丢失问题清单三、阶段二模块化改进核心思路抽取公共库# 公共库agent_common/ # ├── core/ # │ ├── agent.py # Agent 核心逻辑 # │ ├── tool_registry.py # 工具注册中心 # │ └── memory.py # 记忆管理 # ├── tools/ # │ ├── query_order.py # 公共工具 # │ ├── search_kb.py # │ └── base_tool.py # 工具基类 # └── utils/ # ├── llm_client.py # └── prompt_template.py # 客服 Agent 项目简化 from agent_common.core import Agent, ToolRegistry from agent_common.tools import QueryOrderTool, SearchKBTool class CustomerServiceAgent(Agent): def __init__(self): super().__init__() # 注册工具 self.registry.register(QueryOrderTool()) self.registry.register(SearchKBTool()) # 添加客服特有的工具 self.registry.register(CancelOrderTool()) # 销售 Agent 项目简化 from agent_common.core import Agent, ToolRegistry from agent_common.tools import QueryOrderTool, SearchKBTool # 复用 class SalesAgent(Agent): def __init__(self): super().__init__() # 复用公共工具 self.registry.register(QueryOrderTool()) # 直接复用 self.registry.register(SearchKBTool()) # 添加销售特有的工具 self.registry.register(SearchProductsTool()) self.registry.register(CreateOrderTool())生产级实现工具注册中心from abc import ABC, abstractmethod from typing import Dict, List, Any import importlib import json class BaseTool(ABC): 工具基类 abstractmethod def get_name(self) - str: pass abstractmethod def get_description(self) - str: pass abstractmethod def get_parameters(self) - Dict: pass abstractmethod def execute(self, **params) - Any: pass def to_function_calling_schema(self) - Dict: 转换为 Function Calling 格式 return { type: function, function: { name: self.get_name(), description: self.get_description(), parameters: self.get_parameters() } } class ToolRegistry: 工具注册中心单例 _instance None _tools: Dict[str, BaseTool] {} def __new__(cls): if cls._instance is None: cls._instance super().__new__(cls) return cls._instance def register(self, tool: BaseTool): 注册工具 name tool.get_name() if name in self._tools: raise ValueError(fTool {name} already registered) self._tools[name] tool def get_tool(self, name: str) - BaseTool: 获取工具 return self._tools.get(name) def list_tools(self) - List[Dict]: 列出所有工具用于 LLM Function Calling return [tool.to_function_calling_schema() for tool in self._tools.values()] def load_tools_from_config(self, config_path: str): 从配置文件加载工具 with open(config_path, r) as f: config json.load(f) for tool_config in config[tools]: module_path tool_config[module] class_name tool_config[class] # 动态加载 module importlib.import_module(module_path) tool_class getattr(module, class_name) tool_instance tool_class() self.register(tool_instance) # 示例工具实现 class QueryOrderTool(BaseTool): 查询订单工具 def get_name(self) - str: return query_order def get_description(self) - str: return 根据用户 ID 或订单 ID 查询订单信息 def get_parameters(self) - Dict: return { type: object, properties: { user_id: {type: string, description: 用户 ID}, order_id: {type: string, description: 订单 ID} }, required: [] } def execute(self, **params) - Dict: user_id params.get(user_id) order_id params.get(order_id) # 查询数据库简化 if order_id: return {order_id: order_id, status: shipped, amount: 99.9} elif user_id: return {user_id: user_id, orders: [{id: 123, status: shipped}]} else: raise ValueError(user_id or order_id required) # 使用 registry ToolRegistry() registry.register(QueryOrderTool())模块化的局限虽然模块化解决了代码复用问题但还有部署问题每个 Agent 还是独立部署扩缩容问题无法根据负载动态调度监控问题每个 Agent 独立监控缺乏全局视图四、阶段三平台化目标平台化架构核心设计Agent 配置 工具列表关键洞察Agent 的本质是系统 prompt 工具列表 记忆管理策略。from pydantic import BaseModel from typing import List, Optional class AgentConfig(BaseModel): Agent 配置可序列化 agent_id: str name: str description: str # 系统 Prompt system_prompt: str # 工具列表引用工具池中的工具 tools: List[str] # tool names # 记忆策略 memory_config: Dict # 模型配置 model_config: Dict # 其他配置 max_iterations: int 10 timeout: int 60 class AgentPlatform: Agent 平台 def __init__(self): self.tool_registry ToolRegistry() self.agent_configs: Dict[str, AgentConfig] {} self.agent_instances: Dict[str, Agent] {} def register_agent(self, config: AgentConfig): 注册 Agent只需配置 self.agent_configs[config.agent_id] config def get_agent(self, agent_id: str) - Agent: 获取 Agent 实例懒加载 if agent_id not in self.agent_instances: config self.agent_configs[agent_id] # 根据配置创建 Agent agent self._create_agent_from_config(config) self.agent_instances[agent_id] agent return self.agent_instances[agent_id] def _create_agent_from_config(self, config: AgentConfig) - Agent: 根据配置创建 Agent # 1. 加载工具 tools [] for tool_name in config.tools: tool self.tool_registry.get_tool(tool_name) if tool: tools.append(tool) # 2. 创建 Agent agent Agent( system_promptconfig.system_prompt, toolstools, memory_configconfig.memory_config, model_configconfig.model_config ) return agent async def run_agent(self, agent_id: str, user_input: str, context: Dict) - str: 运行 Agent agent self.get_agent(agent_id) return await agent.run(user_input, context) # 使用示例只需配置无需写代码 def create_customer_service_agent_via_config(): 通过配置创建客服 Agent config AgentConfig( agent_idcustomer_service, name智能客服, description处理用户咨询、订单查询、退款, system_prompt你是客服助手友好、专业..., tools[ query_order, cancel_order, search_kb ], memory_config{ type: buffer, max_turns: 10 }, model_config{ model: gpt-4, temperature: 0.7 } ) # 注册到平台 platform AgentPlatform() platform.register_agent(config) return platform # 现在新 Agent 只需写配置文件 # sales_agent_config.json: { agent_id: sales, name: 销售助手, tools: [query_order, search_products, create_order], system_prompt: ... } 平台化核心功能class AgentPlatformAdvanced: 高级 Agent 平台生产级 def __init__(self): self.tool_registry ToolRegistry() self.agent_configs {} self.executor_pool AgentExecutorPool(max_workers100) self.monitor PlatformMonitor() # 功能 1: Agent 模板市场 def list_agent_templates(self) - List[Dict]: 列出 Agent 模板 templates [ { id: customer_service, name: 智能客服, description: 适用于电商客服场景, tools: [query_order, cancel_order, search_kb], preview: https://... }, # ... ] return templates def create_agent_from_template(self, template_id: str, customizations: Dict) - str: 从模板创建 Agent # 复制模板配置 template self._get_template(template_id) config AgentConfig(**template) # 应用自定义 for key, value in customizations.items(): setattr(config, key, value) # 生成唯一 ID config.agent_id f{template_id}_{uuid.uuid4().hex[:8]} # 注册 self.register_agent(config) return config.agent_id # 功能 2: 工具市场 def list_tool_market(self) - List[Dict]: 列出可用工具 return [ { name: query_order, description: 查询订单, usage_count: 1000, rating: 4.5 }, # ... ] # 功能 3: 多 Agent 协作 async def run_multi_agent(self, workflow: Dict, user_input: str) - str: 运行多 Agent 工作流 # workflow 示例 # { # steps: [ # {agent: customer_service, input: {{user_input}}}, # {agent: sales, input: {{previous_output}}}, # ] # } context {user_input: user_input} for step in workflow[steps]: agent_id step[agent] step_input self._render_template(step[input], context) agent self.get_agent(agent_id) output await agent.run(step_input, context) context[f{agent_id}_output] output return context[workflow[steps][-1][agent] _output] # 功能 4: 监控和分析 def get_platform_metrics(self) - Dict: 获取平台指标 return { total_agents: len(self.agent_configs), active_agents: len(self.agent_instances), total_calls_24h: self.monitor.get_total_calls(hours24), avg_latency: self.monitor.get_avg_latency(), error_rate: self.monitor.get_error_rate(), cost_24h: self.monitor.get_cost(hours24) }结论Agent 平台化架构演进阶段一定制化避免问题代码重复、维护困难适用快速验证 1 个月阶段二模块化推荐改进代码复用适用2-5 个 Agent阶段三平台化目标优点快速构建、统一运维、工具共享适用 5 个 Agent平台化核心设计Agent 配置 工具列表工具池共享工具Agent 模板市场多 Agent 协作编排实施路线第 1-2 月模块化抽取公共库第 3-4 月设计平台架构第 5-6 月实现 MVP 平台第 7-12 月迭代优化关键建议不要过早平台化YAGNI 原则先有 3 个 Agent 再考虑平台化平台化的关键是配置化而不是代码化