Agentscope实战：基于Skill调用工具的实现原理与最佳实践

1次阅读

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

在复杂系统中，工具调用常常面临几个典型问题：

接口不一致：不同工具由不同团队开发，调用方式各异，缺乏统一规范
生命周期管理困难：工具实例的初始化、销毁缺乏统一管理，容易造成资源泄漏
依赖混乱：工具之间的依赖关系难以清晰表达，容易出现循环依赖
上下文丢失：调用链中上下文信息（如用户身份、请求 ID）难以传递

Agentscope 的 Skill 机制通过标准化工具定义、集中式管理和自动依赖注入，有效解决了这些问题。Skill 作为工具的抽象层，提供了统一的调用接口和生命周期管理。

定义阶段 ：开发者使用@skill 装饰器定义 Skill
注册阶段：框架扫描并收集所有被装饰的 Skill 类
初始化阶段：根据依赖关系图按顺序实例化 Skill
调用阶段：通过统一的入口方法调用具体功能

# Skill 元数据定义示例
@skill(name="weather_query", 
       desc="查询天气信息",
       version="1.0")
class WeatherQuerySkill:
    # 具体实现...

Skill 可以声明对其他 Skill 或工具的依赖，框架会自动解析并注入：

@skill(name="weather_alert")
class WeatherAlertSkill:
    def __init__(self, weather_query: WeatherQuerySkill):
        self.weather = weather_query  # 自动注入

调用链中的上下文通过 Context 对象传递，包含请求级别的信息：

def query_weather(self, city: str, ctx: Context) -> WeatherInfo:
    logger.info(f"[{ctx.request_id}] 查询 {city} 天气")
    # 实际业务逻辑...

下面是一个完整的 Skill 实现示例：

from agentscope.core import skill, Context
from pydantic import BaseModel, validator

class WeatherRequest(BaseModel):
    city: str
    unit: str = "celsius"

    @validator('city')
    def city_not_empty(cls, v):
        if not v.strip():
            raise ValueError("城市不能为空")
        return v

@skill(name="weather", version="1.1")
class WeatherSkill:
    def __init__(self, cache_tool: CacheTool):
        self.cache = cache_tool

    def query(self, req: WeatherRequest, ctx: Context) -> dict:
        try:
            cache_key = f"weather_{req.city}"
            if cached := self.cache.get(cache_key):
                return cached

            # 调用实际天气 API
            result = call_weather_api(req.city, req.unit)
            self.cache.set(cache_key, result, ttl=3600)
            return result
        except Exception as e:
            logger.error(f"[{ctx.request_id}] 查询失败: {e}")
            raise WeatherQueryError("天气查询服务暂不可用")

对于批量调用场景，建议配置线程池：

# config.yaml
skill_executor:
  thread_pool:
    core_size: 10
    max_size: 50
    queue_capacity: 1000

输入校验：使用 Pydantic 模型严格校验输入参数
权限控制：在 Skill 入口方法检查上下文中的用户权限
速率限制：对高频接口添加限流措施

循环依赖 ：使用agentscope check-deps 命令检测依赖循环
版本冲突：明确指定 Skill 版本 `@skill(version=”1.0″)
资源泄漏 ：实现close() 方法释放资源

当 Skill 需要访问不稳定外部服务时，建议：

重试策略：指数退避重试（如 0.1s, 0.2s, 0.4s…）
降级方案：返回缓存数据或简化版响应
熔断机制：当错误率超过阈值时暂时停止调用

# 带重试的调用示例
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), 
       wait=wait_exponential(multiplier=0.1))
def call_external_api(self, params):
    # 实际调用逻辑...

通过本文介绍的方法，开发者可以构建出健壮、高效的工具调用体系。建议在实际项目中先从核心 Skill 开始实践，逐步扩展复杂功能。

正文完