Agent与Skill架构设计实战：从零构建高效AI协作系统

14次阅读

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

去年我们团队接了一个智能客服升级项目。当尝试把地址识别、情绪分析等 5 个新功能塞进原有 Agent 时，每次调用延迟从 200ms 飙升到 1.2 秒——这就是典型的单体架构瓶颈。更糟的是，修改情绪分析模块时意外影响了工单分类的准确性，这种耦合让迭代变得束手束脚。

我们测试了三种主流方案在 8 核 16G 云主机上的表现（模拟 100 并发）：

gRPC 方案
QPS：3200
错误率：0.3%
痛点：需要预生成桩代码，动态扩展困难
Redis 发布订阅
QPS：5800
错误率：1.2%
优势：无需服务发现，但无堆积能力
RabbitMQ（最终选择）
QPS：5100
错误率：0.8%
决胜点：自带流量削峰和死信队列

class SkillRegistry:
    def __init__(self, mq_conn: pika.BlockingConnection):
        self._channel = mq_conn.channel()
        self._channel.exchange_declare(exchange='skill_registry', exchange_type='topic')

    def register(self, skill_name: str, weight: float=1.0):
        """
        前置条件：AMQP 连接已建立
        后置条件：创建持久化队列并绑定路由键
        """
        queue = self._channel.queue_declare(queue=f'skill_{skill_name}', 
            durable=True,  # 持久化防止重启丢失
            arguments={'x-max-priority': 10}  # 支持优先级
        )
        self._channel.queue_bind(
            exchange='skill_registry',
            queue=queue.method.queue,
            routing_key=skill_name
        )
        self._update_load_balancer(skill_name, weight)

def weighted_round_robin(skills: Dict[str, float]):
    """:param skills: {'ocr': 2.5,'nlu': 1.0} 
    :return: 按权重比例返回技能名
    """
    total = sum(skills.values())
    pick = random.uniform(0, total)
    cumulative = 0
    for skill, weight in skills.items():
        cumulative += weight
        if pick <= cumulative:
            return skill
    return list(skills.keys())[0]  # 兜底逻辑

格式	吞吐量（msg/s）	CPU 占用
JSON	12,000	35%
Protobuf	28,000	18%
MessagePack	19,000	22%

启动时加载轻量级模型
后台线程预跑 10 个典型请求
达到 80% 准确率才标记为就绪

class SkillWrapper:
    def warm_up(self):
        while self._accuracy < 0.8:
            self._run_sample_requests()
            time.sleep(1)

在订单处理等场景必须这样处理：

def handle_message(msg_id: str, content: str):
    if redis.get(f'processed_{msg_id}'):  # 幂等检查
        return

    try:
        process(content)
        redis.setex(f'processed_{msg_id}', 3600, '1')  # 1 小时防重
    except Exception as e:
        if not _should_retry(e):
            send_to_dlq(msg_id)  # 进入死信队列