Using LangChain & Tooling for Real Apps: Practical Recipes

Building production-ready AI applications requires more than just connecting a language model to your code. It demands robust architectures, proper error handling, cost management, and maintainable patterns. This guide provides practical recipes for using LangChain to build real applications that work reliably in production environments.

LangChain has emerged as the de facto framework for orchestrating language model workflows, but many tutorials stop at basic examples. Here, we dive into practical patterns you can adapt for customer support systems, document processing pipelines, content generation tools, and internal automation workflows.

Why LangChain for Real Applications?

Before we dive into recipes, let's understand why LangChain has become essential for production AI applications. LangChain provides a standardized way to:

• Connect multiple LLM providers (OpenAI, Anthropic, local models)
• Manage conversation memory across sessions
• Integrate external tools and APIs
• Process documents and handle retrieval-augmented generation (RAG)
• Chain multiple operations with error recovery

The framework's modular design allows you to swap components as requirements change, making your applications more maintainable and future-proof.

Recipe 1: Document Processing Pipeline with Error Recovery

One of the most common use cases for LangChain is processing documents for question-answering systems. Here's a production-ready implementation:

Core Components

A robust document pipeline needs:

• Multiple document loader options (PDF, DOCX, TXT, HTML)
• Intelligent text splitting with overlap
• Embedding generation with retry logic
• Vector storage with proper indexing
• Query processing with fallback mechanisms

Implementation Pattern

Here's a pattern that includes error handling and monitoring:

from langchain.document_loaders import PyPDFLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI
import logging
from tenacity import retry, stop_after_attempt, wait_exponential

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class DocumentProcessor:
    def __init__(self, persist_directory="./chroma_db"):
        self.embeddings = OpenAIEmbeddings(
            model="text-embedding-ada-002",
            max_retries=3,
            request_timeout=30
        )
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200,
            length_function=len,
            separators=["\n\n", "\n", " ", ""]
        )
        self.persist_directory = persist_directory
        
    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
    def process_document(self, file_path, doc_type="pdf"):
        """Process document with retry logic and error handling"""
        try:
            # Load document based on type
            if doc_type == "pdf":
                loader = PyPDFLoader(file_path)
            elif doc_type == "txt":
                loader = TextLoader(file_path)
            else:
                raise ValueError(f"Unsupported document type: {doc_type}")
            
            documents = loader.load()
            logger.info(f"Loaded {len(documents)} pages from {file_path}")
            
            # Split into chunks
            texts = self.text_splitter.split_documents(documents)
            logger.info(f"Split into {len(texts)} text chunks")
            
            # Create vector store
            vectordb = Chroma.from_documents(
                documents=texts,
                embedding=self.embeddings,
                persist_directory=self.persist_directory
            )
            vectordb.persist()
            
            return vectordb
            
        except Exception as e:
            logger.error(f"Error processing document {file_path}: {str(e)}")
            # Implement fallback strategy
            return self._fallback_processing(file_path)
    
    def _fallback_processing(self, file_path):
        """Fallback processing for when primary method fails"""
        # Implement simpler processing or return cached results
        pass

This pattern includes several production considerations:

• Retry logic with exponential backoff for API calls
• Comprehensive logging for debugging
• Fallback mechanisms for graceful degradation
• Configurable chunking parameters for different document types

Advanced Error Handling

For production systems, consider these additional error handling strategies:

class ResilientDocumentProcessor(DocumentProcessor):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.failed_documents = []
        self.processing_stats = {
            "successful": 0,
            "failed": 0,
            "retried": 0
        }
    
    def batch_process(self, file_paths, max_workers=4):
        """Process multiple documents with parallel execution"""
        from concurrent.futures import ThreadPoolExecutor, as_completed
        
        results = {}
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            future_to_file = {
                executor.submit(self.process_document, fp): fp 
                for fp in file_paths
            }
            
            for future in as_completed(future_to_file):
                file_path = future_to_file[future]
                try:
                    result = future.result(timeout=300)
                    results[file_path] = result
                    self.processing_stats["successful"] += 1
                except Exception as e:
                    logger.error(f"Failed to process {file_path}: {e}")
                    self.failed_documents.append(file_path)
                    self.processing_stats["failed"] += 1
                    # Optionally implement retry with different parameters
                    if self.processing_stats["failed"] < len(file_paths) * 0.1:
                        self._schedule_retry(file_path)
        
        return results

