FastAPI 流式输出实战：用 StreamingResponse 与 yield 流式传输字符串和二进制数据

FastAPI 流式输出实战：用 StreamingResponse 与 yield 流式传输字符串和二进制数据

【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi

本篇基于 FastAPI 官方文档 Stream Data 讲解如何用response_class=StreamingResponse配合yield流式输出纯文本或二进制数据（如 LLM 逐 token 输出、大文件、音视频流），并结合 fastapi/routing.py 的源码实现与 docs_src/stream_data 的完整示例代码，说明其调用链、Content-Type 自定义与异步/线程池注意事项，帮助你在 FastAPI 0.134.0 及以上版本中正确落地原生流式响应。

一、什么时候用原生流式输出

FastAPI 对不同"可流式"的数据提供了不同的原生支持：

• 如果你的数据可以结构化为JSON，应优先参考 Stream JSON Lines 文档，让框架按 JSONL 格式逐行序列化；
• 如果你想流式输出纯二进制数据或纯字符串（不经过任何 JSON 序列化），则使用本文介绍的StreamingResponse+yield方案。

注意：该能力（在path operation function中声明response_class=StreamingResponse并用yield逐块发送数据）是在FastAPI 0.134.0中新增的。当前仓库fastapi/__init__.py中版本为0.141.1（见 fastapi/__init__.py），已包含此特性。

典型使用场景

• 流式输出 AI LLM 的纯文本 token：从大模型服务拿到一个 chunk 就立即转发给客户端，而不是等整段文本生成完毕；
• 流式传输大二进制文件：边读边发，每个数据块读一块发一块，不必把整个文件读进内存；
• 流式传输视频或音频：甚至可以边处理边生成边发送，实现"生成即传输"。

从源码看：FastAPI 如何处理生成器端点

在 fastapi/routing.py 中，当 FastAPI 判断端点是生成器函数（_is_async_gen_callable或_is_gen_callable）时，会进入"原始流式（raw streaming）"分支：

# Raw streaming with explicit response_class (e.g. StreamingResponse) gen = dependant.call(**solved_result.values) if _is_async_gen_callable(dependant.call): async def _async_stream_raw( async_gen: AsyncIterator[Any], ) -> AsyncIterator[Any]: async for chunk in async_gen: yield chunk # To allow for cancellation to trigger # Ref: https://github.com/fastapi/fastapi/issues/14680 await anyio.sleep(0) gen = _async_stream_raw(gen) response_args = _build_response_args( status_code=status_code, solved_result=solved_result ) response = actual_response_class(content=gen, **response_args) response.headers.raw.extend(solved_result.response.headers.raw)

这段代码印证了文档中的三个关键说法：

1. 原样转发，不做序列化：gen（你的生成器）被直接作为content传给actual_response_class（即你声明的response_class），每个 chunk 原封不动地交给StreamingResponse，FastAPI 不会尝试把数据转成 JSON 或做任何序列化；
2. 异步生成器会被包一层"取消检查点"：_async_stream_raw在每次yield后执行await anyio.sleep(0)，这是为了让客户端断连时的取消操作（cancellation）能够被投递进来（源码注释指向 issue #14680）。你在业务代码里不需要关心这一层；
3. 响应头由你控制的 response_class 决定：actual_response_class(content=gen, ...)会按你声明的类构造响应。默认StreamingResponse不设置Content-Type；如果你子类化并设置media_type，就会体现在响应头中（这正是下文PNGStreamingResponse的原理）。此外 FastAPI 会把端点上声明的额外响应头通过response.headers.raw.extend(...)合并进来。

StreamingResponse本身并不是 FastAPI 自己实现的类，fastapi/responses.py 只是从 Starlette 重新导出它：

from starlette.responses import StreamingResponse as StreamingResponse # noqa

所以它的行为（逐块写入 ASGI 响应、media_type类属性决定Content-Type等）遵循 Starlette 的约定。

二、StreamingResponse+yield：逐个 chunk 发送

在path operation function中声明response_class=StreamingResponse后，就可以用yield依次发送每个数据块。官方示例 docs_src/stream_data/tutorial001_py310.py 用一个"Rick and Morty" 对话文本演示了全部变体，核心写法如下：

from collections.abc import AsyncIterable, Iterable from fastapi import FastAPI from fastapi.responses import StreamingResponse app = FastAPI() message = """ Rick: (stumbles in drunkenly, and turns on the lights) Morty! You gotta come on. ... ...（多行对话文本，略） """ @app.get("/story/stream", response_class=StreamingResponse) async def stream_story() -> AsyncIterable[str]: for line in message.splitlines(): yield line

要点：

• 端点函数必须是一个生成器函数（含yield）；
• 返回注解常用AsyncIterable[str]（异步）或Iterable[str]（同步），但注解只是给编辑器和工具看的（见下节）；
• 每个yield出来的块会被原样写给客户端。

非 async 的path operation function也可以

普通的def函数（不带async）同样可以用yield：

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

FastAPI 检测到同步生成器时不会套上_async_stream_raw包装（见上文源码），而是直接把同步生成器交给StreamingResponse，由 Starlette 在线程池中迭代它，避免阻塞事件循环。

可以不写返回类型注解

流式输出二进制数据时，你其实不需要声明返回类型注解：

@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而不是字符串：

@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")

官方示例 tutorial001_py310.py 一共定义了 8 个端点，覆盖 "async/非 async" × "带注解/不带注解" × "str/bytes" 的全部组合（/story/stream、/story/stream-no-async、/story/stream-no-annotation、/story/stream-no-async-no-annotation、/story/stream-bytes、/story/stream-no-async-bytes、/story/stream-no-annotation-bytes、/story/stream-no-async-no-annotation-bytes），对应的测试 tests/test_tutorial/test_stream_data/test_tutorial001.py 对这 8 个路径做了参数化断言：

def test_stream_story(client: TestClient, path: str): response = client.get(path) assert response.status_code == 200, response.text assert response.text == expected_text

即：无论哪种组合，客户端拿到的最终文本都等于各行 chunk 的直接拼接——验证了"chunk 原样转发、无 JSON 包装"的行为。

三、自定义PNGStreamingResponse：补上 Content-Type

上面示例流式发送了数据字节，但响应没有Content-Type头，客户端无从判断收到的是什么类型的数据。解决办法：创建StreamingResponse的子类，把Content-Type设为你要流式传输的数据类型。

例如用media_type类属性把Content-Type设为image/png，官方示例 docs_src/stream_data/tutorial002_py310.py：

import base64 from collections.abc import AsyncIterable, Iterable from io import BytesIO from fastapi import FastAPI from fastapi.responses import StreamingResponse image_base64 = "iVBORw0KGgoAAAANSUhEUgAAAB0AAAAdCAYAAABWk2cP...（Base64 编码的 PNG 图片，略）..." binary_image = base64.b64decode(image_base64) def read_image() -> BytesIO: return BytesIO(binary_image) app = FastAPI() class PNGStreamingResponse(StreamingResponse): media_type = "image/png" @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与binary_image只是一个 Base64 编码图片及其解码后的字节，目的是让图片"住"在同一个文件里，你可以复制下来直接运行；真实项目中它们通常来自数据库、对象存储或本地文件。

测试 tests/test_tutorial/test_stream_data/test_tutorial002.py 对/image/stream等 5 个端点断言了两点：

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

前者验证了自定义media_type生效，后者验证了流式拼回的字节与原始图片完全一致。

用io.BytesIO模拟文件

示例用io.BytesIO（内存中的类文件对象）来模拟文件，好处是它与真实文件接口一致——你可以像遍历文件一样遍历它来消费内容。

使用with块的意义：保证生成器函数（含yield的函数）执行完毕、即响应发送完成后，类文件对象被关闭。对这个纯内存的io.BytesIO来说关闭与否不太要紧；但对真实文件，确保"用完后关闭"非常重要。

文件读取与 async 的关系

大多数类文件对象默认不兼容async/await：没有await file.read()，也没有async for chunk in file。而且很多时候读取它们是阻塞操作（从磁盘或网络读），可能阻塞事件循环。

上面的io.BytesIO示例其实是例外：数据已在内存中，读取它不会阻塞任何东西。但很多场景下读取文件或类文件对象是会阻塞的。

避免阻塞事件循环的简单做法：把path operation function声明为普通def而不是async def，FastAPI 就会把它放到线程池 worker中执行，从而不阻塞主循环（Starlette 迭代同步生成器时走的正是这条路径）：

@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

经验法则：生成数据本身是 I/O 密集且非阻塞的（如已 await 的远程流）用async def；需要阻塞式读文件/读第三方 SDK 的用def。如果需要在 async 函数里调用阻塞代码（或反过来），可以考虑 FastAPI 的"姊妹库" Asyncer（官方文档中以 Tip 形式提及，此处不展开链接）。

yield from小技巧

当你遍历一个可迭代对象（比如类文件对象）并对每个元素做yield时，可以直接用yield from跳过for循环，逐个委托产出。这不是 FastAPI 特有的，而是 Python 语法，但用起来很顺手：

@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

四、OpenAPI 表现：流式端点在 Schema 中的样子

从 test_tutorial001.py 的 OpenAPI 快照可以看到：普通StreamingResponse（无media_type）的端点在/openapi.json中只生成responses: {"200": {"description": "Successful Response"}}，不带content段；而PNGStreamingResponse的端点则会带上：

"200": { "description": "Successful Response", "content": { "image/png": {"schema": {"type": "string"}} } }

（见 test_tutorial002.py 中的快照断言。）也就是说，子类化StreamingResponse并设置media_type后，OpenAPI Schema 会自动为该操作声明对应的响应媒体类型，客户端代码生成工具可以据此感知接口返回的是image/png而非默认 JSON。

五、小结与使用要点

关键文件索引：文档 docs/en/docs/advanced/stream-data.md；示例 docs_src/stream_data/tutorial001_py310.py、docs_src/stream_data/tutorial002_py310.py；核心实现 fastapi/routing.py（原始流式分支）与 fastapi/responses.py（StreamingResponse再导出）；测试 tests/test_tutorial/test_stream_data/test_tutorial001.py、tests/test_tutorial/test_stream_data/test_tutorial002.py。

【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi