思考模式、thinking_budget、JSON Schema 与多轮工具调用

这一课做完你会得到什么

四个可以直接运行的 Python 脚本，分别用来分开打印思考过程和答案、限制思考长度、拿到严格符合 Schema 的 JSON，以及跑通一个完整的工具调用循环。代码沿用第 1 课的 DASHSCOPE_API_KEY 和 WORKSPACE_ID 两个环境变量。

先记住思考模式的三条规则

• qwen3.8-max、qwen3.8-flash 属于混合思考模式，默认开启思考。用 enable_thinking 控制开关：true 时先思考再回答，false 时直接回答。
• enable_thinking 不是 OpenAI 标准参数。Python SDK 要放进 extra_body，Node.js SDK 和 curl 直接写在请求体顶层。
• 思考内容在 reasoning_content 字段里返回，思维链 Token 按输出计费，数量记在 usage.completion_tokens_details.reasoning_tokens。

第 1 步：流式输出，分开打印思考过程和答案

新建 stream_thinking.py：

import os
from openai import OpenAI

ws = os.getenv("WORKSPACE_ID")
client = OpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url=f"https://{ws}.cn-beijing.maas.aliyuncs.com/compatible-mode/v1",
)

completion = client.chat.completions.create(
    model="qwen3.8-max",
    messages=[{"role": "user", "content": "你是谁"}],
    extra_body={"enable_thinking": True},
    stream=True,
    stream_options={"include_usage": True},
)

is_answering = False
print("=" * 20 + "思考过程" + "=" * 20)
for chunk in completion:
    if not chunk.choices:
        print("\nUsage:", chunk.usage)
        continue
    delta = chunk.choices[0].delta
    if getattr(delta, "reasoning_content", None) and not is_answering:
        print(delta.reasoning_content, end="", flush=True)
    if delta.content:
        if not is_answering:
            print("\n" + "=" * 20 + "完整回复" + "=" * 20)
            is_answering = True
        print(delta.content, end="", flush=True)

运行 python3 stream_thinking.py。对照结果的形状（官方示例节选，具体文字每次不同）：

====================思考过程====================
好的，用户问“你是谁”，我需要给出一个准确且友好的回答……
====================完整回复====================
你好！我是千问，是阿里巴巴集团旗下的通义实验室自主研发的超大规模语言模型……
Usage: CompletionUsage(completion_tokens=..., prompt_tokens=10, ...
  completion_tokens_details=CompletionTokensDetails(... reasoning_tokens=...))

reasoning_tokens 是 completion_tokens 的一部分。官方示例中，一次问答的 360 个输出 Token 里有 286 个是思考内容，所以简单问题建议关掉思考。注意：stream_options 只能在流式请求里设置，非流式请求不要带。

第 2 步：关闭思考，或限制思考长度

不需要推理的任务直接关掉思考，把 extra_body 改成：

extra_body={"enable_thinking": False}

需要思考、但不想让它想太久时，用 thinking_budget 设置思维链的最大 Token 数。超过上限后，模型会立即开始输出回复：

extra_body={"enable_thinking": True, "thinking_budget": 50}

在第 1 步的脚本里换上这一行再跑一次，reasoning_tokens 应不超过 50（官方示例中正好是 50）。thinking_budget 的默认值是模型的最大思维链长度，可以在百炼控制台的模型卡片里查到。

多轮对话时，如果希望模型读到上一轮自己的思考过程，可以在 extra_body 里加上 "preserve_thinking": True，并在 messages 的历史 assistant 消息中带上 reasoning_content。qwen3.8-max 支持这个参数。打开后，历史思考内容会计入输入 Token 并计费。

第 3 步：JSON Object 模式

做信息抽取时，把 response_format 设为 {"type": "json_object"}，模型就会返回标准 JSON 字符串，不会带上多余的 Markdown 代码块。新建 json_object.py：

import os
from openai import OpenAI

ws = os.getenv("WORKSPACE_ID")
client = OpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url=f"https://{ws}.cn-beijing.maas.aliyuncs.com/compatible-mode/v1",
)

completion = client.chat.completions.create(
    model="qwen3.8-max",
    messages=[
        {"role": "system", "content": "请抽取用户的姓名与年龄信息，以JSON格式返回"},
        {"role": "user", "content": "大家好，我叫刘五，今年34岁，邮箱是liuwu@example.com"},
    ],
    response_format={"type": "json_object"},
)
print(completion.choices[0].message.content)

对照结果（官方示例）：

{
  "姓名": "刘五",
  "年龄": 34
}

