FastAPI 流式传输数据实战：用 `StreamingResponse` + `yield` 逐块输出字符串与二进制文件

本篇指南讲解 FastAPI 如何「流式传输」不适合用 JSON 表达的数据——包括从 AI LLM 服务直接输出的纯字符串、大体积二进制文件、乃至边生成边发送的视频或音频。读完本文，你将掌握 response_class=StreamingResponse 配合生成器函数（yield）的三种典型姿势（async def / 普通 def / 省略返回类型注解）、如何流式输出 bytes、如何通过自定义子类注入 Content-Type 响应头（如 image/png），并理解文件对象与事件循环、yield from 等底层细节。该能力自 FastAPI 0.134.0 起可用。

适用范围：什么时候该走流式响应

在深入代码之前，先明确本文所讨论功能的边界。若你传输的数据可以被结构化为 JSON，FastAPI 官方推荐优先采用逐行 JSON 的 JSON Lines 流式方案，参见 JSON Lines 流式教程（对应的示例代码位于 docs_src/stream_json_lines）。

而本文面向的是另一类需求——传输纯二进制数据或纯字符串：

• LLM 输出文本流：直接从 AI 大模型服务返回结果，逐块吐给客户端；
• 大体积二进制文件：边读边发，不必一次性把整个文件读进内存；
• 视频 / 音频：甚至可以边处理、边生成、边发送。

判断依据很简单：只要内容"不是 JSON 的形状"，就属于本文 StreamingResponse 的用武之地。

一个带 yield 的 StreamingResponse 路径函数

最直接的流式用法是：在路径操作函数上声明 response_class=StreamingResponse，然后用 yield 把每个数据块依次送出。完整可运行示例见 docs_src/stream_data/tutorial001_py310.py：

from collections.abc import AsyncIterable

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

message = """...一段较长的对话文本..."""


@app.get("/story/stream", response_class=StreamingResponse)
async def stream_story() -> AsyncIterable[str]:
    for line in message.splitlines():
        yield line

这段代码在 /story/stream 上注册了一个 async def 路径函数，返回类型注解为 AsyncIterable[str]。当客户端请求该端点时，函数并不会一次性返回完整正文，而是把 message 按行拆开、一行一行 yield 出去。

关键行为是：FastAPI 会把每个 yield 出来的数据块原样交给 StreamingResponse，不做任何 JSON 转换，也不做其他形式的序列化。数据按你 yield 的顺序被写入响应体并即时推送。

这一点可以从路由层实现得到印证：在 fastapi/routing.py 中，FastAPI 会先判断路径函数是否是一个生成器（async 生成器或普通生成器），若是，则直接走"原始流式"分支：

elif _is_async_gen_callable(dependant.call) or _is_gen_callable(dependant.call):
    # Raw streaming with explicit response_class (e.g. StreamingResponse)
    gen = dependant.call(**solved_result.values)
    ...
    response = actual_response_class(content=gen, **response_args)

也就是说，FastAPI 压根不进入 Pydantic 校验与 JSON 序列化管线，而是把生成器对象本身作为 content 交给响应类。所有字节如何组织、如何编码，都由你的生成器说了算。

非异步的路径操作函数

yield 流式并不要求函数是异步的。普通 def 函数同样可以使用生成器：

@app.get("/story/stream-no-async", response_class=StreamingResponse)
def stream_story_no_async() -> Iterable[str]:
    for line in message.splitlines():
        yield line

两种写法都得到完全一致的输出（下方测试部分会给出证据）。选用哪个取决于数据源本身是同步还是异步的——例如从磁盘读文件通常是同步 I/O，就更适合 def 版本。

可以不写返回类型注解

流式传输二进制数据时，返回类型注解不是必需的：

@app.get("/story/stream-no-annotation", response_class=StreamingResponse)
async def stream_story_no_annotation():
    for line in message.splitlines():
        yield line

原因正是上文提到的那句关键行为：FastAPI 不会用 Pydantic 把数据转成 JSON，也不会做任何形式的序列化。这种情况下，类型注解只对你的编辑器和静态分析工具有意义，FastAPI 根本不会读取它。

