通义千问 Function Calling 接金融实时数据：别用查天气的思路写金融工具描述

你在通义千问开发者后台定义了两个工具，一个查实时行情，一个查历史 K 线。description 都写了“获取市场数据”。Agent 上线第一天，用户问“茅台现在多少钱”，模型调了 K 线工具，返回了昨天的收盘价——因为两个描述在模型眼里没有区别。

!image.png

一、当通用教程遇上金融数据

阿里云 DashScope 的 Function Calling 文档里，示例通常是“获取指定城市的天气”——一个工具，两个参数，查询成功或失败都有明确的话术兜底。但当你把同样的模式直接搬到金融行情上，事情就不一样了。

!image.png

假设你要让通义千问能回答用户“查一下茅台和腾讯的最新价格”。你写好了一个 get_ticker 函数，在 DashScope 的 tools 参数里注册，然后部署上线。很快你会发现几个通用教程没告诉你的问题：

这些问题不是某个数据源独有的——任何金融数据接入都会遇到。本文以 TickDB 作为示例行情后端，拆解通义千问 Function Calling 在金融场景下需要的额外工程约束。即使你换成其他行情 API，这些约束依然存在，只是字段名和错误码不同。

二、工具调用的背后：不是模型不够聪明，是任务太模糊

Function Calling 的本质，是让模型根据用户意图，从你提供的工具列表里选出合适的那个，并生成调用参数。在“查天气”场景下，这套机制很好用，因为城市名参数基本不会出错。但金融场景有三个让复杂度陡升的源头：

1. 工具之间没有天然边界

你很可能同时需要这些数据能力：

• 实时快照（ticker）
• 历史 K 线（kline）
• 最新 K 线（kline/latest）
• 盘口深度（depth）
• 成交明细（trades）

如果每个工具的 description 都写成“获取市场数据”或“查询某品种的行情”，模型就会在它们之间随机选择，因为它分不清“快照”和“已完成 K 线”的区别。你必须主动在描述中划出“不要用 X，应使用 Y”的排他性边界。

2. 调用结果可能包含业务级错误

通用教程里，函数执行只会有两种结果：成功返回数据，或者抛异常。但在真实金融接口中，你的 get_ticker 可能遇到：

• 限流（3001，需要等待 Retry-After 再重试）
• API Key 无效（1001）
• 权限不足（1004）
• 返回空数据（symbol 错或市场休市）

如果这些错误被原样丢回模型，模型可能会把错误信息当成“数据”来处理，比如从“查询超时”推断出“价格可能在 300 左右”。因此，函数返回给模型的内容必须结构化，明确区分 success/fail 状态。

3. 参数在用户嘴里不是代码格式

用户说“茅台”，你的工具需要的是 600519.SH。用户说“苹果”，你可能要区分是 AAPL.US 还是 AAPL 的期权链。通用教程不会告诉你：当用户输入和工具参数之间存在格式鸿沟时，模型的 system prompt 里必须写清楚映射规则。

时间戳真相：经审核工具独立核验的结果

!image.png

以下是 2026-05-29 由 TickDB MCP 工具直接调用核验 的 timestamp 实际返回情况：

⚠️ 文档与实际行为差异：

>

TickDB REST API 官方文档“数据规范”中写道：“所有接口返回的时间戳字段均为毫秒(ms)”。

>

但 MCP 工具实测发现：get_recent_trades 接口对股票类型（AAPL.US 美股、700.HK 港股）实际返回 10 位秒级时间戳，仅加密货币（BTCUSDT）返回 13 位毫秒时间戳。 这一行为与文档描述不一致。

>

这意味着：在处理 trades 数据时，不能假设所有品种的时间戳单位一致。如果你按文档写“全局除 1000”，股票 trades 的时间会被错误转换成 1970 年。

所以时间戳的处理规则不能一刀切：同一个 timestamp 字段名，在不同接口（ticker vs trades）、同一接口的不同品种（股票 vs 加密货币）上，粒度可能不同。使用时需逐接口、逐品种核验。 后面代码中我们会在字段旁加注释，而不是写死一个“全局除 1000”。

三、完整代码：从工具定义到 FC 部署入口

以下代码用 DashScope SDK 调用通义千问，通过 Function Calling 接入 TickDB 行情数据，包含限流退避、Decimal 类型保护、错误状态标记、中文名称提示和阿里云 FC 入口。代码含异常处理的完整可运行示例，本地和函数计算均可，只需替换环境变量中的 API Key。

⚠️ DashScope SDK 调用约定提示：以下代码中 Generation.call 的参数签名、tool_choice 可选值、tool_calls 的返回结构以及 role: "tool" 的回填格式，均以 DashScope 当前官方文档为准。建议在正式接入前核验你所使用的 SDK 版本的实际行为。

环境准备

pip install dashscope==1.20.0 requests python-dotenv

在项目根目录创建 .env 文件，不要提交到版本控制：

DASHSCOPE_API_KEY=你的通义千问APIKey
TICKDB_API_KEY=你的TickDB API Key

主代码：tongyi_tickdb.py

