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 中纯字符串与纯二进制数据的流式传输方案。当数据可以被结构化表达为 JSON 时，应优先采用 JSON Lines 流式传输；而当你要流式输出AI LLM 生成的文本、大文件、视频/音频等二进制或非结构化内容时，本文所讲的StreamingResponse+yield组合就是标准解法。读完本文，你将掌握如何用响应类声明、生成器函数实现逐块发送、处理阻塞式文件读取，以及通过自定义响应子类精准控制Content-Type。

该功能在FastAPI 0.134.0中正式引入，本文对应的源码示例位于 docs_src/stream_data/，并且英文原文（docs/en/docs/advanced/stream-data.md）与韩文翻译（docs/ko/docs/advanced/stream-data.md）内容一致。

典型使用场景

StreamingResponse的流式能力尤其适合以下场景：

• AI LLM 服务输出：直接从大模型服务的输出流中取出纯字符串，逐块转发给客户端，让用户"边生成边看到"结果；
• 大文件传输：读取超大二进制文件时，不必一次性把整个文件载入内存，而是每读到一块数据就立刻流式发送一块；
• 音视频流：以同样的方式流式推送视频、音频内容，甚至可以在"处理一边生成一边发送"；
• 实时生成数据：需要边计算边下发的任意数据块。

需要明确的是：如果你的数据能天然结构化为 JSON（例如一行一个 JSON 对象），请优先参考 JSON Lines 流式传输教程；本文处理的是JSON 之外的"裸数据"——字符串和二进制字节。

基础用法：response_class=StreamingResponse配合yield

流式传输的核心模式非常简单：在路径处理函数中声明response_class=StreamingResponse，然后在函数体内用yield依次产出每一个数据块。

