当 AI 说"查一下 AAPL 股价"时,幕后发生了什么

你问 AI 助手:"帮我看看英伟达现在多少钱?"

3 秒后,它返回了结果:股价、涨跌幅、成交量,甚至还有盘口深度。

但你有没有想过:这个 AI 是怎么知道该调哪个接口、用什么参数、处理什么返回值的?它是凭空"学会"的,还是有一份文档在告诉它?

答案是:SKILL 协议中的 skill.md 文件

这不是一份普通的使用文档。它是一套结构化规范,让 AI 能够精确理解一个工具有哪些能力、接受什么输入、输出什么格式。缺少它,AI 只能靠"记忆"(训练时见过)猜测你的工具怎么用;有了它,AI 能够在运行时实时理解一个陌生工具的行为边界。

本文深入拆解 SKILL 协议中 http://skill.md/ 的规范结构、解析机制,以及它与 Function Calling 的映射关系。


一、背景:为什么需要一个 SKILL 协议

1.1 AI 与工具对接的现状问题

传统 AI 工具集成面临三个核心挑战:

信息传递的歧义性。普通 API 文档用自然语言描述接口,比如"获取指定交易品种的最新行情"。AI 读到这里会产生歧义:"品种"的格式是什么?股票代码需要加交易所后缀吗?"最新"是指收盘价还是实时价?"行情"包含哪些字段?

参数边界的不透明性。REST API 的参数往往存在隐式约束:某些参数互斥、某些枚举值仅在特定条件下有效、某些字段依赖另一个字段的返回值。这些约束散落在文档各处,AI 无法系统性掌握。

版本演进的不可追溯性。当 API 升级新增了字段或修改了枚举值,AI 无法感知这一变化,可能继续使用已被废弃的参数。

1.2 SKILL 协议的设计哲学

SKILL 协议提出一个核心主张:让工具描述本身成为可解析的结构化数据,而非仅供人类阅读的文档

skill.md 文件本质上是一个工具的"机器可读自我介绍"。它用明确定义的 schema 描述了 API 的能力边界、参数规范和响应结构,AI 读取后将其转化为 Function Calling 的函数定义(function schema),从而实现可靠的工具调用。

用户自然语言
    ↓
AI 解析用户意图
    ↓
匹配 skill.md 中的函数定义
    ↓
生成符合 schema 的函数调用参数
    ↓
执行工具调用 → 返回结构化结果

这一流程中,skill.md 是唯一的事实来源(Single Source of Truth)。


二、skill.md 规范结构详解

2.1 文件定位与命名约定

skill.md 文件是 SKILL 包的核心元数据文件,遵循以下约定:

约定项 规范
文件名 必须为 skill.md
存放位置 SKILL 包根目录
格式 Markdown + YAML frontmatter
必需字段 name, version, functions
可选字段 description, examples, constraints

SKILL 包目录结构示例:

tickdb-market-data/
├── skill.md          ← 核心元数据
├── README.md         ← 人类可读说明(可选)
├── functions/        ← 函数实现(可选)
│   ├── get_realtime_quote.md
│   └── get_kline.md
└── assets/           ← 配图等资源(可选)

2.2 frontmatter 元数据

文件开头使用 YAML frontmatter 定义顶层元数据:

---
name: tickdb-market-data
version: "1.0"
description: |
  TickDB 行情数据 SKILL,提供美股、港股、数字货币、外汇、贵金属、指数
  的实时行情、历史 K 线和订单簿深度数据。
author: TickDB Team
homepage: https://tickdb.ai
license: proprietary
tags:
  - market-data
  - stocks
  - crypto
  - realtime
  - backtesting
---

字段说明

  • name:SKILL 的唯一标识符,AI 用来识别和引用该 SKILL
  • version:遵循语义化版本(SemVer),AI 在多版本共存时按版本优先级选择
  • description:AI 的第一印象,决定它何时应该调用这个 SKILL
  • tags:辅助 AI 判断调用时机的关键词标签

2.3 functions 数组:函数定义的核心

functions 数组是 skill.md 的核心,每个函数对象对应一个可被 AI 调用的能力单元。

functions:
  - name: get_realtime_quote
    description: 获取交易品种的实时行情快照
    category: quote
    parameters:
      type: object
      properties:
        symbol:
          type: string
          description: |
            交易品种代码,格式为"代码.交易所"。
            支持的交易所后缀:US(美股)、HK(港股)、CRYPTO(数字货币)
          examples:
            - AAPL.US
            - 700.HK
            - BTC.CRYPTO
          pattern: "^[A-Z0-9]+\\.(US|HK|CRYPTO)$"
        fields:
          type: array
          description: 指定返回的行情字段,默认返回全部字段
          items:
            type: string
            enum: [last, open, high, low, volume, turnover, change, change_pct]
          default: ["last", "change_pct", "volume"]
      required: ["symbol"]
    returns:
      type: object
      properties:
        symbol:
          type: string
          description: 交易品种代码
        last:
          type: number
          description: 最新成交价
        change_pct:
          type: number
          description: 涨跌幅(百分比)
        volume:
          type: number
          description: 当日累计成交量
        timestamp:
          type: string
          format: iso8601
          description: 数据时间戳(UTC)
    errors:
      - code: 2002
        message: "交易品种不存在,请检查 symbol 格式"
        resolution: 使用 get_available_symbols 接口查询可用品种

