FastAPI + OpenAI 兼容协议 + DeepSeek 实战：大模型 Function Calling 工具调用全拆解

haokeizheng

624人浏览 · 2026-08-05 19:40:13

haokeizheng · 2026-08-05 19:40:13 发布

FastAPI + OpenAI 兼容协议 + DeepSeek 实战：大模型 Function Calling 工具调用全拆解

功能概览

功能	核心内容
天气查询	工具声明 `tools`、单轮发起、解析 `tool_calls`（打印函数名+参数，不执行）
学历查询	多工具声明、`get_xueli` 外部调用 + Redis 缓存、工具结果回灌第二轮合成

调用链路总览

用户问题 → messages + tools 发给模型
  → 模型返回 tool_calls（函数名 + JSON 参数）
  → 本地执行对应函数（天气/学历）
  → 把 函数结果 以 role=tool 追加回 messages
  → 再次请求模型 → 模型生成最终自然语言回答

一、环境准备：OpenAI 依赖下载与配置（单列）

工具调用依赖 OpenAI 官方 SDK，配置单独拎出来讲，不与业务代码混在一起。

1.1 依赖下载

pip install openai

代码顶部都是 from openai import OpenAI，装好这个包即可。redis 与 requests 是学历查询里缓存和调用外部 API 用到的，按需安装：pip install redis requests。

1.2 密钥配置（环境变量）

raw_key = os.getenv("DASHSCOPE_API_KEY")
api_key = raw_key.strip()

代码通过 DASHSCOPE_API_KEY 读取阿里云百炼的 API Key，需提前在系统或 .env 中导出该变量。

1.3 base_url 兼容端点

client = OpenAI(
    api_key=api_key,
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",)

关键是 base_url 必须带 /compatible-mode/v1 后缀——百炼把 OpenAI 协议做成了兼容模式，漏写这一截会直接 404。

二、知识点讲解

2.1 什么是 Function Calling

模型本身不能直接查天气、查数据库。我们告诉模型"你有这些函数可用"，模型在觉得需要时返回"我想调用某个函数、参数是这样"，真正的函数由我们本地执行，再把结果交还给模型整理成话。这就是"模型决策 + 本地执行"的混合模式。

2.2 tools 工具声明结构

每个工具是一个 JSON-Schema：type: function + function.name（函数名）+ function.description（给模型看的功能描述）+ function.parameters（入参的 JSON Schema）。description 写得好不好，直接决定模型会不会在该调用时调用。

2.3 tool_calls 是什么

模型认为需要调用工具时，返回的消息里 message.tool_calls 不为 None，里面是列表，每项含 function.name（函数名）、function.arguments（JSON 字符串参数）、id（该次调用的标识，回灌结果时必须带上）。

2.4 单轮识别 vs 完整闭环

单轮：拿到 tool_calls 就结束，仅解析函数名/参数；
闭环：执行函数 → 把结果以 role=tool 追加 → 再请求一次模型，由模型产出最终回答。少了"回灌"这步，用户永远看不到自然语言结果。

三、代码逻辑拆解

3.1 客户端初始化与模型选择

两个功能的顶部完全一致：

raw_key = os.getenv("DASHSCOPE_API_KEY")
api_key = raw_key.strip()
client = OpenAI(
    api_key=api_key,
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",)

第 1 行：从环境变量取密钥原始值；
第 2 行：.strip() 去掉首尾空白，避免复制粘贴带换行导致鉴权失败；
第 3–5 行：构造 OpenAI 客户端，base_url 指向百炼兼容端点。模型名不在客户端里写，而是在每次 create 时指定（见 3.3）。

3.2 第一个功能：天气查询

3.2.1 工具声明与本地函数

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "当你想查询指定城市的天气时非常有用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "城市或县区，比如北京市、杭州市、余杭区等。",
                    }
                },
                "required": ["location"],
            },
        },
    },
]

def get_current_weather(arguments):
    weather_conditions = ["晴天", "多云", "雨天"]
    random_weather = random.choice(weather_conditions)
    location = arguments["location"]
    return f"{location}今天是{random_weather}。"

name：函数标识，必须与本地真实函数同名，模型返回时原样带回；
description：模型的"使用说明书"，决定何时触发；
parameters.properties.location：入参 schema，required 声明该参数必填，模型被要求必须产出它；
get_current_weather 是本地实现：从参数里取 location，随机返回一个天气（演示用，真实场景替换为气象 API）。

3.2.2 发起请求与单轮解析

def get_response(messages):
    completion = client.chat.completions.create(
        model="deepseek-v4-flash-0731",
        messages=messages,
        tools=tools,
    )
    return completion

user_message = [{"role": "user", "content": "你是谁"}]
messages.extend(user_message)
completion = get_response(messages)
messages.append(completion.choices[0].message)
if completion.choices[0].message.tool_calls is None:
    print(f"无需调用工具，直接回复：{completion.choices[0].message.content}")
