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(f"Tool {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())

模块化的局限

虽然模块化解决了代码复用问题，但还有：

1. 部署问题：每个 Agent 还是独立部署
2. 扩缩容问题：无法根据负载动态调度
3. 监控问题：每个 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_prompt=config.system_prompt,
            tools=tools,
            memory_config=config.memory_config,
            model_config=config.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_id="customer_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_workers=100)
        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(hours=24),
            "avg_latency": self.monitor.get_avg_latency(),
            "error_rate": self.monitor.get_error_rate(),
            "cost_24h": self.monitor.get_cost(hours=24)
        }

结论

Agent 平台化架构演进：

阶段一：定制化（避免）

• 问题：代码重复、维护困难
• 适用：快速验证（< 1 个月）

阶段二：模块化（推荐）

• 改进：代码复用
• 适用：2-5 个 Agent

阶段三：平台化（目标）

• 优点：快速构建、统一运维、工具共享
• 适用：> 5 个 Agent

平台化核心设计：

1. Agent = 配置 + 工具列表
2. 工具池（共享工具）
3. Agent 模板市场
4. 多 Agent 协作编排

实施路线：

• 第 1-2 月：模块化（抽取公共库）
• 第 3-4 月：设计平台架构
• 第 5-6 月：实现 MVP 平台
• 第 7-12 月：迭代优化

关键建议：

1. 不要过早平台化（YAGNI 原则）
2. 先有 3+ 个 Agent 再考虑平台化
3. 平台化的关键是"配置化"，而不是"代码化"