Integrating Multiple LLM APIs for Brand Monitoring: Technical Guide

from abc import ABC, abstractmethod
from typing import List, Dict, Optional
import asyncio
from enum import Enum

class LLMProvider(Enum):
    OPENAI = "openai"
    CLAUDE = "claude"
    PERPLEXITY = "perplexity"
    GEMINI = "gemini"

class BaseLLMClient(ABC):
    @abstractmethod
    async def query(self, prompt: str, **kwargs) -> Dict:
        """Query the LLM provider"""
        pass
    
    @abstractmethod
    def get_provider_name(self) -> str:
        """Get provider name"""
        pass

class UnifiedLLMClient:
    def __init__(self):
        self.clients = {
            LLMProvider.OPENAI: OpenAIClient(OPENAI_API_KEY),
            LLMProvider.CLAUDE: ClaudeClient(CLAUDE_API_KEY),
            LLMProvider.PERPLEXITY: PerplexityClient(PERPLEXITY_API_KEY),
            LLMProvider.GEMINI: GeminiClient(GEMINI_API_KEY),
        }
        self.rate_limiters = {}
        self.circuit_breakers = {}
    
    async def query_all_providers(
        self,
        prompt: str,
        providers: Optional[List[LLMProvider]] = None
    ) -> Dict[str, Dict]:
        """Query all or specified providers concurrently"""
        if providers is None:
            providers = list(LLMProvider)
        
        tasks = []
        for provider in providers:
            if self._is_circuit_open(provider):
                tasks.append(self._query_with_fallback(provider, prompt))
        
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        return {
            provider.value: result
            for provider, result in zip(providers, results)
            if not isinstance(result, Exception)
        }
    
    async def _query_with_fallback(
        self,
        provider: LLMProvider,
        prompt: str
    ) -> Dict:
        """Query with automatic fallback and retry logic"""
        max_retries = 3
        backoff_factor = 2
        
        for attempt in range(max_retries):
            try:
                if not self._check_rate_limit(provider):
                    await asyncio.sleep(backoff_factor ** attempt)
                    continue
                
                result = await self.clients[provider].query(prompt)
                self._record_success(provider)
                return result
            
            except Exception as e:
                if attempt == max_retries - 1:
                    self._record_failure(provider)
                    raise
                await asyncio.sleep(backoff_factor ** attempt)
        
        return None

import time
from collections import deque
from typing import Dict
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"  # Normal operation
    OPEN = "open"      # Failing, don't send requests
    HALF_OPEN = "half_open"  # Testing if service recovered

class RateLimiter:
    def __init__(self, max_requests: int, time_window: int):
        self.max_requests = max_requests
        self.time_window = time_window
        self.requests = deque()
    
    def is_allowed(self) -> bool:
        """Check if request is allowed under rate limit"""
        now = time.time()
        
        # Remove old requests outside time window
        while self.requests and self.requests[0] < now - self.time_window:
            self.requests.popleft()
        
        if len(self.requests) < self.max_requests:
            self.requests.append(now)
            return True
        
        return False
    
    def get_wait_time(self) -> float:
        """Get time to wait before next request"""
        if not self.requests:
            return 0.0
        oldest = self.requests[0]
        return max(0, self.time_window - (time.time() - oldest))

class CircuitBreaker:
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: int = 60,
        success_threshold: int = 2
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.success_threshold = success_threshold
        self.failure_count = 0
        self.success_count = 0
        self.state = CircuitState.CLOSED
        self.last_failure_time = None
    
    def call(self, func, *args, **kwargs):
        """Execute function with circuit breaker protection"""
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
                self.success_count = 0
            else:
                raise Exception("Circuit breaker is OPEN")
        
        try:
            result = func(*args, **kwargs)
            self._record_success()
            return result
        except Exception as e:
            self._record_failure()
            raise e
    
    def _record_success(self):
        """Record successful call"""
        self.failure_count = 0
        if self.state == CircuitState.HALF_OPEN:
            self.success_count += 1
            if self.success_count >= self.success_threshold:
                self.state = CircuitState.CLOSED
    
    def _record_failure(self):
        """Record failed call"""
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN