File size: 3,799 Bytes
0a743f9 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 |
"""
Chat engine for FleetMind
Main orchestrator for AI-powered conversations with multi-provider support
"""
import os
import logging
from typing import Tuple, List, Dict
from chat.providers import ClaudeProvider, GeminiProvider
logger = logging.getLogger(__name__)
class ChatEngine:
"""Main orchestrator for AI chat conversations with multi-provider support"""
def __init__(self):
# Get provider selection from environment
provider_name = os.getenv("AI_PROVIDER", "anthropic").lower()
logger.info(f"ChatEngine: Selected provider: {provider_name}")
# Initialize the selected provider
if provider_name == "gemini":
self.provider = GeminiProvider()
logger.info("ChatEngine: Using Gemini provider")
elif provider_name == "anthropic":
self.provider = ClaudeProvider()
logger.info("ChatEngine: Using Claude provider")
else:
# Default to Claude if unknown provider
logger.warning(f"ChatEngine: Unknown provider '{provider_name}', defaulting to Claude")
self.provider = ClaudeProvider()
# Store provider name for UI
self.selected_provider = provider_name
def is_available(self) -> bool:
"""Check if the chat engine is available"""
return self.provider.is_available()
def get_status(self) -> str:
"""Get status message for UI"""
provider_status = self.provider.get_status()
provider_name = self.provider.get_provider_name()
return f"**{provider_name}:** {provider_status}"
def get_provider_name(self) -> str:
"""Get the active provider name"""
return self.provider.get_provider_name()
def get_model_name(self) -> str:
"""Get the active model name"""
return self.provider.get_model_name()
def process_message(
self,
user_message: str,
conversation
) -> Tuple[str, List[Dict]]:
"""
Process user message and return AI response
Args:
user_message: User's message
conversation: ConversationManager instance
Returns:
Tuple of (assistant_response, tool_calls_made)
"""
return self.provider.process_message(user_message, conversation)
def get_welcome_message(self) -> str:
"""Get welcome message for new conversations"""
return self.provider.get_welcome_message()
def get_full_status(self) -> Dict[str, str]:
"""
Get detailed status for all providers
Returns:
Dict with status for each provider
"""
# Get status without creating new instances (avoid API calls)
claude_key = os.getenv("ANTHROPIC_API_KEY", "")
gemini_key = os.getenv("GOOGLE_API_KEY", "")
claude_available = bool(claude_key and not claude_key.startswith("your_"))
gemini_available = bool(gemini_key and not gemini_key.startswith("your_"))
claude_status = "✅ Connected - Model: claude-3-5-sonnet-20241022" if claude_available else "⚠️ Not configured (add ANTHROPIC_API_KEY)"
gemini_status = f"✅ Connected - Model: {self.provider.get_model_name()}" if (self.selected_provider == "gemini" and gemini_available) else "⚠️ Not configured (add GOOGLE_API_KEY)" if not gemini_available else "✅ Configured"
return {
"selected": self.selected_provider,
"claude": {
"name": "Claude (Anthropic)",
"status": claude_status,
"available": claude_available
},
"gemini": {
"name": "Gemini (Google)",
"status": gemini_status,
"available": gemini_available
}
}
|