""" Groq API Integration ------------------ Handles interactions with the Groq API for language model inference. """ import os import logging from typing import Dict, Any, Optional import requests logger = logging.getLogger(__name__) class GroqAPI: """Handles interactions with the Groq API.""" def __init__(self, api_key: Optional[str] = None): """Initialize the Groq API client.""" self.api_key = api_key or os.getenv('GROQ_API_KEY') self.base_url = "https://api.groq.com/v1" self.default_model = "mixtral-8x7b-32768" async def predict(self, prompt: str, model: Optional[str] = None) -> Dict[str, Any]: """Make a prediction using the Groq API.""" try: if not self.api_key: return { "success": False, "error": "GROQ_API_KEY not set in environment variables", "answer": "Error: GROQ_API_KEY not configured" } headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json" } data = { "model": model or self.default_model, "messages": [{"role": "user", "content": prompt}], "temperature": 0.7, "max_tokens": 1000 } response = requests.post( f"{self.base_url}/chat/completions", headers=headers, json=data ) if response.status_code == 200: result = response.json() return { "success": True, "answer": result["choices"][0]["message"]["content"] } else: return { "success": False, "error": f"API Error: {response.status_code}", "answer": f"Error: {response.text}" } except Exception as e: logger.error(f"Error calling Groq API: {str(e)}") return { "success": False, "error": str(e), "answer": f"Error: {str(e)}" }