Agent Skill开发入门指南：从零构建你的第一个智能体技能

11次阅读

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

刚接触 Agent Skill 开发时，我遇到过这些问题：

技能在长时间运行后神秘失踪
用户会话状态莫名其妙被重置
异常发生时连错误日志都找不到

这些问题的根源往往在于：

没有处理好技能的生命周期
忽略了状态持久化的需求
错误处理机制不健全

from typing import Callable
from functools import wraps

class SkillInterface:
    def __init__(self):
        self._handlers = {}

    def register(self, intent: str):
        def decorator(handler: Callable):
            @wraps(handler)
            async def wrapper(*args, **kwargs):
                try:
                    return await handler(*args, **kwargs)
                except Exception as e:
                    print(f"Intent {intent} failed: {str(e)}")
                    raise

            self._handlers[intent] = wrapper
            return wrapper
        return decorator

class WeatherSkill:
    def __init__(self, db_client):
        self.db = db_client  # 数据层依赖注入

    async def get_weather(self, city: str) -> dict:
        # 业务逻辑实现
        cached = await self.db.get(f"weather_{city}")
        if cached:
            return cached

        # 模拟 API 调用
        result = {"city": city, "temp": "25°C"}  # 实际应调用天气 API
        await self.db.set(f"weather_{city}", result, ttl=3600)
        return result

import redis
from redis.asyncio import Redis

class StateManager:
    def __init__(self, redis_url: str):
        self.redis: Redis = redis.from_url(redis_url)

    async def save_session(self, session_id: str, data: dict):
        await self.redis.hset(f"session:{session_id}",
            mapping=data
        )

def require_permission(permission: str):
    def decorator(func):
        @wraps(func)
        async def wrapper(user: dict, *args, **kwargs):
            if permission not in user.get("permissions", []):
                raise PermissionError(f"Missing {permission} permission")
            return await func(user, *args, **kwargs)
        return wrapper
    return decorator

# 使用示例
@require_permission("weather_query")
async def handle_weather_request(user, city):
    return await WeatherSkill().get_weather(city)

并发数	同步模式	异步模式
10	125	980
100	83	2450

预热缓存：在技能启动时预加载常用数据
懒加载：延迟初始化非核心组件
保持热实例：通过心跳机制防止被平台回收

import hashlib

def generate_idempotent_key(params: dict) -> str:
    param_str = "&".join(f"{k}={v}" for k,v in sorted(params.items()))
    return hashlib.md5(param_str.encode()).hexdigest()