第 09 课：流式输出 —— AI 为什么一个字一个字蹦

难度：★★★（需要理解第 08 课的 FastAPI 基础）
教材：worry_debate_game/ 的 app.py（第 138-230 行）+ nodes.py 的 stream 函数 + static/script.js（第 959-1005 行）
预计时间：讲解 70 分钟 + 练习 50 分钟
学完你能：说清「打字机效果」的实现原理，理解 SSE、生成器、流式读取，并能自己加一个新事件

主线实验：一条事件不一定对应一次读取

本课前置除了 FastAPI，还需要函数返回值、循环，以及“字节与文本不同”的概念。先不用模型，只研究同一份内容怎样切开再还原。

在下载包 assets/主线实验/ 运行 py -3 stream_demo.py。预期每块 1、2、7 字节和整段四种情况都恢复 2 条事件。一字节切法必然会拆开中文字节，也会拆开事件之间的空行。

先分清四种边界

层次	单位	能否假定与下一层一一对应
模型输出	token 或内容片段	不能假定等于一个汉字
传输读取	任意字节块	不能假定一块就是一条事件
SSE 应用消息	以空行结束的字段集合	可以包含一段文本，而不是只含一个字
界面动画	每次显示的字符或片段	由前端体验策略决定

“一个字一个字出现”可能是模型增量、网络到达和前端动画共同作用的结果，不能用视觉效果反推每次网络请求恰好传一个字。

解码和组帧各留什么状态

UTF-8 增量解码器保留未完整的字符字节；文本 buffer 保留尚未由空行结束的事件。前者解决“这个字还没到齐”，后者解决“这条消息还没到齐”。这与防抖等待用户停手是不同问题。

实验 decode_chunks 在每次读取后反复提取完整事件，多出来的尾部留到下一块。流结束时如果仍有半条，明确报错，避免静默把它当成成功完成。

本例的协议范围

小实验约定 LF 行尾、单行 JSON data、明确 event 字段。完整 SSE 还包括 CRLF、多行 data、注释、id/retry 等；原课简易 parseSSE 也不是通用解析器。

浏览器原生 EventSource 具有相应的重连机制；本项目使用 fetch POST 手动读流，重连、事件 ID、重放去重需要应用自己设计。不能因为响应的 Content-Type 是 text/event-stream，就假定 fetch 自动恢复了中断。

练习与验收

先运行 py -3 -m unittest test_labs -v，看每个可能的二段切分位置都恢复相同事件。再在自己的副本中去掉最后一个换行，预期报告不完整事件，而不是多出一个空事件。

然后回到原课的 generate/readSSEStream：标出后端事件边界、TextDecoder、buffer/remainder、事件分发四处。在没有 API Key 时，仍能完成协议与分块的核心验收；真实模型流的延迟和断线行为另做联网实验。

排错：中文乱码查增量解码；丢最后一条查结束标记；内容堆到最后出现查服务器/代理缓冲；HTTP 已开始后出错，要在应用协议说明失败或让连接中断，不再尝试改已发送的状态码。断线时客户端收不到任何 error 事件也可能发生，不能把“后端 yield 了错误”当作送达保证。

课前须知：为什么要"一个字一个字蹦"

如果等大模型把 200 字全部写完再一次性返回，用户要盯着空白屏幕等 3-10 秒——
体验很差，还会怀疑程序卡死了。

流式输出的思路：模型每产出一小段，就立刻发给前端显示。
于是就有了打字机效果，用户 1 秒内就能看到第一句话，等待焦虑大大降低。

这需要前后端配合：

后端：模型吐出 chunk → 立刻 yield 出去 → 框架逐块发给浏览器
前端：收到一块 → 立刻追加到气泡里显示

第一步：观察现象

启动项目，打开页面，输入一个烦恼，点开始
F12 → Network → 找到 start-stream 请求 → 点开 Response / EventStream 标签
你会看到内容不是一次性出现的，而是一条条往下冒：

event: worry
data: {"text": "我每天都熬夜..."}

event: tendency
data: {"score": -15}

event: angel_start
data: {}

event: angel
data: {"text": "我"}

event: angel
data: {"text": "理解"}

...

上面这个格式，就是本课的主角：SSE（Server-Sent Events，服务器推送事件）。

第二步：模块地图

后端 app.py
├── sse(event, data)           第 138-140 行  把数据拼成 SSE 格式的字符串
├── start_game_stream()        第 143-187 行  首轮：烦恼 → 天使 → 恶魔 → round_done
└── next_round_stream()        第 190-230 行  后续轮：天使 → 恶魔 → round_done

后端 nodes.py
├── stream_angel(state)        第 66-105 行  逐段产出天使台词（yield）
└── stream_demon(state)        第 108-150 行  逐段产出恶魔台词（yield）

前端 static/script.js
├── parseSSE(buffer)           第 960-977 行  把文本按 SSE 规则切成一条条事件
├── readSSEStream(response,..) 第 980-1005 行 不停读取流，分发给处理函数
└── 排队机制                    第 1007 行起   处理"后端太快、前端显示慢"的问题

第三步：逐块精讲

块 1：先理解"生成器（yield）"（`nodes.py` 第 98-105 行）

full = ""
for chunk in llm.stream([SystemMessage(content=system_prompt), HumanMessage(content=user_prompt)]):
    content = chunk.content
    if content:
        full += content        # 边攒完整版
        yield content          # 边吐出一小块

llm.stream(...) 是 LangChain 的流式调用：模型每生成一小段，就交给你一小段
yield 和 return 的区别：
- return：函数结束，一次性给出结果
- yield：函数可以先歇着，产出一个值，等外面取下一个时再继续——

这种函数叫生成器（generator）

full += content 在攒"完整台词"，最后一行把它写回 state["angel_history"]——

因为历史记录要存整段，而不是碎片

生活比喻：return 是"做好一整桌菜再端上来"；yield 是"做好一道端一道"。

块 2：SSE 是什么（`app.py` 第 138-140 行）

def sse(event: str, data: dict):
    """格式化 SSE 事件字符串"""
    return f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n"

本项目采用的 SSE 子集可用下面的字段与空行表示；完整协议还支持多行 data、注释、id、retry 等：

event: 事件名
data: JSON内容
（空行表示这条事件结束）
（然后下一条）

规则：两个换行 \n\n 分隔不同事件。

为什么用这种"简陋"的文本格式？因为它就是纯文本流，
浏览器和服务器容易检查文本格式；原生 EventSource 有重连机制，但本课 fetch POST 手动读流不会自动重连，恢复与去重需另行设计。

块 3：`StreamingResponse` —— 逐块发出去（第 187 行）

return StreamingResponse(generate(), media_type="text/event-stream")

generate() 是一个生成器函数（里面全是 yield sse(...)）
StreamingResponse 会不断从生成器取下一个值，取到一个就立刻发给浏览器
media_type="text/event-stream" 是 SSE 的标准内容类型，

告诉浏览器"这是事件流，别等全部结束"

对比第 08 课的普通接口：那边是"函数 return 一个字典，框架序列化后一次发完"；
这里是"生成器不断 yield，框架边收边发"。这是本课最核心的差别。

块 4：后端的事件剧本（第 158-185 行）

def generate():
    try:
        state.update(generate_worry_node(state))
        yield sse("worry", {"text": state["worry"]})
        yield sse("tendency", {"score": state["tendency"]})

        yield sse("angel_start", {})
        for chunk in stream_angel(state):
            yield sse("angel", {"text": chunk})
        yield sse("angel_end", {})

        yield sse("demon_start", {})
        for chunk in stream_demon(state):
            yield sse("demon", {"text": chunk})
        yield sse("demon_end", {})

        sessions[session_id] = state
        yield sse("round_done", {...})
    except Exception as e:
        yield sse("error", {"message": ...})

事件序列设计得很有讲究：

事件	作用
`worry`	把烦恼陈述发给前端显示
`tendency`	当前倾向分（正数偏天使、负数偏恶魔）
`angel_start` / `angel_end`	告诉前端"天使开始说了 / 说完了"
`angel`	天使台词的一个片段（会来很多条）
`demon_start` / `demon_end`	恶魔的边界
`demon`	恶魔台词的片段
`round_done`	这一轮结束，带上 session_id 供下一轮用
`error`	可送达时通知应用错误；连接断开时不能保证前端收到，客户端还要处理读流失败

