最佳实践

本文档总结了使用 agent_skills 协议开发技能时的最佳实践和推荐做法，帮助开发者构建高质量、可维护的技能系统。

技能设计最佳实践

1. 清晰的技能定义

推荐做法：

• 提供清晰、准确的技能名称和描述
• 详细说明输入输出参数
• 使用标准的 JSON Schema 定义数据结构
• 提供使用示例

# ✅ 好的示例
skill_definition = {
    "skill_id": "text-sentiment-analysis",
    "name": "文本情感分析",
    "description": "分析文本的情感倾向，返回正面、负面或中性的概率分布。支持中文和英文文本。",
    "input_schema": {
        "type": "object",
        "properties": {
            "text": {
                "type": "string",
                "description": "待分析的文本内容，长度建议在 1000 字符以内",
                "minLength": 1,
                "maxLength": 10000
            },
            "language": {
                "type": "string",
                "description": "文本语言代码，支持 'zh'（中文）和 'en'（英文）",
                "enum": ["zh", "en"],
                "default": "zh"
            }
        },
        "required": ["text"]
    },
    "output_schema": {
        "type": "object",
        "properties": {
            "sentiment": {
                "type": "string",
                "enum": ["positive", "negative", "neutral"],
                "description": "情感倾向"
            },
            "confidence": {
                "type": "number",
                "minimum": 0,
                "maximum": 1,
                "description": "置信度分数"
            }
        }
    }
}

# ❌ 不好的示例
bad_skill_definition = {
    "skill_id": "analyze",
    "name": "分析",
    "description": "分析文本",
    # 缺少详细的输入输出定义
}

2. 单一职责原则

推荐做法：

• 每个技能应该专注于单一功能
• 避免创建"万能"技能
• 将复杂功能拆分为多个简单技能

# ✅ 好的做法：拆分为多个专门技能
skills = [
    {"skill_id": "text-preprocessing", "name": "文本预处理"},
    {"skill_id": "text-sentiment-analysis", "name": "情感分析"},
    {"skill_id": "text-summarization", "name": "文本摘要"}
]

# ❌ 不好的做法：一个技能做所有事情
bad_skill = {
    "skill_id": "text-processor",
    "name": "文本处理器",
    # 包含预处理、情感分析、摘要等多个功能
}

3. 版本管理

推荐做法：

• 使用语义化版本号（Semantic Versioning）
• 提供版本迁移指南
• 保持向后兼容性
• 明确标记废弃版本

# 语义化版本：主版本号.次版本号.修订号
version = "1.2.3"  # 主版本：不兼容的 API 修改
                   # 次版本：向后兼容的功能新增
                   # 修订号：向后兼容的问题修复

# 版本兼容性声明
skill_definition = {
    "version": "2.0.0",
    "backward_compatible": False,  # 不向后兼容
    "migration_guide": "https://docs.example.com/migration-v2",
    "deprecated_versions": ["1.x.x"]
}

实现最佳实践

1. 输入验证

推荐做法：

• 始终验证输入参数
• 提供清晰的错误消息
• 处理边界情况

def process_skill(input_data: dict) -> dict:
    # ✅ 好的做法：完整的输入验证
    if "text" not in input_data:
        raise ValueError("缺少必需参数: text")
    
    text = input_data["text"]
    if not isinstance(text, str):
        raise TypeError("参数 'text' 必须是字符串类型")
    
    if len(text.strip()) == 0:
        raise ValueError("参数 'text' 不能为空")
    
    if len(text) > 10000:
        raise ValueError("参数 'text' 长度不能超过 10000 字符")
    
    # 处理逻辑
    return process(text)

2. 错误处理

推荐做法：

• 使用标准的错误代码
• 提供详细的错误信息
• 区分可恢复和不可恢复的错误

class SkillError(Exception):
    """技能基础错误类"""
    def __init__(self, code: str, message: str, details: dict = None):
        self.code = code
        self.message = message
        self.details = details or {}
        super().__init__(self.message)

class InvalidInputError(SkillError):
    """输入验证错误"""
    pass

class ProcessingError(SkillError):
    """处理过程错误"""
    pass

def process_skill(input_data: dict) -> dict:
    try:
        # 验证输入
        validate_input(input_data)
        
        # 处理逻辑
        result = do_processing(input_data)
        return {"success": True, "result": result}
    
    except InvalidInputError as e:
        # 可恢复的错误
        return {
            "success": False,
            "error": {
                "code": e.code,
                "message": e.message,
                "details": e.details
            }
        }
    except Exception as e:
        # 不可恢复的错误
        logger.error(f"技能处理失败: {e}", exc_info=True)
        return {
            "success": False,
            "error": {
                "code": "INTERNAL_ERROR",
                "message": "处理过程中发生内部错误"
            }
        }