完整的首个示例代码见 docs_src/stream_data/tutorial001_py310.py，核心部分如下：

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. You got--... you gotta come with me. Morty: (rubs his eyes) What, Rick? What's going on? Rick: I got a surprise for you, Morty. Morty: It's the middle of the night. What are you talking about? Rick: (spills alcohol on Morty's bed) Come on, I got a surprise for you. (drags Morty by the ankle) Come on, hurry up. (pulls Morty out of his bed and into the hall) Morty: Ow! Ow! You're tugging me too hard! Rick: We gotta go, gotta get outta here, come on. Got a surprise for you Morty. """ @app.get("/story/stream", response_class=StreamingResponse) async def stream_story() -> AsyncIterable[str]: for line in message.splitlines(): yield line

理解这个机制的关键在于：FastAPI 会把yield出来的每一个数据块原封不动地交给StreamingResponse，完全不会试图把它转换成 JSON 或做任何形式的序列化。因此字符串、bytes、迭代器产出的任意内容都会以最原始的形式流向客户端。

从框架实现上看，StreamingResponse在 fastapi/responses.py 中被直接再导出（re-export），其底层由 Starlette 提供；在 fastapi/routing.py 中，FastAPI 会根据路径处理函数是异步生成器还是普通生成器选择相应的迭代与发送策略，这正是yield模式得以无缝工作的原因。

非 async 的路径处理函数同样可行

如果你写的是不带async的普通def函数，同样可以使用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 不会用 Pydantic 把数据转成 JSON，也不会做任何序列化，这里的类型注解仅对编辑器与静态检查工具有意义，FastAPI 本身不会使用它：

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

这意味着当你使用StreamingResponse时，"按传输要求精确地生成并编码字节数据"的自由与责任完全交到了你手上，类型注解不会帮你兜底。🤓

流式发送字节数据（bytes）

纯字符串之外，流式发送bytes是本文档列出的核心用法之一。你只需在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")

同样的逻辑也可以组合出多种变体：非 async 版本、无注解版本、二者皆无版本等，仓库示例 docs_src/stream_data/tutorial001_py310.py 中逐一给出了这些组合，可在实际项目中按需对照使用。

自定义响应类：构建PNGStreamingResponse

上面几个示例虽然完成了字节流式传输，但响应中没有携带Content-Type头，客户端无从得知收到的到底是什么类型的数据。

解决办法是：创建StreamingResponse的子类，通过media_type属性把Content-Type设为你实际流式传输的数据类型。例如构造一个把Content-Type固定为image/png的PNGStreamingResponse：

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

随后在路径处理函数中通过response_class=PNGStreamingResponse启用它：

@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

完整示例见 docs_src/stream_data/tutorial002_py310.py。仓库的测试 tests/test_tutorial/test_stream_data/test_tutorial002.py 正是这样验证的：对/image/stream等全部端点断言response.headers["content-type"] == "image/png"，并校验返回的response.content与原始binary_image字节完全一致，同时 OpenAPI 文档中也正确出现了image/png的响应内容类型。

用io.BytesIO模拟文件

示例中的文件其实是用io.BytesIO模拟的——它是一种只存在于内存中、却提供与真实文件相同接口的文件类对象（file-like object），例如可以像遍历文件那样for chunk in image_file:逐段消费其内容：

import base64 from collections.abc import AsyncIterable, Iterable from io import BytesIO from fastapi import FastAPI from fastapi.responses import StreamingResponse # image_base64 是一张用 Base64 编码的图片字符串（省略中间冗长内容） 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 编码、再解码为字节，然后交给io.BytesIO——这样整个示例可以完整地存放在单个文件中，方便你直接复制运行。🥚

使用with块的好处是：确保生成器函数（含yield的函数）结束后、也就是响应数据发送完毕之后，文件类对象会被可靠地关闭。在本例中由于是内存里的假文件（io.BytesIO），关闭与否影响不大；但在操作真实文件时，工作完成后确保文件被关闭至关重要。

文件读取与 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 对同步生成器的内部处理策略一致——在 fastapi/routing.py 中可以看到，普通（非 async）生成器会通过线程池方式迭代消费，而异步生成器则直接异步迭代，两者各得其所。

小技巧：如果你需要在 async 函数内部调用阻塞代码，或在阻塞函数内部调用 async 代码，可以考虑 FastAPI 的姊妹库Asyncer，它能帮助你平滑地在两种语境间切换。

用yield from省掉 for 循环

当你遍历某个可迭代对象（比如文件类对象）、并打算把每一项逐个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

这个特性并不是 FastAPI 特有的，而是纯 Python 语法糖，但它是非常实用的技巧。😎

验证与测试依据

仓库为这两份教程示例都提供了完整的端到端测试，可直接作为行为契约参考：

• tests/test_tutorial/test_stream_data/test_tutorial001.py：对/story/stream及其非 async、无注解、字节流等 8 个路径逐一发起请求，断言状态码为 200 且返回文本与预期逐字一致；同时校验生成的 OpenAPI schema 结构正确。
• tests/test_tutorial/test_stream_data/test_tutorial002.py：对/image/stream及其变体端点（含yield from版本）验证content-type为image/png、响应字节与原始图片字节完全一致。

你可以通过如下方式直接体验示例：

1. 查看源码示例文件 docs_src/stream_data/tutorial001_py310.py 与 docs_src/stream_data/tutorial002_py310.py；
2. 使用 FastAPI 自带的TestClient（见上述测试文件中的用法）或启动服务器后访问对应端点，观察数据是否逐块到达；
3. 在浏览器或curl中访问/openapi.json，可以看到文档对 200 响应如实标注了image/png内容类型——这正是media_type属性生效的直接体现。

小结

在 FastAPI 中做"裸数据"流式传输的完整配方可以概括为三点：

1. 声明响应类：在路径处理函数上用response_class=StreamingResponse（或其带media_type的自定义子类）；
2. 用yield逐块产出：每个yield产出的字符串或字节块都会被原样、不经过 JSON 化地发送给客户端；
3. 按需选择async def或def：对于会阻塞的文件/网络读取，用普通def让 FastAPI 在线程池中执行，避免拖垮事件循环，并借助with与yield from优雅地管理资源与迭代。

这套方案覆盖了 AI 生成文本实时推送、超大文件零整载传输、音视频边生成边发送等高频需求，是构建高吞吐、低延迟数据管道的基础设施级能力。

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