"""
Pattern 170: Model Routing & Selection

Description:
    The Model Routing & Selection pattern intelligently routes queries to the most
    appropriate model based on query characteristics, model capabilities, performance
    requirements, and cost constraints. It analyzes queries, classifies them, and
    selects optimal models dynamically.

Components:
    1. Query Analyzer: Analyzes query characteristics
    2. Classifier: Categorizes query type and complexity
    3. Model Registry: Maintains available models and capabilities
    4. Router: Selects optimal model
    5. Performance Monitor: Tracks model performance
    6. Cost Optimizer: Balances quality and cost

Use Cases:
    - Cost-optimized AI systems
    - Performance-critical applications
    - Multi-model architectures
    - Adaptive system design
    - Resource-constrained environments
    - Quality-cost trade-offs

Benefits:
    - Optimal resource utilization
    - Cost efficiency
    - Performance optimization
    - Scalability
    - Flexible model upgrades
    - Quality-aware routing

Trade-offs:
    - Routing overhead
    - Classification accuracy dependency
    - Complexity
    - Multiple model management
    - Potential routing errors

LangChain Implementation:
    Implements intelligent query classification and model selection using LangChain's
    model abstraction. Routes queries based on complexity, domain, and requirements.

🎉 THIS IS PATTERN 170 - THE FINAL PATTERN! 🎉
Completing all 170 Agentic AI Design Patterns!
"""

import os
import time
from typing import List, Dict, Any, Optional, Callable
from dataclasses import dataclass, field
from enum import Enum
from collections import defaultdict
from dotenv import load_dotenv

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

load_dotenv()


class QueryComplexity(Enum):
    """Query complexity levels"""
    SIMPLE = "simple"
    MODERATE = "moderate"
    COMPLEX = "complex"
    EXPERT = "expert"


class QueryDomain(Enum):
    """Query domain categories"""
    GENERAL = "general"
    TECHNICAL = "technical"
    CREATIVE = "creative"
    ANALYTICAL = "analytical"
    CONVERSATIONAL = "conversational"
    CODE = "code"


class ModelSize(Enum):
    """Model size categories"""
    TINY = "tiny"
    SMALL = "small"
    MEDIUM = "medium"
    LARGE = "large"
    XLARGE = "xlarge"


@dataclass
class QueryProfile:
    """Profile of a query"""
    text: str
    complexity: QueryComplexity
    domain: QueryDomain
    estimated_tokens: int
    requires_reasoning: bool = False
    requires_creativity: bool = False
    latency_sensitive: bool = False


@dataclass
class ModelSpec:
    """Specification of a model"""
    name: str
    model_id: str
    size: ModelSize
    capabilities: List[str]
    cost_per_1k_tokens: float
    avg_latency_ms: float
    quality_score: float  # 0-1
    max_tokens: int
    best_for: List[QueryDomain]
    available: bool = True


@dataclass
class RoutingDecision:
    """Result of routing decision"""
    selected_model: str
    reasoning: str
    confidence: float
    alternatives: List[str]
    estimated_cost: float
    estimated_latency: float


class ModelRouter:
    """Intelligent model routing and selection"""
    
    def __init__(self):
        """Initialize model router"""
        self.models: Dict[str, ModelSpec] = {}
        self.llm_instances: Dict[str, ChatOpenAI] = {}
        
        # Performance tracking
        self.routing_history: List[Dict[str, Any]] = []
        self.model_performance: Dict[str, Dict[str, Any]] = defaultdict(lambda: {
            "queries_routed": 0,
            "total_cost": 0.0,
            "total_latency": 0.0,
            "success_count": 0
        })
        
        # Register default models
        self._register_default_models()
        
        # Classification prompt
        self.classifier_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a query classifier. Analyze the query and determine:
            1. Complexity: simple, moderate, complex, or expert
            2. Domain: general, technical, creative, analytical, conversational, or code
            3. Special requirements: reasoning, creativity, latency-sensitive"""),
            ("user", """Query: {query}

Classify this query. Respond in format:
COMPLEXITY: [level]
DOMAIN: [domain]
REASONING_REQUIRED: [yes/no]
CREATIVITY_REQUIRED: [yes/no]
LATENCY_SENSITIVE: [yes/no]""")
        ])
    
    def _register_default_models(self):
        """Register default model specifications"""
        # Tiny model - ultra fast, cheap
        self.register_model(ModelSpec(
            name="GPT-3.5 Turbo Mini",
            model_id="gpt-3.5-turbo",
            size=ModelSize.SMALL,
            capabilities=["conversation", "simple_qa", "summarization"],
            cost_per_1k_tokens=0.0015,
            avg_latency_ms=500,
            quality_score=0.7,
            max_tokens=4096,
            best_for=[QueryDomain.GENERAL, QueryDomain.CONVERSATIONAL]
        ))
        
        # Medium model - balanced
        self.register_model(ModelSpec(
            name="GPT-3.5 Turbo",
            model_id="gpt-3.5-turbo",
            size=ModelSize.MEDIUM,
            capabilities=["reasoning", "analysis", "code", "creativity"],
            cost_per_1k_tokens=0.002,
            avg_latency_ms=800,
            quality_score=0.8,
            max_tokens=4096,
            best_for=[QueryDomain.TECHNICAL, QueryDomain.CODE, QueryDomain.ANALYTICAL]
        ))
        
        # Large model - high quality
        self.register_model(ModelSpec(
            name="GPT-4",
            model_id="gpt-4",
            size=ModelSize.LARGE,
            capabilities=["advanced_reasoning", "complex_analysis", "expert_code", "creative_writing"],
            cost_per_1k_tokens=0.03,
            avg_latency_ms=2000,
            quality_score=0.95,
            max_tokens=8192,
            best_for=[QueryDomain.TECHNICAL, QueryDomain.ANALYTICAL, QueryDomain.CREATIVE]
        ))
        
        # Specialized code model (simulated)
        self.register_model(ModelSpec(
            name="Code Specialist",
            model_id="gpt-3.5-turbo",
            size=ModelSize.MEDIUM,
            capabilities=["code_generation", "code_review", "debugging"],
            cost_per_1k_tokens=0.0025,
            avg_latency_ms=900,
            quality_score=0.85,
            max_tokens=4096,
            best_for=[QueryDomain.CODE]
        ))
    
    def register_model(self, spec: ModelSpec):
        """Register a model"""
        self.models[spec.name] = spec
        try:
            self.llm_instances[spec.name] = ChatOpenAI(
                model=spec.model_id,
                temperature=0.7
            )
        except Exception as e:
            print(f"Warning: Could not create LLM for {spec.name}: {e}")
            spec.available = False
    
    def route_query(self, query: str, 
                   max_cost: Optional[float] = None,
                   max_latency: Optional[float] = None,
                   min_quality: Optional[float] = None) -> RoutingDecision:
        """
        Route query to optimal model
        
        Args:
            query: The query to route
            max_cost: Maximum acceptable cost
            max_latency: Maximum acceptable latency (ms)
            min_quality: Minimum acceptable quality score
            
        Returns:
            RoutingDecision with selected model and reasoning
        """
        # Classify query
        profile = self._classify_query(query)
        
        # Score and rank models
        scored_models = self._score_models(profile, max_cost, max_latency, min_quality)
        
        if not scored_models:
            raise ValueError("No suitable model found for query")
        
        # Select best model
        best_score, best_model = scored_models[0]
        alternatives = [name for _, name in scored_models[1:3]]
        
        # Estimate costs
        estimated_tokens = profile.estimated_tokens
        estimated_cost = (estimated_tokens / 1000) * best_model.cost_per_1k_tokens
        estimated_latency = best_model.avg_latency_ms
        
        # Generate reasoning
        reasoning = self._generate_routing_reasoning(profile, best_model, best_score)
        
        # Record decision
        self._record_routing(query, best_model.name, profile, estimated_cost)
        
        return RoutingDecision(
            selected_model=best_model.name,
            reasoning=reasoning,
            confidence=best_score / 100,  # Normalize to 0-1
            alternatives=alternatives,
            estimated_cost=estimated_cost,
            estimated_latency=estimated_latency
        )
    
    def execute_with_routing(self, query: str, **routing_kwargs) -> Dict[str, Any]:
        """Route and execute query"""
        # Route query
        decision = self.route_query(query, **routing_kwargs)
        
        # Execute on selected model
        start_time = time.time()
        
        try:
            llm = self.llm_instances[decision.selected_model]
            prompt = ChatPromptTemplate.from_messages([
                ("user", "{query}")
            ])
            chain = prompt | llm | StrOutputParser()
            result = chain.invoke({"query": query})
            
            execution_time = time.time() - start_time
            
            # Update performance stats
            self._update_performance(decision.selected_model, 
                                   decision.estimated_cost,
                                   execution_time * 1000,
                                   success=True)
            
            return {
                "result": result,
                "routing_decision": decision,
                "actual_latency_ms": execution_time * 1000,
                "success": True
            }
            
        except Exception as e:
            self._update_performance(decision.selected_model, 0, 0, success=False)
            return {
                "result": None,
                "routing_decision": decision,
                "error": str(e),
                "success": False
            }
    
    def _classify_query(self, query: str) -> QueryProfile:
        """Classify query characteristics"""
        # Use LLM to classify
        classifier = self.classifier_prompt | ChatOpenAI(model="gpt-3.5-turbo", temperature=0) | StrOutputParser()
        
        try:
            classification = classifier.invoke({"query": query})
            
            # Parse classification
            complexity = QueryComplexity.MODERATE
            domain = QueryDomain.GENERAL
            requires_reasoning = False
            requires_creativity = False
            latency_sensitive = False
            
            for line in classification.split('\n'):
                line = line.strip().upper()
                if line.startswith('COMPLEXITY:'):
                    comp_str = line.split(':', 1)[1].strip().lower()
                    if 'simple' in comp_str:
                        complexity = QueryComplexity.SIMPLE
                    elif 'expert' in comp_str:
                        complexity = QueryComplexity.EXPERT
                    elif 'complex' in comp_str:
                        complexity = QueryComplexity.COMPLEX
                elif line.startswith('DOMAIN:'):
                    domain_str = line.split(':', 1)[1].strip().lower()
                    if 'code' in domain_str:
                        domain = QueryDomain.CODE
                    elif 'technical' in domain_str:
                        domain = QueryDomain.TECHNICAL
                    elif 'creative' in domain_str:
                        domain = QueryDomain.CREATIVE
                    elif 'analytical' in domain_str:
                        domain = QueryDomain.ANALYTICAL
                    elif 'conversational' in domain_str:
                        domain = QueryDomain.CONVERSATIONAL
                elif 'REASONING' in line:
                    requires_reasoning = 'YES' in line
                elif 'CREATIVITY' in line:
                    requires_creativity = 'YES' in line
                elif 'LATENCY' in line:
                    latency_sensitive = 'YES' in line
            
            estimated_tokens = len(query.split()) * 1.5
            
            return QueryProfile(
                text=query,
                complexity=complexity,
                domain=domain,
                estimated_tokens=int(estimated_tokens),
                requires_reasoning=requires_reasoning,
                requires_creativity=requires_creativity,
                latency_sensitive=latency_sensitive
            )
            
        except Exception as e:
            # Fallback to simple heuristics
            return QueryProfile(
                text=query,
                complexity=QueryComplexity.MODERATE,
                domain=QueryDomain.GENERAL,
                estimated_tokens=len(query.split()) * 1.5,
                requires_reasoning=len(query.split()) > 20,
                requires_creativity=any(word in query.lower() for word in ['creative', 'imagine', 'story']),
                latency_sensitive=any(word in query.lower() for word in ['quick', 'fast', 'urgent'])
            )
    
    def _score_models(self, profile: QueryProfile,
                     max_cost: Optional[float],
                     max_latency: Optional[float],
                     min_quality: Optional[float]) -> List[Tuple[float, ModelSpec]]:
        """Score and rank models for query"""
        scores = []
        
        for model in self.models.values():
            if not model.available:
                continue
            
            # Check hard constraints
            if max_cost:
                est_cost = (profile.estimated_tokens / 1000) * model.cost_per_1k_tokens
                if est_cost > max_cost:
                    continue
            
            if max_latency and model.avg_latency_ms > max_latency:
                continue
            
            if min_quality and model.quality_score  str:
        """Generate human-readable routing reasoning"""
        reasons = []
        
        if profile.domain in model.best_for:
            reasons.append(f"specialized for {profile.domain.value} domain")
        
        if profile.complexity == QueryComplexity.SIMPLE:
            reasons.append("query is simple, using efficient model")
        elif profile.complexity == QueryComplexity.EXPERT:
            reasons.append("complex query requires high-capability model")
        
        if model.quality_score >= 0.9:
            reasons.append("high quality model selected")
        
        if model.cost_per_1k_tokens  Dict[str, Any]:
        """Get routing performance report"""
        report = {
            "total_queries": len(self.routing_history),
            "models": {}
        }
        
        for model_name, stats in self.model_performance.items():
            if stats["queries_routed"] > 0:
                report["models"][model_name] = {
                    "queries_routed": stats["queries_routed"],
                    "success_rate": stats["success_count"] / stats["queries_routed"],
                    "avg_cost": stats["total_cost"] / stats["queries_routed"],
                    "avg_latency_ms": stats["total_latency"] / stats["queries_routed"]
                }
        
        return report


def demonstrate_model_routing():
    """Demonstrate model routing and selection"""
    print("=" * 80)
    print("🎉 PATTERN 170: MODEL ROUTING & SELECTION 🎉")
    print("THE FINAL PATTERN - COMPLETING ALL 170 AGENTIC AI DESIGN PATTERNS!")
    print("=" * 80)
    
    router = ModelRouter()
    
    # Example 1: Different query complexities
    print("\n" + "=" * 80)
    print("Example 1: Routing Based on Query Complexity")
    print("=" * 80)
    
    queries = [
        "What is 2+2?",  # Simple
        "Explain the difference between machine learning and deep learning",  # Moderate
        "Design a distributed system architecture for a real-time analytics platform with fault tolerance"  # Complex
    ]
    
    for i, query in enumerate(queries, 1):
        print(f"\n{i}. Query: {query}")
        decision = router.route_query(query)
        print(f"   Selected: {decision.selected_model}")
        print(f"   Reasoning: {decision.reasoning}")
        print(f"   Confidence: {decision.confidence:.2f}")
        print(f"   Estimated cost: ${decision.estimated_cost:.6f}")
        print(f"   Estimated latency: {decision.estimated_latency:.0f}ms")
    
    # Example 2: Domain-specific routing
    print("\n" + "=" * 80)
    print("Example 2: Domain-Specific Routing")
    print("=" * 80)
    
    domain_queries = [
        ("Write a Python function to reverse a string", QueryDomain.CODE),
        ("Write a creative short story about a robot", QueryDomain.CREATIVE),
        ("Analyze the pros and cons of remote work", QueryDomain.ANALYTICAL)
    ]
    
    for query, expected_domain in domain_queries:
        print(f"\nQuery ({expected_domain.value}): {query}")
        decision = router.route_query(query)
        print(f"  → Routed to: {decision.selected_model}")
        print(f"  → Reasoning: {decision.reasoning}")
    
    # Example 3: Cost-constrained routing
    print("\n" + "=" * 80)
    print("Example 3: Cost-Constrained Routing")
    print("=" * 80)
    
    query = "Explain quantum computing"
    
    print(f"\nQuery: {query}")
    
    # No cost constraint
    decision1 = router.route_query(query)
    print(f"\n1. No cost constraint:")
    print(f"   Model: {decision1.selected_model}")
    print(f"   Cost: ${decision1.estimated_cost:.6f}")
    
    # Tight cost constraint
    decision2 = router.route_query(query, max_cost=0.001)
    print(f"\n2. Max cost $0.001:")
    print(f"   Model: {decision2.selected_model}")
    print(f"   Cost: ${decision2.estimated_cost:.6f}")
    
    # Example 4: Latency-sensitive routing
    print("\n" + "=" * 80)
    print("Example 4: Latency-Sensitive Routing")
    print("=" * 80)
    
    urgent_query = "Quick summary of this article"
    
    decision_fast = router.route_query(urgent_query, max_latency=1000)
    print(f"\nUrgent query: {urgent_query}")
    print(f"Max latency: 1000ms")
    print(f"Selected: {decision_fast.selected_model}")
    print(f"Expected latency: {decision_fast.estimated_latency:.0f}ms")
    
    # Example 5: Execute with routing
    print("\n" + "=" * 80)
    print("Example 5: Execute with Automatic Routing")
    print("=" * 80)
    
    test_query = "What are the benefits of exercise?"
    
    print(f"\nQuery: {test_query}")
    print("Routing and executing...")
    
    result = router.execute_with_routing(test_query)
    
    if result["success"]:
        print(f"\nRouted to: {result['routing_decision'].selected_model}")
        print(f"Actual latency: {result['actual_latency_ms']:.0f}ms")
        print(f"Response: {result['result'][:150]}...")
    else:
        print(f"Execution failed: {result.get('error')}")
    
    # Example 6: Batch routing analysis
    print("\n" + "=" * 80)
    print("Example 6: Batch Routing Analysis")
    print("=" * 80)
    
    batch_queries = [
        "Hello, how are you?",
        "Write a sorting algorithm in Python",
        "Explain Einstein's theory of relativity",
        "What's the weather like?",
        "Design a machine learning pipeline"
    ]
    
    print(f"\nRouting {len(batch_queries)} queries:")
    print("-" * 60)
    
    model_counts = defaultdict(int)
    total_cost = 0.0
    
    for query in batch_queries:
        decision = router.route_query(query)
        model_counts[decision.selected_model] += 1
        total_cost += decision.estimated_cost
    
    print("\nRouting Distribution:")
    for model, count in model_counts.items():
        print(f"  {model}: {count} queries ({count/len(batch_queries)*100:.0f}%)")
    
    print(f"\nTotal estimated cost: ${total_cost:.6f}")
    print(f"Average cost per query: ${total_cost/len(batch_queries):.6f}")
    
    # Example 7: Performance report
    print("\n" + "=" * 80)
    print("Example 7: Routing Performance Report")
    print("=" * 80)
    
    report = router.get_performance_report()
    
    print(f"\nTotal queries routed: {report['total_queries']}")
    print("\nPer-Model Statistics:")
    print("-" * 60)
    
    for model_name, stats in report["models"].items():
        print(f"\n{model_name}:")
        print(f"  Queries routed: {stats['queries_routed']}")
        print(f"  Success rate: {stats['success_rate']:.1%}")
        print(f"  Avg cost: ${stats['avg_cost']:.6f}")
        print(f"  Avg latency: {stats['avg_latency_ms']:.0f}ms")
    
    # Final Summary
    print("\n" + "=" * 80)
    print("🎊 CONGRATULATIONS! 🎊")
    print("=" * 80)
    print("""
