Python으로 공부하는 OpenAI 6편 — FastAPI에서 OpenAI 스트리밍 응답 만들기

Python으로 공부하는 OpenAI 6편 — FastAPI에서 OpenAI 스트리밍 응답 만들기

여기쯤 오면 이제 욕심이 생깁니다.
“응답을 다 만든 뒤 한 번에 보내는 건 알겠는데… ChatGPT처럼 타이핑되듯이 보내려면 어떻게 하지?”

바로 그게 이번 글 주제입니다.

이전 편까지는 FastAPI에서 OpenAI를 호출하고, Structured Outputs로 JSON을 받고, Pydantic으로 검증하고, 서비스 계층까지 분리했습니다.
그런데 사용자 경험 관점에서는 아직 조금 답답합니다.

왜냐하면 긴 답변은 완성될 때까지 아무것도 안 보이기 때문이죠.

OpenAI 공식 스트리밍 가이드는 기본 요청은 모델의 전체 출력을 다 만든 뒤 한 번에 HTTP 응답으로 보내지만, stream=True를 사용하면 생성되는 도중부터 결과를 조금씩 처리할 수 있다고 설명합니다. 이 가이드는 HTTP 스트리밍 방식으로 SSE(Server-Sent Events) 를 중심으로 다룹니다. (OpenAI Developers)

이번 글에서는 이 흐름을 FastAPI에 붙여보겠습니다.

• 왜 스트리밍이 필요한지
• 일반 응답과 스트리밍 응답 차이
• FastAPI에서 StreamingResponse로 붙이는 방법
• OpenAI Responses API 스트리밍을 어떻게 읽는지
• 프론트엔드에서 어떻게 받는지
• 연결 종료와 예외를 어떻게 다뤄야 하는지

---

1. 왜 스트리밍이 필요한가

처음엔 “어차피 답은 같잖아?” 싶을 수 있습니다.
근데 실제 서비스에서는 차이가 꽤 큽니다.

일반 응답은 이런 느낌입니다.

1. 사용자가 질문한다
2. 서버가 OpenAI에 요청한다
3. 몇 초 동안 화면이 멈춘 것처럼 보인다
4. 결과가 한 번에 툭 나온다

반면 스트리밍은 이렇습니다.

1. 사용자가 질문한다
2. 서버가 OpenAI에 요청한다
3. 첫 토큰부터 조금씩 흘러나온다
4. 사용자는 “아, 지금 생성 중이구나”를 바로 느낀다

OpenAI 공식 가이드는 긴 출력일수록 전체 응답을 기다리는 시간이 체감될 수 있고, 스트리밍을 쓰면 출력의 앞부분을 출력 중간부터 표시하거나 처리할 수 있다고 설명합니다. (OpenAI Developers)

이건 단순히 멋있어 보이는 기능이 아닙니다.
특히 아래 같은 경우 체감이 큽니다.

• 긴 요약문 생성
• 코드 설명
• 블로그 초안 생성
• 고객 응대 답변 생성
• 문서 기반 질의응답

---

2. SSE를 먼저 이해해야 합니다

이번 글에서 핵심 transport는 WebSocket이 아니라 SSE입니다.

OpenAI의 스트리밍 가이드는 stream=True를 통해 server-sent events 로 응답을 흘려보내는 방식을 설명합니다. FastAPI 쪽에서는 이 스트림을 클라이언트에 전달할 때 보통 StreamingResponse를 사용합니다. Starlette 문서도 StreamingResponse가 async generator나 일반 generator를 받아 응답 본문을 스트리밍할 수 있다고 설명합니다. (OpenAI Developers)

쉽게 말하면 SSE는 이런 느낌입니다.

• 서버 → 클라이언트 방향으로 텍스트 이벤트를 계속 흘려보냄
• 브라우저는 그걸 실시간으로 받음
• 채팅 답변, 로그 tail, 진행 상태 표시 같은 데 잘 맞음

즉, 이번 글은
OpenAI 스트리밍 → FastAPI → 브라우저
이 연결을 만드는 과정입니다.

---

