W1 · 流式输出：模型怎么"一个字一个字"说话

1
2

  非流式：  [.......... 等待 5 秒 ..........] → "北京今天晴，26℃"
  流式：    "北" → "京" → "今" → "天" → "晴" → ...  → 全程秒出第一个字

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

  chunk（一截增量）
    │
    ├── choices  ← 数组！为什么是数组？
    │     │
    │     └── [0]  ← 取第 0 个候选
    │           │
    │           └── delta  ← 增量，不是完整答案
    │                 │
    │                 ├── content            正式回答的文字增量
    │                 ├── reasoning_content  思考过程的文字增量（DeepSeek）
    │                 └── role               角色标记（只在第一截出现）
    │
    └── finish_reason  ← 流快结束时才有："stop" / "length" ...

1
2

# n=1（默认）：只生成一份答案 → choices 只有一项
# n=3：生成三份 → choices 有三项，你可以挑最好的那份

1

chunk.choices[0].delta.content   # 永远 [0]

1
2
3
4
5
6
7
8
9

  模型的完整答案："北京今天晴，26℃"

  流式吐出来是：
    chunk 1: delta.content = "北京"
    chunk 2: delta.content = "今天"
    chunk 3: delta.content = "晴，26"
    chunk 4: delta.content = "℃"

  你做的事：content += chunk 的 delta   → 最终拼回 "北京今天晴，26℃"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),     # 密钥走环境变量，不硬编码
    base_url="https://api.deepseek.com",       # DeepSeek 走 OpenAI 兼容协议
)

messages = [{"role": "user", "content": "你是哪个大模型"}]   # 初始对话：只有一条用户消息

# 关键就一个参数：stream=True —— 让响应变成"一截一截吐"
response = client.chat.completions.create(
    model="deepseek-v4-flash",
    messages=messages,
    stream=True,                    # 🎯 开启流式
    reasoning_effort="low",         # DeepSeek 思考强度
    extra_body={"thinking": {"type": "enabled"}},   # 开启深度思考（reasoning 通道）
)

# 两个累加器：一个收思考过程，一个收正式回答
reasoning_content = ""
content = ""

for chunk in response:                        # 遍历每一截 chunk
    # 思考过程的文字增量（DeepSeek 的"草稿纸"）
    if chunk.choices[0].delta.reasoning_content:
        reasoning_content += chunk.choices[0].delta.reasoning_content
    # 正式回答的文字增量（给用户看的"最终稿"）
    if chunk.choices[0].delta.content:
        content += chunk.choices[0].delta.content

print("Reasoning Content:", reasoning_content)   # 模型自己想的（可选展示）
print("Content:", content)                       # 正式回答

1
2
3
4
5
6
7
8

  通道 A  reasoning_content   模型的思考过程（草稿纸）
        —— 模型在心里推理：用户问"你是哪个模型"，
           先想"他可能在测我有没有系统提示泄露"……
        —— 这条流不该直接给用户看（可能泄露 prompt、产生废话）

  通道 B  content             正式回答（最终稿）
        —— "我是 DeepSeek 大模型，由深度求索公司开发……"
        —— 这条才是要展示给用户的

1
2
3

if content:
    return content            # 有正式稿，用正式稿
return reasoning_content      # 没有正式稿（纯思考题），把思考结果当答案

1

TypeError: can only concatenate str (not "NoneType") to str

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

  ① 流刚开始
     chunk = {delta: {role: "assistant"}}      ← 只有角色标记，没有 content
     模型先"开口"声明身份，还没开始说话

  ② 思考阶段
     chunk = {delta: {reasoning_content: "..."}}  ← 只有草稿纸，content 是空的
     模型正在想，还没落到正式稿

  ③ 流快结束
     chunk = {finish_reason: "stop"}           ← 只带结束标志，content 是 None
     模型说完了，最后补一个"我说完了"的信号

1
2
3
4

  chunk 1      →  role="assistant"              （① 开口）
  chunk 2..n   →  reasoning_content="..."       （② 思考，content 为空）
  chunk n+1..  →  content="..."                 （正式开始说人话）
  chunk ...    →  content=None + finish_reason  （③ 收尾）

1
2
3

# 收思考：   delta.reasoning_content 有值 → 累加
# 收正文：   delta.content 有值 → 累加
# 收尾信号： chunk 里没有 delta 文字了，或 finish_reason 出现 → 循环该结束

1

if delta.reasoning_content:      # ← 这个属性，普通模型（不开 thinking）上不存在

1
2
3
4
5

messages = [
    {"role": "system", "content": "你是耐心耐心的 Agent 老师，擅长用通俗语言解释。"},
    {"role": "user", "content": "什么是流式输出？"},       # 本轮问题
    # ← 下一轮：把模型上一轮的回答 append 成 assistant
]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

# ===== 最小流式骨架 =====
import os
from openai import OpenAI

client = OpenAI(api_key=os.getenv("DEEPSEEK_API_KEY"),
                base_url="https://api.deepseek.com")

def get_response(messages, **kwargs):
    """流式请求 + 双通道收集（思考 + 正文）。返回完整正文。"""
    response = client.chat.completions.create(
        model=kwargs.get("model", "deepseek-v4-flash"),
        messages=messages,
        stream=True,                                  # 🎯 流式开关
        reasoning_effort=kwargs.get("reasoning_effort", "low"),
        max_tokens=kwargs.get("max_tokens", 500),
    )
    reasoning_content, content = "", ""
    for chunk in response:
        delta = chunk.choices[0].delta                 # 三层路径：choices[0].delta
        if delta.reasoning_content:                    # 草稿纸（思考）
            reasoning_content += delta.reasoning_content
        if delta.content:                              # 正式稿（正文）—— 先判空！
            content += delta.content
    return content or reasoning_content                # 没有正式稿就把思考当答案

# ===== 三条认知 =====
# 1. delta = 增量（Δ），不是完整答案，必须累加
# 2. choices[0] 是礼貌性下标：n 候选里你只用第一个
# 3. 空 chunk 有三种：开头 role / 思考阶段 / 收尾 finish_reason —— 先判空再拼接