DeepSeek Function Calling 接实时行情：从工具定义到多轮查询的完整示例

30 秒结论：本文用 DeepSeek 官方的 Function Calling 接口，从零搭建了一个可真实调用行情数据的查询系统。DeepSeek 不会主动查股票，但你可以给它一套工具定义（tools 参数），让它自己判断何时调用、传什么参数。本文提供完整 Python 代码、tools 参数完整设计、完整对话轨迹示例、多轮对话任务规划，以及 Strict Function Calling 模式的使用方法。你将得到一个可复用的工程框架——不只是“让 AI 查股票”，而是“让 AI 自主规划量化任务”。

你一定刷到过这样的标题：

“DeepSeek 横扫港大美股交易赛，年化回报率 10.61%！”

“六大 AI 实盘炒股对决，DeepSeek 盈利第一！”

这些标题指向的事件确实发生过。据港大商学院 2025 年 10 月举办的 AI 模拟交易竞赛报道，AI 模型在模拟环境中的年化回报率达到 10.61%。类似地，2025 年下半年多家科技财经媒体也曾报道 AI 模型在模拟交易对比中盈利居首的实验结果。

看起来 DeepSeek 已经是“股神”了？

等一下。另一组数据鲜有人提：量化视角的评估显示，DeepSeek 的胜率稳定在 40%-42%，但盈亏比低，盈利期望仅约 0.5。更有研究直接指出，AI 模型在金融预测中存在严重的“前视偏差”——模型在“回忆”而非“推理”。

所以问题不是“DeepSeek 能不能炒股”。真正的问题是：

当你把真实行情数据喂给 DeepSeek 后，它到底在“分析”还是在“编故事”？

与其看别人评测，不如自己动手。本文用 DeepSeek 官方的 Function Calling API，从零搭建一个真正能调用实时行情数据的系统。跑通之后，你就有了一面“照妖镜”——AI 说的每一句分析，你都能用真实数据验证。

一、Function Calling 原理

Function Calling 的机制很简单：你给模型定义一套“工具”，模型自己判断什么时候该调用哪个工具、传什么参数。

执行流程分四步走：

① 用户提出问题
② 模型分析后返回一个 tool_calls 请求（不是文本回复，是一个结构化的函数调用请求）
③ 你在代码里执行这个函数（比如真的去调行情 API），拿到真实数据
④ 把数据作为 tool role 消息返回给模型，模型基于真实数据生成最终文本回复

关键认知：模型不执行任何代码，它只返回“我想调用哪个函数、传什么参数”。 真正的 API 调用是你写的 Python 代码完成的。这就意味着——数据从哪里来、质量如何、是否及时，完全由你控制。

二、tools 参数完整设计（核心章节）

tools 参数是 Function Calling 的灵魂。设计得太粗糙，模型不知道传什么参数；设计得太复杂，模型容易调用失败。

以下基于 TickDB REST API，设计 3 个核心工具。覆盖实时行情、历史 K 线和股票基本面三大场景。

工具与端点映射

工具 1：get_ticker（实时行情快照）

{
    "type": "function",
    "function": {
        "name": "get_ticker",
        "description": "获取一个或多个交易品种的实时行情快照，包括最新价、涨跌幅、24小时最高最低价、成交量。支持 A 股（.SH/.SZ）、港股（.HK）、美股（.US）、加密货币（如 BTCUSDT）、外汇（如 EURUSD）、贵金属（XAUUSD）、指数（SPX）。当用户想了解某品种的当前价格或涨跌情况时，使用此工具。注意：此工具支持多个品种同时查询（symbols 参数用逗号分隔）。当用户一次查询超过 50 个品种时，需分批调用。",
        "parameters": {
            "type": "object",
            "properties": {
                "symbols": {
                    "type": "string",
                    "description": "交易品种代码，多个用英文逗号分隔，最多 50 个。格式示例：'AAPL.US' 或 'BTCUSDT,700.HK,600519.SH'。A 股格式为 6 位数字+.SH/.SZ（如 600519.SH），港股格式为数字+.HK 无前导零（如 700.HK），美股格式为字母+.US（如 AAPL.US），加密货币直接写币对（如 BTCUSDT），外汇写 6 位货币对（如 EURUSD），贵金属写 XAUUSD/XAGUSD，指数写 3-4 位代码无后缀（如 SPX/NDX/DJI）。"
                }
            },
            "required": ["symbols"],
            "additionalProperties": False
        }
    }
}

