"""
Pattern 200: Model Router Pattern

Intelligently route queries to the most appropriate model based on task type,
complexity, cost, latency requirements, or other criteria.
"""

from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


class TaskType(Enum):
    """Types of tasks for routing."""
    SIMPLE_QA = "simple_qa"
    COMPLEX_REASONING = "complex_reasoning"
    CODE_GENERATION = "code_generation"
    CREATIVE_WRITING = "creative_writing"
    TRANSLATION = "translation"
    SUMMARIZATION = "summarization"
    CLASSIFICATION = "classification"


@dataclass
class Model:
    """Model configuration for routing."""
    name: str
    model_id: str
    strengths: List[TaskType]
    cost_per_token: float
    avg_latency_ms: float
    max_tokens: int
    llm: Optional[ChatOpenAI] = None
    
    def __post_init__(self):
        if self.llm is None:
            # Set temperature based on task strengths
            temp = 0.7 if TaskType.CREATIVE_WRITING in self.strengths else 0.3
            self.llm = ChatOpenAI(model=self.model_id, temperature=temp)


class ModelRouterPattern:
    """
    Model Router Pattern - Route to optimal model.
    """
    
    def __init__(self):
        self.models: List[Model] = []
        self.router_llm = ChatOpenAI(model="gpt-4", temperature=0)
        self.routing_history: List[Dict[str, Any]] = []
    
    def register_model(self, model: Model):
        """Register a model for routing."""
        self.models.append(model)
    
    def classify_task(self, query: str) -> TaskType:
        """Classify the task type using LLM."""
        task_types = ", ".join([t.value for t in TaskType])
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"Classify the task type. Options: {task_types}. Respond with only the task type."),
            ("human", "{query}")
        ])
        
        chain = prompt | self.router_llm
        response = chain.invoke({"query": query})
        
        # Parse response to TaskType
        response_text = str(response.content).lower().strip()
        for task_type in TaskType:
            if task_type.value in response_text:
                return task_type
        
        return TaskType.SIMPLE_QA  # Default
    
    def select_model(self, task_type: TaskType, preferences: Optional[Dict[str, Any]] = None) -> Model:
        """Select best model for task type based on preferences."""
        preferences = preferences or {}
        
        # Filter models by task type strength
        suitable_models = [m for m in self.models if task_type in m.strengths]
        
        if not suitable_models:
            suitable_models = self.models  # Fallback to all models
        
        if not suitable_models:
            raise ValueError("No models registered")
        
        # Select based on preferences
        optimize_for = preferences.get("optimize_for", "quality")
        
        if optimize_for == "cost":
            return min(suitable_models, key=lambda m: m.cost_per_token)
        elif optimize_for == "latency":
            return min(suitable_models, key=lambda m: m.avg_latency_ms)
        elif optimize_for == "quality":
            # Assume first model in strengths list is highest quality
            return suitable_models[0]
        else:
            return suitable_models[0]
    
    def run(self, query: str, preferences: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
        """Route query to appropriate model and execute."""
        # Classify task
        task_type = self.classify_task(query)
        print(f"Task classified as: {task_type.value}")
        
        # Select model
        selected_model = self.select_model(task_type, preferences)
        print(f"Routing to model: {selected_model.name}\n")
        
        # Execute with selected model
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a helpful AI assistant."),
            ("human", "{query}")
        ])
        
        chain = prompt | selected_model.llm
        response = chain.invoke({"query": query})
        
        # Track routing decision
        routing_record = {
            "query": query,
            "task_type": task_type.value,
            "selected_model": selected_model.name,
            "model_cost": selected_model.cost_per_token,
            "model_latency": selected_model.avg_latency_ms
        }
        self.routing_history.append(routing_record)
        
        return {
            "query": query,
            "task_type": task_type.value,
            "routed_to": selected_model.name,
            "result": response.content,
            "routing_metadata": {
                "cost_per_token": selected_model.cost_per_token,
                "avg_latency_ms": selected_model.avg_latency_ms
            }
        }
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get routing statistics."""
        model_usage = {}
        for record in self.routing_history:
            model = record["selected_model"]
            model_usage[model] = model_usage.get(model, 0) + 1
        
        return {
            "total_queries": len(self.routing_history),
            "model_usage": model_usage,
            "registered_models": len(self.models)
        }


# Example usage
if __name__ == "__main__":
    print("=== Model Router Pattern Demo ===\n")
    
    router = ModelRouterPattern()
    
    # Register models with different capabilities
    router.register_model(Model(
        name="GPT-4",
        model_id="gpt-4",
        strengths=[TaskType.COMPLEX_REASONING, TaskType.CODE_GENERATION],
        cost_per_token=0.00003,
        avg_latency_ms=2000,
        max_tokens=8000
    ))
    
    router.register_model(Model(
        name="GPT-3.5-Turbo",
        model_id="gpt-3.5-turbo",
        strengths=[TaskType.SIMPLE_QA, TaskType.SUMMARIZATION, TaskType.CLASSIFICATION],
        cost_per_token=0.000002,
        avg_latency_ms=800,
        max_tokens=4000
    ))
    
    router.register_model(Model(
        name="GPT-4-Creative",
        model_id="gpt-4",
        strengths=[TaskType.CREATIVE_WRITING, TaskType.TRANSLATION],
        cost_per_token=0.00003,
        avg_latency_ms=2000,
        max_tokens=8000
    ))
    
    # Test different queries
    queries = [
        "What is 2 + 2?",
        "Write a Python function to implement quicksort with detailed comments",
        "Write a creative short story about a robot discovering emotions"
    ]
    
    for i, query in enumerate(queries, 1):
        print(f"\nQuery {i}: {query}")
        print("-" * 60)
        result = router.run(query, preferences={"optimize_for": "quality"})
        print(f"Routed to: {result['routed_to']}")
        print(f"Result: {result['result'][:150]}...")
        print()
    
    # Statistics
    print("\n=== Routing Statistics ===")
    stats = router.get_statistics()
    print(f"Total queries: {stats['total_queries']}")
    print(f"Model usage: {stats['model_usage']}")

    

Agentic Architecture Pattern: Model Router

Pattern 200: 200 Model Router Pattern – Line-by-Line Explanation

Overview

Pattern 200: Model Router Pattern

Intelligently route queries to the most appropriate model based on task type, complexity, cost, latency requirements, or other criteria.

Code Structure Analysis

File Information

• Pattern Number: 200
• Pattern Name: 200 Model Router Pattern
• File Path: patterns\llm_specific\200_model_router_pattern.py
• Total Lines: 212

Detailed Code Explanation

Module Docstring

"""
Pattern 200: Model Router Pattern

