智能体(Agent)与技能(Skill)架构设计：从解耦到高效协同的工程实践

6次阅读

共计 1910 个字符，预计需要花费 5 分钟才能阅读完成。

在传统紧耦合的 AI 系统中，开发者常遇到以下典型问题：

技能扩展困难：每新增一个技能都需要修改 Agent 核心代码，导致迭代速度缓慢。例如某对话系统添加天气查询功能时，需要重新部署整个服务。
版本兼容噩梦：技能升级时可能破坏现有调用链。我们曾遇到 NLP 技能升级导致 3 个依赖服务异常的情况。
资源竞争严重：多个技能共享内存时，一个 OCR 技能的内存泄漏可能拖垮整个系统。

采用 RabbitMQ 实现发布 / 订阅模式：

# 技能注册示例
channel.exchange_declare(exchange='skill_events', exchange_type='topic')
channel.queue_bind(queue='vision_q', 
                   exchange='skill_events', 
                   routing_key='vision.*')

使用 Protocol Buffers 定义通用接口：

message SkillRequest {
  string skill_id = 1;
  bytes input_data = 2;
  map<string, string> params = 3;
}

message SkillResponse {
  enum Status {
    SUCCESS = 0;
    TIMEOUT = 1;
    FORBIDDEN = 2;
  }
  Status status = 1;
  bytes output = 2;
}

技能启动时向注册中心发送心跳
Agent 通过 ZooKeeper 监听技能节点变化
路由表自动更新（实测 500 技能注册耗时 <200ms）

from abc import ABC, abstractmethod
from typing import Any, Dict

class BaseSkill(ABC):
    @property
    @abstractmethod
    def version(self) -> str:
        pass

    @abstractmethod
    async def execute(self, 
                     request: Dict[str, Any]) -> Dict[str, Any]:
        """
        :param request: 标准化输入字典
        :return: 必须包含 status 和 data 字段
        """
        pass

    async def shutdown(self):
        """清理资源"""
        pass

class ImageProcessingSkill(BaseSkill):
    def __init__(self):
        self._executor = ThreadPoolExecutor(max_workers=4)

    async def execute(self, request):
        loop = asyncio.get_event_loop()
        # 将 CPU 密集型任务转移到线程池
        result = await loop.run_in_executor(
            self._executor,
            self._process_image,
            request['image_data']
        )
        return {'status': 'success', 'data': result}

from circuitbreaker import circuit

@circuit(failure_threshold=5, 
         recovery_timeout=60)
async def call_skill(skill_id: str, request):
    try:
        return await asyncio.wait_for(skills[skill_id].execute(request),
            timeout=settings.SKILL_TIMEOUT
        )
    except asyncio.TimeoutError:
        raise SkillTimeoutError