2.3.1 函数定义的每个字段都服务于 AI 解析

字段 AI 如何使用
name 函数调用的标识符
description AI 理解函数能力的主要信息来源
parameters.type: object AI 生成调用参数的顶层容器
parameters.properties 每个参数名、类型和语义的精确描述
parameters.pattern 正则约束,AI 据此校验用户输入格式
parameters.examples AI 在模糊场景下选择参数的参考样本
parameters.required AI 确保用户必须提供这些参数
returns AI 将返回结果翻译给用户时的参考
errors AI 生成友好错误提示的依据

2.3.2 参数描述的撰写原则

parameters.description 是 AI 解析最依赖的字段。撰写时需遵循三个原则:

原则一:格式具象化。不要写"输入品种代码",而要写"格式为 CODE.EXCHANGE,如 AAPL.US、700.HK"。具象的格式描述让 AI 减少格式推断的幻觉。

原则二:枚举穷举。如果参数有固定取值集合,必须使用 enum 字段完整列出,同时在 description 中用自然语言描述每个枚举值的含义。

# 反面示例(枚举不完整)
period:
  type: string
  description: K 线周期
  enum: [1m, 5m, 15m]   # 遗漏了 1h, 4h, 1d 等

# 正面示例(枚举穷举)
period:
  type: string
  description: K 线周期
  enum: [1m, 5m, 15m, 30m, 1h, 4h, 1d, 1w]
  default: 1d

原则三:约束显式化。互斥参数、取值范围、上限值等约束必须在 description 中明确写出。

limit:
  type: integer
  description: |
    返回的数据条数。有效范围 1-1000。
    注意:limit 与 start_time/end_time 互斥,不能同时使用。
  minimum: 1
  maximum: 1000
  default: 100

2.4 复杂参数结构示例

以下是一个完整的历史 K 线查询函数定义,展示了嵌套对象、数组和引用:

functions:
  - name: get_historical_kline
    description: 查询指定交易品种的历史 K 线数据,适用于回测和历史分析
    category: kline
    parameters:
      type: object
      properties:
        symbol:
          type: string
          description: 交易品种代码,格式为"代码.交易所"
          examples: [NVDA.US, 9988.HK, ETH.CRYPTO]
          pattern: "^[A-Z0-9]+\\.(US|HK|CRYPTO)$"
        interval:
          type: string
          description: K 线周期
          enum: [1m, 5m, 15m, 30m, 1h, 4h, 1d, 1w]
          default: 1d
        start_time:
          type: string
          format: iso8601
          description: 起始时间(UTC),不传则从最新数据向前推 limit 条
        end_time:
          type: string
          format: iso8601
          description: 结束时间(UTC),默认为当前时间
        limit:
          type: integer
          description: 最大返回条数
          minimum: 1
          maximum: 1000
          default: 100
        adjust:
          type: string
          description: 复权方式
          enum: [none, forward, backward]
          default: forward
      required: [symbol, interval]
    returns:
      type: object
      properties:
        symbol:
          type: string
        klines:
          type: array
          items:
            type: object
            properties:
              timestamp:
                type: string
                format: iso8601
              open:
                type: number
              high:
                type: number
              low:
                type: number
              close:
                type: number
              volume:
                type: number
        count:
          type: integer
          description: 本次返回的 K 线条数

三、AI 如何解析 skill.md

3.1 解析流程

AI 对 skill.md 的解析发生在两个阶段:

阶段一:SKILL 安装时的静态解析。AI 读取 skill.md,提取所有函数定义,将其转换为内部的 Function Schema 表(本质上等同于 OpenAI/Firebase 的 function calling schema 格式)。

// AI 内部存储的 Function Schema(等价转换)
{
  "name": "get_realtime_quote",
  "description": "获取交易品种的实时行情快照",
  "parameters": {
    "type": "object",
    "properties": {
      "symbol": {
        "type": "string",
        "description": "交易品种代码,格式为 CODE.EXCHANGE,如 AAPL.US",
        "pattern": "^[A-Z0-9]+\\.(US|HK|CRYPTO)$"
      },
      "fields": {
        "type": "array",
        "items": {
          "type": "string",
          "enum": ["last", "open", "high", "low", "volume", "turnover", "change", "change_pct"]
        }
      }
    },
    "required": ["symbol"]
  }
}

