当 AI 助手开始查行情:TickDB SKILL 协议的技术全解

你问 AI:"帮我看看英伟达现在的买卖压力比。"

它没有调用任何工具,没有查 API,直接告诉你一个数字。

这个数字从哪来的?不知道。它可能幻觉出来的,也可能是训练数据里偶然拼对的。无论哪种,这都不是工程意义上可靠的行情查询。

真正可靠的方案,是让 AI 拥有一个精确的 Function Calling 接口——它知道有哪些函数可以调用、参数是什么、返回值如何解析。 SKILL 协议,就是为这个目标设计的。

本文完整拆解 SKILL 协议的技术细节:skill.md 文件的内部结构、OpenAPI Schema 的设计逻辑、AI 如何解析并执行函数调用,以及多轮对话状态管理的工程实践。


一、为什么需要 SKILL 协议

1.1 自然语言的歧义陷阱

行情查询听起来简单,但用自然语言描述时充满歧义:

用户说法 实际意图 需要解析的要素
"英伟达现在多少" 查询 NVDA.US 当前价格 标的名称→代码映射
"帮我看看特斯拉的买卖盘" 查询 depth 订单簿 标的名称 + 数据类型
"特斯拉和英伟达哪个波动大" 对比隐含波动率或历史波动率 两个标的 + 波动率指标
"最近一小时成交怎么样" 查询 recent trades 或 1h K 线 时间范围 + 数据类型

自然语言处理(NLU)模型在做这些映射时,准确率受限于训练数据和 Prompt 质量。在专业场景下,这种"概率性正确"是不可接受的。

1.2 Function Calling 的标准困境

OpenAI 在 2023 年引入了 Function Calling 规范,Anthropic、Google、阿里等厂商相继跟进。但各家的函数描述格式不统一:

# OpenAI 格式
{"name": "get_stock_price", "description": "获取股票当前价格", "parameters": {...}}

# Anthropic 格式(Claude)
{"name": "get_stock_price", "description": "获取股票当前价格", "input_schema": {...}}

# Google 格式
{"functionDeclarations": [{"name": "get_stock_price", "parameters": {...}}]}

如果每个数据源各自定义格式,应用层需要针对不同模型写适配器。这是一层不必要的复杂度。

1.3 SKILL 协议的解决思路

SKILL 协议定义了一个与模型无关的函数描述规范,封装在 skill.md 文件中。AI 助手在安装 SKILL 时加载这个文件,从中提取函数签名、参数约束和响应解析规则,然后在自己的 Function Calling 实现中使用这些信息。

核心价值三句话:

  1. 一次定义,多端兼容:同一个 skill.md,适配所有主流模型
  2. 结构化约束:参数校验、类型枚举、必填项声明全部内置
  3. 上下文自包含:SKILL 文件内含使用示例,AI 不依赖外部知识即可正确调用

二、skill.md 文件的内部结构

skill.md 是一个遵循固定 Schema 的 Markdown 文件,内部由多个 YAML/JSON 代码块构成。以下是完整的结构拆解。

2.1 文件结构总览

skill.md
├── meta          # SKILL 元信息(名称、版本、作者、描述)
├── functions     # 函数定义列表(核心区块)
│   ├── function_1
│   ├── function_2
│   └── ...
├── examples      # 调用示例(AI 用于 few-shot 学习)
└── constraints   # 全局约束(限频、认证、错误处理)

2.2 meta 区块:SKILL 的身份档案

meta:
  name: tickdb-market-data
  version: "1.0.0"
  description: >
    TickDB 市场数据 SKILL,提供美股、港股、数字货币、
    外汇、贵金属、指数的实时行情和历史 K 线查询能力。
    支持订单簿深度、逐笔成交、分时 K 线等高颗粒度数据。
  provider: tickdb.ai
  model_compatibility:
    - openai
    - anthropic
    - google
    - deepseek
  tags:
    - market-data
    - real-time
    - financial

字段解析:

