AI Agent 接实时行情：工具设计与上下文裁剪

给 AI Agent 接实时行情，最常见的错误做法是：把 WebSocket 收到的 tick 流不断塞进对话上下文，让模型「看着」价格变化。这个方案会在三个地方迅速崩溃——上下文预算、成本和准确性。

正确的思路是把行情数据封装成工具，而不是上下文。这篇文章讲具体怎么做。

为什么不能把 tick 流塞进上下文

先看数字。一个标的每秒 2 条 tick，一条 tick 序列化后约 180 字节，折合约 45 个 token。看起来不多，但：

10 个标的 × 2 条/秒 × 45 token = 900 token/秒
一分钟对话 = 54,000 token
一个小时的会话 = 324 万 token

这还只是原始数据，没有算模型输出和历史对话。结果是：上下文迅速溢出，成本高到无法接受，而且模型在大量重复的 tick 里根本抓不住重点——它会把注意力花在无意义的价格抖动上，而不是用户真正关心的问题。

正确的分层：流式归系统，查询归模型

关键的设计决策是把实时性和智能性分开：

职责	承担者	原因
持续接收 tick、维护最新状态	你的服务端（WebSocket）	确定性工作，不需要模型
判断是否触发条件、是否推送	你的服务端（状态机）	规则明确，模型反而不可靠
理解意图、选择查什么、解释结果	模型	这才是模型擅长的部分

换句话说：模型不接触流，只接触查询结果。 你的服务端持续订阅 WebSocket，把最新价格维护在内存或缓存里；模型需要时，通过工具调用拿到「当前快照」或「聚合结果」。

这样做的额外好处是：模型不需要知道 WebSocket 协议、重连逻辑或去重，这些都是服务端的确定性问题。

工具设计：三个层次的接口

给模型暴露的工具应该按抽象层次设计，而不是把 REST 接口原样包一层：

快照查询：get_quote(symbols) 返回最新价格、涨跌幅与数据时效性标志。这是最常用的工具。
区间查询：get_price_range(symbol, interval, period) 返回 K 线或聚合统计，用于回答「最近一周怎么样」这类问题。
条件查询：get_movers(market, direction, limit) 返回涨跌幅排名，用于回答「今天哪些标的波动最大」。

为什么不要直接暴露原始 REST 接口？因为模型的推理能力有限，给它「任意参数组合的 HTTP 请求」会让它频繁构造错误参数。把参数收敛成业务语义明确的少数几个工具，成功率会显著提高。

数据裁剪：给模型什么、不给什么

工具返回的数据需要主动裁剪，原则是只返回回答当前问题必需的字段：

必给：标的代码、最新价、涨跌幅、数据时间戳、时效性标志（是否延迟 / 是否不可用）。
可选：买卖价、成交量——只在问题涉及流动性时给。
不给：原始 tick 序列、冗余字段、内部 ID、完整的 50 条批量结果（除非问题确实需要）。

其中数据时间戳是最关键的一个字段。原因在下一节。

时效性：模型幻觉的高发区

语言模型没有时间概念。如果你告诉它「BTC 当前价格 80,400」，它会把这个数字当作「现在」的事实，即使这条数据是五分钟前查的。

这会直接导致三类错误：

把过期数据当实时：模型回答「当前价格是 X」，但 X 已经过时。
编造时间：被追问「什么时候的价格」时，模型可能给出一个看似合理但完全虚构的时间。
混淆延迟与实时：如果套餐返回的是延迟数据，模型不会主动说明，反而会以实时口径回答。

解法是把时效性变成工具的强制输出，而不是让模型去推断：

每个工具返回值里固定包含 as_of 时间戳与 delayed 标志。
在工具描述里明确要求模型在回答中引用数据时间。
服务端在数据过期时主动提示，例如「该数据已超过 N 秒，建议重新查询」。

这样模型不需要「知道」时间，只需要「引用」服务端给它的时间。

Python 实现：工具定义与快照缓存

Python

import json, time
import httpx

API_KEY = "your_api_key"
BASE = "https://api.tiqex.io/api/v1/market"
CACHE_TTL = 2.0                      # 快照缓存 2 秒，避免重复请求

_cache: dict[str, tuple[float, dict]] = {}