这也意味着在使用 StreamingResponse 时，你同时握有自由与责任：最终线路上发送的每个字节，都必须由你自己精确地生产与编码，不能指望注解替你兜底。

流式传输 bytes

流式传输的主要用例之一是输出 bytes 而非 str。做法同样简单——逐块 yield 字节数据即可：

@app.get("/story/stream-bytes", response_class=StreamingResponse)
async def stream_story_bytes() -> AsyncIterable[bytes]:
    for line in message.splitlines():
        yield line.encode("utf-8")

把字符串用 .encode("utf-8") 变成字节后 yield，客户端收到的就是不加任何包装的原始字节流。上面的字符串 / 字节、async / 非 async、带 / 不带注解共可组合出 8 个端点（源码中均已给出），它们产出的响应体完全一致。

自定义 PNGStreamingResponse：为流补上 Content-Type

前文示例虽然在流式传输数据字节，但响应没有 Content-Type 头，客户端无从得知接收到的到底是什么类型。解决办法是对 StreamingResponse 做子类化，用类属性 media_type 声明要流出的内容类型。

以 PNG 图片为例，定义如下子类（见 docs_src/stream_data/tutorial002_py310.py）：

from fastapi.responses import StreamingResponse

class PNGStreamingResponse(StreamingResponse):
    media_type = "image/png"

接着把它放进路径函数的 response_class：

@app.get("/image/stream", response_class=PNGStreamingResponse)
async def stream_image() -> AsyncIterable[bytes]:
    with read_image() as image_file:
        for chunk in image_file:
            yield chunk

StreamingResponse 内部会依据 media_type 在响应上设置 Content-Type 头。你可以在 StreamingResponse 上直接设置，也可以通过继承覆盖，适用于任何自定义媒体类型——文本、字体、音视频均可如法炮制。

顺带说明：StreamingResponse 本身由 Starlette 提供，并在 fastapi/responses.py 中被重新导出，因此 from fastapi.responses import StreamingResponse 即可直接使用；FastAPI 自带的 SSE（Server-Sent Events）响应 fastapi/sse.py 也是同一机制（media_type = "text/event-stream"）的典型实践。

用 io.BytesIO 模拟文件

上面示例里的 read_image() 用内存中的 io.BytesIO 模拟文件读取：

import base64
from io import BytesIO

# 一个 Base64 编码的 PNG 图片（示例内嵌，便于单文件直接运行）
image_base64 = "iVBORw0KGgo...（省略）"
binary_image = base64.b64decode(image_base64)


def read_image() -> BytesIO:
    return BytesIO(binary_image)

BytesIO 是只存活于内存的类文件对象，但接口与真文件一致——可以迭代消费其内容，就像遍历一个真实文件那样：

@app.get("/image/stream", response_class=PNGStreamingResponse)
async def stream_image() -> AsyncIterable[bytes]:
    with read_image() as image_file:
        for chunk in image_file:
            yield chunk

两个内嵌变量 image_base64（Base64 字符串）与 binary_image（base64.b64decode 解码后的字节）只是为了把一张图片"塞进"同一个示例文件，方便你直接复制运行。

注意这里使用了 with 块：它保证生成器函数（含 yield 的函数）运行结束后，类文件对象会被正确关闭——也就是响应发送完毕之后。对本示例而言（内存中的假文件）关闭与否并不重要；但换成真实文件时，用 with 确保用完即关就非常关键，否则可能泄漏文件描述符。

文件与 async 的关系

大多数真实场景里，类文件对象默认不兼容 async/await——没有 await file.read()，也没有 async for chunk in file。而且读取磁盘或网络上的文件通常是阻塞操作，若在事件循环里直接执行，会卡住整个服务进程。

上面 io.BytesIO 的例子其实是例外：数据已在内存中，读取不产生任何阻塞。但对真正的文件，你需要避免阻塞事件循环。最简单的办法是：把路径操作函数声明为普通 def 而非 async def，FastAPI 会把它调度到**线程池（threadpool）**里执行，从而避免阻塞主事件循环：

