Streaming structured output in LangChain lets you parse and validate model responses token by token instead of waiting for the full completion. This langchain streaming structured output tutorial covers the two main approaches — PydanticOutputParser for strict schemas and JsonOutputParser for flexible JSON — plus the production patterns you need when running at scale.
Step 1: Install dependencies and configure the client
Start with a clean environment. You need LangChain core, the OpenAI client (or any OpenAI-compatible endpoint), and Pydantic v2.
pip install langchain-core langchain-openai pydantic python-dotenvCreate a .env file with your API base and key. If you’re using n4n.ai, the base URL is https://api.n4n.ai/v1 and the key comes from your dashboard. The endpoint speaks the OpenAI wire format, so the standard client works without modification.
# .env
OPENAI_API_BASE=https://api.n4n.ai/v1
OPENAI_API_KEY=sk-your-key-hereLoad it in your script:
# config.py
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
load_dotenv()
llm = ChatOpenAI(
model="gpt-4o-mini",
temperature=0,
streaming=True,
max_tokens=2048,
)Verify: Run python -c "from config import llm; print(llm.model_name)" — it should print gpt-4o-mini without errors.
Step 2: Define your schema with Pydantic
Pydantic v2 models are the contract between your code and the model. Use Field descriptions liberally — they become part of the system prompt and materially improve adherence.
# schemas.py
from pydantic import BaseModel, Field, field_validator
from typing import Literal, Optional
from datetime import datetime
class LineItem(BaseModel):
sku: str = Field(description="Product SKU, e.g. 'SKU-1234'")
quantity: int = Field(ge=1, description="Units ordered")
unit_price_cents: int = Field(ge=0, description="Price per unit in cents")
class OrderExtraction(BaseModel):
order_id: str = Field(description="Alphanumeric order reference")
customer_email: str = Field(pattern=r"^[^@]+@[^@]+\.[^@]+$")
currency: Literal["USD", "EUR", "GBP"] = "USD"
items: list[LineItem] = Field(min_length=1)
placed_at: datetime
notes: Optional[str] = Field(default=None, max_length=500)
@field_validator("order_id")
@classmethod
def uppercase_order_id(cls, v: str) -> str:
return v.upper()The validator runs after parsing, so streaming tokens that temporarily violate it (lowercase order ID mid-stream) won’t crash the parser — only the final validated object must pass.
Step 3: Stream with PydanticOutputParser
PydanticOutputParser wraps your model and handles the prompt formatting. The key is stream() on the chain, not invoke().
# stream_pydantic.py
from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.prompts import PromptTemplate
from config import llm
from schemas import OrderExtraction
parser = PydanticOutputParser(pydantic_object=OrderExtraction)
prompt = PromptTemplate(
template=(
"Extract the order details from the following text.\n"
"{format_instructions}\n\n"
"Text: {input_text}"
),
input_variables=["input_text"],
partial_variables={"format_instructions": parser.get_format_instructions()},
)
chain = prompt | llm | parser
sample_text = """
Order ORD-8842 placed by sarah.chen@example.com on 2024-03-15T14:30:00Z.
Items: 2x SKU-A100 at $29.99 each, 1x SKU-B200 at $49.50.
Customer noted: 'Please use sustainable packaging.'
"""
for chunk in chain.stream({"input_text": sample_text}):
# chunk is a *partial* OrderExtraction model
print(chunk.model_dump_json(indent=2))
print("---")What you’ll see: Each iteration prints a progressively more complete OrderExtraction. Early chunks may have None for optional fields or empty lists for items. The parser yields a valid Pydantic instance at every step — it never yields raw JSON strings.
Verify: Run the script. The final chunk should match the sample text exactly, with order_id uppercased by the validator and unit_price_cents as integers (2999, 4950).
Step 4: Stream with JsonOutputParser for flexible schemas
When your schema varies per request or you want to avoid Pydantic overhead, JsonOutputParser streams raw dicts. You lose automatic validation but gain flexibility.
# stream_json.py
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import PromptTemplate
from config import llm
parser = JsonOutputParser()
prompt = PromptTemplate(
template=(
"Return a JSON object with keys: "
"event_type (string), timestamp (ISO8601), payload (object).\n"
"{format_instructions}\n\n"
"Log line: {log_line}"
),
input_variables=["log_line"],
partial_variables={"format_instructions": parser.get_format_instructions()},
)
chain = prompt | llm | parser
log_line = '2024-03-15T14:30:00Z INFO payment_processed {"order_id": "ORD-8842", "amount_cents": 10948, "method": "card"}'
for chunk in chain.stream({"log_line": log_line}):
# chunk is a dict, possibly incomplete
print(chunk)
print("---")Key difference: JsonOutputParser yields dict objects. Missing keys simply don’t appear in early chunks. You must handle KeyError or use .get() when consuming partial results.
Verify: The final chunk should have all three keys with correct types. payload should be a nested dict, not a string.
Step 5: Handle partial parses in your application
Streaming structured output shines when you update UI incrementally or feed downstream consumers before the full response arrives. But partial models require defensive code.
# consumer.py
from typing import Generator
from stream_pydantic import chain # from Step 3
def render_order_stream(input_text: str) -> Generator[dict, None, None]:
"""
Yields UI-ready dicts. Never raises on partial data.
"""
for partial in chain.stream({"input_text": input_text}):
data = partial.model_dump()
# Coerce None/empty to sensible defaults for UI
ui_data = {
"order_id": data.get("order_id") or "—",
"customer_email": data.get("customer_email") or "—",
"currency": data.get("currency") or "USD",
"items": data.get("items") or [],
"subtotal_cents": sum(
i.get("quantity", 0) * i.get("unit_price_cents", 0)
for i in data.get("items") or []
),
"placed_at": data.get("placed_at"),
"notes": data.get("notes"),
"is_complete": all(
k in data and data[k] not in (None, [], "")
for k in ("order_id", "customer_email", "items", "placed_at")
),
}
yield ui_dataThe is_complete flag lets your frontend show a loading skeleton until the required fields populate. Note that items being an empty list [] is valid mid-stream — it becomes non-empty only when the model emits the first item.
Verify: Call list(render_order_stream(sample_text))[-1]["is_complete"] — should be True.
Step 6: Add retry and fallback for production
Streaming doesn’t eliminate provider failures. Wrap the chain with a retry policy and, if you’re using a gateway that supports it, automatic fallback to a healthy model.
# resilient_chain.py
from langchain_core.runnables import RunnableLambda, RunnableConfig
from langchain_core.runnables.utils import Input, Output
from config import llm
from stream_pydantic import prompt, parser
from tenacity import retry, stop_after_attempt, wait_exponential_jitter
base_chain = prompt | llm | parser
@retry(
wait=wait_exponential_jitter(initial=1, max=10),
stop=stop_after_attempt(3),
reraise=True,
)
def _stream_with_retry(input_: dict, config: RunnableConfig) -> Output:
# The gateway (n4n.ai) honors the model parameter and will
# automatically fail over to a healthy provider if the primary
# is rate-limited or degraded. No client-side model list needed.
return base_chain.stream(input_, config=config)
resilient_chain = RunnableLambda(_stream_with_retry)The @retry decorator handles transient network errors. The gateway handles provider-level failures — you send one request to one endpoint, and the gateway routes to whichever of the 240+ models is healthy. Your code stays unchanged.
Verify: Simulate a failure by temporarily setting an invalid API key. The retry logic should attempt three times with exponential backoff before raising.
Step 7: Meter usage per request
If you bill customers or enforce quotas, you need token counts per stream. LangChain exposes usage metadata via RunnableConfig callbacks.
# metered_chain.py
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.runnables import RunnableConfig
from resilient_chain import resilient_chain
from typing import Any
class TokenMeter(BaseCallbackHandler):
def __init__(self):
self.prompt_tokens = 0
self.completion_tokens = 0
self.total_tokens = 0
def on_llm_end(self, response: Any, **kwargs: Any) -> None:
usage = getattr(response, "llm_output", {}).get("token_usage", {})
self.prompt_tokens = usage.get("prompt_tokens", 0)
self.completion_tokens = usage.get("completion_tokens", 0)
self.total_tokens = usage.get("total_tokens", 0)
def stream_metered(input_text: str) -> tuple[Generator, TokenMeter]:
meter = TokenMeter()
config = RunnableConfig(callbacks=[meter])
gen = resilient_chain.stream({"input_text": input_text}, config=config)
return gen, meterAfter the stream exhausts, meter.total_tokens holds the billable count. The gateway also returns per-token usage in the response headers if you call the REST API directly — useful for audit logs.
Verify: Run a request, then assert meter.total_tokens > 0.
Step 8: Test with a CI-friendly harness
Add a pytest fixture that exercises the full pipeline without hitting the network. Use a fake LLM that yields predetermined chunks.
# test_streaming.py
import pytest
from langchain_core.language_models import BaseChatModel
from langchain_core.messages import AIMessageChunk
from langchain_core.outputs import ChatGenerationChunk
from schemas import OrderExtraction
from stream_pydantic import prompt, parser
class FakeStreamingLLM(BaseChatModel):
chunks: list[str]
def _stream(self, *args, **kwargs):
for chunk in self.chunks:
yield ChatGenerationChunk(message=AIMessageChunk(content=chunk))
@property
def _llm_type(self) -> str:
return "fake"
@pytest.fixture
def fake_chain():
# Simulate the model emitting JSON across 4 chunks
chunks = [
'{"order_id": "ORD-8842", "customer_email": "sarah.chen@example.com", ',
'"currency": "USD", "items": [{"sku": "SKU-A100", "quantity": 2, ',
'"unit_price_cents": 2999}], "placed_at": "2024-03-15T14:30:00Z"}',
]
llm = FakeStreamingLLM(chunks=chunks)
return prompt | llm | parser
def test_streaming_parses_correctly(fake_chain):
results = list(fake_chain.stream({"input_text": "dummy"}))
final = results[-1]
assert isinstance(final, OrderExtraction)
assert final.order_id == "ORD-8842"
assert len(final.items) == 1
assert final.items[0].unit_price_cents == 2999This test runs in milliseconds, requires no API key, and catches parser regressions when you upgrade LangChain or Pydantic.
Verify: pytest test_streaming.py -v passes.
Verification checklist
Before shipping, confirm each of these:
- Schema adherence: Final parsed object passes all Pydantic validators (
python -c "from stream_pydantic import chain; list(chain.stream(...))[-1]") - Partial safety: Your consumer code never raises
KeyErrororAttributeErroron mid-stream chunks (runconsumer.pyand watch for crashes) - Retry behavior: Transient 5xx errors trigger exactly three attempts with backoff (check logs)
- Fallback works: Gateway routes around a degraded provider without client changes (inspect response headers for
x-n4n-modelor similar) - Token accounting:
meter.total_tokensmatches the provider’s billing (spot-check against dashboard) - CI passes:
pytestsuite green on every PR
Common pitfalls
| Pitfall | Symptom | Fix |
|---|---|---|
Forgetting streaming=True on ChatOpenAI |
Full response buffers, then yields once | Set streaming=True in config.py |
Using invoke() instead of stream() |
No partial results, defeats the purpose | Call .stream() on the runnable |
| Pydantic model too strict for streaming | Parser raises mid-stream on missing required fields | Mark fields Optional with defaults; validate at the end |
| No retry wrapper | Transient blips surface as 500s to users | Add tenacity retry as shown in Step 6 |
Assuming JsonOutputParser yields valid JSON each chunk |
json.loads() fails on partial dict |
Consume the yielded dict directly; don’t re-serialize |
Streaming structured output moves the parse boundary earlier in your pipeline. You validate incrementally, render progressively, and catch provider failures before they become user-facing errors. The patterns above are battle-tested — use them as the foundation for any LangChain application that needs structured data in real time.