def get_quote(symbols: list[str]) -> dict:
    """工具实现：返回快照，并强制附带时效性字段。"""
    key = ",".join(sorted(symbols))
    now = time.monotonic()
    hit = _cache.get(key)
    if hit and now - hit[0] < CACHE_TTL:
        return hit[1]

    resp = httpx.get(
        f"{BASE}/quote",
        params={"codes": key},              # 单次最多 50 个 codes
        headers={"X-API-Key": API_KEY},
        timeout=10,
    )
    rows = resp.json().get("data", [])

    # 只保留模型回答问题必需的字段，主动裁剪
    payload = {
        "as_of": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "source": "tiqex",
        "quotes": [
            {
                "symbol": r.get("code"),
                "last": r.get("last"),
                "changePercent": r.get("changePercent"),
                # 时效性标志必须给，避免模型把延迟数据当实时
                "delayed": bool(r.get("delayed")),
                "unavailable": bool(r.get("unavailable")),
            }
            for r in rows
        ],
    }
    _cache[key] = (now, payload)
    return payload


# 暴露给模型的工具描述（OpenAI function calling 风格）
TOOLS = [{
    "type": "function",
    "function": {
        "name": "get_quote",
        "description": (
            "查询一个或多个标的的最新行情快照。"
            "返回结果包含 as_of 时间戳与 delayed 标志。"
            "回答时必须说明数据对应的 as_of 时间；"
            "若 delayed 为 true，必须明确告知用户这是延迟数据。"
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "symbols": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "标的代码列表，例如 ['BTCUSDT', 'AAPL']，最多 50 个",
                }
            },
            "required": ["symbols"],
        },
    },
}]

if __name__ == "__main__":
    print(json.dumps(get_quote(["BTCUSDT", "AAPL"]), ensure_ascii=False, indent=2))

三个设计要点：快照缓存（避免模型连续追问时重复请求）、字段裁剪（只给必要字段）、时效性强制输出（把 as_of 与 delayed 写进返回值与工具描述）。

缓存 TTL 设为 2 秒是有意的：它既避免了同一轮对话内的重复请求，又保证了数据足够新鲜。如果设成 60 秒，模型给出的「当前价格」在行情波动时就会明显失真。

条件推送应该放在服务端，不在模型里

一个常见的架构错误是让模型来判断「价格是否突破了阈值」。这不合适，原因有三：

延迟不可控：每次判断都要一次模型调用，延迟远高于本地比较。
结果不确定：同样的价格，模型可能给出不同判断。
成本不可控：高频判断意味着高频调用。

正确分工是：服务端用状态机判断条件（确定性），条件成立后才调用模型生成解释或建议（智能性）。 这样模型只在真正需要时被调用，成本和延迟都可控。

常见问题

能不能把 WebSocket 数据直接喂给模型？

不建议。tick 流的 token 消耗极高（10 个标的约 900 token/秒），会迅速耗尽上下文并产生无法接受的成本，而且模型在大量价格抖动中难以抓住重点。正确做法是服务端维护最新状态，模型按需查询。

怎么防止模型把过期价格当成当前价格？

把时间戳作为工具的强制输出字段，并在工具描述里要求模型引用数据时间。不要依赖模型自己判断时效性——语言模型没有时间概念。

应该暴露原始 REST 接口还是封装业务工具？

封装业务工具。把参数收敛成语义明确的少数几个工具（快照查询、区间查询、条件查询），模型的调用成功率会明显高于让它自由构造 HTTP 请求。

延迟数据可以用在 Agent 里吗？

可以，但必须在回答中明确标注。工具返回值里的 delayed 标志就是为此设计的，工具描述也应要求模型主动说明数据时效性。

条件判断该放服务端还是模型？

放服务端。阈值比较、去抖、状态机都是确定性逻辑，用代码实现更快、更准、更省。模型只负责条件成立后的解释与建议。

下一步

核心原则一句话：让模型查询，不要让模型盯盘。 场景方案见 AI Agent 方案；快照接口的字段定义见多市场行情接口文档；如果服务端需要持续接收数据，订阅协议见 WebSocket 流式接口文档；接入代码可参考 Python 行情接入指南。持续订阅的成本与标的数量基本无关（握手计费、推送不计条），但查询量会计入套餐额度，建议提前估算。