尧图网站建设 
FastAPI 流式 JSON LinesStream JSON Lines实战用 yield 构造逐行推送的 application/jsonl 接口【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi流式streaming响应能够让服务端在整批数据尚未就绪时就开始逐条向客户端发送内容是 AI 大模型逐 token/逐条推理结果、日志与遥测数据实时推送等场景的常见做法。FastAPI 自 0.134.0 起原生支持以JSON LinesJSONL格式返回流式数据——只要在路径操作函数里用yield取代return即可得到application/jsonl类型的逐行 JSON 响应同时保留类型校验、OpenAPI 文档生成与响应过滤能力。读完本文你将掌握 JSONL 流式接口的四种写法async/同步 × 有无返回类型标注并通过阅读 FastAPI 路由与 OpenAPI 源码理解其底层识别、校验、序列化与文档生成机制。什么是流Stream流式发送数据的含义是应用不需要等整批数据全部生成完毕就可以先向客户端发送第一条数据项。随后客户端收到并开始处理该条数据的同时服务端仍在继续生成下一条。换句话说客户端与服务端是边生产、边消费并行推进的而不是全部算完再一次性传输。从上图的时序关系可以看到真正的价值在于管道化发送第 N 条与客户端处理第 N-1 条是重叠进行的。这种模式甚至可以是一个无限流——只要客户端不断读取服务端就一直持续产生并发送数据。JSON Lines 是什么JSON LinesJSONL是适合上述流式场景的一种序列化格式每一行是一个独立的 JSON 对象。对应到 HTTP 上一个 JSONL 响应的 Content-Type 是application/jsonl而不是常见的application/json响应体大致如下{name: Plumbus, description: A multi-purpose household device.} {name: Portal Gun, description: A portal opening device.} {name: Meeseeks Box, description: A box that summons a Meeseeks.}它和 JSON 数组对应 Python 的 list非常相似区别在于JSON 数组用[]包裹、条目之间用,分隔而 JSONL每行一个对象用换行符分隔。逐行分发的关键价值是服务端可以按顺序一行一行地生成而客户端可以同步地消费前面已经收到的行。一个关键技术细节换行符处理因为每个 JSON 对象之间靠换行符分隔所以对象内容里不能包含字面的换行字符不过对象内部可以通过 JSON 标准自带的转义序列\n来表示换行这并不违反格式约束。好消息是当你使用 FastAPI Pydantic 时这一步会由序列化过程自动完成对象会被序列化为单行的 JSON 文本通常不需要手工操心。典型使用场景JSONL 流式响应非常适合产出由多个结构化的 JSON 条目组成、且希望尽早送达的数据典型包括AI / LLM 服务逐个输出推理结果、候选片段或工具调用结构日志logs与遥测telemetry持续推送新的日志行或监控指标项其他可按 JSON 条目结构化的连续数据。需要特别说明的是如果你要流式传输的是二进制数据如视频、音频JSONL 并不是合适的载体应参考进阶指南 Stream Data流式传输二进制数据。用 FastAPI 输出 JSON Lines完整示例FastAPI 输出 JSONL 流的做法非常直观在路径操作函数中用yield代替return每个被yield出来的对象就是一行 JSONL。本节完整代码来自仓库示例 docs_src/stream_json_lines/tutorial001_py310.py文件名标注py310使用了 Python 3.10 的类型语法collections.abc与str | None联合类型写法from collections.abc import AsyncIterable, Iterable from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class Item(BaseModel): name: str description: str | None items [ Item(namePlumbus, descriptionA multi-purpose household device.), Item(namePortal Gun, descriptionA portal opening device.), Item(nameMeeseeks Box, descriptionA box that summons a Meeseeks.), ] app.get(/items/stream) async def stream_items() - AsyncIterable[Item]: for item in items: yield item app.get(/items/stream-no-async) def stream_items_no_async() - Iterable[Item]: for item in items: yield item app.get(/items/stream-no-annotation) async def stream_items_no_annotation(): for item in items: yield item app.get(/items/stream-no-async-no-annotation) def stream_items_no_async_no_annotation(): for item in items: yield item这个文件一共演示了四种排列组合下文逐一解释。启动方式与仓库其他教程一致参见 docs/en/docs/tutorial/first-steps.md例如uv run fastapi dev docs_src/stream_json_lines/tutorial001_py310.py写法一async 函数 AsyncIterable[Item]返回类型当每个要发送的 JSON 条目是类型ItemPydantic 模型且路径操作函数是异步函数时可以把返回类型声明为AsyncIterable[Item]app.get(/items/stream) async def stream_items() - AsyncIterable[Item]: for item in items: yield item只要声明了返回类型FastAPI 就会用该类型完成四件事校验validate逐条校验被 yield 的数据是否符合Item文档document在 OpenAPI schema 中记录该响应使 /docs 交互式文档能正确展示 JSONL 响应结构过滤filter按response_model_include/response_model_exclude等响应模型参数对字段进行取舍序列化serialize交由 Pydantic 完成 JSON 序列化。关于性能官方文档提示因为 Pydanticv2的序列化在其基于 Rust 的 pydantic-core 侧完成声明返回类型会比不声明时获得高得多的序列化性能。这一点建立在类型被提前解析为序列化字段的基础之上见下文源码分析。写法二普通def函数 Iterable[Item]如果路径操作函数是不带async的普通函数同样可以这样yield。FastAPI 会确保这类同步生成器被正确调度不会阻塞事件循环生成器会在合适的执行环境中运行见下文 routing.py 的处理分支。这种情况下对应的返回类型应写为Iterable[Item]app.get(/items/stream-no-async) def stream_items_no_async() - Iterable[Item]: for item in items: yield item写法三/四不声明返回类型也可以完全省略返回类型。此时 FastAPI 无法得知每个条目的静态类型于是会退回到jsonable_encoder把每个yield出来的对象转换为可 JSON 序列化的形式再按行输出app.get(/items/stream-no-annotation) async def stream_items_no_annotation(): for item in items: yield item同步版本同理app.get(/items/stream-no-async-no-annotation) def stream_items_no_async_no_annotation(): for item in items: yield item省略返回类型虽然写法最简但会失去逐条校验、OpenAPI 条目 schema 文档、字段过滤以及 Pydantic 快速序列化这四项收益因此建议优先声明返回类型。底层原理FastAPI 如何识别一个JSONL 生成器端点当函数内出现yield时它就是一个生成器函数。FastAPI 在路由构建阶段fastapi/routing.py会专门检测这类端点并为其选择流式响应路径而不是走普通的 JSON 返回逻辑。第 1 步识别生成器并区分 JSONL / SSE从源码可以确认路由构建逻辑fastapi/routing.py执行如下判定# Detect generator endpoints that should stream as JSONL or SSE is_generator _is_async_gen_callable(route.dependant.call) or _is_gen_callable( route.dependant.call ) route.is_sse_stream is_generator and lenient_issubclass( response_class, EventSourceResponse ) route.is_json_stream is_generator and isinstance( response_class, DefaultPlaceholder )也就是说端点只要是一个生成器函数async 生成器或同步生成器就会进入流式处理当用户没有显式指定response_class即默认占位DefaultPlaceholder时is_json_stream为真走 JSONL 分支当response_class是EventSourceResponse时则走 SSEServer-Sent Events 分支。第 2 步从返回类型注解中提取条目类型端点带返回类型注解时FastAPI 借助 fastapi/dependencies/utils.py 中的get_typed_return_annotation解析真正的返回注解再由get_stream_item_type从AsyncIterable[Item]/Iterable[Item]等泛型中提取出元素类型Item_STREAM_ORIGINS { AsyncIterable, AsyncIterator, AsyncGenerator, Iterable, Iterator, Generator, } def get_stream_item_type(annotation: Any) - Any | None: origin get_origin(annotation) if origin is not None and origin in _STREAM_ORIGINS: type_args get_args(annotation) if type_args: return type_args[0] return Any return None可见只要是AsyncIterable、AsyncIterator、AsyncGenerator、Iterable、Iterator、Generator这六类流式类型来源为collections.abc或typingFastAPI 都会尝试取出其第一个类型实参作为条目类型若是裸的AsyncIterable而没有参数则回退为Any。随后fastapi/routing.py该条目类型会被构造成序列化模式的ModelFieldstream_item_field供后续逐条校验与序列化使用。第 3 步逐条校验 序列化 换行在真正响应时FastAPI 使用一个共享的流条目序列化器fastapi/routing.py声明了返回类型存在stream_item_field先用stream_item_field.validate(...)对每条数据做响应校验校验失败会抛出ResponseValidationError通过后调用serialize_json(...)并把response_model_include/response_model_exclude/by_alias/exclude_unset/exclude_defaults/exclude_none等过滤选项一并传入——这与普通非流式响应的响应模型过滤语义完全一致未声明返回类型走jsonable_encoder(data)再json.dumps(...).encode(utf-8)的兜底路径。序列化得到的每个字节块随后都会追加一个换行符b\ndef _serialize_item(item: Any) - bytes: return _serialize_data(item) b\n这正是每行一个 JSON 对象这一 JSONL 格式约束的来源。第 4 步async 与同步生成器的分发 StreamingResponse在 JSONL 响应分支fastapi/routing.py中FastAPI 会区分生成器类型若是async 生成器则构造_async_stream_jsonl()以async for逐条取出并序列化且每次 yield 后插入await anyio.sleep(0)作为检查点checkpoint保证即使生产者快于消费者、receive()不挂起时取消cancellation信号也能被及时送达——对应流被提前中断时的资源回收场景若是同步生成器普通def端点的yield则构造_sync_stream_jsonl()同步迭代。最终两者都被包装进StreamingResponse并把媒体类型明确设置为response StreamingResponse( jsonl_stream_content, media_typeapplication/jsonl, **response_args, )application/jsonl这一 Content-Type 也正是客户端区分普通 JSON 数组与 JSONL 流的关键信号。第 5 步写入 OpenAPI schemaFastAPI 还会把 JSONL 流式响应文档化到 OpenAPIfastapi/openapi/utils.py在响应的content中登记application/jsonl并把条目 schema 放入itemSchemaif route.is_json_stream: jsonl_content: dict[str, Any] {} if route.stream_item_field: item_schema get_schema_from_model_field(...) jsonl_content[itemSchema] item_schema else: jsonl_content[itemSchema] {} operation.setdefault(responses, {}).setdefault( status_code, {} ).setdefault(content, {})[application/jsonl] jsonl_content也就是说声明了返回类型时itemSchema是对Item等模型的$ref引用/docs 界面能完整呈现每个条目的字段结构没有声明时itemSchema为空对象。仓库测试 tests/test_stream_bare_type.py 对该行为有直接断言assert jsonl_response { description: Successful Response, content: { application/jsonl: {itemSchema: {$ref: #/components/schemas/Item}} }, }客户端如何消费 JSONL 流逐行解析验证由于 JSONL 响应体是一行一个 JSON 对象客户端消费方式非常朴素按换行符切分响应文本再对每一行做json.loads。仓库测试 tests/test_stream_bare_type.py 用 FastAPI 自带的TestClient给出了可直接照搬的消费模式def test_stream_bare_async_iterable(): response client.get(/items/stream-bare-async) assert response.status_code 200 assert response.headers[content-type] application/jsonl lines [json.loads(line) for line in response.text.strip().splitlines()] assert lines [{name: foo}]这段测试同时验证了三件事状态码为 200、content-type精确等于application/jsonl、每一行都是合法的、可独立解析的 JSON。真实浏览器端或下游服务拿到该接口后也可以按同样的读一行、解析一行、处理一行策略实现边收边处理。边界行为条目校验失败与流取消了解了底层实现两个边界行为就很好理解了逐条响应校验声明了返回类型后如果yield出的数据无法通过条目类型校验会在序列化阶段抛出ResponseValidationError见 fastapi/routing.py这与普通响应模型的校验语义保持一致。仓库在 tests/test_stream_json_validation_error.py 中专门覆盖了流式响应包含非法条目的场景。流取消与中断客户端提前断开时FastAPI 需要能取消仍在运行的生成器。前面提到的anyio.sleep(0)检查点保证了取消能在生成器迭代过程中被及时注入tests/test_stream_cancellation.py 通过_run_asgi_and_cancel对/stream-jsonl这类端点验证了取消路径不会挂起或泄漏资源。自定义状态码与依赖覆盖JSONL 端点同样支持status_code、responses、依赖返回 header 等常规能力。tests/test_stream_status_code.py 中就有app.post(/jsonl, status_code201)以及依赖覆盖响应描述 / 状态码的 JSONL 用例其 OpenAPI schema 同样以application/jsonl呈现。这说明 JSONL 流式端点与普通端点在响应配置维度上是平权的。与 Server-Sent EventsSSE的区别JSONL 并非流式响应的唯一选择。FastAPI 还提供对Server-Sent EventsSSE的一等支持二者都是持续推送多条数据但 SSE 增加了事件名、事件 ID、重连提示等额外字段并使用text/event-stream媒体类型与data:前缀的消息帧。如果你需要的是更接近事件总线语义的推送或者希望浏览器端用原生EventSource直接消费可以参考下一篇教程 Server-Sent Events (SSE)而如果你的数据天然就是一批结构相同的 JSON 条目JSONL 是更直接、更轻量的选择。小结本文围绕 FastAPI 0.134.0 起提供的 JSON Lines 流式响应能力从为什么需要流与JSONL 格式长什么样出发完整讲解了yield型路径操作函数的四种组合写法async/同步 × 有无返回类型并沿着 fastapi/routing.py 的构建与响应链路、fastapi/dependencies/utils.py 的流类型提取、fastapi/openapi/utils.py 的 schema 登记梳理了生成器端点从识别为 JSONL 流到逐条校验、序列化、追加换行、包装成application/jsonl的 StreamingResponse的完整调用链。配套测试文件展示了逐行解析消费、条目校验失败与流取消等真实边界行为可直接作为你实现 AI 结果流式输出或日志遥测推送接口的参照。【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 FastAPI 流式 JSON Lines（Stream JSON Lines）实战：用 yield 构造逐行推送的 application/jsonl 接口
FastAPI 流式 JSON LinesStream JSON Lines实战用 yield 构造逐行推送的 application/jsonl 接口【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi流式streaming响应能够让服务端在整批数据尚未就绪时就开始逐条向客户端发送内容是 AI 大模型逐 token/逐条推理结果、日志与遥测数据实时推送等场景的常见做法。FastAPI 自 0.134.0 起原生支持以JSON LinesJSONL格式返回流式数据——只要在路径操作函数里用yield取代return即可得到application/jsonl类型的逐行 JSON 响应同时保留类型校验、OpenAPI 文档生成与响应过滤能力。读完本文你将掌握 JSONL 流式接口的四种写法async/同步 × 有无返回类型标注并通过阅读 FastAPI 路由与 OpenAPI 源码理解其底层识别、校验、序列化与文档生成机制。什么是流Stream流式发送数据的含义是应用不需要等整批数据全部生成完毕就可以先向客户端发送第一条数据项。随后客户端收到并开始处理该条数据的同时服务端仍在继续生成下一条。换句话说客户端与服务端是边生产、边消费并行推进的而不是全部算完再一次性传输。从上图的时序关系可以看到真正的价值在于管道化发送第 N 条与客户端处理第 N-1 条是重叠进行的。这种模式甚至可以是一个无限流——只要客户端不断读取服务端就一直持续产生并发送数据。JSON Lines 是什么JSON LinesJSONL是适合上述流式场景的一种序列化格式每一行是一个独立的 JSON 对象。对应到 HTTP 上一个 JSONL 响应的 Content-Type 是application/jsonl而不是常见的application/json响应体大致如下{name: Plumbus, description: A multi-purpose household device.} {name: Portal Gun, description: A portal opening device.} {name: Meeseeks Box, description: A box that summons a Meeseeks.}它和 JSON 数组对应 Python 的 list非常相似区别在于JSON 数组用[]包裹、条目之间用,分隔而 JSONL每行一个对象用换行符分隔。逐行分发的关键价值是服务端可以按顺序一行一行地生成而客户端可以同步地消费前面已经收到的行。一个关键技术细节换行符处理因为每个 JSON 对象之间靠换行符分隔所以对象内容里不能包含字面的换行字符不过对象内部可以通过 JSON 标准自带的转义序列\n来表示换行这并不违反格式约束。好消息是当你使用 FastAPI Pydantic 时这一步会由序列化过程自动完成对象会被序列化为单行的 JSON 文本通常不需要手工操心。典型使用场景JSONL 流式响应非常适合产出由多个结构化的 JSON 条目组成、且希望尽早送达的数据典型包括AI / LLM 服务逐个输出推理结果、候选片段或工具调用结构日志logs与遥测telemetry持续推送新的日志行或监控指标项其他可按 JSON 条目结构化的连续数据。需要特别说明的是如果你要流式传输的是二进制数据如视频、音频JSONL 并不是合适的载体应参考进阶指南 Stream Data流式传输二进制数据。用 FastAPI 输出 JSON Lines完整示例FastAPI 输出 JSONL 流的做法非常直观在路径操作函数中用yield代替return每个被yield出来的对象就是一行 JSONL。本节完整代码来自仓库示例 docs_src/stream_json_lines/tutorial001_py310.py文件名标注py310使用了 Python 3.10 的类型语法collections.abc与str | None联合类型写法from collections.abc import AsyncIterable, Iterable from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class Item(BaseModel): name: str description: str | None items [ Item(namePlumbus, descriptionA multi-purpose household device.), Item(namePortal Gun, descriptionA portal opening device.), Item(nameMeeseeks Box, descriptionA box that summons a Meeseeks.), ] app.get(/items/stream) async def stream_items() - AsyncIterable[Item]: for item in items: yield item app.get(/items/stream-no-async) def stream_items_no_async() - Iterable[Item]: for item in items: yield item app.get(/items/stream-no-annotation) async def stream_items_no_annotation(): for item in items: yield item app.get(/items/stream-no-async-no-annotation) def stream_items_no_async_no_annotation(): for item in items: yield item这个文件一共演示了四种排列组合下文逐一解释。启动方式与仓库其他教程一致参见 docs/en/docs/tutorial/first-steps.md例如uv run fastapi dev docs_src/stream_json_lines/tutorial001_py310.py写法一async 函数 AsyncIterable[Item]返回类型当每个要发送的 JSON 条目是类型ItemPydantic 模型且路径操作函数是异步函数时可以把返回类型声明为AsyncIterable[Item]app.get(/items/stream) async def stream_items() - AsyncIterable[Item]: for item in items: yield item只要声明了返回类型FastAPI 就会用该类型完成四件事校验validate逐条校验被 yield 的数据是否符合Item文档document在 OpenAPI schema 中记录该响应使 /docs 交互式文档能正确展示 JSONL 响应结构过滤filter按response_model_include/response_model_exclude等响应模型参数对字段进行取舍序列化serialize交由 Pydantic 完成 JSON 序列化。关于性能官方文档提示因为 Pydanticv2的序列化在其基于 Rust 的 pydantic-core 侧完成声明返回类型会比不声明时获得高得多的序列化性能。这一点建立在类型被提前解析为序列化字段的基础之上见下文源码分析。写法二普通def函数 Iterable[Item]如果路径操作函数是不带async的普通函数同样可以这样yield。FastAPI 会确保这类同步生成器被正确调度不会阻塞事件循环生成器会在合适的执行环境中运行见下文 routing.py 的处理分支。这种情况下对应的返回类型应写为Iterable[Item]app.get(/items/stream-no-async) def stream_items_no_async() - Iterable[Item]: for item in items: yield item写法三/四不声明返回类型也可以完全省略返回类型。此时 FastAPI 无法得知每个条目的静态类型于是会退回到jsonable_encoder把每个yield出来的对象转换为可 JSON 序列化的形式再按行输出app.get(/items/stream-no-annotation) async def stream_items_no_annotation(): for item in items: yield item同步版本同理app.get(/items/stream-no-async-no-annotation) def stream_items_no_async_no_annotation(): for item in items: yield item省略返回类型虽然写法最简但会失去逐条校验、OpenAPI 条目 schema 文档、字段过滤以及 Pydantic 快速序列化这四项收益因此建议优先声明返回类型。底层原理FastAPI 如何识别一个JSONL 生成器端点当函数内出现yield时它就是一个生成器函数。FastAPI 在路由构建阶段fastapi/routing.py会专门检测这类端点并为其选择流式响应路径而不是走普通的 JSON 返回逻辑。第 1 步识别生成器并区分 JSONL / SSE从源码可以确认路由构建逻辑fastapi/routing.py执行如下判定# Detect generator endpoints that should stream as JSONL or SSE is_generator _is_async_gen_callable(route.dependant.call) or _is_gen_callable( route.dependant.call ) route.is_sse_stream is_generator and lenient_issubclass( response_class, EventSourceResponse ) route.is_json_stream is_generator and isinstance( response_class, DefaultPlaceholder )也就是说端点只要是一个生成器函数async 生成器或同步生成器就会进入流式处理当用户没有显式指定response_class即默认占位DefaultPlaceholder时is_json_stream为真走 JSONL 分支当response_class是EventSourceResponse时则走 SSEServer-Sent Events 分支。第 2 步从返回类型注解中提取条目类型端点带返回类型注解时FastAPI 借助 fastapi/dependencies/utils.py 中的get_typed_return_annotation解析真正的返回注解再由get_stream_item_type从AsyncIterable[Item]/Iterable[Item]等泛型中提取出元素类型Item_STREAM_ORIGINS { AsyncIterable, AsyncIterator, AsyncGenerator, Iterable, Iterator, Generator, } def get_stream_item_type(annotation: Any) - Any | None: origin get_origin(annotation) if origin is not None and origin in _STREAM_ORIGINS: type_args get_args(annotation) if type_args: return type_args[0] return Any return None可见只要是AsyncIterable、AsyncIterator、AsyncGenerator、Iterable、Iterator、Generator这六类流式类型来源为collections.abc或typingFastAPI 都会尝试取出其第一个类型实参作为条目类型若是裸的AsyncIterable而没有参数则回退为Any。随后fastapi/routing.py该条目类型会被构造成序列化模式的ModelFieldstream_item_field供后续逐条校验与序列化使用。第 3 步逐条校验 序列化 换行在真正响应时FastAPI 使用一个共享的流条目序列化器fastapi/routing.py声明了返回类型存在stream_item_field先用stream_item_field.validate(...)对每条数据做响应校验校验失败会抛出ResponseValidationError通过后调用serialize_json(...)并把response_model_include/response_model_exclude/by_alias/exclude_unset/exclude_defaults/exclude_none等过滤选项一并传入——这与普通非流式响应的响应模型过滤语义完全一致未声明返回类型走jsonable_encoder(data)再json.dumps(...).encode(utf-8)的兜底路径。序列化得到的每个字节块随后都会追加一个换行符b\ndef _serialize_item(item: Any) - bytes: return _serialize_data(item) b\n这正是每行一个 JSON 对象这一 JSONL 格式约束的来源。第 4 步async 与同步生成器的分发 StreamingResponse在 JSONL 响应分支fastapi/routing.py中FastAPI 会区分生成器类型若是async 生成器则构造_async_stream_jsonl()以async for逐条取出并序列化且每次 yield 后插入await anyio.sleep(0)作为检查点checkpoint保证即使生产者快于消费者、receive()不挂起时取消cancellation信号也能被及时送达——对应流被提前中断时的资源回收场景若是同步生成器普通def端点的yield则构造_sync_stream_jsonl()同步迭代。最终两者都被包装进StreamingResponse并把媒体类型明确设置为response StreamingResponse( jsonl_stream_content, media_typeapplication/jsonl, **response_args, )application/jsonl这一 Content-Type 也正是客户端区分普通 JSON 数组与 JSONL 流的关键信号。第 5 步写入 OpenAPI schemaFastAPI 还会把 JSONL 流式响应文档化到 OpenAPIfastapi/openapi/utils.py在响应的content中登记application/jsonl并把条目 schema 放入itemSchemaif route.is_json_stream: jsonl_content: dict[str, Any] {} if route.stream_item_field: item_schema get_schema_from_model_field(...) jsonl_content[itemSchema] item_schema else: jsonl_content[itemSchema] {} operation.setdefault(responses, {}).setdefault( status_code, {} ).setdefault(content, {})[application/jsonl] jsonl_content也就是说声明了返回类型时itemSchema是对Item等模型的$ref引用/docs 界面能完整呈现每个条目的字段结构没有声明时itemSchema为空对象。仓库测试 tests/test_stream_bare_type.py 对该行为有直接断言assert jsonl_response { description: Successful Response, content: { application/jsonl: {itemSchema: {$ref: #/components/schemas/Item}} }, }客户端如何消费 JSONL 流逐行解析验证由于 JSONL 响应体是一行一个 JSON 对象客户端消费方式非常朴素按换行符切分响应文本再对每一行做json.loads。仓库测试 tests/test_stream_bare_type.py 用 FastAPI 自带的TestClient给出了可直接照搬的消费模式def test_stream_bare_async_iterable(): response client.get(/items/stream-bare-async) assert response.status_code 200 assert response.headers[content-type] application/jsonl lines [json.loads(line) for line in response.text.strip().splitlines()] assert lines [{name: foo}]这段测试同时验证了三件事状态码为 200、content-type精确等于application/jsonl、每一行都是合法的、可独立解析的 JSON。真实浏览器端或下游服务拿到该接口后也可以按同样的读一行、解析一行、处理一行策略实现边收边处理。边界行为条目校验失败与流取消了解了底层实现两个边界行为就很好理解了逐条响应校验声明了返回类型后如果yield出的数据无法通过条目类型校验会在序列化阶段抛出ResponseValidationError见 fastapi/routing.py这与普通响应模型的校验语义保持一致。仓库在 tests/test_stream_json_validation_error.py 中专门覆盖了流式响应包含非法条目的场景。流取消与中断客户端提前断开时FastAPI 需要能取消仍在运行的生成器。前面提到的anyio.sleep(0)检查点保证了取消能在生成器迭代过程中被及时注入tests/test_stream_cancellation.py 通过_run_asgi_and_cancel对/stream-jsonl这类端点验证了取消路径不会挂起或泄漏资源。自定义状态码与依赖覆盖JSONL 端点同样支持status_code、responses、依赖返回 header 等常规能力。tests/test_stream_status_code.py 中就有app.post(/jsonl, status_code201)以及依赖覆盖响应描述 / 状态码的 JSONL 用例其 OpenAPI schema 同样以application/jsonl呈现。这说明 JSONL 流式端点与普通端点在响应配置维度上是平权的。与 Server-Sent EventsSSE的区别JSONL 并非流式响应的唯一选择。FastAPI 还提供对Server-Sent EventsSSE的一等支持二者都是持续推送多条数据但 SSE 增加了事件名、事件 ID、重连提示等额外字段并使用text/event-stream媒体类型与data:前缀的消息帧。如果你需要的是更接近事件总线语义的推送或者希望浏览器端用原生EventSource直接消费可以参考下一篇教程 Server-Sent Events (SSE)而如果你的数据天然就是一批结构相同的 JSON 条目JSONL 是更直接、更轻量的选择。小结本文围绕 FastAPI 0.134.0 起提供的 JSON Lines 流式响应能力从为什么需要流与JSONL 格式长什么样出发完整讲解了yield型路径操作函数的四种组合写法async/同步 × 有无返回类型并沿着 fastapi/routing.py 的构建与响应链路、fastapi/dependencies/utils.py 的流类型提取、fastapi/openapi/utils.py 的 schema 登记梳理了生成器端点从识别为 JSONL 流到逐条校验、序列化、追加换行、包装成application/jsonl的 StreamingResponse的完整调用链。配套测试文件展示了逐行解析消费、条目校验失败与流取消等真实边界行为可直接作为你实现 AI 结果流式输出或日志遥测推送接口的参照。【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考