스트리밍 응답 (Stream responses)

LLM 출력이나 오디오 합성처럼 결과가 크거나 점진적으로 생기는 경우, 응답을 한 번에 몰아서 주는 대신 조각으로 나눠 보내면 사용자에게 훨씬 나은 실시간 경험을 줄 수 있어요. BentoML은 이런 스트리밍 응답을 지원하는데, Python 제너레이터를 쓰면 LLM 출력을, WebSocket을 쓰면 오디오 바이트를 흘려보낼 수 있답니다.

출처: https://docs.bentoml.com/en/latest/build-with-bentoml/streaming.html

LLM 출력 스트리밍

BentoML에서 LLM 출력은 Python 제너레이터로 스트리밍해요. 아래는 OpenAI API를 사용하는 예제예요.

import bentoml
from typing import Literal, Generator
from pydantic import BaseModel

class Message(BaseModel):
    content: str
    role: Literal['assistant', 'user', 'system']

@bentoml.service
class LLMExample:
    def __init__(self) -> None:
        self.model_id = MODEL_ID

    @bentoml.api
    async def generate(self, prompt: str) -> Generator[str, None, None]:
        from openai import AsyncOpenAI

        client = AsyncOpenAI()
        message = Message(role="user", content=prompt)

        completion = await client.chat.completions.create(
            model=self.model_id,
            messages=[message.model_dump()],
            stream=True,
        )

        async for chunk in completion:
            yield chunk.choices[0].delta.content or ""

포인트는 두 가지예요. @bentoml.api 메서드의 반환 타입을 Generator로 선언하면 응답이 스트리밍되도록 만들어 주고, 실제로는 OpenAI 클라이언트에 stream=True를 넘긴 chat completion 응답을 async for로 순회하면서 조각(delta.content)을 yield로 하나씩 내보내요. 즉 BentoML이 이 제너레이터를 HTTP 스트리밍 응답으로 바꿔 주는 거죠.

오디오 바이트 스트리밍

오디오 스트리밍은 텍스트-음성 변환(TTS), 실시간 음성 어시스턴트, 실시간 오디오 처리 같은 서비스에 필수예요. 이런 경우 보통 WebSocket 서버를 만들어 오디오 데이터를 클라이언트로 흘려보내야 해요. 아래는 BentoML에서 WebSocket 서버로 오디오 바이트를 스트리밍하는 예제예요.

import bentoml
from fastapi import FastAPI, WebSocket
from typing import Generator

app = FastAPI()

@bentoml.service
@bentoml.asgi_app(app)  # Integrate FastAPI app with BentoML
class TTSExample:
    def __init__(self) -> None:
        self.engine = self.setup_tts_engine()

    def setup_tts_engine(self):
        pass

    def synthesize(self, text: str) -> Generator[bytes, None, None]:
        pass

    @app.websocket("/ws")
    async def speech(self, websocket: WebSocket):
        await websocket.accept()
        try:
            while True:
                data = await websocket.receive_text()
                for chunk in self.engine.synthesize(data):
                    await websocket.send_bytes(chunk)
        except Exception as e:
            print(f"Error in WebSocket connection: {e}")
        finally:
            await websocket.close()

여기서는 FastAPI 앱을 @bentoml.asgi_app(app)으로 BentoML Service에 통합하고, 그 FastAPI 앱의 WebSocket 엔드포인트(/ws)로 커넥션을 받아요. 클라이언트가 텍스트를 보내면 TTS 엔진이 생성한 오디오 조각(bytes)을 send_bytes로 다시 흘려보내는 구조예요.

더 알아보기

BentoML에서 다양한 LLM 서빙하기: LLM 추론: vLLM 예제
오픈소스 모델로 음성 에이전트 만들기: Build a voice agent
WebSocket 엔드포인트 정의하기: Define a WebSocket endpoint