什么是流式输出

非流式：服务端把整段回答生成完，一次性返回，用户要盯着空白页面等好几秒。

流式：服务端生成一个字就返回一个字，用户看到的是文字逐渐蹦出来的效果（像 ChatGPT 那样），等待体验明显更好。

对比	非流式	流式
参数	stream=False	stream=True
返回值	一个完整的响应对象	一个可迭代对象，逐段产出 chunk
取内容	response.choices[0].message.content	chunk.choices[0].delta.content
展示	st.chat_message(...).write(完整内容)	st.empty() 占位 + 循环覆盖写入

NOTE

两种方式拿到的内容是一样的，区别只在"什么时候拿到"。所以同一份 messages 数据可以自由切换。

代码对比

非流式（上一版的写法）

response = client.chat.completions.create(
    model="deepseek-chat",
    messages=[...],
    stream=False
)

st.chat_message("assistant").write(response.choices[0].message.content)

流式（本节的写法）

response = client.chat.completions.create(
    model="deepseek-chat",
    messages=[
        {"role": "system", "content": system_prompt},
        *st.session_state.messages
    ],
    stream=True
)

# 创建一个空的组件, 用于展示大模型返回的结果
response_message = st.empty()

full_response = ""
for chunk in response:
    if chunk.choices[0].delta.content is not None:
        content = chunk.choices[0].delta.content
        full_response += content
        response_message.chat_message("assistant").write(full_response)

# 保存大模型返回的结果
st.session_state.messages.append({"role": "assistant", "content": full_response})

关键点拆解

1. 为什么取值的路径变了

结构	含义
response.choices[0].message.content	非流式：message 是完整消息
chunk.choices[0].delta.content	流式：delta 表示"这一小段增量"

delta 就是"增量"的意思，每次循环拿到的是新增的那几个字，不是全文。

2. 为什么要判断 is not None

if chunk.choices[0].delta.content is not None:

流式返回的最后几个 chunk 里，delta.content 是 None（只带结束标记等元信息，没有文字）。如果不判断就直接字符串拼接，会报 TypeError。

3. st.empty() 的作用

response_message = st.empty()
...
response_message.chat_message("assistant").write(full_response)

• st.empty() 先在页面上占一个空位
• 之后每次循环都往这个空位里重新写一次 full_response（累积后的完整文字），覆盖掉上一次的内容
• 效果就是同一个气泡里的文字不断变长，而不是每加几个字就新建一个气泡

WARNING

如果不用 st.empty()，而是每轮循环都调用 st.chat_message("assistant").write(...)，页面上会不停新增几十个聊天气泡，每个气泡只装一小段文字。

4. 为什么还要 full_response 累加

因为 delta.content 只是增量。要保存到 session_state.messages 里的是完整回答（下次发给模型时必须是完整的 assistant 消息），所以需要边拼边存。

补充：更简洁的官方写法

Streamlit 提供了 st.write_stream()，专门用来展示流式输出，可以省掉手写循环：

response = client.chat.completions.create(..., stream=True)
st.chat_message("assistant").write_stream(response)

它要求传入的是"文本片段的可迭代对象"。课程里手写 chunk 循环的价值在于：看清楚流式数据到底长什么样，以及 delta 与 message 的区别。理解了原理之后，实际项目可以直接用 write_stream。

TIP

想看原始 chunk 结构，可以在循环里临时加一行 print(chunk)，观察每个 chunk 的 choices[0].delta 与最后一个 chunk 的 finish_reason。

相关

• 实战-AI智能伴侣-基本交互
• 实战-AI智能伴侣-会话记忆

练习题

一、知识回顾（读完直接做下面的实践题）

1. 开启流式：client.chat.completions.create(..., stream=True)
2. 流式里每次循环拿到的是"增量"，取值路径是 chunk.choices[0].delta.content
3. 非流式取的是 message.content，流式取的是 delta.content（"增量"的意思）
4. 循环里要判断 content is not None，因为最后几个 chunk 只带结束标记、没有文字
5. 占位组件：response_message = st.empty()，之后可以反复往这个空位里覆盖写入
6. 边拼边存：full_response += content
7. 官方更简洁的写法：st.chat_message("assistant").write_stream(response)
8. 不用 st.empty() 占位、每轮直接 write，页面上会冒出几十个只装几个字的气泡；占位后反复覆盖写入，才只有一个气泡

二、裸写题

• 2-1 看清流式数据长什么样 发一次流式请求，在循环里把每次拿到的"增量片段"打印出来，观察：文字是怎么一段段来的？最后一个片段里是什么？

  - 提示（先自己想，实在想不出再点开）

  一级 · 思路：不要急着拼字符串，先把每个片段整个打印出来看清楚结构，再看它的"内容"字段 二级 · 方法：for chunk in response: + 打印 chunk.choices[0].delta三级 · 骨架：注意有些 chunk 的 delta.content 是 None，要先判断

  - 参考答案（做完再点开）

  import os
  from openai import OpenAI
  
  client = OpenAI(api_key=os.environ.get("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com")
  
  response = client.chat.completions.create(
      model="deepseek-chat",
      messages=[{"role": "user", "content": "从1数到5"}],
      stream=True,
  )
  
  for chunk in response:
      print(chunk)                            # 先看整体结构
      # 看清楚了再换成：print(chunk.choices[0].delta)
  
  # 观察结论：最后一个 chunk 的 delta.content 是 None（只带结束信息），
  # 所以拼接前必须判断 is not None

• 2-2 打字机效果 做一个占位组件，让同一个气泡里的文字不断变长（每次循环覆盖写入累积后的全文），而不是每加几个字就新建一个气泡。

  - 提示（先自己想，实在想不出再点开）

  一级 · 思路：先占一个空位，之后每次循环都往这个空位重写"累积后的全文" 二级 · 方法：st.empty() / response_message.chat_message("assistant").write(full_response)三级 · 骨架：不用 st.empty() 会怎样？试试直接 write，看页面上出现多少个气泡

  - 参考答案（做完再点开）

  import os
  import streamlit as st
  from openai import OpenAI
  
  st.title("打字机效果演示")
  client = OpenAI(api_key=os.environ.get("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com")
  
  prompt = st.chat_input("请输入您要问的问题")
  if prompt:
      st.chat_message("user").write(prompt)
  
      response = client.chat.completions.create(
          model="deepseek-chat",
          messages=[{"role": "user", "content": prompt}],
          stream=True,
      )
  
      response_message = st.empty()          # 先占一个空位，之后反复覆盖它
      full_response = ""
      for chunk in response:
          if chunk.choices[0].delta.content is not None:
              full_response += chunk.choices[0].delta.content
              response_message.chat_message("assistant").write(full_response)
  
  # 对比：把 st.empty() 那两行换成每轮直接 st.chat_message("assistant").write(...)，
  # 页面上会冒出几十个气泡，每个只装几个字