""" Main Chat Service with AI Provider Integration Handles conversations with automatic fallback and error recovery """ from typing import List, Optional, Dict, Any from datetime import datetime import logging from .providers import AIProviderManager, AIMessage, AIProvider from app.config import settings logger = logging.getLogger(__name__) class ChatService: """ Main chat service with AI integration Features: - Multiple AI provider support (OpenAI, Anthropic, Ollama) - Automatic fallback on API failures - Conversation history management - Context-aware responses """ def __init__(self): self.ai_manager = AIProviderManager( openai_key=settings.OPENAI_API_KEY if settings.OPENAI_API_KEY else None, anthropic_key=settings.ANTHROPIC_API_KEY if settings.ANTHROPIC_API_KEY else None, ollama_url=settings.OLLAMA_URL if hasattr(settings, 'OLLAMA_URL') else "http://ollama:11434" ) self.initialized = False async def initialize(self): """Initialize AI providers""" if not self.initialized: await self.ai_manager.initialize() self.initialized = True logger.info("ChatService initialized with AI providers") async def chat( self, user_message: str, conversation_history: List[Dict[str, str]] = None, system_prompt: Optional[str] = None, context: Optional[Dict[str, Any]] = None, preferred_provider: Optional[str] = None, temperature: float = 0.7, max_tokens: int = 1000 ) -> Dict[str, Any]: """ Send a chat message and get AI response Args: user_message: The user's message conversation_history: Previous messages [{"role": "user/assistant", "content": "..."}] system_prompt: System instructions for the AI context: Additional context (user data, business info, etc.) preferred_provider: Preferred AI provider (openai/anthropic/ollama) temperature: Response randomness (0-1) max_tokens: Maximum response length Returns: { "response": "AI response text", "provider": "provider_used", "tokens_used": 123, "latency_ms": 456 } """ start_time = datetime.now() # Ensure initialized if not self.initialized: await self.initialize() # Build message list messages = [] # Add system prompt if system_prompt: messages.append(AIMessage(role="system", content=system_prompt)) elif context: # Create dynamic system prompt from context system_msg = self._build_system_prompt(context) messages.append(AIMessage(role="system", content=system_msg)) # Add conversation history if conversation_history: for msg in conversation_history[-10:]: # Last 10 messages for context messages.append(AIMessage(role=msg["role"], content=msg["content"])) # Add current user message messages.append(AIMessage(role="user", content=user_message)) # Convert preferred provider string to enum provider_enum = None if preferred_provider: try: provider_enum = AIProvider(preferred_provider) except ValueError: logger.warning(f"Invalid provider: {preferred_provider}") try: # Get AI response with automatic fallback response = await self.ai_manager.chat( messages=messages, temperature=temperature, max_tokens=max_tokens, preferred_provider=provider_enum ) # Calculate metrics latency = (datetime.now() - start_time).total_seconds() * 1000 return { "response": response, "provider": self._get_used_provider(), "tokens_used": self._estimate_tokens(messages, response), "latency_ms": int(latency), "success": True } except Exception as e: logger.error(f"Chat service error: {e}") latency = (datetime.now() - start_time).total_seconds() * 1000 return { "response": self._get_fallback_response(), "provider": "fallback", "tokens_used": 0, "latency_ms": int(latency), "success": False, "error": str(e) } def _build_system_prompt(self, context: Dict[str, Any]) -> str: """Build system prompt from context""" parts = [] # Base instruction parts.append("You are a helpful AI assistant.") # Add business context if "business_name" in context: parts.append(f"You work for {context['business_name']}.") if "business_type" in context: parts.append(f"This is a {context['business_type']} business.") # Add capabilities if "capabilities" in context: caps = ", ".join(context["capabilities"]) parts.append(f"You can help with: {caps}.") # Add tone/style if "tone" in context: parts.append(f"Maintain a {context['tone']} tone.") # Add constraints if "constraints" in context: for constraint in context["constraints"]: parts.append(constraint) return " ".join(parts) def _get_used_provider(self) -> str: """Get the provider that was actually used""" if self.ai_manager.available_providers: return self.ai_manager.available_providers[0].value return "unknown" def _estimate_tokens(self, messages: List[AIMessage], response: str) -> int: """Rough token estimation (4 chars ≈ 1 token)""" total_chars = sum(len(msg.content) for msg in messages) + len(response) return int(total_chars / 4) def _get_fallback_response(self) -> str: """Return a fallback response when all AI providers fail""" return ( "I apologize, but I'm having trouble connecting to my AI services at the moment. " "Please try again in a few moments, or contact support if the issue persists." ) async def get_provider_status(self) -> Dict[str, Any]: """Get status of all AI providers""" if not self.initialized: await self.initialize() return { **self.ai_manager.get_status(), "recommendations": self._get_recommendations() } def _get_recommendations(self) -> List[str]: """Get recommendations based on available providers""" recommendations = [] if not self.ai_manager.available_providers: recommendations.append( "⚠️ No AI providers available! Add OpenAI/Anthropic API key or ensure Ollama is running." ) elif AIProvider.OLLAMA in self.ai_manager.available_providers and len(self.ai_manager.available_providers) == 1: recommendations.append( "ℹ️ Using local Ollama only. Add OpenAI/Anthropic API key for better performance." ) elif AIProvider.OPENAI in self.ai_manager.available_providers or AIProvider.ANTHROPIC in self.ai_manager.available_providers: recommendations.append( "✓ Cloud AI providers configured. Ollama available as fallback." ) return recommendations # Singleton instance _chat_service: Optional[ChatService] = None async def get_chat_service() -> ChatService: """Get or create ChatService instance""" global _chat_service if _chat_service is None: _chat_service = ChatService() await _chat_service.initialize() return _chat_service