为什么要 start/end 这种"开始/结束"事件？
因为流式数据只有片段，前端需要知道"这段碎片属于谁、什么时候该换人说话"。
这也是前端能做"天使说完，恶魔才开口"的原因。

注意 except 里也能 yield：流式响应一旦开始（已经发了一部分），
就没法再改成 500 错误了，所以只能在流里发一个 error 事件通知前端。

块 5：前端怎么收流（`script.js` 第 980-1005 行）

async function readSSEStream(response, handlers) {
    const reader = response.body.getReader();      // 拿到"流读取器"
    const decoder = new TextDecoder();
    let buffer = "";

    while (true) {
        const { done, value } = await reader.read();   // 等下一块数据
        if (done) break;
        buffer += decoder.decode(value, { stream: true });  // 字节 → 文本
        const { events, remainder } = parseSSE(buffer);     // 切成完整事件
        buffer = remainder;                                 // 剩下的半条留着
        for (const evt of events) {
            const handler = handlers[evt.event];            // 按事件名找处理函数
            if (handler) handler(evt.data);
        }
    }
}

四个要点：

为什么不用浏览器自带的 EventSource？

因为 EventSource 只支持 GET 请求，而这里要 POST 发送 topic。
所以用 fetch + 手动读流。这是很常见的取舍。

response.body.getReader() + reader.read()：一块一块地读，读到 done 为止
TextDecoder：网络传的是字节，要解码成文字；

{stream: true} 表示"这可能只是一半字符，别急着当成完整内容"

buffer / remainder 的处理：网络传输会随机切块，

可能一次收到半条事件，所以把不完整的部分留到下次拼接——
这是流式解析的消息边界问题；防抖控制触发频率，不能替代字节解码或事件重组

块 6：解析事件（第 960-977 行）

function parseSSE(buffer) {
    const events = buffer.split("\n\n");   // 空行分隔事件
    const remaining = events.pop();        // 最后一段可能不完整，留着
    const parsed = [];
    for (const raw of events) {
        const lines = raw.split("\n");
        let eventType = "message";
        let data = {};
        for (const line of lines) {
            if (line.startsWith("event: ")) eventType = line.slice(7).trim();
            if (line.startsWith("data: ")) {
                try { data = JSON.parse(line.slice(6)); } catch { data = {}; }
            }
        }
        parsed.push({ event: eventType, data });
    }
    return { events: parsed, remainder: remaining };
}

这就是把块 2 讲的 SSE 格式反过来解析：
按 \n\n 切、按 event: / data: 取值、JSON.parse 还原数据。
和后端 sse() 函数严格配对——你写了一个格式，就要写一个对应的解析器。

块 7：排队机制（第 1007-1010 行的注释）

前端注释写得非常清楚，值得单独讲：

后端事件顺序：angel → angel_end → demon → demon_end → round_done
前端打字速度慢于 LLM 生成速度，所以恶魔的文本块会先到达，
但我们希望恶魔等天使全部显示完再开始说。

也就是说：生产速度 > 消费速度，数据会堆积。解决办法是用队列缓冲：
先到的恶魔片段存进 angelPendingQueue，等天使的动画播完再放出来。

这是计算机里非常普遍的思想：当两边速度不匹配时，用缓冲（队列）协调。
（类似：水库调节上游来水和下游用水。）

第四步：对照开源，别人怎么做流式

对照大模型 API 自己的流式格式

OpenAI / DeepSeek 的接口本身就是 SSE：你调用时加 stream: true，
它返回的也是一条条 data: {...}，里面是 delta（增量片段）。
教材的 llm.stream(...) 就是 LangChain 帮忙把这层细节包好了。

所以链路的完整样子是：

DeepSeek（SSE）→ LangChain stream（拆成 chunk）→ 你的 stream_angel（yield）
→ FastAPI StreamingResponse（SSE）→ 浏览器 fetch 读流 → 页面打字机

四层都在传"碎片"，作者只写了中间两层，另外两层是库和框架帮忙做的。

