Agent Skills开发实战：从零构建高可用的智能体技能系统

1次阅读

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

在维护过三个基于对话机器人的 Agent 系统后，我总结出这些典型问题：

技能复用率低：每个业务线重复开发相似技能（如地址解析、时间转换），代码重复率高达 60%
更新成本高：修改一个技能参数需要全量发布，凌晨上线成了家常便饭
性能不稳定：促销期间客服机器人响应时间从 200ms 飙升到 2s+，日志里满是线程阻塞警告

最让我头疼的是去年双十一，因为天气查询技能调用的第三方 API 超时，连带导致支付状态查询技能被阻塞，最终触发了级联故障。这促使我开始探索模块化的技能管理系统。

经过多次迭代，我们形成了如图所示的架构（注：此处描述架构图）：

+-------------------+     +-------------------+     +-------------------+
|   技能注册中心     |     |   技能执行引擎    |     |   上下文管理器    |
| (含版本控制)      |<--->| (含熔断机制)     |<--->| (请求级隔离)     |
+-------------------+     +-------------------+     +-------------------+
        ^                         ^                         ^
        |                         |                         |
+-------------------+     +-------------------+     +-------------------+
|   技能仓库        |     |   监控告警系统    |     |   技能配置中心    |
| (Git/S3 存储)      |     | (Prometheus 埋点) |     | (动态参数热更新)  |
+-------------------+     +-------------------+     +-------------------+

关键设计原则：

标准化接口 ：所有技能必须实现execute(input: Dict, ctx: Context) -> Dict 方法
松耦合通信：通过事件总线传递技能执行结果，避免直接调用
分级隔离：CPU 密集型技能（如图像处理）与 IO 密集型技能（如 API 调用）使用不同线程池

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

class SkillBase(ABC):
    @classmethod
    @abstractmethod
    def meta_info(cls) -> Dict[str, Any]:
        """返回技能元信息（名称、版本、输入输出格式说明）"""
        pass

    @abstractmethod
    async def execute(self, inputs: Dict[str, Any], ctx: 'Context') -> Dict[str, Any]:
        """
        执行技能的核心方法
        :param inputs: 结构化输入参数
        :param ctx: 请求上下文（含用户会话、授权信息等）:return: 标准化输出（必须包含 result_code 字段）"""
        pass

import importlib
from pathlib import Path

class SkillLoader:
    @staticmethod
    def load_from_path(skill_path: str) -> SkillBase:
        """
        从指定路径加载技能类
        :param skill_path: 格式为 "module.submodule:ClassName"
        :return: 技能实例
        """module_path, class_name = skill_path.split(':')
        module = importlib.import_module(module_path)
        skill_class = getattr(module, class_name)
        return skill_class()

    @staticmethod
    def watch_skill_dir(dir_path: str):
        """监控技能目录变化，自动重新加载"""
        # 实际实现需结合 watchdog 等库
        pass

from contextvars import ContextVar
import uuid

request_id = ContextVar('request_id', default='')

class Context:
    def __init__(self):
        self._request_id = str(uuid.uuid4())
        request_id.set(self._request_id)

    @property
    def trace_id(self) -> str:
        """获取全链路追踪 ID"""
        return request_id.get()

在日均调用量 300 万次的客服系统中，我们通过以下优化将 P99 响应时间从 870ms 降到 210ms：

线程池分级配置
IO 密集型：最大线程数 =CPU 核心数 *8，队列长度 =200
CPU 密集型：最大线程数 =CPU 核心数 +1，队列长度 =0（直接拒绝）

技能预热机制

# 系统启动时预加载高频技能
async def warmup_skills():
    hot_skills = ['weather_query', 'payment_status']
    for skill in hot_skills:
        await skill.execute({}, empty_ctx)  # 空参数初始化

结果缓存设计
短期缓存：使用请求级内存缓存（如天气数据缓存 5 分钟）
长期缓存：Redis 存储带版本号的技能输出（适合商品信息等）

现象：两个技能都注册了/parse 路由导致随机响应
解决方案：
– 命名空间隔离：强制技能 ID 包含开发者前缀（如alipay.payment）
– 启动时检查路由冲突

症状：运行 24 小时后 OOM 崩溃，MAT 分析显示 Context 对象堆积
根因：技能中异步回调持有 ctx 引用
修复方案：

# 错误写法（会泄漏）：async def leak_example(ctx):
    loop = asyncio.get_event_loop()
    loop.call_later(10, lambda: print(ctx.user_id))  # 闭包捕获 ctx

# 正确写法：async def safe_example(ctx):
    user_id = ctx.user_id  # 提前提取基本类型
    loop.call_later(10, lambda: print(user_id))

现有系统已稳定运行 9 个月，但还有更多可能性：