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 行情接入指南。持續訂閱的成本與標的數量基本無關（握手計費、推送不計條），但查詢量會計入套餐額度，建議提前估算。