ChatGPT처럼 글자가 흘러나오게 만들기#
일반 API 호출은 응답이 완성될 때까지 기다립니다. 긴 응답은 5~10초 이상 화면이 멈춥니다.
Diagram · Mermaid
스트리밍을 쓰면 첫 토큰이 즉시 출력되어 사용자가 더 빠르다고 느낍니다.
스트리밍 기본 구현#
Python
from openai import OpenAI
client = OpenAI()
stream = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "파이썬의 장점을 설명해줘"}],
stream=True # 스트리밍 활성화
)
for chunk in stream:
if chunk.choices[0].delta.content is not None:
print(chunk.choices[0].delta.content, end="", flush=True)
print() # 마지막 줄바꿈stream=True 하나로 전환됩니다. 응답은 작은 chunk 단위로 순서대로 도착합니다.
스트리밍 청크 구조#
Diagram · Mermaid
Python
for chunk in stream:
delta = chunk.choices[0].delta
finish = chunk.choices[0].finish_reason
if delta.content:
print(delta.content, end="", flush=True)
if finish == "stop":
print("\n[스트리밍 완료]")스트리밍 텍스트 수집#
스트리밍하면서 전체 텍스트를 나중에 사용하고 싶을 때:
Python
def stream_and_collect(messages: list) -> str:
full_response = []
stream = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
stream=True
)
for chunk in stream:
content = chunk.choices[0].delta.content
if content:
full_response.append(content)
print(content, end="", flush=True)
print()
return "".join(full_response)
result = stream_and_collect([
{"role": "user", "content": "간단한 파이썬 정렬 코드 예시"}
])
# 이후 result를 DB에 저장하거나 후처리비동기 처리: 여러 요청 동시에#
Diagram · Mermaid
여러 텍스트를 동시에 처리할 때 비동기를 쓰면 N배 빠릅니다.
비동기 API 호출#
Python
import asyncio
from openai import AsyncOpenAI
async_client = AsyncOpenAI()
async def analyze_one(text: str, idx: int) -> dict:
response = await async_client.chat.completions.create(
model="gpt-4o-mini",
temperature=0,
messages=[
{"role": "system", "content": "감정을 positive/negative/neutral로 분류해줘. 단어 하나만 출력."},
{"role": "user", "content": text}
]
)
return {
"idx": idx,
"text": text[:30],
"sentiment": response.choices[0].message.content.strip()
}
async def analyze_batch(texts: list[str]) -> list[dict]:
tasks = [analyze_one(text, i) for i, text in enumerate(texts)]
results = await asyncio.gather(*tasks) # 동시 실행
return list(results)
# 실행
reviews = [
"정말 좋은 제품이에요!",
"배송이 너무 느렸습니다.",
"그냥 보통이에요.",
"다시는 구매 안 합니다.",
"최고예요, 추천합니다!"
]
results = asyncio.run(analyze_batch(reviews))
for r in sorted(results, key=lambda x: x["idx"]):
print(f"{r['idx']+1}. [{r['sentiment']}] {r['text']}")비동기 스트리밍#
Python
async def stream_async(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:
content = chunk.choices[0].delta.content
if content:
print(content, end="", flush=True)
print()
asyncio.run(stream_async("FastAPI란 무엇인가요?"))동시 요청 수 제한 (Rate Limit 관리)#
OpenAI API는 분당 요청 수(RPM)와 분당 토큰 수(TPM)에 제한이 있습니다.
Diagram · Mermaid
Python
async def analyze_with_limit(texts: list[str], concurrency: int = 10):
semaphore = asyncio.Semaphore(concurrency)
async def limited_call(text, idx):
async with semaphore: # 최대 concurrency개만 동시 실행
return await analyze_one(text, idx)
tasks = [limited_call(t, i) for i, t in enumerate(texts)]
return await asyncio.gather(*tasks)FastAPI + 스트리밍: 웹 API 만들기#
Python
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import AsyncOpenAI
import asyncio
app = FastAPI()
client = AsyncOpenAI()
async def generate_stream(prompt: str):
stream = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
stream=True
)
async for chunk in stream:
content = chunk.choices[0].delta.content
if content:
yield content
@app.get("/chat")
async def chat(prompt: str):
return StreamingResponse(
generate_stream(prompt),
media_type="text/plain"
)이 패턴으로 프론트엔드에서 실시간 타이핑 효과를 구현할 수 있습니다.
성능 비교 측정#
Python
import time
async def benchmark():
texts = ["리뷰 텍스트"] * 20
# 동기 처리
start = time.time()
for text in texts:
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": text}]
)
sync_time = time.time() - start
# 비동기 처리
start = time.time()
await analyze_batch(texts)
async_time = time.time() - start
print(f"동기: {sync_time:.1f}초")
print(f"비동기: {async_time:.1f}초")
print(f"속도 향상: {sync_time/async_time:.1f}배")정리#
| 개념 | 내용 | 사용 상황 |
|---|---|---|
| 스트리밍 | 토큰 단위 실시간 출력 | 챗봇, 글쓰기 보조 |
| AsyncOpenAI | 비동기 API 클라이언트 | 대량 처리, 서버 |
| asyncio.gather | 여러 요청 동시 실행 | 배치 분류, 번역 |
| Semaphore | 동시 요청 수 제한 | Rate limit 관리 |
| StreamingResponse | FastAPI 스트리밍 응답 | 웹 서비스 구축 |
다음 편에서는 Function Calling / Tool Use — LLM이 외부 함수를 호출해 실시간 데이터를 처리하는 강력한 기능을 배웁니다.