字段 含义 设计意图
name SKILL 的唯一标识符 AI 加载多个 SKILL 时避免命名冲突
version 语义化版本号 支持版本校验和增量更新
model_compatibility 兼容的模型列表 同一个 SKILL 适配多模型
tags 分类标签 用于 SKILL 市场搜索和推荐

model_compatibility 字段的存在,是 SKILL 协议与模型无关特性的关键。协议不绑定任何特定模型的 Function Calling 格式,而是通过中间层做格式转换。

2.3 functions 区块:函数签名定义

这是 skill.md 最核心的部分。每个函数以 YAML 格式定义完整签名:

functions:
  - name: get_realtime_price
    description: >
      获取单个交易品种的实时行情数据,包括最新价、24h 成交量、
      买一价、卖一价。适用于需要快速价格参考的场景。
      返回数据延迟通常在 100ms 以内。
    category: quote
    parameters:
      type: object
      properties:
        symbol:
          type: string
          description: 交易品种代码,格式为 CODE.MARKET
          examples:
            - NVDA.US
            - 9988.HK
            - BTC.USDT
          pattern: "^[A-Z0-9]+\\.[A-Z]+$"
        fields:
          type: array
          description: 指定返回的数据字段,不填则返回全部字段
          items:
            type: string
            enum: [price, bid, ask, volume, amount, high, low]
          default: [price, bid, ask, volume]
      required: [symbol]

  - name: get_order_book
    description: >
      获取订单簿深度数据,展示指定档位的买卖盘挂单情况。
      可用于计算买卖压力比、流动性深度等衍生指标。
      depth 频道数据不支持外汇、贵金属和指数品种。
    category: depth
    parameters:
      type: object
      properties:
        symbol:
          type: string
          description: 交易品种代码
          examples: [AAPL.US, 0700.HK, ETH.USDT]
          pattern: "^[A-Z0-9]+\\.[A-Z]+$"
        depth:
          type: integer
          description: 深度档位数,美股仅支持 1 档,港股和数字货币最大 10 档
          minimum: 1
          maximum: 10
          default: 5
      required: [symbol]

  - name: get_kline
    description: >
      获取历史 K 线数据,适用于回测和技术分析。
      支持 1m/5m/15m/1h/4h/1d/1w 等多个时间周期。
      不支持美股和 A 股的 tick 级逐笔成交数据。
    category: history
    parameters:
      type: object
      properties:
        symbol:
          type: string
          description: 交易品种代码
          examples: [AAPL.US, BTC.USDT]
        interval:
          type: string
          description: K 线周期
          enum: [1m, 5m, 15m, 30m, 1h, 4h, 1d, 1w]
          default: 1h
        start_time:
          type: string
          format: iso8601
          description: 起始时间,ISO 8601 格式
        end_time:
          type: string
          format: iso8601
          description: 结束时间,ISO 8601 格式
        limit:
          type: integer
          minimum: 1
          maximum: 1000
          default: 100
      required: [symbol, interval]

设计亮点解析:

pattern 正则约束

pattern: "^[A-Z0-9]+\\.[A-Z]+$"

标的代码的格式校验在 Schema 层完成,AI 在构造请求前就能发现参数错误,而不需要等到 API 调用时再收到 2002(品种不存在)错误。这是一种前置校验策略,减少无效 API 调用。

enum 枚举约束

enum: [1m, 5m, 15m, 30m, 1h, 4h, 1d, 1w]

时间周期的枚举限制,让 AI 不会构造出"1.5h"或"30min"这类无效参数。AI 的 Function Calling 质量,取决于 Schema 约束的精细程度。

examples 示例字段
每个参数都附带 examples,这些示例在 few-shot 场景中帮助 AI 理解正确的参数格式。在复杂标的代码映射(如"特斯拉"→"TSLA.US")场景中,examples 尤为重要。

category 分类标签
用于 AI 在多函数中选择。当用户说"看看订单簿"时,AI 识别到 category: depth,快速筛选候选函数。

2.4 examples 区块:Few-shot 学习样本