3. OpenAI 쪽 스트리밍은 어떻게 켜나

이 부분은 생각보다 간단합니다.

OpenAI 공식 스트리밍 가이드에 따르면 Responses API에서 스트리밍을 시작하려면 요청에 stream=True를 주면 됩니다. 그러면 서버가 응답 생성 중에 여러 스트리밍 이벤트를 보냅니다. (OpenAI Developers)

즉, 기본 호출:

response = client.responses.create(...)

스트리밍 호출:

stream = client.responses.create(..., stream=True)

정도로 이해하면 됩니다.

그리고 OpenAI Python SDK README는 이 라이브러리가 동기/비동기 클라이언트를 모두 제공한다고 설명합니다. FastAPI에서는 async 흐름이 자연스럽기 때문에 이번 글에서는 비동기 클라이언트 기준으로 예제를 가져가겠습니다. (GitHub)

---

4. 이번 글에서 만들 구조

5편 구조를 거의 유지하면서, 스트리밍용 서비스와 라우터를 추가하겠습니다.

app/
├── main.py
├── core/
│   └── config.py
├── schemas/
│   └── chat.py
├── services/
│   └── openai_chat_service.py
└── api/
    └── chat_router.py

이번에는 JSON 구조화 출력보다,
텍스트를 순차적으로 흘려보내는 채팅형 응답에 집중합니다.

---

5. 요청 스키마부터 간단하게

app/schemas/chat.py

from pydantic import BaseModel, Field


class ChatRequest(BaseModel):
    message: str = Field(min_length=1, max_length=2000)

이번 편에서는 응답 모델을 따로 두지 않겠습니다.
왜냐하면 스트리밍 응답은 JSON 객체 하나를 반환하는 게 아니라, 조각난 텍스트 이벤트들을 반환하기 때문입니다.

---

6. 설정은 지난 편처럼 재사용합니다

반응형

app/core/config.py

from functools import lru_cache

from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    app_name: str = "OpenAI FastAPI Streaming Example"
    openai_api_key: str
    openai_model: str = "gpt-5.4-mini"

    model_config = SettingsConfigDict(env_file=".env", extra="ignore")


@lru_cache
def get_settings() -> Settings:
    return Settings()

OpenAI는 API 키를 서버에서 환경변수나 키 관리 시스템으로 안전하게 불러오라고 안내합니다. API 키를 클라이언트 코드에 노출하면 안 됩니다. (OpenAI Developers)

---

7. 핵심: OpenAI 스트림을 FastAPI에 넘기는 서비스 만들기

여기부터가 진짜 중요합니다.

OpenAI 공식 스트리밍 가이드는 Python 예제로 스트림 객체를 순회하면서 이벤트를 처리하는 흐름을 보여줍니다. Responses 스트리밍 이벤트 문서도 stream=true일 때 여러 종류의 이벤트가 순서대로 발생한다고 설명합니다. (OpenAI Developers)

우리는 그중에서 텍스트 조각(delta) 만 뽑아 브라우저로 넘길 겁니다.

app/services/openai_chat_service.py

from openai import AsyncOpenAI

from app.core.config import Settings


class OpenAIChatService:
    def __init__(self, settings: Settings):
        self.settings = settings
        self.client = AsyncOpenAI(api_key=settings.openai_api_key)

    async def stream_text(self, user_message: str):
        stream = await self.client.responses.create(
            model=self.settings.openai_model,
            input=user_message,
            instructions=(
                "당신은 주니어 개발자를 돕는 친절한 Python 멘토입니다. "
                "한국어로 쉽게 설명하세요."
            ),
            stream=True,
        )

        async for event in stream:
            event_type = getattr(event, "type", "")

            if event_type == "response.output_text.delta":
                delta = getattr(event, "delta", "")
                if delta:
                    yield delta

            elif event_type == "response.completed":
                break

여기서 핵심은 두 가지입니다

첫째, AsyncOpenAI를 사용했다는 점
OpenAI Python SDK는 async client를 제공합니다. FastAPI와 붙일 때 이 조합이 자연스럽습니다. (GitHub)