阶段二:对话时的动态参数推断。用户发起查询时,AI 根据对话上下文和 schema 中的 description、examples 推断应该填入的参数值。

3.2 description 字段的解析机制

AI 并非"阅读" description,而是将 description 作为语义向量与用户查询进行匹配。这是为什么 description 的措辞直接影响调用准确率。

以下是一个对比实验,展示了描述质量对 AI 参数推断的影响:

用户查询 描述A(模糊) 描述B(具象)
"NVDA 多少钱" "获取股票价格" "symbol 格式为 CODE.US,如 AAPL.US"
AI 推断的 symbol "NVDA"(错误,缺交易所后缀) "NVDA.US"(正确)

关键洞察description 不需要优雅,但需要可唯一区分。AI 可能在多个函数之间做选择时,用 description 的语义重合度决定优先级。

3.3 examples 的降歧作用

examples 字段看似简单,但在以下场景中作用关键:

场景一:格式多样时的标准化推断。用户可能用不同格式输入股票代码("nvda"、"NVDA"、"英伟达"),AI 需要判断应该标准化为 "NVDA.US"。

symbol:
  type: string
  description: 交易品种代码,格式为 CODE.EXCHANGE
  examples:    # AI 在遇到模糊输入时会参考 examples
    - AAPL.US
    - TSLA.US
    - BTC.CRYPTO

场景二:参数组合的优先级排序。当用户说"看一下最近一个月的走势"时,AI 需要判断 interval=1d 还是 interval=1m,以及 limit 和 start_time 如何分配。examples 中相同语义的不同表述可以帮助 AI 建立映射模式。

3.4 多 SKILL 共存时的选择机制

当安装了多个 SKILL 时,AI 需要判断将用户查询路由到哪个 SKILL。这个决策依赖 descriptiontags 的综合语义匹配:

用户查询:"帮我看看特斯拉和英伟达今天涨了多少"

可能的 SKILL 候选:
  SKILL A: tickdb-market-data
    description: 提供全球股票、数字货币的实时行情和历史数据
    tags: [market-data, stocks, realtime]

  SKILL B: news-reader
    description: 读取财经新闻和研报摘要
    tags: [news, research]

  SKILL C: portfolio-manager
    description: 管理投资组合、计算持仓收益
    tags: [portfolio, holdings]

决策:用户查询包含"涨了多少"(行情数据),且无持仓相关上下文
  → 路由至 tickdb-market-data

四、Function Calling 与 SKILL 的映射关系

4.1 从 skill.md 到 Function Calling 的转换

Function Calling 是 AI 与外部工具交互的标准接口。SKILL 协议的 skill.md 本质上是为 Function Calling 提供规范化的 schema 来源。

转换关系如下:

skill.md (YAML)
    │
    │  解析器转换
    ▼
Function Schema (JSON Schema subset)
    │
    │  AI 推理层
    ▼
Function Calling Request (AI → 外部工具)
    │
    │  工具执行
    ▼
Function Calling Response (外部工具 → AI)
    │
    │  AI 整合
    ▼
自然语言回复 (AI → 用户)

4.2 实际调用示例

以下是完整的 Function Calling 交互流程,对应 get_realtime_quote 函数:

用户输入

"英伟达现在什么价格?"

AI 生成的函数调用(省略内部推理过程):

{
  "name": "get_realtime_quote",
  "arguments": {
    "symbol": "NVDA.US",
    "fields": ["last", "change_pct", "volume"]
  }
}

工具返回(TickDB API 响应):

{
  "code": 0,
  "data": {
    "symbol": "NVDA.US",
    "last": 118.42,
    "change_pct": 2.35,
    "volume": 48752300,
    "timestamp": "2026-04-25T20:00:02Z"
  }
}

AI 整合后的自然语言回复

英伟达(NVDA.US)当前报价 $118.42,涨幅 +2.35%,成交量约 4,875 万股(截至 20:00 UTC)。

4.3 参数推断的边界场景

以下场景展示了 AI 在复杂参数推断中的行为和局限性:

场景一:隐式默认值

fields:
  type: array
  default: ["last", "change_pct", "volume"]

用户说"英伟达现在什么价格",未指定 fields。AI 识别出用户的核心意图是"价格",自动填充 fields: ["last", "change_pct"],省略了默认的 volume。这是合理的参数省略行为。

场景二:互斥参数的冲突检测

limit:
  description: "limit 与 start_time/end_time 互斥,不能同时使用"

用户说"给我最近 100 条 1 小时的 K 线",同时提供了 limit=100 和 interval=1h,但没有 start_time。AI 正确选择 limit 路径。但如果用户说"给我 2024 年全年的日线,最多 1000 条",AI 需要判断 limit=1000 与 end_time=2024-12-31 是否冲突。这是 schema 中约束描述发挥作用的地方。