Recipe 2: Intelligent Chatbot with Memory and Tools

Building a chatbot that remembers conversations and can use external tools requires careful design. Here's a production-ready pattern:

Memory Management

Proper memory management is crucial for maintaining context across conversations:

from langchain.memory import ConversationBufferMemory, RedisChatMessageHistory
from langchain.chains import ConversationChain
from langchain.agents import initialize_agent, AgentType
from langchain.tools import Tool
import redis

class ChatbotWithMemory:
    def __init__(self, session_id, redis_url="redis://localhost:6379"):
        # Initialize memory with Redis backend for persistence
        self.redis_client = redis.from_url(redis_url)
        self.message_history = RedisChatMessageHistory(
            url=redis_url,
            session_id=session_id,
            ttl=3600  # 1 hour TTL
        )
        
        self.memory = ConversationBufferMemory(
            memory_key="chat_history",
            chat_memory=self.message_history,
            return_messages=True
        )
        
        # Initialize LLM with proper configuration
        self.llm = ChatOpenAI(
            temperature=0.7,
            model_name="gpt-4",
            max_tokens=1000,
            request_timeout=30
        )
        
        # Define available tools
        self.tools = self._initialize_tools()
        
    def _initialize_tools(self):
        """Define and initialize external tools"""
        tools = [
            Tool(
                name="Search",
                func=self._search_tool,
                description="Useful for searching current information"
            ),
            Tool(
                name="Calculator",
                func=self._calculator_tool,
                description="Useful for mathematical calculations"
            ),
            Tool(
                name="DatabaseQuery",
                func=self._database_query_tool,
                description="Useful for querying product database"
            )
        ]
        return tools
    
    def create_agent(self):
        """Create conversational agent with tools and memory"""
        agent = initialize_agent(
            tools=self.tools,
            llm=self.llm,
            agent=AgentType.CONVERSATIONAL_REACT_DESCRIPTION,
            memory=self.memory,
            verbose=True,
            max_iterations=5,
            early_stopping_method="generate",
            handle_parsing_errors=True
        )
        return agent
    
    def process_message(self, user_input, agent):
        """Process message with error handling and context management"""
        try:
            # Add context if needed
            context = self._get_relevant_context(user_input)
            if context:
                user_input = f"Context: {context}\n\nUser: {user_input}"
            
            # Process through agent
            response = agent.run(user_input)
            
            # Update conversation summary for long conversations
            if len(self.message_history.messages) > 10:
                self._summarize_conversation()
            
            return response
            
        except Exception as e:
            logger.error(f"Error processing message: {e}")
            return self._get_fallback_response(user_input)

Tool Implementation Patterns

Tools extend your chatbot's capabilities. Here are patterns for robust tool implementation:

class ProductionTools:
    @staticmethod
    def _search_tool(query: str) -> str:
        """Search tool with caching and rate limiting"""
        # Implement caching to reduce API calls
        cache_key = f"search:{hash(query)}"
        cached = redis_client.get(cache_key)
        if cached:
            return cached.decode('utf-8')
        
        # Implement actual search with rate limiting
        try:
            results = search_api(query, limit=5)
            # Cache results for 5 minutes
            redis_client.setex(cache_key, 300, json.dumps(results))
            return json.dumps(results)
        except RateLimitError:
            return "Search rate limit exceeded. Please try again in a moment."
    
    @staticmethod
    def _database_query_tool(query_params: dict) -> str:
        """Database query with validation and sanitization"""
        # Validate input parameters
        required_fields = ['table', 'columns', 'filters']
        if not all(field in query_params for field in required_fields):
            return "Missing required query parameters"
        
        # Sanitize inputs to prevent SQL injection
        sanitized = {
            'table': sanitize_table_name(query_params['table']),
            'columns': [sanitize_column_name(col) for col in query_params['columns']],
            'filters': sanitize_filters(query_params['filters'])
        }
        
        # Execute query with timeout
        try:
            result = execute_safe_query(sanitized, timeout=10)
            return format_query_results(result)
        except TimeoutError:
            return "Query timed out. Please try a more specific query."
        except DatabaseError as e:
            logger.error(f"Database error: {e}")
            return "Unable to query database at this time."

Recipe 3: Workflow Automation with Conditional Logic

LangChain excels at orchestrating complex workflows with conditional branching. Here's a pattern for business process automation:

from langchain.chains import LLMChain, SequentialChain, TransformChain
from langchain.prompts import PromptTemplate
from langchain.callbacks import get_openai_callback

class WorkflowAutomation:
    def __init__(self):
        self.llm = ChatOpenAI(temperature=0.3, model_name="gpt-3.5-turbo")
        
    def create_customer_support_workflow(self):
        """Create a multi-step customer support workflow"""
        
        # Step 1: Classify query type
        classification_prompt = PromptTemplate(
            input_variables=["user_query"],
            template="""
            Classify the following customer query into one of these categories:
            - billing
            - technical
            - account
            - general
            
            Query: {user_query}
            
            Return only the category name.
            """
        )
        
        classification_chain = LLMChain(
            llm=self.llm,
            prompt=classification_prompt,
            output_key="query_type"
        )
        
        # Step 2: Route to appropriate handler based on classification
        def routing_function(inputs):
            query_type = inputs["query_type"].lower().strip()
            if query_type == "billing":
                return {"next_step": "billing_team"}
            elif query_type == "technical":
                return {"next_step": "technical_support"}
            elif query_type == "account":
                return {"next_step": "account_management"}
            else:
                return {"next_step": "general_support"}
        
        routing_chain = TransformChain(
            input_variables=["query_type"],
            output_variables=["next_step"],
            transform=routing_function
        )
        
        # Step 3: Generate appropriate response
        response_prompt = PromptTemplate(
            input_variables=["user_query", "next_step"],
            template="""
            Based on the query type ({next_step}), generate an appropriate response.
            
            Query: {user_query}
            
            Response should be helpful and direct the user to the right team if needed.
            """
        )
        
        response_chain = LLMChain(
            llm=self.llm,
            prompt=response_prompt,
            output_key="final_response"
        )
        
        # Combine into sequential chain
        overall_chain = SequentialChain(
            chains=[classification_chain, routing_chain, response_chain],
            input_variables=["user_query"],
            output_variables=["query_type", "next_step", "final_response"],
            verbose=True
        )
        
        return overall_chain
    
    def execute_with_monitoring(self, chain, user_query):
        """Execute chain with cost monitoring and performance tracking"""
        with get_openai_callback() as cb:
            start_time = time.time()
            
            try:
                result = chain({"user_query": user_query})
                
                execution_time = time.time() - start_time
                
                # Log performance metrics
                self._log_metrics({
                    "execution_time": execution_time,
                    "total_tokens": cb.total_tokens,
                    "total_cost": cb.total_cost,
                    "success": True
                })
                
                return result
                
            except Exception as e:
                execution_time = time.time() - start_time
                
                self._log_metrics({
                    "execution_time": execution_time,
                    "total_tokens": cb.total_tokens if 'cb' in locals() else 0,
                    "total_cost": cb.total_cost if 'cb' in locals() else 0,
                    "success": False,
                    "error": str(e)
                })
                
                raise

Conditional Workflow Patterns

Advanced workflows often need conditional execution paths:

class ConditionalWorkflow:
    def create_dynamic_workflow(self):
        """Create workflow with dynamic path selection"""
        
        # Decision chain to determine workflow path
        decision_prompt = PromptTemplate(
            input_variables=["input_data", "business_rules"],
            template="""Analyze the input and business rules to decide which workflow path to take.
            Input: {input_data}
            Rules: {business_rules}
            
            Return: 'path_a', 'path_b', or 'path_c'
            """
        )
        
        decision_chain = LLMChain(
            llm=self.llm,
            prompt=decision_prompt,
            output_key="selected_path"
        )
        
        # Define different paths
        def execute_path_a(inputs):
            # Path A implementation
            return {"result": "Path A executed", "status": "complete"}
        
        def execute_path_b(inputs):
            # Path B implementation
            return {"result": "Path B executed", "status": "complete"}
        
        def execute_path_c(inputs):
            # Path C implementation
            return {"result": "Path C executed", "status": "complete"}
        
        # Route to appropriate path
        def route_execution(inputs):
            selected_path = inputs["selected_path"]
            
            if selected_path == "path_a":
                return execute_path_a(inputs)
            elif selected_path == "path_b":
                return execute_path_b(inputs)
            elif selected_path == "path_c":
                return execute_path_c(inputs)
            else:
                return {"result": "Default path executed", "status": "complete"}
        
        routing_chain = TransformChain(
            input_variables=["selected_path", "input_data"],
            output_variables=["result", "status"],
            transform=route_execution
        )
        
        return SequentialChain(
            chains=[decision_chain, routing_chain],
            input_variables=["input_data", "business_rules"],
            output_variables=["selected_path", "result", "status"]
        )

Recipe 4: Cost Optimization and Performance Tuning

Production applications need careful cost management and performance optimization. Here are practical strategies:

Token Usage Optimization

class CostOptimizer:
    def __init__(self):
        self.token_usage = {}
        self.cost_tracker = {}
        
    def optimize_prompt(self, prompt, max_tokens=500):
        """Optimize prompt to reduce token usage"""
        optimization_rules = {
            "remove_redundancy": True,
            "shorten_examples": True,
            "use_abbreviations": False,
            "max_examples": 3
        }
        
        optimized = self._apply_optimization_rules(prompt, optimization_rules)
        
        # Estimate token count
        token_count = self._estimate_tokens(optimized)
        
        if token_count > max_tokens:
            optimized = self._truncate_prompt(optimized, max_tokens)
        
        return optimized
    
    def select_model_based_on_complexity(self, task_complexity, budget_constraints):
        """Dynamically select model based on task requirements"""
        model_selection_rules = [
            {
                "complexity": "low",
                "models": ["gpt-3.5-turbo", "claude-instant"],
                "max_tokens": 500
            },
            {
                "complexity": "medium",
                "models": ["gpt-3.5-turbo-16k", "claude-2"],
                "max_tokens": 2000
            },
            {
                "complexity": "high",
                "models": ["gpt-4", "claude-2-100k"],
                "max_tokens": 4000
            }
        ]
        
        for rule in model_selection_rules:
            if task_complexity == rule["complexity"]:
                # Consider cost if budget is constrained
                if budget_constraints == "strict":
                    return rule["models"][0], rule["max_tokens"]
                else:
                    return rule["models"][-1], rule["max_tokens"]
        
        return "gpt-3.5-turbo", 1000
    
    def implement_caching_layer(self, query, ttl=300):
        """Implement caching to reduce repeated API calls"""
        cache_key = self._generate_cache_key(query)
        
        # Check cache
        cached_response = cache_store.get(cache_key)
        if cached_response:
            self.token_usage["cache_hits"] = self.token_usage.get("cache_hits", 0) + 1
            return cached_response
        
        # Execute and cache
        response = self._execute_query(query)
        cache_store.setex(cache_key, ttl, response)
        
        return response

Batch Processing for Efficiency

class BatchProcessor:
    def __init__(self, batch_size=10, max_concurrent=5):
        self.batch_size = batch_size
        self.max_concurrent = max_concurrent
        
    def process_batch(self, items, process_function):
        """Process items in batches for efficiency"""
        results = []
        
        for i in range(0, len(items), self.batch_size):
            batch = items[i:i + self.batch_size]
            
            # Process batch with rate limiting
            batch_result = self._process_with_limits(
                batch, 
                process_function,
                self.max_concurrent
            )
            
            results.extend(batch_result)
            
            # Add delay between batches if needed
            if i + self.batch_size < len(items):
                time.sleep(0.1)  # Prevent rate limiting
        
        return results
    
    def _process_with_limits(self, batch, process_function, max_concurrent):
        """Process with concurrency limits"""
        from concurrent.futures import ThreadPoolExecutor
        
        with ThreadPoolExecutor(max_workers=max_concurrent) as executor:
            futures = [executor.submit(process_function, item) for item in batch]
            
            results = []
            for future in futures:
                try:
                    result = future.result(timeout=30)
                    results.append(result)
                except Exception as e:
                    logger.error(f"Error in batch processing: {e}")
                    results.append(None)  # Or implement retry logic
        
        return results

Recipe 5: Monitoring, Logging, and Alerting

Production applications need proper observability. Here's how to implement monitoring for LangChain applications:

class LangChainMonitor:
    def __init__(self, application_name="langchain_app"):
        self.application_name = application_name
        self.metrics = {
            "request_count": 0,
            "success_count": 0,
            "error_count": 0,
            "total_tokens": 0,
            "total_cost": 0.0
        }
        
    def setup_monitoring(self):
        """Setup comprehensive monitoring"""
        # Initialize logging
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('langchain_app.log'),
                logging.StreamHandler()
            ]
        )
        
        # Initialize metrics collection
        self._initialize_metrics_export()
        
        # Setup alerting rules
        self._setup_alerting()
    
    def track_execution(self, chain_name, start_time, end_time, 
                       token_usage=None, cost=None, success=True):
        """Track execution of a chain"""
        execution_time = end_time - start_time
        
        self.metrics["request_count"] += 1
        
        if success:
            self.metrics["success_count"] += 1
        else:
            self.metrics["error_count"] += 1
        
        if token_usage:
            self.metrics["total_tokens"] += token_usage
        
        if cost:
            self.metrics["total_cost"] += cost
        
        # Log detailed metrics
        self._log_metrics({
            "chain_name": chain_name,
            "execution_time": execution_time,
            "token_usage": token_usage,
            "cost": cost,
            "success": success,
            "timestamp": datetime.now().isoformat()
        })
        
        # Check for alert conditions
        self._check_alerts()
    
    def _setup_alerting(self):
        """Setup alerting rules"""
        self.alert_rules = {
            "error_rate": {
                "threshold": 0.05,  # 5% error rate
                "window_minutes": 15,
                "alert_channel": "slack"
            },
            "high_cost": {
                "threshold": 100.0,  # $100 per hour
                "window_minutes": 60,
                "alert_channel": "email"
            },
            "slow_response": {
                "threshold": 10.0,  # 10 seconds
                "window_minutes": 5,
                "alert_channel": "pagerduty"
            }
        }
    
    def create_custom_callbacks(self):
        """Create custom callbacks for detailed monitoring"""
        from langchain.callbacks.base import BaseCallbackHandler
        
        class MonitoringCallbackHandler(BaseCallbackHandler):
            def __init__(self, monitor):
                self.monitor = monitor
            
            def on_chain_start(self, serialized, inputs, **kwargs):
                self.start_time = time.time()
                self.chain_name = serialized.get("name", "unknown")
                
            def on_chain_end(self, outputs, **kwargs):
                end_time = time.time()
                
                # Extract token usage if available
                token_usage = kwargs.get("token_usage", {})
                total_tokens = sum(token_usage.values()) if token_usage else None
                
                # Track execution
                self.monitor.track_execution(
                    chain_name=self.chain_name,
                    start_time=self.start_time,
                    end_time=end_time,
                    token_usage=total_tokens,
                    success=True
                )
            
            def on_chain_error(self, error, **kwargs):
                end_time = time.time()
                self.monitor.track_execution(
                    chain_name=self.chain_name,
                    start_time=self.start_time,
                    end_time=end_time,
                    success=False
                )
        
        return MonitoringCallbackHandler(self)

