Building Production-Ready AI Agent Middleware with LangGraph: Enterprise Patterns, Guardrails, Human Approval & LLM Gateways
Executive Summary
In this comprehensive guide, you’ll master production AI agent architecture for LangGraph with practical enterprise patterns:
✓ Middleware vs Infrastructure — Architectural distinctions and component placement
✓ LangGraph Human-in-the-Loop — True interrupt() patterns with Command(resume=...)
✓ Cache Layers - Prompt, response, embedding, semantic, tool, and RAG caching
✓ LLM Gateways - LiteLLM, OpenRouter, Azure AI Foundry, and enterprise routing
✓ Security Middleware - Prompt injection, jailbreak, secret detection, content moderation
✓ Agent Guardrails - Llama Guard, NeMo Guardrails, and policy enforcement
✓ Memory Systems - Short-term, long-term, semantic, episodic with LangGraph Store
✓ Cost Enforcement - Budget limits with automatic model downgrading
✓ Multi-Agent Middleware - Supervisor patterns with per-agent validation
✓ Observability - OpenTelemetry, LangSmith, and distributed tracing
✓ Production Storage - PostgresSaver, Redis, and enterprise databases
What you’ll build: Enterprise-grade AI systems that balance control, observability, security, and performance.
Architecture Foundation: Beyond Simple Middleware
The AI Execution Pipeline
Before diving into code, understand the enterprise architecture:
text
┌─────────────────────────────────────────────────────────────┐│ Client Layer ││ (Web, Mobile, API, Slack, etc.) │└─────────────────────────────────────────────────────────────┘ │ ▼┌─────────────────────────────────────────────────────────────┐│ API Gateway Layer ││ • Authentication (JWT, OAuth2) • Rate Limiting ││ • Request Validation • Content Moderation │└─────────────────────────────────────────────────────────────┘ │ ▼┌─────────────────────────────────────────────────────────────┐│ Security Middleware Layer ││ • Prompt Injection Detection • Jailbreak Detection ││ • Secret Detection/Redaction • PII Detection ││ • Policy Enforcement (OPA/Cedar) • Audit Logging │└─────────────────────────────────────────────────────────────┘ │ ▼┌─────────────────────────────────────────────────────────────┐│ Application Middleware ││ • Trace ID Setup • Performance Monitoring ││ • Feature Flags • Request Correlation │└─────────────────────────────────────────────────────────────┘ │ ▼┌─────────────────────────────────────────────────────────────┐│ LLM Gateway Layer ││ • Model Routing (LiteLLM) • Budget Enforcement ││ • Fallback Strategy • Prompt Versioning ││ • Cost Tracking • Model Selection │└─────────────────────────────────────────────────────────────┘ │ ▼┌─────────────────────────────────────────────────────────────┐│ LangGraph Orchestration Layer ││ • State Management • Agent Logic ││ • Branching Decisions • Multi-Agent Coordination ││ • Human-in-the-Loop (interrupt) • Checkpointing │└─────────────────────────────────────────────────────────────┘ │ ▼┌─────────────────────────────────────────────────────────────┐│ Tool Execution Layer ││ • Tool Approval (interrupt) • Retry + Circuit Breaker ││ • MCP Validation • Tool Caching ││ • Timeout Enforcement • Error Handling │└─────────────────────────────────────────────────────────────┘ │ ▼┌─────────────────────────────────────────────────────────────┐│ Output Processing Layer ││ • Guardrails (Llama Guard) • PII Redaction ││ • Content Moderation • Observability │└─────────────────────────────────────────────────────────────┘ │ ▼┌─────────────────────────────────────────────────────────────┐│ Observability Layer ││ • OpenTelemetry Traces • LangSmith ││ • Prometheus Metrics • Structured Logging ││ • Distributed Tracing • Alerting │└─────────────────────────────────────────────────────────────┘
Critical Distinction: Middleware vs Infrastructure
Understanding what is actually middleware versus broader infrastructure is crucial:
Application Middleware (Cross-cutting concerns)
- Authentication/authorization
- Rate limiting
- Request validation
- PII detection/redaction
- Logging and metrics
- Timeout enforcement
- Structured error handling
Infrastructure Components (Platform services)
- LLM Gateway (routing, fallback, cost management)
- Memory Layer (short-term, long-term, semantic)
- Prompt Management (versioning, A/B testing)
- Model Registry (model selection, capabilities)
- Cache Layer (response, embedding, semantic)
Orchestration (Workflow components)
- Agent nodes (decision logic)
- Tool selection and execution
- State transitions
- Branching logic
- Multi-agent coordination
Production Principle: Use middleware for cross-cutting concerns, infrastructure for platform services, and orchestration for workflow logic.
LangGraph Human-in-the-Loop: The Correct Pattern
True interrupt() Implementation
The modern, officially recommended approach uses interrupt() with Command(resume=...):
python
from langgraph.types import interruptfrom langgraph.graph import StateGraph, START, END, Commandfrom typing import Dict, Any, Literalimport datetime
php
def tool_approval_node(state: Dict[str, Any]) -> Command[Literal["execute_tool", "reject_tool"]]: """ execution for human approval using true interrupt()""" tool_name = state.get("selected_tool") sensitive_tools = ["send_email", "transfer_money", "delete_data"] if tool_name not in sensitive_tools: # Continue without interruption return Command(goto="execute_tool") # TRUE interrupt - graph s here # This is the official LangGraph pattern approval = interrupt({ "request_type": "approval", "tool": tool_name, "arguments": state.get("tool_args"), "timestamp": datetime.now().isoformat() }) # After human response via Command(resume=...) if approval.get("approved"): state["tool_approved"] = True return Command(goto="execute_tool", update=state) else: state["tool_rejected"] = True state["rejection_reason"] = approval.get("reason", "No reason provided") return Command(goto="reject_tool", update=state)
python
Why this is better:
- ✅ Official LangGraph pattern with true interrupt()
- ✅ Cleaner control flow with Command(resume=...)
- ✅ Automatic resumption with context preservation
- ✅ Built-in state preservation across interruptions
- ✅ Better checkpointing support
- ✅ Type-safe with Literal for routing
- ✅ Production-ready with PostgresSaver
Security Middleware: Enterprise Protection
Prompt Injection Detection
python
import refrom typing import Dict, Any
class PromptInjectionMiddleware: """Detect and block prompt injection attempts""" def __init__(self): # Illustrative patterns - production systems should use ML-based detection self.injection_patterns = [ r"ignore (all|previous|above) (instructions|commands)", r"you are (now|going to) (ignore|override)", r"new (instruction|command|rule):", r"system:", r"assistant: override", r"forget (all|your) (instructions|previous)", r"disregard (?:all|any) (?:previous|above|prior) (?:prompt|instruction|rule)", r"stop being (?:an?|the) AI", r"act as (?:if|though) (?:you are|you're) (?:not|no longer)", r"from now on, (?:you will|you are)", r"you must (?:now|always) (?:respond|answer|act)" ] async def validate_input(self, state: Dict[str, Any]) -> Dict[str, Any]: """Check for prompt injection patterns""" user_input = state.get("input", "") for pattern in self.injection_patterns: if re.search(pattern, user_input, re.IGNORECASE): return { **state, "injection_detected": True, "injection_type": "direct", "error": "[BLOCKED: Prompt injection attempt detected]" } # Check for indirect injection via data/content # Production: Use ML-based detection for complex indirect injections if await self._check_indirect_injection(user_input): return { **state, "injection_detected": True, "injection_type": "indirect", "error": "[BLOCKED: Indirect injection via content]" } return {**state, "input_validated": True} async def _check_indirect_injection(self, content: str) -> bool: """Check for indirect injection patterns""" # Production: Use more sophisticated detection # Example: Check for encoded instructions, hidden system prompts return False
Jailbreak Detection
python
class JailbreakDetectionMiddleware: """Detect attempts to jailbreak the model""" def __init__(self): # Illustrative patterns - production should use ensemble detection self.jailbreak_patterns = [ r"dANgerous.*(?:aCt|behAv|exPL)", # Obfuscated text r"\[SYSTEM\].*\[/SYSTEM\]", # System tag injection r"(?:role|system).*assistant", # Role confusion r"(?:developer|admin) mode", # Privilege escalation r"my (?:grandma|grandfather)", # Common social engineering r"you are (?:now|going to be) (?:an?|the) (?:administrator|developer)", r"enable (?:developer|admin) mode", r"jailbreak" ] async def detect_jailbreak(self, state: Dict[str, Any]) -> Dict[str, Any]: """Check for jailbreak attempts""" combined_input = state.get("input", "") + " " + state.get("context", "") for pattern in self.jailbreak_patterns: if re.search(pattern, combined_input, re.IGNORECASE): return { **state, "jailbreak_detected": True, "violation_type": "jailbreak_attempt", "error": "[BLOCKED: Jailbreak attempt detected]" } return {**state, "safety_checked": True}
Secret Detection
python
class SecretDetectionMiddleware: """Detect and redact secrets before sending to LLM""" def __init__(self): # Illustrative patterns - production should use comprehensive detection self.secret_patterns = { "aws_key": r"AKIA[0-9A-Z]{16}", "aws_secret": r"[A-Za-z0-9/+=]{40}", "openai_key": r"sk-[A-Za-z0-9]{48}", "github_token": r"ghp_[A-Za-z0-9]{36}", "password": r"password\s*[:=]\s*['\"]?[^\s'\"]+['\"]?", "api_key": r"api[_-]?key\s*[:=]\s*['\"]?[^\s'\"]+['\"]?" } async def detect_and_redact(self, state: Dict[str, Any]) -> Dict[str, Any]: """Find and redact secrets in prompts""" content = state.get("input", "") redacted_content = content for secret_type, pattern in self.secret_patterns.items(): if re.search(pattern, content, re.IGNORECASE): redacted_content = re.sub( pattern, f"[REDACTED_{secret_type.upper()}]", redacted_content, flags=re.IGNORECASE ) # Log for security audit await self._log_secret_attempt( user_id=state.get("user_id"), secret_type=secret_type ) return { **state, "input": redacted_content, "secrets_redacted": redacted_content != content, "security_audit_logged": True } async def _log_secret_attempt(self, user_id: str, secret_type: str): """Log security events to SIEM""" # Production: Send to Splunk, Datadog, or ELK pass
Content Moderation
python
class ContentModerationMiddleware: """Comprehensive content moderation""" def __init__(self): self.toxic_categories = [ "violence", "hate", "sexual", "self_harm", "extremism", "harassment" ] self.moderation_threshold = 0.7 # Confidence threshold async def moderate_content(self, state: Dict[str, Any]) -> Dict[str, Any]: """Moderate content for safety""" content = state.get("input", "") # Check each category violations = [] for category in self.toxic_categories: score = await self._get_toxicity_score(content, category) if score > self.moderation_threshold: violations.append({ "category": category, "score": score }) if violations: return { **state, "moderation_violation": True, "violations": violations, "error": f"[BLOCKED: Content moderation violation - {', '.join([v['category'] for v in violations])}]" } # Check against policy engine (OPA/Cedar) if not await self._check_policy_compliance(content, state.get("user_id")): return { **state, "policy_violation": True, "error": "[BLOCKED: Policy violation]" } return {**state, "content_moderated": True} async def _check_policy_compliance(self, content: str, user_id: str) -> bool: """Check against enterprise policy using OPA or Cedar""" # Production: Integrate with Open Policy Agent # or Cedar for fine-grained policy enforcement return True
Agent Guardrails: Modern Safety
Guardrail Framework
Note: Llama Guard and NeMo Guardrails have specific deployment patterns. The following is illustrative and should be adapted to your deployment method (Hugging Face Transformers, serving endpoints, or cloud APIs).
python
python
import logging
logger = logging.getLogger(__name__)
class AgentGuardrails: """Multi-layer output safety with enterprise guardrails""" def __init__(self): # Production: Deploy Llama Guard through Hugging Face or serving endpoint # self.llama_guard = AutoModelForSequenceClassification.from_pretrained( # "meta-llama/Llama-Guard-3-1B" # ) # self.tokenizer = AutoTokenizer.from_pretrained("meta-llama/Llama-Guard-3-1B") # NeMo Guardrails: Configure through rails definitions # from nemoguardrails import RailsConfig, LLMRails # config = RailsConfig.from_path("./config") # self.rails = LLMRails(config) pass async def validate_output(self, state: Dict[str, Any]) -> Dict[str, Any]: """Multi-layer safety checks - illustrative implementation""" output = state.get("output", "") # Layer 1: Safety classification (Llama Guard) # Production: Call Llama Guard through appropriate interface if await self._check_safety(output) == "unsafe": return { **state, "guardrail_violated": True, "violation_type": "safety_violation", "output": "[CONTENT BLOCKED: Safety violation detected]" } # Layer 2: Policy compliance (NeMo Guardrails) # Production: Use NeMo Rails application if not await self._check_policy(output): return { **state, "guardrail_violated": True, "violation_type": "policy", "output": "[CONTENT BLOCKED: Policy violation]" } # Layer 3: Hallucination detection if await self._contains_hallucination(state, output): return { **state, "guardrail_violated": True, "violation_type": "hallucination", "confidence": state.get("hallucination_score", 0.0) } # Layer 4: Toxicity detection if await self._is_toxic(output): return { **state, "guardrail_violated": True, "violation_type": "toxicity", "output": self._moderate_content(output) } return {**state, "guardrail_passed": True} async def _check_safety(self, output: str) -> str: """Check safety using classifier""" # Production: Use Llama Guard through appropriate interface # Return "safe" or "unsafe" return "safe" async def _check_policy(self, output: str) -> bool: """Check policy compliance""" # Production: Use NeMo Guardrails or custom policy engine return True async def _is_toxic(self, output: str) -> bool: """Check toxicity using detoxify or similar""" # Production: Use detoxify, google-cloud-moderation, etc. return False
Recommended Libraries:
- Llama Guard: Deploy via Hugging Face Transformers or serving endpoint
- NeMo Guardrails: Configure through rails definitions
- Microsoft Presidio: PII detection
- Detoxify: Toxicity detection
Enterprise LLM Gateway Design
LLM Gateway with LiteLLM Integration
python
python
import litellmfrom typing import Dict, Any, Optional
class EnterpriseLLMGateway: """Production LLM gateway with enterprise features""" def __init__(self): # Configure LiteLLM for production litellm.set_verbose = False # Illustrative model routing - update based on available models self.model_routing = { "simple_qa": ["gpt-4o-mini", "claude-3-haiku"], "complex_reasoning": ["gpt-4o", "claude-3-7-sonnet"], "long_context": ["claude-3-7-sonnet", "gemini-1.5-pro"], "code_generation": ["claude-3-7-sonnet", "gpt-4o"], "fast_response": ["gpt-4o-mini", "gemini-1.5-flash"], "vision": ["gpt-4o", "claude-3-5-sonnet"], "cost_sensitive": ["gpt-4o-mini", "claude-3-haiku"] } # Fallback chains self.fallback_chains = { "premium": ["gpt-4o", "claude-3-7-sonnet", "gemini-1.5-pro"], "standard": ["gpt-4o", "claude-3-5-sonnet", "gemini-1.5-pro"], "budget": ["gpt-4o-mini", "claude-3-haiku", "gemini-1.5-flash"] } async def route_request(self, state: Dict[str, Any]) -> Dict[str, Any]: """Route to appropriate model with fallback""" # Determine use case use_case = state.get("use_case", "standard") tier = state.get("tier", "standard") # Get model chain if use_case in self.model_routing: model_chain = self.model_routing[use_case] else: model_chain = self.fallback_chains.get(tier, self.fallback_chains["standard"]) # Try each model with fallback last_error = None for model in model_chain: try: response = await self._call_model_with_retry(model, state) return { **state, "model_used": model, "response": response, "fallback_chain": model_chain } except Exception as e: last_error = e await self._record_model_failure(model, str(e)) continue # All models failed return { **state, "error": f"All models failed in chain: {model_chain}. Last error: {last_error}", "model_failures": True } async def _call_model_with_retry(self, model: str, state: Dict[str, Any]) -> str: """Call model with retry logic""" # Production: Use tenacity for retry with exponential backoff # Simplified for illustration return "Response from model"
Model Selection Rules
python
class ModelSelector: """Intelligent model selection based on requirements""" def select_model(self, state: Dict[str, Any]) -> str: """Select optimal model based on requirements""" requirements = state.get("requirements", {}) # Simple queries → cost-efficient model if requirements.get("complexity") == "simple": return "gpt-4o-mini" # Complex reasoning → premium model if requirements.get("complexity") == "complex": if requirements.get("reasoning_depth", 0) > 0.8: return "gpt-4o" return "gpt-4o" # Long context → specialized model if requirements.get("context_length", 0) > 100000: return "claude-3-7-sonnet" # Code generation → Claude (known for code) if requirements.get("task_type") == "code": return "claude-3-7-sonnet" # Vision tasks → multimodal model if requirements.get("has_images"): return "gpt-4o" # Fast response required if requirements.get("latency_sensitive"): return "gpt-4o-mini" # Cost sensitive if requirements.get("cost_sensitive"): return "gpt-4o-mini" # Default return "gpt-4o"
Cost Enforcement Middleware
python
class CostEnforcementMiddleware: """Enforce budgets with automatic model downgrade""" def __init__(self): # Illustrative pricing - update from provider docs # Prices change frequently - always use current rates self.model_pricing = { "gpt-4o": {"input": 0.005, "output": 0.015}, # per 1K tokens "gpt-4o-mini": {"input": 0.00015, "output": 0.0006}, "claude-3-7-sonnet": {"input": 0.003, "output": 0.015}, "claude-3-haiku": {"input": 0.00025, "output": 0.00125}, "gemini-1.5-pro": {"input": 0.0025, "output": 0.010}, "gemini-1.5-flash": {"input": 0.0003, "output": 0.0012} } async def check_budget(self, state: Dict[str, Any]) -> Dict[str, Any]: """Check and enforce budget limits""" user_id = state.get("user_id") monthly_spent = await self.get_monthly_spend(user_id) monthly_budget = state.get("monthly_budget", 100.00) # $100 remaining = monthly_budget - monthly_spent if remaining <= 0: return { **state, "budget_exceeded": True, "monthly_spent": monthly_spent, "error": f"Monthly budget exceeded. Spent: ${monthly_spent:.2f}" } # Check request cost estimate estimated_cost = await self._estimate_cost(state) if estimated_cost > remaining: original_model = state.get("selected_model") downgraded_model = self._downgrade_model(original_model) return { **state, "selected_model": downgraded_model, "downgraded": True, "reason": f"Budget protection: {estimated_cost:.2f}$ > ${remaining:.2f}", "original_model": original_model } return {**state, "budget_checked": True} def _downgrade_model(self, model: str) -> str: """Intelligent model downgrade based on capabilities""" # Illustrative downgrades - update based on available models downgrades = { "gpt-4o": "gpt-4o-mini", "claude-3-7-sonnet": "claude-3-haiku", "gemini-1.5-pro": "gemini-1.5-flash" } return downgrades.get(model, "gpt-4o-mini")
Caching Strategy: Multiple Layers
Cache Layer Architecture
text
Request │ ├─ [1] Prompt Cache (API-level) │ └─ LLM-provided prompt caching │ (OpenAI, Anthropic, Google) │ ├─ [2] Response Cache (Application-level) │ └─ Full response caching │ (Redis, Memcached) │ ├─ [3] Embedding Cache │ └─ Cached embeddings from inputs │ (Pinecone, Qdrant) │ ├─ [4] Semantic Cache │ └─ Cache based on query similarity │ (Vector similarity > 0.95) │ ├─ [5] Tool Cache │ └─ Tool execution results │ (Database queries, API calls) │ └─ [6] RAG Cache └─ Retrieved documents (Vector DB lookups)
Implementation Strategy
python
import redisfrom typing import Dict, Any, Optionalimport hashlibimport jsonimport asyncio
class CachingStrategy: """Multi-layer caching with proper TTLs and invalidation""" def __init__(self): # Production: Use Redis with proper connection pooling self.redis_client = redis.Redis( host='localhost', port=6379, decode_responses=True, connection_pool=redis.ConnectionPool(max_connections=50) ) self.ttl_seconds = { "response": 3600, # 1 hour "embedding": 86400, # 24 hours "semantic": 7200, # 2 hours "tool": 300, # 5 minutes "rag": 3600 # 1 hour } async def get_cached_response(self, state: Dict[str, Any]) -> Dict[str, Any]: """Try each cache layer""" query = state.get("input", "") cache_key = self._generate_cache_key(query) # Layer 1: Check response cache cached_response = await self._get_response_cache(cache_key) if cached_response: state["cache_hit"] = "response_cache" state["output"] = cached_response return state # Layer 2: Check semantic cache (similarity) semantic_response = await self._get_semantic_cache(query) if semantic_response: state["cache_hit"] = "semantic_cache" state["output"] = semantic_response return state # No cache hit - proceed return state async def _get_response_cache(self, key: str) -> Optional[str]: """Get from Redis response cache""" # Production: Check if cache is stale # Invalidation strategy: Version or TTL cached = self.redis_client.get(f"response:{key}") if cached: # Check if still valid (TTL handled by Redis) return json.loads(cached) return None async def _get_semantic_cache(self, query: str) -> Optional[str]: """Semantic cache using vector similarity""" # Production: Use vector DB (Pinecone, Qdrant, Weaviate) # Check if similar query exists with score > 0.95 # Simplified for illustration return None def _generate_cache_key(self, query: str) -> str: """Generate cache key from query""" return hashlib.md5(query.encode()).hexdigest() def _invalidate_cache(self, pattern: str): """Invalidate cache by pattern""" # Production: Use Redis SCAN for pattern deletion # Important for cache invalidation pass
Cache Selection:
- Prompt Cache: Use when available (reduce input tokens)
- Response Cache: Use for repeated exact queries
- Semantic Cache: Use for similar queries (fuzzy matching)
- Embedding Cache: Use for RAG systems
- Tool Cache: Use for deterministic tools
Memory Systems: LangGraph Store API
Comprehensive Memory Management
Note: LangGraph Store APIs have evolved across releases. Verify against your target LangGraph version. The following is illustrative for current patterns.
python
python
from langgraph.store.postgres import PostgresStorefrom typing import Dict, Any, Listimport datetime
class MemorySystem: """Comprehensive memory management with LangGraph Store""" def __init__(self): # Production: Use PostgresStore, not InMemoryStore # Verify exact API for your LangGraph version self.store = PostgresStore.from_conn_string( "postgresql://user:pass@localhost/db" ) async def retrieve_memory(self, state: Dict[str, Any]) -> Dict[str, Any]: """Retrieve all memory types using LangGraph Store""" user_id = state.get("user_id") current_query = state.get("input", "") # 1. Short-term memory (conversation history) conversation = await self.get_conversation_memory( user_id, last_n=10, ttl_minutes=30 ) # 2. Long-term memory (user profile) profile = await self.get_profile_memory(user_id) # 3. Semantic memory (knowledge graph) relevant_facts = await self.get_semantic_memory( user_id, query=current_query, similarity_threshold=0.8 ) # 4. Episodic memory (significant events) recent_events = await self.get_episodic_memory( user_id, last_days=7 ) # 5. Working memory (current context) working = state.get("working_memory", {}) # 6. Compose context context = self._compose_context( profile=profile, conversation=conversation, relevant_facts=relevant_facts, recent_events=recent_events, current_query=current_query ) return { **state, "injected_context": context, "memory_sources": { "conversation": len(conversation), "profile": bool(profile), "semantic": len(relevant_facts), "episodic": len(recent_events) } } async def get_conversation_memory(self, user_id: str, last_n: int, ttl_minutes: int): """Get short-term conversation memory""" # Production: Use LangGraph Store with TTL key = f"conversation:{user_id}" # API may vary by version - check documentation memories = self.store.search(key) return memories[-last_n:] if memories else [] async def get_profile_memory(self, user_id: str): """Get long-term user profile""" # Production: Use LangGraph Store with persistence key = f"profile:{user_id}" return self.store.get(key) async def get_semantic_memory(self, user_id: str, query: str, similarity_threshold: float): """Get semantic memory via vector similarity""" # Production: Use vector DB with proper indexing return [] async def get_episodic_memory(self, user_id: str, last_days: int): """Get episodic memory (significant events)""" # Production: Use time-series store return [] def _compose_context(self, **kwargs) -> str: """Compose memory context for prompt injection""" # Format all memory types into context string context_parts = [] if kwargs.get("profile"): context_parts.append(f"USER PROFILE:\n{kwargs['profile']}") if kwargs.get("conversation"): context_parts.append(f"CONVERSATION HISTORY:\n{kwargs['conversation']}") if kwargs.get("relevant_facts"): context_parts.append(f"RELEVANT FACTS:\n{kwargs['relevant_facts']}") if kwargs.get("recent_events"): context_parts.append(f"RECENT EVENTS:\n{kwargs['recent_events']}") context_parts.append(f"CURRENT TASK:\n{kwargs.get('current_query', '')}") return "\n\n".join(context_parts)
Observability: Production Monitoring
OpenTelemetry with LangSmith Integration
Note: OpenTelemetry instrumentation package names may vary by release. The following is illustrative for current patterns. Verify against your OpenTelemetry and LangChain versions.
python
python
from opentelemetry import tracefrom opentelemetry.instrumentation.langchain import LangChainInstrumentorfrom langsmith import Client as LangSmithClientimport loggingimport jsonimport uuidimport asyncio
logger = logging.getLogger(__name__)
class ObservabilityMiddleware: """Production observability with OpenTelemetry and LangSmith""" def __init__(self): # Setup OpenTelemetry self.tracer = trace.get_tracer(__name__) # LangChainInstrumentor may require specific version # Verify against your installation LangChainInstrumentor().instrument() # Setup LangSmith for production self.langsmith = LangSmithClient() # Setup structured logging self.logger = logging.getLogger(__name__) self.logger.setLevel(logging.INFO) # JSON formatter for structured logs formatter = logging.Formatter( '{"timestamp": "%(asctime)s", "level": "%(levelname)s", "name": "%(name)s", "message": "%(message)s"}' ) handler = logging.StreamHandler() handler.setFormatter(formatter) self.logger.addHandler(handler) async def trace_agent_execution(self, state: Dict[str, Any]) -> Dict[str, Any]: """Trace agent execution with OpenTelemetry""" # Generate correlation ID correlation_id = str(uuid.uuid4()) with self.tracer.start_as_current_span("agent_execution") as span: # Add attributes span.set_attribute("user_id", state.get("user_id", "unknown")) span.set_attribute("model", state.get("selected_model", "unknown")) span.set_attribute("agent_type", state.get("agent_type", "unknown")) span.set_attribute("correlation_id", correlation_id) span.set_attribute("span_id", span.get_span_context().span_id) span.set_attribute("trace_id", span.get_span_context().trace_id) # Add to state for downstream state["correlation_id"] = correlation_id state["trace_id"] = str(span.get_span_context().trace_id) state["span_id"] = str(span.get_span_context().span_id) # Log to LangSmith await self._log_to_langsmith(state) # Structured logging self.logger.info( json.dumps({ "event": "agent_execution_started", "correlation_id": correlation_id, "trace_id": str(span.get_span_context().trace_id), "user_id": state.get("user_id"), "model": state.get("selected_model") }) ) return state async def _log_to_langsmith(self, state: Dict[str, Any]): """Log to LangSmith for agent observability""" try: # LangSmith API may vary - verify against current SDK self.langsmith.create_run( name="agent_execution", run_type="chain", inputs={"state": state}, metadata={ "user_id": state.get("user_id"), "correlation_id": state.get("correlation_id"), "trace_id": state.get("trace_id") } ) except Exception as e: logger.error(f"Failed to log to LangSmith: {e}")
Recommended Tools:
- OpenTelemetry: Distributed tracing
- LangSmith: Agent observability
- Langfuse: Open-source LLM observability
- Phoenix: LLM observability
- Weights & Biases: Experiment tracking
- Braintrust: Evaluation and monitoring
- Prometheus: Metrics collection
Retry and Circuit Breaker
Production Retry with Exponential Backoff + Jitter
python
import asyncioimport randomfrom tenacity import retry, stop_after_attempt, wait_exponential
class RetryMiddleware: """Production retry with exponential backoff + jitter""" @retry( stop=stop_after_attempt(5), wait=wait_exponential(multiplier=1, min=2, max=60), reraise=True ) async def call_with_retry(self, func, *args, **kwargs): """Call function with exponential backoff + jitter""" # tenacity handles exponential backoff # Add jitter manually for production jitter = random.uniform(0, 1) * 0.5 await asyncio.sleep(jitter) # Add jitter return await func(*args, **kwargs)
Circuit Breaker with PyBreaker
python
from pybreaker import CircuitBreaker, CircuitBreakerErrorimport time
class CircuitBreakerMiddleware: """Production circuit breaker with proper thresholds""" def __init__(self): # Create a circuit breaker instance self.breaker = CircuitBreaker( fail_max=5, # Max failures before opening reset_timeout=60, # Time before transitioning to half-open half_open_timeout=30, # Time in half-open state half_open_success_threshold=3, # Successful requests to close exclude=[TimeoutError] # Don't count timeouts as failures ) async def call_model(self, model: str, state: Dict[str, Any]) -> str: """Call model with circuit breaker protection""" # Use the instance's call method, not a decorator on the method try: response = await self.breaker.call( self._call_model_impl, model, state ) return response except CircuitBreakerError: # Circuit is open - use fallback return await self._handle_open_circuit(model, state) async def _call_model_impl(self, model: str, state: Dict[str, Any]) -> str: """Actual model call implementation""" # Simplified for illustration return "Response from model" async def _handle_open_circuit(self, model: str, state: Dict[str, Any]) -> str: """Handle open circuit with fallback""" # Circuit is open - use fallback model fallback_model = self._get_fallback_model() return await self._call_model_impl(fallback_model, state)
Production Storage
Enterprise Checkpoint Storage
python
from langgraph.checkpoint.postgres import PostgresSaverfrom langgraph.checkpoint.redis import RedisSaverfrom langgraph.checkpoint.sqlite import SqliteSaver
Prompt Versioning with Enterprise Tools
python
class PromptVersioning: """Enterprise prompt management with versioning tools""" async def get_prompt(self, state: Dict[str, Any]) -> Dict[str, Any]: """Retrieve prompt version from enterprise management system""" prompt_id = state.get("prompt_id", "default_prompt") # Option 1: LangSmith Prompt Hub # from langsmith import Client # ls_client = Client() # API may vary - check current LangSmith documentation # prompt = ls_client.get_prompt(prompt_id) # Option 2: Langfuse # from langfuse import Langfuse # langfuse = Langfuse() # prompt = langfuse.get_prompt(prompt_id) # Option 3: PromptLayer # from promptlayer import PromptLayer # pl = PromptLayer() # prompt = pl.get_prompt(prompt_id) # Apply version from config version = state.get("prompt_version", "production") # Illustrative return return { **state, "prompt": f"Prompt content for {prompt_id} version {version}", "prompt_version_used": version, "prompt_id": prompt_id }
Recommended Tools:
- LangSmith Prompt Hub: LangChain’s native prompt management
- Langfuse: Open-source prompt management with observability
- PromptLayer: Enterprise prompt management
- Promptfoo: Prompt testing and evaluation
- Braintrust: AI evaluation platform
Middleware Ordering: Critical Pitfalls
Correct Middleware Ordering
python
import asyncio
php
async def correct_middleware_pipeline(state: Dict[str, Any]) -> Dict[str, Any]: """Proper middleware ordering for production""" # 1. Authentication FIRST state = await validate_auth(state) if state.get("auth_failed"): return state # 2. Rate limiting SECOND state = await check_rate_limit(state) if state.get("rate_limited"): return state # 3. Security checks BEFORE transformation state = await detect_secrets(state) state = await detect_injection(state) state = await moderate_content(state) # 4. Cache check BEFORE LLM call state = await check_cache(state) # 5. Memory retrieval BEFORE prompt construction state = await retrieve_memory(state) # 6. Prompt transformation AFTER security checks state = await transform_prompt(state) # 7. Budget check BEFORE LLM call state = await check_budget(state) # 8. Observability context state = await setup_tracing(state) return state
Common Pitfalls
❌ WRONG: Cache before AuthenticationProblem: Caches unauthenticated requests
❌ WRONG: PII Detection before Prompt InjectionProblem: Injection pattern might be missed
❌ WRONG: Budget check after LLM callProblem: Can't prevent overspending
✅ CORRECT: Authentication → CacheProblem: Only cache authenticated requests
✅ CORRECT: Prompt Injection → PII DetectionProblem: Block injection before processing content
✅ CORRECT: Budget check BEFORE LLM callProblem: Prevent overspending proactively
Production Checklist
Core Architecture
- Use interrupt() for HITL, not custom routing
- Use Command(resume=...) for resumption
- Distinguish middleware from infrastructure
- Use production libraries (tenacity, pybreaker)
- Implement proper cache invalidation
- Use structured JSON logging
- Implement distributed tracing
Security
- Implement prompt injection detection (ML + regex)
- Implement jailbreak detection
- Implement secret detection/redaction
- Implement content moderation
- Policy enforcement with OPA/Cedar
- Budget enforcement
- PII redaction
- State validation with Pydantic
Observability
- OpenTelemetry traces with correlation IDs
- LangSmith run tracking
- Langfuse/Phoenix observability
- Prometheus metrics
- Structured logging
- Error categorization
Performance
- Multiple cache layers with TTL
- Async middleware pipeline
- Connection pooling
- Model fallover configured
- Timeout handling
- Proper middleware ordering
Governance
- Prompt versioning with enterprise tools
- A/B testing framework
- Compliance auditing
- Security incident response plan
- Cost tracking and reporting
References & Tools
Core Frameworks
Observability
Security & Guardrails
Infrastructure
Libraries
Summary
These patterns provide a strong foundation for building enterprise AI agent systems. Adapt the architecture to your organization’s security, compliance, scalability, and operational requirements.
Key architectural decisions:
Correctness — Use modern patterns (interrupt() with Command(resume=...)) Distinction — Middleware ≠ Infrastructure ≠ OrchestrationCompleteness — Multiple cache layers, guardrails, security, streamingLibraries — Use proven solutions (tenacity, pybreaker, litellm) Observability — OpenTelemetry, LangSmith, structured loggingPerformance — Async pipelines, concurrent operations, proper ordering Security — Comprehensive protection at all layersGovernance — Policy enforcement, auditing, complianceStorage — Production databases (Postgres, Redis), not MemorySaver
Remember:
- All code examples are illustrative and require production hardening
- API names and availability change frequently — verify against current documentation
- Security requires multiple layers (regex + ML + policies)
- Monitor both technical and business metrics
- Plan for fallback and failure modes
Master these patterns to build AI systems that are reliable, observable, production-ready, and enterprise-compliant.
Building Production-Ready AI Agent Middleware with LangGraph: Enterprise Patterns, Guardrails… was originally published in Towards AI on Medium, where people are continuing the conversation by highlighting and responding to this story.
source & further reading
pub.towardsai.net — original article
Kimi K3 beats Opus 4.8 but costs the same as Sonnet 5: The End of the "Open Equals Cheap" Era
Speculative Decoding Collapses at Batch 224 on Llama-3-70B and Never on gpt-oss-120B
The New Rules of Context Engineering for Claude 5