Upload app/agents/llm_orchestrator_agent.py with huggingface_hub

app/agents/llm_orchestrator_agent.py

@@ -0,0 +1,868 @@
+"""
+LLM Orchestrator Agent
+
+Multi-model orchestration with rate limit handling:
+- ChatGPT (OpenAI)
+- Gemini (Google AI)
+- Automatic retry on rate limits
+- Fallback mechanisms
+
+Inspired by GestLLM/GestOS research for gesture-to-LLM integration.
+"""
+
+import asyncio
+import time
+import json
+from typing import Dict, List, Any, Optional
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta
+from enum import Enum
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class LLMProvider(Enum):
+    CHATGPT = "chatgpt"
+    GEMINI = "gemini"
+    CLAUDE = "claude"
+    DEEPSEEK = "deepseek"
+    OLLAMA = "ollama"
+    GROQ = "groq"
+
+
+@dataclass
+class RateLimitConfig:
+    """Rate limit configuration per provider"""
+    requests_per_minute: int = 60
+    requests_per_day: int = 500
+    tokens_per_minute: int = 90000
+    retry_after_seconds: int = 60
+    max_retries: int = 5
+
+
+@dataclass
+class RateLimitState:
+    """Current rate limit state"""
+    request_timestamps: List[datetime] = field(default_factory=list)
+    daily_requests: int = 0
+    daily_reset: datetime = field(default_factory=lambda: datetime.now() + timedelta(days=1))
+    token_usage: List[tuple] = field(default_factory=list)
+    is_rate_limited: bool = False
+    retry_after: Optional[datetime] = None
+
+
+@dataclass
+class LLMRequest:
+    """A request to an LLM"""
+    prompt: str
+    system_prompt: Optional[str] = None
+    model: str = "gpt-4"
+    max_tokens: int = 2000
+    temperature: float = 0.7
+    providers: List[LLMProvider] = field(default_factory=lambda: [LLMProvider.CHATGPT, LLMProvider.GEMINI])
+    timeout: int = 60
+    user_id: str = "default"
+    models: Optional[Dict[str, str]] = None
+
+
+@dataclass
+class LLMResponse:
+    """Response from an LLM"""
+    content: str
+    provider: LLMProvider
+    model: str
+    tokens_used: int
+    latency_ms: float
+    success: bool
+    error: Optional[str] = None
+    cached: bool = False
+
+
+class RateLimitHandler:
+    """
+    Handles rate limiting with automatic retry.
+    
+    Features:
+    - Token bucket algorithm
+    - Per-provider limits
+    - Exponential backoff
+    - Automatic retry when limits refresh
+    """
+    
+    def __init__(self, config: Optional[Dict[str, RateLimitConfig]] = None):
+        self.configs = config or {
+            LLMProvider.CHATGPT: RateLimitConfig(
+                requests_per_minute=60,
+                requests_per_day=500,
+                retry_after_seconds=60
+            ),
+            LLMProvider.GEMINI: RateLimitConfig(
+                requests_per_minute=60,
+                requests_per_day=150,
+                retry_after_seconds=30
+            ),
+            LLMProvider.CLAUDE: RateLimitConfig(
+                requests_per_minute=50,
+                requests_per_day=200,
+                retry_after_seconds=60
+            ),
+            LLMProvider.DEEPSEEK: RateLimitConfig(
+                requests_per_minute=60,
+                requests_per_day=200,
+                retry_after_seconds=60
+            ),
+            LLMProvider.OLLAMA: RateLimitConfig(
+                requests_per_minute=1000,
+                requests_per_day=100000,
+                retry_after_seconds=1
+            ),
+            LLMProvider.GROQ: RateLimitConfig(
+                requests_per_minute=30,
+                requests_per_day=10000,
+                retry_after_seconds=60
+            )
+        }
+        
+        self.states: Dict[LLMProvider, RateLimitState] = {
+            provider: RateLimitState() for provider in LLMProvider
+        }
+        
+        self.request_queue: Dict[LLMProvider, List[asyncio.Task]] = {}
+        self._lock = asyncio.Lock()
+    
+    async def acquire(self, provider: LLMProvider, priority: int = 0) -> bool:
+        """Acquire permission to make a request"""
+        async with self._lock:
+            state = self.states[provider]
+            config = self.configs.get(provider, RateLimitConfig())
+            
+            now = datetime.now()
+            
+            if state.retry_after and now < state.retry_after:
+                return False
+            
+            if state.daily_reset < now:
+                state.daily_requests = 0
+                state.daily_reset = now + timedelta(days=1)
+            
+            minute_ago = now - timedelta(minutes=1)
+            state.request_timestamps = [
+                ts for ts in state.request_timestamps if ts > minute_ago
+            ]
+            
+            if state.daily_requests >= config.requests_per_day:
+                state.is_rate_limited = True
+                state.retry_after = state.daily_reset
+                return False
+            
+            if len(state.request_timestamps) >= config.requests_per_minute:
+                oldest = min(state.request_timestamps)
+                wait_time = (oldest - minute_ago).total_seconds()
+                if wait_time > 0:
+                    state.retry_after = now + timedelta(seconds=wait_time)
+                    return False
+            
+            return True
+    
+    async def release(self, provider: LLMProvider, tokens_used: int = 0):
+        """Release a request slot"""
+        async with self._lock:
+            state = self.states[provider]
+            now = datetime.now()
+            
+            state.request_timestamps.append(now)
+            state.daily_requests += 1
+            
+            if tokens_used > 0:
+                state.token_usage.append((now, tokens_used))
+            
+            state.retry_after = None
+            state.is_rate_limited = False
+    
+    def set_rate_limited(self, provider: LLMProvider, retry_after_seconds: int):
+        """Manually set a provider as rate limited from API response"""
+        state = self.states[provider]
+        state.is_rate_limited = True
+        state.retry_after = datetime.now() + timedelta(seconds=retry_after_seconds)
+        
+        config = self.configs.get(provider, RateLimitConfig())
+        logger.warning(
+            f"Rate limited for {provider.value}: retrying after {retry_after_seconds}s"
+        )
+    
+    async def wait_for_slot(self, provider: LLMProvider, max_wait: int = 120) -> bool:
+        """Wait for a rate limit slot to become available"""
+        start = time.time()
+        
+        while time.time() - start < max_wait:
+            if await self.acquire(provider):
+                return True
+            
+            state = self.states[provider]
+            if state.retry_after:
+                wait_seconds = (state.retry_after - datetime.now()).total_seconds()
+                if wait_seconds > 0:
+                    await asyncio.sleep(min(wait_seconds, 5))
+            else:
+                await asyncio.sleep(1)
+        
+        return False
+    
+    def get_status(self) -> Dict:
+        """Get rate limit status for all providers"""
+        now = datetime.now()
+        status = {}
+        
+        for provider in LLMProvider:
+            state = self.states[provider]
+            config = self.configs.get(provider, RateLimitConfig())
+            
+            minute_ago = now - timedelta(minutes=1)
+            recent_requests = sum(1 for ts in state.request_timestamps if ts > minute_ago)
+            
+            status[provider.value] = {
+                "rate_limited": state.is_rate_limited,
+                "requests_this_minute": recent_requests,
+                "requests_per_minute_limit": config.requests_per_minute,
+                "requests_today": state.daily_requests,
+                "requests_per_day_limit": config.requests_per_day,
+                "retry_after_seconds": (
+                    (state.retry_after - now).total_seconds() 
+                    if state.retry_after and state.retry_after > now else 0
+                )
+            }
+        
+        return status
+
+
+class LLMOrchestrator:
+    """
+    Multi-model LLM orchestration with gesture triggers.
+    
+    Features:
+    - Parallel queries to multiple LLMs
+    - Rate limit handling with auto-retry
+    - Response synthesis
+    - Gesture-triggered actions
+    
+    Inspired by:
+    - GestLLM: LLM-powered gesture interpretation
+    - GestOS: Multi-robot gesture orchestration
+    - GestureGPT: Free-form gesture understanding
+    """
+    
+    def __init__(self, api_keys: Optional[Dict[str, str]] = None):
+        self.api_keys = api_keys or {}
+        self.rate_limiter = RateLimitHandler()
+        
+        self.provider_configs = {
+            LLMProvider.CHATGPT: {
+                "model": "gpt-4",
+                "model_name": "gpt-4o",
+                "base_url": "https://api.openai.com/v1",
+                "supports_vision": True
+            },
+            LLMProvider.GEMINI: {
+                "model": "gemini-pro",
+                "model_name": "gemini-2.0-flash",
+                "base_url": "https://generativelanguage.googleapis.com/v1beta",
+                "supports_vision": True
+            },
+            LLMProvider.CLAUDE: {
+                "model": "claude-3-opus-20240229",
+                "model_name": "claude-3-5-sonnet",
+                "base_url": "https://api.anthropic.com/v1",
+                "supports_vision": True
+            },
+            LLMProvider.DEEPSEEK: {
+                "model": "deepseek-chat",
+                "model_name": "deepseek-chat",
+                "base_url": "https://api.deepseek.com/v1",
+                "supports_vision": False
+            },
+            LLMProvider.OLLAMA: {
+                "model": "llama3",
+                "model_name": "llama3",
+                "base_url": "http://localhost:11434/v1",
+                "supports_vision": False
+            },
+            LLMProvider.GROQ: {
+                "model": "llama-3.1-70b-versatile",
+                "model_name": "llama-3.1-70b-versatile",
+                "base_url": "https://api.groq.com/openai/v1",
+                "supports_vision": False
+            }
+        }
+        
+        self.cache: Dict[str, LLMResponse] = {}
+        self.cache_ttl = 3600
+        
+        self.pending_requests: List[Dict] = []
+        self.response_history: List[LLMResponse] = []
+    
+    async def query(
+        self, 
+        request: LLMRequest,
+        preferred_provider: Optional[LLMProvider] = None
+    ) -> LLMResponse:
+        """Query an LLM with rate limit handling"""
+        
+        providers_to_try = (
+            [preferred_provider] if preferred_provider 
+            else request.providers
+        )
+        
+        last_error = None
+        
+        for attempt in range(self.rate_limiter.configs.get(
+            providers_to_try[0], RateLimitConfig()
+        ).max_retries):
+            
+            for provider in providers_to_try:
+                if not await self.rate_limiter.acquire(provider):
+                    continue
+                
+                try:
+                    response = await self._call_provider(provider, request)
+                    await self.rate_limiter.release(provider, response.tokens_used)
+                    
+                    self.response_history.append(response)
+                    return response
+                    
+                except RateLimitError as e:
+                    await self.rate_limiter.release(provider, 0)
+                    self.rate_limiter.set_rate_limited(
+                        provider, 
+                        e.retry_after or 60
+                    )
+                    last_error = e
+                    
+                except Exception as e:
+                    await self.rate_limiter.release(provider, 0)
+                    last_error = e
+                    logger.error(f"LLM call failed: {e}")
+            
+            if last_error and not isinstance(last_error, RateLimitError):
+                break
+            
+            if providers_to_try[0]:
+                await asyncio.sleep(2 ** attempt)
+        
+        return LLMResponse(
+            content="",
+            provider=providers_to_try[0] if providers_to_try else LLMProvider.CHATGPT,
+            model="",
+            tokens_used=0,
+            latency_ms=0,
+            success=False,
+            error=str(last_error) if last_error else "All providers failed"
+        )
+    
+    async def query_parallel(
+        self, 
+        request: LLMRequest
+    ) -> List[LLMResponse]:
+        """Query multiple LLMs in parallel and return all responses"""
+        tasks = []
+        
+        for provider in request.providers:
+            provider_request = LLMRequest(
+                prompt=request.prompt,
+                system_prompt=request.system_prompt,
+                model=request.model,
+                max_tokens=request.max_tokens,
+                temperature=request.temperature,
+                providers=[provider],
+                timeout=request.timeout,
+                user_id=request.user_id
+            )
+            tasks.append(self.query(provider_request, provider))
+        
+        responses = await asyncio.gather(*tasks, return_exceptions=True)
+        
+        valid_responses = []
+        for r in responses:
+            if isinstance(r, Exception):
+                valid_responses.append(LLMResponse(
+                    content="",
+                    provider=LLMProvider.CHATGPT,
+                    model="",
+                    tokens_used=0,
+                    latency_ms=0,
+                    success=False,
+                    error=str(r)
+                ))
+            else:
+                valid_responses.append(r)
+        
+        return valid_responses
+    
+    async def query_with_retry(
+        self,
+        request: LLMRequest,
+        max_attempts: int = 3
+    ) -> LLMResponse:
+        """Query with automatic retry on rate limits"""
+        
+        for attempt in range(max_attempts):
+            response = await self.query(request)
+            
+            if response.success:
+                return response
+            
+            if "rate_limit" in (response.error or "").lower():
+                await asyncio.sleep(30 * (attempt + 1))
+                continue
+            
+            return response
+        
+        return response
+    
+    async def _call_provider(
+        self, 
+        provider: LLMProvider, 
+        request: LLMRequest
+    ) -> LLMResponse:
+        """Call a specific LLM provider"""
+        start_time = time.time()
+        
+        config = self.provider_configs.get(provider, {})
+        model = config.get("model_name", request.model)
+        
+        if request.models and provider.value in request.models:
+            model = request.models[provider.value]
+        
+        if provider == LLMProvider.CHATGPT:
+            return await self._call_chatgpt(request, model, start_time)
+        elif provider == LLMProvider.GEMINI:
+            return await self._call_gemini(request, model, start_time)
+        elif provider == LLMProvider.CLAUDE:
+            return await self._call_claude(request, model, start_time)
+        elif provider == LLMProvider.DEEPSEEK:
+            return await self._call_deepseek(request, model, start_time)
+        elif provider == LLMProvider.OLLAMA:
+            return await self._call_ollama(request, model, start_time)
+        elif provider == LLMProvider.GROQ:
+            return await self._call_groq(request, model, start_time)
+        else:
+            raise ValueError(f"Unknown provider: {provider}")
+    
+    async def _call_chatgpt(
+        self, 
+        request: LLMRequest, 
+        model: str, 
+        start_time: float
+    ) -> LLMResponse:
+        """Call OpenAI ChatGPT"""
+        try:
+            import aiohttp
+            
+            api_key = self.api_keys.get("openai", "")
+            if not api_key:
+                api_key = "dummy-key"
+            
+            headers = {
+                "Authorization": f"Bearer {api_key}",
+                "Content-Type": "application/json"
+            }
+            
+            payload = {
+                "model": model,
+                "messages": [
+                    {"role": "system", "content": request.system_prompt or "You are a helpful AI assistant."},
+                    {"role": "user", "content": request.prompt}
+                ],
+                "max_tokens": request.max_tokens,
+                "temperature": request.temperature
+            }
+            
+            async with aiohttp.ClientSession() as session:
+                async with session.post(
+                    f"https://api.openai.com/v1/chat/completions",
+                    headers=headers,
+                    json=payload,
+                    timeout=aiohttp.ClientTimeout(total=request.timeout)
+                ) as resp:
+                    if resp.status == 429:
+                        raise RateLimitError("Rate limit exceeded", retry_after=60)
+                    
+                    if resp.status != 200:
+                        text = await resp.text()
+                        raise Exception(f"API error: {resp.status} - {text}")
+                    
+                    data = await resp.json()
+                    
+                    return LLMResponse(
+                        content=data["choices"][0]["message"]["content"],
+                        provider=LLMProvider.CHATGPT,
+                        model=model,
+                        tokens_used=data.get("usage", {}).get("total_tokens", 0),
+                        latency_ms=(time.time() - start_time) * 1000,
+                        success=True
+                    )
+        except aiohttp.ClientError as e:
+            raise Exception(f"Network error: {e}")
+    
+    async def _call_gemini(
+        self, 
+        request: LLMRequest, 
+        model: str, 
+        start_time: float
+    ) -> LLMResponse:
+        """Call Google Gemini"""
+        try:
+            import aiohttp
+            
+            api_key = self.api_keys.get("gemini", "")
+            if not api_key:
+                api_key = "dummy-key"
+            
+            payload = {
+                "contents": [{
+                    "parts": [{"text": request.prompt}]
+                }],
+                "generationConfig": {
+                    "maxOutputTokens": request.max_tokens,
+                    "temperature": request.temperature
+                }
+            }
+            
+            if request.system_prompt:
+                payload["systemInstruction"] = {"parts": [{"text": request.system_prompt}]}
+            
+            async with aiohttp.ClientSession() as session:
+                async with session.post(
+                    f"https://generativelanguage.googleapis.com/v1beta/models/{model}:generateContent?key={api_key}",
+                    json=payload,
+                    timeout=aiohttp.ClientTimeout(total=request.timeout)
+                ) as resp:
+                    if resp.status == 429:
+                        raise RateLimitError("Rate limit exceeded", retry_after=30)
+                    
+                    if resp.status != 200:
+                        text = await resp.text()
+                        raise Exception(f"API error: {resp.status} - {text}")
+                    
+                    data = await resp.json()
+                    
+                    content = data["candidates"][0]["content"]["parts"][0]["text"]
+                    
+                    return LLMResponse(
+                        content=content,
+                        provider=LLMProvider.GEMINI,
+                        model=model,
+                        tokens_used=0,
+                        latency_ms=(time.time() - start_time) * 1000,
+                        success=True
+                    )
+        except aiohttp.ClientError as e:
+            raise Exception(f"Network error: {e}")
+    
+    async def _call_claude(
+        self, 
+        request: LLMRequest, 
+        model: str, 
+        start_time: float
+    ) -> LLMResponse:
+        """Call Anthropic Claude"""
+        try:
+            import aiohttp
+            
+            api_key = self.api_keys.get("claude", "") or self.api_keys.get("anthropic", "")
+            if not api_key:
+                api_key = "dummy-key"
+            
+            headers = {
+                "x-api-key": api_key,
+                "Content-Type": "application/json",
+                "anthropic-version": "2023-06-01"
+            }
+            
+            payload = {
+                "model": model,
+                "max_tokens": request.max_tokens,
+                "temperature": request.temperature,
+                "messages": [
+                    {"role": "user", "content": request.prompt}
+                ]
+            }
+            
+            if request.system_prompt:
+                payload["system"] = request.system_prompt
+            
+            async with aiohttp.ClientSession() as session:
+                async with session.post(
+                    "https://api.anthropic.com/v1/messages",
+                    headers=headers,
+                    json=payload,
+                    timeout=aiohttp.ClientTimeout(total=request.timeout)
+                ) as resp:
+                    if resp.status == 429:
+                        raise RateLimitError("Rate limit exceeded", retry_after=60)
+                    
+                    if resp.status != 201:
+                        text = await resp.text()
+                        raise Exception(f"API error: {resp.status} - {text}")
+                    
+                    data = await resp.json()
+                    
+                    return LLMResponse(
+                        content=data["content"][0]["text"],
+                        provider=LLMProvider.CLAUDE,
+                        model=model,
+                        tokens_used=data.get("usage", {}).get("input_tokens", 0) + data["usage"].get("output_tokens", 0),
+                        latency_ms=(time.time() - start_time) * 1000,
+                        success=True
+                    )
+        except aiohttp.ClientError as e:
+            raise Exception(f"Network error: {e}")
+    
+    async def _call_deepseek(
+        self, 
+        request: LLMRequest, 
+        model: str, 
+        start_time: float
+    ) -> LLMResponse:
+        """Call DeepSeek"""
+        try:
+            import aiohttp
+            
+            api_key = self.api_keys.get("deepseek", "")
+            if not api_key:
+                return LLMResponse(
+                    content="",
+                    provider=LLMProvider.DEEPSEEK,
+                    model=model,
+                    tokens_used=0,
+                    latency_ms=0,
+                    success=False,
+                    error="API key not configured"
+                )
+            
+            headers = {
+                "Authorization": f"Bearer {api_key}",
+                "Content-Type": "application/json"
+            }
+            
+            payload = {
+                "model": model,
+                "messages": [
+                    {"role": "system", "content": request.system_prompt or "You are a helpful AI assistant."},
+                    {"role": "user", "content": request.prompt}
+                ],
+                "max_tokens": request.max_tokens,
+                "temperature": request.temperature
+            }
+            
+            async with aiohttp.ClientSession() as session:
+                async with session.post(
+                    "https://api.deepseek.com/chat/completions",
+                    headers=headers,
+                    json=payload,
+                    timeout=aiohttp.ClientTimeout(total=request.timeout)
+                ) as resp:
+                    if resp.status == 429:
+                        raise RateLimitError("Rate limit exceeded", retry_after=60)
+                    
+                    if resp.status != 200:
+                        text = await resp.text()
+                        raise Exception(f"API error: {resp.status} - {text}")
+                    
+                    data = await resp.json()
+                    
+                    return LLMResponse(
+                        content=data["choices"][0]["message"]["content"],
+                        provider=LLMProvider.DEEPSEEK,
+                        model=model,
+                        tokens_used=data.get("usage", {}).get("total_tokens", 0),
+                        latency_ms=(time.time() - start_time) * 1000,
+                        success=True
+                    )
+        except aiohttp.ClientError as e:
+            raise Exception(f"Network error: {e}")
+    
+    async def _call_ollama(
+        self, 
+        request: LLMRequest, 
+        model: str, 
+        start_time: float
+    ) -> LLMResponse:
+        """Call Ollama (local)"""
+        try:
+            import aiohttp
+            
+            headers = {
+                "Content-Type": "application/json"
+            }
+            
+            payload = {
+                "model": model,
+                "messages": [
+                    {"role": "system", "content": request.system_prompt or "You are a helpful AI assistant."},
+                    {"role": "user", "content": request.prompt}
+                ],
+                "options": {
+                    "temperature": request.temperature
+                },
+                "stream": False
+            }
+            
+            base_url = self.provider_configs.get(LLMProvider.OLLAMA, {}).get("base_url", "http://localhost:11434/v1")
+            
+            async with aiohttp.ClientSession() as session:
+                async with session.post(
+                    f"{base_url}/chat/completions",
+                    headers=headers,
+                    json=payload,
+                    timeout=aiohttp.ClientTimeout(total=request.timeout)
+                ) as resp:
+                    if resp.status != 200:
+                        text = await resp.text()
+                        raise Exception(f"Ollama error: {resp.status} - {text}")
+                    
+                    data = await resp.json()
+                    
+                    return LLMResponse(
+                        content=data["message"]["content"],
+                        provider=LLMProvider.OLLAMA,
+                        model=model,
+                        tokens_used=0,
+                        latency_ms=(time.time() - start_time) * 1000,
+                        success=True
+                    )
+        except aiohttp.ClientError as e:
+            raise Exception(f"Network error: {e}")
+    
+    async def _call_groq(
+        self, 
+        request: LLMRequest, 
+        model: str, 
+        start_time: float
+    ) -> LLMResponse:
+        """Call Groq (fast GPU inference)"""
+        try:
+            import aiohttp
+            
+            api_key = self.api_keys.get("groq", "")
+            if not api_key:
+                return LLMResponse(
+                    content="",
+                    provider=LLMProvider.GROQ,
+                    model=model,
+                    tokens_used=0,
+                    latency_ms=0,
+                    success=False,
+                    error="API key not configured"
+                )
+            
+            headers = {
+                "Authorization": f"Bearer {api_key}",
+                "Content-Type": "application/json"
+            }
+            
+            payload = {
+                "model": model,
+                "messages": [
+                    {"role": "system", "content": request.system_prompt or "You are a helpful AI assistant."},
+                    {"role": "user", "content": request.prompt}
+                ],
+                "max_tokens": request.max_tokens,
+                "temperature": request.temperature
+            }
+            
+            async with aiohttp.ClientSession() as session:
+                async with session.post(
+                    "https://api.groq.com/openai/v1/chat/completions",
+                    headers=headers,
+                    json=payload,
+                    timeout=aiohttp.ClientTimeout(total=request.timeout)
+                ) as resp:
+                    if resp.status == 429:
+                        raise RateLimitError("Rate limit exceeded", retry_after=30)
+                    
+                    if resp.status != 200:
+                        text = await resp.text()
+                        raise Exception(f"API error: {resp.status} - {text}")
+                    
+                    data = await resp.json()
+                    
+                    return LLMResponse(
+                        content=data["choices"][0]["message"]["content"],
+                        provider=LLMProvider.GROQ,
+                        model=model,
+                        tokens_used=data.get("usage", {}).get("total_tokens", 0),
+                        latency_ms=(time.time() - start_time) * 1000,
+                        success=True
+                    )
+        except aiohttp.ClientError as e:
+            raise Exception(f"Network error: {e}")
+    
+    def get_rate_limit_status(self) -> Dict:
+        """Get current rate limit status"""
+        return self.rate_limiter.get_status()
+    
+    def get_pending_requests(self) -> int:
+        """Get number of pending requests"""
+        return len(self.pending_requests)
+
+
+class RateLimitError(Exception):
+    """Raised when rate limited"""
+    def __init__(self, message: str, retry_after: Optional[int] = None):
+        super().__init__(message)
+        self.retry_after = retry_after
+
+
+class LLMSession:
+    """Manages an LLM conversation session"""
+    
+    def __init__(self, orchestrator: LLMOrchestrator, user_id: str):
+        self.orchestrator = orchestrator
+        self.user_id = user_id
+        self.messages: List[Dict] = []
+        self.system_prompt = "You are a helpful learning assistant."
+    
+    def add_message(self, role: str, content: str):
+        """Add a message to the conversation"""
+        self.messages.append({"role": role, "content": content})
+    
+    async def send(
+        self, 
+        message: str,
+        providers: Optional[List[LLMProvider]] = None
+    ) -> List[LLMResponse]:
+        """Send a message and get responses from all providers"""
+        self.add_message("user", message)
+        
+        request = LLMRequest(
+            prompt=self._format_conversation(),
+            system_prompt=self.system_prompt,
+            providers=providers or [LLMProvider.CHATGPT, LLMProvider.GEMINI],
+            user_id=self.user_id
+        )
+        
+        responses = await self.orchestrator.query_parallel(request)
+        
+        for response in responses:
+            if response.success:
+                self.add_message("assistant", response.content)
+        
+        return responses
+    
+    def _format_conversation(self) -> str:
+        """Format conversation history for LLM"""
+        formatted = []
+        for msg in self.messages[-10:]:
+            role = msg["role"].capitalize()
+            formatted.append(f"{role}: {msg['content']}")
+        return "\n".join(formatted)
+    
+    def clear(self):
+        """Clear conversation history"""
+        self.messages = []
+
+
+def create_orchestrator(api_keys: Optional[Dict[str, str]] = None) -> LLMOrchestrator:
+    """Create a new LLM orchestrator instance"""
+    return LLMOrchestrator(api_keys)