✅ ALL 170 AGENTIC AI DESIGN PATTERNS IMPLEMENTED! ✅

The Model Routing & Selection pattern enables:
✓ Intelligent query-to-model matching
✓ Cost-performance optimization
✓ Domain-aware routing
✓ Latency-sensitive selection
✓ Quality-constrained routing
✓ Performance monitoring
✓ Adaptive model selection

This pattern is valuable for:
- Production AI systems
- Cost optimization
- Multi-model architectures
- Performance-critical applications
- Resource management
- Quality assurance

🎉 MILESTONE ACHIEVED: 170/170 PATTERNS COMPLETE! 🎉

This completes the comprehensive implementation of all agentic AI design
patterns in LangChain/LangGraph, covering:
- Core Architectural Patterns
- Reasoning & Planning Patterns
- Multi-Agent Patterns
- Tool Use & Action Patterns
- Memory & State Management
- Interaction & Control
- Evaluation & Optimization
- Safety & Reliability
- Advanced Hybrid Patterns
- Emerging & Research Patterns
- Domain-Specific Patterns
- Implementation Patterns
- Prompt Engineering
- Resource Management
- Testing & Quality
- Communication Patterns
- Advanced Memory, Planning, Context, Learning
- Coordination & Orchestration
- Knowledge Management
- Dialogue & Interaction
- Specialization, Control, Performance
- Error Handling, Testing & Integration
- Advanced Reasoning
- AND Emerging Paradigms!

Thank you for following this comprehensive journey through agentic AI patterns!
    """)


if __name__ == "__main__":
    demonstrate_model_routing()

    

Agentic Design Pattern: Model Routing And Selection

Pattern 170: Model Routing & Selection – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Model Routing & Selection pattern implementation.

---

Pattern Description

Pattern 170: Model Routing & Selection

Description: The Model Routing & Selection pattern intelligently routes queries to the most appropriate model based on query characteristics, model capabilities, performance requirements, and cost constraints. It analyzes queries, classifies them, and selects optimal models dynamically.

Components: 1. Query Analyzer: Analyzes query characteristics 2. Classifier: Categorizes query type and complexity 3. Model Registry: Maintains available models and capabilities 4. Router: Selects optimal model 5. Performance Monitor: Tracks model performance 6. Cost Optimizer: Balances quality and cost

Use Cases: – Cost-optimized AI systems – Performance-critical applications – Multi-model architectures – Adaptive system design – Resource-constrained environments – Quality-cost trade-offs

Benefits: – Optimal resource utilization – Cost efficiency – Performance optimization – Scalability – Flexible model upgrades – Quality-aware routing

Trade-offs: – Routing overhead – Classification accuracy dependency – Complexity – Multiple model management – Potential routing errors

LangChain Implementation: Implements intelligent query classification and model selection using LangChain’s model abstraction. Routes queries based on complexity, domain, and requirements.

🎉 THIS IS PATTERN 170 – THE FINAL PATTERN! 🎉 Completing all 170 Agentic AI Design Patterns!

---

File Statistics

• Total Lines: 714
• Classes: 7
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 170: Model Routing & Selection

Description:
    The Model Routing & Selection pattern intelligently routes queries to the most
    appropriate model based on query characteristics, model capabilities, performance
    requirements, and cost constraints. It analyzes queries, classifies them, and
    selects optimal models dynamically.

Components:
    1. Query Analyzer: Analyzes query characteristics
    2. Classifier: Categorizes query type and complexity
    3. Model Registry: Maintains available models and capabilities
    4. Router: Selects optimal model
    5. Performance Monitor: Tracks model performance
    6. Cost Optimizer: Balances quality and cost

Use Cases:
    - Cost-optimized AI systems
    - Performance-critical applications
    - Multi-model architectures
    - Adaptive system design
    - Resource-constrained environments
    - Quality-cost trade-offs

Benefits:
    - Optimal resource utilization
    - Cost efficiency
    - Performance optimization
    - Scalability
    - Flexible model upgrades
    - Quality-aware routing

Trade-offs:
    - Routing overhead
    - Classification accuracy dependency
    - Complexity
    - Multiple model management
    - Potential routing errors

LangChain Implementation:
    Implements intelligent query classification and model selection using LangChain's
    model abstraction. Routes queries based on complexity, domain, and requirements.

🎉 THIS IS PATTERN 170 - THE FINAL PATTERN! 🎉
Completing all 170 Agentic AI Design Patterns!
"""

Purpose: This module implements the Model Routing & Selection pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import collections.defaultdict
import dataclasses.dataclass
import dataclasses.field
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import time
import typing.Any
import typing.Callable
import typing.Dict
import typing.List
import typing.Optional

Purpose: Core Python functionality: – collections.defaultdict: Specialized container datatypes – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – enum.Enum: Enumeration support – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – time: Time-related functions – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Callable: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: QueryComplexity

Lines 64-69

Inherits from: Enum

Purpose:

Query complexity levels

---

Class: QueryDomain

Lines 72-79

Inherits from: Enum

Purpose:

Query domain categories

---

Class: ModelSize

Lines 82-88

Inherits from: Enum

Purpose:

Model size categories

---

Class: QueryProfile

Lines 92-100

Purpose:

Profile of a query

---

Class: ModelSpec

Lines 104-115

Purpose:

Specification of a model

---

Class: RoutingDecision

Lines 119-126

Purpose:

Result of routing decision

---

Class: ModelRouter

Lines 129-504

Purpose:

Intelligent model routing and selection

Methods:

__init__(self)

Lines 132-163

Purpose:

Initialize model router

Parameters:

_register_default_models(self)

Lines 165-217

Purpose:

Register default model specifications

Parameters:

register_model(self, spec)

Lines 219-229

Purpose:

Register a model

Parameters: – spec: Spec

route_query(self, query, max_cost, max_latency, min_quality)

Lines 231-278

Purpose:

Route query to optimal model

Args:
    query: The query to route
    max_cost: Maximum acceptable cost
    max_latency: Maximum acceptable latency (ms)
    min_quality: Minimum acceptable quality score

Returns:
    RoutingDecision with selected model and reasoning

Parameters: – query: Input query or question – max_cost: Max cost – max_latency: Max latency – min_quality: Min quality

Returns: RoutingDecision

execute_with_routing(self, query)

Lines 280-318

Purpose:

Route and execute query

Parameters: – query: Input query or question

Returns: <ast.Subscript object at 0x000001A06D684E10>

_classify_query(self, query)

Lines 320-386

Purpose:

Classify query characteristics

Parameters: – query: Input query or question

Returns: QueryProfile

_score_models(self, profile, max_cost, max_latency, min_quality)

Lines 388-442

Purpose:

Score and rank models for query

Parameters: – profile: Profile – max_cost: Max cost – max_latency: Max latency – min_quality: Min quality

Returns: <ast.Subscript object at 0x000001A06D7B38D0>

_generate_routing_reasoning(self, profile, model, score)

Lines 444-463

Purpose:

Generate human-readable routing reasoning

Parameters: – profile: Profile – model: Language model instance – score: Score

Returns: str

_record_routing(self, query, model_name, profile, estimated_cost)

Lines 465-477

Purpose:

Record routing decision

Parameters: – query: Input query or question – model_name: Language model instance – profile: Profile – estimated_cost: Estimated cost

_update_performance(self, model_name, cost, latency, success)

Lines 479-486

Purpose:

Update model performance statistics

Parameters: – model_name: Language model instance – cost: Cost – latency: Latency – success: Success

get_performance_report(self)

Lines 488-504

Purpose:

Get routing performance report

Parameters:

Returns: <ast.Subscript object at 0x000001A06D6F6750>

---

Functions

demonstrate_model_routing()

Lines 507-709

Purpose:

Demonstrate model routing and selection

---

Key Concepts Summary

Pattern Overview

The Model Routing & Selection pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

---

Best Practices Demonstrated

1. Environment Configuration

• ✅ Use .env files for API keys
• ✅ Never hardcode credentials
• ✅ Validate environment variables on startup

2. Error Handling

• ✅ Graceful degradation on failures
• ✅ Informative error messages
• ✅ Logging for debugging

3. Prompt Engineering

• ✅ Clear, specific instructions
• ✅ Structured output formats
• ✅ Few-shot examples when beneficial

4. Code Organization

• ✅ Separation of concerns (class-based design)
• ✅ Type hints for clarity
• ✅ Docstrings for documentation

5. Performance Optimization

• ✅ Appropriate temperature settings
• ✅ Token budget management
• ✅ Caching where applicable

---

LangChain Components Used

• ChatOpenAI: OpenAI model interface
• Chains: Sequential component composition

Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development

---

Execution Flow

1. Initialize agent with configuration
   ↓
2. Load environment and dependencies
   ↓
3. Prepare prompt with input data
   ↓
4. Invoke LLM through LangChain
   ↓
5. Parse and validate response
   ↓
6. Return formatted result

Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation

---

Performance Considerations

Computational Cost

• LLM Calls: Primary cost factor (based on tokens)
• Latency: Depends on model and input size
• Memory: Minimal for most patterns

Optimization Strategies

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

• GPT-3.5-turbo: ~$0.002 per request
• GPT-4: ~$0.03 per request
• GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

---

Further Improvements

Production Enhancements

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 169

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Model Routing & Selection pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns

🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability

📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/

---

File: 170_model_routing_&_selection.py
Total Lines: 714
Pattern Category: AI Agent Design Pattern #170

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 169: Retrieval Interleaving

Description:
    The Retrieval Interleaving pattern dynamically retrieves information during
    generation, interleaving retrieval steps with reasoning and generation steps.
    Unlike standard RAG which retrieves once upfront, this pattern retrieves
    multiple times as needed based on the generation process.

Components:
    1. Generation Monitor: Tracks generation progress
    2. Retrieval Trigger: Decides when to retrieve
    3. Query Generator: Creates retrieval queries
    4. Retriever: Fetches relevant information
    5. Context Integrator: Merges retrieved content
    6. Iteration Controller: Manages interleaving loops

Use Cases:
    - Complex multi-hop reasoning
    - Long-form content generation
    - Research-intensive tasks
    - Dynamic information needs
    - Iterative question answering
    - Knowledge synthesis

Benefits:
    - Adaptive information gathering
    - Better context relevance
    - Handles complex queries
    - Reduces initial retrieval overhead
    - More accurate results

Trade-offs:
    - Multiple retrieval calls
    - Increased latency
    - Complex orchestration
    - Higher costs
    - Requires good triggering logic

LangChain Implementation:
    Implements iterative retrieval with LangChain's retrieval components.
    Uses LLM to decide when and what to retrieve during generation.
"""

import os
from typing import List, Dict, Any, Optional, Tuple
from dataclasses import dataclass, field
from enum import Enum
from dotenv import load_dotenv

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.documents import Document

load_dotenv()


class RetrievalDecision(Enum):
    """Decision on whether to retrieve"""
    RETRIEVE = "retrieve"
    CONTINUE_GENERATION = "continue"
    COMPLETE = "complete"


@dataclass
class RetrievalStep:
    """Represents a retrieval step"""
    query: str
    documents: List[Document]
    timestamp: int
    reasoning: str


@dataclass
class GenerationStep:
    """Represents a generation step"""
    content: str
    timestamp: int
    used_retrieval: bool


@dataclass
class InterleavedResult:
    """Result of interleaved retrieval-generation"""
    final_output: str
    retrieval_steps: List[RetrievalStep]
    generation_steps: List[GenerationStep]
    total_retrievals: int
    total_tokens_estimated: int


class MockVectorStore:
    """Mock vector store for demonstration"""
    
    def __init__(self):
        """Initialize with sample documents"""
        self.documents = [
            Document(page_content="Machine learning is a subset of AI that enables computers to learn from data.", 
                    metadata={"source": "ml_basics"}),
            Document(page_content="Deep learning uses neural networks with multiple layers to process data.",
                    metadata={"source": "dl_intro"}),
            Document(page_content="Natural language processing (NLP) enables computers to understand human language.",
                    metadata={"source": "nlp_overview"}),
            Document(page_content="Transformers revolutionized NLP with attention mechanisms.",
                    metadata={"source": "transformers"}),
            Document(page_content="GPT models are large language models trained on vast amounts of text data.",
                    metadata={"source": "gpt_info"}),
            Document(page_content="RAG combines retrieval with generation for better accuracy.",
                    metadata={"source": "rag_concept"}),
            Document(page_content="Computer vision enables machines to interpret visual information.",
                    metadata={"source": "cv_basics"}),
            Document(page_content="Reinforcement learning trains agents through rewards and penalties.",
                    metadata={"source": "rl_intro"}),
        ]
        self.embeddings = OpenAIEmbeddings()
    
    def similarity_search(self, query: str, k: int = 3) -> List[Document]:
        """Mock similarity search"""
        # Simple keyword matching for demonstration
        query_lower = query.lower()
        scored_docs = []
        
        for doc in self.documents:
            content_lower = doc.page_content.lower()
            score = sum(1 for word in query_lower.split() if word in content_lower)
            scored_docs.append((score, doc))
        
        scored_docs.sort(key=lambda x: x[0], reverse=True)
        return [doc for _, doc in scored_docs[:k]]


class RetrievalInterleavingAgent:
    """Agent that interleaves retrieval with generation"""
    
    def __init__(self, vector_store: Optional[MockVectorStore] = None,
                 max_retrievals: int = 5):
        """
        Initialize interleaving agent
        
        Args:
            vector_store: Vector store for retrieval
            max_retrievals: Maximum number of retrieval steps
        """
        self.vector_store = vector_store or MockVectorStore()
        self.max_retrievals = max_retrievals
        self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
        
        # Prompts
        self.decision_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are an AI that decides when to retrieve more information.
            Analyze the current generation state and decide if you need more information.
            
            Respond with one of: RETRIEVE, CONTINUE, COMPLETE
            If RETRIEVE, also provide a search query."""),
            ("user", """Query: {query}

Current generation:
{current_generation}

Context so far:
{context}

Decision (RETRIEVE/CONTINUE/COMPLETE) and query if retrieving:""")
        ])
        
        self.generation_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a helpful AI assistant. Generate content based on 
            the query and available context. Be factual and cite information when possible."""),
            ("user", """Query: {query}

Available context:
{context}

Previous generation:
{previous_generation}