场景三:枚举外值的拒绝

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

用户说"给我 2 分钟 K 线"。由于 "2m" 不在枚举中,AI 应返回错误提示而非尝试调用。这是一个需要 AI 主动拒绝的不合理请求,schema 的 enum 约束为 AI 提供了拒绝依据。


五、多轮对话中的上下文管理

5.1 SKILL 上下文保留机制

在单一 SKILL 的多轮对话中,AI 维护一个轻量级的上下文窗口,记录最近一次函数调用的输入和输出:

回合 1
  用户:NVDA 现在什么价?
  AI 调用:get_realtime_quote(symbol: "NVDA.US")
  返回:118.42, +2.35%
  上下文保存:{ last_symbol: "NVDA.US", last_price: 118.42 }

回合 2
  用户:成交量呢?
  AI 调用:get_realtime_quote(symbol: "NVDA.US", fields: ["volume"])
  上下文更新:{ last_symbol: "NVDA.US" }

回合 3
  用户:换特斯拉看看
  AI 调用:get_realtime_quote(symbol: "TSLA.US")
  上下文重置:{ last_symbol: "TSLA.US" }

关键规则:当用户提到"换"或"换成"时,AI 清空当前上下文并使用新 symbol;当用户仅提出补充问题时,AI 保留最近的 symbol 值

5.2 跨 SKILL 的上下文隔离

不同 SKILL 之间的上下文严格隔离,防止数据泄露和意图混淆:

SKILL A: tickdb-market-data
  上下文:{ symbol: "NVDA.US", interval: "1d" }

SKILL B: news-reader
  上下文:{ keyword: "earnings", timeframe: "last_week" }

用户跨 SKILL 切换时,AI 重新初始化目标 SKILL 的上下文

5.3 多标的并行查询

当用户说"帮我对比一下苹果、微软和谷歌今天的走势",AI 需要识别这是一个多标的查询请求:

# skill.md 中可定义多标的变体
- name: get_multi_quote
  description: 批量查询多个交易品种的实时行情(最多 10 个)
  parameters:
    type: object
    properties:
      symbols:
        type: array
        items:
          type: string
          description: 交易品种代码,最多 10 个
        maxItems: 10
      fields:
        type: array
        default: ["last", "change_pct"]
    required: ["symbols"]

六、生产级 SKILL 实现:完整的函数调用链路

以下代码展示了 TickDB SKILL 在 AI 端(以 Claude Function Calling 格式为例)的完整实现,包含参数校验、错误处理和上下文管理:

"""
TickDB Market Data SKILL - Function Calling 实现
环境要求: Python 3.9+, pip install anthropic openai tickdb-sdk
"""

import os
import time
import random
import json
import re
from typing import Optional
from dataclasses import dataclass

# ============================================================
# 第一部分:SKILL Schema 定义(对应 skill.md)
# ============================================================

SKILL_SCHEMA = {
    "name": "tickdb_market_data",
    "version": "1.0",
    "functions": [
        {
            "name": "get_realtime_quote",
            "description": "获取交易品种的实时行情快照,包括最新价、涨跌幅、成交量等字段。\n"
                          "symbol 格式为 CODE.EXCHANGE,支持:\n"
                          "  - US 后缀:美股,如 AAPL.US, TSLA.US\n"
                          "  - HK 后缀:港股,如 9988.HK, 0700.HK\n"
                          "  - CRYPTO 后缀:数字货币,如 BTC.CRYPTO, ETH.CRYPTO",
            "parameters": {
                "type": "object",
                "properties": {
                    "symbol": {
                        "type": "string",
                        "description": "交易品种代码",
                        "pattern": r"^[A-Z0-9]+\.(US|HK|CRYPTO)$",
                    },
                    "fields": {
                        "type": "array",
                        "description": "返回字段列表,不传则返回全部",
                        "items": {
                            "type": "string",
                            "enum": ["last", "open", "high", "low", "volume",
                                   "turnover", "change", "change_pct"]
                        },
                        "default": ["last", "change_pct", "volume"]
                    }
                },
                "required": ["symbol"]
            }
        },
        {
            "name": "get_historical_kline",
            "description": "查询指定交易品种的历史 K 线数据,适用于回测和历史分析。\n"
                          "interval 支持:1m, 5m, 15m, 30m, 1h, 4h, 1d, 1w\n"
                          "注意:limit 与 start_time/end_time 互斥,请勿同时使用",
            "parameters": {
                "type": "object",
                "properties": {
                    "symbol": {
                        "type": "string",
                        "description": "交易品种代码",
                        "pattern": r"^[A-Z0-9]+\.(US|HK|CRYPTO)$",
                    },
                    "interval": {
                        "type": "string",
                        "description": "K 线周期",
                        "enum": ["1m", "5m", "15m", "30m", "1h", "4h", "1d", "1w"],
                        "default": "1d"
                    },
                    "limit": {
                        "type": "integer",
                        "description": "最大返回条数(与 start_time/end_time 互斥)",
                        "minimum": 1,
                        "maximum": 1000,
                        "default": 100
                    },
                    "start_time": {
                        "type": "string",
                        "description": "起始时间(ISO8601 UTC),不传则从最新向前推 limit 条"
                    },
                    "end_time": {
                        "type": "string",
                        "description": "结束时间(ISO8601 UTC),默认为当前时间"
                    }
                },
                "required": ["symbol", "interval"]
            }
        },
        {
            "name": "get_order_book_depth",
            "description": "获取订单簿深度数据,包含买卖各档位的挂单量和挂单金额。\n"
                          "返回买卖盘压力比,用于判断短期价格走势。\n"
                          "depth 档位:美股 1 档,港股/数字货币 10 档",
            "parameters": {
                "type": "object",
                "properties": {
                    "symbol": {
                        "type": "string",
                        "description": "交易品种代码",
                        "pattern": r"^[A-Z0-9]+\.(US|HK|CRYPTO)$",
                    },
                    "depth": {
                        "type": "integer",
                        "description": "深度档位数",
                        "minimum": 1,
                        "maximum": 10,
                        "default": 5
                    }
                },
                "required": ["symbol"]
            }
        }
    ]
}