examples:
  - user: "英伟达现在多少钱"
    thought: >
      用户想查 NVDA.US 的实时价格。标的名称"英伟达"
      需要映射为代码"NVDA.US"。调用 get_realtime_price。
    function: get_realtime_price
    params:
      symbol: NVDA.US
      fields: [price, bid, ask]

  - user: "特斯拉和苹果最近一小时成交怎么样"
    thought: >
      用户想对比 TSLA.US 和 AAPL.US 的近期成交情况。
      需要同时调用两次 get_kline,周期 1h,取最近 3 根 K 线。
    function: get_kline
    params:
      symbol: TSLA.US
      interval: 1h
      limit: 3
    parallel_calls:
      - function: get_kline
        params:
          symbol: AAPL.US
          interval: 1h
          limit: 3

  - user: "比特币深度怎么样"
    thought: >
      用户想看 BTC.USDT 的订单簿深度。调用 get_order_book,
      depth 设为 5 档(默认值)。
    function: get_order_book
    params:
      symbol: BTC.USDT

examples 的关键作用:它不只是告诉 AI"参数怎么填",更重要的是告诉 AI 用户意图如何映射到函数选择thought 字段模拟了 AI 的内部推理过程,这是 Chain-of-Thought 在 SKILL 层的应用。

2.5 constraints 区块:全局约束声明

constraints:
  rate_limit:
    global:
      max_requests_per_minute: 60
      error_code: 3001
      retry_header: Retry-After
    per_endpoint:
      get_realtime_price:
        max_requests_per_minute: 30
      get_kline:
        max_requests_per_minute: 20

  authentication:
    method: api_key
    header: X-API-Key
    env_variable: TICKDB_API_KEY

  error_handling:
    retryable_codes: [3001, 5000, 5001]
    non_retryable_codes: [1001, 1002, 2002]
    max_retries: 3
    backoff_strategy: exponential_jitter

  data_coverage:
    markets:
      - 美股: 支持 K 线(10 年)、实时行情、depth(1 档)
      - 港股: 支持 K 线、实时行情、depth(10 档)
      - 数字货币: 支持 K 线、实时行情、depth(10 档)、trades
    exclusions:
      - 美股和 A 股不支持 tick 级逐笔成交
      - 外汇、贵金属、指数不支持 depth 订单簿

constraints 的价值:它让 AI 具备"知道自己不知道什么"的能力。当用户请求"标普 500 的订单簿"时,AI 查 constraints 发现标普属于指数品种且不支持 depth,应该主动告知用户,而不是发起一个注定失败的请求。


三、AI 如何解析 skill.md

3.1 解析链路

skill.md 文件
    │
    ▼
[解析层] 提取 YAML/JSON 代码块 → 构建中间表示(IR)
    │
    ├── meta → SKILL 注册表(名称、版本、标签)
    ├── functions → 函数索引(name → signature 映射)
    ├── examples → Few-shot 示例库
    └── constraints → 全局约束规则集
    │
    ▼
[适配层] 中间表示 → 模型特定格式
    │
    ├── OpenAI → tools 格式
    ├── Anthropic → tools 格式
    └── Google → functionDeclarations 格式
    │
    ▼
[调用层] 模型生成函数调用 → 执行 → 解析响应 → 返回给用户

关键在于中间表示层的设计:skill.md 不直接生成特定模型的格式,而是先构建一个与模型无关的结构化表示,再由适配层转换为目标格式。

3.2 中间表示的构建逻辑

解析层从 skill.md 中提取的核心数据结构(以 TypeScript 类型表示):

interface SKILL {
  meta: {
    name: string;
    version: string;
    description: string;
    tags: string[];
    modelCompatibility: string[];
  };
  functions: FunctionDefinition[];
  examples: Example[];
  constraints: ConstraintSet;
}

interface FunctionDefinition {
  name: string;
  description: string;
  category: string;
  parameters: {
    type: "object";
    properties: Record<string, ParameterSchema>;
    required: string[];
  };
}

interface ParameterSchema {
  type: string;
  description: string;
  enum?: string[];
  pattern?: string;
  minimum?: number;
  maximum?: number;
  default?: any;
  examples?: string[];
}