Recipe 6: Security Best Practices

When deploying LangChain applications, security must be a priority:

class SecurityManager:
    def __init__(self):
        self.sensitive_patterns = [
            r"\b\d{3}-\d{2}-\d{4}\b",  # SSN
            r"\b\d{16}\b",  # Credit card
            r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b",  # Email
        ]
    
    def sanitize_input(self, user_input):
        """Sanitize user input before processing"""
        # Remove sensitive information
        sanitized = user_input
        for pattern in self.sensitive_patterns:
            sanitized = re.sub(pattern, "[REDACTED]", sanitized)
        
        # Prevent prompt injection
        sanitized = self._prevent_prompt_injection(sanitized)
        
        # Validate length and content
        if len(sanitized) > 10000:
            raise ValueError("Input too long")
        
        return sanitized
    
    def _prevent_prompt_injection(self, text):
        """Attempt to detect and prevent prompt injection"""
        injection_indicators = [
            "ignore previous instructions",
            "disregard earlier directions",
            "system prompt",
            "as an AI language model",
            "your initial instructions"
        ]
        
        text_lower = text.lower()
        for indicator in injection_indicators:
            if indicator in text_lower:
                # Log potential injection attempt
                logger.warning(f"Potential prompt injection detected: {indicator}")
                # Optionally: raise error or return safe response
                return "I cannot process this request due to security policies."
        
        return text
    
    def implement_rate_limiting(self, user_id, endpoint):
        """Implement rate limiting per user per endpoint"""
        redis_key = f"rate_limit:{user_id}:{endpoint}"
        
        current_count = redis_client.incr(redis_key)
        if current_count == 1:
            # Set expiration for 1 minute
            redis_client.expire(redis_key, 60)
        
        if current_count > 60:  # 60 requests per minute
            raise RateLimitExceeded("Rate limit exceeded")
        
        return True
    
    def secure_api_keys(self):
        """Secure handling of API keys"""
        # Never hardcode API keys
        # Use environment variables or secure secret management
        api_key = os.getenv("OPENAI_API_KEY")
        
        if not api_key:
            raise ValueError("API key not found in environment variables")
        
        # Rotate keys regularly
        self._schedule_key_rotation()
        
        return api_key

Recipe 7: Deployment and Scaling Patterns

Deploying LangChain applications requires consideration of scalability and reliability:

Containerized Deployment

# Dockerfile example for LangChain application
"""
FROM python:3.9-slim

WORKDIR /app

# Install system dependencies
RUN apt-get update && apt-get install -y \
    build-essential \
    curl \
    && rm -rf /var/lib/apt/lists/*

# Copy requirements
COPY requirements.txt .

# Install Python dependencies
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code
COPY . .

# Create non-root user
RUN useradd -m -u 1000 langchainuser
USER langchainuser

# Health check
HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1

# Run application
CMD ["gunicorn", "app:app", "--bind", "0.0.0.0:8000", "--workers", "4"]
"""

# requirements.txt should include:
"""
langchain==0.0.200
openai==0.27.8
chromadb==0.4.0
redis==4.5.5
fastapi==0.100.0
uvicorn[standard]==0.23.0
prompt-toolkit==3.0.38
tenacity==8.2.2
"""

Horizontal Scaling Strategy

