从零开始理解Skill：概念解析与实战应用指南

2次阅读

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

在技术语境中，Skill 特指封装特定领域能力的可复用组件。与普通函数或 API 不同，它强调完整的业务闭环处理能力——就像人类技能一样，给定输入就能独立完成从处理到反馈的全流程。

自包含性 ：内置领域知识库与处理逻辑
上下文感知 ：支持多轮交互和状态保持
容错能力 ：内置错误恢复和降级策略

特性	Skill	传统 API
交互模式	多轮 / 异步	单次请求 - 响应
状态管理	内置会话上下文	无状态
错误处理	自动重试 + 降级	依赖调用方处理
性能指标	端到端延迟	单次响应时间

安装基础依赖（Python≥3.8）：

pip install retrying>=2.0.0 aiohttp>=3.8.0

from typing import Any, Dict
import asyncio
from retrying import retry

class BaseSkill:
    def __init__(self, max_retries: int = 3):
        self._max_retries = max_retries

    @retry(stop_max_attempt_number=3, wait_fixed=1000)
    async def execute(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
        """异步执行入口"""
        try:
            return await self._process(input_data)
        except Exception as e:
            print(f"Execution failed: {str(e)}")
            raise

    async def _process(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
        raise NotImplementedError

并发控制

使用信号量限制并行请求数

from asyncio import Semaphore

class ConcurrentSkill(BaseSkill):
    def __init__(self, concurrency_limit=10):
        self.semaphore = Semaphore(concurrency_limit)

    async def execute(self, input_data):
        async with self.semaphore:
            return await super().execute(input_data)

状态管理
会话级：Redis 存储上下文
请求级：Python ContextVars

超时熔断

from async_timeout import timeout

async def execute_with_timeout(self, input_data, timeout_sec=5):
    try:
        async with timeout(timeout_sec):
            return await self.execute(input_data)
    except asyncio.TimeoutError:
        return {"error": "Request timeout"}

扩展要求：
– 实现城市参数校验（使用 pydantic）
– 添加 OpenWeatherMap API 调用
– 实现温度单位转换功能

启动模板：

from pydantic import BaseModel, validator

class WeatherRequest(BaseModel):
    city: str
    country_code: str = "us"

    @validator('city')
    def validate_city(cls, v):
        if not v.isalpha():
            raise ValueError("Invalid city name")
        return v.title()

class WeatherSkill(BaseSkill):
    async def _process(self, input_data: Dict) -> Dict:
        request = WeatherRequest(**input_data)
        # 在这里添加 API 调用逻辑
        return {"status": "TODO"}

使用 Locust 压测（100 并发）：