Agent Skill实现：从零构建高可扩展的技能调度系统

11次阅读

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

在传统 Agent 开发中，技能实现常遇到三个典型问题：

硬编码严重 ：新技能上线需要修改核心调度代码，每次变更都伴随系统重启风险
扩展性差 ：技能间存在隐式依赖，添加新功能时容易引发连锁反应
并发冲突 ：共享状态管理导致技能执行结果不可预测，尤其在多线程环境下

这些痛点使得系统难以应对业务快速迭代的需求。比如某客服 Agent 系统，当需要新增『工单转接』技能时，不得不停服更新调度逻辑。

采用发布 - 订阅机制实现技能解耦：

技能提供者通过装饰器注册到事件总线
事件生产者触发事件时不关心具体执行者
总线负责匹配事件类型与技能订阅关系

class EventBus:
    def __init__(self):
        self._handlers = defaultdict(list)

    def subscribe(self, event_type, handler):
        self._handlers[event_type].append(handler)

为关键技能设置执行优先级（0- 9 级），总线根据优先级排序执行：

紧急事件（如系统告警）设为最高优先级 9
常规业务技能（如查询操作）设为 5
后台任务（如日志清理）设为 1

每个技能执行时创建独立沙箱环境：

通过 ThreadLocal 存储技能运行时数据
自动注入请求级上下文（如用户会话 ID）
强制技能声明输入 / 输出契约

def skill(event_type, priority=5):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 创建隔离上下文
            ctx = SkillContext()
            try:
                return func(ctx, *args, **kwargs)
            finally:
                ctx.cleanup()

        # 注册到事件总线
        EventBus().subscribe(event_type, (priority, wrapper))
        return wrapper
    return decorator

@skill('user_query', priority=6)
def handle_user_query(ctx, query):
    # 获取依赖服务实例（通过上下文注入）db = ctx.get_service('database')
    return db.execute(query)

将同步执行改为 async/await 模式后，在 4 核机器上测试结果：

模式	QPS	平均延迟
同步	1,200	83ms
异步	8,700	11ms

采用预加载策略：

系统启动时加载基础技能
按需动态加载非核心技能
维护常驻技能缓存池

为每个技能分配唯一技能 ID
执行前检查是否已处理相同请求 ID
实现自动重试时的去重逻辑

class SkillContext:
    def __init__(self):
        self._resources = []

    def cleanup(self):
        for res in self._resources:
            res.release()