3. 性能优化

推荐做法：

• 实现结果缓存
• 优化资源使用
• 使用异步处理长时间任务

from functools import lru_cache
import asyncio

class OptimizedSkill:
    def __init__(self):
        self.cache = {}
        self.cache_ttl = 3600  # 1小时
    
    @lru_cache(maxsize=1000)
    def process_simple(self, text: str) -> dict:
        """简单处理，使用 LRU 缓存"""
        return self._process(text)
    
    async def process_async(self, input_data: dict) -> dict:
        """异步处理长时间任务"""
        # 在后台线程中执行
        loop = asyncio.get_event_loop()
        result = await loop.run_in_executor(
            None,
            self._long_running_task,
            input_data
        )
        return result
    
    def _long_running_task(self, input_data: dict) -> dict:
        """长时间运行的任务"""
        # 实际处理逻辑
        pass

4. 日志记录

推荐做法：

• 记录关键操作和错误
• 使用结构化日志
• 避免记录敏感信息

import logging
import json

logger = logging.getLogger(__name__)

def process_skill(input_data: dict) -> dict:
    # 记录请求开始
    logger.info("技能调用开始", extra={
        "skill_id": "text-sentiment-analysis",
        "input_size": len(json.dumps(input_data))
    })
    
    try:
        result = do_processing(input_data)
        
        # 记录成功
        logger.info("技能调用成功", extra={
            "skill_id": "text-sentiment-analysis",
            "processing_time_ms": result.get("processing_time", 0)
        })
        
        return result
    
    except Exception as e:
        # 记录错误（不包含敏感信息）
        logger.error("技能调用失败", extra={
            "skill_id": "text-sentiment-analysis",
            "error_type": type(e).__name__,
            "error_message": str(e)
        }, exc_info=True)
        raise

安全最佳实践

1. 输入清理

推荐做法：

• 清理和验证所有输入
• 防止注入攻击
• 限制输入大小

import re
from html import escape

def sanitize_input(text: str, max_length: int = 10000) -> str:
    """清理输入文本"""
    # 限制长度
    if len(text) > max_length:
        raise ValueError(f"输入长度超过限制: {max_length}")
    
    # 移除潜在的恶意内容
    text = escape(text)  # HTML 转义
    
    # 移除控制字符（保留换行和制表符）
    text = re.sub(r'[\x00-\x08\x0B-\x0C\x0E-\x1F]', '', text)
    
    return text.strip()

2. 访问控制

推荐做法：

• 实施基于角色的访问控制
• 使用令牌认证
• 限制调用频率

from functools import wraps
from datetime import datetime, timedelta

class RateLimiter:
    def __init__(self, max_calls: int, period_seconds: int):
        self.max_calls = max_calls
        self.period = timedelta(seconds=period_seconds)
        self.calls = {}  # agent_id -> [timestamps]
    
    def check_rate_limit(self, agent_id: str) -> bool:
        """检查是否超过速率限制"""
        now = datetime.now()
        
        # 清理过期记录
        if agent_id in self.calls:
            self.calls[agent_id] = [
                ts for ts in self.calls[agent_id]
                if now - ts < self.period
            ]
        else:
            self.calls[agent_id] = []
        
        # 检查是否超过限制
        if len(self.calls[agent_id]) >= self.max_calls:
            return False
        
        # 记录本次调用
        self.calls[agent_id].append(now)
        return True

def require_authentication(func):
    """要求身份验证的装饰器"""
    @wraps(func)
    def wrapper(self, *args, **kwargs):
        if not self.is_authenticated():
            raise PermissionError("需要身份验证")
        return func(self, *args, **kwargs)
    return wrapper

3. 数据加密

推荐做法：

• 使用 HTTPS 传输
• 加密敏感数据
• 安全存储密钥

import hashlib
import hmac
import os

class SecureSkill:
    def __init__(self, secret_key: str):
        self.secret_key = secret_key.encode()
    
    def generate_token(self, data: dict) -> str:
        """生成数据签名令牌"""
        message = json.dumps(data, sort_keys=True).encode()
        signature = hmac.new(
            self.secret_key,
            message,
            hashlib.sha256
        ).hexdigest()
        return signature
    
    def verify_token(self, data: dict, token: str) -> bool:
        """验证数据签名令牌"""
        expected_token = self.generate_token(data)
        return hmac.compare_digest(expected_token, token)

测试最佳实践

1. 单元测试

推荐做法：

• 测试所有边界情况
• 使用模拟数据
• 保持测试独立性

import pytest
from unittest.mock import Mock, patch