Continue or complete the response:""")
        ])
        
        self.decision_chain = self.decision_prompt | self.llm | StrOutputParser()
        self.generation_chain = self.generation_prompt | self.llm | StrOutputParser()
    
    def generate_with_interleaved_retrieval(self, query: str) -> InterleavedResult:
        """Generate response with interleaved retrieval"""
        retrieval_steps: List[RetrievalStep] = []
        generation_steps: List[GenerationStep] = []
        
        current_generation = ""
        retrieved_context = ""
        step_counter = 0
        
        # Initial generation attempt
        initial_gen = self._generate(query, "", "")
        current_generation = initial_gen
        generation_steps.append(GenerationStep(
            content=initial_gen,
            timestamp=step_counter,
            used_retrieval=False
        ))
        step_counter += 1
        
        # Interleaving loop
        for retrieval_count in range(self.max_retrievals):
            # Decide if we need more information
            decision, retrieval_query = self._decide_next_step(
                query, current_generation, retrieved_context
            )
            
            if decision == RetrievalDecision.COMPLETE:
                break
            
            if decision == RetrievalDecision.RETRIEVE and retrieval_query:
                # Retrieve documents
                docs = self.vector_store.similarity_search(retrieval_query, k=2)
                
                retrieval_steps.append(RetrievalStep(
                    query=retrieval_query,
                    documents=docs,
                    timestamp=step_counter,
                    reasoning=f"Needed information about: {retrieval_query}"
                ))
                step_counter += 1
                
                # Update context
                new_context = "\n\n".join([doc.page_content for doc in docs])
                retrieved_context += "\n\n" + new_context
                
                # Generate with new context
                new_generation = self._generate(query, retrieved_context, current_generation)
                current_generation = new_generation
                
                generation_steps.append(GenerationStep(
                    content=new_generation,
                    timestamp=step_counter,
                    used_retrieval=True
                ))
                step_counter += 1
            
            elif decision == RetrievalDecision.CONTINUE_GENERATION:
                # Continue generation without retrieval
                new_generation = self._generate(query, retrieved_context, current_generation)
                current_generation = new_generation
                
                generation_steps.append(GenerationStep(
                    content=new_generation,
                    timestamp=step_counter,
                    used_retrieval=False
                ))
                step_counter += 1
        
        # Estimate tokens
        total_tokens = sum(len(step.content.split()) for step in generation_steps) * 1.3
        total_tokens += sum(len(step.query.split()) for step in retrieval_steps) * 1.3
        
        return InterleavedResult(
            final_output=current_generation,
            retrieval_steps=retrieval_steps,
            generation_steps=generation_steps,
            total_retrievals=len(retrieval_steps),
            total_tokens_estimated=int(total_tokens)
        )
    
    def _decide_next_step(self, query: str, current_generation: str,
                         context: str) -> Tuple[RetrievalDecision, Optional[str]]:
        """Decide whether to retrieve, continue, or complete"""
        decision_text = self.decision_chain.invoke({
            "query": query,
            "current_generation": current_generation[:200],
            "context": context[:300] if context else "No context yet"
        })
        
        # Parse decision
        decision_text_upper = decision_text.upper()
        
        if "COMPLETE" in decision_text_upper:
            return RetrievalDecision.COMPLETE, None
        elif "RETRIEVE" in decision_text_upper:
            # Extract query from response
            lines = decision_text.split('\n')
            query_line = next((line for line in lines if "query" in line.lower()), None)
            if query_line and ':' in query_line:
                retrieval_query = query_line.split(':', 1)[1].strip()
            else:
                retrieval_query = query  # Fallback to original query
            return RetrievalDecision.RETRIEVE, retrieval_query
        else:
            return RetrievalDecision.CONTINUE_GENERATION, None
    
    def _generate(self, query: str, context: str, previous: str) -> str:
        """Generate content with available context"""
        response = self.generation_chain.invoke({
            "query": query,
            "context": context if context else "No additional context available",
            "previous_generation": previous if previous else "Starting response"
        })
        return response


def demonstrate_retrieval_interleaving():
    """Demonstrate retrieval interleaving pattern"""
    print("=" * 80)
    print("RETRIEVAL INTERLEAVING PATTERN DEMONSTRATION")
    print("=" * 80)
    
    # Create agent with mock vector store
    agent = RetrievalInterleavingAgent(max_retrievals=3)
    
    # Example 1: Complex query requiring multiple retrievals
    print("\n" + "=" * 80)
    print("Example 1: Multi-Hop Question")
    print("=" * 80)
    
    query1 = "Explain how transformers relate to GPT models and why they're important for NLP"
    
    print(f"\nQuery: {query1}")
    print("\nExecuting with interleaved retrieval...")
    print("-" * 60)
    
    result1 = agent.generate_with_interleaved_retrieval(query1)
    
    print(f"\nFinal Output:")
    print(result1.final_output)
    
    print(f"\n" + "-" * 60)
    print(f"Execution Summary:")
    print(f"  Total retrievals: {result1.total_retrievals}")
    print(f"  Total generation steps: {len(result1.generation_steps)}")
    print(f"  Estimated tokens: {result1.total_tokens_estimated}")
    
    print(f"\nRetrieval Steps:")
    for i, step in enumerate(result1.retrieval_steps, 1):
        print(f"\n  Step {i} (timestamp {step.timestamp}):")
        print(f"    Query: {step.query}")
        print(f"    Retrieved {len(step.documents)} documents")
        for j, doc in enumerate(step.documents, 1):
            print(f"      Doc {j}: {doc.page_content[:60]}...")
    
    print(f"\nGeneration Steps:")
    for i, step in enumerate(result1.generation_steps, 1):
        used = "with retrieval" if step.used_retrieval else "without retrieval"
        print(f"  Step {i} (timestamp {step.timestamp}): Generated {used}")
        print(f"    Content: {step.content[:80]}...")
    
    # Example 2: Simple query (fewer retrievals)
    print("\n" + "=" * 80)
    print("Example 2: Simple Query")
    print("=" * 80)
    
    query2 = "What is machine learning?"
    
    print(f"\nQuery: {query2}")
    result2 = agent.generate_with_interleaved_retrieval(query2)
    
    print(f"\nFinal Output:")
    print(result2.final_output)
    
    print(f"\nExecution Summary:")
    print(f"  Total retrievals: {result2.total_retrievals}")
    print(f"  (Simpler query required fewer retrievals)")
    
    # Example 3: Comparing with standard RAG
    print("\n" + "=" * 80)
    print("Example 3: Comparison - Interleaved vs Standard RAG")
    print("=" * 80)
    
    query3 = "Explain deep learning and how it relates to computer vision and NLP"
    
    print(f"\nQuery: {query3}")
    
    # Interleaved approach
    print("\n1. Interleaved Retrieval Approach:")
    result_interleaved = agent.generate_with_interleaved_retrieval(query3)
    print(f"   Retrievals: {result_interleaved.total_retrievals}")
    print(f"   Output: {result_interleaved.final_output[:150]}...")
    
    # Standard RAG (single retrieval)
    print("\n2. Standard RAG (single retrieval):")
    docs = agent.vector_store.similarity_search(query3, k=3)
    context = "\n\n".join([doc.page_content for doc in docs])
    
    standard_prompt = ChatPromptTemplate.from_messages([
        ("system", "You are a helpful AI assistant. Use the provided context to answer the query."),
        ("user", "Context:\n{context}\n\nQuery: {query}\n\nAnswer:")
    ])
    standard_chain = standard_prompt | agent.llm | StrOutputParser()
    standard_result = standard_chain.invoke({"context": context, "query": query3})
    
    print(f"   Retrievals: 1")
    print(f"   Output: {standard_result[:150]}...")
    
    print("\n   Analysis:")
    print(f"   - Interleaved made {result_interleaved.total_retrievals} targeted retrievals")
    print(f"   - Standard RAG made 1 broad retrieval")
    print(f"   - Interleaved can gather more specific information as needed")
    
    # Example 4: Retrieval pattern analysis
    print("\n" + "=" * 80)
    print("Example 4: Retrieval Pattern Analysis")
    print("=" * 80)
    
    queries = [
        "What is NLP?",
        "Explain the relationship between machine learning, deep learning, and transformers",
        "How does reinforcement learning work?"
    ]
    
    print("\nAnalyzing retrieval patterns for different query types:\n")
    
    for query in queries:
        result = agent.generate_with_interleaved_retrieval(query)
        print(f"Query: {query}")
        print(f"  Retrievals: {result.total_retrievals}")
        print(f"  Generation steps: {len(result.generation_steps)}")
        
        if result.retrieval_steps:
            print(f"  Retrieval queries:")
            for step in result.retrieval_steps:
                print(f"    - {step.query}")
        print()
    
    # Summary
    print("\n" + "=" * 80)
    print("SUMMARY")
    print("=" * 80)
    print("""
The Retrieval Interleaving pattern enables:
✓ Dynamic, adaptive information gathering
✓ Multiple retrieval steps during generation
✓ More targeted and relevant retrievals
✓ Better handling of complex multi-hop queries
✓ Reduced initial retrieval overhead
✓ Context-aware retrieval decisions
✓ Improved accuracy for knowledge-intensive tasks

This pattern is valuable for:
- Complex research questions
- Multi-hop reasoning tasks
- Long-form content generation
- Dynamic information needs
- Iterative question answering
- Knowledge synthesis tasks
- Exploratory queries

Compared to standard RAG:
- Standard RAG: One retrieval upfront, simpler, faster
- Interleaved: Multiple targeted retrievals, more complex, better for complex queries
    """)


if __name__ == "__main__":
    demonstrate_retrieval_interleaving()

    

Agentic Design Pattern: Retrieval Interleaving

Pattern 169: Retrieval Interleaving – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Retrieval Interleaving pattern implementation.

---

Pattern Description

Pattern 169: Retrieval Interleaving

Description: The Retrieval Interleaving pattern dynamically retrieves information during generation, interleaving retrieval steps with reasoning and generation steps. Unlike standard RAG which retrieves once upfront, this pattern retrieves multiple times as needed based on the generation process.

Components: 1. Generation Monitor: Tracks generation progress 2. Retrieval Trigger: Decides when to retrieve 3. Query Generator: Creates retrieval queries 4. Retriever: Fetches relevant information 5. Context Integrator: Merges retrieved content 6. Iteration Controller: Manages interleaving loops

Use Cases: – Complex multi-hop reasoning – Long-form content generation – Research-intensive tasks – Dynamic information needs – Iterative question answering – Knowledge synthesis

Benefits: – Adaptive information gathering – Better context relevance – Handles complex queries – Reduces initial retrieval overhead – More accurate results

Trade-offs: – Multiple retrieval calls – Increased latency – Complex orchestration – Higher costs – Requires good triggering logic

LangChain Implementation: Implements iterative retrieval with LangChain’s retrieval components. Uses LLM to decide when and what to retrieve during generation.

---

File Statistics

• Total Lines: 453
• Classes: 6
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 169: Retrieval Interleaving

Description:
    The Retrieval Interleaving pattern dynamically retrieves information during
    generation, interleaving retrieval steps with reasoning and generation steps.
    Unlike standard RAG which retrieves once upfront, this pattern retrieves
    multiple times as needed based on the generation process.

Components:
    1. Generation Monitor: Tracks generation progress
    2. Retrieval Trigger: Decides when to retrieve
    3. Query Generator: Creates retrieval queries
    4. Retriever: Fetches relevant information
    5. Context Integrator: Merges retrieved content
    6. Iteration Controller: Manages interleaving loops

Use Cases:
    - Complex multi-hop reasoning
    - Long-form content generation
    - Research-intensive tasks
    - Dynamic information needs
    - Iterative question answering
    - Knowledge synthesis

Benefits:
    - Adaptive information gathering
    - Better context relevance
    - Handles complex queries
    - Reduces initial retrieval overhead
    - More accurate results

Trade-offs:
    - Multiple retrieval calls
    - Increased latency
    - Complex orchestration
    - Higher costs
    - Requires good triggering logic

LangChain Implementation:
    Implements iterative retrieval with LangChain's retrieval components.
    Uses LLM to decide when and what to retrieve during generation.
"""

Purpose: This module implements the Retrieval Interleaving pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import dataclasses.dataclass
import dataclasses.field
import enum.Enum
import langchain_core.documents.Document
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Optional
import typing.Tuple

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – enum.Enum: Enumeration support – langchain_core.documents.Document: Regular expression operations – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Tuple: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI
from langchain_openai.OpenAIEmbeddings

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: RetrievalDecision

Lines 59-63

Inherits from: Enum

Purpose:

Decision on whether to retrieve

---

Class: RetrievalStep

Lines 67-72

Purpose:

Represents a retrieval step

---

Class: GenerationStep

Lines 76-80

Purpose:

Represents a generation step

---

Class: InterleavedResult

Lines 84-90

Purpose:

Result of interleaved retrieval-generation

---

Class: MockVectorStore

Lines 93-130

Purpose:

Mock vector store for demonstration

Methods:

__init__(self)

Lines 96-116

Purpose:

Initialize with sample documents

Parameters:

similarity_search(self, query, k)

Lines 118-130

Purpose:

Mock similarity search

Parameters: – query: Input query or question – k: K

Returns: <ast.Subscript object at 0x000001A06E0ABAD0>

---

Class: RetrievalInterleavingAgent

Lines 133-297

Purpose:

Agent that interleaves retrieval with generation

Methods:

__init__(self, vector_store, max_retrievals)

Lines 136-182

Purpose:

Initialize interleaving agent

Args:
    vector_store: Vector store for retrieval
    max_retrievals: Maximum number of retrieval steps

Parameters: – vector_store: Vector store – max_retrievals: Max retrievals

generate_with_interleaved_retrieval(self, query)

Lines 184-262

Purpose:

Generate response with interleaved retrieval

Parameters: – query: Input query or question

Returns: InterleavedResult

_decide_next_step(self, query, current_generation, context)

Lines 264-288

Purpose:

Decide whether to retrieve, continue, or complete

Parameters: – query: Input query or question – current_generation: Current generation – context: Context

Returns: <ast.Subscript object at 0x000001A06D60AC90>

_generate(self, query, context, previous)

Lines 290-297

Purpose:

Generate content with available context

Parameters: – query: Input query or question – context: Context – previous: Previous

Returns: str

---

Functions

demonstrate_retrieval_interleaving()

Lines 300-448

Purpose:

Demonstrate retrieval interleaving pattern

---

Key Concepts Summary

Pattern Overview

The Retrieval Interleaving pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

---

Best Practices Demonstrated

1. Environment Configuration

• ✅ Use .env files for API keys
• ✅ Never hardcode credentials
• ✅ Validate environment variables on startup

2. Error Handling

• ✅ Graceful degradation on failures
• ✅ Informative error messages
• ✅ Logging for debugging

3. Prompt Engineering

• ✅ Clear, specific instructions
• ✅ Structured output formats
• ✅ Few-shot examples when beneficial

4. Code Organization

• ✅ Separation of concerns (class-based design)
• ✅ Type hints for clarity
• ✅ Docstrings for documentation

5. Performance Optimization

• ✅ Appropriate temperature settings
• ✅ Token budget management
• ✅ Caching where applicable

---

LangChain Components Used

• ChatOpenAI: OpenAI model interface
• Chains: Sequential component composition

Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development

---

Execution Flow

1. Initialize agent with configuration
   ↓
2. Load environment and dependencies
   ↓
3. Prepare prompt with input data
   ↓
4. Invoke LLM through LangChain
   ↓
5. Parse and validate response
   ↓
6. Return formatted result

Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation

---

Performance Considerations

Computational Cost

• LLM Calls: Primary cost factor (based on tokens)
• Latency: Depends on model and input size
• Memory: Minimal for most patterns

Optimization Strategies

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

• GPT-3.5-turbo: ~$0.002 per request
• GPT-4: ~$0.03 per request
• GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

---

Further Improvements

Production Enhancements

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 168
• Pattern 170

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Retrieval Interleaving pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns

🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability

📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/

---

File: 169_retrieval_interleaving.py
Total Lines: 453
Pattern Category: AI Agent Design Pattern #169

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 168: Constitutional Chain

Description:
    The Constitutional Chain pattern ensures AI outputs align with predefined principles,
    values, and rules through iterative self-critique and revision. The agent generates
    an initial response, critiques it against constitutional principles, and refines it
    until it meets all requirements.

Components:
    1. Constitution: Set of principles and rules
    2. Generator: Creates initial responses
    3. Critic: Evaluates responses against constitution
    4. Reviser: Improves responses based on critique
    5. Compliance Checker: Validates final alignment
    6. Iteration Manager: Controls refinement loops

Use Cases:
    - Ethical AI systems
    - Content moderation
    - Policy-compliant generation
    - Bias reduction
    - Safety-critical applications
    - Regulated industries

Benefits:
    - Value alignment
    - Explicit principles
    - Iterative improvement
    - Audit trail
    - Consistent behavior
    - Safety guarantees

Trade-offs:
    - Multiple LLM calls
    - Increased latency
    - May be overly cautious
    - Requires good constitution
    - Complexity overhead

LangChain Implementation:
    Implements iterative critique-revision loops with LangChain. Uses structured
    prompts for critique and revision stages with constitutional principles.
"""

import os
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from enum import Enum
from dotenv import load_dotenv

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

load_dotenv()


class PrincipleType(Enum):
    """Types of constitutional principles"""
    ETHICAL = "ethical"
    SAFETY = "safety"
    ACCURACY = "accuracy"
    BIAS = "bias"
    TONE = "tone"
    LEGAL = "legal"
    PRIVACY = "privacy"


class ComplianceLevel(Enum):
    """Compliance assessment levels"""
    COMPLIANT = "compliant"
    MINOR_ISSUES = "minor_issues"
    MAJOR_VIOLATIONS = "major_violations"
    CRITICAL = "critical"


@dataclass
class ConstitutionalPrinciple:
    """Represents a principle in the constitution"""
    name: str
    description: str
    principle_type: PrincipleType
    priority: int = 5  # 1-10, higher is more important
    examples_good: List[str] = field(default_factory=list)
    examples_bad: List[str] = field(default_factory=list)


@dataclass
class CritiqueResult:
    """Result of critiquing a response"""
    compliance_level: ComplianceLevel
    violations: List[str]
    suggestions: List[str]
    score: float  # 0-1 scale
    critique_text: str


@dataclass
class ConstitutionalResponse:
    """Final response with compliance information"""
    response: str
    iterations: int
    final_compliance: ComplianceLevel
    improvements_made: List[str]
    critique_history: List[CritiqueResult]


class Constitution:
    """Defines a set of principles for AI behavior"""
    
    def __init__(self, name: str):
        self.name = name
        self.principles: List[ConstitutionalPrinciple] = []
    
    def add_principle(self, principle: ConstitutionalPrinciple):
        """Add a principle to the constitution"""
        self.principles.append(principle)
    
    def get_principles_text(self) -> str:
        """Get formatted text of all principles"""
        lines = [f"Constitution: {self.name}", "=" * 60, ""]
        
        for i, principle in enumerate(sorted(self.principles, 
                                            key=lambda p: p.priority, 
                                            reverse=True), 1):
            lines.append(f"{i}. {principle.name} (Priority: {principle.priority})")
            lines.append(f"   Type: {principle.principle_type.value}")
            lines.append(f"   {principle.description}")
            if principle.examples_good:
                lines.append(f"   Good: {principle.examples_good[0]}")
            if principle.examples_bad:
                lines.append(f"   Bad: {principle.examples_bad[0]}")
            lines.append("")
        
        return "\n".join(lines)


class ConstitutionalChain:
    """Implements constitutional AI pattern with critique-revision loops"""
    
    def __init__(self, constitution: Constitution, max_iterations: int = 3):
        """
        Initialize constitutional chain
        
        Args:
            constitution: The constitution to follow
            max_iterations: Maximum number of revision iterations
        """
        self.constitution = constitution
        self.max_iterations = max_iterations
        self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
        
        # Prompts
        self.generation_prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a helpful AI assistant."),
            ("user", "{query}")
        ])
        
        self.critique_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a constitutional AI critic. Evaluate the response 
            against the following principles:

{principles}

Identify any violations and provide specific suggestions for improvement."""),
            ("user", """Query: {query}

Response to evaluate:
{response}

Provide:
1. Compliance level: compliant, minor_issues, major_violations, or critical
2. List of violations (if any)
3. Specific suggestions for improvement
4. Overall score (0-1)

Format your response as:
COMPLIANCE: [level]
VIOLATIONS: [list violations or "none"]
SUGGESTIONS: [list suggestions]
SCORE: [0-1 score]
CRITIQUE: [detailed explanation]""")
        ])
        
        self.revision_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are an AI assistant that revises responses to align with 
            constitutional principles. Improve the response based on the critique while 
            maintaining helpfulness and relevance."""),
            ("user", """Original query: {query}

Previous response:
{response}

Critique:
{critique}

Revise the response to address all issues while staying helpful and relevant.""")
        ])
        
        self.generation_chain = self.generation_prompt | self.llm | StrOutputParser()
        self.critique_chain = self.critique_prompt | self.llm | StrOutputParser()
        self.revision_chain = self.revision_prompt | self.llm | StrOutputParser()
    
    def generate_constitutional_response(self, query: str) -> ConstitutionalResponse:
        """Generate a response that complies with constitutional principles"""
        # Generate initial response
        response = self.generation_chain.invoke({"query": query})
        
        critique_history = []
        improvements = []
        iterations = 0
        
        principles_text = self.constitution.get_principles_text()
        
        for iteration in range(self.max_iterations):
            iterations += 1
            
            # Critique current response
            critique = self._critique_response(query, response, principles_text)
            critique_history.append(critique)
            
            # Check if compliant
            if critique.compliance_level == ComplianceLevel.COMPLIANT:
                break
            
            if critique.compliance_level == ComplianceLevel.CRITICAL:
                # Critical issues require immediate attention
                improvements.append(f"Iteration {iteration + 1}: Fixed critical issues")
            elif critique.compliance_level == ComplianceLevel.MAJOR_VIOLATIONS:
                improvements.append(f"Iteration {iteration + 1}: Fixed major violations")
            else:
                improvements.append(f"Iteration {iteration + 1}: Addressed minor issues")
            
            # Revise response
            response = self._revise_response(query, response, critique.critique_text)
        
        # Final critique
        final_critique = critique_history[-1]
        
        return ConstitutionalResponse(
            response=response,
            iterations=iterations,
            final_compliance=final_critique.compliance_level,
            improvements_made=improvements,
            critique_history=critique_history
        )
    
    def _critique_response(self, query: str, response: str, 
                          principles: str) -> CritiqueResult:
        """Critique a response against constitutional principles"""
        critique_text = self.critique_chain.invoke({
            "query": query,
            "response": response,
            "principles": principles
        })
        
        # Parse critique
        compliance_level = ComplianceLevel.COMPLIANT
        violations = []
        suggestions = []
        score = 1.0
        
        for line in critique_text.split('\n'):
            line = line.strip()
            if line.startswith('COMPLIANCE:'):
                level_str = line.split(':', 1)[1].strip().lower()
                if 'critical' in level_str:
                    compliance_level = ComplianceLevel.CRITICAL
                elif 'major' in level_str:
                    compliance_level = ComplianceLevel.MAJOR_VIOLATIONS
                elif 'minor' in level_str:
                    compliance_level = ComplianceLevel.MINOR_ISSUES
                else:
                    compliance_level = ComplianceLevel.COMPLIANT
            elif line.startswith('VIOLATIONS:'):
                viol_text = line.split(':', 1)[1].strip()
                if viol_text.lower() != 'none':
                    violations = [v.strip() for v in viol_text.split(',')]
            elif line.startswith('SUGGESTIONS:'):
                sugg_text = line.split(':', 1)[1].strip()
                suggestions = [s.strip() for s in sugg_text.split(',')]
            elif line.startswith('SCORE:'):
                try:
                    score = float(line.split(':', 1)[1].strip())
                except:
                    score = 0.8
        
        return CritiqueResult(
            compliance_level=compliance_level,
            violations=violations,
            suggestions=suggestions,
            score=score,
            critique_text=critique_text
        )
    
    def _revise_response(self, query: str, response: str, critique: str) -> str:
        """Revise response based on critique"""
        revised = self.revision_chain.invoke({
            "query": query,
            "response": response,
            "critique": critique
        })
        return revised


def demonstrate_constitutional_chain():
    """Demonstrate constitutional chain pattern"""
    print("=" * 80)
    print("CONSTITUTIONAL CHAIN PATTERN DEMONSTRATION")
    print("=" * 80)
    
    # Example 1: Basic constitutional AI
    print("\n" + "=" * 80)
    print("Example 1: Basic Constitutional AI")
    print("=" * 80)
    
    # Create a simple constitution
    basic_constitution = Constitution("Basic Safety Constitution")
    
    basic_constitution.add_principle(ConstitutionalPrinciple(
        name="Helpful and Harmless",
        description="Responses must be helpful while avoiding harmful content",
        principle_type=PrincipleType.ETHICAL,
        priority=10,
        examples_good=["I can help you learn programming"],
        examples_bad=["I can help you hack systems"]
    ))
    
    basic_constitution.add_principle(ConstitutionalPrinciple(
        name="No Misinformation",
        description="Avoid spreading false or misleading information",
        principle_type=PrincipleType.ACCURACY,
        priority=9,
        examples_good=["According to research..."],
        examples_bad=["I think this is true but not sure..."]
    ))
    
    basic_constitution.add_principle(ConstitutionalPrinciple(
        name="Respectful Tone",
        description="Maintain respectful and professional communication",
        principle_type=PrincipleType.TONE,
        priority=7,
        examples_good=["Let me help you understand this"],
        examples_bad=["That's a stupid question"]
    ))
    
    print("\n" + basic_constitution.get_principles_text())
    
    # Create chain
    chain = ConstitutionalChain(basic_constitution, max_iterations=3)
    
    # Test query
    query = "Tell me how to get rich quick with cryptocurrency"
    
    print("\nQuery:", query)
    print("\nGenerating constitutional response...")
    print("-" * 60)
    
    result = chain.generate_constitutional_response(query)
    
    print(f"\nIterations required: {result.iterations}")
    print(f"Final compliance: {result.final_compliance.value}")
    
    if result.improvements_made:
        print("\nImprovements made:")
        for improvement in result.improvements_made:
            print(f"  - {improvement}")
    
    print(f"\nFinal response:")
    print(result.response)
    
    print("\n" + "-" * 60)
    print("Critique History:")
    for i, critique in enumerate(result.critique_history, 1):
        print(f"\nIteration {i}:")
        print(f"  Compliance: {critique.compliance_level.value}")
        print(f"  Score: {critique.score:.2f}")
        if critique.violations:
            print(f"  Violations: {', '.join(critique.violations[:2])}")
        if critique.suggestions:
            print(f"  Suggestions: {', '.join(critique.suggestions[:2])}")
    
    # Example 2: Content moderation constitution
    print("\n" + "=" * 80)
    print("Example 2: Content Moderation")
    print("=" * 80)
    
    moderation_constitution = Constitution("Content Moderation")
    
    moderation_constitution.add_principle(ConstitutionalPrinciple(
        name="No Hate Speech",
        description="Responses must not contain discriminatory or hateful content",
        principle_type=PrincipleType.ETHICAL,
        priority=10
    ))
    
    moderation_constitution.add_principle(ConstitutionalPrinciple(
        name="Age Appropriate",
        description="Content should be suitable for general audiences",
        principle_type=PrincipleType.SAFETY,
        priority=9
    ))
    
    moderation_constitution.add_principle(ConstitutionalPrinciple(
        name="No Personal Attacks",
        description="Avoid personal attacks or aggressive language",
        principle_type=PrincipleType.TONE,
        priority=8
    ))
    
    mod_chain = ConstitutionalChain(moderation_constitution, max_iterations=2)
    
    queries = [
        "What do you think about people who disagree with me?",
        "How do I respond to criticism online?"
    ]
    
    for query in queries:
        print(f"\nQuery: {query}")
        result = mod_chain.generate_constitutional_response(query)
        print(f"Compliance: {result.final_compliance.value}")
        print(f"Response: {result.response[:120]}...")
    
    # Example 3: Professional communication constitution
    print("\n" + "=" * 80)
    print("Example 3: Professional Communication")
    print("=" * 80)
    
    prof_constitution = Constitution("Professional Communication")
    
    prof_constitution.add_principle(ConstitutionalPrinciple(
        name="Clarity",
        description="Responses must be clear and unambiguous",
        principle_type=PrincipleType.TONE,
        priority=8
    ))
    
    prof_constitution.add_principle(ConstitutionalPrinciple(
        name="Evidence-Based",
        description="Claims should be supported by evidence or clearly marked as opinions",
        principle_type=PrincipleType.ACCURACY,
        priority=9
    ))
    
    prof_constitution.add_principle(ConstitutionalPrinciple(
        name="Conciseness",
        description="Responses should be concise while complete",
        principle_type=PrincipleType.TONE,
        priority=6
    ))
    
    prof_chain = ConstitutionalChain(prof_constitution, max_iterations=3)
    
    query = "Explain quantum computing"
    print(f"\nQuery: {query}")
    print("\nGenerating professional response...")
    
    result = prof_chain.generate_constitutional_response(query)
    
    print(f"\nIterations: {result.iterations}")
    print(f"Final compliance: {result.final_compliance.value}")
    print(f"\nResponse:\n{result.response}")
    
    # Example 4: Comparing with and without constitutional chain
    print("\n" + "=" * 80)
    print("Example 4: Comparison - With vs Without Constitution")
    print("=" * 80)
    
    query = "Should I invest all my money in a single stock?"
    
    # Without constitution
    print(f"\nQuery: {query}")
    print("\n1. Without Constitutional Chain:")
    basic_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
    basic_chain = ChatPromptTemplate.from_messages([
        ("user", "{query}")
    ]) | basic_llm | StrOutputParser()
    
    basic_response = basic_chain.invoke({"query": query})
    print(f"   {basic_response[:200]}...")
    
    # With constitution
    print("\n2. With Constitutional Chain:")
    safety_constitution = Constitution("Financial Safety")
    safety_constitution.add_principle(ConstitutionalPrinciple(
        name="Risk Disclosure",
        description="Always disclose risks in financial advice",
        principle_type=PrincipleType.ETHICAL,
        priority=10
    ))
    safety_constitution.add_principle(ConstitutionalPrinciple(
        name="No Guarantees",
        description="Avoid making guarantees about financial outcomes",
        principle_type=PrincipleType.ACCURACY,
        priority=9
    ))
    
    safety_chain = ConstitutionalChain(safety_constitution, max_iterations=2)
    result = safety_chain.generate_constitutional_response(query)
    print(f"   {result.response[:200]}...")
    print(f"\n   Iterations: {result.iterations}")
    print(f"   Compliance: {result.final_compliance.value}")
    
    # Summary
    print("\n" + "=" * 80)
    print("SUMMARY")
    print("=" * 80)
    print("""
The Constitutional Chain pattern enables:
✓ Value-aligned AI responses
✓ Iterative self-critique and improvement
✓ Explicit principle enforcement
✓ Audit trail of improvements
✓ Consistent ethical behavior
✓ Safety and compliance guarantees
✓ Transparent reasoning process

This pattern is valuable for:
- Production AI systems requiring safety
- Content moderation platforms
- Regulated industries (finance, healthcare)
- Ethical AI development
- Bias mitigation
- Policy-compliant generation
- High-stakes decision support
    """)


if __name__ == "__main__":
    demonstrate_constitutional_chain()

    

Agentic Design Pattern: Constitutional Chain

Pattern 168: Constitutional Chain – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Constitutional Chain pattern implementation.

---

Pattern Description

Pattern 168: Constitutional Chain

Description: The Constitutional Chain pattern ensures AI outputs align with predefined principles, values, and rules through iterative self-critique and revision. The agent generates an initial response, critiques it against constitutional principles, and refines it until it meets all requirements.

Components: 1. Constitution: Set of principles and rules 2. Generator: Creates initial responses 3. Critic: Evaluates responses against constitution 4. Reviser: Improves responses based on critique 5. Compliance Checker: Validates final alignment 6. Iteration Manager: Controls refinement loops

Use Cases: – Ethical AI systems – Content moderation – Policy-compliant generation – Bias reduction – Safety-critical applications – Regulated industries

Benefits: – Value alignment – Explicit principles – Iterative improvement – Audit trail – Consistent behavior – Safety guarantees

Trade-offs: – Multiple LLM calls – Increased latency – May be overly cautious – Requires good constitution – Complexity overhead

LangChain Implementation: Implements iterative critique-revision loops with LangChain. Uses structured prompts for critique and revision stages with constitutional principles.

---

File Statistics

• Total Lines: 532
• Classes: 7
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 168: Constitutional Chain

Description:
    The Constitutional Chain pattern ensures AI outputs align with predefined principles,
    values, and rules through iterative self-critique and revision. The agent generates
    an initial response, critiques it against constitutional principles, and refines it
    until it meets all requirements.

Components:
    1. Constitution: Set of principles and rules
    2. Generator: Creates initial responses
    3. Critic: Evaluates responses against constitution
    4. Reviser: Improves responses based on critique
    5. Compliance Checker: Validates final alignment
    6. Iteration Manager: Controls refinement loops

Use Cases:
    - Ethical AI systems
    - Content moderation
    - Policy-compliant generation
    - Bias reduction
    - Safety-critical applications
    - Regulated industries

Benefits:
    - Value alignment
    - Explicit principles
    - Iterative improvement
    - Audit trail
    - Consistent behavior
    - Safety guarantees

Trade-offs:
    - Multiple LLM calls
    - Increased latency
    - May be overly cautious
    - Requires good constitution
    - Complexity overhead

LangChain Implementation:
    Implements iterative critique-revision loops with LangChain. Uses structured
    prompts for critique and revision stages with constitutional principles.
"""

Purpose: This module implements the Constitutional Chain pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import dataclasses.dataclass
import dataclasses.field
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Optional

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – enum.Enum: Enumeration support – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: PrincipleType

Lines 59-67

Inherits from: Enum

Purpose:

Types of constitutional principles

---

Class: ComplianceLevel

Lines 70-75

Inherits from: Enum

Purpose:

Compliance assessment levels

---

Class: ConstitutionalPrinciple

Lines 79-86

Purpose:

Represents a principle in the constitution

---

Class: CritiqueResult

Lines 90-96

Purpose:

Result of critiquing a response

---

Class: ConstitutionalResponse

Lines 100-106

Purpose:

Final response with compliance information

---

Class: Constitution

Lines 109-136

Purpose:

Defines a set of principles for AI behavior

Methods:

__init__(self, name)

Lines 112-114

Purpose: Initialize the class instance with required parameters.

Parameters: – name: Name

add_principle(self, principle)

Lines 116-118

Purpose:

Add a principle to the constitution

Parameters: – principle: Principle

get_principles_text(self)

Lines 120-136

Purpose:

Get formatted text of all principles

Parameters:

Returns: str

---

Class: ConstitutionalChain

Lines 139-304

Purpose:

Implements constitutional AI pattern with critique-revision loops

Methods:

__init__(self, constitution, max_iterations)

Lines 142-203

Purpose:

Initialize constitutional chain

Args:
    constitution: The constitution to follow
    max_iterations: Maximum number of revision iterations

Parameters: – constitution: Constitution – max_iterations: Max iterations

generate_constitutional_response(self, query)

Lines 205-247

Purpose:

Generate a response that complies with constitutional principles

Parameters: – query: Input query or question

Returns: ConstitutionalResponse

_critique_response(self, query, response, principles)

Lines 249-295

Purpose:

Critique a response against constitutional principles

Parameters: – query: Input query or question – response: Response – principles: Principles

Returns: CritiqueResult

_revise_response(self, query, response, critique)

Lines 297-304

Purpose:

Revise response based on critique

Parameters: – query: Input query or question – response: Response – critique: Critique

Returns: str

---

Functions

demonstrate_constitutional_chain()

Lines 307-527

Purpose:

Demonstrate constitutional chain pattern

---

Key Concepts Summary

Pattern Overview

The Constitutional Chain pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

---

Best Practices Demonstrated

1. Environment Configuration

• ✅ Use .env files for API keys
• ✅ Never hardcode credentials
• ✅ Validate environment variables on startup

2. Error Handling

• ✅ Graceful degradation on failures
• ✅ Informative error messages
• ✅ Logging for debugging

3. Prompt Engineering

• ✅ Clear, specific instructions
• ✅ Structured output formats
• ✅ Few-shot examples when beneficial

4. Code Organization

• ✅ Separation of concerns (class-based design)
• ✅ Type hints for clarity
• ✅ Docstrings for documentation

5. Performance Optimization

• ✅ Appropriate temperature settings
• ✅ Token budget management
• ✅ Caching where applicable

---

LangChain Components Used

• ChatOpenAI: OpenAI model interface
• Chains: Sequential component composition

Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development

---

Execution Flow

1. Initialize agent with configuration
   ↓
2. Load environment and dependencies
   ↓
3. Prepare prompt with input data
   ↓
4. Invoke LLM through LangChain
   ↓
5. Parse and validate response
   ↓
6. Return formatted result

Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation

---

Performance Considerations

Computational Cost

• LLM Calls: Primary cost factor (based on tokens)
• Latency: Depends on model and input size
• Memory: Minimal for most patterns

Optimization Strategies

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

• GPT-3.5-turbo: ~$0.002 per request
• GPT-4: ~$0.03 per request
• GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

---

Further Improvements

Production Enhancements

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 167
• Pattern 169

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Constitutional Chain pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns

🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability

📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/

---

File: 168_constitutional_chain.py
Total Lines: 532
Pattern Category: AI Agent Design Pattern #168

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 167: Agentic Workflows

Description:
    The Agentic Workflows pattern implements complex, multi-step processes where agents
    autonomously navigate through workflow states, make decisions, handle branching logic,
    and coordinate multiple tasks. It uses state machines and graph-based execution for
    sophisticated task orchestration.

Components:
    1. Workflow Definition: Graph structure with states and transitions
    2. State Manager: Tracks current workflow state
    3. Decision Engine: Makes branching decisions
    4. Task Executor: Executes tasks at each state
    5. Condition Evaluator: Evaluates transition conditions
    6. Error Handler: Manages workflow failures

Use Cases:
    - Multi-step business processes
    - Complex data pipelines
    - Decision trees and workflows
    - Approval workflows
    - Multi-stage content generation
    - Automated testing workflows

Benefits:
    - Complex process automation
    - Clear workflow visualization
    - Conditional branching support
    - State persistence
    - Error recovery
    - Audit trail

Trade-offs:
    - Increased complexity
    - State management overhead
    - Debugging challenges
    - Performance considerations
    - Requires workflow design

LangChain Implementation:
    Uses LangGraph-inspired state machines with LangChain agents for task execution.
    Implements graph-based workflows with conditional branching and state management.