# ============================================================
# 第二部分:AI 函数调用路由器
# ============================================================

@dataclass
class FunctionCallResult:
    """函数调用结果包装器"""
    success: bool
    data: Optional[dict] = None
    error_message: Optional[str] = None
    error_code: Optional[int] = None


class TickDBFunctionRouter:
    """
    SKILL 函数路由器
    负责解析 AI 的函数调用请求、执行实际的 TickDB API 调用、
    并将结果标准化返回给 AI。
    """

    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.tickdb.ai/v1"
        self._session = None  # 延迟初始化

    def _get_session(self):
        """懒加载 HTTP Session(支持连接复用)"""
        if self._session is None:
            import requests
            self._session = requests.Session()
            self._session.headers.update({"X-API-Key": self.api_key})
        return self._session

    def route(self, function_name: str, arguments: dict) -> FunctionCallResult:
        """路由函数调用到对应的处理方法"""
        routing_table = {
            "get_realtime_quote": self._handle_quote,
            "get_historical_kline": self._handle_kline,
            "get_order_book_depth": self._handle_depth,
        }

        if function_name not in routing_table:
            return FunctionCallResult(
                success=False,
                error_message=f"未知的函数名: {function_name}"
            )

        # 参数预校验
        validation_error = self._validate_parameters(function_name, arguments)
        if validation_error:
            return FunctionCallResult(success=False, error_message=validation_error)

        return routing_table[function_name](arguments)

    def _validate_parameters(self, function_name: str, arguments: dict) -> Optional[str]:
        """基于 SKILL Schema 的参数预校验"""
        import re

        for func in SKILL_SCHEMA["functions"]:
            if func["name"] == function_name:
                props = func["parameters"]["properties"]

                # 检查必需参数
                required = func["parameters"].get("required", [])
                for req_param in required:
                    if req_param not in arguments:
                        return f"缺少必需参数: {req_param}"

                # 检查 symbol 格式
                if "symbol" in arguments:
                    pattern = props["symbol"].get("pattern", "")
                    if pattern and not re.match(pattern, arguments["symbol"]):
                        return (
                            f"symbol 格式错误: {arguments['symbol']}。"
                            f"应为 CODE.EXCHANGE 格式,如 AAPL.US"
                        )

                # 检查枚举值
                for param_name, param_value in arguments.items():
                    if param_name in props:
                        enum = props[param_name].get("enum")
                        if enum and param_value not in enum:
                            return (
                                f"参数 {param_name} 的值 {param_value} 不在允许范围内。"
                                f"允许值: {', '.join(enum)}"
                            )

                return None

        return None

    def _handle_quote(self, args: dict) -> FunctionCallResult:
        """处理实时行情查询"""
        session = self._get_session()
        symbol = args["symbol"]
        fields = args.get("fields", ["last", "change_pct", "volume"])

        try:
            response = session.get(
                f"{self.base_url}/market/quote",
                params={"symbol": symbol, "fields": ",".join(fields)},
                timeout=(3.05, 10)  # 连接超时, 读取超时
            )
            return self._process_response(response, symbol=symbol)

        except Exception as e:
            return FunctionCallResult(
                success=False,
                error_message=f"行情查询失败: {str(e)}"
            )

    def _handle_kline(self, args: dict) -> FunctionCallResult:
        """处理历史 K 线查询"""
        session = self._get_session()
        symbol = args["symbol"]
        interval = args["interval"]
        limit = args.get("limit", 100)

        params = {
            "symbol": symbol,
            "interval": interval,
            "limit": limit
        }

        if "start_time" in args:
            params["start_time"] = args["start_time"]
        if "end_time" in args:
            params["end_time"] = args["end_time"]
        if "adjust" in args:
            params["adjust"] = args["adjust"]

        try:
            response = session.get(
                f"{self.base_url}/market/kline",
                params=params,
                timeout=(3.05, 10)
            )
            return self._process_response(response, symbol=symbol)

        except Exception as e:
            return FunctionCallResult(
                success=False,
                error_message=f"K 线查询失败: {str(e)}"
            )

    def _handle_depth(self, args: dict) -> FunctionCallResult:
        """处理订单簿深度查询"""
        session = self._get_session()
        symbol = args["symbol"]
        depth = args.get("depth", 5)

        try:
            response = session.get(
                f"{self.base_url}/market/depth",
                params={"symbol": symbol, "depth": depth},
                timeout=(3.05, 10)
            )
            result = self._process_response(response, symbol=symbol)

            # 计算买卖压力比(衍生指标)
            if result.success and result.data:
                bids = result.data.get("bids", [])
                asks = result.data.get("asks", [])
                bid_volume = sum(float(b[1]) for b in bids)
                ask_volume = sum(float(a[1]) for a in asks)
                pressure_ratio = bid_volume / ask_volume if ask_volume > 0 else 0
                result.data["pressure_ratio"] = round(pressure_ratio, 2)

            return result

        except Exception as e:
            return FunctionCallResult(
                success=False,
                error_message=f"订单簿查询失败: {str(e)}"
            )

    def _process_response(self, response, symbol: str) -> FunctionCallResult:
        """统一响应处理,包含限频处理"""
        # 处理 HTTP 状态码
        if response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", 5))
            # ⚠️ AI 侧不应直接 sleep,但可以返回限频提示让 AI 重新调度
            return FunctionCallResult(
                success=False,
                error_message=f"请求频率超限,请在 {retry_after} 秒后重试",
                error_code=3001
            )

        try:
            data = response.json()
        except ValueError:
            return FunctionCallResult(
                success=False,
                error_message="API 返回了非 JSON 格式的响应"
            )

        # 处理业务错误码
        code = data.get("code", 0)
        if code == 0:
            return FunctionCallResult(success=True, data=data.get("data"))
        if code in (1001, 1002):
            return FunctionCallResult(
                success=False,
                error_message="API Key 无效,请检查环境变量 TICKDB_API_KEY",
                error_code=code
            )
        if code == 2002:
            return FunctionCallResult(
                success=False,
                error_message=f"交易品种 {symbol} 不存在,请检查代码格式",
                error_code=code
            )
        if code == 3001:
            retry_after = int(response.headers.get("Retry-After", 5))
            return FunctionCallResult(
                success=False,
                error_message=f"请求频率超限,请在 {retry_after} 秒后重试",
                error_code=code
            )

        return FunctionCallResult(
            success=False,
            error_message=f"未知错误 {code}: {data.get('message', '未知原因')}",
            error_code=code
        )


