import os import aiohttp from typing import Optional, List, Dict, Any from utils.logger import discord_logger as log def _ensure_env_loaded(): """Asegura que las variables de entorno estén cargadas.""" from dotenv import load_dotenv load_dotenv() def _get_groq_config(): _ensure_env_loaded() try: from botdiscord.config import get_groq_config return get_groq_config() except Exception: return {} def _load_config(): _ensure_env_loaded() # Primero tomar de variables de entorno (Docker), luego de config.yaml como backup env_key = os.getenv("GROQ_API_KEY", "") env_model = os.getenv("GROQ_MODEL", "llama-3.3-70b-versatile") env_rag_url = os.getenv("RAG_API_URL", "http://localhost:8004") cfg = _get_groq_config() return { "api_key": env_key or cfg.get("api_key", ""), "model": env_model or cfg.get("model", "llama-3.3-70b-versatile"), "rag_url": env_rag_url or cfg.get("rag_url", "http://localhost:8004") } def _get_system_prompt(): try: from botdiscord.database import get_bot_config prompt = get_bot_config("groq_system_prompt") if prompt: return prompt except Exception: pass try: prompt_file = os.path.join(os.path.dirname(__file__), "..", "prompt_general.md") if os.path.exists(prompt_file): with open(prompt_file, "r", encoding="utf-8") as f: return f.read() except Exception: pass return """Eres el General Reserves, comandante del ejército de Last War: Survival Game. IDIOMA - IMPORTANTE: 1. Detecta el idioma de la PREGUNTA del usuario 2. Si NO es inglés, tradúcela al inglés ANTES de consultar el RAG 3. Cuando recibas la respuesta del RAG, tradúcela al MISMO IDIOMA de la pregunta original 4. RESPONDE SIEMPRE en el mismo idioma que te habló el usuario SALUDOS: Saluda como "¡A la orden, recruit! 🎖️" o "¡Reporting for duty!" FORMATO DE RESPUESTA: - Primero saluda al usuario - Da la información encontrada - NUNCA repitas información varias veces - Sé conciso RESTRICCIONES: 1. SOLO responde sobre Last War: Survival Game 2. NUNCA inventes información 3. Si no hay datos en el RAG, responde con humor gentil: "¡Mi radar no detectó eso, recruit! 🤔" Usa el sistema RAG para buscar información.""" _config_cache = None def get_config(): global _config_cache if _config_cache is None: _config_cache = _load_config() log.info(f"Groq Config loaded - API Key: {'set' if _config_cache.get('api_key') else 'NOT SET'}, Model: {_config_cache.get('model')}, RAG URL: {_config_cache.get('rag_url')}") return _config_cache def reload_config(): global _config_cache _config_cache = _load_config() async def query_rag(question: str, top_k: int = 3) -> Dict[str, Any]: """Consulta la API RAG y retorna la respuesta.""" config = get_config() rag_url = config.get("rag_url", "http://localhost:8004") log.info(f"Querying RAG at {rag_url} with question: {question[:50]}...") try: async with aiohttp.ClientSession() as session: async with session.post( f"{rag_url}/query", json={"question": question, "top_k": top_k}, timeout=30 ) as resp: if resp.status == 200: return await resp.json() else: log.error(f"RAG API error: {resp.status}") return {"answer": "Error querying knowledge base", "sources": [], "cached": False} except Exception as e: log.error(f"RAG query failed: {e}") return {"answer": "Error connecting to knowledge base", "sources": [], "cached": False} def build_messages(question: str, rag_context: Optional[Dict[str, Any]] = None) -> List[Dict[str, str]]: """Construye los mensajes para la API de Groq.""" system_prompt = _get_system_prompt() messages = [{"role": "system", "content": system_prompt}] if rag_context: context_text = f"Context from knowledge base:\n{rag_context.get('answer', '')}\n\nSources: {rag_context.get('sources', [])}" messages.append({"role": "system", "content": context_text}) messages.append({"role": "user", "content": question}) return messages async def chat_with_rag(question: str, use_rag: bool = True) -> Dict[str, Any]: """ Procesa una pregunta usando Groq con RAG opcional. Args: question: La pregunta del usuario use_rag: Si True, consulta la base de conocimientos primero Returns: Dict con 'response' (respuesta final), 'rag_result' (resultado RAG), 'sources' (fuentes) """ config = get_config() api_key = config.get("api_key") model = config.get("model", "llama-3.3-70b-versatile") if not api_key: return { "response": "Error: GROQ_API_KEY not configured", "rag_result": None, "sources": [] } rag_result = None if use_rag: rag_result = await query_rag(question, top_k=2) log.info(f"RAG result length: {len(str(rag_result.get('answer', '')))} chars") if not rag_result or not rag_result.get('answer'): # No hay contexto del RAG, responder directamente return { "response": "¡Mi radar no detectó información relevante en la base de datos, recruit! 🤔 No encontré información sobre eso.", "rag_result": rag_result, "sources": [] } messages = build_messages(question, rag_result) payload = { "model": model, "messages": messages, "temperature": 0.7, "max_tokens": 1024 } headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" } try: async with aiohttp.ClientSession() as session: async with session.post( "https://api.groq.com/openai/v1/chat/completions", json=payload, headers=headers, timeout=30 ) as resp: if resp.status == 200: data = await resp.json() choice = data.get("choices", [{}])[0] response = choice.get("message", {}).get("content", "") sources = [] if rag_result and rag_result.get("sources"): sources = rag_result["sources"] return { "response": response, "rag_result": rag_result, "sources": sources } else: error = await resp.text() log.error(f"Groq API error: {resp.status} - {error}") return { "response": "Error processing request", "rag_result": rag_result, "sources": [] } except Exception as e: log.error(f"Groq request failed: {e}") return { "response": "Error connecting to AI service", "rag_result": rag_result, "sources": [] }