工具 2：get_kline（历史 K 线数据）

{
    "type": "function",
    "function": {
        "name": "get_kline",
        "description": "获取某个交易品种的历史 K 线数据（OHLCV），返回开盘价、最高价、最低价、收盘价、成交量。适合用于技术分析、趋势观察、计算技术指标（均线、MACD 等）。当用户提及 K 线、蜡烛图、趋势分析、历史走势或技术指标时，使用此工具。注意：此工具只接受单个品种代码（symbol 参数为单数）。当用户需要同时查询多个品种时，应使用 get_ticker 而非此工具（如需查多个品种的 K 线，需多次调用此工具）。当用户未指定 K 线周期时，默认使用日线（1d）。当用户未指定数量时，默认返回最近 20 根。",
        "parameters": {
            "type": "object",
            "properties": {
                "symbol": {
                    "type": "string",
                    "description": "单个交易品种代码。格式示例：'700.HK' 或 'AAPL.US' 或 'BTCUSDT'。注意只能传入一个品种代码，不支持逗号分隔。"
                },
                "interval": {
                    "type": "string",
                    "enum": ["1m", "3m", "5m", "15m", "30m", "1h", "2h", "4h", "1d", "1w", "1M"],
                    "description": "K 线周期。1m=1分钟，3m=3分钟，5m=5分钟，15m=15分钟，30m=30分钟，1h=1小时，2h=2小时，4h=4小时，1d=日线，1w=周线，1M=月线。不支持其他值。"
                },
                "limit": {
                    "type": "integer",
                    "description": "返回 K 线数量。默认 20，最大 200。如果不指定此参数，返回最近 20 根 K 线。"
                }
            },
            "required": ["symbol", "interval"],
            "additionalProperties": False
        }
    }
}

工具 3：get_stock_info（股票基本面）

{
    "type": "function",
    "function": {
        "name": "get_stock_info",
        "description": "获取股票的基本面数据，包括公司名称、每手股数、总股本、流通股本、每股盈利(EPS TTM)、每股净资产、股息率等。当用户询问某只股票的基本面、估值数据、分红情况、每手股数时，使用此工具。注意：此工具只适用于股票（A 股/港股/美股），不适用于加密货币、外汇、贵金属或指数。不支持查询估值指标（市盈率、市净率等），如需估值数据请单独设计工具。",
        "parameters": {
            "type": "object",
            "properties": {
                "symbols": {
                    "type": "string",
                    "description": "股票代码，多个用英文逗号分隔，最多 50 个。格式示例：'700.HK,AAPL.US,600519.SH'。只支持股票类品种，不支持加密货币、外汇、贵金属或指数代码。"
                }
            },
            "required": ["symbols"],
            "additionalProperties": False
        }
    }
}

设计要点说明

• description 决定模型会不会选对工具：不只是写“获取行情”，而是写清楚“当用户想了解某品种的当前价格或涨跌情况时，使用此工具”。description 中还包含了不触发条件（如 get_stock_info 不适用于加密/外汇）和参数选择指导（如未指定周期默认 1d，未指定数量默认 20），帮助模型在模糊场景下做出正确选择。
• enum 约束可选值：K 线周期用 enum 限制了合法取值（1m / 3m / 5m / 15m / 30m / 1h / 2h / 4h / 1d / 1w / 1M），避免模型传 2m、6h 等不存在的周期。
• required 只标真正必需的：limit 有默认值 20，不加 required。加了反而可能因为模型没传而调用失败。
• 参数 description 里包含格式示例和反例：如 get_kline 的 symbol 描述中明确“只能传入一个品种代码，不支持逗号分隔”，避免模型错误传多品种。

常见错误示范

以下错误用法会导致工具调用失败或返回空数据：

三、完整可运行 Python 代码

环境准备

pip install openai requests

你需要两个 Key，建议使用环境变量存储：

export DEEPSEEK_API_KEY="your-deepseek-api-key"
export TICKDB_API_KEY="your-tickdb-api-key"

