Handling Partial JSON Streams
Waiting for an entire model response before parsing JSON feels sluggish in production applications. You don't want users staring at a blank screen. It's much better to render data into your user interface as tokens arrive from the provider. Language models output strings token by token. Standard JSON syntax requires a complete closing brace and matching quotes before a normal parser accepts it. If you try to run json.loads() on an incomplete chunk, Python throws a JSONDecodeError immediately. You can't just pass half a dictionary to standard library parsers. It won't work.
You need a strategy that handles partial payloads gracefully without breaking on half-finished keys. While you might be tempted to build custom regular expression routines to guess missing quotes and brackets, you'll end up writing a fragile parser that fails on edge cases. Instead, you can combine asynchronous Python generators with strict schema validation libraries to process streaming data safely. It's tricky. If you're also managing token economics across different endpoints, you might want to look at llm-token-counting-cost-control-in-laravel-and-nextjs to keep API expenses predictable. Don't skip cost monitoring when streaming large payloads.
Building a reliable consumer requires anticipating how network buffers and provider chunks arrive at the client application. A single TCP packet rarely contains an entire token or a logical phrase. Instead, chunks arrive split across arbitrary character boundaries. Your parser must accumulate these fragments in a local string buffer and only attempt structural inspection when a new token arrives. This design avoids CPU spikes caused by excessive parsing attempts on every single network read event. It also keeps your event loop free to handle other concurrent tasks.
Defining the Extraction Schema
Pydantic gives you a straightforward way to declare the exact shape of the data you expect back from the model. You define a base model with typed attributes. The validation layer enforces those constraints at runtime. Here's how you define a target schema for extracting user details from unstructured text input. It's clean and readable.
from pydantic import BaseModel, Field
class UserProfile(BaseModel):
username: str = Field(description="The handle of the user")
age: int = Field(description="User age in years")
skills: list[str] = Field(description="Programming languages known")
The model definition acts as the single source of truth for both your application logic and the prompt instructions sent to the model. You pass this schema definition to the provider's structured output API so the completion endpoint restricts its token sampling to valid keys and types. Don't rely on system prompts alone to enforce types. Models drift. They forget constraints when context windows grow large. Schema enforcement at the provider level changes everything. It guarantees that every single token adheres to your structural rules, reducing malformed responses to near zero.
When designing these schemas, keep your type annotations explicit. Avoid using generic dictionaries or arbitrary keyword arguments inside your Pydantic models because the underlying provider validation engines require concrete types to construct their internal context grammars. If you use optional fields, provide sensible defaults so the model doesn't stumble when a piece of data is missing from the source text. Every field constraint you add shrinks the search space for the model, yielding faster completions and cleaner outputs.
Streaming and Validating Chunks
Most major API clients support an asynchronous streaming mode that yields delta events as they leave the inference server. You iterate over the asynchronous generator, accumulate the raw text, and attempt partial validation against your Pydantic model at each step. Here's a complete script demonstrating how to consume the stream correctly. It uses the modern async client initialization pattern.
import asyncio
from openai import AsyncOpenAI
from pydantic import BaseModel, Field
client = AsyncOpenAI()
class UserProfile(BaseModel):
username: str
age: int
skills: list[str]
async def main():
stream = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "Extract: Jane Doe is 28 and writes Python and Rust."}],
response_format={"type": "json_object"},
stream=True,
)
accumulated_content = ""
async for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
accumulated_content += delta
print(delta, end="", flush=True)
asyncio.run(main())
The gotcha here is that raw delta printing only gives you raw text strings, not validated objects. To get typed models mid-stream, you need to catch parsing exceptions on intermediate string states until the JSON buffer becomes complete. For testing these integrations without hitting live rate limits or incurring costs on every test run, check out testing-llm-integration-without-flaky-ci-runs for mocking strategies. You'll thank yourself later when your test suite runs locally in seconds.
As the stream progresses, you can feed the accumulated string into a JSON repair utility or a partial parser before passing it to Pydantic. Because standard json loads will crash on an unclosed quote, you might want to inspect the string length and ensure matching brace counts before running validation. Once the final chunk arrives, your code converts the complete string into a fully typed instance of your Pydantic model, giving you safe, validated data ready for database insertion or downstream processing.
Managing Production Failure Modes
Models sometimes drop a trailing bracket or emit a trailing comma when hitting maximum token limits mid-sentence. When that happens, your final parse step will fail even if ninety percent of the object arrived intact. You should always wrap your final validation block in a try-except statement and log the raw accumulator for debugging purposes. Never trust an incoming stream to match your schema on the very first try without exception handling. Network drops also happen more often than you think during long streaming sessions. If the transmission severs halfway through a payload, your async iterator will raise an exception. Implement a retry wrapper around your generator call with exponential backoff to handle transient gateway errors cleanly. Don't let a dropped TCP connection crash your background workers.
Memory consumption is another quiet trap in long-running streaming loops. If you accumulate massive text responses without capping string lengths or checking payload bounds, an erratic model can leak memory inside your async worker processes. Set a strict ceiling on the total characters your buffer can accept before throwing an exception and aborting the stream. Monitoring these edge cases ensures your application stays resilient when third-party AI providers experience degraded performance or unexpected output drifts.
