💻 AI Orchestrator Implementation Example

See how to coordinate multiple AI models with this simple orchestrator pattern

import asyncio
from typing import List, Dict, Any
from dataclasses import dataclass

@dataclass
class AIModel:
    name: str
    specialty: str
    cost_per_token: float

class AIOrchestrator:
    """
    Simple orchestrator that routes tasks to appropriate AI models
    based on their specialties and cost efficiency
    """
    
    def __init__(self):
        self.models = [
            AIModel("code-llama", "programming", 0.002),
            AIModel("claude-3", "reasoning", 0.008),
            AIModel("gpt-4", "general", 0.03)
        ]
    
    async def route_task(self, task: str, task_type: str) -> Dict[str, Any]:
        """
        Route task to the most appropriate AI model
        """
        # Filter models by specialty
        suitable_models = [
            model for model in self.models 
            if task_type in model.specialty or model.specialty == "general"
        ]
        
        if not suitable_models:
            return {"error": "No suitable model found"}
        
        # Select most cost-effective model
        selected_model = min(suitable_models, key=lambda m: m.cost_per_token)
        
        # Simulate API call
        result = await self.call_model(selected_model, task)
        
        return {
            "model": selected_model.name,
            "result": result,
            "cost_estimate": selected_model.cost_per_token * len(task)
        }
    
    async def call_model(self, model: AIModel, task: str) -> str:
        """Simulate AI model API call"""
        await asyncio.sleep(0.1)  # Simulate network delay
        return f"Processed by {model.name}: {task[:50]}..."

# Usage example
async def main():
    orchestrator = AIOrchestrator()
    
    # Route a programming task
    result = await orchestrator.route_task(
        "Write a Python function to sort a list",
        "programming"
    )
    print(f"Selected: {result['model']}")
    print(f"Result: {result['result']}")
    print(f"Cost: ${result['cost_estimate']:.4f}")

# Run the orchestrator
if __name__ == "__main__":
    asyncio.run(main())