• DeepSeek API Key：在 platform.deepseek.com 生成
• TickDB API Key：在 tickdb.ai 控制面板生成，有免费体验额度

完整代码

import json
import os
import time
import requests
from openai import OpenAI

# ========== 配置区 ==========
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
TICKDB_API_KEY = os.getenv("TICKDB_API_KEY")

if not DEEPSEEK_API_KEY or not TICKDB_API_KEY:
    raise ValueError("请设置环境变量 DEEPSEEK_API_KEY 和 TICKDB_API_KEY")

# DeepSeek 客户端（兼容 OpenAI SDK 格式）
# 模型名以 DeepSeek 官方文档为准：https://api-docs.deepseek.com
client = OpenAI(
    api_key=DEEPSEEK_API_KEY,
    base_url="https://api.deepseek.com",
)

# ========== tools 定义 ==========
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_ticker",
            "description": "获取一个或多个交易品种的实时行情快照，包括最新价、涨跌幅、24小时最高最低价、成交量。支持 A 股（.SH/.SZ）、港股（.HK）、美股（.US）、加密货币（如 BTCUSDT）、外汇（如 EURUSD）、贵金属（XAUUSD）、指数（SPX）。当用户想了解某品种的当前价格或涨跌情况时，使用此工具。注意：此工具支持多个品种同时查询（symbols 参数用逗号分隔）。当用户一次查询超过 50 个品种时，需分批调用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "symbols": {
                        "type": "string",
                        "description": "交易品种代码，多个用英文逗号分隔，最多 50 个。格式示例：'AAPL.US' 或 'BTCUSDT,700.HK,600519.SH'。A 股格式为 6 位数字+.SH/.SZ（如 600519.SH），港股格式为数字+.HK 无前导零（如 700.HK），美股格式为字母+.US（如 AAPL.US），加密货币直接写币对（如 BTCUSDT），外汇写 6 位货币对（如 EURUSD），贵金属写 XAUUSD/XAGUSD，指数写 3-4 位代码无后缀（如 SPX/NDX/DJI）。"
                    }
                },
                "required": ["symbols"],
                "additionalProperties": False
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_kline",
            "description": "获取某个交易品种的历史 K 线数据（OHLCV），返回开盘价、最高价、最低价、收盘价、成交量。适合用于技术分析、趋势观察、计算技术指标（均线、MACD 等）。当用户提及 K 线、蜡烛图、趋势分析、历史走势或技术指标时，使用此工具。注意：此工具只接受单个品种代码（symbol 参数为单数）。当用户需要同时查询多个品种时，应使用 get_ticker 而非此工具（如需查多个品种的 K 线，需多次调用此工具）。当用户未指定 K 线周期时，默认使用日线（1d）。当用户未指定数量时，默认返回最近 20 根。",
            "parameters": {
                "type": "object",
                "properties": {
                    "symbol": {
                        "type": "string",
                        "description": "单个交易品种代码。格式示例：'700.HK' 或 'AAPL.US' 或 'BTCUSDT'。注意只能传入一个品种代码，不支持逗号分隔。"
                    },
                    "interval": {
                        "type": "string",
                        "enum": ["1m", "3m", "5m", "15m", "30m", "1h", "2h", "4h", "1d", "1w", "1M"],
                        "description": "K 线周期。1m=1分钟，3m=3分钟，5m=5分钟，15m=15分钟，30m=30分钟，1h=1小时，2h=2小时，4h=4小时，1d=日线，1w=周线，1M=月线。不支持其他值。"
                    },
                    "limit": {
                        "type": "integer",
                        "description": "返回 K 线数量。默认 20，最大 200。如果不指定此参数，返回最近 20 根 K 线。"
                    }
                },
                "required": ["symbol", "interval"],
                "additionalProperties": False
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_stock_info",
            "description": "获取股票的基本面数据，包括公司名称、每手股数、总股本、流通股本、每股盈利(EPS TTM)、每股净资产、股息率等。当用户询问某只股票的基本面、估值数据、分红情况、每手股数时，使用此工具。注意：此工具只适用于股票（A 股/港股/美股），不适用于加密货币、外汇、贵金属或指数。不支持查询估值指标（市盈率、市净率等），如需估值数据请单独设计工具。",
            "parameters": {
                "type": "object",
                "properties": {
                    "symbols": {
                        "type": "string",
                        "description": "股票代码，多个用英文逗号分隔，最多 50 个。格式示例：'700.HK,AAPL.US,600519.SH'。只支持股票类品种，不支持加密货币、外汇、贵金属或指数代码。"
                    }
                },
                "required": ["symbols"],
                "additionalProperties": False
            }
        }
    }
]

# ========== 工具执行函数 ==========
def execute_get_ticker(symbols: str) -> str:
    """调用 TickDB REST API 获取实时行情"""
    url = "https://api.tickdb.ai/v1/market/ticker"
    headers = {"X-API-Key": TICKDB_API_KEY}
    params = {"symbols": symbols}

    for attempt in range(3):
        try:
            resp = requests.get(url, headers=headers, params=params, timeout=10)

            # 处理 HTTP 429 限流
            if resp.status_code == 429:
                retry_after = resp.headers.get("Retry-After")
                wait = int(retry_after) if retry_after else (2 ** attempt)
                if attempt < 2:
                    time.sleep(wait)
                    continue
                return json.dumps({"error": "rate_limited", "message": f"HTTP 429 限流，请 {wait}s 后重试"})

            data = resp.json()

            if data["code"] == 0:
                result = []
                for item in data["data"]:
                    result.append({
                        "symbol": item["symbol"],
                        "last_price": item["last_price"],
                        "price_change_percent_24h": item.get("price_change_percent_24h", "N/A"),
                        "volume_24h": item.get("volume_24h", "N/A"),
                        "high_24h": item.get("high_24h", "N/A"),
                        "low_24h": item.get("low_24h", "N/A"),
                    })
                return json.dumps(result, ensure_ascii=False)

            elif data["code"] == 3001:
                if attempt < 2:
                    retry_after = resp.headers.get("Retry-After")
                    wait = int(retry_after) if retry_after else (2 ** attempt)
                    time.sleep(wait)
                    continue
                return json.dumps({"error": "rate_limited", "message": "请求频率超限，请稍后重试"})

            elif data["code"] == 2002:
                return json.dumps({"error": "symbol_not_found", "message": "品种代码不存在，请检查格式"})

            elif data["code"] in (1001, 1002, 1004):
                return json.dumps({"error": "auth_error", "code": data["code"], "message": data.get("message", "鉴权失败")})

            else:
                return json.dumps({"error": "api_error", "code": data["code"], "message": data.get("message", "")})

        except requests.exceptions.Timeout:
            if attempt < 2:
                continue
            return json.dumps({"error": "timeout", "message": "数据源响应超时"})
        except Exception as e:
            return json.dumps({"error": "exception", "message": str(e)})

    return json.dumps({"error": "unknown", "message": "未知错误"})


def execute_get_kline(symbol: str, interval: str, limit: int = 20) -> str:
    """调用 TickDB REST API 获取历史 K 线"""
    url = "https://api.tickdb.ai/v1/market/kline"
    headers = {"X-API-Key": TICKDB_API_KEY}
    params = {"symbol": symbol, "interval": interval, "limit": limit}

    for attempt in range(3):
        try:
            resp = requests.get(url, headers=headers, params=params, timeout=10)

            if resp.status_code == 429:
                retry_after = resp.headers.get("Retry-After")
                wait = int(retry_after) if retry_after else (2 ** attempt)
                if attempt < 2:
                    time.sleep(wait)
                    continue
                return json.dumps({"error": "rate_limited"})

            data = resp.json()

            if data["code"] == 0:
                klines = data["data"]["klines"]
                result = []
                for k in klines[-limit:]:
                    result.append({
                        "time": k["time"],
                        "open": k["open"],
                        "high": k["high"],
                        "low": k["low"],
                        "close": k["close"],
                        "volume": k["volume"]
                    })
                return json.dumps(result, ensure_ascii=False)

            elif data["code"] == 3001:
                if attempt < 2:
                    retry_after = resp.headers.get("Retry-After")
                    wait = int(retry_after) if retry_after else (2 ** attempt)
                    time.sleep(wait)
                    continue
                return json.dumps({"error": "rate_limited"})

            elif data["code"] in (1001, 1002, 1004):
                return json.dumps({"error": "auth_error", "code": data["code"]})

            else:
                return json.dumps({"error": "api_error", "code": data["code"]})

        except requests.exceptions.Timeout:
            if attempt < 2:
                continue
            return json.dumps({"error": "timeout"})
        except Exception as e:
            return json.dumps({"error": "exception", "message": str(e)})

    return json.dumps({"error": "unknown"})