@app.get("/image/stream-no-async", response_class=PNGStreamingResponse)
def stream_image_no_async() -> Iterable[bytes]:
    with read_image() as image_file:
        for chunk in image_file:
            yield chunk

这与 FastAPI 对"阻塞型依赖/函数跑线程池、async def 跑事件循环"的整体模型是一致的。若你必须在异步函数中调用阻塞代码、或在阻塞代码中调用异步函数，原文档给出的建议是使用 FastAPI 的姊妹库 Asyncer 这类工具来桥接。

yield from 简化迭代转发

当你遍历某个对象（比如类文件对象）并逐项 yield 时，可以直接改用 yield from，省去手写 for 循环：

@app.get("/image/stream-no-async-yield-from", response_class=PNGStreamingResponse)
def stream_image_no_async_yield_from() -> Iterable[bytes]:
    with read_image() as image_file:
        yield from image_file

yield from 会把可迭代对象中的每一项逐个"转交"出去，是纯 Python 语法，并非 FastAPI 特性，但配合流式响应非常顺手——一句话概括：当你"只是要把整个可迭代对象原样转发"时，yield from 就是最简洁的写法。

官方测试如何验证这些端点

仓库测试对上述所有端点做了端到端断言，是理解各变体等价性的最佳佐证。

test_stream_data/test_tutorial001.py 对 8 个 /story/stream* 变体逐一请求，断言：

response = client.get(path)
assert response.status_code == 200, response.text
assert response.text == expected_text

即字符串版本、字节版本、async/非 async、带/不带注解的所有组合，最终响应文本都必须与 expected_text 完全一致——证明这些写法产出的流内容没有区别。同一文件中的 test_openapi_schema 还以快照方式断言 OpenAPI schema 中 200 响应不携带 content 媒体类型描述，印证了"无注解时 FastAPI 不尝试序列化"的设计。

test_stream_data/test_tutorial002.py 则对 5 个 /image/stream* 变体断言了三条关键事实：

assert response.status_code == 200
assert response.headers["content-type"] == "image/png"
assert response.content == mod.binary_image

• 状态码 200：流式传输正常完成；
• content-type 为 image/png：自定义 media_type 生效，Content-Type 头被正确注入；
• 响应体与原始 binary_image 逐字节相等：图片数据被"原样"流出，没有走 Base64 或 JSON 之类的中转。

同时该测试的 OpenAPI 快照显示，一旦设置 media_type，schema 的 200 响应就会带上 "content": {"image/png": {"schema": {"type": "string"}}} 描述——这正是自定义响应类的额外收益：流式之外，OpenAPI 文档也能正确标注媒体类型。

实践要点小结

把以上内容收束为几条可直接落地的准则：

1. 结构化 JSON 数据 → 用 JSON Lines 流式（相关教程）；纯字符串 / 原始二进制 / 音视频 → 用 StreamingResponse 本文方案。
2. 逐块推送只需在路径函数上声明 response_class=StreamingResponse 并在函数内 yield 每个块即可，FastAPI 不做任何转换。
3. 阻塞型 I/O（读真文件）用普通 def，让 FastAPI 丢到线程池执行，避免卡住事件循环；纯内存对象（如 BytesIO）用 async def 也无妨。
4. 别忘了 with 管理真实文件的关闭时机；用 yield from 简化"原样转发整个可迭代对象"的写法。
5. 需要明确媒体类型时，子类化 StreamingResponse 并设置 media_type，它既会写入 Content-Type 头，也会让 OpenAPI schema 正确呈现。
6. 返回类型注解不会被执行，只服务于编辑器与工具，因此字节的编码正确性完全由你负责。

核心示例代码位于 docs_src/stream_data/tutorial001_py310.py 与 docs_src/stream_data/tutorial002_py310.py，可直接复制运行；路由层的流式分发逻辑可对照 fastapi/routing.py 与 fastapi/responses.py 深入阅读，测试基准见 tests/test_tutorial/test_stream_data。