Why this matters
A context pipeline in development runs in under 100ms. The same pipeline in production needs to serve thousands of concurrent users, each requiring fresh embeddings, database lookups, and possibly LLM-based summarization before the main model call starts.
Production context engineering is about making that pipeline fast, reliable, and observable.
Core concepts
Component caching
| Component | Cache duration | Cache key |
|---|---|---|
| User profile | 1 hour | user:{id}:profile |
| System prompt tokens | Indefinitely | system_prompt:{hash} |
| Session summary | Until next summarization | session:{id}:summary |
| Query embeddings | 15 minutes | embed:{hash(query)} |
| Retrieved chunks | 5 minutes | retrieval:{hash(query)}:{k} |
import hashlib
import json
from typing import Any, Callable
class ContextCache:
def __init__(self, backend):
self.backend = backend
def _key(self, prefix: str, content: str) -> str:
return f"{prefix}:{hashlib.sha256(content.encode()).hexdigest()[:16]}"
async def get_or_compute(
self, prefix: str, content: str, compute_fn: Callable[[], Any], ttl_seconds: int = 300
) -> Any:
key = self._key(prefix, content)
cached = await self.backend.get(key)
if cached is not None:
return json.loads(cached)
result = await compute_fn()
await self.backend.set(key, json.dumps(result), ex=ttl_seconds)
return result
Parallel context fetching
import asyncio
from dataclasses import dataclass
from typing import Optional
@dataclass
class ContextComponents:
user_profile: Optional[str]
retrieved_docs: list[str]
session_history: list[dict]
tool_results: list[str]
async def fetch_context_parallel(
user_id: str, query: str, session_id: str,
vector_store, user_store, session_store, retrieval_k: int = 5,
) -> ContextComponents:
profile, retrieval, history = await asyncio.gather(
user_store.get_profile(user_id),
vector_store.search(query, k=retrieval_k),
session_store.get_history(session_id),
return_exceptions=True,
)
return ContextComponents(
user_profile=profile if not isinstance(profile, Exception) else None,
retrieved_docs=[hit["content"] for hit in retrieval] if not isinstance(retrieval, Exception) else [],
session_history=history if not isinstance(history, Exception) else [],
tool_results=[],
)
return_exceptions=True ensures a single slow fetch does not block the others.
Latency budgets
CONTEXT_FETCH_TIMEOUT_SEC = 1.5
async def with_timeout(coro, timeout: float, fallback=None):
try:
return await asyncio.wait_for(coro, timeout=timeout)
except asyncio.TimeoutError:
return fallback
A response without retrieved documents is better than no response at all.
Code: ProductionContextPipeline
import asyncio
import time
import logging
from dataclasses import dataclass, field
logger = logging.getLogger(__name__)
@dataclass
class PipelineMetrics:
total_latency_ms: float = 0
fetch_latency_ms: float = 0
assemble_latency_ms: float = 0
cache_hits: int = 0
cache_misses: int = 0
context_tokens: int = 0
warnings: list[str] = field(default_factory=list)
@dataclass
class ProductionContextPipeline:
vector_store: object
user_store: object
session_store: object
cache: ContextCache
memory_manager: MemoryManager
formatter: ToolResultFormatter
debugger: ContextDebugger
async def assemble(
self,
user_id: str,
session_id: str,
query: str,
system_prompt: str,
tool_results: list[tuple[str, object]] | None = None,
) -> tuple[list[dict], PipelineMetrics]:
metrics = PipelineMetrics()
t0 = time.monotonic()
t1 = time.monotonic()
cache_key = f"{user_id}:{session_id}:{hash(query) % 100000}"
components = await self.cache.get_or_compute(
"ctx", cache_key,
lambda: fetch_context_parallel(
user_id, query, session_id,
self.vector_store, self.user_store, self.session_store,
),
ttl_seconds=60,
)
metrics.fetch_latency_ms = (time.monotonic() - t1) * 1000
if tool_results:
await self.formatter.process_multiple(tool_results, query=query)
t2 = time.monotonic()
await self.memory_manager.load_session(session_id)
for turn in components.session_history:
self.memory_manager.add_turn(turn["role"], turn["content"])
messages = await self.memory_manager.assemble_context(
system_prompt=system_prompt,
user_input=query,
retrieved_docs=components.retrieved_docs,
)
metrics.assemble_latency_ms = (time.monotonic() - t2) * 1000
report = self.debugger.analyze_messages(messages)
metrics.context_tokens = report.total_tokens
metrics.warnings = report.warnings
metrics.total_latency_ms = (time.monotonic() - t0) * 1000
logger.info(
"context_pipeline_complete",
extra={
"user_id": user_id,
"total_latency_ms": metrics.total_latency_ms,
"context_tokens": metrics.context_tokens,
},
)
return messages, metrics
Production monitoring
Track these metrics in your observability platform:
- p50/p95/p99 context assembly latency — catch slowdowns before they affect users
- Context tokens per request — monitor for gradual bloat
- Cache hit rate — a low hit rate suggests incorrect TTL or cache keys
- Context warning rate — fraction of requests triggering quality warnings
def emit_context_metrics(metrics: PipelineMetrics, statsd_client) -> None:
statsd_client.timing("context.latency_ms", metrics.total_latency_ms)
statsd_client.gauge("context.tokens", metrics.context_tokens)
statsd_client.increment("context.cache_hits", metrics.cache_hits)
statsd_client.increment("context.cache_misses", metrics.cache_misses)
if metrics.warnings:
statsd_client.increment("context.quality_warnings", len(metrics.warnings))
Common mistakes
Sequential fetching in the hot path. Confirm that independent fetches run concurrently.
No timeout on context assembly. Always set a hard timeout with a graceful degradation path.
Caching assembled messages rather than components. Cache components and re-assemble each turn.
Not alarming on quality warnings. Wire warning counts to an alert threshold.
Try it yourself
- Implement a benchmark that runs
assemble()100 times and reports p50, p95, and p99 latency. - Add a circuit breaker to the vector store fetch: skip retrieval after 5 failures in 30 seconds.
- Design a canary deployment strategy for context pipeline changes. What metrics would you monitor during a 5% rollout?