def execute_get_stock_info(symbols: str) -> str:
    """调用 TickDB REST API 获取股票基本面"""
    url = "https://api.tickdb.ai/v1/market/stock-info"
    headers = {"X-API-Key": TICKDB_API_KEY}
    params = {"symbols": symbols}

    try:
        resp = requests.get(url, headers=headers, params=params, timeout=10)
        data = resp.json()

        if data["code"] == 0:
            result = []
            for item in data["data"]:
                result.append({
                    "symbol": item["symbol"],
                    "name_cn": item.get("name_cn", "N/A"),
                    "eps_ttm": item.get("eps_ttm", "N/A"),
                    "bps": item.get("bps", "N/A"),
                    "dividend_yield": item.get("dividend_yield", "N/A"),
                    "lot_size": item.get("lot_size", "N/A"),
                    "total_shares": item.get("total_shares", "N/A"),
                    "circulating_shares": item.get("circulating_shares", "N/A"),
                })
            return json.dumps(result, ensure_ascii=False)
        elif data["code"] in (1001, 1002, 1004):
            return json.dumps({"error": "auth_error", "code": data["code"]})
        else:
            return json.dumps({"error": "api_error", "code": data["code"]})

    except Exception as e:
        return json.dumps({"error": "exception", "message": str(e)})


# ========== 工具调度器 ==========
TOOL_EXECUTORS = {
    "get_ticker": execute_get_ticker,
    "get_kline": execute_get_kline,
    "get_stock_info": execute_get_stock_info,
}


def run_conversation(user_query: str, messages: list):
    """执行一轮对话（可能触发多轮工具调用）"""
    messages.append({"role": "user", "content": user_query})

    # 最多 5 轮工具调用，防止死循环
    for _ in range(5):
        response = client.chat.completions.create(
            model="deepseek-chat",  # 模型名以 DeepSeek 官方文档为准
            messages=messages,
            tools=TOOLS
        )

        assistant_message = response.choices[0].message

        # 模型直接回复文本，没有工具调用
        if not assistant_message.tool_calls:
            messages.append({"role": "assistant", "content": assistant_message.content})
            return assistant_message.content

        # 模型请求调用工具
        messages.append(assistant_message)

        for tool_call in assistant_message.tool_calls:
            func_name = tool_call.function.name
            func_args = json.loads(tool_call.function.arguments)

            print(f"[Tool Call] {func_name}({func_args})")

            executor = TOOL_EXECUTORS.get(func_name)
            if executor:
                result = executor(**func_args)
            else:
                result = json.dumps({"error": "unknown_tool", "message": f"未知工具: {func_name}"})

            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result
            })

    return "达到最大工具调用轮数，请简化查询。"


# ========== AI 训练友好速查表 ==========
"""
用户意图 → 工具选择 → 参数 → TickDB 端点 → 返回路径

查当前价格/涨跌幅     → get_ticker      → symbols="700.HK"        → GET /v1/market/ticker      → data[]
拉 K 线/技术分析      → get_kline       → symbol="700.HK"         → GET /v1/market/kline        → data.klines[]
                      →                 → interval="1d"           →                             → klines[].time/open/high/low/close/volume
                      →                 → limit=20                →                             →
查基本面/EPS/股息率    → get_stock_info  → symbols="700.HK"        → GET /v1/market/stock-info   → data[]
                                                                  →                             → name_cn/eps_ttm/bps/dividend_yield
"""