else:
    print("需要调用工具:")
    tool_calls = completion.choices[0].message.tool_calls
    for i in tool_calls:
        f_name = i.function.name
        f_arg = i.function.arguments
        print(f"调用工具是：{f_name}，参数：{f_arg}")

model="deepseek-v4-flash-0731"：具体模型名，写在每次请求里；
tools=tools：把工具清单一并提交，模型才会返回 tool_calls；
messages.extend：把用户问题追加进对话列表（模块级 messages = []）；
completion.choices[0].message：模型原始消息，先整体 append 进 messages，保证多轮上下文连续；
if ... tool_calls is None：判断是否触发工具——None 说明模型直接回答了，否则进入工具分支；
工具分支里 i.function.name 取函数名、i.function.arguments 取参数字符串（是字符串不是字典，要用 json.loads 解析）；
注意：这一段到这里只 print，没真正执行函数、也没回灌，所以它只是"识别"演示。

3.3 第二个功能：学历查询

3.3.1 Redis 客户端与外部学历查询工具

r = Redis(host='127.0.0.1', port=6379, db=11, decode_responses=True)

def get_xueli(arguments):
    vcode = arguments["vcode"]
    key = f"boos:llm:academic_credential_verification:{vcode}"
    redis_vreif = r.get(key)
    if redis_vreif is None:
        API_KEY = "MY_KEY_le0KRXAsCh6cNphDEURqJCs02jt3x1"
        BASE_URL = "https://www.apimy.cn/api/xxw/bgcx"
        params = {"key": API_KEY, "vcode": arguments["vcode"]}
        headers = {"Content-Type": "application/json"}
        response = requests.get(BASE_URL, params=params, timeout=30)
        response.raise_for_status()
        data = response.json()
        r.set(key, json.dumps(data, ensure_ascii=False))
        return json.dumps(data, ensure_ascii=False)
    else:
        return redis_vreif

第 1–4 行：Redis 客户端，db=11 与多轮对话的 db=10 区分开，decode_responses=True 让取出的 value 直接是字符串；
key：用学历验证码拼键，同一验证码只查一次；
if redis_vreif is None：缓存未命中才真打外部 API，命中直接返回，省额度；
requests.get(..., timeout=30)：timeout 必带，外部接口抽风时不会把连接挂死；
r.set(key, json.dumps(data))：把结果以 JSON 字符串写回 Redis；返回时 json.dumps(data) 与缓存写入格式保持一致，工具结果对模型而言都是字符串即可。

3.3.2 多工具声明

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "当你想查询指定城市的天气时非常有用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "城市或县区，比如北京市、杭州市、余杭区等。",
                    }
                },
                "required": ["location"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_xueli",
            "description": "当你想查询学历或验证学历时非常有用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "vcode": {
                        "type": "string",
                        "description": "学历验证码",
                    }
                },
                "required": ["vcode"],
            },
        },
    },
]

在天气工具之外还声明了 get_xueli 工具，结构一致，参数换成了 vcode（学历验证码），required: ["vcode"]。两个工具并列放在同一个 tools 列表里，模型可自由选其一或都用。

3.3.3 完整闭环：执行工具 + 结果回灌

user_message = [{"role": "user", "content": "帮我查下一下学历验证码是:你自己的学历吗"}]
messages.extend(user_message)
completion = get_response(messages)
messages.append(completion.choices[0].message)
if completion.choices[0].message.tool_calls is None:
    print(f"无需调用工具，直接回复：{completion.choices[0].message.content}")
else:
    print("需要调用工具:")
    tool_calls = completion.choices[0].message.tool_calls
    fun = {
        "get_current_weather": get_current_weather,
        "get_xueli": get_xueli
    }
    for i in tool_calls:
        f_name = i.function.name
        f_arg = i.function.arguments
        print(f"调用工具是：{f_name}，参数：{f_arg}")
        tool_result = fun[f_name](json.loads(f_arg))
        print(f"工具返回结果是：{tool_result}")
        # 把工具结果追加消息，再次请求模型生成自然文本
        messages.append({"role": "tool", "tool_call_id": i.id, "content": tool_result})
        final_resp = get_response(messages)
        print("模型整理后的文本回答：", final_resp.choices[0].message.content)

fun = {...}：函数名到本地函数对象的映射表，模型返回函数名后据此分发执行；
fun[f_name](json.loads(f_arg))：用 json.loads 把参数字符串还原成字典后调用；
messages.append({"role": "tool", ...})：这是闭环的关键——role 必须是 "tool"，tool_call_id 填模型给的 i.id，content 填工具返回值；
get_response(messages) 再次发起，此时模型已拿到工具结果，会生成最终自然语言回答。源码注释也点明"把工具结果追加消息，再次请求模型生成自然文本"就是回灌 + 二次请求。