긴 응답이 버벅일 때 — Streaming API로 토큰을 실시간으로 받아 처리하는 법

언제 Streaming이 필수인가

일반 messages.create는 모델이 응답을 완전히 생성한 뒤 한 번에 반환합니다. 응답이 500 토큰만 넘어가도 수 초의 공백이 생깁니다. Streaming은 토큰이 생성되는 즉시 청크 단위로 전송하기 때문에 사용자 체감 속도가 극적으로 개선됩니다.

추가로, stream.finalMessage()를 활용하면 실제 소비 토큰을 정확히 추적할 수 있어 비용 모니터링에도 유용합니다.

Streaming 작동 방식 이해하기

[모델 내부]  토큰1 → 토큰2 → 토큰3 → ... → 토큰N
                ↓        ↓        ↓              ↓
[클라이언트] 즉시렌더  즉시렌더  즉시렌더  ...  완료 이벤트

Anthropic SDK는 stream() 헬퍼를 통해 이벤트 기반으로 청크를 받습니다. 주요 이벤트:

| 이벤트 | 설명 | |--------|------| | text | 텍스트 토큰 한 조각 | | message | 전체 완성 메시지 (마지막) | | finalMessage() | 토큰 사용량 포함 최종 객체 |

Python으로 Streaming + 토큰 모니터링 구현

import anthropic
import time

client = anthropic.Anthropic()

def stream_with_token_tracking(prompt: str, max_tokens: int = 1024) -> dict:
    """
    스트리밍으로 응답을 받으면서 토큰 사용량과 TTFT를 측정합니다.
    TTFT = Time To First Token (첫 번째 토큰까지 걸린 시간)
    """
    collected_text = []
    start_time = time.time()
    first_token_time = None

    print("[응답 시작]\n")

    with client.messages.stream(
        model="claude-sonnet-4-6",
        max_tokens=max_tokens,
        messages=[{"role": "user", "content": prompt}],
    ) as stream:
        for text_chunk in stream.text_stream:
            # 첫 토큰 도착 시간 기록
            if first_token_time is None:
                first_token_time = time.time()
                ttft = first_token_time - start_time
                print(f"  ⚡ TTFT: {ttft:.2f}초\n")

            print(text_chunk, end="", flush=True)  # 실시간 출력
            collected_text.append(text_chunk)

        # 스트림 종료 후 토큰 사용량 확인
        final = stream.get_final_message()
        usage = final.usage

    total_time = time.time() - start_time
    full_text = "".join(collected_text)

    print(f"\n\n[스트리밍 완료]")
    print(f"  입력 토큰  : {usage.input_tokens:,}")
    print(f"  출력 토큰  : {usage.output_tokens:,}")
    print(f"  총 소요시간: {total_time:.2f}초")
    print(f"  토큰 처리속도: {usage.output_tokens / total_time:.1f} tokens/sec")

    return {
        "text": full_text,
        "input_tokens": usage.input_tokens,
        "output_tokens": usage.output_tokens,
        "ttft_sec": ttft,
        "total_sec": total_time,
    }


# 사용 예
result = stream_with_token_tracking(
    prompt="파이썬에서 비동기 프로그래밍이 필요한 이유와 asyncio 기본 패턴을 설명해줘.",
    max_tokens=800,
)

토큰 절약 팁: max_tokens를 실제 필요 분량보다 20~30% 더 잡아두면 중간 잘림을 방지하면서도 낭비를 줄일 수 있습니다. 스트리밍 중 원하는 구간에서 .close()를 호출해 조기 종료도 가능합니다.

운영 환경 적용 체크리스트

• [ ] 프론트엔드에서 SSE(Server-Sent Events) 또는 WebSocket으로 청크를 실시간 렌더링하고 있는가?
• [ ] 네트워크 단절 시 재연결 로직(retry 또는 reconnect)이 있는가?
• [ ] output_tokens를 로깅해 일별·사용자별 비용을 추적하고 있는가?
• [ ] temperature: 0이 필요한 태스크에서 실수로 streaming과 혼용하지 않았는가? (streaming은 temperature와 무관하게 동작)
• [ ] 스트림 완료 전에 UI가 '로딩 중' 상태를 명확히 표시하는가?
• [ ] 타임아웃 설정이 일반 요청보다 충분히 길게 잡혀 있는가? (스트림은 전체 생성 시간만큼 연결 유지 필요)