AI Agent Skills 入门指南：从零构建你的第一个智能体技能

1次阅读

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

在智能体技术栈中，AI Agent Skills 是具体能力的执行单元，相当于机器人的「技能库」。它们与 LLM（大语言模型）的关系类似于「手」和「大脑」：LLM 负责理解意图和生成决策，Skills 则负责具体动作执行。例如当用户说 ” 查天气 ” 时，LLM 判断需要调用 WeatherQuerySkill，而该技能会实际完成 API 调用和数据格式化。

方案类型	适用场景	优点	缺点
纯 Prompt 工程	简单文本处理	开发快速，无需编码	复杂逻辑难以实现
函数调用	需要外部数据交互的场景	执行可靠，便于调试	需要编写业务逻辑代码
混合模式	需要动态决策的复合任务	兼顾灵活性和可控性	架构复杂度较高

from typing import Dict, Any
from pydantic import BaseModel, Field

class WeatherParams(BaseModel):
    city: str = Field(..., description="城市名称")
    date: str = Field(default="today", description="查询日期")

@skill_registry.register(
    name="weather_query",
    description="查询指定城市天气情况",
    params_schema=WeatherParams
)
class WeatherQuerySkill:
    # 后续代码将在这里实现

import re

WEATHER_PATTERNS = [r"(.*) 的天气 (怎么样 | 如何 | 怎样)",
    r"(.*)(今天 | 明天 | 后天) 天气"
]

def match_weather_intent(text: str) -> Dict[str, str]|None:
    for pattern in WEATHER_PATTERNS:
        match = re.fullmatch(pattern, text)
        if match:
            return {"city": match.group(1).strip(),
                "date": match.group(2) if len(match.groups())>1 else "today"
            }
    return None

from typing import Optional
import requests

class WeatherQuerySkill:
    # ... 接上文元数据定义

    async def execute(self, params: Dict[str, Any]) -> Dict[str, Any]:
        try:
            validated = WeatherParams(**params)
            # 实际业务逻辑
            api_url = f"https://weather-api.example.com/{validated.city}"
            resp = requests.get(api_url, timeout=5)
            resp.raise_for_status()

            return {
                "status": "success",
                "data": resp.json(),
                "metadata": {
                    "city": validated.city,
                    "date": validated.date
                }
            }
        except Exception as e:
            return {
                "status": "error",
                "message": str(e)
            }

为每个技能请求生成唯一 request_id
执行前检查是否已处理过相同请求
记录执行状态和结果

from uuid import uuid4

request_store = {}

def generate_request_id() -> str:
    return str(uuid4())

import asyncio
from concurrent.futures import ThreadPoolExecutor

async def async_execute(skill, params):
    loop = asyncio.get_event_loop()
    with ThreadPoolExecutor() as pool:
        return await loop.run_in_executor(pool, skill.execute, params)

敏感参数加密存储
输入参数白名单校验
API 密钥轮换机制

from cryptography.fernet import Fernet

key = Fernet.generate_key()
cipher_suite = Fernet(key)

def encrypt_data(data: str) -> str:
    return cipher_suite.encrypt(data.encode()).decode()