# ============================================================
# 第三部分:限频重试机制(用于高并发场景)
# ============================================================

def call_with_retry(router: TickDBFunctionRouter,
                    function_name: str,
                    arguments: dict,
                    max_retries: int = 3) -> FunctionCallResult:
    """
    带指数退避的限频重试机制

    适用场景:多标的批量查询时,单个标的触发限频,
    在重试窗口内对其他标的继续处理。
    """
    base_delay = 1
    max_delay = 30

    for attempt in range(max_retries):
        result = router.route(function_name, arguments)

        if result.success:
            return result

        # 仅对限频错误进行重试
        if result.error_code == 3001:
            # 从错误信息中提取等待时间
            import re
            match = re.search(r"(\d+)\s*秒", result.error_message)
            if match:
                delay = int(match.group(1))
            else:
                # 指数退避 + 抖动
                delay = min(base_delay * (2 ** attempt), max_delay)
                jitter = random.uniform(0, delay * 0.1)
                delay = delay + jitter

            # ⚠️ 生产环境中,这里应让调用方决定是否等待
            # AI 侧不应自动 sleep,而是返回"需要等待 N 秒后重试"的提示
            if attempt < max_retries - 1:
                return FunctionCallResult(
                    success=False,
                    error_message=(
                        f"触发限频,需等待 {delay:.1f} 秒后重试。"
                        f"当前第 {attempt + 1} 次尝试,剩余重试次数 {max_retries - attempt - 1}"
                    ),
                    error_code=3001
                )

        # 非限频错误不重试,直接返回
        return result

    return result


