如何写好一个Skill：从架构设计到最佳实践

2次阅读

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

开发对话式 AI 技能时，开发者常遇到几个典型问题：

意图冲突 ：当用户输入模糊时，系统难以准确识别意图。比如在电商场景中，用户说 ” 查看便宜手机 ”，可能同时触发 ” 价格筛选 ” 和 ” 商品浏览 ” 两个意图。
多轮对话状态丢失 ：在复杂流程中，系统容易丢失上下文。例如用户先询问 ” 两千元以下的手机 ”，再说 ” 要华为的 ”，若未保持状态，系统会忘记之前的价格条件。
第三方 API 集成复杂 ：对接支付、物流等外部服务时，需要处理各种异常情况。比如调用物流 API 超时后，如何优雅降级而不中断对话。

规则引擎 ：适合流程固定、意图明确的场景（如客服 FAQ）。优点是开发简单、响应快，但难以处理复杂语义。
机器学习 ：适合意图多变的情况（如开放域聊天）。基于 BERT 等模型能更好理解用户表达，但需要大量训练数据。

flowchart TD
    A[NLU 层] -->| 结构化意图 | B[对话管理层]
    B -->| 调用指令 | C[服务集成层]
    C -->| 返回数据 | B
    B -->| 生成回复 | D[用户]

NLU 层 ：将用户输入转为结构化数据（意图 + 实体）
对话管理层 ：通过状态机驱动流程，维护上下文
服务集成层 ：封装外部 API 调用

stateDiagram-v2
    [*] --> 商品浏览
    商品浏览 --> 价格筛选: 用户指定价格
    商品浏览 --> 品牌筛选: 用户指定品牌
    价格筛选 --> 商品浏览: 用户修改条件

from enum import Enum, auto
from datetime import datetime, timedelta

class DialogState(Enum):
    PRODUCT_BROWSING = auto()  # 商品浏览状态
    PRICE_FILTER = auto()      # 价格筛选
    BRAND_FILTER = auto()      # 品牌筛选

class DialogManager:
    def __init__(self):
        self.state = DialogState.PRODUCT_BROWSING
        self.context = {'filters': {},
            'last_active': datetime.now()}
        self.TTL = timedelta(minutes=5)  # 上下文有效期

    def update_state(self, user_input):
        # 处理用户中断
        if user_input == '退出':
            self._cleanup()
            return '对话已结束'

        # 状态转移逻辑
        if '价格' in user_input:
            self.state = DialogState.PRICE_FILTER
            self._extract_price(user_input)
        elif '品牌' in user_input:
            self.state = DialogState.BRAND_FILTER
            self._extract_brand(user_input)

        self.context['last_active'] = datetime.now()
        return self._generate_response()

    def _extract_price(self, text):
        # 实现价格提取逻辑（示例）if '以下' in text:
            self.context['filters']['max_price'] = 2000

    def _check_ttl(self):
        return datetime.now() - self.context['last_active'] < self.TTL

import requests
from tenacity import retry, stop_after_attempt

@retry(stop=stop_after_attempt(3))
def call_api(endpoint, params):
    try:
        response = requests.get(endpoint, params=params, timeout=3)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"API 调用失败: {e}")
        raise

压力测试：在 4 核 8G 服务器上，单节点可处理 1000 并发请求，平均延迟 <500ms
建议：
使用 Redis 缓存高频意图模型
对 NLU 服务做水平扩展

# OAuth2.0 token 管理示例
class TokenManager:
    def __init__(self, client_id, client_secret):
        self.token = None
        self.expires_at = None

    def get_token(self):
        if not self.token or datetime.now() >= self.expires_at:
            self._refresh_token()
        return self.token