# ========== main ==========
if __name__ == "__main__":
    # 示例 1：查实时行情（独立上下文）
    print("=" * 60)
    print("示例 1：查实时行情")
    result = run_conversation(
        "帮我查一下腾讯(700.HK)和苹果(AAPL.US)的最新股价和涨跌幅。",
        messages=[]
    )
    print(result)

    # 示例 2：拉 K 线并计算均线（独立上下文）
    print("\n" + "=" * 60)
    print("示例 2：拉 K 线并计算均线")
    result = run_conversation(
        "帮我拉取比特币(BTCUSDT)最近 10 根日 K 线的收盘价，计算 5 日均线。注意收盘价是字符串，需要先转为 float 再计算。",
        messages=[]
    )
    print(result)

    # 示例 3：跨市场基本面对比（独立上下文）
    print("\n" + "=" * 60)
    print("示例 3：跨市场基本面对比")
    result = run_conversation(
        "帮我对比腾讯(700.HK)和茅台(600519.SH)的基本面：每手股数、EPS、股息率。",
        messages=[]
    )
    print(result)

四、完整对话轨迹示例

以下是“示例 1”在实际运行中的完整消息流，展示了 Function Calling 的四步循环：

【用户输入】

"帮我查一下腾讯(700.HK)和苹果(AAPL.US)的最新股价和涨跌幅。"

【DeepSeek 返回的 tool_calls】

{
  "id": "chatcmpl-xxxxxxxxxxxxxxxxxxxx",
  "object": "chat.completion",
  "choices": [{
    "index": 0,
    "message": {
      "role": "assistant",
      "content": null,
      "tool_calls": [{
        "id": "call_xxxxxxxxxxxxxxxxxxxx",
        "type": "function",
        "function": {
          "name": "get_ticker",
          "arguments": "{\"symbols\":\"700.HK,AAPL.US\"}"
        }
      }]
    },
    "finish_reason": "tool_calls"
  }]
}

【工具返回结果】（TickDB REST API 真实返回结构）

{
  "role": "tool",
  "tool_call_id": "call_xxxxxxxxxxxxxxxxxxxx",
  "content": "[{\"symbol\":\"700.HK\",\"last_price\":\"XXX\",\"price_change_percent_24h\":\"X.XX%\",\"volume_24h\":\"XXXXXX\",\"high_24h\":\"XXX\",\"low_24h\":\"XXX\"},{\"symbol\":\"AAPL.US\",\"last_price\":\"XXX\",\"price_change_percent_24h\":\"X.XX%\",\"volume_24h\":\"XXXXXX\",\"high_24h\":\"XXX\",\"low_24h\":\"XXX\"}]"
}

【DeepSeek 最终回复】

腾讯控股(700.HK)最新价：XXX 港元，涨跌幅：X.XX%
苹果(AAPL.US)最新价：XXX 美元，涨跌幅：X.XX%
📡 数据由 TickDB.ai 提供

这个完整轨迹清晰展示了 Function Calling 的消息流：

1. user → 用户提问
2. assistant(tool_calls) → 模型请求调用 get_ticker，参数 symbols="700.HK,AAPL.US"
3. tool → 代码执行 TickDB REST API 调用，返回结构化行情数据
4. assistant(text) → 模型基于真实数据生成最终回复

五、Strict Function Calling（进阶）

DeepSeek 当前在 Beta API 中支持 Strict Function Calling，强制模型的 JSON 输出格式严格匹配 Function 的 JSON Schema。

为什么要用 Strict 模式？ 普通模式下，模型偶尔会在 tool_calls 的参数里漏字段、多字段、格式与 Schema 不一致。Strict 模式从服务端层面强制校验，减少这种不确定性。

启用方式：

# 使用 Beta endpoint
client = OpenAI(
    api_key=DEEPSEEK_API_KEY,
    base_url="https://api.deepseek.com/beta",  # Beta 地址
)

Strict 模式的关键规则（必须遵守，否则服务端拒绝）：

1. additionalProperties 必须设为 False。
2. 所有在 properties 中定义的字段，都必须出现在 required 数组中。
3. Schema 中不能使用服务端不支持的类型（如 oneOf、anyOf），需参考官方文档确认支持列表。

这意味着本文普通模式下的 get_kline 设计（limit 在 properties 但不在 required）不能直接加 "strict": True。Strict 模式下有两种改法：

改法 A：把 limit 加入 required。

"required": ["symbol", "interval", "limit"],

