大模型skill开发实战：从零构建你的第一个AI技能模块

2次阅读

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

大模型 skill（技能模块）的核心价值在于：扩展大语言模型 (LLM) 的能力边界，将通用 AI 转化为垂直领域解决方案；实现业务逻辑与技术架构的闭环，让 AI 真正落地；通过模块化开发降低迭代成本，快速响应需求变化。

过度依赖 prompt engineering（提示词工程）：试图通过复杂提示词解决所有问题，导致维护成本飙升
忽视技能边界定义：未明确划分系统能力与技能职责，造成功能混乱
忽略异常流处理：只考虑理想交互路径，实际使用时频频报错

状态管理混乱：用全局变量存储会话状态，导致多用户数据污染
同步阻塞调用：直接在大模型请求线程执行耗时操作，影响系统吞吐量
硬编码配置：将 API 密钥等敏感信息直接写入代码，存在安全风险

建议使用分类模型 + 规则引擎的双重校验机制：

# 基于 FastAPI 的意图识别示例
from enum import Enum

class Intent(str, Enum):
    WEATHER = "weather_query"
    NEWS = "news_search"
    UNKNOWN = "unknown"

def recognize_intent(text: str) -> Intent:
    """
    使用轻量级 BERT 模型进行意图分类
    返回预设的意图枚举值
    """
    # 实际项目应替换为真实的模型推理代码
    if "天气" in text:
        return Intent.WEATHER
    elif "新闻" in text:
        return Intent.NEWS
    return Intent.UNKNOWN

保持处理函数纯净（无状态）：

def handle_weather_query(params: dict, session_id: str) -> dict:
    """
    天气查询业务逻辑
    :param params: 包含 location 等参数的字典
    :param session_id: 当前会话唯一标识
    :return: 标准化结果字典
    """
    # 示例调用第三方天气 API
    api_url = f"https://weather.example.com/api?city={params['location']}"
    response = requests.get(api_url, timeout=3)

    return {"data": response.json(),
        "tts": f"{params['location']}当前气温 {response.json()['temp']} 度",
        "display": generate_weather_card(response.json())
    }

遵循平台接口规范：

def format_response(intent: Intent, result: dict) -> dict:
    """
    标准化输出结构
    包含语音播报 (TTS)、视觉展示(Display)、原始数据(Data) 三层
    """return {"version":"1.0","session_attributes": {},"response": {"output_speech": {"type":"PlainText","text": result.get("tts","")
            },
            "card": {
                "type": "Standard",
                "title": intent.value,
                "content": result.get("display", "")
            }
        }
    }

异步化处理：对第三方 API 调用使用 async/await

import aiohttp

async def async_weather_query(location: str):
    async with aiohttp.ClientSession() as session:
        async with session.get(f"https://api.weatherapi.com/v1/current.json?key=YOUR_KEY&q={location}"
        ) as resp:
            return await resp.json()

缓存策略：对高频查询结果进行缓存

from datetime import timedelta
from fastapi_cache.decorator import cache

@cache(expire=timedelta(minutes=10))
async def get_cached_weather(city: str):
    return await async_weather_query(city)

输入校验层：

from pydantic import BaseModel, constr

class WeatherRequest(BaseModel):
    location: constr(strip_whitespace=True, min_length=2, max_length=20)
    date: str | None = None

权限控制：

# JWT 验证装饰器示例
from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def validate_token(token: str = Depends(oauth2_scheme)):
    if not verify_jwt(token):  # 实际项目需实现 JWT 验证逻辑
        raise HTTPException(status_code=401, detail="Invalid token")
    return token