interface ConstraintSet {
  rateLimit: RateLimitConfig;
  authentication: AuthConfig;
  errorHandling: ErrorHandlingConfig;
  dataCoverage: DataCoverageConfig;
}

这个中间表示是对 skill.md 内容的结构化抽象,任何模型适配器只需读取这个结构,不需要重新解析 Markdown。

3.3 模型适配层的工作方式

以 OpenAI 适配为例,中间表示 → OpenAI tools 格式的转换:

import yaml
import json
import re

def parse_skill_md(skill_md_content: str) -> dict:
    """
    从 skill.md 中提取 YAML 代码块,构建中间表示
    """
    # 提取所有 YAML 代码块
    yaml_blocks = re.findall(
        r'```yaml\n(.*?)```',
        skill_md_content,
        re.DOTALL
    )
    
    result = {}
    for block in yaml_blocks:
        parsed = yaml.safe_load(block)
        # 按顶级 key 分类
        for key, value in parsed.items():
            result[key] = value
    
    return result


def to_openai_tools(skill_ir: dict) -> list[dict]:
    """
    将中间表示转换为 OpenAI tools 格式
    """
    tools = []
    
    for func in skill_ir.get("functions", []):
        tool = {
            "type": "function",
            "function": {
                "name": func["name"],
                "description": func["description"],
                "parameters": {
                    "type": "object",
                    "properties": {},
                    "required": func["parameters"].get("required", [])
                }
            }
        }
        
        # 转换参数定义
        for param_name, param_schema in func["parameters"].get("properties", {}).items():
            param_def = {
                "type": param_schema.get("type", "string"),
                "description": param_schema.get("description", "")
            }
            
            if "enum" in param_schema:
                param_def["enum"] = param_schema["enum"]
            if "minimum" in param_schema:
                param_def["minimum"] = param_schema["minimum"]
            if "maximum" in param_schema:
                param_def["maximum"] = param_schema["maximum"]
            
            tool["function"]["parameters"]["properties"][param_name] = param_def
        
        tools.append(tool)
    
    return tools


# 实际使用示例
if __name__ == "__main__":
    with open("skill.md", "r") as f:
        content = f.read()
    
    skill_ir = parse_skill_md(content)
    openai_tools = to_openai_tools(skill_ir)
    
    print(json.dumps(openai_tools[0], indent=2, ensure_ascii=False))

输出示例(get_realtime_price 函数):

{
  "type": "function",
  "function": {
    "name": "get_realtime_price",
    "description": "获取单个交易品种的实时行情数据,包括最新价、24h 成交量、买一价、卖一价。返回数据延迟通常在 100ms 以内。",
    "parameters": {
      "type": "object",
      "properties": {
        "symbol": {
          "type": "string",
          "description": "交易品种代码,格式为 CODE.MARKET,示例:NVDA.US"
        },
        "fields": {
          "type": "array",
          "description": "指定返回的数据字段,不填则返回全部字段",
          "items": {
            "type": "string",
            "enum": ["price", "bid", "ask", "volume", "amount", "high", "low"]
          }
        }
      },
      "required": ["symbol"]
    }
  }
}

四、Function Calling 的执行流程

4.1 完整调用链路时序图

用户: "英伟达现在多少钱"
    │
    ▼
┌─────────────────────────────────────────────────────────┐
│  AI 模型(已加载 SKILL tools)                          │
│  1. 识别用户意图 → 匹配函数: get_realtime_price        │
│  2. 构造参数: symbol="NVDA.US"                          │
│  3. 输出 function_call 对象                             │
└─────────────────────────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────────────────────────┐
│  SKILL 运行时(skillon_runtime)                        │
│  1. 解析 function_call                                  │
│  2. 读取 constraints → 检查 rate_limit                  │
│  3. 注入认证头 → X-API-Key: {env:TICKDB_API_KEY}       │
│  4. 发起 HTTP 请求                                      │
│  5. 处理响应或错误码                                    │
└─────────────────────────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────────────────────────┐
│  TickDB API                                             │
│  GET /v1/market/realtime?symbol=NVDA.US                │
└─────────────────────────────────────────────────────────┘
    │
    ▼