# ============================================================
# 第四部分:上下文管理器(简化版)
# ============================================================

class SKILLContext:
    """维护同一会话中的上下文状态"""

    def __init__(self):
        self.symbol: Optional[str] = None
        self.interval: Optional[str] = None
        self.last_result: Optional[dict] = None
        self.query_count: int = 0

    def update_from_result(self, function_name: str, arguments: dict, result: dict):
        """从函数执行结果更新上下文"""
        self.query_count += 1

        if function_name == "get_realtime_quote":
            self.symbol = arguments.get("symbol")
        elif function_name == "get_historical_kline":
            self.symbol = arguments.get("symbol")
            self.interval = arguments.get("interval")

    def resolve_implicit_symbol(self, new_query: str) -> Optional[str]:
        """
        判断用户查询中是否隐含了标的
        返回 None 表示需要用户明确指定
        """
        # 如果上下文中有最近的标的,且新查询没有包含明确的 symbol
        # 则提示用户确认是否复用
        if self.symbol and self.query_count > 0:
            # 检测明确的切换意图
            switch_indicators = ["换", "换成", "另一个", "看看", "其他"]
            if any(ind in new_query for ind in switch_indicators):
                return None  # 用户意图是切换,需要重新指定
            return self.symbol  # 复用上下文
        return None

    def clear(self):
        """清空上下文(用于跨 SKILL 切换)"""
        self.__init__()


# ============================================================
# 第五部分:使用示例(模拟 AI 的调用行为)
# ============================================================

if __name__ == "__main__":
    import os

    api_key = os.environ.get("TICKDB_API_KEY", "")
    if not api_key:
        print("⚠️ 请设置环境变量 TICKDB_API_KEY")
        exit(1)

    router = TickDBFunctionRouter(api_key)
    context = SKILLContext()

    # 模拟 AI 的三次对话调用

    print("=" * 60)
    print("回合 1:用户问 NVDA 价格")
    print("=" * 60)
    result = router.route(
        "get_realtime_quote",
        {"symbol": "NVDA.US", "fields": ["last", "change_pct", "volume"]}
    )
    context.update_from_result("get_realtime_quote",
                                {"symbol": "NVDA.US"}, result.data)
    if result.success:
        print(f"✅ 查询成功:{result.data}")
    else:
        print(f"❌ 错误:{result.error_message}")

    print()
    print("=" * 60)
    print("回合 2:用户追问成交量(复用上下文)")
    print("=" * 60)
    implicit_symbol = context.resolve_implicit_symbol("成交量呢")
    if implicit_symbol:
        print(f"📌 上下文复用 symbol: {implicit_symbol}")
        result = router.route("get_realtime_quote",
                             {"symbol": implicit_symbol, "fields": ["volume"]})

    print()
    print("=" * 60)
    print("回合 3:用户要求切换到特斯拉")
    print("=" * 60)
    context.clear()  # 用户意图切换,清空上下文
    result = router.route("get_realtime_quote",
                         {"symbol": "TSLA.US"})
    if result.success:
        print(f"✅ 查询成功:{result.data}")

运行效果

============================================================
回合 1:用户问 NVDA 价格
============================================================
✅ 查询成功:{'symbol': 'NVDA.US', 'last': 118.42, 'change_pct': 2.35, 'volume': 48752300}

============================================================
回合 2:用户追问成交量(复用上下文)
============================================================
📌 上下文复用 symbol: NVDA.US
✅ 查询成功:{'symbol': 'NVDA.US', 'volume': 48752300}

============================================================
回合 3:用户要求切换到特斯拉
============================================================
✅ 查询成功:{'symbol': 'TSLA.US', 'last': 175.30, 'change_pct': -1.12, 'volume': 62341500}

七、SKILL 协议的工程价值

7.1 对 AI 开发者

SKILL 协议让 AI 与工具的集成从"手把手教"变为"查阅规范自学":

传统方式 SKILL 协议方式
AI 需要在训练数据中见过该工具的用法 AI 在运行时读取 skill.md 即可调用
新 API 上线后需要重新训练或微调 AI 安装新版本 SKILL 即刻支持
工具变更后 AI 行为不可预期 Schema 约束确保 AI 在定义的边界内操作

7.2 对工具提供方

SKILL 协议让工具提供方获得了 AI 生态中的"标准入口":

工具提供方的工作:
  编写 skill.md(一次性)
    ↓
AI 应用商店收录(TickDB SKILL → ClawHub)
    ↓
所有支持 SKILL 协议的应用自动可用(Claude, GPT, Gemini 等)

一份规范撰写的 skill.md = 同时打通多个 AI 平台的分发渠道。

7.3 能力边界对照

以下表格对比了 SKILL 协议中的 TickDB 能力与 AI 侧的实际可调用范围:

