Spaces:
Paused
Paused
| """ | |
| Fallback LLM implementation without LlamaIndex dependency. | |
| Direct API clients for maximum compatibility. | |
| """ | |
| import logging | |
| from typing import List, Dict, Optional | |
| import json | |
| # Direct API imports (no LlamaIndex) | |
| try: | |
| import groq | |
| except ImportError: | |
| groq = None | |
| try: | |
| import anthropic | |
| except ImportError: | |
| anthropic = None | |
| try: | |
| import google.generativeai as genai | |
| except ImportError: | |
| genai = None | |
| from .config import config | |
| logger = logging.getLogger(__name__) | |
| class DirectLLMProvider: | |
| """Direct LLM provider without LlamaIndex dependency""" | |
| def __init__(self): | |
| self.providers_available = { | |
| 'groq': groq is not None and config.groq_api_key, | |
| 'anthropic': anthropic is not None and config.anthropic_api_key, | |
| 'gemini': genai is not None and config.google_api_key | |
| } | |
| async def chat(self, messages: List[Dict[str, str]], temperature: float = 0.1) -> str: | |
| """Chat completion with fallback chain: Groq -> Anthropic -> Gemini -> Mock""" | |
| # Try Groq first | |
| if self.providers_available['groq']: | |
| try: | |
| client = groq.Groq(api_key=config.groq_api_key) | |
| response = client.chat.completions.create( | |
| model="llama-3.1-8b-instant", | |
| messages=messages, | |
| temperature=temperature, | |
| max_tokens=1000 | |
| ) | |
| return response.choices[0].message.content | |
| except Exception as e: | |
| logger.warning(f"Groq failed: {e}") | |
| # Fallback to Anthropic | |
| if self.providers_available['anthropic']: | |
| try: | |
| client = anthropic.Anthropic(api_key=config.anthropic_api_key) | |
| # Separate system message | |
| system_msg = "" | |
| user_messages = [] | |
| for msg in messages: | |
| if msg["role"] == "system": | |
| system_msg = msg["content"] | |
| else: | |
| user_messages.append(msg) | |
| response = client.messages.create( | |
| model="claude-3-sonnet-20240229", | |
| max_tokens=1000, | |
| temperature=temperature, | |
| system=system_msg, | |
| messages=user_messages | |
| ) | |
| return response.content[0].text | |
| except Exception as e: | |
| logger.warning(f"Anthropic failed: {e}") | |
| # Fallback to Gemini | |
| if self.providers_available['gemini']: | |
| try: | |
| genai.configure(api_key=config.google_api_key) | |
| model = genai.GenerativeModel('gemini-pro') | |
| # Convert messages to Gemini format | |
| prompt = "" | |
| for msg in messages: | |
| if msg["role"] == "system": | |
| prompt += f"System: {msg['content']}\n\n" | |
| elif msg["role"] == "user": | |
| prompt += f"User: {msg['content']}\n" | |
| elif msg["role"] == "assistant": | |
| prompt += f"Assistant: {msg['content']}\n" | |
| response = model.generate_content(prompt) | |
| return response.text | |
| except Exception as e: | |
| logger.warning(f"Gemini failed: {e}") | |
| # Final fallback to mock | |
| return self._mock_response(messages) | |
| def _mock_response(self, messages: List[Dict[str, str]]) -> str: | |
| """Mock response for development/fallback""" | |
| last_msg = messages[-1]["content"].lower() if messages else "hello" | |
| if any(word in last_msg for word in ["book", "schedule", "appointment"]): | |
| return "I'd be happy to help you book an appointment! Please provide your name, preferred date and time." | |
| elif any(word in last_msg for word in ["cancel", "delete"]): | |
| return "I can help you cancel an appointment. Which meeting would you like to cancel?" | |
| elif any(word in last_msg for word in ["available", "availability"]): | |
| return "Let me check Peter's availability. What dates are you considering?" | |
| else: | |
| return "Hello! I'm ChatCal, your voice-enabled scheduling assistant. How can I help you today?" | |
| # Global instance | |
| direct_llm = DirectLLMProvider() |