class TestSentimentAnalysisSkill:
    def test_positive_sentiment(self):
        skill = SentimentAnalysisSkill()
        result = skill.process("这个产品很棒！", "zh")
        assert result["sentiment"] == "positive"
        assert result["confidence"] > 0.8
    
    def test_empty_text(self):
        skill = SentimentAnalysisSkill()
        with pytest.raises(ValueError, match="不能为空"):
            skill.process("", "zh")
    
    def test_invalid_language(self):
        skill = SentimentAnalysisSkill()
        # 应该使用默认语言或返回错误
        result = skill.process("Hello", "invalid")
        assert "sentiment" in result
    
    @patch('external_api.call')
    def test_external_api_failure(self, mock_api):
        mock_api.side_effect = Exception("API 错误")
        skill = SentimentAnalysisSkill()
        with pytest.raises(ProcessingError):
            skill.process("test", "zh")

2. 集成测试

推荐做法：

• 测试与注册中心的交互
• 测试技能组合工作流
• 使用测试环境

class TestSkillIntegration:
    def test_skill_registration(self):
        provider = SkillProvider(test_mode=True)
        provider.register_skill(test_skill_definition)
        
        registry = SkillRegistry(test_mode=True)
        skills = registry.discover_skills(category="test")
        
        assert len(skills) > 0
        assert any(s["skill_id"] == "test-skill" for s in skills)
    
    def test_skill_workflow(self):
        workflow = SkillWorkflow("test-workflow")
        workflow.add_step("skill-1", parameters={"input": "test"})
        workflow.add_step("skill-2", parameters={"input": "${steps.0.output}"})
        
        result = workflow.execute({"input": "initial"})
        assert result["success"] == True

部署最佳实践

1. 配置管理

推荐做法：

• 使用环境变量
• 分离开发和生产配置
• 使用配置管理工具

import os
from dataclasses import dataclass

@dataclass
class SkillConfig:
    registry_url: str
    api_key: str
    log_level: str
    cache_enabled: bool
    cache_ttl: int
    
    @classmethod
    def from_env(cls):
        return cls(
            registry_url=os.getenv("AGENT_SKILLS_REGISTRY_URL", "https://registry.agent-skills.org"),
            api_key=os.getenv("AGENT_SKILLS_API_KEY"),
            log_level=os.getenv("LOG_LEVEL", "INFO"),
            cache_enabled=os.getenv("CACHE_ENABLED", "true").lower() == "true",
            cache_ttl=int(os.getenv("CACHE_TTL", "3600"))
        )

2. 监控和告警

推荐做法：

• 监控关键指标
• 设置告警阈值
• 使用 APM 工具

from prometheus_client import Counter, Histogram, Gauge

# 定义指标
skill_calls_total = Counter('skill_calls_total', '技能调用总数', ['skill_id', 'status'])
skill_latency = Histogram('skill_latency_seconds', '技能调用延迟', ['skill_id'])
active_skills = Gauge('active_skills', '活跃技能数量')

def process_with_metrics(skill_id: str, func):
    """带指标记录的处理函数"""
    start_time = time.time()
    
    try:
        result = func()
        skill_calls_total.labels(skill_id=skill_id, status='success').inc()
        return result
    except Exception as e:
        skill_calls_total.labels(skill_id=skill_id, status='error').inc()
        raise
    finally:
        latency = time.time() - start_time
        skill_latency.labels(skill_id=skill_id).observe(latency)

文档最佳实践

1. API 文档

推荐做法：

• 提供完整的 API 文档
• 包含使用示例
• 说明错误情况

2. 代码注释

推荐做法：

• 为复杂逻辑添加注释
• 使用类型提示
• 保持注释与代码同步

def process_sentiment(text: str, language: str = "zh") -> dict:
    """
    分析文本的情感倾向
    
    Args:
        text: 待分析的文本内容，长度应在 1-10000 字符之间
        language: 文本语言代码，支持 'zh'（中文）和 'en'（英文）
    
    Returns:
        包含以下字段的字典：
        - sentiment: 情感倾向（"positive", "negative", "neutral"）
        - confidence: 置信度分数（0-1）
        - probabilities: 各情感的概率分布
    
    Raises:
        ValueError: 当输入文本为空或长度超限时
        ProcessingError: 当处理过程中发生错误时
    
    Example:
        >>> result = process_sentiment("这个产品很棒！", "zh")
        >>> print(result["sentiment"])
        'positive'
    """
    # 实现逻辑
    pass

总结

遵循这些最佳实践可以帮助你：

• ✅ 构建高质量、可维护的技能
• ✅ 提高系统性能和可靠性
• ✅ 确保安全性和合规性
• ✅ 改善开发体验和协作效率

记住，最佳实践是指导原则，应根据具体场景灵活应用。