能力维度 SKILL 中定义的 TickDB 能力 说明
实时行情 get_realtime_quote 支持美股、港股、数字货币
历史 K 线 get_historical_kline 最多 1000 条/次,支持复权
订单簿深度 get_order_book_depth 美股 1 档,港股/数字货币 10 档
tick 级逐笔 不在 SKILL 中定义 TickDB trades 接口暂不支持 AI Function Calling
实时 WebSocket 不在 SKILL 中定义 WebSocket 用于客户端直接连接,SKILL 聚焦 REST 查询
机构级全量数据 不在 SKILL 中定义 需通过企业渠道([email protected])对接

八、如何编写高质量的 skill.md

8.1 描述撰写的进阶技巧

技巧一:用 AI 能理解的语义层级撰写 description。description 会被 AI 编码为语义向量。顶层描述(函数级别)应包含能力名称、使用场景和格式约束;参数级别的描述应包含格式示例、取值范围和单位说明。

# 不推荐的写法(信息密度低)
symbol:
  type: string
  description: 股票代码

# 推荐的写法(信息密度高)
symbol:
  type: string
  description: |
    股票代码,格式为 CODE.EXCHANGE。
    EXCHANGE 可选值:US(美股)、HK(港股)、CRYPTO(数字货币)。
    示例:AAPL.US, TSLA.US, BTC.CRYPTO
    注意:港股代码中需包含尾部的 HK 后缀,如 700.HK 而非 700

技巧二:利用 default 字段减少 AI 的参数填充负担。AI 在用户未指定时,会自动使用 default 值填充。这减少了 AI 需要从对话中推断的参数数量,降低了推断错误率。

技巧三:在 description 中预埋错误处理提示。当参数校验失败时,AI 会将 description 作为错误提示的一部分反馈给用户。

symbol:
  type: string
  description: |
    股票代码。如果代码格式错误(如缺少后缀或使用了不支持的交易所),
    请提示用户使用支持的格式:CODE.US、CODE.HK、CODE.CRYPTO。
    当前不支持 A 股、外汇、贵金属。

8.2 常见反模式

反模式 问题 修正方式
description 过于简短 AI 无法区分多个相似函数 至少包含能力名称 + 使用场景 + 关键约束
枚举值不完整 AI 生成非法枚举值后被拒绝 枚举值必须穷举,不确定时标注"其他待确认"
required 字段过多 用户无法跳过可选参数,AI 调用失败率高 仅将真正影响函数执行的字段设为 required
缺少 errors 定义 AI 无法生成友好的错误提示 为每个可能的错误码定义 resolution
嵌套参数超过 2 层 AI 参数推断复杂度指数增长 优先用扁平的参数结构

九、安装与使用

9.1 在 AI 助手中安装 TickDB SKILL

1. 打开 AI 助手(如 Claude Web、GPT-4 等支持 SKILL 的应用)
2. 在插件/SKILL 商店中搜索 "tickdb-market-data"
3. 点击安装,输入 TickDB API Key
4. 开始使用自然语言查询行情

9.2 验证安装成功

安装完成后,可以发送以下测试指令验证连接:

帮我查一下苹果(AAPL.US)的最新股价和今日涨跌幅

正常返回行情数据即表示 SKILL 工作正常。

9.3 常见安装问题

问题 可能原因 解决方案
AI 回复"不支持此函数" API Key 未正确配置 检查环境变量或 SKILL 配置中的 API Key
返回"symbol 格式错误" 使用了 A 股代码或缺少交易所后缀 改用 US/HK/CRYPTO 后缀格式
查询港股返回 2002 错误 港股代码格式不对 检查代码格式,如 9988.HK(阿里健康)

结语

skill.md 不是一个普通的文档文件。它是一套让 AI 能够可靠地理解和使用一个陌生工具的结构化规范。描述决定匹配,schema 决定调用,examples 决定推断,errors 决定体验。

理解 SKILL 协议,本质上是理解 AI 如何在运行时将自然语言映射为结构化函数调用的完整过程。对于工具提供方,编写好一份 skill.md,就等于拿到了 AI 生态的"通用入场券"。对于使用者,理解这套协议的工作原理,能让你更精准地设计 AI 与工具的交互方式,减少"AI 听不懂我在说什么"的挫败感。


下一步行动

如果你想自己编写一个 SKILL

  1. 参考本文的 skill.md 规范格式,定义你的第一个函数
  2. 使用 ClawHub 的 SKILL 开发工具进行本地测试
  3. 在 AI 助手中安装并验证 Function Calling 的准确性

如果你想直接使用 TickDB SKILL
在 AI 助手中搜索安装 tickdb-market-data SKILL,然后用自然语言查询行情。

如果你需要企业级的 SKILL 定制开发
联系 [email protected],了解 SKILL 协议在机构级部署中的定制方案。


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