import os import re import httpx import asyncio from typing import List, Optional, Tuple from contextlib import asynccontextmanager from fastapi import FastAPI, HTTPException, Request, Response from pydantic import BaseModel, Field # --- IMPORTS DE LANGCHAIN --- from langchain_community.document_loaders import WebBaseLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.vectorstores import FAISS from langchain_huggingface import HuggingFaceEmbeddings from langchain_core.runnables import RunnablePassthrough from langchain_core.prompts import PromptTemplate from langchain_groq import ChatGroq # ------------------------------------------------------------------------------ # 1. CONFIGURACIÓN DE ENTORNO, TOKENS Y CACHÉ # ------------------------------------------------------------------------------ TEMP_CACHE_DIR = '/tmp/huggingface_cache' os.environ['TRANSFORMERS_CACHE'] = TEMP_CACHE_DIR os.environ['HF_HOME'] = TEMP_CACHE_DIR os.environ['SENTENCE_TRANSFORMERS_HOME'] = TEMP_CACHE_DIR os.makedirs(TEMP_CACHE_DIR, exist_ok=True) # Variables de entorno obtenidas desde Secrets VERIFY_TOKEN = os.getenv("VERIFY_TOKEN") WHATSAPP_TOKEN = os.getenv("WHATSAPP_TOKEN") WHATSAPP_PHONE_ID = os.getenv("WHATSAPP_PHONE_ID") # ------------------------------------------------------------------------------ # 2. MODELOS DE DATOS (DTOs) # ------------------------------------------------------------------------------ class QueryRequest(BaseModel): query: str = Field(..., min_length=2, max_length=500, description="Consulta del usuario sobre Godot Engine") phone_number: Optional[str] = Field(None, max_length=20, pattern=r"^\+?[0-9\s\-]+$") class QueryResponse(BaseModel): query: str response: str intent: str sources: List[str] # ------------------------------------------------------------------------------ # 3. NOTIFICADOR EXTERNO (Meta / WhatsApp Cloud API) # ------------------------------------------------------------------------------ class WhatsAppNotifier: """Envía mensajes directamente usando la API Oficial de Meta WhatsApp Cloud.""" def __init__(self): # trust_env=False evita interferencias del entorno del contenedor HF self.client = httpx.AsyncClient( timeout=httpx.Timeout(30.0, connect=15.0), trust_env=False ) async def send_message(self, phone_number: str, message_text: str) -> bool: token = os.getenv("WHATSAPP_TOKEN") phone_id = os.getenv("WHATSAPP_PHONE_ID") api_url = f"https://graph.facebook.com/v19.0/{phone_id}/messages" if phone_id else None if not phone_number or not token or not api_url: print(f"[WhatsAppNotifier Error] Faltan credenciales: phone_number={bool(phone_number)}, token={bool(token)}, api_url={bool(api_url)}") return False headers = { "Authorization": f"Bearer {token.strip()}", "Content-Type": "application/json" } payload = { "messaging_product": "whatsapp", "recipient_type": "individual", "to": phone_number, "type": "text", "text": {"preview_url": False, "body": message_text} } try: response = await self.client.post(api_url, json=payload, headers=headers) if response.status_code != 200: print(f"[WhatsAppNotifier Error] Meta API respondió Status {response.status_code}: {response.text}") return False print(f"[WhatsAppNotifier Success] Mensaje enviado correctamente a {phone_number}") return True except Exception as err: print(f"[WhatsAppNotifier Error] Excepción al realizar POST a Meta: {type(err).__name__} - {err}") return False async def close(self): await self.client.aclose() # ------------------------------------------------------------------------------ # 4. RAG PIPELINE ASÍNCRONO # ------------------------------------------------------------------------------ class GodotRAGPipeline: def __init__(self, doc_url: str): self.doc_url = doc_url self.embeddings = None self.vectorstore = None self.retriever = None self.llm = None self.intent_chain = None self.saludo_chain = None self.rag_chain = None self._init_prompts() def _init_prompts(self): self.intent_prompt = PromptTemplate( template="""Eres un sistema clasificador estricto de intenciones para un asistente de Godot Engine. Tu ÚNICA tarea es devolver una de estas palabras clave: SALUDO, GODOT o OTRO. Categorías: - SALUDO: saludos, despedidas, conversación casual. - GODOT: preguntas técnicas sobre código, nodos, funciones, errores o GDScript. - OTRO: cualquier otra pregunta ajena a Godot. Entrada del usuario a clasificar: \"\"\"{query}\"\"\" Responde ÚNICAMENTE con la palabra de la categoría (SALUDO, GODOT o OTRO):""", input_variables=["query"] ) self.saludo_prompt = PromptTemplate( template="""Hola, soy tu asistente experto en Godot Engine. ¿En qué script o problema técnico te puedo ayudar hoy? 🤖 Mensaje del usuario: \"\"\"{query}\"\"\" Respuesta:""", input_variables=["query"] ) self.rag_prompt = PromptTemplate( template="""Eres un desarrollador experto en Godot Engine. Responde la pregunta del usuario utilizando la información del contexto extraído de la documentación oficial. Contexto oficial: {context} Pregunta del usuario: \"\"\"{question}\"\"\" Respuesta técnica:""", input_variables=["context", "question"] ) def _sanitize_input(self, text: str) -> str: text = text.replace('"""', '""') text = re.sub(r'[\x00-\x08\x0B\x0C\x0E-\x1F\x7F]', '', text) return text.strip() def initialize(self): try: print(f"[GodotRAGPipeline] Cargando contenido desde: {self.doc_url}") loader = WebBaseLoader(self.doc_url) docs = loader.load() text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200) splits = text_splitter.split_documents(docs) print("[GodotRAGPipeline] Cargando modelo de Embeddings...") self.embeddings = HuggingFaceEmbeddings( model_name="sentence-transformers/all-MiniLM-L6-v2", model_kwargs={'device': 'cpu'}, cache_folder=TEMP_CACHE_DIR ) print("[GodotRAGPipeline] Construyendo índice vectorial FAISS...") self.vectorstore = FAISS.from_documents(splits, self.embeddings) self.retriever = self.vectorstore.as_retriever(search_kwargs={"k": 3}) print("[GodotRAGPipeline] Inicializando modelo LLM (Groq)...") self.llm = ChatGroq(temperature=0.2, model_name="qwen/qwen3.6-27b", max_tokens=1024) self.intent_chain = self.intent_prompt | self.llm self.saludo_chain = self.saludo_prompt | self.llm print("[GodotRAGPipeline] Pipeline inicializado de forma segura.") except Exception as e: print(f"[GodotRAGPipeline CRITICAL] Error al inicializar: {e}") raise RuntimeError(f"Falla crítica en RAG Pipeline: {e}") async def process_query(self, query_text: str) -> Tuple[str, str, List[str]]: clean_query = self._sanitize_input(query_text) intent_res = await self.intent_chain.ainvoke({"query": clean_query}) intent = intent_res.content.strip().upper() sources = [] if "SALUDO" in intent: response = await self.saludo_chain.ainvoke({"query": clean_query}) return response.content, "SALUDO", sources elif "OTRO" in intent: response = "Disculpa, solo puedo ayudarte con temas de programación y desarrollo en Godot Engine. 🕹️" return response, "OTRO", sources else: docs = await self.retriever.ainvoke(clean_query) context_text = "\n\n".join([doc.page_content for doc in docs]) response = await (self.rag_prompt | self.llm).ainvoke({ "context": context_text, "question": clean_query }) sources = list(set([doc.metadata.get("source", self.doc_url) for doc in docs])) return response.content, "GODOT_INFO", sources # ------------------------------------------------------------------------------ # 5. MANEJO DE TAREAS Y CONTROLADOR API (FastAPI) # ------------------------------------------------------------------------------ DOC_URL = "https://docs.godotengine.org/es/stable/tutorials/scripting/gdscript/gdscript_basics.html" rag_pipeline = GodotRAGPipeline(doc_url=DOC_URL) notifier = WhatsAppNotifier() # Estructura para almacenar referencias fuertes de tareas asíncronas background_tasks_set = set() async def process_and_send_whatsapp(sender_phone: str, user_text: str): """Procesa la respuesta del RAG en el Event Loop en segundo plano.""" try: print(f"[AsyncTask] Procesando consulta RAG para {sender_phone}: '{user_text}'") response_text, intent, _ = await rag_pipeline.process_query(user_text) print(f"[AsyncTask] Respuesta generada ({intent}). Enviando POST a Meta...") await notifier.send_message(sender_phone, response_text) except Exception as e: print(f"[AsyncTask Error] Error durante la ejecución asíncrona: {type(e).__name__} - {e}") @asynccontextmanager async def lifespan(app: FastAPI): try: rag_pipeline.initialize() except Exception as e: print(f"Error al iniciar el servicio: {e}") yield await notifier.close() app = FastAPI(title="Godot Link RAG API", lifespan=lifespan) @app.get("/") def health_check(): return {"status": "active", "message": "Servicio RAG de Godot operativo y protegido."} # --- WEBHOOK PARA VALIDACIÓN Y RECEPCIÓN DE WHATSAPP --- @app.get("/webhook") async def verify_webhook(request: Request): params = request.query_params mode = params.get("hub.mode") token = params.get("hub.verify_token") challenge = params.get("hub.challenge") if mode == "subscribe" and token == VERIFY_TOKEN: print("[Webhook] Verificación exitosa de Meta.") return Response(content=challenge, status_code=200) return Response(content="Token de verificación inválido", status_code=403) @app.post("/webhook") async def receive_whatsapp_message(request: Request): data = await request.json() try: entries = data.get("entry", []) for entry in entries: for change in entry.get("changes", []): value = change.get("value", {}) messages = value.get("messages", []) if messages: message = messages[0] sender_phone = message.get("from") if message.get("type") == "text": user_text = message["text"]["body"] print(f"[Webhook] Mensaje de {sender_phone}: {user_text}") # Crear tarea y guardar referencia fuerte para evitar GC (Garbage Collection) task = asyncio.create_task(process_and_send_whatsapp(sender_phone, user_text)) background_tasks_set.add(task) task.add_done_callback(background_tasks_set.discard) except Exception as e: print(f"[Webhook Error] Fallo al procesar evento entrante: {e}") # Retorna HTTP 200 inmediatamente a Meta para responder antes de los 3 segundos return {"status": "success"} @app.post("/query", response_model=QueryResponse) async def handle_query(request: QueryRequest): try: response_text, intent, sources = await rag_pipeline.process_query(request.query) if request.phone_number: await notifier.send_message(request.phone_number, response_text) return QueryResponse( query=request.query, response=response_text, intent=intent, sources=sources ) except Exception as e: raise HTTPException(status_code=500, detail=f"Error interno procesando la consulta: {str(e)}")