响应数据 → SKILL 运行时解析 → 格式化输出 → 返回给用户

4.2 SKILL 运行时核心实现

以下是 SKILL 运行时的核心逻辑,展示如何执行 Function Calling 并处理各种边界情况:

import os
import time
import random
import json
import logging
from typing import Any
from dataclasses import dataclass
from enum import Enum

logger = logging.getLogger(__name__)


class ErrorCode(Enum):
    SUCCESS = 0
    INVALID_API_KEY = (1001, 1002)
    SYMBOL_NOT_FOUND = 2002
    RATE_LIMITED = 3001
    SERVER_ERROR = (5000, 5001)
    
    def __init__(self, codes):
        self.codes = codes if isinstance(codes, tuple) else (codes,)
    
    @classmethod
    def from_code(cls, code: int):
        for member in cls:
            if code in member.codes:
                return member
        return None


@dataclass
class SKILLConfig:
    api_key: str
    base_url: str = "https://api.tickdb.ai/v1"
    max_retries: int = 3
    timeout: tuple[float, float] = (3.05, 10)


class SKILLRuntime:
    """
    SKILL 运行时:执行 AI 生成的函数调用
    遵循 constraints 中的全局约束
    """
    
    def __init__(self, config: SKILLConfig):
        self.api_key = config.api_key
        self.base_url = config.base_url
        self.max_retries = config.max_retries
        self.timeout = config.timeout
        
        # 限频追踪(滑动窗口)
        self.request_timestamps: list[float] = []
        self.rate_limit_per_minute = 60
    
    def _check_rate_limit(self, endpoint: str = "global") -> bool:
        """检查是否触发限频"""
        now = time.time()
        # 清理 60 秒前的记录
        self.request_timestamps = [
            ts for ts in self.request_timestamps if now - ts < 60
        ]
        
        if len(self.request_timestamps) >= self.rate_limit_per_minute:
            sleep_time = 60 - (now - self.request_timestamps[0])
            logger.warning(f"限频触发,等待 {sleep_time:.1f}s")
            time.sleep(sleep_time)
        
        self.request_timestamps.append(time.time())
    
    def _execute_http(
        self,
        method: str,
        endpoint: str,
        params: dict | None = None,
        json_data: dict | None = None
    ) -> dict[str, Any]:
        """执行 HTTP 请求,带重试逻辑"""
        import requests
        
        url = f"{self.base_url}{endpoint}"
        headers = {
            "X-API-Key": self.api_key,
            "Content-Type": "application/json"
        }
        
        base_delay = 1.0
        max_delay = 30.0
        
        for attempt in range(self.max_retries):
            try:
                response = requests.request(
                    method=method,
                    url=url,
                    headers=headers,
                    params=params,
                    json=json_data,
                    timeout=self.timeout
                )
                
                data = response.json()
                code = data.get("code", 0)
                
                if code == 0:
                    return data.get("data", {})
                
                error_type = ErrorCode.from_code(code)
                
                if error_type in (ErrorCode.INVALID_API_KEY, ErrorCode.SYMBOL_NOT_FOUND):
                    # 非重试类错误,直接抛出
                    raise ValueError(
                        f"API 错误 {code}: {data.get('message', '未知错误')}"
                    )
                
                if error_type == ErrorCode.RATE_LIMITED:
                    # 限频错误:读取 Retry-After 头
                    retry_after = int(response.headers.get("Retry-After", 5))
                    logger.warning(f"限频 (3001),等待 {retry_after}s")
                    time.sleep(retry_after)
                    continue
                
                if error_type == ErrorCode.SERVER_ERROR:
                    # 服务端错误:指数退避重试
                    delay = min(base_delay * (2 ** attempt), max_delay)
                    jitter = random.uniform(0, delay * 0.1)
                    sleep_time = delay + jitter
                    logger.warning(f"服务端错误 ({code}),{sleep_time:.1f}s 后重试")
                    time.sleep(sleep_time)
                    continue
                
                raise RuntimeError(f"未知错误码 {code}: {data.get('message')}")
                
            except requests.exceptions.Timeout:
                delay = min(base_delay * (2 ** attempt), max_delay)
                jitter = random.uniform(0, delay * 0.1)
                logger.warning(f"请求超时,{delay + jitter:.1f}s 后重试")
                time.sleep(delay + jitter)
                
            except requests.exceptions.RequestException as e:
                if attempt == self.max_retries - 1:
                    raise RuntimeError(f"网络请求失败: {e}")
                time.sleep(base_delay * (2 ** attempt))
        
        raise RuntimeError(f"达到最大重试次数 ({self.max_retries})")
    
    def call_function(self, function_name: str, arguments: dict[str, Any]) -> Any:
        """执行 AI 生成的函数调用"""
        self._check_rate_limit()
        
        if function_name == "get_realtime_price":
            return self._get_realtime_price(**arguments)
        elif function_name == "get_order_book":
            return self._get_order_book(**arguments)
        elif function_name == "get_kline":
            return self._get_kline(**arguments)
        else:
            raise ValueError(f"未知函数: {function_name}")
    
    def _get_realtime_price(self, symbol: str, fields: list[str] | None = None) -> dict:
        """get_realtime_price 函数实现"""
        params = {"symbol": symbol}
        if fields:
            params["fields"] = ",".join(fields)
        
        return self._execute_http(
            method="GET",
            endpoint="/market/realtime",
            params=params
        )
    
    def _get_order_book(self, symbol: str, depth: int = 5) -> dict:
        """get_order_book 函数实现"""
        return self._execute_http(
            method="GET",
            endpoint="/market/depth",
            params={"symbol": symbol, "depth": depth}
        )
    
    def _get_kline(
        self,
        symbol: str,
        interval: str,
        start_time: str | None = None,
        end_time: str | None = None,
        limit: int = 100
    ) -> dict:
        """get_kline 函数实现"""
        params = {
            "symbol": symbol,
            "interval": interval,
            "limit": limit
        }
        if start_time:
            params["start_time"] = start_time
        if end_time:
            params["end_time"] = end_time
        
        return self._execute_http(
            method="GET",
            endpoint="/market/kline",
            params=params
        )

