流式输出

如果想获取全套课程资料，请关注微信公众号 「AI面包君」，回复 「课程资料」 即可领取。

1. 边收到内容，边显示回复

ch03 会等模型生成完整回答，再一次打印。如果回复比较长，终端就会有一段时间没有新内容。

流式输出把回答分成小段陆续返回。程序每收到一段就显示出来，用户可以先阅读前面的内容，不必等到全部生成完。

设置 stream=True 后，SDK 返回可迭代的流式响应。for 循环每次拿到一个 chunk，再从中读取本次新增的文本。

需要多处理的一种情况是工具调用：tool_calls 也可能分段到达，名称和参数未必一次就齐。完整实现要按 index 把这些片段合并，得到完整请求后再执行。

为了先看清文本流的处理，本章主线采用一个简化方案：

流式只用于"模型输出最终文本"的回合。 如果检测到模型要调工具，放弃流式，改用 ch03 的非流式方式重新请求一次。

遇到工具调用时，这种做法会多发一次请求，增加耗时和用量。附录 APPENDIX_full_streaming.py 演示怎样直接拼接工具调用片段，可以在主线跑通后再读。

<!-- bread-visual:start -->

图解： 本章主线选择了便于理解的工具调用回退方案。 本章的“工具回合卡一下”有明确原因：额外发起了非流式请求。

<!-- bread-visual:end -->

---

2. 跑起来

cd ch04_stream
python main.py

先让 AI 写一段长回答（不会触发工具）：

你 > 用 5 段话介绍一下中国的四大发明
AI > 中国的四大发明——造纸术、印刷术、火药和指南针，是中华民族对世界文明的杰出贡献...
        ↑↑↑ 这些字应该是一个一个蹦出来的（流式）

再让它做一个会调工具的回合：

你 > 列出当前目录都有什么
AI > [切换到非流式模式获取工具调用...]
  [tool] list_dir({'path': '.'})
AI > 当前目录有 ... 等文件。

注意 [切换到非流式模式获取工具调用...] 这行——这就是我们的简化在工作：遇到工具调用就回退非流式。

---

3. 对照代码看实现

核心函数：stream_one_turn

def stream_one_turn(messages: list) -> tuple[str, bool]:
    stream = client.chat.completions.create(
        model=MODEL,
        messages=messages,
        tools=TOOLS_SCHEMA,
        stream=True,   # ← 关键开关
    )

    accumulated_text = ""
    detected_tool_call = False

    for chunk in stream:
        if not chunk.choices:
            continue
        delta = chunk.choices[0].delta

        # ★ 工具调用检测
        if delta.tool_calls:
            detected_tool_call = True
            for _ in stream:    # 把流剩下的部分消费掉
                pass
            break

        # ★ 文本 delta
        if delta.content:
            print(delta.content, end="", flush=True)
            accumulated_text += delta.content

    if accumulated_text:
        print()
    return accumulated_text, detected_tool_call