둘째, 이벤트 타입을 보고 필요한 것만 흘려보낸다는 점
Responses 스트리밍 이벤트 문서는 response.output_text.delta 같은 이벤트와 완료 이벤트가 있음을 설명합니다. 즉, 모든 이벤트를 그대로 사용자에게 던지는 게 아니라, 내가 필요한 이벤트만 골라야 합니다. (OpenAI Developers)

---

8. FastAPI 라우터에서 SSE 형태로 감싸기

이제 OpenAI에서 받은 delta를 브라우저가 읽기 쉬운 SSE 포맷으로 감싸야 합니다.

SSE는 기본적으로 아래 형식을 많이 씁니다.

data: 첫 번째 조각

data: 두 번째 조각

data: 세 번째 조각

그리고 FastAPI/Starlette에서는 StreamingResponse로 이런 텍스트 스트림을 보낼 수 있습니다. Starlette 문서는 StreamingResponse가 generator를 받아 chunked 응답을 보낸다고 설명합니다. (OpenAI Developers)

app/api/chat_router.py

from typing import Annotated

from fastapi import APIRouter, Depends
from fastapi.responses import StreamingResponse

from app.core.config import Settings, get_settings
from app.schemas.chat import ChatRequest
from app.services.openai_chat_service import OpenAIChatService

router = APIRouter(prefix="/chat", tags=["chat"])


def get_chat_service(
    settings: Annotated[Settings, Depends(get_settings)],
) -> OpenAIChatService:
    return OpenAIChatService(settings)


@router.post("/stream")
async def stream_chat(
    request: ChatRequest,
    service: Annotated[OpenAIChatService, Depends(get_chat_service)],
):
    async def event_generator():
        try:
            async for chunk in service.stream_text(request.message):
                yield f"data: {chunk}\n\n"
            yield "event: done\ndata: [DONE]\n\n"
        except Exception as e:
            yield f"event: error\ndata: {str(e)}\n\n"

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
        },
    )

왜 media_type="text/event-stream" 인가

SSE 응답은 브라우저와 프론트엔드가 “이건 이벤트 스트림이구나”라고 이해해야 합니다.
OpenAI 자체 스트리밍도 SSE 기반으로 설명되고 있고, 우리 서버도 브라우저 쪽에는 같은 계열의 스트림 형식으로 넘겨주는 게 자연스럽습니다. (OpenAI Developers)

---

9. 앱 등록

app/main.py

from fastapi import FastAPI

from app.api.chat_router import router as chat_router
from app.core.config import get_settings

settings = get_settings()

app = FastAPI(title=settings.app_name)

app.include_router(chat_router)

실행:

pip install fastapi uvicorn openai pydantic pydantic-settings python-dotenv
uvicorn app.main:app --reload

OpenAI 공식 라이브러리 설치는 pip install openai이고, SDK는 Python 3.9+ 애플리케이션을 지원합니다. (OpenAI Developers)

---

10. 프론트엔드에서는 어떻게 받나

여기서 한 가지 현실적인 포인트가 있습니다.

브라우저의 EventSource는 기본적으로 GET 요청에 잘 맞습니다.
지금 우리는 POST로 메시지를 보내고 있으니, 실제 프론트에서는 fetch()로 응답 body를 스트림처럼 읽는 방식이 더 유연합니다.

아래는 아주 단순한 예제입니다.

<!DOCTYPE html>
<html lang="ko">
<head>
  <meta charset="UTF-8" />
  <title>Streaming Chat Example</title>
</head>
<body>
  <input id="message" type="text" value="파이썬 데코레이터를 쉽게 설명해줘" />
  <button id="send">전송</button>
  <pre id="output"></pre>

  <script>
    const button = document.getElementById("send");
    const output = document.getElementById("output");
    const messageInput = document.getElementById("message");

    button.addEventListener("click", async () => {
      output.textContent = "";

      const response = await fetch("/chat/stream", {
        method: "POST",
        headers: {
          "Content-Type": "application/json"
        },
        body: JSON.stringify({
          message: messageInput.value
        })
      });

      const reader = response.body.getReader();
      const decoder = new TextDecoder("utf-8");

      while (true) {
        const { value, done } = await reader.read();
        if (done) break;

        const chunk = decoder.decode(value, { stream: true });
        const lines = chunk.split("\n");

        for (const line of lines) {
          if (line.startsWith("data: ")) {
            output.textContent += line.slice(6);
          }
        }
      }
    });
  </script>
