Streaming API로 답변 대기 시간을 체감 0으로 만드는 UX 설계

왜 Streaming인가?

일반 API 호출은 모델이 전체 응답을 완성한 뒤 한 번에 반환합니다. 응답이 500토큰이라면 사용자는 수 초간 빈 화면을 봐야 합니다.

Streaming은 토큰이 생성되는 즉시 클라이언트로 전달하여:

• 체감 응답 속도를 획기적으로 개선
• 사용자가 틀린 방향의 답변을 중간에 중단 가능
• 긴 문서 생성 시 진행 상황 파악 가능

Streaming vs Non-Streaming 비교

| 항목 | 일반 호출 | Streaming | |------|-----------|----------| | 첫 응답까지 대기 | 전체 생성 완료 후 | 첫 토큰 즉시 (~200ms) | | 중간 취소 | 불가 | 가능 (controller.abort()) | | 구현 복잡도 | 낮음 | 중간 | | 적합한 상황 | 짧은 응답, 배치 처리 | 챗봇, 문서 생성, 실시간 분석 |

Anthropic SDK 실전 예제 (Python)

import anthropic
import sys

client = anthropic.Anthropic()


def stream_response(user_message: str) -> str:
    """스트리밍으로 응답을 출력하고 전체 텍스트를 반환합니다."""
    full_text = ""
    input_tokens = 0
    output_tokens = 0

    try:
        # stream() 컨텍스트 매니저 사용
        with client.messages.stream(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            system="당신은 친절한 기술 문서 작성 전문가입니다.",
            messages=[{"role": "user", "content": user_message}],
        ) as stream:
            print("AI: ", end="", flush=True)

            # 토큰 단위로 즉시 출력
            for text_chunk in stream.text_stream:
                print(text_chunk, end="", flush=True)
                full_text += text_chunk

            # 스트림 종료 후 사용량 확인
            final_message = stream.get_final_message()
            input_tokens = final_message.usage.input_tokens
            output_tokens = final_message.usage.output_tokens

    except anthropic.APIConnectionError:
        print("\n[오류] 네트워크 연결을 확인하세요.", file=sys.stderr)
        raise
    except anthropic.RateLimitError:
        print("\n[오류] API 요청 한도 초과. 잠시 후 재시도하세요.", file=sys.stderr)
        raise
    except anthropic.APIStatusError as e:
        print(f"\n[오류] API 오류 {e.status_code}: {e.message}", file=sys.stderr)
        raise
    finally:
        print()  # 줄바꿈
        if input_tokens:
            print(f"[토큰 사용: 입력 {input_tokens} / 출력 {output_tokens}]")

    return full_text


# 실제 사용 예
result = stream_response(
    "Python asyncio의 gather와 wait의 차이점을 "
    "실무 예제와 함께 설명해 주세요."
)

# 결과를 파일 저장 등 후처리에 활용
with open("output.md", "w", encoding="utf-8") as f:
    f.write(result)

주의: stream.text_stream 은 텍스트 델타만 반환합니다. 도구 호출(tool use) 결과가 포함된 경우 stream.event_stream 으로 전체 이벤트를 핸들링해야 합니다.

마무리 체크리스트

• [ ] 응답 길이가 300토큰 이상 예상되면 Streaming 적용 검토
• [ ] flush=True (Python) 또는 즉시 DOM 업데이트로 버퍼링 방지
• [ ] APIConnectionError, RateLimitError 각각 다르게 핸들링
• [ ] 스트리밍 중 사용자 취소 버튼 UI 제공 (긴 생성 시 필수)
• [ ] get_final_message() 로 토큰 사용량 로깅 → 비용 모니터링
• [ ] 짧은 응답(분류, Yes/No 판단)은 일반 호출이 코드가 더 단순