Skip to content

lpgcast.llm.jsonparse

Tolerant parsing of a model's JSON reply.

Repairs only what a model is known to break without changing meaning — escaped delimiting quotes, // comments, trailing commas, a fenced block, closing brackets of the wrong kind — then parses strictly. Anything else fails loudly, with the decoder's position and the text around it, so a re-ask can show the model exactly where its reply broke.

Attributes

FEEDBACK_AFTER_ERROR = 200 module-attribute

FEEDBACK_BEFORE_ERROR = 600 module-attribute

FEEDBACK_HEAD = 800 module-attribute

FEEDBACK_TAIL = 1200 module-attribute

JSON_ERROR_WINDOW = 150 module-attribute

Classes

LLMJsonParseError

Bases: ValueError

A reply that is not a JSON object, with the decoder's message and position.

Source code in lpgcast/llm/jsonparse.py
class LLMJsonParseError(ValueError):
    """A reply that is not a JSON object, with the decoder's message and position."""

    def __init__(self, message: str, *, msg: str, pos: int, doc: str) -> None:
        super().__init__(message)
        self.msg = msg
        self.pos = pos
        self.doc = doc

Attributes

doc = doc instance-attribute
msg = msg instance-attribute
pos = pos instance-attribute

Methods:

__init__(message, *, msg, pos, doc)
Source code in lpgcast/llm/jsonparse.py
def __init__(self, message: str, *, msg: str, pos: int, doc: str) -> None:
    super().__init__(message)
    self.msg = msg
    self.pos = pos
    self.doc = doc

Functions:

feedback_excerpt(content, error)

A bounded slice of a failed reply to show the model when re-asking.

Source code in lpgcast/llm/jsonparse.py
def feedback_excerpt(content: str, error: Exception) -> str:
    """A bounded slice of a failed reply to show the model when re-asking."""
    if isinstance(error, LLMJsonParseError) and error.pos:
        doc = error.doc or content
        start = max(0, error.pos - FEEDBACK_BEFORE_ERROR)
        end = min(len(doc), error.pos + FEEDBACK_AFTER_ERROR)
        prefix = "..." if start else ""
        suffix = "..." if end < len(doc) else ""
        return f"{prefix}{doc[start:end]}{suffix}"

    if len(content) <= FEEDBACK_HEAD + FEEDBACK_TAIL:
        return content
    elided = len(content) - FEEDBACK_HEAD - FEEDBACK_TAIL
    return (
        f"{content[:FEEDBACK_HEAD]}\n... [{elided} characters elided] ...\n"
        f"{content[-FEEDBACK_TAIL:]}"
    )

parse_json_object(text)

Parse a reply as one JSON object.

Strict parse first; then the first fenced block; then the bracket-kind repair. Raw control characters inside strings are tolerated.

Returns:

Type Description
tuple[dict[str, Any], int]

The object, and the number of brackets repaired (0 if none).

Raises:

Type Description
LLMJsonParseError

with the strict error's position and context.

Source code in lpgcast/llm/jsonparse.py
def parse_json_object(text: str) -> tuple[dict[str, Any], int]:
    """Parse a reply as one JSON object.

    Strict parse first; then the first fenced block; then the bracket-kind
    repair. Raw control characters inside strings are tolerated.

    Returns:
        The object, and the number of brackets repaired (0 if none).

    Raises:
        LLMJsonParseError: with the strict error's position and context.
    """
    text = sanitize_json_text(text)
    try:
        return _loads_object(text), 0
    except json.JSONDecodeError as strict_error:
        reported = strict_error
        match = _JSON_FENCE_RE.search(text)
        if match and match.group(1):
            fenced = match.group(1)
            try:
                return _loads_object(fenced), 0
            except json.JSONDecodeError as fence_error:
                text, reported = fenced, fence_error

        repaired, fixes = repair_bracket_kinds(text)
        if fixes:
            try:
                return _loads_object(repaired), fixes
            except json.JSONDecodeError:
                pass

        raise LLMJsonParseError(
            _format_json_error(text, reported),
            msg=reported.msg,
            pos=reported.pos,
            doc=text,
        ) from strict_error

repair_bracket_kinds(text)

Rewrite each closing bracket to the kind its opener demands.

Only substitutes characters; never inserts, deletes or reorders. Gives up (returns the text unchanged and 0) on a closer with no open frame or on frames left open at the end — that is truncation, and closing it would invent a payload.

Returns:

Type Description
tuple[str, int]

The repaired text and the number of characters substituted.

Source code in lpgcast/llm/jsonparse.py
def repair_bracket_kinds(text: str) -> tuple[str, int]:
    """Rewrite each closing bracket to the kind its opener demands.

    Only substitutes characters; never inserts, deletes or reorders. Gives up
    (returns the text unchanged and 0) on a closer with no open frame or on
    frames left open at the end — that is truncation, and closing it would
    invent a payload.

    Returns:
        The repaired text and the number of characters substituted.
    """
    out = list(text)
    stack: list[str] = []
    fixes = 0
    in_string = False
    i = 0
    n = len(text)

    while i < n:
        ch = text[i]
        if in_string:
            if ch == "\\":
                i += 2
                continue
            if ch == '"':
                in_string = False
            i += 1
            continue

        if ch == '"':
            in_string = True
        elif ch in "{[":
            stack.append(ch)
        elif ch in "}]":
            if not stack:
                return text, 0
            wanted = "}" if stack.pop() == "{" else "]"
            if wanted != ch:
                out[i] = wanted
                fixes += 1
        i += 1

    if stack:
        return text, 0
    return "".join(out), fixes

sanitize_json_text(text)

Apply the meaning-preserving text repairs, in order.

Source code in lpgcast/llm/jsonparse.py
def sanitize_json_text(text: str) -> str:
    """Apply the meaning-preserving text repairs, in order."""
    return strip_trailing_commas(strip_json_comments(unescape_json_delimiters(text)))

strip_json_comments(text)

Remove // line comments outside string literals.

Source code in lpgcast/llm/jsonparse.py
def strip_json_comments(text: str) -> str:
    """Remove ``//`` line comments outside string literals."""

    def _replace(match: re.Match[str]) -> str:
        matched = match.group()
        return matched if matched.startswith('"') else ""

    return _JSON_COMMENT_RE.sub(_replace, text)

strip_trailing_commas(text)

Remove commas directly before } or ].

Source code in lpgcast/llm/jsonparse.py
def strip_trailing_commas(text: str) -> str:
    """Remove commas directly before ``}`` or ``]``."""
    previous = None
    while previous != text:
        previous = text
        text = _TRAILING_COMMA_RE.sub(r"\1", text)
    return text

unescape_json_delimiters(text)

Repair strings whose delimiting quotes were escaped ("k": \"v\",).

A \" where a value must start opens a string; inside such a string a \" followed by a delimiter closes it. Escapes inside normally delimited strings are left alone. Outside strings, \n, \t and \r become the whitespace they denote.

Source code in lpgcast/llm/jsonparse.py
def unescape_json_delimiters(text: str) -> str:
    """Repair strings whose *delimiting* quotes were escaped (``"k": \\"v\\",``).

    A ``\\"`` where a value must start opens a string; inside such a string a
    ``\\"`` followed by a delimiter closes it. Escapes inside normally
    delimited strings are left alone. Outside strings, ``\\n``, ``\\t`` and
    ``\\r`` become the whitespace they denote.
    """
    out: list[str] = []
    i = 0
    n = len(text)
    in_string = False
    repaired_open = False
    last_significant = ""

    while i < n:
        ch = text[i]
        if in_string:
            if ch == "\\" and i + 1 < n:
                nxt = text[i + 1]
                if nxt == '"' and repaired_open:
                    j = i + 2
                    while j < n and text[j] in " \t\r\n":
                        j += 1
                    if j >= n or text[j] in ",}]:":
                        out.append('"')
                        i += 2
                        in_string = False
                        repaired_open = False
                        last_significant = '"'
                        continue
                out.append(ch)
                out.append(nxt)
                i += 2
                continue
            out.append(ch)
            if ch == '"':
                in_string = False
                repaired_open = False
                last_significant = '"'
            i += 1
            continue

        if (
            ch == "\\"
            and i + 1 < n
            and text[i + 1] == '"'
            and last_significant in (":", ",", "{", "[", "")
        ):
            out.append('"')
            i += 2
            in_string = True
            repaired_open = True
            continue

        if ch == "\\" and i + 1 < n and text[i + 1] in "nrt":
            out.append({"n": "\n", "r": "\r", "t": "\t"}[text[i + 1]])
            i += 2
            continue

        out.append(ch)
        if ch == '"':
            in_string = True
            repaired_open = False
        if not ch.isspace():
            last_significant = ch
        i += 1

    return "".join(out)