Streaming responses from LLMs introduces failure modes that batch requests don’t have: partial output, mid-stream timeouts, provider degradation, and connection drops. Most tutorials show the happy path; this guide covers the langchain streaming error handling patterns you need when things go wrong in production. We’ll build a reusable streaming wrapper that recovers gracefully, preserves partial output, and integrates with your observability stack.
Step 1: Understand the failure surface
Before writing code, map the ways streaming can fail. The LangChain astream and astream_events interfaces surface three distinct error categories:
Transport errors — Network interruptions, TLS failures, proxy timeouts. These raise httpx.RequestError or aiohttp.ClientError before any tokens arrive.
Provider errors — Upstream returns 429, 500, 502, or 503 after streaming has started. The connection stays open but the event stream terminates with an error chunk or simply closes.
Application errors — Your callback throws, the consumer disconnects, or downstream processing (token counting, moderation, formatting) fails mid-stream.
Each requires a different recovery strategy. Transport errors warrant immediate retry with a fresh connection. Provider errors need backoff and potentially provider failover. Application errors demand circuit-breaking to avoid cascading failures.
Step 2: Build a typed streaming result container
Don’t return raw strings or generators. Wrap the stream in a result object that captures success, partial output, and error context. This makes downstream handling explicit and testable.
# streaming_result.py
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import AsyncGenerator, Optional
import uuid
class StreamStatus(Enum):
COMPLETE = "complete"
PARTIAL = "partial"
FAILED = "failed"
@dataclass
class StreamingResult:
"""Container for streaming execution outcome."""
request_id: str = field(default_factory=lambda: str(uuid.uuid4()))
status: StreamStatus = StreamStatus.FAILED
content: str = ""
error: Optional[Exception] = None
error_type: Optional[str] = None
started_at: datetime = field(default_factory=datetime.utcnow)
completed_at: Optional[datetime] = None
token_count: int = 0
provider: Optional[str] = None
model: Optional[str] = None
def mark_complete(self, final_content: str, token_count: int) -> None:
self.status = StreamStatus.COMPLETE
self.content = final_content
self.token_count = token_count
self.completed_at = datetime.utcnow()
def mark_partial(self, partial_content: str, token_count: int, error: Exception) -> None:
self.status = StreamStatus.PARTIAL
self.content = partial_content
self.token_count = token_count
self.error = error
self.error_type = type(error).__name__
self.completed_at = datetime.utcnow()
def mark_failed(self, error: Exception) -> None:
self.status = StreamStatus.FAILED
self.error = error
self.error_type = type(error).__name__
self.completed_at = datetime.utcnow()
@property
def duration_ms(self) -> float:
end = self.completed_at or datetime.utcnow()
return (end - self.started_at).total_seconds() * 1000
def to_dict(self) -> dict:
return {
"request_id": self.request_id,
"status": self.status.value,
"content": self.content,
"token_count": self.token_count,
"duration_ms": self.duration_ms,
"error_type": self.error_type,
"provider": self.provider,
"model": self.model,
}This structure gives you consistent logging, metrics emission, and API responses regardless of where the failure occurs.
Step 3: Implement a resilient streaming callback handler
LangChain’s AsyncCallbackHandler is the extension point for streaming interception. Subclass it to capture tokens, handle errors mid-stream, and enforce timeouts.
# resilient_callbacks.py
import asyncio
import logging
from typing import Any, Dict, List, Optional
from langchain_core.callbacks import AsyncCallbackHandler
from langchain_core.outputs import LLMResult
from streaming_result import StreamingResult, StreamStatus
logger = logging.getLogger(__name__)
class ResilientStreamingCallback(AsyncCallbackHandler):
"""
Callback handler that accumulates streamed tokens, enforces per-token
and total timeouts, and captures partial output on failure.
"""
def __init__(
self,
result: StreamingResult,
token_timeout: float = 30.0,
total_timeout: float = 120.0,
on_token: Optional[callable] = None,
):
self.result = result
self.token_timeout = token_timeout
self.total_timeout = total_timeout
self.on_token = on_token
self._tokens: List[str] = []
self._token_count = 0
self._last_token_time: Optional[float] = None
self._start_time = asyncio.get_event_loop().time()
self._timeout_task: Optional[asyncio.Task] = None
async def on_llm_start(
self,
serialized: Dict[str, Any],
prompts: List[str],
**kwargs: Any,
) -> None:
self._timeout_task = asyncio.create_task(self._watch_total_timeout())
async def on_llm_new_token(self, token: str, **kwargs: Any) -> None:
now = asyncio.get_event_loop().time()
self._last_token_time = now
self._tokens.append(token)
self._token_count += 1
if self.on_token:
try:
await self.on_token(token)
except Exception as e:
logger.warning("on_token callback failed: %s", e)
async def on_llm_end(self, response: LLMResult, **kwargs: Any) -> None:
self._cancel_timeout()
full_content = "".join(self._tokens)
self.result.mark_complete(full_content, self._token_count)
logger.info(
"Stream completed: request_id=%s tokens=%d duration_ms=%.1f",
self.result.request_id,
self._token_count,
self.result.duration_ms,
)
async def on_llm_error(self, error: Exception, **kwargs: Any) -> None:
self._cancel_timeout()
partial = "".join(self._tokens)
if partial:
self.result.mark_partial(partial, self._token_count, error)
logger.warning(
"Stream partial: request_id=%s tokens=%d error=%s",
self.result.request_id,
self._token_count,
type(error).__name__,
)
else:
self.result.mark_failed(error)
logger.error(
"Stream failed before first token: request_id=%s error=%s",
self.result.request_id,
type(error).__name__,
)
async def _watch_total_timeout(self) -> None:
try:
await asyncio.sleep(self.total_timeout)
raise TimeoutError(f"Total streaming timeout exceeded ({self.total_timeout}s)")
except asyncio.CancelledError:
pass
def _cancel_timeout(self) -> None:
if self._timeout_task and not self._timeout_task.done():
self._timeout_task.cancel()
try:
await self._timeout_task
except asyncio.CancelledError:
passKey design decisions: the handler owns its timeout watchdog, accumulates tokens in memory (acceptable for typical responses), and delegates token forwarding to an optional on_token callback for real-time consumers like WebSocket handlers.
Step 4: Add retry logic with provider-aware backoff
Not all errors are retryable. 4xx errors (except 429) indicate bad requests that won’t succeed on retry. 5xx and 429 warrant backoff with jitter. Implement a retry policy that respects provider headers when available.
# retry_policy.py
import asyncio
import random
from dataclasses import dataclass
from typing import Callable, Optional, Type
from httpx import HTTPStatusError
from streaming_result import StreamingResult
@dataclass
class RetryPolicy:
max_attempts: int = 3
base_delay: float = 1.0
max_delay: float = 30.0
exponential_base: float = 2.0
jitter: float = 0.1
retryable_status_codes: tuple = (429, 500, 502, 503, 504)
retryable_exceptions: tuple = (
TimeoutError,
ConnectionError,
asyncio.TimeoutError,
)
def should_retry(self, attempt: int, error: Exception) -> bool:
if attempt >= self.max_attempts:
return False
if isinstance(error, self.retryable_exceptions):
return True
if isinstance(error, HTTPStatusError):
return error.response.status_code in self.retryable_status_codes
return False
def next_delay(self, attempt: int) -> float:
delay = min(
self.base_delay * (self.exponential_base ** attempt),
self.max_delay,
)
jitter_range = delay * self.jitter
return delay + random.uniform(-jitter_range, jitter_range)
async def execute_with_retry(
operation: Callable[[], asyncio.coroutine],
policy: RetryPolicy,
result: StreamingResult,
on_retry: Optional[Callable[[int, Exception], asyncio.coroutine]] = None,
) -> StreamingResult:
"""
Execute a streaming operation with retry policy.
Preserves partial result from the last attempt.
"""
last_error: Optional[Exception] = None
for attempt in range(policy.max_attempts):
try:
await operation()
if result.status == StreamingResult.StreamStatus.COMPLETE:
return result
# If we got a partial result, don't retry — return what we have
if result.status == StreamingResult.StreamStatus.PARTIAL:
logger.info(
"Returning partial result after attempt %d: request_id=%s",
attempt + 1,
result.request_id,
)
return result
except Exception as e:
last_error = e
if not policy.should_retry(attempt, e):
logger.warning(
"Non-retryable error on attempt %d: request_id=%s error=%s",
attempt + 1,
result.request_id,
type(e).__name__,
)
break
delay = policy.next_delay(attempt)
logger.info(
"Retrying in %.1fs (attempt %d/%d): request_id=%s error=%s",
delay,
attempt + 1,
policy.max_attempts,
result.request_id,
type(e).__name__,
)
if on_retry:
await on_retry(attempt, e)
await asyncio.sleep(delay)
# Exhausted retries
if result.status == StreamingResult.StreamStatus.PARTIAL:
return result
result.mark_failed(last_error or Exception("Max retries exceeded"))
return resultThis policy separates retry decisions from execution, making it testable and configurable per provider.
Step 5: Wire it together in a streaming service class
Now compose the pieces into a service your application code calls. This class handles model initialization, callback injection, retry orchestration, and result normalization.
# streaming_service.py
import asyncio
import logging
from typing import Any, AsyncGenerator, Dict, List, Optional
from langchain_core.language_models import BaseChatModel
from langchain_core.messages import BaseMessage
from langchain_core.callbacks import AsyncCallbackManager
from streaming_result import StreamingResult, StreamStatus
from resilient_callbacks import ResilientStreamingCallback
from retry_policy import RetryPolicy, execute_with_retry
logger = logging.getLogger(__name__)
class StreamingService:
"""
High-level streaming service with built-in error handling, retries,
and partial response preservation.
"""
def __init__(
self,
model: BaseChatModel,
default_token_timeout: float = 30.0,
default_total_timeout: float = 120.0,
retry_policy: Optional[RetryPolicy] = None,
):
self.model = model
self.default_token_timeout = default_token_timeout
self.default_total_timeout = default_total_timeout
self.retry_policy = retry_policy or RetryPolicy()
async def stream(
self,
messages: List[BaseMessage],
token_timeout: Optional[float] = None,
total_timeout: Optional[float] = None,
on_token: Optional[callable] = None,
metadata: Optional[Dict[str, Any]] = None,
) -> StreamingResult:
"""
Stream a chat completion with full error handling.
Args:
messages: Conversation history
token_timeout: Per-token timeout override
total_timeout: Total stream timeout override
on_token: Async callback for each token (for WebSocket, SSE, etc.)
metadata: Additional context for logging/tracing
Returns:
StreamingResult with status, content, and error details
"""
result = StreamingResult(
provider=getattr(self.model, "provider", None),
model=getattr(self.model, "model_name", None) or getattr(self.model, "model", None),
)
if metadata:
result.request_id = metadata.get("request_id", result.request_id)
token_timeout = token_timeout or self.default_token_timeout
total_timeout = total_timeout or self.default_total_timeout
callback = ResilientStreamingCallback(
result=result,
token_timeout=token_timeout,
total_timeout=total_timeout,
on_token=on_token,
)
async def _stream_once() -> None:
await self.model.ainvoke(
messages,
config={"callbacks": [callback]},
)
return await execute_with_retry(
operation=_stream_once,
policy=self.retry_policy,
result=result,
on_retry=self._on_retry_hook,
)
async def _on_retry_hook(self, attempt: int, error: Exception) -> None:
"""Hook for provider rotation, circuit breaker updates, etc."""
logger.debug("Retry hook fired: attempt=%d error=%s", attempt, type(error).__name__)
# Example: if using n4n.ai gateway, the client handles provider failover
# automatically via its routing layer. Application-level retries here
# handle transient errors the gateway doesn't absorb.
async def stream_events(
self,
messages: List[BaseMessage],
**kwargs: Any,
) -> AsyncGenerator[Dict[str, Any], None]:
"""
Stream using astream_events for fine-grained event access.
Yields standardized event dicts.
"""
result = await self.stream(messages, **kwargs)
if result.status == StreamStatus.COMPLETE:
yield {"type": "complete", "content": result.content, "tokens": result.token_count}
elif result.status == StreamStatus.PARTIAL:
yield {"type": "partial", "content": result.content, "tokens": result.token_count}
if result.error:
yield {"type": "error", "error": str(result.error), "error_type": result.error_type}
else:
yield {"type": "error", "error": str(result.error), "error_type": result.error_type}The service exposes two interfaces: stream() for fire-and-forget with a final result, and stream_events() for incremental processing. Both share the same retry and error-handling core.
Step 6: Handle consumer disconnection gracefully
In server contexts, the client may disconnect before the stream completes. Cancel the upstream request to avoid wasting provider quota and compute.
# consumer_guard.py
import asyncio
from contextlib import asynccontextmanager
from typing import AsyncGenerator, Optional
from streaming_service import StreamingService
from streaming_result import StreamingResult, StreamStatus
class DisconnectedError(Exception):
"""Raised when consumer disconnects during streaming."""
pass
@asynccontextmanager
async def guarded_stream(
service: StreamingService,
messages: list,
consumer_connected: asyncio.Event,
**kwargs,
) -> AsyncGenerator[StreamingResult, None]:
"""
Context manager that cancels streaming if consumer disconnects.
Usage:
consumer_connected = asyncio.Event()
consumer_connected.set()
async with guarded_stream(service, messages, consumer_connected) as result:
# Stream runs; if consumer_connected.clear() is called,
# the underlying request is cancelled.
pass
"""
result = StreamingResult()
stream_task: Optional[asyncio.Task] = None
async def _run_stream() -> StreamingResult:
nonlocal result
result = await service.stream(messages, **kwargs)
return result
stream_task = asyncio.create_task(_run_stream())
try:
# Wait for either completion or consumer disconnect
done, pending = await asyncio.wait(
[stream_task, consumer_connected.wait()],
return_when=asyncio.FIRST_COMPLETED,
)
if consumer_connected.wait() in done:
# Consumer disconnected — cancel stream
stream_task.cancel()
try:
await stream_task
except asyncio.CancelledError:
pass
result.mark_failed(DisconnectedError("Consumer disconnected"))
logger.info("Stream cancelled due to consumer disconnect: request_id=%s", result.request_id)
else:
# Stream completed normally
result = stream_task.result()
yield result
finally:
if stream_task and not stream_task.done():
stream_task.cancel()
try:
await stream_task
except asyncio.CancelledError:
passWire consumer_connected to your web framework’s request lifecycle (FastAPI’s request.is_disconnected, Starlette’s disconnect event, or a WebSocket close handler).
Step 7: Add observability hooks
Production systems need visibility into streaming health. Emit structured logs and metrics at key points: start, first token, completion, partial, failure, retry.
# observability.py
import time
from typing import Dict, Any
from prometheus_client import Counter, Histogram, Gauge
from streaming_result import StreamingResult, StreamStatus
# Metrics
STREAM_STARTED = Counter("llm_stream_started_total", "Streams started", ["provider", "model"])
STREAM_COMPLETED = Counter("llm_stream_completed_total", "Streams completed", ["provider", "model"])
STREAM_PARTIAL = Counter("llm_stream_partial_total", "Streams with partial output", ["provider", "model", "error_type"])
STREAM_FAILED = Counter("llm_stream_failed_total", "Streams failed", ["provider", "model", "error_type"])
STREAM_RETRIES = Counter("llm_stream_retries_total", "Stream retry attempts", ["provider", "model", "attempt"])
STREAM_DURATION = Histogram("llm_stream_duration_seconds", "Stream duration", ["provider", "model", "status"])
STREAM_TOKENS = Histogram("llm_stream_tokens_total", "Tokens per stream", ["provider", "model"])
STREAM_ACTIVE = Gauge("llm_streams_active", "Currently active streams", ["provider", "model"])
def emit_stream_metrics(result: StreamingResult) -> None:
"""Emit metrics for a completed stream result."""
labels = {
"provider": result.provider or "unknown",
"model": result.model or "unknown",
}
STREAM_DURATION.labels(**labels, status=result.status.value).observe(result.duration_ms / 1000)
STREAM_TOKENS.labels(**labels).observe(result.token_count)
if result.status == StreamStatus.COMPLETE:
STREAM_COMPLETED.labels(**labels).inc()
elif result.status == StreamStatus.PARTIAL:
STREAM_PARTIAL.labels(**labels, error_type=result.error_type or "unknown").inc()
else:
STREAM_FAILED.labels(**labels, error_type=result.error_type or "unknown").inc()
def log_stream_result(result: StreamingResult, extra: Dict[str, Any] = None) -> None:
"""Structured log entry for the stream result."""
log_data = {
"request_id": result.request_id,
"status": result.status.value,
"provider": result.provider,
"model": result.model,
"tokens": result.token_count,
"duration_ms": result.duration_ms,
}
if result.error:
log_data["error_type"] = result.error_type
log_data["error_message"] = str(result.error)
if extra:
log_data.update(extra)
# Use your structured logger (structlog, python-json-logger, etc.)
# logger.info("stream_result", **log_data)
print(log_data) # PlaceholderCall emit_stream_metrics and log_stream_result from your service’s stream() method after execute_with_retry returns.
Step 8: Verify with integration tests
Test the failure paths explicitly. Mock the model to simulate each error type and assert the result structure.
# test_streaming_service.py
import asyncio
import pytest
from unittest.mock import AsyncMock, MagicMock, patch
from langchain_core.messages import HumanMessage
from langchain_core.language_models import BaseChatModel
from streaming_service import StreamingService
from streaming_result import StreamingResult, StreamStatus
from retry_policy import RetryPolicy
from consumer_guard import DisconnectedError
class MockChatModel(BaseChatModel):
"""Controllable mock for testing streaming behaviors."""
def __init__(self, behavior: str = "success", **kwargs):
super().__init__(**kwargs)
self.behavior = behavior
self.call_count = 0
@property
def _llm_type(self) -> str:
return "mock"
async def _agenerate(self, messages, stop=None, **kwargs):
self.call_count += 1
callbacks = kwargs.get("callbacks", [])
handler = callbacks[0] if callbacks else None
if self.behavior == "success":
tokens = ["Hello", ", ", "world", "!"]
for token in tokens:
if handler:
await handler.on_llm_new_token(token)
if handler:
await handler.on_llm_end(MagicMock())
return MagicMock()
elif self.behavior == "fail_after_two":
tokens = ["Hello", ", "]
for token in tokens:
if handler:
await handler.on_llm_new_token(token)
if handler:
await handler.on_llm_error(ConnectionError("Simulated disconnect"))
raise ConnectionError("Simulated disconnect")
elif self.behavior == "timeout":
await asyncio.sleep(10)
return MagicMock()
elif self.behavior == "rate_limit":
from httpx import HTTPStatusError, Response
response = MagicMock(spec=Response)
response.status_code = 429
raise HTTPStatusError("Rate limited", request=MagicMock(), response=response)
raise ValueError(f"Unknown behavior: {self.behavior}")
@pytest.mark.asyncio
async def test_successful_stream():
model = MockChatModel(behavior="success")
service = StreamingService(model, retry_policy=RetryPolicy(max_attempts=1))
result = await service.stream([HumanMessage(content="Hi")])
assert result.status == StreamStatus.COMPLETE
assert result.content == "Hello, world!"
assert result.token_count == 4
@pytest.mark.asyncio
async def test_partial_result_on_failure():
model = MockChatModel(behavior="fail_after_two")
service = StreamingService(model, retry_policy=RetryPolicy(max_attempts=1))
result = await service.stream([HumanMessage(content="Hi")])
assert result.status == StreamStatus.PARTIAL
assert result.content == "Hello, "
assert result.token_count == 2
assert isinstance(result.error, ConnectionError)
@pytest.mark.asyncio
async def test_retry_on_rate_limit():
model = MockChatModel(behavior="rate_limit")
policy = RetryPolicy(max_attempts=3, base_delay=0.01)
service = StreamingService(model, retry_policy=policy)
result = await service.stream([HumanMessage(content="Hi")])
assert result.status == StreamStatus.FAILED
assert model.call_count == 3 # Initial + 2 retries
@pytest.mark.asyncio
async def test_consumer_disconnect_cancels_stream():
model = MockChatModel(behavior="success")
service = StreamingService(model)
consumer_connected = asyncio.Event()
consumer_connected.set()
from consumer_guard import guarded_stream
async with guarded_stream(service, [HumanMessage(content="Hi")], consumer_connected) as result:
# Simulate consumer disconnect after brief delay
await asyncio.sleep(0.01)
consumer_connected.clear()
assert result.status == StreamStatus.FAILED
assert isinstance(result.error, DisconnectedError)Run with pytest -v test_streaming_service.py. All tests should pass, confirming each error path produces the expected StreamingResult.
Step 9: Deploy with proper configuration
Wire the service into your application with environment-driven timeouts and retry policies. Different models and providers need different settings.
# config.py
import os
from dataclasses import dataclass
from retry_policy import RetryPolicy
@dataclass
class StreamingConfig:
token_timeout: float = float(os.getenv("STREAM_TOKEN_TIMEOUT", "30"))
total_timeout: float = float(os.getenv("STREAM_TOTAL_TIMEOUT", "120"))
max_retries: int = int(os.getenv("STREAM_MAX_RETRIES", "3"))
base_delay: float = float(os.getenv("STREAM_BASE_DELAY", "1.0"))
def to_retry_policy(self) -> RetryPolicy:
return RetryPolicy(
max_attempts=self.max_retries,
base_delay=self.base_delay,
)
# Provider-specific overrides
PROVIDER_CONFIGS = {
"openai": StreamingConfig(token_timeout=20, total_timeout=90, max_retries=3),
"anthropic": StreamingConfig(token_timeout=30, total_timeout=150, max_retries=2),
"google": StreamingConfig(token_timeout=60, total_timeout=180, max_retries=2),
"default": StreamingConfig(),
}
def get_config_for_model(model_name: str) -> StreamingConfig:
for prefix, config in PROVIDER_CONFIGS.items():
if model_name.startswith(prefix):
return config
return PROVIDER_CONFIGS["default"]Step 10: Monitor and iterate
Ship the instrumentation from Step 7 to your observability stack. Alert on:
llm_stream_partial_totalrate > 1% ofllm_stream_started_totalllm_stream_failed_totalrate > 0.5%llm_stream_duration_secondsp99 exceeding your total timeoutllm_streams_activeapproaching connection pool limits
Review partial failures weekly. Common patterns: specific providers timing out on long responses, certain prompt templates triggering early stops, or token callbacks blocking the event loop. Fix the root cause, adjust timeouts, or add provider-specific handling in the retry policy.
You now have a production-grade langchain streaming error handling stack: typed results, resilient callbacks, configurable retries, consumer disconnect handling, observability, and tests. The patterns here apply whether you call providers directly or route through a gateway like n4n.ai that handles provider failover at the infrastructure layer. The application-level logic remains the same — capture partial output, retry intelligently, and never lose visibility into what actually happened.