유효한 JSON을 안정적으로 반환하는 모델이라 할지라도, 텍스트를 생성(typing)하는 도중에 유효한 JSON을 반환하는 것은 아닙니다. 이 둘은 서로 다른 보장 사항이며, 대부분의 스트리밍(streaming) 클라이언트는 이 둘이 같다고 조용히 가정해 버립니다. 제약된 디코딩(Constrained decoding)과 스키마 강제 출력(schema-enforced output) 모드는 최종 응답이 파싱될 것임을 보장합니다. 하지만 열일곱 번째 청크(chunk)가 키(key)의 중간에서 끝나거나, 이스케이프 시퀀스(escape sequence) 내부에서 끝나거나, 혹은 아직 뒤에 요소가 없는 쉼표(comma) 바로 뒤에서 끝날 수도 있다는 점에 대해서는 아무것도 보장하지 않습니다.

따라서 일반적인 타협안은 진행률 스피너(progress spinner)를 위해 토큰을 스트리밍하면서도 어쨌든 모두 버퍼(buffer)에 쌓아두었다가, 마지막에 한 번에 파싱하는 것입니다. 사용자는 애플리케이션 자체가 읽기를 거부하는 문자들을 나타나는 것을 지켜보게 됩니다. 모델이 이미 결정한 모든 것 — 제목, 처음 세 개의 리스트 항목, 첫 50개 토큰 안에 도착한 분류 레이블(classification label) — 은 닫는 중괄호가 도착할 때까지 문자열 버퍼 내에서 사용할 수 없는 상태로 머물러 있습니다.

이 간극은 메울 수 있습니다. 이어지는 내용은 스트림이 현재 정당화할 수 있는 최선의 완전한 값으로 모든 청크를 변환하는 부분적 JSON 리더(partial JSON reader)를 구축하는 방법을 다루며, 그 후 이 아이디어의 일반적인 버전이 데이터를 조용히 손상시키는 두 가지 사례를 보여줍니다.

스트리밍 모델이 실제로 당신에게 보내는 것

