Parsing an LLM's JSON Before the Last Token Arrives

A model that reliably returns valid JSON does not return valid JSON while it is typing. Those are two different guarantees, and most streaming clients quietly assume they are the same one. Constrained decoding and schema-enforced output modes promise that the finished response parses. They promise nothing about the seventeenth chunk, which might end in the middle of a key, inside an escape sequence, or right after a comma that has no element behind it yet.

So the usual compromise is to stream tokens for a progress spinner and buffer them all anyway, then parse once at the end. The user watches characters appear that the application itself refuses to read. Everything the model has already decided — the title, the first three list items, the classification label that arrived in the first fifty tokens — sits unusable in a string buffer until the closing brace lands.

The gap is closable. What follows builds a partial JSON reader that turns every chunk into the best complete value the stream can currently justify, then shows the two places where the obvious version of that idea silently corrupts data.

What a Streaming Model Actually Sends You

A streaming completion arrives as deltas of text with no relationship to JSON structure. Token boundaries follow the tokenizer's vocabulary, not the grammar. A single delta can carry {"ti, or tle": "Sh, or a lone backslash that only becomes meaningful when the next delta supplies the n.

The client loop is the easy part:

import json
from openai import OpenAI

client = OpenAI()

def stream_text(prompt: str):
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"},
        stream=True,
    )
    for chunk in response:
        delta = chunk.choices[0].delta.content
        if delta:
            yield delta

Every element of that generator is a fragment of a document that is not yet a document. The question is what a consumer can do with each one besides append it.

Why json.loads Is the Wrong Tool Mid-Stream

The standard library parser is all-or-nothing by design. Hand it a prefix and it raises, because a prefix of a JSON document is not a JSON document.

buffer = ""
for delta in stream_text(prompt):
    buffer += delta
    try:
        value = json.loads(buffer)   # raises on every chunk but the last
    except json.JSONDecodeError:
        continue

json.JSONDecoder.raw_decode looks like an escape hatch, and it is a useful one, but for a different problem: it decodes one complete value from the front of a string and reports where it stopped. That solves concatenated documents in a stream of separate objects. It does not help when the single object you want is truncated, because there is still no complete value at the front to decode.

The ijson package is closer. It is an event-driven parser that yields start_map, map_key, string, and end_map events as bytes arrive, which is exactly the streaming shape you want. Its constraint is that it is built for very large but eventually complete documents; a truncated feed ends in IncompleteJSONError, and you get events only for values that have already closed. For a rendering UI that wants to show a half-written string as it grows, event-per-completed-value is one step too coarse.

Writing a real incremental parser is the thorough answer, and it is more work than the problem deserves. A tokenizer, a state machine over the grammar, and a value builder that can be interrogated halfway through is a few hundred lines to write and considerably more to trust. The remaining approach is cheaper and reuses a parser that is already correct: repair the prefix into a valid document, parse the repair, and throw the repair away. The buffer itself is never modified, so a mistake in the repair logic costs one bad snapshot rather than a poisoned stream.

Closing What You Have So Far

The core idea is small. Track which containers are open, and when a snapshot is requested, append the closers that would finish them.

def naive_snapshot(buffer: str):
    stack = []
    for ch in buffer:
        if ch == "{":
            stack.append("}")
        elif ch == "[":
            stack.append("]")
        elif ch in "}]" and stack:
            stack.pop()
    try:
        return json.loads(buffer + "".join(reversed(stack)))
    except json.JSONDecodeError:
        return None

