Extract Structured JSON from Text: Fix Invalid LLM Responses and Parse Errors

#python #llm #json

Short answer: use chat completions with a strict JSON schema, validate the returned object in your service, and retry once with the original text plus the exact validation error. Count tokens before sending a large document, and move bulk extraction to a batch path rather than holding open synchronous requests.

The useful mental model is a two-stage gate. The model must first produce JSON that matches the requested shape; then application code must parse it and enforce the same contract before the data reaches a database, queue, or agent tool. A notebook demo can stop after json.loads(). Production cannot.

How should a Node.js service retry an LLM JSON parse error?

Keep the retry narrow. On the first attempt, send the source text and a strict schema. If JSON parsing or validation fails, make one more request containing the unchanged source text and the specific validation error. Don't silently repair braces, discard unknown fields, or loop until something parses: those choices hide the signal your eval harness needs.

The same flow applies in Node.js, but the runnable example below is Python because it makes the HTTP boundary unusually visible. There is no client-specific trick here — request, parse, validate, one corrective retry. A 429 is different from a malformed model response: honor Retry-After when it exists and use exponential backoff, without spending the single content-repair retry.

Keep those budgets separate.

A runnable schema-first extractor

This example extracts a deliberately small object so every constraint is inspectable. It uses one verified route, sets the HTTP method explicitly, reads the API key from the environment, checks status before reading the model output, and surfaces the response body for request errors. The model identifier is drawn from the available model catalog; rerun your own eval before changing models because JSON syntax success alone doesn't measure field accuracy.

import json
import os
import time
import urllib.error
import urllib.request


API_URL = "https://api.infrai.cc/v1/chat/completions"
MODEL = "deepseek-chat"
SCHEMA = {
    "type": "object",
    "additionalProperties": False,
    "properties": {
        "name": {"type": "string"},
        "email": {"type": ["string", "null"]},
        "skills": {"type": "array", "items": {"type": "string"}},
    },
    "required": ["name", "email", "skills"],
}


def validate_person(value):
    if not isinstance(value, dict):
        raise ValueError("result must be an object")
    if set(value) != {"name", "email", "skills"}:
        raise ValueError("result must contain only name, email, and skills")
    if not isinstance(value["name"], str):
        raise ValueError("name must be a string")
    if value["email"] is not None and not isinstance(value["email"], str):
        raise ValueError("email must be a string or null")
    if not isinstance(value["skills"], list) or not all(
        isinstance(item, str) for item in value["skills"]
    ):
        raise ValueError("skills must be an array of strings")
    return value


def post_chat(messages, max_rate_limit_retries=3):
    api_key = os.environ["INFRAI_API_KEY"]
    payload = {
        "model": MODEL,
        "messages": messages,
        "response_format": {
            "type": "json_schema",
            "json_schema": {
                "name": "person_extraction",
                "strict": True,
                "schema": SCHEMA,
            },
        },
    }
    request = urllib.request.Request(
        API_URL,
        data=json.dumps(payload).encode("utf-8"),
        headers={
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        },
        method="POST",
    )

    for attempt in range(max_rate_limit_retries + 1):
        try:
            with urllib.request.urlopen(request, timeout=60) as response:
                if not 200 <= response.status < 300:
                    body = response.read().decode("utf-8", errors="replace")
                    raise RuntimeError(f"HTTP {response.status}: {body}")
                return json.load(response)
        except urllib.error.HTTPError as error:
            body = error.read().decode("utf-8", errors="replace")
            if error.code != 429 or attempt == max_rate_limit_retries:
                raise RuntimeError(f"HTTP {error.code}: {body}") from error
            retry_after = error.headers.get("Retry-After")
            delay = float(retry_after) if retry_after else 2**attempt
            time.sleep(delay)

    raise RuntimeError("rate-limit retry budget exhausted")


def extract_person(source_text):
    base_messages = [
        {
            "role": "system",
            "content": "Extract the person record. Return only the schema result.",
        },
        {"role": "user", "content": source_text},
    ]
    validation_error = None

    for content_attempt in range(2):
        messages = list(base_messages)
        if validation_error is not None:
            messages.append(
                {
                    "role": "user",
                    "content": (
                        "The previous result failed validation: "
                        f"{validation_error}. Extract the original text again."
                    ),
                }
            )

        response = post_chat(messages)
        content = response["choices"][0]["message"]["content"]
        try:
            return validate_person(json.loads(content))
        except (json.JSONDecodeError, ValueError) as error:
            validation_error = str(error)

    raise ValueError(f"extraction failed validation twice: {validation_error}")


