스트리밍 출력 — 덜 끝난 JSON과 처리 흐름 · Prompt Quest

스트리밍에서는 완성 전의 JSON을 받아

구조화 출력을 흘려보내면 {"verdict": "app처럼 아직 닫히지 않은 JSON 조각이 도착해. 매 조각에 곧장 json.loads()를 부르면 모두 실패하지. 처리가 끝날 때까지 버퍼에 모으거나, 스트리밍 JSON 파서로 완성된 부분부터 읽어야 해.

그래도 스트리밍하는 까닭

사용자 경험 — 8초를 기다리게 하지 않고 200밀리초 안에 첫 결과를 보여줄 수 있어.
단계 겹치기 — 전체 응답이 끝나기 전에 첫 필드가 완성되는 순간 다음 작업을 시작할 수 있어.
취소 — 사용자가 화면을 떠나거나 검증기가 거절하면 곧장 요청을 멈출 수 있어.

스트리밍에서만 두드러지는 실패

max_tokens 도달이나 통신 끊김으로 JSON이 중간에서 잘려.
도구 호출이 start, delta, stop 조각으로 나뉘어 와.
추론 토큰이 최종 출력과 별도 흐름으로 올 수 있어.
소비자가 연결을 끊어도 취소 신호를 연결하지 않은 서버는 계속 계산해.

Code

스트리밍 JSON을 안전하게 읽기·python

import json_stream

buffer = []
with client.messages.stream(
    model="claude-opus-4-7",
    messages=[...],
    response_format={"type": "json_object"},
) as stream:
    for delta in stream.text_stream:
        buffer.append(delta)
        # incremental: try to parse, ignore JSONDecodeError
        try:
            partial = json_stream.loads("".join(buffer))
            on_partial(partial)  # update UI as fields arrive
        except Exception:
            pass
final = json.loads("".join(buffer))

스트리밍 출력 — 덜 끝난 JSON과 처리 흐름

스트리밍에서는 완성 전의 JSON을 받아

그래도 스트리밍하는 까닭

스트리밍에서만 두드러지는 실패

Code

External links

Exercise

Progress

댓글 0