DeepSeek 接实时行情，上一篇代码需要补上的 3 个可信度约束

5 月 23 日，我发布过一篇《DeepSeek Function Calling 接实时行情：从工具定义到多轮查询的完整示例》。

那篇文章解决的是：如何把行情接口包装成模型可调用的工具。

但复盘之后，我发现还有一个更重要的问题需要单独说明：

工具接通了，不等于当前行情答案就可信。

对于“苹果现在多少钱”“腾讯今天涨跌如何”这类问题，模型必须先完成一次成功的数据查询，才能回答价格。否则，即便代码配置了工具，模型仍可能直接生成文本，或者在工具失败后继续组织一个看似合理的回答。

另外，为与 DeepSeek 当前官方文档保持一致，本文统一使用其当前表述：Tool Calls。

需要修正的，不是接口，而是调用约束

上一篇的核心方向没有问题：

用户提问 -> DeepSeek 请求工具 -> 程序查询行情 API -> 工具结果回填 -> 模型回答

问题出在三个容易忽略的实现细节。

!image.png

修正一：当前行情查询，不应只依赖 tool_choice="auto"

auto 的含义是：模型可以选择调用工具，也可以直接回答。

这在普通问答里没有问题。例如用户问“什么是 K 线”，模型不需要调用行情接口。

但如果用户问：

请查询 AAPL 当前行情快照。

这已经不是知识问答，而是一次外部事实查询。此时如果仍允许模型自行决定是否调用工具，就留下了一个风险：模型可能没有查询数据，却生成了当前价格描述。

如果当前请求只开放一个行情快照工具，可以这样写：

first = client.chat.completions.create(
    model="deepseek-v4-pro",
    messages=messages,
    tools=ticker_tools,
    tool_choice="required",
)

如果同一次请求中开放了多个工具，例如行情、K 线、基本面，那么对“查当前价格”这种明确任务，更稳妥的方式是指定行情工具：

first = client.chat.completions.create(
    model="deepseek-v4-pro",
    messages=messages,
    tools=tools,
    tool_choice={
        "type": "function",
        "function": {"name": "get_market_ticker"},
    },
)

重点不在于参数写法本身，而在于这条规则：

当前价格必须来自一次实际工具查询，而不是模型自行判断要不要查。

修正二：工具查询失败时，不应继续生成行情答案

另一个容易被忽略的问题是：工具调用失败之后，代码是否还把错误结果交给模型，让它继续输出“分析”。

行情查询可能失败，常见情况包括：

• API Key 缺失、过期或权限不足；
• HTTP 429 或业务限流；
• 请求超时；
• 返回内容不是合法 JSON；
• data 为空；
• 返回结果中找不到用户查询的标的；
• 核心字段缺失。

这些情况下，正确行为不是让模型继续回答价格，而是明确告诉用户：本次查询没有成功。

下面以 TickDB 的行情快照接口为例，写一个只保留关键字段的工具函数：

import os
import requests

TICKER_URL = "https://api.tickdb.ai/v1/market/ticker"
ALLOWED_SYMBOLS = {"AAPL.US", "BTCUSDT"}


def get_market_ticker(symbol: str) -> dict:
    symbol = symbol.upper()

    if symbol not in ALLOWED_SYMBOLS:
        return {
            "ok": False,
            "error": "unsupported_symbol",
            "symbol": symbol,
        }

    try:
        response = requests.get(
            TICKER_URL,
            params={"symbols": symbol},
            headers={"X-API-Key": os.environ["TICKDB_API_KEY"]},
            timeout=10,
        )
    except requests.RequestException as exc:
        return {
            "ok": False,
            "error": "request_failed",
            "detail": str(exc),
        }

    retry_after = response.headers.get("Retry-After")

    if response.status_code == 429:
        return {
            "ok": False,
            "error": "rate_limited",
            "retry_after": retry_after,
        }

    try:
        payload = response.json()
    except ValueError:
        return {
            "ok": False,
            "error": "invalid_json",
        }

    code = payload.get("code")

    if code == 3001:
        return {
            "ok": False,
            "error": "rate_limited",
            "retry_after": retry_after,
        }

    if code in {1001, 1002, 1004}:
        return {
            "ok": False,
            "error": "authentication_or_permission_error",
            "code": code,
            "message": payload.get("message"),
        }

    if response.status_code >= 500 or code != 0:
        return {
            "ok": False,
            "error": "api_error",
            "code": code,
            "message": payload.get("message"),
        }

    rows = payload.get("data")
    if not isinstance(rows, list) or not rows:
        return {
            "ok": False,
            "error": "empty_data",
            "symbol": symbol,
        }

    row = next((item for item in rows if item.get("symbol") == symbol), None)
    if row is None:
        return {
            "ok": False,
            "error": "symbol_not_found",
            "symbol": symbol,
        }

    required_fields = {"symbol", "last_price", "timestamp"}
    if not required_fields.issubset(row):
        return {
            "ok": False,
            "error": "missing_fields",
            "symbol": symbol,
        }

    return {
        "ok": True,
        "symbol": row["symbol"],
        "last_price": row["last_price"],
        "timestamp": row["timestamp"],
    }