4.3 与 AI 模型的集成

from openai import OpenAI

def chat_with_skill(user_message: str, skill_md_path: str):
    """
    带 SKILL 能力的对话入口
    """
    # 加载并解析 skill.md
    with open(skill_md_path, "r") as f:
        skill_md_content = f.read()
    
    skill_ir = parse_skill_md(skill_md_content)
    tools = to_openai_tools(skill_ir)
    
    # 初始化运行时
    config = SKILLConfig(
        api_key=os.environ.get("TICKDB_API_KEY", ""),
        base_url="https://api.tickdb.ai/v1"
    )
    runtime = SKILLRuntime(config)
    
    # 初始化模型
    client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
    
    messages = [{"role": "user", "content": user_message}]
    
    # 最大 Tool Call 轮次,防止无限循环
    max_turns = 10
    turn = 0
    
    while turn < max_turns:
        turn += 1
        
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
            tool_choice="auto"
        )
        
        assistant_message = response.choices[0].message
        messages.append(assistant_message)
        
        if not assistant_message.tool_calls:
            # 没有更多 Tool Call,返回最终回复
            return assistant_message.content
        
        # 执行所有 Tool Call
        for tool_call in assistant_message.tool_calls:
            function_name = tool_call.function.name
            arguments = json.loads(tool_call.function.arguments)
            
            try:
                result = runtime.call_function(function_name, arguments)
                tool_result = {
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": json.dumps(result, ensure_ascii=False)
                }
            except Exception as e:
                tool_result = {
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": json.dumps({"error": str(e)}, ensure_ascii=False)
                }
            
            messages.append(tool_result)
    
    return "已达到最大对话轮次限制"