Intelligently route queries to the most appropriate model based on task type,
complexity, cost, latency requirements, or other criteria.
"""

Explanation: This module docstring describes the purpose and key features of the 200 Model Router Pattern pattern.

Imports Section

from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, Callable: Type hints for better code documentation – from dataclasses import dataclass: Data class decorators for clean data structures – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Enumerations

TaskType (Line 15)

class TaskType(Enum):
    SIMPLE_QA
    COMPLEX_REASONING
    CODE_GENERATION
    CREATIVE_WRITING
    TRANSLATION
    SUMMARIZATION
    CLASSIFICATION

Purpose: Types of tasks for routing.

Values: – SIMPLE_QA: Represents simple qa – COMPLEX_REASONING: Represents complex reasoning – CODE_GENERATION: Represents code generation – CREATIVE_WRITING: Represents creative writing – TRANSLATION: Represents translation – SUMMARIZATION: Represents summarization – CLASSIFICATION: Represents classification

Classes

TaskType (Line 15)

Description: Types of tasks for routing.

Model (Line 27)

Description: Model configuration for routing.

Methods:

__post_init__ (Line 37)

def __post_init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

ModelRouterPattern (Line 44)

Description: Model Router Pattern – Route to optimal model.

Methods:

__init__ (Line 49)

def __init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

register_model (Line 54)

def register_model(self, model):
    """Register a model for routing."""

Purpose: Register a model for routing.

Parameters: – self: Instance reference – model: Parameter for model

classify_task (Line 58)

def classify_task(self, query):
    """Classify the task type using LLM."""

Purpose: Classify the task type using LLM.

Parameters: – self: Instance reference – query: Parameter for query

select_model (Line 78)

def select_model(self, task_type, preferences):
    """Select best model for task type based on preferences."""

Purpose: Select best model for task type based on preferences.

Parameters: – self: Instance reference – task_type: Parameter for task_type – preferences: Parameter for preferences

run (Line 104)

def run(self, query, preferences):
    """Route query to appropriate model and execute."""

Purpose: Route query to appropriate model and execute.

Parameters: – self: Instance reference – query: Parameter for query – preferences: Parameter for preferences

get_statistics (Line 144)

def get_statistics(self):
    """Get routing statistics."""

Purpose: Get routing statistics.

Parameters: – self: Instance reference

Design Patterns Used

• Strategy Pattern: Different execution strategies for various scenarios
• State Pattern: Managing different states in the workflow
• Chain of Responsibility: Sequential processing through workflow nodes
• Observer Pattern: Tracking and monitoring pattern execution
• Factory Pattern: Creating instances with proper configuration

Usage Example

# Initialize the pattern
pattern = TaskType(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

• LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
• Workflow Orchestration: LangGraph manages execution flow through nodes
• State Management: TypedDict defines workflow state structure
• Error Handling: Comprehensive error handling and fallback mechanisms
• Extensibility: Designed for easy customization and extension

Extension Points

This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities