Building Production-Ready AI Agent Middleware with LangGraph: Enterprise Patterns, Guardrails…

LangGraph, a framework by LangChain, is the focus of a new enterprise guide detailing production-ready AI agent middleware patterns, including human-in-the-loop interrupts, LLM gateways, security middleware, and observability. The guide covers architectural layers from client to tool execution, emphasizing control, security, and performance for enterprise AI systems.

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 python from langgraph.types import interruptfrom langgraph.graph import StateGraph, START, END, Commandfrom typing import Dict, Any, Literalimport datetime NOTE: This example illustrates the core interrupt pattern. Production code requires: error handling, logging, monitoring, configuration, authentication, testing, and complete type hints. php def tool approval node state: Dict str, Any - Command Literal "execute tool", "reject tool" : """Pause 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 pauses 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 Graph constructiongraph = StateGraph AgentState graph.add node "tool approval", tool approval node graph.add edge START, "tool approval" graph.add edge "execute tool", END graph.add edge "reject tool", END python Production usage with checkpointingfrom langgraph.checkpoint.postgres import PostgresSaver Use PostgresSaver in production, not MemorySaverwith PostgresSaver.from conn string "postgresql://user:pass@localhost/db" as checkpointer: graph = graph.compile checkpointer=checkpointer Execute with human-in-the-loopdef execute with approval initial state, thread id="user session" : """Production HITL execution""" First run - graph will pause at interrupt result = graph.invoke initial state, config={"configurable": {"thread id": thread id}} Check if graph paused at interrupt Human approves or rejects based on interrupt data human decision = {"approved": True, "reason": "Looks safe"} Resume with human decision using Command resume=... final result = graph.invoke Command resume=human decision , config={"configurable": {"thread id": thread id}} return final result 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 python import refrom typing import Dict, Any NOTE: These regex patterns are illustrative and should be supplemented with classifier-based detection in production. Regular expressions alone are not sufficient for comprehensive prompt injection detection. 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 Llama Guard typically used through Hugging Face Transformers This is illustrative - adapt to your deployment method from transformers import AutoModelForSequenceClassification, AutoTokenizer NeMo Guardrails is configured through rails definitions This is illustrative - see NeMo Guardrails documentation for correct API 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 NOTE: Model availability and exact identifiers vary by provider and over time. The following model names are illustrative examples. Always check current provider documentation for available models and exact identifiers. See: https://docs.litellm.ai/docs/providers 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 NOTE: All pricing figures are illustrative only. Always use current pricing from provider documentation: https://openai.com/pricing https://www.anthropic.com/pricing https://cloud.google.com/vertex-ai/pricing 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 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 NOTE: Store APIs may vary by LangGraph version. Verify against your target version documentation. https://langchain-ai.github.io/langgraph/reference/store/ 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 NOTE: OpenTelemetry instrumentation may require specific package versions. Check documentation for your LangChain and OpenTelemetry versions: https://opentelemetry.io/docs/instrumentation/python/ https://docs.smith.langchain.com/ 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 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 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 python from langgraph.checkpoint.postgres import PostgresSaverfrom langgraph.checkpoint.redis import RedisSaverfrom langgraph.checkpoint.sqlite import SqliteSaver Production options for checkpoints Option 1: PostgreSQL Recommended for most enterprises with PostgresSaver.from conn string "postgresql://user:pass@localhost:5432/langgraph" as checkpointer: graph = graph.compile checkpointer=checkpointer Option 2: Redis For high-throughput/low-latency checkpointer = RedisSaver.from conn string "redis://localhost:6379/0" graph = graph.compile checkpointer=checkpointer Option 3: SQLite For smaller deployments checkpointer = SqliteSaver.from conn string "sqlite:///langgraph.db" graph = graph.compile checkpointer=checkpointer ⚠️ NOT FOR PRODUCTION: MemorySaver is for testing only from langgraph.checkpoint.memory import MemorySaver checkpointer = MemorySaver Prompt Versioning with Enterprise Tools python NOTE: Prompt management APIs may vary by tool version. The following is illustrative - verify against current SDK documentation. LangSmith Prompt Hub: https://docs.smith.langchain.com/ Langfuse: https://langfuse.com/docs/prompts PromptLayer: https://docs.promptlayer.com/ 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 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 ≠ Orchestration Completeness — Multiple cache layers, guardrails, security, streaming Libraries — Use proven solutions tenacity, pybreaker, litellm Observability — OpenTelemetry, LangSmith, structured logging Performance — Async pipelines, concurrent operations, proper ordering Security — Comprehensive protection at all layers Governance — Policy enforcement, auditing, compliance Storage — 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… https://pub.towardsai.net/building-production-ready-ai-agent-middleware-with-langgraph-enterprise-patterns-guardrails-2ed7ba5a8486 was originally published in Towards AI https://pub.towardsai.net on Medium, where people are continuing the conversation by highlighting and responding to this story.