# 使用示例
if __name__ == "__main__":
    result = chat_with_skill(
        user_message="英伟达和特斯拉现在买一卖一价分别是多少?",
        skill_md_path="skill.md"
    )
    print(result)

五、多轮对话状态管理

5.1 状态管理的必要性

多轮对话中,上下文状态的累积直接影响函数调用的准确性:

对话轮次 用户输入 需要的上下文
第 1 轮 "英伟达现在多少" NVDA.US 价格
第 2 轮 "加上下特斯拉" NVDA.US + TSLA.US 价格
第 3 轮 "最近一小时呢" 1h K 线(两个标的)
第 4 轮 "哪个波动大" 对比两个标的的波动率

如果 AI 每次只看到当前轮次的输入,第 4 轮就无法回答"哪个波动大",因为它不知道前几轮查的是哪两个标的。

5.2 上下文累积策略

SKILL 运行时不存储对话历史(那是 AI 模型的职责),但提供上下文注入接口,让 AI 在构造下一轮调用时知道之前查了什么:

class ConversationContext:
    """
    多轮对话上下文追踪器
    记录每轮对话中调用的函数和返回值摘要
    """
    
    def __init__(self):
        self.history: list[dict] = []
    
    def record(self, function_name: str, params: dict, result: Any):
        """记录一轮调用"""
        summary = self._summarize_result(result)
        self.history.append({
            "function": function_name,
            "params": params,
            "result_summary": summary,
            "timestamp": time.time()
        })
    
    def _summarize_result(self, result: Any) -> str:
        """对结果进行摘要,节省上下文 token"""
        if isinstance(result, dict):
            # 只保留关键字段的摘要
            return str({k: v for k, v in list(result.items())[:5]})
        return str(result)[:200]
    
    def get_context_summary(self) -> str:
        """生成上下文摘要,供 AI 在下一轮使用"""
        if not self.history:
            return "(暂无历史查询)"
        
        lines = ["【已查询的数据】"]
        for i, entry in enumerate(self.history, 1):
            symbol = entry["params"].get("symbol", "N/A")
            func = entry["function"]
            summary = entry["result_summary"]
            lines.append(f"{i}. {func}(symbol={symbol}): {summary}")
        
        return "\n".join(lines)
    
    def clear(self):
        """清空上下文(用户开启新话题时)"""
        self.history = []

在 SKILL 运行时的 call_function 方法中集成上下文记录:

class SKILLRuntime:
    def __init__(self, config: SKILLConfig):
        # ... 现有初始化 ...
        self.context = ConversationContext()
    
    def call_function(self, function_name: str, arguments: dict) -> Any:
        self._check_rate_limit()
        
        result = self._execute_function(function_name, arguments)
        
        # 记录到上下文
        self.context.record(function_name, arguments, result)
        
        return result
    
    def get_context_for_prompt(self) -> str:
        """获取当前上下文摘要,注入到 system prompt"""
        return self.context.get_context_summary()

使用时,将上下文摘要注入 system prompt:

SYSTEM_PROMPT = """你是一个专业的金融行情助手。

【上下文规则】
1. 如果用户没有指定标的,优先使用【已查询的数据】中的标的
2. 如果用户说"最近一小时",使用 interval=1h,limit=3
3. 对比类问题,需要同时调用两次函数

【当前上下文】
{context_summary}
"""

def chat_with_skill_v2(user_message: str, skill_md_path: str):
    runtime = SKILLRuntime(config)
    context_summary = runtime.get_context_for_prompt()
    
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT.format(
            context_summary=context_summary
        )},
        {"role": "user", "content": user_message}
    ]
    
    # 后续交互中,AI 根据上下文正确理解用户意图
    # ...

5.3 上下文边界:何时清空

不是每次对话都要累积上下文。以下情况应该清空:

触发条件 原因
用户明确换话题 "好了,我们换个话题"
超过 10 轮未交互 上下文可能已过期
检测到时间跳跃 用户问"上周的行情"而非当前
标的完全变更 从查股票转为查数字货币

SKILL 运行时通过检测函数名变化或时间参数异常来自动触发上下文清空。


