Python 스트리밍은 화면과 최종 회계를 나눠 다뤄

편의 도우미와 원시 반복자 중 골라

client.messages.stream(...)은 text_stream, get_final_message, on_event를 제공하는 문맥 관리자야. 채팅 화면처럼 텍스트와 최종 결과가 필요한 곳에 편해. client.messages.create(stream=True)는 사건을 직접 다루는 낮은 수준의 반복자를 돌려주므로, 모든 사건의 순서를 직접 통제할 때 써.

사용량은 스트림이 닫힌 뒤 확정돼

텍스트 조각이 오는 동안에는 최종 토큰 사용량을 알 수 없어. 편의 도우미에서는 반복이 끝난 뒤 stream.get_final_message()로 확인하고, 원시 반복자에서는 message_delta 사건을 모아야 해. 화면에 보낼 내용과 비용 장부를 같은 시점의 값으로 착각하지 마.

사용자가 보기 전에 기록해

세션을 재생해야 하는 cwkPippa 같은 애플리케이션은 각 텍스트 델타를 화면에 보내기 전에 영구 저장소에 덧붙여. JSONL 한 줄마다 명시적으로 비우면 중간에 프로세스가 죽어도 사용자가 실제로 본 내용이 기록에 남아. 재생 기록과 화면이 갈라지지 않는 순서야.

원칙: 스트리밍은 체감 속도를, 최종 메시지는 사용량 회계를 맡아. 둘의 완료 시점을 분리해.

Code

헬퍼 스타일 스트리밍 + usage·python

from anthropic import Anthropic

client = Anthropic()

with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Three tips for streaming UX."}],
) as stream:
    for chunk in stream.text_stream:
        print(chunk, end="", flush=True)
    final = stream.get_final_message()

print()
print("input:", final.usage.input_tokens, "output:", final.usage.output_tokens)

Low-level iterator + JSONL persistence·python

import json, pathlib
from anthropic import Anthropic

client = Anthropic()
session_log = pathlib.Path("/tmp/session.jsonl").open("a", buffering=1)  # line-buffered

stream = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Stream and persist"}],
    stream=True,
)

for event in stream:
    if event.type == "content_block_delta" and event.delta.type == "text_delta":
        session_log.write(json.dumps({"delta": event.delta.text}) + "\n")
        session_log.flush()
        print(event.delta.text, end="", flush=True)
    elif event.type == "message_stop":
        # Final usage는 message에 — 필요하면 조립된 응답에서 캡처
        print()

session_log.close()

편의 도우미와 원시 반복자 중 골라

사용량은 스트림이 닫힌 뒤 확정돼

사용자가 보기 전에 기록해

Code

External links

Exercise

Progress

댓글 0