Agent Skill详解：从核心原理到生产环境最佳实践

9次阅读

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

在智能 Agent 开发中，技能（Skill）是完成特定任务的基本单元。随着业务复杂度提升，开发者常面临以下问题：

技能复用困难：相似功能重复开发，缺乏标准化接口
上下文隔离不足：多个技能共享变量导致意外覆盖
并发控制复杂：资源竞争和死锁问题频发
监控调试不便：缺乏统一的执行追踪机制

输入契约：明确定义参数类型、取值范围和必填项
处理逻辑：核心业务实现，需保持无状态设计
输出规范：统一返回结构包含状态码、错误信息和业务数据

采用插件化架构实现，主要功能：

维护技能元信息（名称、版本、依赖）
处理技能发现与生命周期管理
提供执行时的上下文隔离环境

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

class SkillExecutionError(Exception):
    """自定义技能执行异常"""
    pass

class BaseSkill(ABC):
    def __init__(self, skill_name: str):
        self.name = skill_name
        self._timeout = 5  # 默认超时(秒)

    @property
    def timeout(self) -> int:
        return self._timeout

    @abstractmethod
    async def execute(self, **kwargs) -> Dict[str, Any]:
        """
        执行入口
        :param kwargs: 动态输入参数
        :return: 必须包含{'status': int, 'data': Any}
        """
        raise NotImplementedError

    def _validate_input(self, **kwargs):
        """参数校验模板方法"""
        # 实现具体校验逻辑
        pass

_skill_registry = {}

def register_skill(name: str, desc: str = ''):
    def decorator(cls):
        if name in _skill_registry:
            raise ValueError(f'Skill {name} already registered')

        cls.metadata = {'name': name, 'desc': desc}
        _skill_registry[name] = cls
        return cls
    return decorator

@register_skill(name='weather_query', desc='查询实时天气')
class WeatherSkill(BaseSkill):
    async def execute(self, city: str) -> dict:
        # 具体实现...
        return {'status': 200, 'data': {...}}

类型	适用场景	优缺点
同步	CPU 密集型计算简单 IO 操作	实现简单但会阻塞事件循环
异步	高并发 IO 需要协程协作	资源利用率高需要显式 await 管理

分层超时设置：
全局默认超时（如 5 秒）
技能级自定义超时
特殊操作独立配置（如文件上传）

智能重试策略：

class RetryPolicy:
    def __init__(self, 
                 max_retries: int = 3,
                 backoff_factor: float = 1.0):
        self.max_retries = max_retries
        self.backoff = backoff_factor

    async def execute_with_retry(self, skill, **kwargs):
        for attempt in range(self.max_retries):
            try:
                return await skill.execute(**kwargs)
            except TimeoutError:
                if attempt == self.max_retries - 1:
                    raise
                await asyncio.sleep(self.backoff * (attempt + 1))

显式声明：在技能元数据中定义 requires 字段
动态注入：通过 DI 容器管理依赖实例
循环检测：启动时验证依赖无环图

# 在基类中添加监控逻辑
async def execute(self, **kwargs):
    start_time = time.monotonic()
    try:
        result = await self._execute_impl(**kwargs)
        emit_metric('skill_success', tags={'name': self.name})
        return result
    except Exception as e:
        emit_metric('skill_failed', tags={'name': self.name, 'error': type(e).__name__})
        raise
    finally:
        latency = time.monotonic() - start_time
        emit_metric('skill_latency', value=latency)