오늘은 llm 채팅 구현에 사용되는 sse에 대해 간단히 정리해 보고 코드도 살펴보겠습니다.
최근 AI agent 를 공부/구현하고 있는데, api를 처리하는 방식을 알아보던 중 sse를 알게 되었고 대부분 sse로 llm 채팅을 구현하고 있다는 것을 알게 되었습니다. 또한 OpenAI 도 SSE로 구현하고 있습니다. 잘 나가는 친구들 따라가면 반이라도 가겠죠
아무튼 시작해 보겠습니다. 제가 CS전공이 아니어서 정확하지 않을 수 있습니다. 제가 이해한 대로 글을 작성합니다. 틀린 부분은 마음껏 지적해 주세요.
그래서 SSE 그게 뭔데?
SSE... Server-Sent Event
서버 전송 이벤트로 클라이언트가 HTTP 연결을 통해 서버로부터 자동 업데이트를 수신할 수 있도록 하는 서버 푸시 기술입니다.
이런 특징을 가지고 있습니다.
- text/event-stream이라는 MIME 타입으로 이벤트를 줄줄이 전송
- 서버 → 클라이언트 단방향
- HTTP 위에서 동작하는 스트리밍 방식
- 요청 1번을 보내면, 서버가 응답을 끝내지 않고 계속 열린 상태로 데이터를 보냄
쉽게 말해서 한번 연결해 두고 서버가 이벤트가 발생할 때마다 보내준다는 것이다.
Why SSE?
그래서 llm 채팅 구현하는데 왜 SSE로 하는데? 하면 바로 다음과 같은 이유입니다.
1. 사용자 체감 latency를 줄일 수 있다.
맞습니다. chatGPT, Claude 많이 쓰시죠? 내가 질문을 하면 얘들이 채팅 치는 것 마냥 글자가 쭈루루르륵 생깁니다. 첫 토큰이 나오자마자 이벤트로 토큰마다 클라이언트로 전송해 주니까 그걸 그대로 받아서 화면에 띄울 수 있습니다.
모델 출력이 다 끝나고 출력하게 되면 저희는 질문 하나 던지고 멀뚱멀뚱 빈 화면만 봐야 하는 거죠. 토큰마다 쏴주면서 사용자 경험을 높여줄 수 있습니다.
2. 양방향 통신이 필요 없다.
그럼 WebSocket 쓰면 되는 거 아니야? 라고 생각할 수 있습니다. 위에서 말한 stream 이 가능한 방식이니까요.
하지만 둘 중에 SSE를 선택하게 된 이유는 llm 채팅의 특성에 있습니다.
사용자가 입력/질문을 던지게 되면 서버에서 llm 출력이 끝나기 전까지 저희는 또 다른 입력을 보내지 않습니다. 단방향 스트리밍 까지가 우리가 필요한 바운더리라는 것입니다.
게임/협업이 필요한 그런 상황에서는 스트리밍에 양방향 통신까지 가능한 웹소켓을 쓰면 되겠지만 지금 상황에서는 헤비하고 구조적으로 SSE가 더 단순하고 효율적입니다!
3. HTTP 위에서 동작한다.
또한 SSE는 별도 프로토콜이 아닌 HTTP 그대로 사용합니다. 그래서 기존 인증 체계, 호환, 서버 구현 모두 쉽다는 거죠. 즉 기존 웹 인프라에 이질감 없이 그대로 얹을 수 있다는 점입니다.
이렇게 보니 SSE를 왜 사용하는지 어느 정도 이해가 됩니다.
Fastapi로 SSE 구현해 보기
간단한 웹 인터페이스를 (claude code에게 시켜서) 만들어서 fastapi로 구현한 SSE가 잘 작동하는지 확인해 보는 것 까지가 오늘 목표입니다.
아래와 같은 준비물이 간단하게 필요합니다.
준비물
- 노트북
- vscode (이건 아무거나 편한 거)
- fastapi 설치 필요
- openrouter
claude code
이 정도인데요, openrouter는 다양한 모델들의 api를 이용할 수 있는 플랫폼입니다.
OpenRouter
The unified interface for LLMs. Find the best models & prices for your prompts
openrouter.ai
위 페이지에 접속하셔서 회원가입/로그인하시면 사용이 가능합니다.
아니 돈 내는 거 아니냐고요? 맞습니다. 저는 결제를 했습니다. 하지만 openrouter에서는 새로운 모델이나 몇 가지 모델들에 대해서 무료로 api를 제공합니다. 선물 모양이 있는 놈을 찾으면 됩니다.
바로 이런 놈들입니다. 오늘은 업스테이지 solar pro 3 로 한 번 해보겠습니다. 후원 감사합니다.
**주의사항** : 무료 모델은 프롬프트가 학습에 사용되니 비밀 얘기를 물어보는 것은 자제해 주세요
아무튼 fastapi 설치는 다들 하셨을 것 같아서 시작하겠습니다. 물론 안 하셨으면 지금 하고 오시면 됩니다. 모르시는 분들은 인터넷에 검색..... 혹은 chatgpt, gemini, claude의 도움을 받으시면 금방 하실 것 같습니다.
아무튼 이런 프로젝트가 생겼는데요,
# main.py
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def read_root():
return {"Hello": "World"}
시작하기에 앞서 저는 미리 openrouter api키를 넣어주겠습니다.
main.py 상단 부분에 넣어주면 되겠죠.
# main.py
from dotenv import load_dotenv
load_dotenv()
OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY")
OPENROUTER_URL = "https://openrouter.ai/api/v1/chat/completions"
api key 같은 경우는 .env에 잘 저장을 해두셔야 합니다. 아래와 같이 말이죠
# .env
OPENROUTER_API_KEY=sk-or-v1-hereisyourapikey
그렇다면 클라이언트가 서버에게 보내는 JSON body 구조부터 정의를 해보겠습니다.
즉, 사용자 우리가 모델에게 보내고 싶은 정보를 정의하는 것이죠. 저희가 실제 서비스를 개발하고 있다면 다양한 정보를 뭐 많이 넣어야겠지만? 지금은 단순히 sse 구현해 보는 것이 목적이니까 prompt 하나만 보내는 것으로 하겠습니다.
class ChatRequest(BaseModel):
prompt: str
쉽죠?
그런 다음 openrouter에서 우리에게 보내줄 놈들을 SSE 형식으로 변경해 주는 파이프라인을 추가해줘야 합니다.
# main.py
async def stream_openrouter(prompt: str):
headers = {
"Authorization": f"Bearer {OPENROUTER_API_KEY}", # api 인증
"Content-Type": "application/json",
}
payload = {
"model": "upstage/solar-pro-3:free",
"stream": True, # stream을 true로 설정해야 토큰 단위로 쪼개서 보내줌
"messages": [{"role": "user", "content": prompt}],
}header 에는 인증을 위한 openrouter api key가 들어갑니다. 그리고 model, stream, message 가 payload에 들어갑니다. 여기서 중요한 것은stream을 true로 설정해야 openrouter에서 토큰 단위로 쪼개서 보내줍니다.
다음은 이제 openrouter에서 보내주는 이벤트를 파싱을 해야 합니다. openrouter가 보내주는 SSE는 아래와 같은 형태를 지닙니다.
data: {"choices":[{"delta":{"content":"안"}}]}
data: {"choices":[{"delta":{"content":"녕"}}]}
data: [DONE]
줄마다 파싱 해주고 [Done] 이 나오면 스트림을 종료하고 그런 부분들을 구현해야 합니다.
구현하면 요런 식으로 됩니다.
# 스트리밍 요청
async with httpx.AsyncClient() as client:
# 청크 단위로 비동기 수신
async with client.stream(
"POST", OPENROUTER_URL, json=payload, headers=headers, timeout=60.0
) as response:
# 줄 단위로 파싱해주는 것
async for line in response.aiter_lines():
if not line.startswith("data: "):
continue
data = line[len("data: "):]
# done 나오면 스트림 종료
if data.strip() == "[DONE]":
yield "data: [DONE]\n\n"
return
# json 파싱
try:
chunk = json.loads(data)
content = chunk["choices"][0]["delta"].get("content", "")
if content:
yield f"data: {content}\n\n"
except (json.JSONDecodeError, KeyError, IndexError):
continue
그러면 엔드포인트가 필요하겠죠?
@app.post("/chat")
async def chat(req: ChatRequest):
return StreamingResponse(
stream_openrouter(req.prompt),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"X-Accel-Buffering": "no",
},
)req로는 앞서 정의해 준 ChatRequest를 넣어줍니다. 그리고 StreamingResponse로 return 하는데 이것은 FastAPI가 자체 구현한 게 아니라 Starlette에서 온 것입니다. 그럼 Starlette은 뭐냐? Starlette는 FastAPI의 기반 프레임워크로, HTTP 요청/응답 처리, 라우팅, 미들웨어 등 저수준 웹 기능을 담당합니다.
아무튼 이런 식으로 구현이 가능합니다.
이렇게 SSE 구현이 끝났는데요, 이러면 눈으로 확인을 못해보니 저의 충신 claude code를 시켜서 웹 페이지를 하나 만들겠습니다.
"해줘"
SSE 구현해서 확인해려고 하는데 간단하게 html로 페이지 작성해 달라고 했더니 아래와 같은 것을 던져주었습니다. 길어서 필요하신 분만 보시면 될 듯합니다.
index.html
# index.html
<!DOCTYPE html>
<html lang="ko">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>SSE Streaming Chat</title>
<style>
* { margin: 0; padding: 0; box-sizing: border-box; }
body {
font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif;
max-width: 720px;
margin: 0 auto;
padding: 24px;
background: #f5f5f5;
}
h1 { margin-bottom: 16px; font-size: 1.4rem; }
#input-area {
display: flex;
gap: 8px;
margin-bottom: 16px;
}
#prompt {
flex: 1;
padding: 10px 14px;
font-size: 1rem;
border: 1px solid #ccc;
border-radius: 6px;
outline: none;
}
#prompt:focus { border-color: #555; }
button {
padding: 10px 20px;
font-size: 1rem;
background: #333;
color: #fff;
border: none;
border-radius: 6px;
cursor: pointer;
}
button:disabled { background: #999; cursor: not-allowed; }
#output {
background: #fff;
border: 1px solid #ddd;
border-radius: 8px;
padding: 16px;
min-height: 200px;
white-space: pre-wrap;
word-break: break-word;
line-height: 1.6;
font-size: 1rem;
}
</style>
</head>
<body>
<h1>SSE Streaming Chat</h1>
<div id="input-area">
<input id="prompt" type="text" placeholder="메시지를 입력하세요..." />
<button id="send" onclick="sendChat()">전송</button>
</div>
<div id="output"></div>
<script>
const promptInput = document.getElementById("prompt");
const sendBtn = document.getElementById("send");
const output = document.getElementById("output");
promptInput.addEventListener("keydown", (e) => {
if (e.key === "Enter" && !sendBtn.disabled) sendChat();
});
async function sendChat() {
const prompt = promptInput.value.trim();
if (!prompt) return;
output.textContent = "";
sendBtn.disabled = true;
try {
const response = await fetch("/chat", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ prompt }),
});
const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = "";
while (true) {
const { done, value } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
const lines = buffer.split("\n");
buffer = lines.pop(); // 마지막 불완전한 줄은 버퍼에 유지
for (const line of lines) {
if (!line.startsWith("data: ")) continue;
const data = line.slice(6);
if (data === "[DONE]") break;
output.textContent += data;
}
}
} catch (err) {
output.textContent += "\n[오류: " + err.message + "]";
} finally {
sendBtn.disabled = false;
promptInput.value = "";
promptInput.focus();
}
}
</script>
</body>
</html>
그리고 페이지 서빙하려면 main에도 엔드포인트를 하나 추가해야 합니다.
@app.get("/")
async def root():
html = Path("index.html").read_text(encoding="utf-8")
return HTMLResponse(html)
다 완성하셨다면
uvicorn main:app --reload를 통해서 확인해 보시면 됩니다.
동영상 녹화는 귀찮아서 안 했고요. 질문을 하면 응답이 타자를 치는 것처럼 stream 됩니다.
오늘은 SSE 가 무엇인지 알아보고 간단한 코드 구현 실습까지 진행해 보았는데요. 네 그렇습니다. 감사합니다.
'식빵 > AI' 카테고리의 다른 글
| Claude opus 4.6 시스템 프롬프트 유출 (0) | 2026.02.09 |
|---|---|
| Codex 에이전트 구조 분석 : 완전 쉬움 (2) | 2026.01.31 |