Server-Sent Events를 사용한 FastAPI 스트리밍
StreamingResponse와 text/event-stream 콘텐츠 유형을 사용해 LLM 스트리밍 응답을 브라우저 클라이언트로 전달하는 FastAPI 엔드포인트를 구축합니다.
Server-Sent Events를 사용한 FastAPI 스트리밍은(는) CoddyKit의 무료 AI Engineering Academy 강의입니다. 이것은 4개 중 3번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 AI 튜터와 함께 브라우저에서 직접 실습할 수 있습니다. 이 강의는 AI Engineering Academy 학습 경로의 일부이며, 진행 상황이 웹과 CoddyKit 앱에 동기화됩니다. AI Engineering Academy 강의에는 총 4개의 강의가 포함되어 있습니다.
LLM 스트리밍에 Server-Sent Events를 사용하는 이유
Server-Sent Events (SSE)는 서버가 하나의 장시간 HTTP 연결을 통해 텍스트 이벤트 스트림을 브라우저 클라이언트로 푸시할 수 있게 하는 W3C 표준입니다. WebSockets와 달리 SSE는 단방향(서버에서 클라이언트 방향)이며, 표준 HTTP/1.1을 통해 작동하고, 연결이 끊어지면 자동으로 다시 연결하며, 특수한 브라우저 라이브러리가 필요하지 않습니다. 이러한 특성 때문에 FastAPI 백엔드에서 웹 프런트엔드로 LLM 토큰을 스트리밍하기에 이상적인 전송 방식입니다.
SSE 전송 형식
SSE는 줄 바꿈으로 구분된 일련의 필드 형태로 텍스트 데이터를 전송합니다. 각 이벤트에는 선택적인 event 유형 필드, 페이로드가 담긴 data 필드, 재연결을 위한 선택적인 id가 포함됩니다. 이벤트는 빈 줄로 구분됩니다. LLM 스트리밍에서는 각 토큰을 data: token_text\n\n 줄로 보내고, 마지막에는 스트림 완료를 알리는 특수한 data: [DONE]\n\n 이벤트를 보내십시오.
# SSE wire format example
'''
data: The\n\n
data: capital\n\n
data: of\n\n
data: France\n\n
data: is\n\n
data: Paris\n\n
data: [DONE]\n\n
'''
# Each 'data:' line is one event.
# The double newline (\n\n) terminates each event.
# The client receives these as EventSource message events.
# The content-type must be 'text/event-stream'.FastAPI의 StreamingResponse
FastAPI의 StreamingResponse는 문자열을 생성하는 비동기 생성기를 받아 클라이언트로 스트리밍합니다. media_type을 'text/event-stream'으로 설정하고 생성되는 각 문자열을 SSE 이벤트 형식으로 지정하면, 모든 비동기 생성기를 올바른 SSE 스트림으로 변환할 수 있습니다. FastAPI가 연결 수명 주기, 플러시, HTTP 헤더를 자동으로 처리합니다.
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import AsyncOpenAI
import asyncio
app = FastAPI()
async_client = AsyncOpenAI()
async def token_generator(prompt: str):
stream = await async_client.chat.completions.create(
model='gpt-4o-mini',
messages=[{'role': 'user', 'content': prompt}],
stream=True,
)
async for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
yield f'data: {delta}\n\n' # SSE format
yield 'data: [DONE]\n\n'
@app.get('/stream')
async def stream_endpoint(prompt: str):
return StreamingResponse(
token_generator(prompt),
media_type='text/event-stream',
headers={'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no'},
)SSE에 중요한 HTTP 헤더
프록시와 CDN을 거쳐 SSE가 올바르게 작동하려면 세 가지 HTTP 헤더가 중요합니다. Cache-Control: no-cache는 중간 서버가 스트림을 캐시하지 못하게 합니다. Connection: keep-alive는 TCP 연결을 열린 상태로 유지합니다. X-Accel-Buffering: no는 Nginx의 응답 버퍼링을 비활성화합니다. 이 버퍼링은 청크를 한꺼번에 처리하여 스트리밍 효과를 없애기 때문입니다. 마지막 헤더가 없으면 Nginx는 모든 출력을 브라우저로 전달하기 전에 버퍼링합니다.
from fastapi.responses import StreamingResponse
SSE_HEADERS = {
'Cache-Control': 'no-cache',
'Connection': 'keep-alive',
'X-Accel-Buffering': 'no', # disable nginx buffering
'Access-Control-Allow-Origin': '*', # CORS for cross-origin clients
}
@app.get('/chat')
async def chat_stream(prompt: str):
return StreamingResponse(
token_generator(prompt),
media_type='text/event-stream',
headers=SSE_HEADERS,
)JSON 페이로드를 사용하는 구조화된 SSE 이벤트
더 풍부한 스트리밍 API를 위해 각 이벤트 페이로드를 일반 텍스트가 아닌 JSON으로 인코딩하십시오. 그러면 토큰과 함께 메타데이터를 포함할 수 있습니다. 예를 들어 토큰 유형(콘텐츠 또는 도구 호출), 메시지 ID, 지연 시간 타임스탬프 등을 넣을 수 있습니다. 브라우저 클라이언트는 각 이벤트의 JSON을 구문 분석하고 서로 다른 이벤트 유형을 서로 다른 UI 구성 요소로 전달합니다.
import json
import time
async def json_token_generator(prompt: str, session_id: str):
t_start = time.perf_counter()
stream = await async_client.chat.completions.create(
model='gpt-4o-mini',
messages=[{'role': 'user', 'content': prompt}],
stream=True,
)
async for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
payload = json.dumps({
'type': 'token',
'content': delta,
'session_id': session_id,
't_ms': round((time.perf_counter() - t_start) * 1000),
})
yield f'data: {payload}\n\n'
# Send completion event
yield f'data: {json.dumps({"type": "done", "session_id": session_id})}\n\n'브라우저에서 SSE 사용하기(JavaScript)
브라우저 측 EventSource API는 SSE 엔드포인트에 연결하고 이벤트가 도착할 때마다 이벤트를 발생시킵니다. 토큰 스트리밍에서는 기본 message 이벤트를 수신하고, 데이터를 JSON으로 구문 분석하거나 일반 문자열로 처리한 다음 각 토큰을 DOM에 추가하십시오. [DONE] 센티널을 받으면 EventSource 연결을 닫으십시오.
// Browser-side JavaScript
const prompt = 'Explain hybrid search in one paragraph.';
const url = '/stream?prompt=' + encodeURIComponent(prompt);
const source = new EventSource(url);
const output = document.getElementById('output');
source.onmessage = (event) => {
if (event.data === '[DONE]') {
source.close(); // stop listening
return;
}
output.textContent += event.data; // append each token
};
source.onerror = (err) => {
console.error('SSE error:', err);
source.close();
};스트리밍을 위한 fetch POST 요청
EventSource는 GET 요청만 지원하므로 복잡한 프롬프트에는 제약이 있습니다. POST 요청(대화 기록이 포함된 JSON 본문 전송)에는 브라우저의 fetch API와 Streams API를 사용하여 응답 본문을 조금씩 읽으십시오. 이 패턴은 ChatGPT 웹 인터페이스와 대부분의 프로덕션 LLM 채팅 UI에서 사용됩니다.
// Browser-side: POST with fetch and ReadableStream
async function streamPost(messages) {
const response = await fetch('/chat', {
method: 'POST',
headers: {'Content-Type': 'application/json'},
body: JSON.stringify({messages}),
});
const reader = response.body.getReader();
const decoder = new TextDecoder();
const output = document.getElementById('output');
while (true) {
const {done, value} = await reader.read();
if (done) break;
const text = decoder.decode(value, {stream: true});
// Parse SSE lines
for (const line of text.split('\n')) {
if (line.startsWith('data: ') && line !== 'data: [DONE]') {
output.textContent += line.slice(6);
}
}
}
}채팅 스트리밍을 위한 FastAPI POST 엔드포인트
POST 기반 채팅 스트리밍을 구현하려면 요청 본문을 위한 Pydantic 모델을 정의하고, 메시지 목록을 받은 뒤 LLM 응답을 스트리밍하십시오. 이렇게 하면 각 요청에 전체 대화 기록을 전달할 수 있어 여러 차례 대화하는 채팅 애플리케이션을 지원할 수 있습니다. 요청 본문에서 프롬프트를 추출한다는 점을 제외하면 GET 스트리밍과 패턴이 동일합니다.
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
class ChatRequest(BaseModel):
messages: list[dict]
model: str = 'gpt-4o-mini'
@app.post('/chat')
async def chat_post(request: ChatRequest):
async def generate():
stream = await async_client.chat.completions.create(
model=request.model,
messages=request.messages,
stream=True,
)
async for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
yield f'data: {delta}\n\n'
yield 'data: [DONE]\n\n'
return StreamingResponse(
generate(),
media_type='text/event-stream',
headers=SSE_HEADERS,
)클라이언트 연결 끊김 처리
브라우저 사용자가 다른 페이지로 이동하거나 탭을 닫으면 HTTP 연결이 닫히고 FastAPI는 스트리밍 생성기에서 asyncio.CancelledError를 발생시킵니다. 항상 이를 처리해야 LLM 스트리밍 요청이 열린 채로 남아 불필요한 API 비용이 발생하는 것을 막을 수 있습니다. 생성기를 CancelledError에 대한 try/except로 감싸고, 이를 감지하면 OpenAI 스트림을 취소하십시오.
from fastapi import Request
@app.get('/stream')
async def stream_with_disconnect(prompt: str, request: Request):
async def generate_with_cancel():
try:
stream = await async_client.chat.completions.create(
model='gpt-4o-mini',
messages=[{'role': 'user', 'content': prompt}],
stream=True,
)
async for chunk in stream:
if await request.is_disconnected():
break # client gone, stop generating
delta = chunk.choices[0].delta.content
if delta:
yield f'data: {delta}\n\n'
except asyncio.CancelledError:
pass # client disconnected
finally:
yield 'data: [DONE]\n\n'
return StreamingResponse(generate_with_cancel(), media_type='text/event-stream')요청 인증 추가
프로덕션 스트리밍 엔드포인트는 승인되지 않은 LLM 사용을 막기 위해 요청을 인증해야 합니다. API 키 또는 JWT 헤더 확인과 함께 FastAPI의 Depends를 사용하십시오. 인증은 생성기가 시작되기 전에 수행되므로 오버헤드가 작고, 사용자가 확인된 후에만 스트림이 시작됩니다.
from fastapi import Header, HTTPException, Depends
VALID_API_KEYS = {'sk-demo-key-1', 'sk-demo-key-2'}
async def verify_api_key(x_api_key: str = Header(None)):
if x_api_key not in VALID_API_KEYS:
raise HTTPException(status_code=401, detail='Invalid API key')
return x_api_key
@app.post('/chat')
async def authenticated_chat(
request: ChatRequest,
api_key: str = Depends(verify_api_key),
):
async def generate():
stream = await async_client.chat.completions.create(
model=request.model,
messages=request.messages,
stream=True,
)
async for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
yield f'data: {delta}\n\n'
yield 'data: [DONE]\n\n'
return StreamingResponse(generate(), media_type='text/event-stream', headers=SSE_HEADERS)SSE 엔드포인트 테스트하기
FastAPI의 TestClient를 스트리밍 모드로 사용하여 스트리밍 엔드포인트를 테스트하십시오. with client.stream('GET', '/stream', params={...}) as r를 사용하고 r.iter_lines()를 순회하여 SSE 이벤트를 받으십시오. 이를 통해 토큰 형식이 올바른지, DONE 센티널이 전송되는지, 오류 상황에서 적절한 SSE 오류 이벤트가 생성되는지 확인할 수 있습니다.
from fastapi.testclient import TestClient
def test_sse_endpoint():
with TestClient(app) as client:
with client.stream('GET', '/stream', params={'prompt': 'Say hi'}) as r:
assert r.status_code == 200
assert 'text/event-stream' in r.headers['content-type']
events = []
for line in r.iter_lines():
if line.startswith('data: '):
events.append(line[6:])
assert events[-1] == '[DONE]'
full_text = ''.join(e for e in events if e != '[DONE]')
assert len(full_text) > 0빠른 확인
이 레슨에서 배운 FastAPI의 SSE 스트리밍에 대한 이해도를 확인해 보십시오.
레슨 요약
이 레슨에서는 다음을 배웠습니다. Server-Sent Events는 LLM 토큰을 브라우저 클라이언트로 스트리밍하기 위한 표준 HTTP 전송 방식이고, text/event-stream을 사용하는 StreamingResponse는 FastAPI에서 모든 비동기 생성기를 SSE 스트림으로 변환하며, 프록시 뒤에서 올바르게 작동하려면 X-Accel-Buffering과 Cache-Control을 비롯한 중요한 헤더가 필요합니다. 고아 상태로 남는 LLM API 호출을 방지하려면 클라이언트 연결 끊김을 처리하십시오. 다음으로 도구 호출이 포함된 스트리밍 응답을 다루겠습니다.
AI 튜터와 함께 Python을(를) 배우세요 — 무료
브라우저에서 실제 코드를 작성하고 실행하며, 24/7 AI 튜터로부터 즉각적인 도움을 받고, 웹이나 앱에서 중단한 부분부터 계속 학습하세요.
- 코스
- 30
- 레슨
- 120
자주 묻는 질문
“Server-Sent Events를 사용한 FastAPI 스트리밍” 강의는 무료인가요?
네 — “Server-Sent Events를 사용한 FastAPI 스트리밍” 전체 내용을 이 웹사이트에서 무료로 읽을 수 있습니다. 인터랙티브하게 실습하려면(내장 코드 에디터와 24/7 AI 튜터), CoddyKit PRO로 업그레이드하면 AI Engineering Academy 강의 전체를 잠금 해제할 수 있습니다. AI Engineering Academy 강의에는 총 4개의 강의가 포함되어 있습니다.
“Server-Sent Events를 사용한 FastAPI 스트리밍”에서 뭘 배우나요?
StreamingResponse와 text/event-stream 콘텐츠 유형을 사용해 LLM 스트리밍 응답을 브라우저 클라이언트로 전달하는 FastAPI 엔드포인트를 구축합니다. 브라우저에서 직접 실행하는 실습 코드로 AI Engineering Academy을(를) 배우며, 24/7 AI 튜터가 강의를 진행하면서 질문에 답변해줍니다.
AI Engineering Academy을(를) 시작하는 데 경험이 필요한가요?
사전 경험은 필요하지 않습니다. CoddyKit의 AI Engineering Academy은(는) 초급자부터 고급 학습자까지를 위해 구성되어 있으므로, 여기서 시작하거나 처음부터 시작할 수 있으며 자신의 속도대로 진행할 수 있습니다. 이것은 4개 중 3번째 강의입니다.
“Server-Sent Events를 사용한 FastAPI 스트리밍” 강의는 얼마나 걸리나요?
대부분의 CoddyKit 강의는 약 5~10분이 소요됩니다. 각 강의는 간결하고 인터랙티브하여 꾸준한 진행이 가능하며, 웹과 앱에서 중단한 부분부터 바로 시작할 수 있습니다.
이 AI Engineering Academy 강의에서 코드를 작성하고 실행할 수 있나요?
네. 모든 AI Engineering Academy 강의에는 내장 코드 에디터가 포함되어 있으므로, 브라우저에서 바로 실제 코드를 작성하고 실행한 후 즉시 AI 피드백을 받을 수 있습니다 — 로컬 설정이 필요 없습니다.
이 강의의 모든 강의
- 토큰 스트리밍 이해
- Python SDK로 스트림 소비
- Server-Sent Events를 사용한 FastAPI 스트리밍
- 스트리밍 응답에서 도구 호출 처리