스트리밍 완료(streaming completion)는 JSON 구조와 아무런 관계가 없는 텍스트의 델타(deltas) 형태로 도착합니다. 토큰 경계는 문법(grammar)이 아니라 토크나이저(tokenizer)의 어휘(vocabulary)를 따릅니다. 단일 델타는 {"ti를 포함하거나, tle": "Sh를 포함하거나, 혹은 다음 델타가 n을 제공할 때 비로소 의미를 갖게 되는 단독 백슬래시(\)를 포함할 수 있습니다.

클라이언트 루프(client loop)는 쉬운 부분입니다:

import json
from openai import OpenAI

...

해당 제너레이터(generator)의 모든 요소는 아직 문서가 아닌 문서의 파편입니다. 문제는 소비자가 단순히 이를 추가(append)하는 것 외에 각 요소로 무엇을 할 수 있느냐는 것입니다.

스트리밍 도중에 json.loads가 잘못된 도구인 이유

표준 라이브러리 파서(parser)는 설계상 '전부 아니면 전무(all-or-nothing)' 방식입니다. 접두사(prefix)를 전달하면 에러를 발생시키는데, JSON 문서의 접두사는 JSON 문서가 아니기 때문입니다.

buffer = ""
for delta in stream_text(prompt):
    buffer += delta
...

json.JSONDecoder.raw_decode는 탈출구(escape hatch)처럼 보이며 실제로도 유용하지만, 다른 문제를 해결하기 위한 것입니다. 이 함수는 문자열의 맨 앞에서 하나의 완전한 값을 디코딩하고 어디서 멈췄는지를 보고합니다. 이는 별개의 객체들이 스트림(stream) 내에 연속적으로 이어져 있는 경우를 해결해 줍니다. 하지만 원하는 단일 객체가 잘려 있는(truncated) 경우에는 도움이 되지 않는데, 디코딩할 수 있는 완전한 값이 맨 앞에 아직 존재하지 않기 때문입니다.

ijson 패키지는 그나마 더 가깝습니다. 이는 바이트가 도착함에 따라 start_map, map_key, string, end_map 이벤트를 생성하는 이벤트 기반 파서(event-driven parser)이며, 이는 정확히 사용자가 원하는 스트리밍 형태입니다. 다만 이 패키지의 제약 사항은 매우 크지만 결국에는 완전한 문서들을 위해 구축되었다는 점입니다. 잘려 있는 피드(feed)는 IncompleteJSONError로 종료되며, 이미 닫힌 값들에 대해서만 이벤트를 받을 수 있습니다. 문자열이 작성되는 동안 이를 보여주고자 하는 렌더링 UI(rendering UI) 입장에서는, '완료된 값당 하나의 이벤트' 방식은 너무 거칠고(coarse) 단계가 높습니다.

진정한 증분 파서(incremental parser)를 작성하는 것이 철저한 해답이겠지만, 이는 문제의 난이도에 비해 너무 많은 작업이 필요합니다. 토크나이저(tokenizer), 문법(grammar)에 대한 상태 머신(state machine), 그리고 중간 단계에서 조회가 가능한 값 빌더(value builder)를 작성하려면 수백 줄의 코드가 필요하며, 이를 신뢰하기 위해서는 훨씬 더 많은 노력이 필요합니다. 남은 접근 방식은 더 저렴하며 이미 올바르게 작동하는 파서를 재사용하는 것입니다. 즉, 접두사(prefix)를 유효한 문서로 수리(repair)하고, 그 수리된 내용을 파싱한 뒤, 수리된 내용을 버리는 방식입니다. 버퍼 자체는 절대 수정되지 않으므로, 수리 로직에 실수가 있더라도 스트림이 오염되는 대신 잘못된 스냅샷(snapshot) 하나를 생성하는 것으로 끝납니다.

지금까지의 내용 정리

핵심 아이디어는 간단합니다. 어떤 컨테이너(container)들이 열려 있는지 추적하고, 스냅샷이 요청되면 그것들을 닫아줄 닫기 기호(closers)들을 추가하는 것입니다.

def naive_snapshot(buffer: str):
    stack = []
    for ch in buffer:
...

잘 구성된 접두사의 경우 이 방식은 즉시 작동합니다. {"title": "Ship it", "tags": ["python"는 {"title": "Ship it", "tags": ["python"]}가 되어, 응답이 끝나기 수백 토큰 전이라도 템플릿이 즉시 렌더링할 수 있는 딕셔너리(dictionary)로 파싱됩니다.

또한 끊임없이 실패가 발생하며, 이러한 실패 사례들은 성공 사례보다 더 흥미롭습니다. {"tags": ["python",는 trailing comma(마지막 쉼표)로 닫힙니다. {"score":는 값이 없는 키로 닫힙니다. {"score": 1.은 소수점으로 끝나는 숫자 리터럴(number literal)로 닫힙니다. 이 각각은 모두 디코딩 에러(decode error)이므로, 스냅샷은 None으로 반환되며 스트림이 운 좋게 경계선에 걸릴 때까지 UI는 멈춰 있게 됩니다.

문자열, 이스케이프(Escapes), 그리고 데이터를 오염시키는 복구 작업

위험한 실패는 소음이 발생하는 실패와는 다릅니다. 소음이 발생하는 실패는 스스로를 알립니다. trailing comma가 발생하면 에러가 발생하고 스냅샷은 None이 되며, 다음 청크(chunk)가 보통 이를 해결합니다. 하지만 문자열 값 안에 있는 중괄호는 중괄호가 아니며, 이때는 아무런 에러도 발생하지 않습니다.

{"note": "use {curly} braces라는 접두사를 생각해 보십시오. 위의 스캐너(scanner)는 산문(prose) 내의 {를 카운트하고, 스택(stack)에 두 번째 }를 푸시(push)하며, 일치하는 닫는 괄호를 찾지 못해 {"note": "use {curly} braces}}를 생성합니다. 이것은 디코딩 에러가 아닙니다. 스트림이 멈춘 위치에 따라, 이러한 복구 작업은 모델의 산문으로부터 구조가 발명되어 잘못된 형태(shape)로 파싱되는 문서를 만들어낼 수 있습니다.

이것은 드문 입력도 아닙니다. 코드, 파일 경로, 또는 템플릿 구문(template syntax)에 대해 작성하는 모든 모델은 문자열 값 내부에서 중괄호와 대괄호를 끊임없이 생성하며, JSON에 관한 JSON 응답은 단순한 스캐너에게 최악의 사례입니다. 잘못된 형태로 파싱되는 스냅샷은 파싱에 실패하는 스냅샷보다 더 나쁩니다. 왜냐하면 소비자(consumer)는 무언가 잘못되었다는 신호를 전혀 받을 수 없기 때문입니다.

자신이 문자열 내부에 있는지 알지 못하는 스캐너는 추측을 하고 있는 것입니다. 이스케이프(escapes)에도 마찬가지입니다. 단일 \로 끝나는 버퍼는 부분적인 이스케이프 시퀀스(escape sequence)이며, "로 문자열을 닫으면 해당 trailing backslash(마지막 백슬래시)가 이스케이프된 따옴표가 되어, 닫는 따옴표를 삼켜버리고 오염을 한 단계 더 바깥으로 밀어냅니다.

올바른 처리를 위해서는 청크 간에 유지되어야 하는 세 가지 상태(state)가 필요합니다: 컨테이너 스택(container stack), 문자열 내부 플래그(in-string flag), 그리고 이스케이프 플래그(escape flag)입니다.

자신의 위치를 기억하는 스캐너

각 문자는 도착했을 때 단 한 번만 분류되며, 다시 스캔되지 않습니다. 또한 모든 프레임(frame)은 안전한 절단 오프셋(safe truncation offset) — 컨테이너가 완전한 요소들만 보유하고 있었던 시점 — 을 기록합니다. 따라서 복구(repair)에 실패하더라도 아무것도 반환하는 대신 마지막으로 확인된 양호한 경계(last known-good boundary)로 되돌아갈 수 있습니다.

import json
from dataclasses import dataclass

...

스냅샷(snapshot) 메서드는 가장 완전한 형태부터 가장 보수적인 형태까지 복구 후보군을 생성하며, 파싱에 성공하는 첫 번째 후보를 반환합니다:

    def _closers(self, upto: int | None = None) -> str:
        frames = self._stack if upto is None else self._stack[:upto]
        return "".join(frame.close for frame in reversed(frames))
...

폴백(fallback) 경로는 불완전한 숫자나 절반만 입력된 리터럴(literals)을 무해하게 만듭니다. 1. 또는 tru로 끝나는 버퍼의 경우, 먼저 실패하는 첫 번째 후보를 생성한 다음, 미완성된 요소를 완전히 제거하여 파싱 가능한 절단된 후보를 생성합니다. 호출자는 아무것도 보지 못하는 대신, 해당 키가 없는 객체를 보게 됩니다.

이 설계의 한 가지 속성은 이를 사용하는 모든 이에게 명확히 고지되어야 합니다: 스냅샷 내의 문자열 값은 실제 값의 접두사(prefix)일 수 있습니다. 값이 커짐에 따라 부분적인 설명을 렌더링하는 것이 의도된 용도입니다. 부분적인 값을 열거형(enum)과 비교하거나, URL로 취급하거나, 특정 동작을 수행하는 무언가에 전달하는 것은 느린 토큰(slow token)을 기다리다 발생할 버그입니다.

스냅샷을 필드 이벤트로 전환하기

필드가 완료되었을 때만 관심이 있는 소비자에게 청크(chunk)마다 전체 스냅샷을 폴링(polling)하는 것은 낭비입니다. JSON 객체는 순서대로 스트리밍되므로, 간단하고 신뢰할 수 있는 완료 규칙을 제공합니다: 두 번째 키가 나타나는 즉시, 첫 번째 키의 값은 더 이상 변경될 수 없습니다. 스냅샷의 마지막 키를 제외한 모든 키는 확정(settled)됩니다.

class FieldEmitter:
    def __init__(self) -> None:
        self.stream = PartialJSONStream()
...

이제 다운스트림 핸들러(downstream handler)는 모델이 explanation을 작성하는 동안에도 classification에 대한 작업을 시작할 수 있습니다. 이것이 바로 산문 형태의 덩어리(blob of prose)가 아닌 구조화된 응답(structured response)을 스트리밍하는 핵심 목적입니다. 라우팅 결정(routing decision)을 발송하거나, 데이터베이스 행(database row)을 예약하거나, UI 섹션을 스켈레톤(skeleton) 형태가 아닌 최종 형태로 렌더링할 수 있습니다.

이 순서 규칙에는 한 가지 조건이 붙습니다. 모델이 고정된 키 순서(fixed key order)로 작성하는 객체(object)에 대해서만 유효하며, 이는 스키마 제약 디코딩(schema-constrained decoding)이 생성하는 방식입니다. 나중에 나오는 요소가 이전 요소를 수정하지 않지만 마지막 요소가 여전히 성장 중인 객체 배열(array of objects)에는 적용되지 않으므로, 리스트의 마지막 요소는 객체의 마지막 키와 정확히 동일한 방식으로 임시(provisional) 상태로 취급하십시오.

검증, 취소, 그리고 회복 불가능한 출력

스냅샷(snapshot)은 초안이므로, 이를 엄격한 출력 모델(strict output model)에 따라 검증하는 것은 구조적으로 잘못된 방식입니다. 필수 필드(required fields)가 의도적으로 누락되어 있기 때문입니다. 스냅샷용으로는 모델의 완화된 미러(relaxed mirror)를 구축하고, 최종 값(final value)을 위해서는 엄격한 모델을 유지하십시오.

from pydantic import BaseModel, create_model

class Review(BaseModel):
...

이와 관련하여 두 가지 실패 모드(failure modes)에 대한 명시적인 처리가 필요합니다. 첫 번째는 모델이 구조 생성을 중단하고 사과 문구나 펜스 코드 블록(fenced code block)을 생성하기 시작하는 경우입니다. 이 경우 스냅샷은 None이 되고 계속 None 상태로 남게 되므로, 연속적으로 파싱할 수 없는 청크(chunk)를 추적하여 오지 않을 종료(close)를 기다리는 대신 스트림을 포기하십시오. 두 번째는 조기 종료(early exit)입니다. 스냅샷이 이미 엄격한 모델을 충족하고 남은 필드들이 선택 사항(optional)인 경우, 요청을 취소하여 아무도 읽지 않을 토큰에 비용을 지불하는 것을 중단하십시오.

버퍼(buffer)에도 상한선을 두십시오. 부분 파서(partial parser)는 반복 루프(repetition loop)에 빠진 모델로부터 메가바이트 단위의 데이터를 기꺼이 축적할 수 있으며, 버퍼에 크기 제한(size ceiling)을 두는 것이 이에 대한 가장 저렴한 보호책입니다.

한 번에 한 글자씩 테스트하기

작성할 가치가 있는 테스트는 가능한 최악의 입도(granularity)로 입력을 제공하는 테스트입니다. 왜냐하면 크기가 1인 청크(chunk)는 실제 토크나이저(tokenizer)가 생성할 수 있는 모든 경계 조건을 실행하기 때문입니다.

import json
import pytest
from partial_json import PartialJSONStream
...

두 번째 테스트가 중요한 테스트입니다. 이 테스트는 생성기(emitter)가 소비자(consumer)에게 하는 약속, 즉 확정된 값으로 전달된 값은 절대 수정되지 않는다는 약속을 인코딩합니다. 또한, 문자열 내부에 중괄호가 나타날 때만 발생하는 스캐너(scanner) 버그를 잡아내는 테스트이기도 합니다. 손상된 스냅샷이 이미 보고된 키를 변경해 버리기 때문입니다.

속성 기반 테스트(Property-based testing)는 이를 저렴하게 확장합니다. Hypothesis를 사용하여 임의의 중첩된 구조를 생성하고, 이를 직렬화(serialize)한 뒤, 모든 접두사(prefix)를 입력으로 제공하며 동일한 두 가지 불변성(invariants)을 확인합니다. 이스케이프 처리(escape-handling) 실수는 망가진 필드에 대한 고객 지원 티켓이 아니라, 축소된 반례(shrunk counterexample)로 나타납니다.

이것을 통해 얻는 것은 더 빠른 모델이 아닙니다. 결코 필요하지 않았던 대기 시간, 즉 값이 결정된 순간과 닫는 중괄호가 나타나 애플리케이션이 해당 값을 인지할 수 있게 되는 순간 사이의 간격을 제거하는 것입니다. 파서는 대략 100줄 정도이며, 파서가 유지하는 상태는 세 개의 변수와 스택(stack)입니다. 그리고 정당성 논거는 두 개의 테스트 안에 들어갑니다. 이는 답변이 작성되는 동안 사용자에게 답변을 보여주는 것에 대한 합리적인 대가입니다.

원문은 Dispatch에 처음 게시되었습니다.