agentic_json

Hey there. If you build AI applications, you probably know the pain of working with structured JSON outputs from Large Language Models.

Normally, AI programs ask the model for a JSON object and then just sit there. You have to wait for the entire JSON string to finish generating before you can call json.loads(), update your UI, or trigger background computations. If the model is generating a long response or a detailed conversational reply, your users are left staring at a loading spinner.

I built agentic_json to fix that.

This library provides a real-time stream parser designed specifically for LLM outputs. Instead of waiting for the full payload, it parses the JSON stream chunk by chunk. You can trigger events the exact millisecond a specific key-value pair completes, or even stream the text of a specific value as it generates.

Because LLMs can be unpredictable, the library also comes with built-in fail-safes. You define a schema with expected types and default values. If the AI hallucinates malformed content or outputs a string when you wanted an integer, agentic_json catches the mismatch and falls back to your safe default. This drastically reduces mismatch risks and keeps your application from crashing.

Key Features

• Real-Time Key Extraction: Fire callbacks the moment a key is fully generated, long before the rest of the JSON object finishes.

• Value Streaming: Hook into specific text fields to stream characters directly to your UI.

• Fail-Safe Schema: Define strict types and fallback defaults. If the AI breaks the JSON formatting, your app survives.

• Seamless Async Integration: Built from the ground up to work with modern Python async workflows.

Quick Start

Here is a look at how you can use agentic_json to build an interactive AI interviewer. In this example, we stream the AI's response and trigger different actions based on the keys being generated.

Install it using pip install agentic-json

import asyncio
from agentic_json import StreamEngine, Field
from ollama import AsyncClient

# Defining your strict schema with fail-safe defaults
schema = {
    "question_number": Field(type=int, default=0),
    "type": Field(type=str, default="General"),
    "end": Field(type=bool, default=False),
    "text": Field(type=str, default=""),
}

engine = StreamEngine(schema=schema)

# Set up your real-time callbacks

@engine.on_complete("question_number") # Triggers after a key-value has completed streaming the value
async def handle_ready(value):
    print(f"\n[SYSTEM] Moving to Question Number: {value}")

@engine.on_complete("type") 
async def handle_progress(value: str):
    print(f"[SYSTEM] Question Category: {value}")

@engine.on_stream("text") # Prints/Streams each character as they arrive
async def handle_description_stream(delta: str, current_raw: str):
  
    print(delta, end="", flush=True)

@engine.on_complete("end")
async def handle_user(value: bool):
    print(f"\n[EVENT] Interview complete status: {value}")

# Feed your LLM stream into the engine
async def stream_from_ollama():
    print("Starting AI stream...\n")
    
    messages = [
        {
            "role": "system",
            "content": "You are a job interviewer. Reply in strict JSON with keys: question_number, type, text, and end."
        }
    ]
    
    client = AsyncClient()
    stream = await client.chat(
        model="any model",
        messages=messages,
        stream=True,
    )
    
    async for chunk in stream:
        text_chunk = chunk["message"]["content"]
        await engine.consume(text_chunk) 
        
    print("\n\nStream finished. Final Validated State:", engine.state)

if __name__ == "__main__":
    asyncio.run(stream_from_ollama())

The Roadmap

I am actively developing agentic_json and continuously releasing better versions. This is still in the prototype phase but my goal is to make this library the absolute industry standard for handling streaming JSON in agentic workflows.

Feel free to open issues whenever anything breaks, submit pull requests, and help the world build some incredibly responsive AI apps faster.

License

Apache-2.0 license