# tongyi_tickdb.py
# 通义千问 Function Calling 接入 TickDB 行情数据
# 部署环境：阿里云函数计算 Python 3.11 / 本地 Python 3.10+

import os
import json
import time
import requests
from decimal import Decimal, InvalidOperation
from dotenv import load_dotenv
from dashscope import Generation
from dashscope.api_entities.dashscope_response import GenerationResponse

load_dotenv()

# ================== 配置区 ==================
DASHSCOPE_API_KEY = os.getenv("DASHSCOPE_API_KEY")
TICKDB_API_KEY = os.getenv("TICKDB_API_KEY")
TICKDB_REST_URL = "https://api.tickdb.ai"
MAX_RETRIES = 3

# ================== 数据获取层：TickDB REST 封装 ==================
def get_ticker(symbols_str: str, retry_count: int = 0) -> dict:
    """
    获取实时行情快照。
    不要使用此函数获取历史K线——历史K线应使用 get_kline。
    
    返回统一结构：
        {"success": True, "data": [...], "error_code": ""}
        {"success": False, "data": [], "error_code": "RATE_LIMITED|...}
    """
    if retry_count > MAX_RETRIES:
        return {"success": False, "data": [], "error_code": "MAX_RETRIES_EXCEEDED"}

    headers = {"X-API-Key": TICKDB_API_KEY}
    params = {"symbols": symbols_str}

    try:
        resp = requests.get(
            f"{TICKDB_REST_URL}/v1/market/ticker",
            headers=headers,
            params=params,
            timeout=10
        )
    except requests.exceptions.Timeout:
        return {"success": False, "data": [], "error_code": "TIMEOUT"}
    except requests.exceptions.ConnectionError:
        return {"success": False, "data": [], "error_code": "CONNECTION_ERROR"}

    data = resp.json()

    # 限流处理：读取 Retry-After，保护非整数情况
    if resp.status_code == 429 or data.get("code") == 3001:
        retry_after = resp.headers.get("Retry-After", "5")
        try:
            wait_seconds = float(retry_after)
        except (ValueError, TypeError):
            wait_seconds = 5
        time.sleep(wait_seconds)
        return get_ticker(symbols_str, retry_count + 1)

    # 鉴权错误码（阻断，不重试）
    if data.get("code") == 1001:
        return {"success": False, "data": [], "error_code": "INVALID_API_KEY"}
    if data.get("code") == 1002:
        return {"success": False, "data": [], "error_code": "MISSING_API_KEY"}
    if data.get("code") == 1004:
        return {"success": False, "data": [], "error_code": "PERMISSION_DENIED"}

    if data.get("code") != 0:
        return {"success": False, "data": [], "error_code": f"API_ERROR_{data.get('code')}"}

    # 字段映射与类型保护：volume_24h 可能是浮点数字符串，必须用 Decimal
    results = []
    for d in data.get("data", []):
        try:
            price = Decimal(str(d["last_price"]))
            vol = Decimal(str(d.get("volume_24h", "0")))
            ts = d["timestamp"]
            # ticker 接口的 timestamp 经 MCP 实测为 13 位毫秒 UTC。
            # 注意：get_recent_trades 接口中，股票（美股/港股）为 10 位秒级，
            # 加密货币为 13 位毫秒。时间戳单位需逐接口、逐品种核验，不可一刀切。
        except (InvalidOperation, ValueError):
            continue  # 跳过解析异常的数据条目

        results.append({
            "symbol": d["symbol"],
            "last_price": str(price),
            "volume_24h": str(vol),
            "timestamp_ms": ts,
            "timestamp_note": "毫秒UTC (ticker接口，已核验)"
        })

    if not results:
        return {"success": False, "data": [], "error_code": "EMPTY_DATA"}

    return {"success": True, "data": results, "error_code": ""}


# ================== 通义千问工具定义 ==================
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_ticker",
            "description": (
                "获取品种实时行情快照，包含最新价、24小时成交量、毫秒UTC时间戳。"
                "不要使用此函数获取历史K线数据——历史K线应使用 get_kline 函数。"
                "不要使用此函数获取盘口深度——盘口深度应使用 get_order_book 函数。"
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "symbols": {
                        "type": "string",
                        "description": (
                            "逗号分隔的品种代码，例如 '600519.SH,700.HK,AAPL.US'。"
                            "A 股格式为代码.SH/SZ/BJ，港股为代码.HK（无前导零），美股为代码.US。"
                            "用户输入中文名称时，请先告知正确代码格式，不要自行映射或猜测。"
                        )
                    }
                },
                "required": ["symbols"]
            }
        }
    }
]

SYSTEM_PROMPT = """你是一个金融数据助手。你只能通过 get_ticker 工具获取实时行情数据。

严格遵循以下规则：
1. 工具返回 success=true 时，基于 data 中的结构化数据回答用户问题。
2. 工具返回 success=false 时，直接告知用户“当前无法获取行情数据，错误原因：<error_code>”，不要猜测或编造任何数值。
3. 用户输入中文名称（如“茅台”）时，不要尝试转换为代码，而是回复：“请提供品种代码，例如 600519.SH（贵州茅台）、700.HK（腾讯控股）、AAPL.US（苹果）。”
4. 所有价格、成交量、时间戳必须来自工具返回结果，绝不使用训练数据中的记忆值。"""