class ScalingManager:
    def __init__(self, min_instances=2, max_instances=10):
        self.min_instances = min_instances
        self.max_instances = max_instances
        
    def auto_scale_based_on_metrics(self, current_metrics):
        """Auto-scale based on performance metrics"""
        scaling_decisions = []
        
        # CPU-based scaling
        if current_metrics["cpu_percent"] > 80:
            scaling_decisions.append(self._scale_up("high_cpu"))
        
        # Memory-based scaling
        if current_metrics["memory_percent"] > 85:
            scaling_decisions.append(self._scale_up("high_memory"))
        
        # Request queue-based scaling
        if current_metrics["queue_length"] > 100:
            scaling_decisions.append(self._scale_up("long_queue"))
        
        # Scale down during low usage
        if (current_metrics["cpu_percent"] < 30 and 
            current_metrics["memory_percent"] < 40 and
            current_metrics["queue_length"] < 10):
            scaling_decisions.append(self._scale_down("low_usage"))
        
        return scaling_decisions
    
    def implement_session_affinity(self):
        """Implement session affinity for stateful applications"""
        # Use consistent hashing for session routing
        # Store session state in shared Redis
        # Implement session recovery mechanisms
        
        return {
            "routing_strategy": "consistent_hashing",
            "session_store": "redis_cluster",
            "recovery_enabled": True
        }

Common Pitfalls and Solutions

Based on real-world deployments, here are common issues and their solutions:

1. Memory Leaks in Long-Running Processes

Problem: LangChain applications can accumulate memory over time when processing large volumes.

Solution: Implement periodic garbage collection and monitor memory usage:

def manage_memory_usage():
    import gc
    import psutil
    
    process = psutil.Process()
    
    # Check memory usage
    if process.memory_percent() > 70:
        logger.warning("High memory usage detected")
        
        # Force garbage collection
        gc.collect()
        
        # Clear unused caches
        clear_unused_caches()
        
        # Consider restarting worker if memory remains high
        if process.memory_percent() > 85:
            graceful_restart()

2. Rate Limiting and API Timeouts

Problem: External API calls can fail due to rate limits or timeouts.

Solution: Implement robust retry logic with exponential backoff:

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=4, max=60),
    retry=retry_if_exception_type((RateLimitError, TimeoutError, ConnectionError))
)
def call_external_api_with_retry(api_func, *args, **kwargs):
    return api_func(*args, **kwargs)

3. Inconsistent Output Quality

Problem: LLM outputs can vary in quality and format.

Solution: Implement output validation and post-processing:

class OutputValidator:
    def validate_and_clean(self, output, expected_format="json"):
        """Validate and clean LLM output"""
        
        # Try to parse as expected format
        try:
            if expected_format == "json":
                parsed = json.loads(output)
                return self._clean_json_output(parsed)
            elif expected_format == "text":
                return self._clean_text_output(output)
        except json.JSONDecodeError:
            # Attempt to extract JSON from text
            json_match = re.search(r'\{.*\}', output, re.DOTALL)
            if json_match:
                try:
                    parsed = json.loads(json_match.group())
                    return self._clean_json_output(parsed)
                except:
                    pass
        
        # If validation fails, return structured error
        return {
            "error": "output_validation_failed",
            "original_output": output[:500],  # Truncate for logging
            "fallback": self._generate_fallback_response()
        }

Testing Strategies for LangChain Applications

Testing AI applications requires special considerations:

class LangChainTestSuite:
    def setup_test_environment(self):
        """Setup test environment with mocked LLM calls"""
        # Use LangChain's FakeListLLM for testing
        from langchain.llms import FakeListLLM
        
        self.test_llm = FakeListLLM(responses=[
            "Test response 1",
            "Test response 2",
            "Test response 3"
        ])
        
        # Mock external APIs
        self._setup_api_mocks()
    
    def test_chain_logic(self, chain, test_inputs):
        """Test chain logic without actual API calls"""
        test_results = []
        
        for test_input in test_inputs:
            # Run with test LLM
            result = chain.run(test_input)
            
            # Validate structure
            assert isinstance(result, str), "Result should be string"
            assert len(result) > 0, "Result should not be empty"
            
            # Check for expected patterns
            if "expected_pattern" in test_input:
                assert re.search(test_input["expected_pattern"], result), \
                    f"Pattern not found in result: {test_input['expected_pattern']}"
            
            test_results.append({
                "input": test_input,
                "output": result,
                "passed": True
            })
        
        return test_results
    
    def performance_test(self, chain, concurrent_requests=10):
        """Performance testing with concurrent requests"""
        from concurrent.futures import ThreadPoolExecutor
        import time
        
        def make_request(request_id):
            start_time = time.time()
            try:
                result = chain.run(f"Test request {request_id}")
                end_time = time.time()
                return {
                    "request_id": request_id,
                    "success": True,
                    "response_time": end_time - start_time
                }
            except Exception as e:
                end_time = time.time()
                return {
                    "request_id": request_id,
                    "success": False,
                    "response_time": end_time - start_time,
                    "error": str(e)
                }
        
        with ThreadPoolExecutor(max_workers=concurrent_requests) as executor:
            futures = [executor.submit(make_request, i) for i in range(concurrent_requests)]
            results = [f.result() for f in futures]
        
        # Calculate metrics
        successful = [r for r in results if r["success"]]
        avg_response_time = sum(r["response_time"] for r in successful) / len(successful)
        
        return {
            "total_requests": len(results),
            "successful_requests": len(successful),
            "success_rate": len(successful) / len(results),
            "average_response_time": avg_response_time,
            "p95_response_time": self._calculate_percentile(
                [r["response_time"] for r in successful], 95
            )
        }