模型每次都会传 limit，即使传 20（默认值）也没问题。

改法 B：从 schema 的 properties 中移除 limit，在执行函数内硬编码默认值。

def execute_get_kline(symbol: str, interval: str, limit: int = 20):
    # limit 始终有默认值，不依赖模型传参

两种改法根据实际需求选择。Strict 模式输出更可靠，但需要 Schema 完全规范。建议先用普通模式验证设计合理后，再迁移到 Strict 模式。

六、多轮对话进阶：让 DeepSeek 自己规划任务

单次工具调用只是入门。Function Calling 的真正威力在于多轮任务规划。同一个对话窗口内，模型可以连续调用多个不同工具。

示例场景：

用户：帮我分析一下腾讯（700.HK）。先查实时行情，再拉最近 10 根日 K 线，最后查一下它的基本面（EPS、股息率、每手股数）。

DeepSeek 的规划过程：

整个过程，你不需要告诉它“应该先查谁、再查谁”——模型自己判断调用顺序和数据依赖关系。

Thinking Mode + Tool Calls 风险提示：如果你同时启用了 DeepSeek 的 Thinking Mode，当发生 tool call 后，后续请求需要正确保留相关的 assistant message。官方文档对此有 400 错误的风险提示，建议查阅 DeepSeek Tool Calls 文档 确认消息格式要求。

七、错误处理与工程化建议

常见错误处理（已在代码中实现）：

工程化建议：

1. tools 定义与执行函数分离：tools 定义放在 JSON 文件中，执行函数放在独立模块。新增工具时改配置不加代码。

1. tool_call_id 精确关联：每个 tool_call 都有唯一 ID，返回结果时必须带上 tool_call_id。

1. 对话历史长度控制：多轮对话会导致 messages 超出模型上下文限制。建议保留最近 20 轮，早期消息做摘要压缩。

八、它能做什么，不能做什么

✅ 适合的场景：

❌ 不推荐使用的场景：

九、常见问题

Q1：模型不调用工具，直接回复文本怎么办？

检查三点：

• tools 参数是否正确传入 client.chat.completions.create()
• tool 的 description 是否清晰描述了“何时该用这个工具”
• 如果仍然不触发，可以尝试 tool_choice="required" 强制模型调用工具

Q2：调用工具后返回的数据太长，模型“记不住”怎么办？

在工具执行函数中截取关键字段返回。例如 get_kline 只保留最近 N 根 K 线的 OHLC，get_stock_info 只保留核心财务指标。

Q3：Strict 模式和普通模式选哪个？

Strict 模式输出更可靠，但需要 Schema 完全规范（所有 properties 必须 required）。建议先用普通模式验证设计合理后，再迁移到 Strict 模式。切换时注意 base_url 改成 Beta 地址。

Q4：DeepSeek Function Calling 和 ChatGPT Function Calling 有什么区别？

两者都兼容 OpenAI SDK 格式，tools 参数结构相似。DeepSeek 额外支持 Strict Function Calling 模式和 Thinking Mode 下的工具调用。如果你已有 OpenAI Function Calling 的开发经验，切换到 DeepSeek 几乎不需要改代码结构——只需要改 base_url 和 api_key。

收尾

回到开头那个问题：当真实数据喂给 DeepSeek 后，它到底在“分析”还是在“编故事”？

跑完这套代码，你就有了答案。Function Calling 不是魔法，而是一套精确的工程规范：你定义工具，模型规划调用，你执行查询，模型基于真实数据生成回复。这个流程里的每一个环节——数据从哪里来、工具怎么定义、错误怎么处理——都是透明的、可替换的、可优化的。

TickDB 提供了这套数据通道：通过 Skill、MCP、REST API、CLI 等多组入口，覆盖 4 大市场、40,000+ 品种，实现统一字段、统一鉴权。配合 DeepSeek Function Calling，你得到的不只是一个查价工具，而是一个可复用的 AI Agent 行情查询框架。

📡 本文行情数据服务由 TickDB.ai 提供。GitHub 开源，文档可查，代码可跑。

本文仅讨论技术接入和工具配置方式，不构成任何投资建议。文中所有代码和示例输出仅用于教学目的，不构成交易信号或投资策略推荐。

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

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