# ================== Function Calling 主流程 ==================
def call_tongyi_with_fc(user_query: str) -> str:
    """调用通义千问 Function Calling，处理一次用户查询"""
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": user_query}
    ]

    # 第一次调用：模型决定是否使用工具
    response = Generation.call(
        api_key=DASHSCOPE_API_KEY,
        model="qwen-max",
        messages=messages,
        tools=TOOLS,
        tool_choice="auto",
        result_format="message"
    )

    if response.status_code != 200:
        return f"模型调用失败: {response.code} - {response.message}"

    assistant_output = response.output.choices[0].message

    # 如果模型不需要工具，直接返回回复
    if not hasattr(assistant_output, "tool_calls") or not assistant_output.tool_calls:
        return assistant_output.content or "模型未返回有效响应"

    # 处理工具调用
    messages.append(assistant_output)

    for tool_call in assistant_output.tool_calls:
        if tool_call.function.name == "get_ticker":
            try:
                args = json.loads(tool_call.function.arguments)
            except json.JSONDecodeError:
                tool_result = {"success": False, "data": [], "error_code": "INVALID_ARGUMENTS"}
            else:
                tool_result = get_ticker(args.get("symbols", ""))

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

    # 第二次调用：模型基于工具结果生成最终回复
    final_response = Generation.call(
        api_key=DASHSCOPE_API_KEY,
        model="qwen-max",
        messages=messages,
        result_format="message"
    )

    if final_response.status_code == 200:
        return final_response.output.choices[0].message.content
    else:
        return f"模型生成回复失败: {final_response.code}"


# ================== 阿里云函数计算入口（部署时取消注释） ==================
# def handler(event, context):
#     """FC 事件处理函数。触发时从 event 中提取 query 参数。"""
#     body = json.loads(event)
#     user_query = body.get("query", "")
#     if not user_query:
#         return {"statusCode": 400, "body": "缺少 query 参数"}
#     answer = call_tongyi_with_fc(user_query)
#     return {"statusCode": 200, "body": answer}


# ================== 本地测试入口 ==================
if __name__ == "__main__":
    print("=== 测试1: 标准英文代码查询 ===")
    print(call_tongyi_with_fc("查一下 600519.SH 和 AAPL.US 的实时价格"))
    print("\n=== 测试2: 中文名称输入（应触发格式引导，不编造数据） ===")
    print(call_tongyi_with_fc("茅台现在多少钱"))

核心解读

• 工具描述：description 开头直接写“不要使用此函数获取历史K线”，划定排他性边界，避免模型在多个行情工具间随机选择。
• 返回结构：函数始终返回 {"success": bool, "data": [], "error_code": ""}，而不是把原始 JSON 或异常直接抛给模型。error_code 为后续监控和告警提供了机器可读标签。
• 限流退避：精确解析 Retry-After，保护非整数情况；最大重试 3 次后返回 MAX_RETRIES_EXCEEDED，防止死循环。
• 字段类型：成交量 volume_24h 用 Decimal(str(...)) 解析，避免浮点数字符串直接转整数而抛错（加密货币常见精度）。
• 时间戳注释：代码中对 ticker 接口的 timestamp 加了接口来源注释，并注明“trades 接口中股票为秒级、加密货币为毫秒”，防止后续维护者一刀切处理。

四、生产落地的最后一个思考：你部署在哪里

写完本地代码只是第一步。通义千问 Function Calling 的生产环境通常有两种部署选择，决策取决于你对冷启动、网络策略和管理复杂度的承受力。

如果选择 FC，有两个容易忽略的细节：

1. 函数超时设置：要大于 TickDB REST API 的超时（10 秒）+ 两次 DashScope 调用的耗时。建议设为 30 秒以上。
2. 冷启动优化：把 TOOLS、SYSTEM_PROMPT 等大对象放在函数全局作用域，不要放在 handler 函数内，这样实例复用时可以减少初始化开销。

五、结尾

给通义千问接上行情数据，步骤上只是定义了一个 tools 数组。但让这个工具在真实金融场景下不犯错，靠的是：

• 描述里的“不要用 X，应选 Y”
• 返回结构里的 success 字段
• 时间戳旁边的接口注释——以及一个重要的实测发现：TickDB 的 get_recent_trades 接口对股票返回秒级时间戳、对加密货币返回毫秒，这与官方文档“所有时间戳均为毫秒”的表述不一致。 这个细节如果在代码里被忽略，你的 Agent 在分析成交数据时会把 2026 年的交易对到 1970 年。

一个反直觉的观察：模型能力越强，工具描述越需要写得“啰嗦”——因为强模型能从模糊描述中推断出多种可能，而金融数据场景下，推断就是风险。

你在接入通义千问的时候，有没有遇到过模型“自作主张”的情况？欢迎在评论区聊聊你踩过的坑。

📡 数据示例由 TickDB.ai 提供

标签：通义千问 / Function Calling / DashScope / 金融数据 / 阿里云函数计算 / TickDB

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

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