# Building Production-Ready AI Agent Middleware with LangGraph: Enterprise Patterns, Guardrails… > Source: > Published: 2026-08-01 16:36:19+00:00 ### 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.