</body>
</html>

이 코드는 아주 단순화한 버전입니다.
실전에서는 chunk 경계가 SSE 이벤트 경계와 꼭 일치하지 않을 수 있어서 버퍼링 처리를 더 신경 써야 합니다. 하지만 입문 단계에서는 “FastAPI가 SSE 스타일 텍스트를 흘려주고, 프론트가 body stream을 읽는다”는 흐름을 먼저 이해하는 게 중요합니다.

---

11. 연결 종료와 예외 처리는 왜 중요할까

스트리밍은 일반 JSON 응답보다 다루기 까다롭습니다.
왜냐하면 응답이 “한 번에 끝나는 객체”가 아니라 “흐르는 연결”이기 때문입니다.

실무에서 꼭 신경 써야 하는 건 아래입니다.

11-1. 완료 이벤트를 명확히 보내기

위 예제의 [DONE] 같은 마커는 프론트가 “이제 끝났구나”를 알 수 있게 해줍니다.
OpenAI 스트리밍도 완료 이벤트를 따로 보냅니다. (OpenAI Developers)

11-2. 에러를 일반 텍스트와 구분하기

event: error 같이 이벤트 이름을 따로 주면 프론트가 예외 상황을 분기 처리하기 쉽습니다.

11-3. 중간 취소를 고려하기

사용자가 화면을 나가거나 새 요청을 보내면 연결이 끊길 수 있습니다.
스트리밍은 이 상황을 전제로 설계해야 합니다.

11-4. 너무 긴 작업은 별도 설계를 고민하기

OpenAI 공식 문서는 HTTP 스트리밍과 별도로, 더 지속적인 상호작용이 필요하면 Realtime API나 WebSocket 모드를 고려하라고 안내합니다. 즉, 단방향 텍스트 출력은 SSE가 잘 맞지만, 양방향 저지연 상호작용은 다른 선택지가 더 적합할 수 있습니다. (OpenAI Developers)

---

12. 스트리밍과 일반 응답은 언제 나눠 써야 하나

이건 진짜 실무적인 질문입니다.

모든 응답을 스트리밍으로 만들 필요는 없습니다.

일반 응답이 더 좋은 경우

• 짧은 분류 결과
• JSON 구조화 출력
• DB 저장용 응답
• 후처리 완료 후 한 번에 내려야 하는 결과

스트리밍이 더 좋은 경우

• 긴 설명문
• 블로그 초안
• 코드 생성/해설
• 사용자 대기 시간이 체감되는 답변
• 채팅 UI

OpenAI Structured Outputs는 JSON Schema를 정확히 지키는 응답을 만드는 데 강점이 있고, 스트리밍은 텍스트를 점진적으로 보여주는 데 강점이 있습니다. 즉, 구조화 응답과 스트리밍은 목적이 다릅니다. (OpenAI Developers)

저는 보통 이렇게 나눕니다.

• 내부 처리용 데이터 → JSON / Structured Outputs
• 사용자에게 읽히는 긴 텍스트 → Streaming

---

13. 주니어가 여기서 자주 하는 실수

실수 1. 스트리밍인데도 최종 문자열만 만들고 마지막에 한 번에 보낸다

이러면 이름만 스트리밍이고 실제로는 일반 응답과 다를 게 없습니다.

실수 2. OpenAI 이벤트를 전부 그대로 프론트에 노출한다

스트림 이벤트는 내부용 메타 이벤트도 섞일 수 있습니다.
필요한 delta만 골라 내보내는 게 깔끔합니다. (OpenAI Developers)

실수 3. text/event-stream를 빼먹는다

프론트가 스트림으로 처리하지 못할 수 있습니다.

실수 4. 에러를 스트림 밖 예외로만 처리한다

