推理强度、流式思考、JSON Schema 与多轮工具调用

这一课做完你会得到什么

四个脚本：一个对比三档推理强度，一个流式打印思考过程和答案，一个用严格 JSON Schema 拿结构化结果，一个能连续调用工具直到给出最终答案的 Agent 循环。全部复用上一课的 MOONSHOT_API_KEY。

K3 和 K2.x 的参数差别，先记住这三条

• kimi-k3 始终思考，不接受 thinking 参数，用请求顶层的 reasoning_effort 调深浅："low"、"high"、"max"，默认 "max"。从 K2.x 迁移时要把 thinking 配置删掉。
• temperature=1.0、top_p=0.95、n=1、presence_penalty=0、frequency_penalty=0 在 K3 上是固定值，官方建议不要显式传。
• 多轮对话和工具调用时，必须把接口返回的完整 assistant message 原样放回 messages，包括 reasoning_content 和 tool_calls，不能只留 content。

第 1 步：对比三档推理强度

新建 effort_demo.py：

import os
import time
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["MOONSHOT_API_KEY"],
    base_url="https://api.moonshot.cn/v1",
)

question = "请推导一下这个数列的通项公式：1, 4, 9, 25, 64, ..."

for effort in ["low", "high", "max"]:
    start = time.time()
    completion = client.chat.completions.create(
        model="kimi-k3",
        messages=[{"role": "user", "content": question}],
        reasoning_effort=effort,
    )
    usage = completion.usage
    print(f"[{effort}] {time.time() - start:.1f}s  output_tokens={usage.completion_tokens}")
    print(completion.choices[0].message.content[:120])
    print("-" * 40)

python3 effort_demo.py

对照结果：会打印三段，每段第一行是档位、耗时和输出 token 数，后面是答案开头。通常 low 最快、输出 token 最少，max 最慢、最多；具体数字每次都不一样。官方说明思考内容不能关闭，嫌思考太长就用 low。

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

流式响应里，推理增量在 delta.reasoning_content，答案增量在 delta.content，并且思考一定先于答案出现。OpenAI SDK 的类型里没有 reasoning_content 字段，要用 getattr 读。新建 stream_demo.py：

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["MOONSHOT_API_KEY"],
    base_url="https://api.moonshot.cn/v1",
)

stream = client.chat.completions.create(
    model="kimi-k3",
    messages=[{"role": "user", "content": "解释为什么天空是蓝色的。"}],
    reasoning_effort="low",
    stream=True,
)

thinking = False
for chunk in stream:
    if not chunk.choices:
        continue
    delta = chunk.choices[0].delta
    reasoning = getattr(delta, "reasoning_content", None)
    if reasoning:
        if not thinking:
            thinking = True
            print("===== 思考 =====")
        print(reasoning, end="", flush=True)
    if delta.content:
        if thinking:
            thinking = False
            print("\n===== 回答 =====")
        print(delta.content, end="", flush=True)
print()

对照结果的形状：

===== 思考 =====
（模型的推理过程，逐字流出）
===== 回答 =====
（关于瑞利散射的最终解释，逐字流出）

非流式请求超过 900 秒会被网关以 504 断开，长任务一律建议用流式。

第 3 步：多轮对话要原样回传 assistant message

K3 的保留式思考始终开启，第二轮请求里必须带上第一轮完整的 assistant message。正确写法是直接把 SDK 返回的 message 对象追加进去：

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["MOONSHOT_API_KEY"],
    base_url="https://api.moonshot.cn/v1",
)

messages = [{"role": "user", "content": "我叫李雷，帮我想一个 Python 项目名。"}]
first = client.chat.completions.create(model="kimi-k3", messages=messages, reasoning_effort="low")
messages.append(first.choices[0].message)  # 整个 message 放回去，含 reasoning_content

messages.append({"role": "user", "content": "我刚才说我叫什么？"})
second = client.chat.completions.create(model="kimi-k3", messages=messages, reasoning_effort="low")
print(second.choices[0].message.content)

对照结果：第二次回答里会提到「李雷」。常见的错误写法是只追加 {"role": "assistant", "content": first.choices[0].message.content}，那样会丢掉思考内容，官方明确要求不要这样做。

注意：历史里的思考内容会继续占上下文并按 token 计费，对话越长越要留意 usage.prompt_tokens。

第 4 步：用 JSON Schema 拿到严格的结构化结果

response_format 有两种：json_object 只保证是合法 JSON；json_schema 加 strict: true 会强制按你给的字段输出。下游程序要消费数据时，官方建议始终用后者。新建 schema_demo.py：

