★ FEATURED ARTICLE 模型推理服务人工智能后端大模型MLOpsLLMOps【免费下载链接】BentoMLThe easiest way to serve AI apps and models - Build Model Inference APIs, Job queues, LLM apps, Multi-model pipelines, and more!项目地址https://gitcode.com/gh_mirrors/be/BentoML点击查看免费下载本文基于 BentoML 官方文档 Stream responses 展开讲解如何用 Python 生成器Generator / AsyncGenerator在bentoml.api中实现 LLM 文本的流式输出以及如何通过挂载 FastAPI 的 WebSocket 端点实现 TTS 音频字节流。读完本文你将掌握流式接口在 BentoML 中的声明方式与自动识别机制、响应底层如何被包装为StreamingResponse、客户端如何消费流式数据以及 WebSocket 音频流的完整落地路径。为什么需要流式响应流式Streaming让你把大块或增量数据分片chunk发送给客户端而不必等整个结果生成完毕再一次性返回从而显著改善实时用户体验。BentoML 对两类典型场景提供了原生支持大语言模型LLM输出用户希望边生成边显示而不是等待完整回答音频合成TTS等实时数据语音助手、实时音频处理需要把音频字节持续推送给客户端。从源码结构看BentoML 的流式能力分为两条技术路线普通推理 API 走生成器返回值 StreamingResponse的 HTTP 流式协议而需要双工、长连接的场景如音频则走挂载 ASGI 应用 WebSocket路线。流式输出 LLM用 async generator 声明接口BentoML 中流式接口的核心约定非常简洁只要bentoml.api修饰的方法是一个生成器函数同步yield或异步生成器函数async ... yield服务端就会自动将其按流式响应处理。下面是一个基于 OpenAI API 的完整示例继承自原文档import bentoml from typing import Literal, Generator from pydantic import BaseModel # Define message structure for LLM input class Message(BaseModel): content: str role: Literal[assistant, user, system] bentoml.service class LLMExample: def __init__(self) - None: # Initialize your model configuration # A dummy example here self.model_id MODEL_ID bentoml.api async def generate(self, prompt: str) - Generator[str, None, None]: # Yields text chunks from the LLM response from openai import AsyncOpenAI # Initialize OpenAI client client AsyncOpenAI() message Message(roleuser, contentprompt) # Call OpenAIs chat completion API with streaming enabled completion await client.chat.completions.create( modelself.model_id, messages[message.model_dump()], # type: ignore streamTrue, ) # Stream and yield the response chunks async for chunk in completion: yield chunk.choices[0].delta.content or 要点说明generate是async def且包含yield即异步生成器每次yield出的字符串片段会被即时发送到客户端输入参数prompt: str会被 BentoML 的 IO 描述符系统解析为请求体字段流式输出Generator[str, None, None]决定响应以文本形式分块下发该写法对任意边生成边产出的后端都成立替换 OpenAI 为 vLLM 或其他推理引擎时接口形态不变BentoML 社区还有基于 vLLM 部署不同 LLM 的示例仓库 BentoVLLM 可参考。源码级机制流式接口是如何被识别的这一约定优于配置的行为可以在 SDK 源码中得到印证流式判定。在 src/_bentoml_sdk/method.py 中API 元数据有一个is_stream属性其默认值由函数类型自动推导is_stream.default def default_is_stream(self) - bool: return inspect.isasyncgenfunction(self.func) or inspect.isgeneratorfunction( self.func )即只要是异步生成器函数或普通生成器函数该 API 即被标记为流式无需任何额外装饰器参数。流式响应的 MIME 类型。同一文件中定义了默认流媒体类型并在 API 初始化时回填到输出规格DEFAULT_STREAM_MEDIA_TYPE text/event-stream # src/_bentoml_sdk/method.py#L29 ... def __attrs_post_init__(self) - None: if self.is_stream and not self.output_spec.media_type: self.output_spec.media_type DEFAULT_STREAM_MEDIA_TYPE这意味着流式 API 的响应头Content-Type默认是text/event-stream与主流前端/SDK 对 SSE 流的处理习惯兼容。响应包装。真正的生成器 → HTTP 流转换发生在 src/_bentoml_sdk/io_models.py 的IODescriptor.to_http_response中对异步生成器inspect.isasyncgen(obj)服务端会先await obj.__anext__()试探性地取第一个 chunk——如果生成器在首个yield之前就抛错例如初始化 OpenAI client 失败、鉴权失败错误可以在此阶段被捕获并体现在响应中而不是挂起为一个永远没有数据的流随后将后续 chunk 包装进 Starlette 的StreamingResponseasync def async_stream() - t.AsyncGenerator[str | bytes, None]: try: async for item in gen(): if isinstance(item, (str, bytes)): yield item else: obj_item (cls(item) if issubclass(cls, IORootModel) else item) for chunk in serde.serialize_model(...).data: yield chunk except Exception: logger.exception(Error while streaming response) return StreamingResponse(async_stream(), media_typecls.mime_type())对同步生成器inspect.isgenerator(obj)实现上先itertools.tee出一份试探流消费第一个元素以暴露早发错误再把原流交给content_stream()并同样以StreamingResponse返回每个 chunk 可以是str/bytes直接透传也可以是 Pydantic 模型实例经 serde 序列化后再分块下发流中途抛错会被记录日志Error while streaming response并终止该响应。客户端如何消费流式输出BentoML 内置 Python 客户端同样支持流式消费详见 Python 客户端文档 的 Streaming 一节同步客户端SyncHTTPClient流式 API 返回一个 Python generator随接收随产出with bentoml.SyncHTTPClient(base_urlhttp://localhost:3000) as client: for data_chunk in client.stream_data(): # 对应你的流式 API print(data_chunk, end, flushTrue)异步客户端AsyncHTTPClient返回 async generator可用async for迭代async with bentoml.AsyncHTTPClient(base_urlhttp://localhost:3000) as client: async for data_chunk in client.generate(prompt...): print(data_chunk, end, flushTrue)流式音频字节WebSocket FastAPI 挂载TTS、实时语音助手等场景需要双向、长连接的通信模型单纯 HTTP 流式响应不够通常要构建 WebSocket 服务器。BentoML 的做法是用 FastAPI 定义 WebSocket 端点再通过bentoml.asgi_app把 FastAPI 应用整体挂载到 BentoML Service 上完整原理见 ASGI 应用挂载文档 与 WebSocket 端点文档。原文档给出的示例如下import bentoml from fastapi import FastAPI, WebSocket from typing import Generator # Create a FastAPI app app FastAPI() bentoml.service bentoml.asgi_app(app) # Integrate FastAPI app with BentoML class TTSExample: def __init__(self) - None: # Initialize your TTS engine here self.engine self.setup_tts_engine() def setup_tts_engine(self): # Configure your TTS engine here pass def synthesize(self, text: str) - Generator[bytes, None, None]: # Implement your TTS logic here pass # Define a WebSocket endpoint for streaming audio app.websocket(/ws) async def speech(self, websocket: WebSocket): await websocket.accept() try: while True: # Receive text from client data await websocket.receive_text() # Stream audio chunks back to client for chunk in self.engine.synthesize(data): await websocket.send_bytes(chunk) except Exception as e: print(fError in WebSocket connection: {e}) finally: await websocket.close()结构与要点app FastAPI()在 Service 外部创建bentoml.asgi_app(app)负责将其挂到 Service 的 ASGI 层上。从源码看装饰器定义在 src/_bentoml_sdk/decorators.py最终调用 Service 的mount_asgi_app实现见 src/_bentoml_sdk/service/factory.py并把挂载应用的既有路由一并提取注册到 OpenAPI/路由表app.websocket(/ws)是 FastAPI 原生 WebSocket 路由路由函数定义在 Service 类内部因此可以直接用self.engine访问 TTS 引擎实例通信模式是客户端send文本 → 服务端调用self.engine.synthesize(data)得到一个同步生成器 → 循环send_bytes把音频 chunk 推回客户端连接断开时走finally关闭asgi_app支持path参数为挂载的 FastAPI 应用设置前缀。例如bentoml.asgi_app(app, path/chat)后上述端点的完整地址就是ws://localhost:3000/chat/ws前缀 路由名。由于 WebSocket 端点不属于普通 HTTP 推理 APIBentoML 的 Python 客户端尚不支持调用它需要自行实现客户端。使用websockets库的调用示例import asyncio import websockets async def test_websocket(): # /chat comes from the asgi_app path, /ws from the endpoint # Adjust URL as needed uri ws://localhost:3000/chat/ws async with websockets.connect(uri) as websocket: # Send a test message await websocket.send(Hello BentoML) response await websocket.recv() print(fResponse: {response}) # Run the test asyncio.run(test_websocket())另外bentoml.service的traffic{timeout: N}参数可调整请求超时对于长连接的 WebSocket 场景参照 WebSocket 文档 中的做法可以显式设置一个较大的超时值以避免连接被服务端超时策略提前切断。流式方案选型小结场景推荐方案关键机制LLM 文本逐词输出、大对象分块下载bentoml.api返回生成器 / async generator自动识别为流式 APIStreamingResponse下发默认text/event-streamTTS 音频推送、实时语音、双向通信FastAPIapp.websocketbentoml.asgi_app挂载ASGI 应用与 BentoML Service 并存于同一服务进程WebSocket 走双工长连接两条路线的共同前提是流式逻辑都写在 Service 内部或通过 ASGI 挂载应用访问 Service 实例从而共享模型的加载与生命周期管理。延伸阅读与验证路径服务端的流式识别与响应包装src/_bentoml_sdk/method.py、src/_bentoml_sdk/io_models.pyASGI 挂载装饰器与路由提取src/_bentoml_sdk/decorators.py、src/_bentoml_sdk/service/factory.py相关文档ASGI 应用挂载、WebSocket 端点、Python 客户端流式消费原文档还指向两个社区示例项目基于 vLLM 服务不同 LLM 的 BentoVLLM以及用开源模型构建语音 Agent 的 BentoVoiceAgent适合在掌握本文基础后进一步对照实践。赞分享模型推理服务人工智能后端大模型MLOpsLLMOps【免费下载链接】BentoMLThe easiest way to serve AI apps and models - Build Model Inference APIs, Job queues, LLM apps, Multi-model pipelines, and more!项目地址https://gitcode.com/gh_mirrors/be/BentoML点击查看免费下载相关推荐vLLM流式输出实现实时LLM响应技术细节vLLM流式输出实现实时LLM响应技术细节 引言LLM响应延迟的行业痛点 在大型语言模型LLM应用中用户体验与响应速度直接相关。传统的完整生成模式下人工智能大模型模型推理服务推理引擎本地部署从字节流到字素簇AIRI 流式文本动画与 Clustr 读取库实战解析从字节流到字素簇AIRI 流式文本动画与 Clustr 读取库实战解析 本文以 AIRI 项目 2025.08.01 DevLog 为基础展开当聊天消息、语AI 应用人工智能大模型数字人AI Agent语音前端后端桌面应用移动开发即时通讯3D渲染mistral.rs 服务端流式输出实战用 OpenAI Python 客户端实现逐 Token 流式对话mistral.rs 服务端流式输出实战用 OpenAI Python 客户端实现逐 Token 流式对话 导读 本文讲解 mistral.rs 以 Open推理引擎模型推理服务AI Agent多模态上一篇【亲测免费】 推荐开源项目FastAPI-Utils - 提升你的FastAPI开发效率下一篇【亲测免费】 探索微软开源的表格数据转换工具Table Transformer创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
阅读完成 · 觉得有帮助？
