스트리밍 응답
스트리밍 응답(Streaming)을 사용하면 모델 생성 과정에서 실시간으로 출력을 수신할 수 있어 사용자 경험과 체감 속도를 향상시킬 수 있습니다.
동작 원리
OfoxAI는 Server-Sent Events (SSE) 프로토콜을 사용하여 스트리밍 응답을 구현합니다:
- 클라이언트가 요청을 보낼 때
stream: true를 설정합니다 - 서버가 생성된 콘텐츠 조각(chunk)을 순차적으로 반환합니다
- 각 chunk는
data:접두사와 함께 SSE를 통해 전송됩니다 - 생성이 완료되면
data: [DONE]을 전송합니다
OpenAI 프로토콜 스트리밍
cURL
Terminal
curl https://api.ofox.ai/v1/chat/completions \
-H "Authorization: Bearer $OFOX_API_KEY" \
-H "Content-Type: application/json" \
-N \
-d '{
"model": "openai/gpt-4o",
"messages": [{"role": "user", "content": "프로그래밍에 대한 시를 써 주세요"}],
"stream": true
}'Anthropic 프로토콜 스트리밍
Python
stream_anthropic.py
import anthropic
client = anthropic.Anthropic(
base_url="https://api.ofox.ai/anthropic", # OfoxAI API 주소
api_key="<귀하의 OFOXAI_API_KEY>" # https://app.ofox.ai 에서 발급
)
with client.messages.stream(
model="anthropic/claude-sonnet-4.5",
max_tokens=1024,
messages=[{"role": "user", "content": "프로그래밍에 대한 시를 써 주세요"}]
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)스트리밍 + Function Calling
스트리밍 응답은 함수 호출 시나리오도 지원합니다. 모델이 먼저 도구 호출 요청을 스트리밍으로 출력하고, 처리 완료 후 대화를 계속합니다:
stream_with_tools.py
stream = client.chat.completions.create(
model="openai/gpt-4o",
messages=[{"role": "user", "content": "오늘 서울 날씨는 어떤가요?"}],
tools=[{
"type": "function",
"function": {
"name": "get_weather",
"description": "지정된 도시의 날씨를 조회합니다",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "도시 이름"}
},
"required": ["city"]
}
}
}],
stream=True
)
for chunk in stream:
delta = chunk.choices[0].delta
if delta.tool_calls:
# 도구 호출 처리
print(f"도구 호출: {delta.tool_calls[0].function}")
elif delta.content:
print(delta.content, end="", flush=True)오류 처리 및 재연결
스트리밍 연결은 네트워크 문제로 인해 중단될 수 있습니다. 재연결 로직을 구현하는 것을 권장합니다.
stream_retry.py
import time
def stream_with_retry(client, max_retries=3, **kwargs):
for attempt in range(max_retries):
try:
stream = client.chat.completions.create(stream=True, **kwargs)
for chunk in stream:
yield chunk
return # 성공적으로 완료
except Exception as e:
if attempt < max_retries - 1:
wait = 2 ** attempt # 지수 백오프
print(f"\n연결이 중단되었습니다. {wait}초 후 재시도...")
time.sleep(wait)
else:
raise e모범 사례
- 항상 타임아웃을 설정하세요 — 무한 대기를 방지합니다
- 불완전한 chunk를 처리하세요 — 일부 chunk에는 content가 없을 수 있습니다
- 재연결 메커니즘을 구현하세요 — 지수 백오프 전략을 사용합니다
- 프론트엔드에서
flush를 사용하세요 — 콘텐츠가 즉시 표시되도록 합니다
Last updated on