这个模式有两个坑：一是 System 或 User 消息里必须出现 "JSON" 这个词（大小写不限），否则会报错 'messages' must contain the word 'json' in some form；二是键名和字段类型不保证稳定，换一种问法可能就变成 name / age。

第 4 步：JSON Schema 模式，结构固定不漂移

需要稳定的结构时，用 OpenAI SDK 的 parse 方法，直接传入 Pydantic 类，SDK 会自动转成 json_schema。官方列出支持 JSON Schema 的有 Qwen3.8-Max、Qwen3.8-Flash、Qwen3.7-Max、Qwen3.7-Plus、Qwen3.7-Flash 系列。

pip install -U openai pydantic

新建 json_schema.py：

import os
from openai import OpenAI
from pydantic import BaseModel, Field

ws = os.getenv("WORKSPACE_ID")
client = OpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url=f"https://{ws}.cn-beijing.maas.aliyuncs.com/compatible-mode/v1",
)


class UserInfo(BaseModel):
    name: str = Field(description="用户的姓名")
    age: int = Field(description="用户的年龄，单位为岁")


completion = client.chat.completions.parse(
    model="qwen3.8-max",
    messages=[
        {"role": "system", "content": "提取姓名与年龄信息。"},
        {"role": "user", "content": "我叫刘五，今年25岁。"},
    ],
    response_format=UserInfo,
)
result = completion.choices[0].message.parsed
print(f"姓名：{result.name}，年龄：{result.age}")

对照结果：

姓名：刘五，年龄：25

不用 Pydantic 时，也可以手写 response_format，形如 {"type": "json_schema", "json_schema": {"name": "user_info", "strict": true, "schema": {...}}}。官方建议把 strict 设为 true，并在 schema 里加上 "additionalProperties": false。注意：输入里含图像、视频、音频时，json_schema 会自动降级为 json_object，schema 约束不生效。

第 5 步：写一个完整的工具调用循环

Function Calling 的流程是：把问题和工具清单一起发给模型，模型返回要调用的函数名和参数，你在本地执行函数，把结果作为 role: "tool" 消息追加进去，再调一次模型拿到最终回答。新建 tool_loop.py：

import json
import os
import random
from openai import OpenAI

ws = os.getenv("WORKSPACE_ID")
client = OpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url=f"https://{ws}.cn-beijing.maas.aliyuncs.com/compatible-mode/v1",
)

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 = random.choice(["晴天", "多云", "雨天"])
    return f"{arguments['location']}今天是{weather}。"


def get_response(messages):
    return client.chat.completions.create(
        model="qwen3.8-max",
        extra_body={"enable_thinking": False},
        messages=messages,
        tools=tools,
    )


messages = [{"role": "user", "content": "北京天气咋样"}]
assistant_output = get_response(messages).choices[0].message
if assistant_output.content is None:
    assistant_output.content = ""
messages.append(assistant_output)

while assistant_output.tool_calls:
    # 一轮里可能返回多个 tool_calls，要全部处理
    for tool_call in assistant_output.tool_calls:
        arguments = json.loads(tool_call.function.arguments)
        print(f"正在调用工具 [{tool_call.function.name}]，参数：{arguments}")
        tool_result = get_current_weather(arguments)
        print(f"工具返回：{tool_result}")
        messages.append(
            {"role": "tool", "tool_call_id": tool_call.id, "content": tool_result}
        )
    assistant_output = get_response(messages).choices[0].message
    if assistant_output.content is None:
        assistant_output.content = ""
    messages.append(assistant_output)

print(f"助手最终回复：{assistant_output.content}")

对照结果（官方示例；天气是脚本随机生成的）：

正在调用工具 [get_current_weather]，参数：{'location': '北京'}
工具返回：北京今天是多云。
助手最终回复：北京今天是多云的天气。

三个要点：

• 执行工具的是你的程序，不是模型。模型只负责决定调哪个工具、传什么参数。
• 每轮都要把模型返回的完整 assistant 消息追加进 messages，再追加 tool 消息，tool_call_id 要一一对应。
• 一轮可能返回多个 tool_calls，要用 for 循环全部处理，不能只取第一个。

自查清单

• stream_thinking.py 先打印思考过程、再打印回复，最后输出 reasoning_tokens
• 设置 thinking_budget: 50 后，reasoning_tokens 不超过 50
• json_object.py 的提示词里含有 "JSON"，输出可以被 json.loads 解析
• json_schema.py 打印出「姓名：刘五，年龄：25」
• tool_loop.py 先打印工具调用，再给出最终回复