对照 WebSocket

	SSE（本项目）	WebSocket
方向	服务器 → 客户端（单向）	双向
协议	普通 HTTP	独立协议
复杂度	低	较高
适合	推通知、AI 打字机	聊天室、协同编辑

「AI 边生成边显示」是典型的单向场景，SSE 足够且更简单。
需要双向实时互动（比如多人游戏）时再考虑 WebSocket。

对照：轮询 vs 推送

老办法是前端每隔 1 秒问一次"好了吗"（轮询），既费资源又有延迟。
SSE 是"服务器主动推"，有内容就推——这就是推送模式的价值。

动手练习（做完才算过关）

练习 1（热身）：看原始数据流
用 F12 的 Network → start-stream → EventStream 面板，数一数：
（1）一共有多少条 angel 事件？
（2）angel_start 和 angel_end 分别出现在第几条？
把这个序列和 app.py 第 158-185 行的代码对照着看一遍。

练习 2（必做）：加一个新事件
在 generate()（第 158 行）的开头加一条：

yield sse("notice", {"text": "辩论即将开始"})

然后在 script.js 里找到处理函数表（await readSSEStream(res, { ... })，第 1147 行附近），
加一个 notice 处理，用 console.log(data.text) 打印出来。
运行后看控制台——你就完成了一次"后端发事件、前端收事件"的完整扩展。

练习 3（必做）：改事件名，观察后果
把后端 yield sse("angel", ...) 的事件名改成 "angel_chunk"，但前端不改。
运行后会发现天使台词不显示了——为什么？
说说看：事件名是什么？为什么前后端必须约定一致？
（改完记得改回来。）

练习 4（思考题）：
为什么这个项目用 SSE，而不是 WebSocket？用 2-3 行写清楚。

看答案

看数据是单向还是双向。

练习 5（选做，性能对比）：
在 Network 面板里对比 /api/start（非流式）和 /api/start-stream（流式）：
看「等待首字节的时间」和「总时间」有什么不同。
说说看：为什么流式的"首字节"来得快得多？这对用户体验意味着什么？

练习 6（选做，进阶）：
parseSSE 里对 JSON.parse 做了 try/catch（第 971 行）。
如果去掉这个容错，当某条 data: 不是合法 JSON 时，前端会发生什么？

看答案

一个抛出异常会让整个读取循环怎样？

本课小结

流式输出解决"等大模型太慢"的体验问题：边生成边显示
yield（生成器）是流式的发动机：产出一个、歇一下、再产出
SSE 格式：event: 名字 + data: JSON + 空行；前后端各写一个函数配对
后端 StreamingResponse(generate(), media_type="text/event-stream") 逐块发送
前端不用 EventSource（只支持 GET），而是 fetch + getReader() 手动读流
流式解析必须处理"半条数据"（buffer/remainder），这是流式的通用难题
速度不匹配时用队列缓冲；生产速度 > 消费速度是经典问题
出错了要在流里发 error 事件（因为流已开始，无法再改状态码）

术语表

术语	人话解释
流式（streaming）	数据分多次、小块小块地传输
生成器 / yield	能"暂停并产出"的函数，流式的发动机
chunk	一小块数据（模型吐出的一小段字）
SSE	服务器推送事件的纯文本协议
`text/event-stream`	SSE 的标准内容类型
StreamingResponse	FastAPI 里"边取边发"的响应类型
getReader / TextDecoder	前端读字节流 / 把字节解码成文字
缓冲（buffer）	暂存不完整或来不及处理的数据
队列（queue）	先进先出的缓冲结构
轮询 / 推送	主动反复问 / 服务器主动告诉你

下节预告

第 10 课：调用大模型：prompt 与链。
还是 nodes.py，但这次看"怎么跟模型说话"：
get_llm 怎么切换 DeepSeek/OpenAI、system prompt 怎么写、
怎么把历史对话塞回去、evaluate_tendency 怎么让模型返回可编程的 JSON，
以及 graph.py 用 LangGraph 把节点连成流程图的思路。

← FastAPI 入门 —— 从零到接口目录调用大模型 —— prompt 与链 →