Migration and Version Management

As LangChain evolves, managing migrations becomes crucial:

class MigrationManager:
    def __init__(self, current_version="0.0.200", target_version="0.0.210"):
        self.current_version = current_version
        self.target_version = target_version
        
    def check_backward_compatibility(self):
        """Check for breaking changes between versions"""
        breaking_changes = self._load_breaking_changes_list()
        
        issues_found = []
        for change in breaking_changes:
            if self._affects_our_codebase(change):
                issues_found.append({
                    "issue": change["description"],
                    "severity": change["severity"],
                    "mitigation": change["mitigation"]
                })
        
        return issues_found
    
    def create_migration_plan(self):
        """Create step-by-step migration plan"""
        return {
            "phase_1": {
                "actions": [
                    "Update requirements to target version",
                    "Run compatibility checks",
                    "Update deprecated imports"
                ],
                "estimated_time": "2 hours",
                "rollback_plan": "Revert to previous requirements.txt"
            },
            "phase_2": {
                "actions": [
                    "Update chain configurations",
                    "Test core functionality",
                    "Update prompt templates"
                ],
                "estimated_time": "4 hours",
                "rollback_plan": "Use feature flags for new configurations"
            },
            "phase_3": {
                "actions": [
                    "Deploy to staging",
                    "Run full test suite",
                    "Monitor for regressions"
                ],
                "estimated_time": "8 hours",
                "rollback_plan": "Blue-green deployment with quick rollback"
            }
        }
    
    def implement_feature_flags(self):
        """Implement feature flags for gradual migration"""
        return {
            "use_new_memory_system": os.getenv("USE_NEW_MEMORY", "false") == "true",
            "enable_new_tools": os.getenv("ENABLE_NEW_TOOLS", "false") == "true",
            "migration_phase": os.getenv("MIGRATION_PHASE", "1")
        }

Conclusion: Building Production-Ready LangChain Applications

Building real applications with LangChain requires going beyond basic tutorials. By implementing the patterns and recipes in this guide, you can create robust, maintainable, and scalable AI applications. Remember these key principles:

1. Start with error handling - Assume things will fail and plan for graceful degradation
2. Monitor everything - Track costs, performance, and usage patterns
3. Optimize incrementally - Start simple and add optimizations based on actual usage
4. Plan for scale - Design with horizontal scaling in mind from the beginning
5. Security first - Implement input validation, rate limiting, and sensitive data handling

LangChain provides a powerful foundation for AI applications, but production readiness comes from the patterns and practices you implement around it. Use these recipes as starting points, adapt them to your specific needs, and continually iterate based on real-world feedback.

Visuals Produced by AI

Further Reading

• Retrieval-Augmented Generation (RAG) — Advanced Practical Guide
• Vector Databases Explained: From Embeddings to Search
• LLMOps 101: Deploying, Monitoring and Managing Models