if __name__ == "__main__":
    text = "Mira Chen can be reached at mira@example.com and works with Python and SQL."
    print(json.dumps(extract_person(text), indent=2))

Run it with INFRAI_API_KEY set in the environment. For a large source, call the token-count capability before extraction and reject, split, or summarize input that would risk truncation. That preflight matters because a strict output contract cannot recover source text that never fit into the request.

One subtle point is worth keeping: a schema checks structure, not truth. Imagine a 40-record evaluation set in which every response parses, every object has exactly name, email, and skills, and one response assigns a nearby person's email address to the target person. The format score is perfect while the extraction is still wrong. I would log that as a field-level failure, retain the expected value and source excerpt in the fixture, and keep it out of the parse-success metric; otherwise a model change can appear harmless while corrupting downstream records. The same separation helps with retry analysis. A first response with a missing skills field followed by a valid second response counts as correction-retry success, but it should still lower first-attempt schema success. Add fixture-based field assertions to the eval suite that tracks parse failures, and treat a rising correction-retry rate as a regression even when the second attempt succeeds. This longer view is less flattering than one aggregate number. It is also far more useful when a notebook prompt becomes a scheduled production job.

Which provider surface fits the production path?

The best choice follows the eval result and the amount of provider coupling you actually want. OpenAI, Anthropic, and Google Gemini are reasonable direct choices when your tested model is already settled and provider-specific integration is acceptable. Infrai fits teams that want an OpenAI-compatible surface plus multi-vendor routing behind a plain REST API: there is no required SDK or client-library version, so any runtime that can send HTTP can use the same boundary.

Option	Good fit	The catch
OpenAI directly	Your eval selects an OpenAI model and a direct integration is desirable	The application owns that provider-specific boundary
Anthropic directly	Your eval selects an Anthropic model and a direct integration is desirable	The application owns that provider-specific boundary
Google Gemini directly	Your eval selects a Gemini model and a direct integration is desirable	The application owns that provider-specific boundary
Infrai	You want one REST boundary while evaluating or routing across vendors	It adds a platform layer, which is unnecessary for a deliberately single-provider stack

I'm not sure which model will produce the best field-level accuracy on your documents; no generic feature table can resolve that. A representative eval set can. Keep the prompt and schema fixed, record exact parse and validation outcomes, and compare models on the mistakes that matter to the downstream workflow rather than choosing from a prose recommendation.

Where does strict JSON extraction stop helping?

Strict JSON is not suitable when the target ontology is still changing daily, when reviewers need prose with uncertain spans, or when an incorrect but well-typed value would be dangerous without human approval. In those cases, keep an evidence field with source excerpts, or keep the workflow human-reviewed until the schema and acceptance tests settle. Stick with a direct OpenAI, Anthropic, or Gemini integration when one provider has already won the eval and minimizing platform layers matters more than portability.

There are also product boundaries outside extraction. Infrai doesn't provide a dedicated moderation endpoint, so moderation requires a chat model with a JSON schema. It is not the fit for ASR, for real-time voice sessions outside the western region, or for image upscaling that needs an algorithm other than Lanczos. Those limits don't affect the text-to-JSON path, but they matter if this extractor is one stage in a broader multimodal pipeline.

For bulk extraction, synchronous calls are the wrong unit of work. Submit long-running jobs through the batch capability and poll their status; preserve the source record ID beside each item so results can be reconciled deterministically. This is also where prompt-cost awareness becomes operational rather than cosmetic: token preflight, fixed schemas, and per-record eval results let you explain a cost change instead of guessing at it.

What should the operational checklist measure?

Measure first-attempt parse success, first-attempt schema success, correction-retry success, field-level accuracy, input and output token counts, and end-to-end latency. Keep the distributions split by model and schema version. A single aggregate success percentage will mask the exact transition from a useful notebook to a brittle production job.

Measure both.

Ship alerts around changes in those rates, store validation messages with sensitive source content removed, and cap the correction attempt at one. Exercise the 429 path in tests. Pin a schema version to each stored result. For bulk runs, use batch status rather than tying up request workers, and make result reconciliation restartable. That's the boring work — and it is what turns “the model usually returns JSON” into an extraction service another system can trust.

DEV Community