aishastr.live

Agentic Design Pattern: Model Routing And Selection

Written by

admin

in


    """
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

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *

More posts