스트리밍은 첫 토큰을 앞당기는 대신 상태를 떠안아

사용자가 기다릴 때 값이 생겨

일반 Messages 호출은 응답 전체가 완성될 때까지 돌려주지 않아. 긴 답변에서는 화면이 몇 초씩 멈춘 것처럼 보이지. 스트리밍은 Server-Sent Events로 생성 중인 내용을 보내 마지막 토큰까지의 총시간보다 첫 토큰까지의 체감 시간을 줄여 줘.

다섯 사건의 순서를 익혀

SDK는 SSE 전송 형식을 타입이 있는 사건으로 바꿔 줘. message_start에서 응답 껍데기가 열리고, content_block_start에서 텍스트나 도구 블록이 시작돼. content_block_delta가 내용을 이어 붙이고, content_block_stop이 블록을 닫으며, message_stop이 전체 응답을 끝내. 도구 호출은 자체 델타 형식을 가진 별도 블록으로 와.

부분 결과가 생긴 순간 책임도 생겨

중간에 연결이 끊기면 이미 받은 출력 때문에 자동 재시도가 안전하지 않을 수 있어. 최종 토큰 사용량은 message_stop에서 확정되고, 도구 입력은 델타를 모두 모은 뒤에야 해석할 수 있어. 빠른 화면을 얻는 대신 부분 상태와 중단 복구를 직접 설계해야 해.

원칙: 사람이 기다리는 경로는 스트리밍하고, 백그라운드 작업은 완성본을 받아. 필요 없는 복잡성까지 사지는 마.

Code

Python SDK 헬퍼로 스트리밍·python

with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=2048,
    messages=[{"role": "user", "content": "Explain prompt caching in three short paragraphs."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    final_message = stream.get_final_message()

print()
print("usage:", final_message.usage)

직접 짠 이벤트 처리 (TypeScript)·typescript

import Anthropic from '@anthropic-ai/sdk';
const client = new Anthropic();

const stream = await client.messages.stream({
  model: 'claude-sonnet-4-6',
  max_tokens: 1024,
  messages: [{ role: 'user', content: 'List three SSE pitfalls.' }],
});

for await (const event of stream) {
  if (event.type === 'content_block_delta' && event.delta.type === 'text_delta') {
    process.stdout.write(event.delta.text);
  }
  if (event.type === 'message_stop') {
    process.stdout.write('\n');
  }
}

const final = await stream.finalMessage();
console.log('usage:', final.usage);

Exercise

Messages 응답을 스트리밍해 각 델타를 JSONL에 먼저 쓰고 stdout에 출력하는 작은 CLI를 만들어. 중간에 프로세스를 끊은 뒤 화면에 나온 토큰이 모두 파일에 있는지 확인해.

Hint

파일을 줄 단위 버퍼로 열고 매 쓰기 뒤 flush()를 호출해 마지막 완성 줄을 지켜.

사용자가 기다릴 때 값이 생겨

다섯 사건의 순서를 익혀

부분 결과가 생긴 순간 책임도 생겨

Code

External links

Exercise

Progress

댓글 0