스트리밍 중간 에러는 이미 응답이 시작된 뒤라, 일반 JSON 에러 응답처럼 처리하기 어렵습니다. 그래서 스트림 내부에서 에러 이벤트를 흘려주는 설계가 유용합니다.

실수 5. Structured Outputs와 스트리밍을 같은 문제로 본다

둘은 비슷해 보여도 목적이 다릅니다.
정확한 데이터 구조가 필요하면 Structured Outputs,
사용자 체감 반응성이 중요하면 Streaming이 우선입니다. (OpenAI Developers)

---

14. 지금 단계에서 바로 써먹을 수 있는 개선 포인트

이번 예제를 그대로 운영에 넣기보다, 아래를 붙이면 훨씬 좋아집니다.

요청별 request id 붙이기

로그 추적이 쉬워집니다.

delta를 누적해서 최종 로그도 저장하기

실시간 표시와 저장은 분리하는 게 좋습니다.

프롬프트/모델명을 로그에 남기기

나중에 품질 이슈 추적이 쉬워집니다.

클라이언트 disconnect 감지 추가

중간 연결 종료 시 불필요한 작업을 줄일 수 있습니다.

프론트 버퍼링 개선

chunk가 끊기는 경계를 정확히 처리하도록 버퍼를 두는 편이 좋습니다.

---

15. 오늘 글의 핵심 요약

이번 글에서 꼭 가져가야 할 건 이것입니다.

FastAPI에서 OpenAI 스트리밍을 만들려면, OpenAI의 stream=True 응답을 서비스 계층에서 읽고, 필요한 delta만 골라 StreamingResponse로 SSE 형태로 흘려보내면 된다.

핵심 흐름은 이렇게 정리할 수 있습니다.

1. OpenAI Responses API 호출 시 stream=True를 준다. (OpenAI Developers)
2. 스트림 이벤트 중 response.output_text.delta 같은 텍스트 조각만 골라낸다. (OpenAI Developers)
3. FastAPI StreamingResponse로 text/event-stream 응답을 만든다. (OpenAI Developers)
4. 프론트는 fetch()의 body stream이나 SSE 방식으로 이를 읽는다.
5. 완료 이벤트와 에러 이벤트를 따로 설계한다.

이 감각이 잡히면, 이제 진짜 “채팅형 서비스” 느낌이 나기 시작합니다.

---

다음 편 예고

다음 글에서는 이 스트리밍 구조를 한 단계 더 실무적으로 가져가겠습니다.

FastAPI에서 대화 히스토리와 함께 OpenAI 채팅 API 만들기

이 주제로,

• 메시지 배열 설계
• 이전 대화 이어가기
• 세션별 대화 저장 방식
• previous_response_id와 대화 상태 감각
• 채팅 서비스 구조화

까지 이어가보겠습니다.

---

출처

• OpenAI Streaming API responses 가이드 — stream=True, SSE 기반 스트리밍, 생성 중간부터 출력 가능. (OpenAI Developers)
• OpenAI Responses streaming events 레퍼런스 — 스트리밍 시 여러 이벤트가 발생하며 response.output_text.delta, 완료 이벤트 등을 설명. (OpenAI Developers)
• OpenAI API Overview — API 키는 서버에서 안전하게 관리해야 하며 Bearer 인증 사용. (OpenAI Developers)
• OpenAI Libraries / Python SDK README — Python SDK 설치, Python 3.9+ 지원, sync/async client 제공. (OpenAI Developers)
• OpenAI Structured Outputs 가이드 — JSON Schema를 따르는 구조화 응답 용도. 스트리밍과는 목적이 다름. (OpenAI Developers)
• OpenAI Realtime API 가이드 — 더 지속적이고 저지연인 양방향 상호작용에는 Realtime/WebSocket 계열 고려 가능. (OpenAI Developers)

 

Python, OpenAI, FastAPI, StreamingResponse, SSE, Server Sent Events, OpenAI Streaming, Responses API, AsyncOpenAI, AI 백엔드, Python 백엔드, 주니어 개발자, 채팅 서버, 실시간 응답, FastAPI 스트리밍