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

12次阅读

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

Agent Skill 是智能体系统中用于处理特定任务的核心功能单元。在智能体架构中，它负责将用户输入（如自然语言）转化为可执行动作，并通过上下文保持、意图识别等技术实现连续对话。一个典型的技能包含对话策略、槽位填充和业务逻辑三个核心模块。

开发 Agent Skill 主要有三种实现方式：

规则引擎 ：基于预定义规则树实现，响应延迟低（通常 <50ms），但准确率依赖规则覆盖率。适合流程固定的场景（如密码重置）
机器学习模型 ：通过意图分类模型处理请求，准确率高（可达 90%+），但需要大量标注数据。适合开放域场景（如客服问答）
混合方案 ：关键节点使用规则保障稳定性，其他分支采用模型预测。维护成本适中，适合大多数业务场景

以下是用 Python 实现订餐技能的完整示例，采用有限状态机（FSM）管理对话流程：

from enum import Enum, auto

# 定义状态枚举
class State(Enum):
    START = auto()
    SELECT_FOOD = auto()
    CONFIRM_ORDER = auto()
    PAYMENT = auto()

class FoodOrderSkill:
    def __init__(self):
        self.state = State.START
        self.food_type = None
        self.quantity = None

    def handle_input(self, user_input: str) -> str:
        try:
            if self.state == State.START:
                return self._handle_start()
            elif self.state == State.SELECT_FOOD:
                return self._handle_food_selection(user_input)
            # 其他状态处理...
        except Exception as e:
            self.state = State.START  # 重置状态
            return f"系统错误：{str(e)}，请重新开始"

    def _handle_start(self) -> str:
        self.state = State.SELECT_FOOD
        return "请选择餐品（1. 汉堡 2. 披萨 3. 沙拉）："

    def _handle_food_selection(self, choice: str) -> str:
        food_map = {"1": "汉堡", "2": "披萨", "3": "沙拉"}
        if choice not in food_map:
            return "无效选择，请重新输入"

        self.food_type = food_map[choice]
        self.state = State.CONFIRM_ORDER
        return f"已选择 {self.food_type}，请输入数量："

关键设计要点：

使用 Enum 明确划分对话状态，避免魔法字符串
每个状态对应独立处理方法，保持单一职责
try-catch 包裹主流程，实现技能熔断
用户输入验证在前端状态处理中完成

对于多轮对话场景，建议采用分级缓存：

短期缓存（<5 分钟）：使用内存存储，如 Redis String 结构，存储当前对话槽位
长期缓存（>5 分钟）：持久化到数据库，仅保留必要字段

示例配置：

# Redis 缓存配置示例
CACHE_CONFIG = {
    "host": "redis.prod",
    "port": 6379,
    "db": 1,
    "socket_timeout": 0.5  # 防止技能阻塞
}

对于需要调用外部 API 的技能（如支付验证），必须采用异步执行：

import aiohttp

async def verify_payment(payment_id: str) -> bool:
    async with aiohttp.ClientSession() as session:
        async with session.post(
            "https://payment-api/verify",
            json={"payment_id": payment_id},
            timeout=3.0
        ) as resp:
            return await resp.json()

注意事项：

设置合理的超时时间（建议≤3 秒）
使用连接池复用 TCP 连接
异步方法需统一异常处理

推荐采用分阶段发布策略：

内部测试：10% 内部流量验证基础功能
小规模发布：1% 生产流量测试稳定性
全量发布：分 3 批次，每批次间隔 30 分钟

监控指标应包括：

技能响应时间 P99
异常触发率
对话完成率

敏感信息必须经过处理方可存储：

import re

def sanitize_log(text: str) -> str:
    # 脱敏手机号
    text = re.sub(r"(1[3-9])\d{9}", r"\1******", text)
    # 脱敏身份证号
    text = re.sub(r"[1-9]\d{5}(?:19|20)\d{2}(?:0[1-9]|1[0-2])(?:0[1-9]|[12]\d|3[01])\d{3}[0-9Xx]", "ID_CARD", text)
    return text