这段代码刻意没有让工具返回过多字段。对于“当前价格是否真实查询过”这个问题，只保留：

• symbol
• last_price
• timestamp

已经足够支持回答与追溯。

下面是一份脱敏后的 REST 请求样本。它证明数据层可以取得用于回填的字段；其中价格仅表示截图记录时那一次查询结果。

!image.png

修正三：成功回填后，最终回答阶段关闭继续调用工具

拿到行情结果并回填给模型后，程序通常只需要模型解释这一次已经取得的数据。

如果第二轮仍继续允许模型调用工具，就可能出现新的工具请求，而代码却直接读取文本答案，最终造成流程不完整。

因此，最终生成回答时，应显式设置：

tool_choice="none"

完整的关键控制流程如下：

import json
import os

from openai import OpenAI

client = OpenAI(
    api_key=os.environ["DEEPSEEK_API_KEY"],
    base_url="https://api.deepseek.com",
)

ticker_tools = [{
    "type": "function",
    "function": {
        "name": "get_market_ticker",
        "description": "Query the current market ticker snapshot for one symbol.",
        "parameters": {
            "type": "object",
            "properties": {
                "symbol": {
                    "type": "string",
                    "description": "Trading symbol such as AAPL.US or BTCUSDT.",
                }
            },
            "required": ["symbol"],
        },
    },
}]

messages = [
    {
        "role": "system",
        "content": (
            "用户询问当前行情时，必须调用行情工具。"
            "只有工具返回 ok=true 时，才能说明价格和时间戳；"
            "不得补写工具未返回的数据。"
        ),
    },
    {
        "role": "user",
        "content": "请查询 AAPL 当前行情快照。",
    },
]

# 第一步：要求模型先调用行情工具
first = client.chat.completions.create(
    model="deepseek-v4-pro",
    messages=messages,
    tools=ticker_tools,
    tool_choice="required",
)

assistant_message = first.choices[0].message
tool_calls = assistant_message.tool_calls or []

if not tool_calls:
    raise RuntimeError("模型没有返回必需的工具调用。")

# 保留模型返回的完整 assistant message
messages.append(assistant_message)

tool_results = []

for tool_call in tool_calls:
    if tool_call.function.name != "get_market_ticker":
        result = {
            "ok": False,
            "error": "unexpected_tool",
        }
    else:
        try:
            arguments = json.loads(tool_call.function.arguments)
            result = get_market_ticker(arguments["symbol"])
        except (KeyError, json.JSONDecodeError):
            result = {
                "ok": False,
                "error": "invalid_tool_arguments",
            }

    tool_results.append(result)

    messages.append({
        "role": "tool",
        "tool_call_id": tool_call.id,
        "content": json.dumps(result, ensure_ascii=False),
    })

# 第二步：任何查询失败，都不让模型继续生成价格答案
if not all(item.get("ok") for item in tool_results):
    print("行情查询未成功：", json.dumps(tool_results, ensure_ascii=False))

else:
    # 第三步：数据已经取得，只允许模型解释结果，不再发起新工具调用
    final = client.chat.completions.create(
        model="deepseek-v4-pro",
        messages=messages,
        tools=ticker_tools,
        tool_choice="none",
    )

    print(final.choices[0].message.content)

auto 不是错误，只是不适合所有问题

这里需要补充一个边界：tool_choice="auto" 并不是错误写法。

如果用户问：

请介绍一下苹果公司的业务结构，并在需要时补充当前行情。

模型可以先回答通用背景，再根据任务需要决定是否调用工具，auto 是合理的。

但如果用户的问题本身就是：

AAPL 现在多少钱？

那么工具调用不是可选增强，而是回答成立的前提。此时使用 required 或指定工具，更符合数据可信原则。

可以将选择规则简单归纳为：

为什么这三处修改比增加更多工具更重要

工具数量多，并不会自动提高回答可信度。

即使你为模型提供了行情、K 线、基本面、新闻甚至更多外部能力，只要当前价格仍可能在未查询成功时被生成出来，整条链路就存在根本问题。

在实时数据应用里，更关键的是：

1. 当前事实必须来自外部查询；
2. 查询失败必须被明确暴露；
3. 最终回答只能解释已经取得的数据；
4. 每条回答都能回溯到接口结果与时间字段。

做到这几点，模型才是在解释事实，而不是补全一个听起来合理的答案。

结语

上一篇解决了如何把行情接口交给 DeepSeek 调用；这一篇补上一个更严格的约束：

当用户问的是当前行情，工具调用必须成为回答的前置条件。

模型负责理解问题和表达结果，程序负责查询数据、处理失败和保存证据。只要这个边界守住，AI 查询行情才真正具备工程上的可信基础。

参考资料：

• DeepSeek Tool Calls 官方指南
• DeepSeek Chat Completion API Reference
• TickDB Ticker Snapshot 文档
• TickDB Error Codes 文档

本文仅讨论工具调用与行情数据接入的工程方法，不构成投资建议或交易建议。

通过 TickDB API 获取实时行情数据

一个 API 接入外汇、加密货币、美股、港股、A股、贵金属和全球指数的实时行情。支持 WebSocket 低延迟推送，免费开始使用。