import json
import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["MOONSHOT_API_KEY"],
    base_url="https://api.moonshot.cn/v1",
)

completion = client.chat.completions.create(
    model="kimi-k3",
    reasoning_effort="low",
    messages=[{"role": "user", "content": "小林今年 28 岁。提取姓名和年龄。"}],
    response_format={
        "type": "json_schema",
        "json_schema": {
            "name": "person",
            "strict": True,
            "schema": {
                "type": "object",
                "properties": {
                    "name": {"type": "string"},
                    "age": {"type": "integer"},
                },
                "required": ["name", "age"],
                "additionalProperties": False,
            },
        },
    },
)

person = json.loads(completion.choices[0].message.content or "{}")
print(person)

对照结果：

{'name': '小林', 'age': 28}

只解析 message.content，不要去解析 reasoning_content。如果打印出的 JSON 被截断，通常是输出撞到了 max_completion_tokens，调大它即可（K3 默认 131072，最大 1048576）。

第 5 步：写一个多轮工具调用循环

模型不会替你执行工具：它返回 finish_reason == "tool_calls" 和要调用的函数名、参数，由你的代码执行，再把结果用 role: "tool" 和对应的 tool_call_id 发回去，直到它返回 finish_reason == "stop"。下面在官方「最小天气 Agent Loop」的基础上加了一个循环和第二个工具。新建 tool_loop.py：

import json
import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["MOONSHOT_API_KEY"],
    base_url="https://api.moonshot.cn/v1",
)

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询城市今天的天气",
            "parameters": {
                "type": "object",
                "properties": {"city": {"type": "string", "description": "城市名"}},
                "required": ["city"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "计算一个算术表达式",
            "parameters": {
                "type": "object",
                "properties": {"expression": {"type": "string"}},
                "required": ["expression"],
            },
        },
    },
]


def get_weather(args):
    fake = {"北京": ("晴", 24), "上海": ("小雨", 21)}
    weather, temp = fake.get(args["city"], ("多云", 20))
    return {"city": args["city"], "weather": weather, "temperature_c": temp}


def calculate(args):
    allowed = set("0123456789+-*/(). ")
    expr = args["expression"]
    if not set(expr) <= allowed:
        return {"error": "只支持数字和 + - * / ( )"}
    return {"expression": expr, "result": eval(expr)}


tool_map = {"get_weather": get_weather, "calculate": calculate}

messages = [{"role": "user", "content": "北京和上海今天的气温分别是多少？两地温差是多少度？"}]

step = 0
while True:
    step += 1
    completion = client.chat.completions.create(
        model="kimi-k3",
        messages=messages,
        tools=tools,
        reasoning_effort="high",
    )
    choice = completion.choices[0]
    messages.append(choice.message)  # 完整回传，含 reasoning_content 和 tool_calls
    if choice.finish_reason != "tool_calls":
        break
    for tool_call in choice.message.tool_calls:
        name = tool_call.function.name
        args = json.loads(tool_call.function.arguments)
        result = tool_map[name](args)
        print(f"[step {step}] {name}({args}) -> {result}")
        messages.append(
            {
                "role": "tool",
                "tool_call_id": tool_call.id,
                "name": name,
                "content": json.dumps(result, ensure_ascii=False),
            }
        )

print("最终回答：", choice.message.content)

python3 tool_loop.py

对照结果（工具调用顺序和措辞可能不同，数字应一致）：

[step 1] get_weather({'city': '北京'}) -> {'city': '北京', 'weather': '晴', 'temperature_c': 24}
[step 1] get_weather({'city': '上海'}) -> {'city': '上海', 'weather': '小雨', 'temperature_c': 21}
[step 2] calculate({'expression': '24 - 21'}) -> {'expression': '24 - 21', 'result': 3}
最终回答： 北京今天晴，24°C；上海小雨，21°C；两地温差 3°C。

模型可能在同一步里并行请求两个城市的天气，所以会看到两行 step 1；也可能自己心算温差而不调 calculate，两种都算正确。

想强制模型第一步必须调工具，可以在首轮请求里加 tool_choice="required"（官方 K3 示例就是这么做的），后续轮次去掉它，否则模型永远没法给出最终回答。

自查清单

• 三档推理强度的耗时和输出 token 有明显差别
• 流式脚本先打印「思考」再打印「回答」
• 第二轮对话能记住你的名字，而且你追加的是整个 message 对象
• schema_demo.py 打印出 {'name': '小林', 'age': 28}
• tool_loop.py 至少调用一次 get_weather 并给出温差