| """ |
| Fallback LLM implementation without LlamaIndex dependency. |
| Direct API clients for maximum compatibility. |
| """ |
|
|
| import logging |
| from typing import List, Dict, Optional |
| import json |
|
|
| |
| 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""" |
| |
| |
| 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}") |
| |
| |
| if self.providers_available['anthropic']: |
| try: |
| client = anthropic.Anthropic(api_key=config.anthropic_api_key) |
| |
| |
| 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}") |
| |
| |
| if self.providers_available['gemini']: |
| try: |
| genai.configure(api_key=config.google_api_key) |
| model = genai.GenerativeModel('gemini-pro') |
| |
| |
| 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}") |
| |
| |
| 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?" |
|
|
|
|
| |
| direct_llm = DirectLLMProvider() |