스트리밍 헬퍼 · final message · usage

Python에서 스트리밍 두 가지

SDK가 스트리밍 진입점 두 개 — client.messages.stream(...)은 풍부한 편의 메서드(text_stream, get_final_message, on_event) 가진 context-manager 스타일 헬퍼. client.messages.create(stream=True)는 raw 이벤트 iterator 반환하는 lower-level 경로. 챗 UI엔 헬퍼; 이벤트 처리 풀 컨트롤 원하면 iterator.

get_final_message가 usage 줘

텍스트 스트림 동안엔 토큰 usage 아직 없음. Usage는 message_stop에서 확정. 헬퍼는 iteration 완료 후 stream.get_final_message()로 노출; iterator 경로는 message_delta 이벤트 캡처해서 누적해야.

스트림 진행하면서 persist

세션 로그·replay하는 애플리케이션(cwkPippa)이면, 각 텍스트 delta를 user에게 렌더 전에 영구 저장에 써. 매 라인 후 explicit flush와 함께 JSONL append-only면 stream 중간 hard kill에도 user가 실제 본 게 완전한 기록으로 남아.

원칙: UX엔 stream, accounting엔 finalize. Usage는 마감 accounting 라인이지 per-delta 숫자 아님.

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()

Exercise

스트리밍 챗 핸들러를 각 텍스트 delta가 HTTP stream으로 yield되기 전에 JSONL 파일에 쓰이게 다시 짜. SIGTERM으로 응답 중간 프로세스 죽이고 JSONL에 모든 visible 토큰이 그대로 있는지 확인.

Hint

buffering=1(line-buffered)로 파일 열고 매 write 후 flush() 호출.

스트리밍 헬퍼 · final message · usage

Python에서 스트리밍 두 가지

get_final_message가 usage 줘

스트림 진행하면서 persist

Code

External links

Exercise

Progress

댓글 0