Agent pipelines that depend on tool calls live or die by strict JSON contracts. When a model emits a truncated object, a stray comma, or a code fence, the whole run stalls. A self-healing JSON parser LLM layer intercepts those failures, applies cheap local fixes first, and only spends a model call when the syntax is beyond regex repair. This tutorial builds that layer from scratch and shows where it pays off in a real ReAct loop.
Prerequisites
- Python 3.10+ with
pip install openai - An OpenAI-compatible API endpoint and key. If you want the repair call to survive provider outages, point the client at a gateway that offers automatic fallback; an OpenAI-compatible endpoint such as n4n.ai forwards requests to 240+ models and reroutes on rate limits without code changes.
- Familiarity with
json.JSONDecodeErrorand basic string manipulation.
You should be able to run every snippet below in a single module.
Step 1: The naive parse and why it breaks
A tool-calling agent typically does this:
import json
def parse_tool_output(raw: str) -> dict:
return json.loads(raw)That throws on anything but pristine JSON. Common failure modes from real model outputs:
- Markdown fences:
```json - Trailing commas:
{"a": 1,} - Truncated streams:
{"name": "Alice", "roles": ["admin" - Prose wrappers:
Here is the result: {"ok": true}
Each of these aborts the agent step. We need a repair stage before the hard parse.
Step 2: Local heuristic repair
Local repair is free and fast. We strip fences, extract the first balanced brace block, and kill trailing commas. This covers 80% of cases I see in production logs.
import re
def local_repair(raw: str) -> str:
# Strip ```json ... ``` or ``` ... ```
cleaned = re.sub(r"^```(?:json)?\s*|\s*```$", "", raw.strip(), flags=re.MULTILINE)
start = cleaned.find("{")
if start == -1:
raise ValueError("No JSON object delimiter found")
depth = 0
end = -1
for i, ch in enumerate(cleaned[start:], start):
if ch == "{":
depth += 1
elif ch == "}":
depth -= 1
if depth == 0:
end = i
break
if end == -1:
# Truncated: close what's open
candidate = cleaned[start:] + "}" * depth
else:
candidate = cleaned[start:end + 1]
# Remove trailing commas before } or ]
candidate = re.sub(r",\s*([}\]])", r"\1", candidate)
return candidateCheckpoint output with a typical bad string:
bad = '```json\n{"name": "Alice", "roles": ["admin", "user",],}\n```'
print(local_repair(bad))
# {"name": "Alice", "roles": ["admin", "user"]}The fences are gone, the trailing comma inside the array and the one before the closing brace are removed. json.loads now succeeds.
Handling truncated streams
truncated = '{"query": "sales", "limit": 10'
print(local_repair(truncated))
# {"query": "sales", "limit": 10}The function appends one } because depth was 1. It will not invent values; if the truncation cut a string mid-word, you still get a decode error and escalate.
Extracting from prose
prose = 'Sure! Here is the payload: {"ok": true, "id": 42} Hope that helps'
print(local_repair(prose))
# {"ok": true, "id": 42}The first { anchors extraction; the depth counter stops at the matching }.
Step 3: Escalating to an LLM repair
When local repair still raises JSONDecodeError, we call a model. The prompt is tight: output only valid JSON. We run the response through local_repair again as a safety net.
from openai import OpenAI
client = OpenAI(base_url="https://api.n4n.ai/v1", api_key="YOUR_KEY")
def llm_repair(raw: str, client: OpenAI) -> dict:
resp = client.chat.completions.create(
model="gpt-4o-mini",
temperature=0,
messages=[
{"role": "system", "content": "You repair malformed JSON. Respond with only valid JSON, no commentary."},
{"role": "user", "content": f"Repair this into valid JSON:\n{raw}"}
],
)
text = resp.choices[0].message.content or ""
return json.loads(local_repair(text))Expected behavior on a deeply broken input:
broken = "{'name': 'Bob', 'age': 'thirty'}" # single quotes, string age
repaired = llm_repair(broken, client)
print(repaired)
# {'name': 'Bob', 'age': 'thirty'} (now double-quoted, valid)The LLM converts Python-style dict syntax to JSON. Local repair would have failed on single quotes.
Step 4: Composing the self-healing parser
Wrap both strategies in a class with a single parse entry point. Track which path succeeded for observability.
class SelfHealingJSONParser:
def __init__(self, client: OpenAI | None = None):
self.client = client
self.last_repair = "none"
def parse(self, raw: str) -> dict:
try:
return json.loads(raw)
except json.JSONDecodeError:
pass
try:
result = json.loads(local_repair(raw))
self.last_repair = "local"
return result
except json.JSONDecodeError:
pass
if self.client is None:
raise ValueError("Local repair failed and no LLM client configured")
self.last_repair = "llm"
return llm_repair(raw, self.client)Usage:
parser = SelfHealingJSONParser(client)
data = parser.parse(bad)
print(data, parser.last_repair)
# {'name': 'Alice', 'roles': ['admin', 'user']} localStep 5: Wiring into an agent pipeline
In a ReAct loop, tool output is often a raw model string. Swap the naive json.loads for the parser:
def handle_tool_response(tool_raw: str, parser: SelfHealingJSONParser):
try:
args = parser.parse(tool_raw)
except Exception as e:
# Surface to agent as a tool error, do not crash the run
return {"error": f"parse_failed: {e}"}
if "error" in args:
return args
# Continue with tool execution using validated args
return execute_tool(args)This keeps a single malformed response from killing the episode. The last_repair flag feeds your metrics: if llm repairs spike, your primary model’s JSON mode needs tuning.
Step 6: Guardrails and observability
LLM repair costs tokens. Cap it per parse and log counts:
import collections
repair_counts = collections.Counter()
def instrumented_parse(parser: SelfHealingJSONParser, raw: str) -> dict:
result = parser.parse(raw)
repair_counts[parser.last_repair] += 1
return result
# Later: print(repair_counts) -> Counter({'local': 142, 'llm': 3, 'none': 89})If you use a gateway with per-token usage metering, the repair calls show up as ordinary completions—no extra instrumentation needed. Set max_tokens low (e.g., 256) on the repair call; a well-formed JSON object for tool args rarely exceeds that.
Edge cases you must handle
- Strings containing braces:
{"code": "def f(): return {"}fools the naive depth scanner. Local repair will misbalance. In practice, escalate to LLM when the extracted block fails and contains colon-quote patterns outside brackets. - Numeric truncation:
{"temp": 22.is unrecoverable locally. LLM repair can infer22.0or22, but you should validate against a schema afterward. - Schema validation: Use
pydanticafter parse. Self-healing gets you valid JSON, not semantically correct JSON.
Step 7: Adding a schema check
from pydantic import BaseModel, ValidationError
class ToolArgs(BaseModel):
name: str
roles: list[str]
def parse_and_validate(raw: str, parser: SelfHealingJSONParser) -> ToolArgs:
data = parser.parse(raw)
try:
return ToolArgs(**data)
except ValidationError as e:
raise ValueError(f"Schema mismatch: {e}")Now the pipeline rejects {"name": "Alice"} (missing roles) even if the JSON is syntactically fine. That separation—syntax healing vs. schema enforcement—keeps the self-healing JSON parser LLM logic focused and your agent logic strict.
Closing notes for production
Run local repair synchronously in the hot path; it adds microseconds. Make the LLM repair asynchronous or behind a circuit breaker so a slow provider does not block the agent. The self-healing JSON parser LLM pattern is not a substitute for asking the model for JSON mode, but it is the difference between a retry storm and a quiet recovery when the context window clips the last brace.