"""

import os
from typing import List, Dict, Any, Optional, Callable
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime
from dotenv import load_dotenv

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

load_dotenv()


class WorkflowStatus(Enum):
    """Status of workflow execution"""
    NOT_STARTED = "not_started"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"
    PAUSED = "paused"


class TransitionType(Enum):
    """Types of state transitions"""
    AUTOMATIC = "automatic"
    CONDITIONAL = "conditional"
    MANUAL = "manual"


@dataclass
class WorkflowState:
    """Represents a state in the workflow"""
    name: str
    description: str
    task: Optional[Callable] = None
    is_terminal: bool = False
    metadata: Dict[str, Any] = field(default_factory=dict)


@dataclass
class StateTransition:
    """Represents a transition between states"""
    from_state: str
    to_state: str
    condition: Optional[Callable] = None
    transition_type: TransitionType = TransitionType.AUTOMATIC
    description: str = ""


@dataclass
class WorkflowContext:
    """Context passed through workflow execution"""
    data: Dict[str, Any] = field(default_factory=dict)
    history: List[str] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    def add_result(self, state: str, result: Any):
        """Add result from a state execution"""
        self.data[f"{state}_result"] = result
        self.history.append(state)
    
    def get_result(self, state: str) -> Any:
        """Get result from previous state"""
        return self.data.get(f"{state}_result")


class AgenticWorkflow:
    """Implements agentic workflow pattern with state machines"""
    
    def __init__(self, name: str):
        """Initialize workflow"""
        self.name = name
        self.states: Dict[str, WorkflowState] = {}
        self.transitions: List[StateTransition] = []
        self.start_state: Optional[str] = None
        self.current_state: Optional[str] = None
        self.status = WorkflowStatus.NOT_STARTED
        self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
        
        # Execution log
        self.execution_log: List[Dict[str, Any]] = []
    
    def add_state(self, state: WorkflowState, is_start: bool = False):
        """Add a state to the workflow"""
        self.states[state.name] = state
        if is_start:
            self.start_state = state.name
    
    def add_transition(self, transition: StateTransition):
        """Add a transition between states"""
        if transition.from_state not in self.states:
            raise ValueError(f"Unknown from_state: {transition.from_state}")
        if transition.to_state not in self.states:
            raise ValueError(f"Unknown to_state: {transition.to_state}")
        self.transitions.append(transition)
    
    def execute(self, initial_context: Optional[WorkflowContext] = None) -> WorkflowContext:
        """Execute the workflow"""
        if not self.start_state:
            raise ValueError("No start state defined")
        
        context = initial_context or WorkflowContext()
        self.current_state = self.start_state
        self.status = WorkflowStatus.IN_PROGRESS
        
        self._log_execution("workflow_started", {
            "start_state": self.start_state,
            "timestamp": datetime.now().isoformat()
        })
        
        try:
            while True:
                state = self.states[self.current_state]
                
                self._log_execution("state_entered", {
                    "state": self.current_state,
                    "timestamp": datetime.now().isoformat()
                })
                
                # Execute state task
                if state.task:
                    try:
                        result = state.task(context, self.llm)
                        context.add_result(self.current_state, result)
                        
                        self._log_execution("state_executed", {
                            "state": self.current_state,
                            "result": str(result)[:100]
                        })
                    except Exception as e:
                        self._log_execution("state_error", {
                            "state": self.current_state,
                            "error": str(e)
                        })
                        self.status = WorkflowStatus.FAILED
                        raise
                
                # Check if terminal state
                if state.is_terminal:
                    self.status = WorkflowStatus.COMPLETED
                    self._log_execution("workflow_completed", {
                        "final_state": self.current_state,
                        "timestamp": datetime.now().isoformat()
                    })
                    break
                
                # Find next state
                next_state = self._get_next_state(context)
                
                if next_state is None:
                    self.status = WorkflowStatus.FAILED
                    raise ValueError(f"No valid transition from state: {self.current_state}")
                
                self._log_execution("transition", {
                    "from": self.current_state,
                    "to": next_state,
                    "timestamp": datetime.now().isoformat()
                })
                
                self.current_state = next_state
        
        except Exception as e:
            self.status = WorkflowStatus.FAILED
            self._log_execution("workflow_failed", {
                "error": str(e),
                "timestamp": datetime.now().isoformat()
            })
            raise
        
        return context
    
    def _get_next_state(self, context: WorkflowContext) -> Optional[str]:
        """Determine next state based on transitions"""
        # Find applicable transitions from current state
        applicable = [
            t for t in self.transitions 
            if t.from_state == self.current_state
        ]
        
        if not applicable:
            return None
        
        # Evaluate conditions for conditional transitions
        for transition in applicable:
            if transition.transition_type == TransitionType.AUTOMATIC:
                return transition.to_state
            elif transition.transition_type == TransitionType.CONDITIONAL:
                if transition.condition and transition.condition(context):
                    return transition.to_state
        
        # If no condition matched, try first automatic transition
        automatic = [t for t in applicable if t.transition_type == TransitionType.AUTOMATIC]
        if automatic:
            return automatic[0].to_state
        
        return None
    
    def _log_execution(self, event_type: str, data: Dict[str, Any]):
        """Log workflow execution event"""
        self.execution_log.append({
            "event": event_type,
            "data": data
        })
    
    def get_execution_summary(self) -> Dict[str, Any]:
        """Get summary of workflow execution"""
        return {
            "workflow_name": self.name,
            "status": self.status.value,
            "states_visited": len(set(e["data"].get("state", "") for e in self.execution_log)),
            "total_events": len(self.execution_log),
            "current_state": self.current_state
        }
    
    def visualize(self) -> str:
        """Create text visualization of workflow"""
        lines = [f"Workflow: {self.name}", "=" * 60, ""]
        
        lines.append("States:")
        for name, state in self.states.items():
            marker = "→" if name == self.start_state else " "
            terminal = " (terminal)" if state.is_terminal else ""
            lines.append(f"  {marker} {name}{terminal}: {state.description}")
        
        lines.append("\nTransitions:")
        for trans in self.transitions:
            cond = f" [conditional]" if trans.transition_type == TransitionType.CONDITIONAL else ""
            lines.append(f"  {trans.from_state} -> {trans.to_state}{cond}")
            if trans.description:
                lines.append(f"    {trans.description}")
        
        return "\n".join(lines)


def demonstrate_agentic_workflows():
    """Demonstrate agentic workflows"""
    print("=" * 80)
    print("AGENTIC WORKFLOWS PATTERN DEMONSTRATION")
    print("=" * 80)
    
    # Example 1: Content generation workflow
    print("\n" + "=" * 80)
    print("Example 1: Content Generation Workflow")
    print("=" * 80)
    
    def research_task(context: WorkflowContext, llm: ChatOpenAI) -> str:
        """Research phase"""
        topic = context.data.get("topic", "AI")
        prompt = f"List 3 key points about {topic} for a blog post"
        chain = ChatPromptTemplate.from_messages([
            ("user", prompt)
        ]) | llm | StrOutputParser()
        return chain.invoke({})
    
    def outline_task(context: WorkflowContext, llm: ChatOpenAI) -> str:
        """Outline creation"""
        research = context.get_result("research")
        prompt = f"Create a blog post outline based on: {research}"
        chain = ChatPromptTemplate.from_messages([
            ("user", prompt)
        ]) | llm | StrOutputParser()
        return chain.invoke({})
    
    def write_task(context: WorkflowContext, llm: ChatOpenAI) -> str:
        """Writing phase"""
        outline = context.get_result("outline")
        prompt = f"Write the first paragraph based on: {outline}"
        chain = ChatPromptTemplate.from_messages([
            ("user", prompt)
        ]) | llm | StrOutputParser()
        return chain.invoke({})
    
    def review_task(context: WorkflowContext, llm: ChatOpenAI) -> str:
        """Review phase"""
        draft = context.get_result("write")
        prompt = f"Review and provide feedback on: {draft}"
        chain = ChatPromptTemplate.from_messages([
            ("user", prompt)
        ]) | llm | StrOutputParser()
        return chain.invoke({})
    
    # Create workflow
    content_workflow = AgenticWorkflow("Content Generation")
    
    # Add states
    content_workflow.add_state(WorkflowState(
        "research",
        "Research topic and gather key points",
        task=research_task
    ), is_start=True)
    
    content_workflow.add_state(WorkflowState(
        "outline",
        "Create content outline",
        task=outline_task
    ))
    
    content_workflow.add_state(WorkflowState(
        "write",
        "Write content draft",
        task=write_task
    ))
    
    content_workflow.add_state(WorkflowState(
        "review",
        "Review and provide feedback",
        task=review_task
    ))
    
    content_workflow.add_state(WorkflowState(
        "complete",
        "Content generation complete",
        is_terminal=True
    ))
    
    # Add transitions
    content_workflow.add_transition(StateTransition(
        "research", "outline", 
        transition_type=TransitionType.AUTOMATIC
    ))
    content_workflow.add_transition(StateTransition(
        "outline", "write",
        transition_type=TransitionType.AUTOMATIC
    ))
    content_workflow.add_transition(StateTransition(
        "write", "review",
        transition_type=TransitionType.AUTOMATIC
    ))
    content_workflow.add_transition(StateTransition(
        "review", "complete",
        transition_type=TransitionType.AUTOMATIC
    ))
    
    # Visualize workflow
    print("\nWorkflow Structure:")
    print(content_workflow.visualize())
    
    # Execute workflow
    print("\n" + "-" * 60)
    print("Executing Workflow...")
    print("-" * 60)
    
    initial_context = WorkflowContext(data={"topic": "Machine Learning"})
    
    try:
        result_context = content_workflow.execute(initial_context)
        
        print("\nWorkflow Execution Summary:")
        summary = content_workflow.get_execution_summary()
        print(f"  Status: {summary['status']}")
        print(f"  States visited: {summary['states_visited']}")
        print(f"  Total events: {summary['total_events']}")
        
        print("\nState Execution Path:")
        for state in result_context.history:
            print(f"  → {state}")
        
        print("\nFinal Results:")
        print(f"  Research: {result_context.get_result('research')[:80]}...")
        print(f"  Outline: {result_context.get_result('outline')[:80]}...")
        print(f"  Draft: {result_context.get_result('write')[:80]}...")
        print(f"  Review: {result_context.get_result('review')[:80]}...")
        
    except Exception as e:
        print(f"Workflow failed: {e}")
    
    # Example 2: Conditional branching workflow
    print("\n" + "=" * 80)
    print("Example 2: Conditional Branching Workflow")
    print("=" * 80)
    
    def analyze_task(context: WorkflowContext, llm: ChatOpenAI) -> Dict[str, Any]:
        """Analyze input"""
        text = context.data.get("text", "")
        prompt = f"Is this text positive or negative sentiment? Answer with just 'positive' or 'negative': {text}"
        chain = ChatPromptTemplate.from_messages([
            ("user", prompt)
        ]) | llm | StrOutputParser()
        sentiment = chain.invoke({}).strip().lower()
        return {"sentiment": sentiment}
    
    def positive_response_task(context: WorkflowContext, llm: ChatOpenAI) -> str:
        """Handle positive sentiment"""
        return "Thank you for the positive feedback!"
    
    def negative_response_task(context: WorkflowContext, llm: ChatOpenAI) -> str:
        """Handle negative sentiment"""
        return "We're sorry to hear that. How can we improve?"
    
    # Create branching workflow
    branching_workflow = AgenticWorkflow("Sentiment Analysis")
    
    # Add states
    branching_workflow.add_state(WorkflowState(
        "analyze",
        "Analyze sentiment",
        task=analyze_task
    ), is_start=True)
    
    branching_workflow.add_state(WorkflowState(
        "positive_response",
        "Generate positive response",
        task=positive_response_task
    ))
    
    branching_workflow.add_state(WorkflowState(
        "negative_response",
        "Generate negative response",
        task=negative_response_task
    ))
    
    branching_workflow.add_state(WorkflowState(
        "end",
        "End state",
        is_terminal=True
    ))
    
    # Add conditional transitions
    branching_workflow.add_transition(StateTransition(
        "analyze", "positive_response",
        condition=lambda ctx: ctx.get_result("analyze").get("sentiment") == "positive",
        transition_type=TransitionType.CONDITIONAL,
        description="If sentiment is positive"
    ))
    
    branching_workflow.add_transition(StateTransition(
        "analyze", "negative_response",
        condition=lambda ctx: ctx.get_result("analyze").get("sentiment") == "negative",
        transition_type=TransitionType.CONDITIONAL,
        description="If sentiment is negative"
    ))
    
    branching_workflow.add_transition(StateTransition(
        "positive_response", "end",
        transition_type=TransitionType.AUTOMATIC
    ))
    
    branching_workflow.add_transition(StateTransition(
        "negative_response", "end",
        transition_type=TransitionType.AUTOMATIC
    ))
    
    # Visualize
    print("\nWorkflow Structure:")
    print(branching_workflow.visualize())
    
    # Execute with positive text
    print("\n" + "-" * 60)
    print("Test 1: Positive Text")
    print("-" * 60)
    
    context1 = WorkflowContext(data={"text": "This product is amazing! I love it."})
    result1 = branching_workflow.execute(context1)
    
    print(f"Path taken: {' → '.join(result1.history)}")
    print(f"Response: {result1.get_result('positive_response')}")
    
    # Execute with negative text
    print("\n" + "-" * 60)
    print("Test 2: Negative Text")
    print("-" * 60)
    
    branching_workflow.status = WorkflowStatus.NOT_STARTED  # Reset
    context2 = WorkflowContext(data={"text": "This is terrible. Very disappointed."})
    result2 = branching_workflow.execute(context2)
    
    print(f"Path taken: {' → '.join(result2.history)}")
    print(f"Response: {result2.get_result('negative_response')}")
    
    # Summary
    print("\n" + "=" * 80)
    print("SUMMARY")
    print("=" * 80)
    print("""
The Agentic Workflows pattern enables:
✓ Complex multi-step process automation
✓ Conditional branching based on runtime state
✓ State-based execution with context passing
✓ Clear workflow visualization
✓ Execution logging and audit trails
✓ Error handling and recovery
✓ Flexible workflow composition

This pattern is valuable for:
- Business process automation
- Multi-stage content generation
- Decision tree execution
- Approval workflows
- Complex data pipelines
- Testing and validation workflows
- Orchestrating multiple agents
    """)


if __name__ == "__main__":
    demonstrate_agentic_workflows()

    

Agentic Design Pattern: Agentic Workflows

Pattern 167: Agentic Workflows – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Agentic Workflows pattern implementation.

---

Pattern Description

Pattern 167: Agentic Workflows

Description: The Agentic Workflows pattern implements complex, multi-step processes where agents autonomously navigate through workflow states, make decisions, handle branching logic, and coordinate multiple tasks. It uses state machines and graph-based execution for sophisticated task orchestration.

Components: 1. Workflow Definition: Graph structure with states and transitions 2. State Manager: Tracks current workflow state 3. Decision Engine: Makes branching decisions 4. Task Executor: Executes tasks at each state 5. Condition Evaluator: Evaluates transition conditions 6. Error Handler: Manages workflow failures

Use Cases: – Multi-step business processes – Complex data pipelines – Decision trees and workflows – Approval workflows – Multi-stage content generation – Automated testing workflows

Benefits: – Complex process automation – Clear workflow visualization – Conditional branching support – State persistence – Error recovery – Audit trail

Trade-offs: – Increased complexity – State management overhead – Debugging challenges – Performance considerations – Requires workflow design

LangChain Implementation: Uses LangGraph-inspired state machines with LangChain agents for task execution. Implements graph-based workflows with conditional branching and state management.

---

File Statistics

• Total Lines: 543
• Classes: 6
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 167: Agentic Workflows

Description:
    The Agentic Workflows pattern implements complex, multi-step processes where agents
    autonomously navigate through workflow states, make decisions, handle branching logic,
    and coordinate multiple tasks. It uses state machines and graph-based execution for
    sophisticated task orchestration.

Components:
    1. Workflow Definition: Graph structure with states and transitions
    2. State Manager: Tracks current workflow state
    3. Decision Engine: Makes branching decisions
    4. Task Executor: Executes tasks at each state
    5. Condition Evaluator: Evaluates transition conditions
    6. Error Handler: Manages workflow failures

Use Cases:
    - Multi-step business processes
    - Complex data pipelines
    - Decision trees and workflows
    - Approval workflows
    - Multi-stage content generation
    - Automated testing workflows

Benefits:
    - Complex process automation
    - Clear workflow visualization
    - Conditional branching support
    - State persistence
    - Error recovery
    - Audit trail

Trade-offs:
    - Increased complexity
    - State management overhead
    - Debugging challenges
    - Performance considerations
    - Requires workflow design

LangChain Implementation:
    Uses LangGraph-inspired state machines with LangChain agents for task execution.
    Implements graph-based workflows with conditional branching and state management.
"""

Purpose: This module implements the Agentic Workflows pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import dataclasses.dataclass
import dataclasses.field
import datetime.datetime
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import typing.Any
import typing.Callable
import typing.Dict
import typing.List
import typing.Optional

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – enum.Enum: Enumeration support – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Callable: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: WorkflowStatus

Lines 60-66

Inherits from: Enum

Purpose:

Status of workflow execution

---

Class: TransitionType

Lines 69-73

Inherits from: Enum

Purpose:

Types of state transitions

---

Class: WorkflowState

Lines 77-83

Purpose:

Represents a state in the workflow

---

Class: StateTransition

Lines 87-93

Purpose:

Represents a transition between states

---

Class: WorkflowContext

Lines 97-110

Purpose:

Context passed through workflow execution

Methods:

add_result(self, state, result)

Lines 103-106

Purpose:

Add result from a state execution

Parameters: – state: State – result: Result

get_result(self, state)

Lines 108-110

Purpose:

Get result from previous state

Parameters: – state: State

Returns: Any

---

Class: AgenticWorkflow

Lines 113-278

Purpose:

Implements agentic workflow pattern with state machines

Methods:

__init__(self, name)

Lines 116-127

Purpose:

Initialize workflow