For a well-behaved prefix this works immediately. {"title": "Ship it", "tags": ["python" becomes {"title": "Ship it", "tags": ["python"]} and parses into a dictionary a template can render right now, several hundred tokens before the response ends.

It also fails constantly, and the failures are more interesting than the successes. {"tags": ["python", closes into a trailing comma. {"score": closes into a key with no value. {"score": 1. closes into a number literal that ends in a decimal point. Each of these is a decode error, so the snapshot comes back as None and the UI stalls until the stream happens to land on a lucky boundary.

Strings, Escapes, and the Repairs That Corrupt Data

The dangerous failure is different from the noisy ones. The noisy ones announce themselves: a trailing comma raises, the snapshot is None, and the next chunk usually fixes it. A brace inside a string value is not a brace, and nothing raises.

Consider the prefix {"note": "use {curly} braces. The scan above counts the { in the prose, pushes a second } onto the stack, never sees a matching close, and produces {"note": "use {curly} braces}}. That is not a decode error. Depending on where the stream stopped, a repair like this can produce a document that parses into the wrong shape — structure invented out of the model's prose.

This is not a rare input either. Any model writing about code, file paths, or template syntax emits braces and brackets inside string values constantly, and a JSON response about JSON is the worst case for a naive scanner. A snapshot that parses into the wrong shape is worse than one that fails to parse, because a consumer has no signal that anything went wrong.

Any scanner that does not know whether it is inside a string is guessing. The same applies to escapes: a buffer ending in a single \ is a partial escape sequence, and closing the string with " turns that trailing backslash into an escaped quote, which swallows the closer and pushes the corruption one level further out.

Correct handling needs three pieces of state carried between chunks: the container stack, an in-string flag, and an escape flag.

A Scanner That Remembers Where It Is

Each character is classified once, when it arrives, and never re-scanned. Every frame also records a safe truncation offset — the point at which the container held only complete elements — so a repair that fails can retreat to the last known-good boundary instead of returning nothing.

import json
from dataclasses import dataclass


@dataclass
class _Frame:
    close: str
    cut: int  # offset where this container held only finished elements


class PartialJSONStream:
    def __init__(self) -> None:
        self._buf: list[str] = []
        self._len = 0
        self._stack: list[_Frame] = []
        self._in_string = False
        self._escaped = False

    def feed(self, chunk: str) -> None:
        for ch in chunk:
            self._step(ch)
            self._buf.append(ch)
            self._len += 1

    def _step(self, ch: str) -> None:
        i = self._len
        if self._in_string:
            if self._escaped:
                self._escaped = False
            elif ch == "\\":
                self._escaped = True
            elif ch == '"':
                self._in_string = False
            return
        if ch == '"':
            self._in_string = True
        elif ch == "{":
            self._stack.append(_Frame("}", i + 1))
        elif ch == "[":
            self._stack.append(_Frame("]", i + 1))
        elif ch in "}]":
            if self._stack:
                self._stack.pop()
        elif ch == "," and self._stack:
            self._stack[-1].cut = i

The snapshot method generates repair candidates from most complete to most conservative and returns the first one that parses:

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

    def _candidates(self, text: str):
        head = text[:-1] if self._escaped else text
        if self._in_string:
            head += '"'
        head = head.rstrip()
        if head.endswith(","):
            head = head[:-1].rstrip()
        elif head.endswith(":"):
            head += " null"
        yield head + self._closers()
        for depth in range(len(self._stack) - 1, -1, -1):
            cut = self._stack[depth].cut
            yield text[:cut].rstrip().rstrip(",") + self._closers(depth + 1)

    def snapshot(self):
        text = "".join(self._buf)
        for candidate in self._candidates(text):
            try:
                return json.loads(candidate)
            except json.JSONDecodeError:
                continue
        return None

The fallback path is what makes partial numbers and half-typed literals harmless. A buffer ending in 1. or tru produces a first candidate that fails, then a truncated candidate that drops the unfinished element entirely and parses. The caller sees the object without that key rather than seeing nothing at all.

One property of this design has to be stated plainly to whoever consumes it: a string value in a snapshot may be a prefix of the real value. Rendering a partial description as it grows is the intended use. Comparing a partial value against an enum, treating it as a URL, or passing it to something that performs an action is a bug waiting for a slow token.

Turning Snapshots Into Field Events

Polling a whole snapshot per chunk is wasteful for consumers that only care when a field is done. JSON objects stream in order, which gives a simple and reliable completion rule: once a second key appears, the first key's value can no longer change. Every key except the last one in the snapshot is settled.

class FieldEmitter:
    def __init__(self) -> None:
        self.stream = PartialJSONStream()
        self._emitted: set[str] = set()

    def feed(self, chunk: str) -> list[tuple[str, object]]:
        self.stream.feed(chunk)
        snap = self.stream.snapshot()
        if not isinstance(snap, dict):
            return []
        events = []
        for key in list(snap)[:-1]:
            if key not in self._emitted:
                self._emitted.add(key)
                events.append((key, snap[key]))
        return events

    def finish(self) -> list[tuple[str, object]]:
        snap = self.stream.snapshot()
        if not isinstance(snap, dict):
            return []
        return [(k, v) for k, v in snap.items() if k not in self._emitted]

A downstream handler can now start work on classification while the model is still writing explanation, which is the whole point of streaming a structured response rather than a blob of prose. A routing decision can be dispatched, a database row can be reserved, a UI section can render as final rather than as a skeleton.

The ordering rule has one condition attached: it holds for objects the model writes in a fixed key order, which is what schema-constrained decoding produces. It does not hold for arrays of objects where a later element revises nothing but the last element is still growing, so treat the final element of a list as provisional in exactly the same way as the final key of an object.

Validation, Cancellation, and Output That Never Recovers

A snapshot is a draft, so validating it against the strict output model is wrong by construction — required fields are missing on purpose. Build a relaxed mirror of the model for snapshots and keep the strict one for the final value.

from pydantic import BaseModel, create_model


class Review(BaseModel):
    verdict: str
    reasons: list[str]
    score: int


def draft_of(model: type[BaseModel]) -> type[BaseModel]:
    fields = {
        name: (info.annotation | None, None)
        for name, info in model.model_fields.items()
    }
    return create_model(f"Draft{model.__name__}", **fields)


DraftReview = draft_of(Review)

Two failure modes need explicit handling around this. The first is a model that stops producing structure and starts producing an apology or a fenced code block; the snapshot goes None and stays None, so track consecutive unparseable chunks and abandon the stream rather than waiting for a close that is not coming. The second is early exit: when a snapshot already satisfies the strict model and the remaining fields are optional, cancelling the request stops paying for tokens nobody will read.

Cap the buffer as well. A partial parser will happily accumulate megabytes from a model stuck in a repetition loop, and a size ceiling on the buffer is the cheapest protection against that.

Testing One Character at a Time

The tests worth writing are the ones that feed input at the worst possible granularity, because a chunk size of one exercises every boundary a real tokenizer could ever produce.

import json
import pytest
from partial_json import PartialJSONStream

PAYLOADS = [
    {"verdict": "ship", "reasons": ["tests pass", "small diff"], "score": 8},
    {"note": 'use {curly} braces and a "quote"', "path": "C:\\tmp\\out.json"},
    {"nested": {"a": [1, 2, {"b": None}], "c": True}, "trailing": 1.5},
]


@pytest.mark.parametrize("payload", PAYLOADS)
def test_every_prefix_parses_or_declines(payload):
    text = json.dumps(payload)
    stream = PartialJSONStream()
    for ch in text:
        stream.feed(ch)
        snap = stream.snapshot()
        assert snap is None or isinstance(snap, (dict, list))
    assert stream.snapshot() == payload


@pytest.mark.parametrize("payload", PAYLOADS)
def test_settled_keys_never_change(payload):
    text = json.dumps(payload)
    stream = PartialJSONStream()
    seen: dict[str, object] = {}
    for ch in text:
        stream.feed(ch)
        snap = stream.snapshot()
        if isinstance(snap, dict):
            for key in list(snap)[:-1]:
                if key in seen:
                    assert seen[key] == snap[key]
                else:
                    seen[key] = snap[key]

The second test is the important one. It encodes the promise the emitter makes to its consumers — that a value handed out as settled is never revised — and it is the test that catches a scanner bug that only shows up when a brace appears inside a string, because the corrupted snapshot changes a key that had already been reported.

Property-based testing extends this cheaply. Generate arbitrary nested structures with Hypothesis, serialise them, feed every prefix, and assert the same two invariants. Any escape-handling mistake surfaces as a shrunk counterexample rather than as a support ticket about a mangled field.

What this buys is not a faster model. It is the removal of a wait that was never necessary — the interval between the moment a value is decided and the moment the closing brace lets the application admit it knows. The parser is roughly a hundred lines, the state it carries is three variables and a stack, and the correctness argument fits in two tests. That is a reasonable price for showing users an answer while it is still being written.

---

Originally published on Dispatch.