从零开始编写一个高效skill：新手避坑指南与实践

3次阅读

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

一个完整的 skill 通常由三个核心组件构成：

意图识别：负责理解用户的输入，将其映射到预定义的意图（如查询天气、播放音乐等）。这通常通过自然语言处理（NLP）模型或规则匹配实现。
对话管理：处理多轮对话的上下文，管理对话状态（例如用户当前处于哪个步骤），并决定下一步的响应。
服务集成：调用外部 API 或本地服务完成实际功能（如从天气服务获取数据）。

状态维护混乱：多轮对话中，未能清晰管理用户上下文，导致逻辑错乱。
异常处理缺失：未考虑网络超时、API 失败等场景，导致 skill 崩溃或返回无意义响应。
过度耦合：业务逻辑与对话管理混杂，难以扩展新功能。

以下是一个简单的 skill 框架，使用 Python 的 flask 库实现：

from flask import Flask, request, jsonify

app = Flask(__name__)

# 示例：简单的意图识别函数
def recognize_intent(user_input):
    if "天气" in user_input:
        return "weather_query"
    elif "播放" in user_input:
        return "play_music"
    return "unknown"

@app.route('/skill', methods=['POST'])
def handle_skill():
    data = request.json
    user_input = data.get('user_input', '')

    # 识别意图
    intent = recognize_intent(user_input)

    # 根据意图处理逻辑
    if intent == "weather_query":
        response = handle_weather_query()
    elif intent == "play_music":
        response = handle_play_music()
    else:
        response = {"error": "未知意图"}

    return jsonify(response)

if __name__ == '__main__':
    app.run(debug=True)

状态机是管理多轮对话的关键。以下是一个简单的实现：

class ConversationState:
    def __init__(self):
        self.state = "START"  # 初始状态
        self.context = {}     # 存储对话上下文

    def transition(self, intent):
        if self.state == "START" and intent == "weather_query":
            self.state = "ASK_CITY"
            return "请问您想查询哪个城市的天气？"
        elif self.state == "ASK_CITY" and "city" in intent:
            self.context["city"] = intent["city"]
            self.state = "CONFIRM"
            return f"您想查询 {self.context['city']} 的天气，对吗？"
        # 其他状态转换...
        return "抱歉，我没有理解您的意思。"

集成外部 API 时，务必注意安全：

使用 HTTPS：确保所有 API 调用通过加密通道。
敏感数据加密：如 API 密钥，不要硬编码在代码中，使用环境变量或密钥管理服务。
限流与重试：避免频繁调用导致 API 拒绝服务。

示例代码：

import os
import requests
from requests.exceptions import RequestException

# 从环境变量获取 API 密钥
API_KEY = os.getenv('WEATHER_API_KEY')


def fetch_weather(city):
    try:
        response = requests.get(f"https://api.weather.com/v1?city={city}&key={API_KEY}",
            timeout=5  # 设置超时
        )
        response.raise_for_status()  # 检查 HTTP 错误
        return response.json()
    except RequestException as e:
        # 记录错误并返回友好提示
        print(f"API 调用失败: {e}")
        return {"error": "天气服务暂时不可用"}