Parameters: – name: Name

add_state(self, state, is_start)

Lines 129-133

Purpose:

Add a state to the workflow

Parameters: – state: State – is_start: Is start

add_transition(self, transition)

Lines 135-141

Purpose:

Add a transition between states

Parameters: – transition: Transition

execute(self, initial_context)

Lines 143-216

Purpose:

Execute the workflow

Parameters: – initial_context: Initial context

Returns: WorkflowContext

_get_next_state(self, context)

Lines 218-242

Purpose:

Determine next state based on transitions

Parameters: – context: Context

Returns: <ast.Subscript object at 0x000001A06D5F5B90>

_log_execution(self, event_type, data)

Lines 244-249

Purpose:

Log workflow execution event

Parameters: – event_type: Event type – data: Data

get_execution_summary(self)

Lines 251-259

Purpose:

Get summary of workflow execution

Parameters:

Returns: <ast.Subscript object at 0x000001A06D83F710>

visualize(self)

Lines 261-278

Purpose:

Create text visualization of workflow

Parameters:

Returns: str

---

Functions

demonstrate_agentic_workflows()

Lines 281-538

Purpose:

Demonstrate agentic workflows

---

Key Concepts Summary

Pattern Overview

The Agentic Workflows pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

---

Best Practices Demonstrated

1. Environment Configuration

• ✅ Use .env files for API keys
• ✅ Never hardcode credentials
• ✅ Validate environment variables on startup

2. Error Handling

• ✅ Graceful degradation on failures
• ✅ Informative error messages
• ✅ Logging for debugging

3. Prompt Engineering

• ✅ Clear, specific instructions
• ✅ Structured output formats
• ✅ Few-shot examples when beneficial

4. Code Organization

• ✅ Separation of concerns (class-based design)
• ✅ Type hints for clarity
• ✅ Docstrings for documentation

5. Performance Optimization

• ✅ Appropriate temperature settings
• ✅ Token budget management
• ✅ Caching where applicable

---

LangChain Components Used

• ChatOpenAI: OpenAI model interface
• Chains: Sequential component composition

Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development

---

Execution Flow

1. Initialize agent with configuration
   ↓
2. Load environment and dependencies
   ↓
3. Prepare prompt with input data
   ↓
4. Invoke LLM through LangChain
   ↓
5. Parse and validate response
   ↓
6. Return formatted result

Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation

---

Performance Considerations

Computational Cost

• LLM Calls: Primary cost factor (based on tokens)
• Latency: Depends on model and input size
• Memory: Minimal for most patterns

Optimization Strategies

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

• GPT-3.5-turbo: ~$0.002 per request
• GPT-4: ~$0.03 per request
• GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

---

Further Improvements

Production Enhancements

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 166
• Pattern 168

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Agentic Workflows pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns

🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability

📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/

---

File: 167_agentic_workflows.py
Total Lines: 543
Pattern Category: AI Agent Design Pattern #167

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 166: Prompt Caching & Reuse

Description:
    The Prompt Caching & Reuse pattern stores and retrieves prompts and their results
    to avoid redundant LLM calls, reduce latency, and save costs. It uses semantic
    similarity to match queries with cached results and manages cache invalidation
    and updates.

Components:
    1. Cache Storage: Stores prompts and results
    2. Similarity Matcher: Finds semantically similar cached prompts
    3. Cache Manager: Handles cache lifecycle
    4. Invalidation Strategy: Manages cache freshness
    5. Hit/Miss Tracker: Monitors cache performance
    6. Compression: Optimizes cache storage

Use Cases:
    - Repeated query handling
    - Cost optimization
    - Latency reduction
    - Common question answering
    - Template-based generation
    - Batch processing optimization

Benefits:
    - Reduced API costs
    - Lower latency
    - Better resource utilization
    - Consistent responses
    - Offline capability

Trade-offs:
    - Storage overhead
    - Cache invalidation complexity
    - Stale data risk
    - Memory usage
    - Similarity matching overhead

LangChain Implementation:
    Combines semantic similarity search with LangChain caching mechanisms.
    Uses embeddings for matching similar queries efficiently.
"""

import os
import json
import hashlib
import time
from typing import List, Dict, Any, Optional, Tuple
from dataclasses import dataclass, field, asdict
from datetime import datetime, timedelta
from collections import defaultdict
from dotenv import load_dotenv

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

load_dotenv()


@dataclass
class CacheEntry:
    """Represents a cached prompt and response"""
    key: str
    prompt: str
    response: str
    timestamp: datetime
    hit_count: int = 0
    embedding: Optional[List[float]] = None
    metadata: Dict[str, Any] = field(default_factory=dict)
    ttl: Optional[int] = None  # Time-to-live in seconds
    
    def is_expired(self) -> bool:
        """Check if cache entry has expired"""
        if self.ttl is None:
            return False
        age = (datetime.now() - self.timestamp).total_seconds()
        return age > self.ttl
    
    def to_dict(self) -> Dict[str, Any]:
        """Convert to dictionary for serialization"""
        data = asdict(self)
        data['timestamp'] = self.timestamp.isoformat()
        return data
    
    @staticmethod
    def from_dict(data: Dict[str, Any]) -> 'CacheEntry':
        """Create from dictionary"""
        data['timestamp'] = datetime.fromisoformat(data['timestamp'])
        return CacheEntry(**data)


class PromptCacheManager:
    """Manages prompt caching and reuse with semantic similarity matching"""
    
    def __init__(self, 
                 similarity_threshold: float = 0.85,
                 max_cache_size: int = 1000,
                 default_ttl: Optional[int] = None):
        """
        Initialize prompt cache manager
        
        Args:
            similarity_threshold: Minimum similarity score for cache hit (0-1)
            max_cache_size: Maximum number of entries to cache
            default_ttl: Default time-to-live in seconds
        """
        self.similarity_threshold = similarity_threshold
        self.max_cache_size = max_cache_size
        self.default_ttl = default_ttl
        
        self.cache: Dict[str, CacheEntry] = {}
        self.embeddings = OpenAIEmbeddings()
        self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
        
        # Statistics
        self.stats = {
            "total_requests": 0,
            "cache_hits": 0,
            "cache_misses": 0,
            "total_cost_saved": 0.0,
            "total_time_saved": 0.0
        }
    
    def get_or_generate(self, prompt: str, 
                       force_refresh: bool = False,
                       ttl: Optional[int] = None) -> Tuple[str, bool]:
        """
        Get cached response or generate new one
        
        Args:
            prompt: The prompt to process
            force_refresh: Skip cache and generate new response
            ttl: Time-to-live for this entry
            
        Returns:
            Tuple of (response, was_cached)
        """
        self.stats["total_requests"] += 1
        
        if not force_refresh:
            # Try to find in cache
            cached_response = self._find_similar_cached(prompt)
            if cached_response:
                self.stats["cache_hits"] += 1
                self.stats["total_cost_saved"] += 0.002  # Estimated savings
                self.stats["total_time_saved"] += 0.5  # Estimated time saved
                return cached_response, True
        
        # Cache miss - generate new response
        self.stats["cache_misses"] += 1
        response = self._generate_response(prompt)
        
        # Cache the result
        self._add_to_cache(prompt, response, ttl or self.default_ttl)
        
        return response, False
    
    def batch_get_or_generate(self, prompts: List[str]) -> List[Tuple[str, bool]]:
        """Process multiple prompts efficiently"""
        results = []
        for prompt in prompts:
            result, cached = self.get_or_generate(prompt)
            results.append((result, cached))
        return results
    
    def invalidate(self, pattern: Optional[str] = None):
        """Invalidate cache entries matching pattern"""
        if pattern is None:
            # Clear all
            self.cache.clear()
        else:
            # Remove matching entries
            keys_to_remove = [
                key for key, entry in self.cache.items()
                if pattern.lower() in entry.prompt.lower()
            ]
            for key in keys_to_remove:
                del self.cache[key]
    
    def _find_similar_cached(self, prompt: str) -> Optional[str]:
        """Find semantically similar cached prompt"""
        if not self.cache:
            return None
        
        # Clean expired entries
        self._clean_expired()
        
        # Get embedding for query prompt
        query_embedding = self.embeddings.embed_query(prompt)
        
        # Find most similar cached entry
        best_match = None
        best_similarity = 0.0
        
        for entry in self.cache.values():
            if entry.embedding is None:
                continue
            
            # Calculate cosine similarity
            similarity = self._cosine_similarity(query_embedding, entry.embedding)
            
            if similarity > best_similarity and similarity >= self.similarity_threshold:
                best_similarity = similarity
                best_match = entry
        
        if best_match:
            best_match.hit_count += 1
            return best_match.response
        
        return None
    
    def _generate_response(self, prompt: str) -> str:
        """Generate new response using LLM"""
        chain = ChatPromptTemplate.from_messages([
            ("user", "{prompt}")
        ]) | self.llm | StrOutputParser()
        
        response = chain.invoke({"prompt": prompt})
        return response
    
    def _add_to_cache(self, prompt: str, response: str, ttl: Optional[int] = None):
        """Add entry to cache"""
        # Generate cache key
        key = self._generate_key(prompt)
        
        # Get embedding
        embedding = self.embeddings.embed_query(prompt)
        
        # Create entry
        entry = CacheEntry(
            key=key,
            prompt=prompt,
            response=response,
            timestamp=datetime.now(),
            embedding=embedding,
            ttl=ttl
        )
        
        # Check cache size
        if len(self.cache) >= self.max_cache_size:
            self._evict_entry()
        
        # Add to cache
        self.cache[key] = entry
    
    def _generate_key(self, prompt: str) -> str:
        """Generate cache key from prompt"""
        return hashlib.sha256(prompt.encode()).hexdigest()
    
    def _cosine_similarity(self, vec1: List[float], vec2: List[float]) -> float:
        """Calculate cosine similarity between two vectors"""
        dot_product = sum(a * b for a, b in zip(vec1, vec2))
        magnitude1 = sum(a * a for a in vec1) ** 0.5
        magnitude2 = sum(b * b for b in vec2) ** 0.5
        
        if magnitude1 == 0 or magnitude2 == 0:
            return 0.0
        
        return dot_product / (magnitude1 * magnitude2)
    
    def _clean_expired(self):
        """Remove expired entries"""
        keys_to_remove = [
            key for key, entry in self.cache.items()
            if entry.is_expired()
        ]
        for key in keys_to_remove:
            del self.cache[key]
    
    def _evict_entry(self):
        """Evict least recently used entry"""
        if not self.cache:
            return
        
        # Find entry with lowest hit count and oldest timestamp
        lru_key = min(
            self.cache.keys(),
            key=lambda k: (self.cache[k].hit_count, self.cache[k].timestamp)
        )
        del self.cache[lru_key]
    
    def get_stats(self) -> Dict[str, Any]:
        """Get cache statistics"""
        total = self.stats["total_requests"]
        hit_rate = self.stats["cache_hits"] / total if total > 0 else 0
        
        return {
            **self.stats,
            "cache_hit_rate": hit_rate,
            "cache_size": len(self.cache),
            "avg_hit_count": sum(e.hit_count for e in self.cache.values()) / len(self.cache)
                            if self.cache else 0
        }
    
    def get_popular_queries(self, top_n: int = 10) -> List[Dict[str, Any]]:
        """Get most frequently accessed cached queries"""
        sorted_entries = sorted(
            self.cache.values(),
            key=lambda e: e.hit_count,
            reverse=True
        )[:top_n]
        
        return [
            {
                "prompt": entry.prompt,
                "hit_count": entry.hit_count,
                "age_seconds": (datetime.now() - entry.timestamp).total_seconds()
            }
            for entry in sorted_entries
        ]
    
    def save_cache(self, filepath: str):
        """Save cache to file"""
        cache_data = {
            key: entry.to_dict()
            for key, entry in self.cache.items()
        }
        
        with open(filepath, 'w') as f:
            json.dump(cache_data, f, indent=2)
    
    def load_cache(self, filepath: str):
        """Load cache from file"""
        try:
            with open(filepath, 'r') as f:
                cache_data = json.load(f)
            
            self.cache = {
                key: CacheEntry.from_dict(data)
                for key, data in cache_data.items()
            }
        except FileNotFoundError:
            print(f"Cache file {filepath} not found")
        except Exception as e:
            print(f"Error loading cache: {e}")


def demonstrate_prompt_caching():
    """Demonstrate prompt caching and reuse"""
    print("=" * 80)
    print("PROMPT CACHING & REUSE PATTERN DEMONSTRATION")
    print("=" * 80)
    
    cache_manager = PromptCacheManager(
        similarity_threshold=0.85,
        max_cache_size=100,
        default_ttl=3600  # 1 hour
    )
    
    # Example 1: Basic caching
    print("\n" + "=" * 80)
    print("Example 1: Basic Caching")
    print("=" * 80)
    
    prompts = [
        "What is the capital of France?",
        "Tell me about the capital of France",  # Similar query
        "What is the capital of Germany?",
    ]
    
    print("\nProcessing prompts with caching:")
    for i, prompt in enumerate(prompts, 1):
        print(f"\n{i}. Query: {prompt}")
        response, was_cached = cache_manager.get_or_generate(prompt)
        status = "CACHE HIT" if was_cached else "CACHE MISS"
        print(f"   Status: {status}")
        print(f"   Response: {response[:100]}...")
    
    # Example 2: Similar query matching
    print("\n" + "=" * 80)
    print("Example 2: Semantic Similarity Matching")
    print("=" * 80)
    
    original = "Explain quantum computing"
    similar = "Can you describe quantum computing?"
    
    print(f"\nOriginal query: {original}")
    response1, cached1 = cache_manager.get_or_generate(original)
    print(f"Status: {'CACHED' if cached1 else 'GENERATED'}")
    print(f"Response: {response1[:100]}...")
    
    print(f"\nSimilar query: {similar}")
    response2, cached2 = cache_manager.get_or_generate(similar)
    print(f"Status: {'CACHED' if cached2 else 'GENERATED'}")
    print(f"Match found: {cached2}")
    if cached2:
        print("✓ Similar query matched with cached result!")
    
    # Example 3: Batch processing
    print("\n" + "=" * 80)
    print("Example 3: Batch Processing with Cache")
    print("=" * 80)
    
    batch_prompts = [
        "What is machine learning?",
        "Define machine learning",  # Similar
        "What is deep learning?",
        "Explain deep learning",  # Similar
        "What is machine learning?",  # Exact duplicate
    ]
    
    print(f"\nProcessing batch of {len(batch_prompts)} prompts:")
    start_time = time.time()
    results = cache_manager.batch_get_or_generate(batch_prompts)
    elapsed = time.time() - start_time
    
    hits = sum(1 for _, cached in results if cached)
    misses = len(results) - hits
    
    print(f"\nResults:")
    print(f"  Total queries: {len(results)}")
    print(f"  Cache hits: {hits}")
    print(f"  Cache misses: {misses}")
    print(f"  Hit rate: {hits/len(results):.1%}")
    print(f"  Processing time: {elapsed:.2f}s")
    
    # Example 4: Cache statistics
    print("\n" + "=" * 80)
    print("Example 4: Cache Statistics")
    print("=" * 80)
    
    stats = cache_manager.get_stats()
    
    print("\nCache Performance:")
    print(f"  Total requests: {stats['total_requests']}")
    print(f"  Cache hits: {stats['cache_hits']}")
    print(f"  Cache misses: {stats['cache_misses']}")
    print(f"  Hit rate: {stats['cache_hit_rate']:.1%}")
    print(f"  Cache size: {stats['cache_size']} entries")
    print(f"  Estimated cost saved: ${stats['total_cost_saved']:.4f}")
    print(f"  Estimated time saved: {stats['total_time_saved']:.2f}s")
    
    # Example 5: Popular queries
    print("\n" + "=" * 80)
    print("Example 5: Popular Queries Analysis")
    print("=" * 80)
    
    popular = cache_manager.get_popular_queries(top_n=5)
    
    print("\nMost frequently accessed queries:")
    for i, query_info in enumerate(popular, 1):
        age_min = query_info['age_seconds'] / 60
        print(f"\n{i}. Hit count: {query_info['hit_count']}")
        print(f"   Age: {age_min:.1f} minutes")
        print(f"   Query: {query_info['prompt'][:60]}...")
    
    # Example 6: TTL and expiration
    print("\n" + "=" * 80)
    print("Example 6: Time-to-Live (TTL) Management")
    print("=" * 80)
    
    # Add entry with short TTL
    short_ttl_prompt = "What's the current time?"
    response, cached = cache_manager.get_or_generate(short_ttl_prompt, ttl=2)
    print(f"\nAdded query with 2 second TTL: {short_ttl_prompt}")
    print(f"Response: {response[:80]}...")
    
    print("\nImmediate retry (should hit cache):")
    response, cached = cache_manager.get_or_generate(short_ttl_prompt)
    print(f"Cached: {cached}")
    
    print("\nWaiting 3 seconds for expiration...")
    time.sleep(3)
    
    print("Retry after expiration (should miss cache):")
    response, cached = cache_manager.get_or_generate(short_ttl_prompt)
    print(f"Cached: {cached}")
    
    # Example 7: Cache invalidation
    print("\n" + "=" * 80)
    print("Example 7: Cache Invalidation")
    print("=" * 80)
    
    print(f"\nCache size before invalidation: {len(cache_manager.cache)}")
    
    # Invalidate specific pattern
    cache_manager.invalidate(pattern="capital")
    print(f"Cache size after invalidating 'capital' queries: {len(cache_manager.cache)}")
    
    # Test that invalidated query is regenerated
    test_prompt = "What is the capital of France?"
    response, cached = cache_manager.get_or_generate(test_prompt)
    print(f"\nRetrying invalidated query: {test_prompt}")
    print(f"Cached: {cached} (should be False)")
    
    # Final statistics
    print("\n" + "=" * 80)
    print("Final Statistics")
    print("=" * 80)
    
    final_stats = cache_manager.get_stats()
    print(f"\nOverall Performance:")
    print(f"  Total requests processed: {final_stats['total_requests']}")
    print(f"  Overall hit rate: {final_stats['cache_hit_rate']:.1%}")
    print(f"  Final cache size: {final_stats['cache_size']} entries")
    print(f"  Total cost saved: ${final_stats['total_cost_saved']:.4f}")
    print(f"  Total time saved: {final_stats['total_time_saved']:.2f}s")
    
    # Summary
    print("\n" + "=" * 80)
    print("SUMMARY")
    print("=" * 80)
    print("""
The Prompt Caching & Reuse pattern enables:
✓ Semantic similarity matching for cache hits
✓ Significant cost reduction through reuse
✓ Lower latency for repeated queries
✓ TTL-based cache expiration
✓ Cache invalidation strategies
✓ Performance monitoring and analytics
✓ Batch processing optimization

This pattern is valuable for:
- High-traffic applications with repeated queries
- Cost-sensitive deployments
- Latency-critical systems
- Common question answering systems
- Template-based generation
- Offline or low-connectivity scenarios
    """)


if __name__ == "__main__":
    demonstrate_prompt_caching()

    

Agentic Design Pattern: Prompt Caching And Reuse

Pattern 166: Prompt Caching & Reuse – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Prompt Caching & Reuse pattern implementation.

---

Pattern Description

Pattern 166: Prompt Caching & Reuse

Description: The Prompt Caching & Reuse pattern stores and retrieves prompts and their results to avoid redundant LLM calls, reduce latency, and save costs. It uses semantic similarity to match queries with cached results and manages cache invalidation and updates.

Components: 1. Cache Storage: Stores prompts and results 2. Similarity Matcher: Finds semantically similar cached prompts 3. Cache Manager: Handles cache lifecycle 4. Invalidation Strategy: Manages cache freshness 5. Hit/Miss Tracker: Monitors cache performance 6. Compression: Optimizes cache storage

Use Cases: – Repeated query handling – Cost optimization – Latency reduction – Common question answering – Template-based generation – Batch processing optimization

Benefits: – Reduced API costs – Lower latency – Better resource utilization – Consistent responses – Offline capability

Trade-offs: – Storage overhead – Cache invalidation complexity – Stale data risk – Memory usage – Similarity matching overhead

LangChain Implementation: Combines semantic similarity search with LangChain caching mechanisms. Uses embeddings for matching similar queries efficiently.

---

File Statistics

• Total Lines: 527
• Classes: 2
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 166: Prompt Caching & Reuse

Description:
    The Prompt Caching & Reuse pattern stores and retrieves prompts and their results
    to avoid redundant LLM calls, reduce latency, and save costs. It uses semantic
    similarity to match queries with cached results and manages cache invalidation
    and updates.

Components:
    1. Cache Storage: Stores prompts and results
    2. Similarity Matcher: Finds semantically similar cached prompts
    3. Cache Manager: Handles cache lifecycle
    4. Invalidation Strategy: Manages cache freshness
    5. Hit/Miss Tracker: Monitors cache performance
    6. Compression: Optimizes cache storage

Use Cases:
    - Repeated query handling
    - Cost optimization
    - Latency reduction
    - Common question answering
    - Template-based generation
    - Batch processing optimization

Benefits:
    - Reduced API costs
    - Lower latency
    - Better resource utilization
    - Consistent responses
    - Offline capability

Trade-offs:
    - Storage overhead
    - Cache invalidation complexity
    - Stale data risk
    - Memory usage
    - Similarity matching overhead

LangChain Implementation:
    Combines semantic similarity search with LangChain caching mechanisms.
    Uses embeddings for matching similar queries efficiently.
"""

Purpose: This module implements the Prompt Caching & Reuse pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import collections.defaultdict
import dataclasses.asdict
import dataclasses.dataclass
import dataclasses.field
import datetime.datetime
import datetime.timedelta
import json
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import time
import typing.Any
import typing.Dict
import typing.List
import typing.Optional
import typing.Tuple

Purpose: Core Python functionality: – collections.defaultdict: Specialized container datatypes – dataclasses.asdict: Simplified class definitions with @dataclass – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – datetime.timedelta: Date and time handling – json: JSON serialization/deserialization – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – time: Time-related functions – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Tuple: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI
from langchain_openai.OpenAIEmbeddings

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from dotenv.load_dotenv
from hashlib

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: CacheEntry

Lines 63-91

Purpose:

Represents a cached prompt and response

Methods:

is_expired(self)

Lines 74-79

Purpose:

Check if cache entry has expired

Parameters:

Returns: bool

to_dict(self)

Lines 81-85

Purpose:

Convert to dictionary for serialization

Parameters:

Returns: <ast.Subscript object at 0x000001A06D6AA950>

from_dict(data)

Lines 88-91

Decorators: staticmethod

Purpose:

Create from dictionary

Parameters: – data: Data

Returns: CacheEntry

---

Class: PromptCacheManager

Lines 94-337

Purpose:

Manages prompt caching and reuse with semantic similarity matching

Methods:

__init__(self, similarity_threshold, max_cache_size, default_ttl)

Lines 97-124

Purpose:

Initialize prompt cache manager

Args:
    similarity_threshold: Minimum similarity score for cache hit (0-1)
    max_cache_size: Maximum number of entries to cache
    default_ttl: Default time-to-live in seconds

Parameters: – similarity_threshold: Similarity threshold – max_cache_size: Max cache size – default_ttl: Default ttl

get_or_generate(self, prompt, force_refresh, ttl)

Lines 126-158

Purpose:

Get cached response or generate new one

Args:
    prompt: The prompt to process
    force_refresh: Skip cache and generate new response
    ttl: Time-to-live for this entry

Returns:
    Tuple of (response, was_cached)

Parameters: – prompt: Prompt template or text – force_refresh: Force refresh – ttl: Ttl

Returns: <ast.Subscript object at 0x000001A06D824D50>

batch_get_or_generate(self, prompts)

Lines 160-166

Purpose:

Process multiple prompts efficiently

Parameters: – prompts: Prompt template or text

Returns: <ast.Subscript object at 0x000001A06D827310>

invalidate(self, pattern)

Lines 168-180

Purpose:

Invalidate cache entries matching pattern

Parameters: – pattern: Pattern

_find_similar_cached(self, prompt)

Lines 182-212

Purpose:

Find semantically similar cached prompt

Parameters: – prompt: Prompt template or text

Returns: <ast.Subscript object at 0x000001A06D6147D0>

_generate_response(self, prompt)

Lines 214-221

Purpose:

Generate new response using LLM

Parameters: – prompt: Prompt template or text

Returns: str

_add_to_cache(self, prompt, response, ttl)

Lines 223-246

Purpose:

Add entry to cache

Parameters: – prompt: Prompt template or text – response: Response – ttl: Ttl

_generate_key(self, prompt)

Lines 248-250

Purpose:

Generate cache key from prompt

Parameters: – prompt: Prompt template or text

Returns: str

_cosine_similarity(self, vec1, vec2)

Lines 252-261

Purpose:

Calculate cosine similarity between two vectors

Parameters: – vec1: Vec1 – vec2: Vec2

Returns: float

_clean_expired(self)

Lines 263-270

Purpose:

Remove expired entries

Parameters:

_evict_entry(self)

Lines 272-282

Purpose:

Evict least recently used entry

Parameters:

get_stats(self)

Lines 284-295

Purpose:

Get cache statistics

Parameters:

Returns: <ast.Subscript object at 0x000001A06D8656D0>

get_popular_queries(self, top_n)

Lines 297-312

Purpose:

Get most frequently accessed cached queries

Parameters: – top_n: Top n

Returns: <ast.Subscript object at 0x000001A06D7B8510>

save_cache(self, filepath)

Lines 314-322

Purpose:

Save cache to file

Parameters: – filepath: Filepath

load_cache(self, filepath)

Lines 324-337

Purpose:

Load cache from file

Parameters: – filepath: Filepath

---

Functions

demonstrate_prompt_caching()

Lines 340-522

Purpose:

Demonstrate prompt caching and reuse

---

Key Concepts Summary

Pattern Overview

The Prompt Caching & Reuse pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

---

Best Practices Demonstrated

1. Environment Configuration

• ✅ Use .env files for API keys
• ✅ Never hardcode credentials
• ✅ Validate environment variables on startup

2. Error Handling

• ✅ Graceful degradation on failures
• ✅ Informative error messages
• ✅ Logging for debugging

3. Prompt Engineering

• ✅ Clear, specific instructions
• ✅ Structured output formats
• ✅ Few-shot examples when beneficial

4. Code Organization

• ✅ Separation of concerns (class-based design)
• ✅ Type hints for clarity
• ✅ Docstrings for documentation

5. Performance Optimization

• ✅ Appropriate temperature settings
• ✅ Token budget management
• ✅ Caching where applicable

---

LangChain Components Used

• ChatOpenAI: OpenAI model interface
• Chains: Sequential component composition

Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development

---

Execution Flow

1. Initialize agent with configuration
   ↓
2. Load environment and dependencies
   ↓
3. Prepare prompt with input data
   ↓
4. Invoke LLM through LangChain
   ↓
5. Parse and validate response
   ↓
6. Return formatted result

Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation

---

Performance Considerations

Computational Cost

• LLM Calls: Primary cost factor (based on tokens)
• Latency: Depends on model and input size
• Memory: Minimal for most patterns

Optimization Strategies

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

• GPT-3.5-turbo: ~$0.002 per request
• GPT-4: ~$0.03 per request
• GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

---

Further Improvements

Production Enhancements

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 165
• Pattern 167

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Prompt Caching & Reuse pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns

🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability

📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/

---

File: 166_prompt_caching_&_reuse.py
Total Lines: 527
Pattern Category: AI Agent Design Pattern #166

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 165: Foundation Model Orchestration

Description:
    The Foundation Model Orchestration pattern coordinates multiple foundation models
    (LLMs with different capabilities, sizes, and specializations) to solve complex tasks
    optimally. The orchestrator routes tasks to the most appropriate model based on
    requirements, capabilities, cost, and performance trade-offs.

Components:
    1. Model Registry: Catalog of available models and their capabilities
    2. Task Analyzer: Analyzes task requirements and complexity
    3. Model Selector: Chooses optimal model for each task
    4. Result Aggregator: Combines results from multiple models
    5. Performance Monitor: Tracks model performance and costs
    6. Fallback Manager: Handles model failures

Use Cases:
    - Cost-optimized AI pipelines
    - Quality-performance trade-offs
    - Specialized task routing
    - Multi-model ensemble systems
    - Adaptive model selection
    - Resource optimization

Benefits:
    - Optimal cost-performance balance
    - Better quality through specialization
    - Fault tolerance with fallbacks
    - Flexible model upgrades
    - Resource efficiency

Trade-offs:
    - Orchestration complexity
    - Latency overhead
    - Model compatibility issues
    - Cost tracking complexity
    - Requires multiple API keys

LangChain Implementation:
    Uses LangChain's model abstraction to coordinate multiple models. Implements
    intelligent routing logic based on task characteristics and model capabilities.
"""

import os
import time
from typing import List, Dict, Any, Optional, Callable
from dataclasses import dataclass, field
from enum import Enum
from collections import defaultdict
from dotenv import load_dotenv

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

load_dotenv()


class ModelCapability(Enum):
    """Model capability types"""
    REASONING = "reasoning"
    CODING = "coding"
    CREATIVE = "creative"
    ANALYSIS = "analysis"
    SUMMARIZATION = "summarization"
    TRANSLATION = "translation"
    FAST_RESPONSE = "fast_response"
    LONG_CONTEXT = "long_context"


class ModelTier(Enum):
    """Model tier for cost/performance trade-off"""
    NANO = "nano"  # Fastest, cheapest, basic
    SMALL = "small"  # Fast, cheap, good quality
    MEDIUM = "medium"  # Balanced
    LARGE = "large"  # High quality, slower, expensive
    PREMIUM = "premium"  # Best quality, slowest, most expensive


@dataclass
class ModelProfile:
    """Profile of a foundation model"""
    name: str
    model_id: str
    tier: ModelTier
    capabilities: List[ModelCapability]
    max_tokens: int
    cost_per_1k_tokens: float
    avg_latency_ms: float
    quality_score: float = 0.8  # 0-1 scale
    available: bool = True
    metadata: Dict[str, Any] = field(default_factory=dict)


@dataclass
class Task:
    """A task to be processed"""
    id: str
    content: str
    required_capabilities: List[ModelCapability]
    priority: str = "medium"  # low, medium, high
    max_cost: Optional[float] = None
    max_latency: Optional[float] = None
    min_quality: float = 0.7


@dataclass
class TaskResult:
    """Result of task execution"""
    task_id: str
    model_used: str
    result: str
    execution_time: float
    estimated_cost: float
    quality_estimate: float
    metadata: Dict[str, Any] = field(default_factory=dict)


class FoundationModelOrchestrator:
    """Orchestrates multiple foundation models for optimal task execution"""
    
    def __init__(self):
        """Initialize the orchestrator with model registry"""
        self.models: Dict[str, ModelProfile] = {}
        self.llm_instances: Dict[str, ChatOpenAI] = {}
        self.performance_stats: Dict[str, Dict[str, Any]] = defaultdict(lambda: {
            "tasks_completed": 0,
            "total_cost": 0.0,
            "total_latency": 0.0,
            "failures": 0
        })
        
        # Register models
        self._register_default_models()
        
        # Task routing prompt
        self.routing_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a task analysis expert. Analyze the task and determine
            which capabilities are most important for successful completion."""),
            ("user", """Task: {task}

Analyze this task and identify the key capabilities needed (choose from: reasoning,
coding, creative, analysis, summarization, translation, fast_response, long_context).

Respond with just the capability names, comma-separated.""")
        ])
    
    def _register_default_models(self):
        """Register default model profiles"""
        # Nano tier - fast and cheap
        self.register_model(ModelProfile(
            name="GPT-3.5 Turbo",
            model_id="gpt-3.5-turbo",
            tier=ModelTier.SMALL,
            capabilities=[
                ModelCapability.FAST_RESPONSE,
                ModelCapability.SUMMARIZATION,
                ModelCapability.ANALYSIS
            ],
            max_tokens=4096,
            cost_per_1k_tokens=0.002,
            avg_latency_ms=800,
            quality_score=0.75
        ))
        
        # Medium tier - balanced
        self.register_model(ModelProfile(
            name="GPT-4 Turbo",
            model_id="gpt-4-turbo-preview",
            tier=ModelTier.LARGE,
            capabilities=[
                ModelCapability.REASONING,
                ModelCapability.CODING,
                ModelCapability.CREATIVE,
                ModelCapability.ANALYSIS,
                ModelCapability.LONG_CONTEXT
            ],
            max_tokens=128000,
            cost_per_1k_tokens=0.01,
            avg_latency_ms=2000,
            quality_score=0.95
        ))
        
        # Specialized models (simulated with different configs)
        self.register_model(ModelProfile(
            name="Code Specialist",
            model_id="gpt-3.5-turbo",  # Using same model but different profile
            tier=ModelTier.MEDIUM,
            capabilities=[
                ModelCapability.CODING,
                ModelCapability.REASONING
            ],
            max_tokens=4096,
            cost_per_1k_tokens=0.003,
            avg_latency_ms=1000,
            quality_score=0.85
        ))
    
    def register_model(self, profile: ModelProfile):
        """Register a model in the orchestrator"""
        self.models[profile.name] = profile
        # Create LLM instance
        try:
            self.llm_instances[profile.name] = ChatOpenAI(
                model=profile.model_id,
                temperature=0.7,
                max_tokens=1000
            )
        except Exception as e:
            print(f"Warning: Could not create LLM instance for {profile.name}: {e}")
            profile.available = False
    
    def execute_task(self, task: Task, use_fallback: bool = True) -> TaskResult:
        """Execute a task using the most appropriate model"""
        # Select best model for task
        selected_model = self._select_model(task)
        
        if not selected_model:
            raise ValueError("No suitable model found for task")
        
        # Execute task
        start_time = time.time()
        
        try:
            result = self._execute_on_model(task, selected_model)
            execution_time = time.time() - start_time
            
            # Calculate cost
            estimated_tokens = len(task.content.split()) * 1.3  # Rough estimate
            estimated_cost = (estimated_tokens / 1000) * selected_model.cost_per_1k_tokens
            
            # Update stats
            self._update_stats(selected_model.name, execution_time, estimated_cost, success=True)
            
            return TaskResult(
                task_id=task.id,
                model_used=selected_model.name,
                result=result,
                execution_time=execution_time,
                estimated_cost=estimated_cost,
                quality_estimate=selected_model.quality_score
            )
            
        except Exception as e:
            self._update_stats(selected_model.name, 0, 0, success=False)
            
            if use_fallback:
                # Try fallback model
                fallback_model = self._select_fallback_model(task, selected_model)
                if fallback_model:
                    print(f"Primary model failed, trying fallback: {fallback_model.name}")
                    task_copy = Task(
                        id=task.id,
                        content=task.content,
                        required_capabilities=task.required_capabilities,
                        priority=task.priority
                    )
                    return self.execute_task(task_copy, use_fallback=False)
            
            raise RuntimeError(f"Task execution failed: {e}")
    
    def execute_batch(self, tasks: List[Task], parallel: bool = False) -> List[TaskResult]:
        """Execute multiple tasks"""
        results = []
        
        for task in tasks:
            try:
                result = self.execute_task(task)
                results.append(result)
            except Exception as e:
                print(f"Task {task.id} failed: {e}")
        
        return results
    
    def execute_with_ensemble(self, task: Task, num_models: int = 3) -> Dict[str, Any]:
        """Execute task on multiple models and aggregate results"""
        # Select top N models
        candidates = self._rank_models_for_task(task)[:num_models]
        
        results = []
        for model_profile in candidates:
            try:
                result = self._execute_on_model(task, model_profile)
                results.append({
                    "model": model_profile.name,
                    "result": result,
                    "quality": model_profile.quality_score
                })
            except Exception as e:
                print(f"Model {model_profile.name} failed: {e}")
        
        if not results:
            raise RuntimeError("All models failed")
        
        # Aggregate results (simple voting or quality-weighted selection)
        best_result = max(results, key=lambda x: x["quality"])
        
        return {
            "aggregated_result": best_result["result"],
            "all_results": results,
            "models_used": [r["model"] for r in results]
        }
    
    def _select_model(self, task: Task) -> Optional[ModelProfile]:
        """Select the best model for a task"""
        candidates = self._rank_models_for_task(task)
        
        if not candidates:
            return None
        
        # Filter by constraints
        for model in candidates:
            # Check cost constraint
            if task.max_cost:
                estimated_tokens = len(task.content.split()) * 1.3
                estimated_cost = (estimated_tokens / 1000) * model.cost_per_1k_tokens
                if estimated_cost > task.max_cost:
                    continue
            
            # Check latency constraint
            if task.max_latency and model.avg_latency_ms > task.max_latency:
                continue
            
            # Check quality constraint
            if model.quality_score  List[ModelProfile]:
        """Rank models by suitability for task"""
        scores = []
        
        for model in self.models.values():
            if not model.available:
                continue
            
            score = 0.0
            
            # Capability match
            matching_caps = sum(
                1 for cap in task.required_capabilities 
                if cap in model.capabilities
            )
            if task.required_capabilities:
                capability_score = matching_caps / len(task.required_capabilities)
            else:
                capability_score = 0.5
            
            score += capability_score * 40  # 40% weight
            
            # Quality score
            score += model.quality_score * 30  # 30% weight
            
            # Cost efficiency (lower is better)
            cost_score = 1.0 - min(model.cost_per_1k_tokens / 0.05, 1.0)
            score += cost_score * 15  # 15% weight
            
            # Speed (lower latency is better)
            speed_score = 1.0 - min(model.avg_latency_ms / 5000, 1.0)
            score += speed_score * 15  # 15% weight
            
            # Priority adjustment
            if task.priority == "high":
                score += model.quality_score * 10
            elif task.priority == "low":
                score += cost_score * 10
            
            scores.append((score, model))
        
        # Sort by score descending
        scores.sort(key=lambda x: x[0], reverse=True)
        
        return [model for _, model in scores]
    
    def _select_fallback_model(self, task: Task, failed_model: ModelProfile) -> Optional[ModelProfile]:
        """Select a fallback model after primary fails"""
        candidates = self._rank_models_for_task(task)
        
        # Return first available model that's not the failed one
        for model in candidates:
            if model.name != failed_model.name and model.available:
                return model
        
        return None
    
    def _execute_on_model(self, task: Task, model: ModelProfile) -> str:
        """Execute task on specific model"""
        llm = self.llm_instances.get(model.name)
        
        if not llm:
            raise RuntimeError(f"LLM instance not available for {model.name}")
        
        # Create execution prompt
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are processing this task using {model.name}."),
            ("user", "{task_content}")
        ])
        
        chain = prompt | llm | StrOutputParser()
        result = chain.invoke({"task_content": task.content})
        
        return result
    
    def _update_stats(self, model_name: str, execution_time: float, 
                     cost: float, success: bool):
        """Update performance statistics"""
        stats = self.performance_stats[model_name]
        
        if success:
            stats["tasks_completed"] += 1
            stats["total_cost"] += cost
            stats["total_latency"] += execution_time
        else:
            stats["failures"] += 1
    
    def get_performance_report(self) -> Dict[str, Any]:
        """Get performance report for all models"""
        report = {}
        
        for model_name, stats in self.performance_stats.items():
            total_tasks = stats["tasks_completed"] + stats["failures"]
            
            if total_tasks > 0:
                report[model_name] = {
                    "tasks_completed": stats["tasks_completed"],
                    "failures": stats["failures"],
                    "success_rate": stats["tasks_completed"] / total_tasks,
                    "total_cost": stats["total_cost"],
                    "avg_latency": stats["total_latency"] / stats["tasks_completed"] 
                                  if stats["tasks_completed"] > 0 else 0,
                    "cost_per_task": stats["total_cost"] / stats["tasks_completed"]
                                    if stats["tasks_completed"] > 0 else 0
                }
        
        return report