六、SKILL 协议 vs 传统 API 调用的对比

维度 传统 API 调用 SKILL 协议
交互方式 程序员写代码,固定参数 自然语言描述,AI 理解后构造参数
参数校验 服务端校验,错误时返回错误码 Schema 层前置校验,减少无效调用
多标的处理 循环调用,代码复杂 AI 自动识别多标的,生成并行调用
上下文感知 无,每次请求独立 通过上下文管理器累积状态
错误恢复 程序员处理 retry 逻辑 SKILL 运行时内置退避重试
模型适配 每个模型写一套适配代码 中间表示层统一,一次定义多端兼容
限频管理 需手动实现 constraints 中声明,运行时自动控制
学习成本 需要阅读 API 文档 AI 直接从 examples 学习用法

SKILL 协议并不替代 API,而是在 API 之上加了一层自然语言接口。对开发者而言,底层仍然是可靠的 REST/WebSocket 调用,只是调用方式从"写代码"变成了"说需求"。


七、部署与使用

7.1 SKILL 安装流程

1. 访问 AI 助手(如 ClawHub 集成的助手)
2. 搜索 "tickdb-market-data" SKILL
3. 点击安装,AI 助手加载 skill.md 并解析
4. 用户输入 API Key(存储在 AI 助手的密钥管理中)
5. 开始对话

7.2 分场景使用指南

场景 推荐用法 示例问题
快速价格查询 单函数调用 "比特币现在多少"
策略回测准备 get_kline + 上下文累积 "帮我拉取特斯拉最近一个月的 1h K 线"
流动性分析 get_order_book + 衍生计算 "苹果的买卖压力比怎么样"
跨品种对比 并行多函数调用 "英伟达和 AMD 哪个成交量大"
异常监控 循环调用 + 阈值告警 "特斯拉买一量跌破 5000 时提醒我"

7.3 开发者自定义 SKILL

如果你需要扩展 SKILL 的函数集,只需在 skill.md 的 functions 区块添加新的函数定义:

functions:
  # ... 现有函数 ...
  
  - name: calculate_pressure_ratio
    description: >
      根据订单簿深度数据计算买卖压力比。
      压力比 = Σ(前 N 档买盘量) / Σ(前 N 档卖盘量)。
      比值 > 1 表示买盘主导,< 1 表示卖盘主导。
    category: analysis
    parameters:
      type: object
      properties:
        order_book:
          type: object
          description: get_order_book 返回的订单簿数据
        depth:
          type: integer
          description: 参与计算的档位数
          minimum: 1
          maximum: 10
          default: 5
      required: [order_book]

添加后,AI 就能在对话中调用 calculate_pressure_ratio,对订单簿数据做二次计算。


结语

SKILL 协议的核心价值,不是让 AI"能查行情",而是让 AI 查行情的方式变得可靠、可控、可解释

通过 skill.md 的 Schema 约束,AI 的函数调用参数在发出前就经过校验。通过 constraints 的全局声明,AI 知道自己能做什么、不能做什么。通过 examples 的 few-shot 样本,AI 学会正确的意图映射方式。通过多轮上下文管理,历史查询不再丢失。

这不是一个用自然语言包装 API 的玩具,而是一套严肃的工程协议。 它的目标,是在 AI 助手这个充满概率性的环境中,引入足够的确定性,让专业场景的自动化成为可能。


下一步行动

如果你是 AI 开发者,想为自己的数据源构建 SKILL 协议支持:

  1. 访问 tickdb.ai 了解 TickDB SKILL 的完整规范和开源运行时
  2. 参考本文的 Schema 设计,为你的 API 定义 skill.md
  3. 使用 parse_skill_md() + to_openai_tools() 构建适配层

如果你想直接使用 TickDB SKILL,在 ClawHub 或支持的 AI 助手中搜索安装 tickdb-market-data,配置你的 API Key 即可开始对话。

如果你在构建多模型兼容的系统,SKILL 协议的中间表示层设计可以作为一个与模型无关的函数注册方案参考。


本文不构成任何投资建议。市场有风险,投资需谨慎。