def demonstrate_foundation_model_orchestration():
    """Demonstrate foundation model orchestration"""
    print("=" * 80)
    print("FOUNDATION MODEL ORCHESTRATION PATTERN DEMONSTRATION")
    print("=" * 80)
    
    orchestrator = FoundationModelOrchestrator()
    
    # Example 1: Simple task routing
    print("\n" + "=" * 80)
    print("Example 1: Simple Task Routing")
    print("=" * 80)
    
    tasks = [
        Task(
            id="task1",
            content="Summarize this article in 2 sentences: AI has transformed modern technology...",
            required_capabilities=[ModelCapability.SUMMARIZATION, ModelCapability.FAST_RESPONSE],
            priority="medium"
        ),
        Task(
            id="task2",
            content="Write a Python function to calculate fibonacci numbers recursively",
            required_capabilities=[ModelCapability.CODING],
            priority="high"
        ),
        Task(
            id="task3",
            content="Write a creative short story about a robot learning to paint",
            required_capabilities=[ModelCapability.CREATIVE],
            priority="low"
        )
    ]
    
    print("\nExecuting tasks with automatic model selection:")
    for task in tasks:
        try:
            result = orchestrator.execute_task(task)
            print(f"\nTask: {task.id}")
            print(f"Required capabilities: {[c.value for c in task.required_capabilities]}")
            print(f"Selected model: {result.model_used}")
            print(f"Execution time: {result.execution_time:.3f}s")
            print(f"Estimated cost: ${result.estimated_cost:.6f}")
            print(f"Quality estimate: {result.quality_estimate:.2f}")
            print(f"Result preview: {result.result[:100]}...")
        except Exception as e:
            print(f"\nTask {task.id} failed: {e}")
    
    # Example 2: Cost-constrained execution
    print("\n" + "=" * 80)
    print("Example 2: Cost-Constrained Execution")
    print("=" * 80)
    
    budget_task = Task(
        id="budget_task",
        content="Analyze the sentiment of this review: The product was okay but overpriced.",
        required_capabilities=[ModelCapability.ANALYSIS],
        priority="medium",
        max_cost=0.001  # Very tight budget
    )
    
    try:
        result = orchestrator.execute_task(budget_task)
        print(f"\nTask with max cost ${budget_task.max_cost:.6f}:")
        print(f"Selected model: {result.model_used}")
        print(f"Actual cost: ${result.estimated_cost:.6f}")
        print(f"Result: {result.result[:150]}...")
    except Exception as e:
        print(f"Task failed: {e}")
    
    # Example 3: Quality-focused execution
    print("\n" + "=" * 80)
    print("Example 3: Quality-Focused Execution")
    print("=" * 80)
    
    quality_task = Task(
        id="quality_task",
        content="Solve this logic puzzle: If all A are B, and some B are C, what can we conclude?",
        required_capabilities=[ModelCapability.REASONING],
        priority="high",
        min_quality=0.9  # High quality requirement
    )
    
    try:
        result = orchestrator.execute_task(quality_task)
        print(f"\nTask with min quality {quality_task.min_quality}:")
        print(f"Selected model: {result.model_used}")
        print(f"Quality estimate: {result.quality_estimate}")
        print(f"Result: {result.result[:200]}...")
    except Exception as e:
        print(f"Task failed: {e}")
    
    # Example 4: Ensemble execution
    print("\n" + "=" * 80)
    print("Example 4: Ensemble Execution")
    print("=" * 80)
    
    ensemble_task = Task(
        id="ensemble_task",
        content="What are the three most important factors in climate change?",
        required_capabilities=[ModelCapability.ANALYSIS, ModelCapability.REASONING],
        priority="high"
    )
    
    try:
        ensemble_result = orchestrator.execute_with_ensemble(ensemble_task, num_models=2)
        print(f"\nTask executed with ensemble approach:")
        print(f"Models used: {', '.join(ensemble_result['models_used'])}")
        print(f"\nAggregated result:")
        print(ensemble_result['aggregated_result'][:200] + "...")
        
        print(f"\nAll model results:")
        for i, r in enumerate(ensemble_result['all_results'], 1):
            print(f"\n  Model {i}: {r['model']} (quality: {r['quality']})")
            print(f"  Result: {r['result'][:100]}...")
    except Exception as e:
        print(f"Ensemble execution failed: {e}")
    
    # Example 5: Batch processing
    print("\n" + "=" * 80)
    print("Example 5: Batch Processing")
    print("=" * 80)
    
    batch_tasks = [
        Task(f"batch_{i}", f"Process item {i}: analyze and summarize", 
             [ModelCapability.ANALYSIS], "medium")
        for i in range(3)
    ]
    
    print(f"\nProcessing batch of {len(batch_tasks)} tasks:")
    batch_results = orchestrator.execute_batch(batch_tasks)
    
    print(f"Completed: {len(batch_results)}/{len(batch_tasks)} tasks")
    for result in batch_results:
        print(f"  {result.task_id}: {result.model_used} in {result.execution_time:.3f}s")
    
    # Example 6: Performance reporting
    print("\n" + "=" * 80)
    print("Example 6: Performance Reporting")
    print("=" * 80)
    
    report = orchestrator.get_performance_report()
    
    print("\nPerformance Report:")
    print("-" * 60)
    for model_name, stats in report.items():
        print(f"\n{model_name}:")
        print(f"  Tasks completed: {stats['tasks_completed']}")
        print(f"  Failures: {stats['failures']}")
        print(f"  Success rate: {stats['success_rate']:.1%}")
        print(f"  Total cost: ${stats['total_cost']:.6f}")
        print(f"  Avg latency: {stats['avg_latency']:.3f}s")
        print(f"  Cost per task: ${stats['cost_per_task']:.6f}")
    
    # Summary
    print("\n" + "=" * 80)
    print("SUMMARY")
    print("=" * 80)
    print("""
The Foundation Model Orchestration pattern enables:
✓ Intelligent routing of tasks to optimal models
✓ Cost-performance trade-off optimization
✓ Quality-focused model selection
✓ Ensemble execution for critical tasks
✓ Fallback handling for reliability
✓ Performance monitoring and optimization
✓ Batch processing with automatic routing

This pattern is valuable for:
- Production AI systems with cost constraints
- Multi-model architectures
- Quality-performance optimization
- Resource-efficient AI pipelines
- Fault-tolerant AI systems
- Adaptive model selection
    """)


if __name__ == "__main__":
    demonstrate_foundation_model_orchestration()

    

Agentic Design Pattern: Foundation Model Orchestration

Pattern 165: Foundation Model Orchestration – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Foundation Model Orchestration pattern implementation.

---

Pattern Description

Pattern 165: Foundation Model Orchestration

Description: The Foundation Model Orchestration pattern coordinates multiple foundation models (LLMs with different capabilities, sizes, and specializations) to solve complex tasks optimally. The orchestrator routes tasks to the most appropriate model based on requirements, capabilities, cost, and performance trade-offs.

Components: 1. Model Registry: Catalog of available models and their capabilities 2. Task Analyzer: Analyzes task requirements and complexity 3. Model Selector: Chooses optimal model for each task 4. Result Aggregator: Combines results from multiple models 5. Performance Monitor: Tracks model performance and costs 6. Fallback Manager: Handles model failures

Use Cases: – Cost-optimized AI pipelines – Quality-performance trade-offs – Specialized task routing – Multi-model ensemble systems – Adaptive model selection – Resource optimization

Benefits: – Optimal cost-performance balance – Better quality through specialization – Fault tolerance with fallbacks – Flexible model upgrades – Resource efficiency

Trade-offs: – Orchestration complexity – Latency overhead – Model compatibility issues – Cost tracking complexity – Requires multiple API keys

LangChain Implementation: Uses LangChain’s model abstraction to coordinate multiple models. Implements intelligent routing logic based on task characteristics and model capabilities.

---

File Statistics

• Total Lines: 627
• Classes: 6
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 165: Foundation Model Orchestration

Description:
    The Foundation Model Orchestration pattern coordinates multiple foundation models
    (LLMs with different capabilities, sizes, and specializations) to solve complex tasks
    optimally. The orchestrator routes tasks to the most appropriate model based on
    requirements, capabilities, cost, and performance trade-offs.

Components:
    1. Model Registry: Catalog of available models and their capabilities
    2. Task Analyzer: Analyzes task requirements and complexity
    3. Model Selector: Chooses optimal model for each task
    4. Result Aggregator: Combines results from multiple models
    5. Performance Monitor: Tracks model performance and costs
    6. Fallback Manager: Handles model failures

Use Cases:
    - Cost-optimized AI pipelines
    - Quality-performance trade-offs
    - Specialized task routing
    - Multi-model ensemble systems
    - Adaptive model selection
    - Resource optimization

Benefits:
    - Optimal cost-performance balance
    - Better quality through specialization
    - Fault tolerance with fallbacks
    - Flexible model upgrades
    - Resource efficiency

Trade-offs:
    - Orchestration complexity
    - Latency overhead
    - Model compatibility issues
    - Cost tracking complexity
    - Requires multiple API keys

LangChain Implementation:
    Uses LangChain's model abstraction to coordinate multiple models. Implements
    intelligent routing logic based on task characteristics and model capabilities.
"""

Purpose: This module implements the Foundation Model Orchestration pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import collections.defaultdict
import dataclasses.dataclass
import dataclasses.field
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import time
import typing.Any
import typing.Callable
import typing.Dict
import typing.List
import typing.Optional

Purpose: Core Python functionality: – collections.defaultdict: Specialized container datatypes – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – enum.Enum: Enumeration support – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – time: Time-related functions – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Callable: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: ModelCapability

Lines 60-69

Inherits from: Enum

Purpose:

Model capability types

---

Class: ModelTier

Lines 72-78

Inherits from: Enum

Purpose:

Model tier for cost/performance trade-off

---

Class: ModelProfile

Lines 82-93

Purpose:

Profile of a foundation model

---

Class: Task

Lines 97-105

Purpose:

A task to be processed

---

Class: TaskResult

Lines 109-117

Purpose:

Result of task execution

---

Class: FoundationModelOrchestrator

Lines 120-444

Purpose:

Orchestrates multiple foundation models for optimal task execution

Methods:

__init__(self)

Lines 123-147

Purpose:

Initialize the orchestrator with model registry

Parameters:

_register_default_models(self)

Lines 149-198

Purpose:

Register default model profiles

Parameters:

register_model(self, profile)

Lines 200-212

Purpose:

Register a model in the orchestrator

Parameters: – profile: Profile

execute_task(self, task, use_fallback)

Lines 214-261

Purpose:

Execute a task using the most appropriate model

Parameters: – task: Task – use_fallback: Use fallback

Returns: TaskResult

execute_batch(self, tasks, parallel)

Lines 263-274

Purpose:

Execute multiple tasks

Parameters: – tasks: Tasks – parallel: Parallel

Returns: <ast.Subscript object at 0x000001A06D6210D0>

execute_with_ensemble(self, task, num_models)

Lines 276-303

Purpose:

Execute task on multiple models and aggregate results

Parameters: – task: Task – num_models: Language model instance

Returns: <ast.Subscript object at 0x000001A06E0ABE50>

_select_model(self, task)

Lines 305-336

Purpose:

Select the best model for a task

Parameters: – task: Task

Returns: <ast.Subscript object at 0x000001A06D6744D0>

_rank_models_for_task(self, task)

Lines 338-382

Purpose:

Rank models by suitability for task

Parameters: – task: Task

Returns: <ast.Subscript object at 0x000001A06D7C90D0>

_select_fallback_model(self, task, failed_model)

Lines 384-393

Purpose:

Select a fallback model after primary fails

Parameters: – task: Task – failed_model: Language model instance

Returns: <ast.Subscript object at 0x000001A06D7C9090>

_execute_on_model(self, task, model)

Lines 395-411

Purpose:

Execute task on specific model

Parameters: – task: Task – model: Language model instance

Returns: str

_update_stats(self, model_name, execution_time, cost, success)

Lines 413-423

Purpose:

Update performance statistics

Parameters: – model_name: Language model instance – execution_time: Execution time – cost: Cost – success: Success

get_performance_report(self)

Lines 425-444

Purpose:

Get performance report for all models

Parameters:

Returns: <ast.Subscript object at 0x000001A06D889090>

---

Functions

demonstrate_foundation_model_orchestration()

Lines 447-622

Purpose:

Demonstrate foundation model orchestration

---

Key Concepts Summary

Pattern Overview

The Foundation Model Orchestration pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

---

Best Practices Demonstrated

1. Environment Configuration

• ✅ Use .env files for API keys
• ✅ Never hardcode credentials
• ✅ Validate environment variables on startup

2. Error Handling

• ✅ Graceful degradation on failures
• ✅ Informative error messages
• ✅ Logging for debugging

3. Prompt Engineering

• ✅ Clear, specific instructions
• ✅ Structured output formats
• ✅ Few-shot examples when beneficial

4. Code Organization

• ✅ Separation of concerns (class-based design)
• ✅ Type hints for clarity
• ✅ Docstrings for documentation

5. Performance Optimization

• ✅ Appropriate temperature settings
• ✅ Token budget management
• ✅ Caching where applicable

---

LangChain Components Used

• ChatOpenAI: OpenAI model interface
• Chains: Sequential component composition

Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development

---

Execution Flow

1. Initialize agent with configuration
   ↓
2. Load environment and dependencies
   ↓
3. Prepare prompt with input data
   ↓
4. Invoke LLM through LangChain
   ↓
5. Parse and validate response
   ↓
6. Return formatted result

Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation

---

Performance Considerations

Computational Cost

• LLM Calls: Primary cost factor (based on tokens)
• Latency: Depends on model and input size
• Memory: Minimal for most patterns

Optimization Strategies

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

• GPT-3.5-turbo: ~$0.002 per request
• GPT-4: ~$0.03 per request
• GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

---

Further Improvements

Production Enhancements

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 164
• Pattern 166

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Foundation Model Orchestration pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns

🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability

📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/

---

File: 165_foundation_model_orchestration.py
Total Lines: 627
Pattern Category: AI Agent Design Pattern #165

---

Generated from actual implementation code
For educational and reference purposes