""" backend/agents/grid_rag.py — Grid-Enhanced RAG (S766-GRID-4) Sistema RAG (Retrieval-Augmented Generation) avanzato che indicizza: - Memoria distribuita (Supabase A, B, C, D) - Log di sistema e di Railway - Documentazione interna (.agents/memory/) Architettura: - GridIndexer: Indicizza i dati provenienti da diverse fonti - ContextRetriever: Recupera il contesto più rilevante per il goal corrente - KnowledgeGraph: Mappa le relazioni tra i diversi profili e i loro stati """ import os import asyncio import logging from typing import List, Dict, Any, Optional from datetime import datetime import json _logger = logging.getLogger("grid_rag") # ── Configurazione ───────────────────────────────────────────────────────── RAG_INDEX_SIZE = 100 # Numero di elementi da mantenere nel buffer RAG RAG_SIMILARITY_THRESHOLD = 0.75 class GridIndexer: """Indicizzatore per la Grid.""" def __init__(self): self.index = [] self._lock = asyncio.Lock() async def add_to_index(self, source: str, content: str, metadata: Dict): """Aggiunge un elemento all'indice RAG.""" async with self._lock: entry = { "source": source, "content": content, "metadata": metadata, "timestamp": datetime.now().isoformat(), } self.index.append(entry) # Mantieni dimensione fissa if len(self.index) > RAG_INDEX_SIZE: self.index.pop(0) async def index_railway_logs(self, profile: str, logs: str): """Indicizza i log di Railway per identificare crash passati.""" lines = logs.split("\n") for line in lines[-50:]: # Ultime 50 righe if "error" in line.lower() or "crash" in line.lower() or "failed" in line.lower(): await self.add_to_index( source=f"railway_logs_{profile}", content=line, metadata={"type": "log_error", "profile": profile} ) class ContextRetriever: """Recuperatore di contesto per l'agente.""" def __init__(self, indexer: GridIndexer): self.indexer = indexer async def retrieve_relevant_context(self, query: str) -> List[Dict]: """ Recupera il contesto rilevante basato sulla query. Attualmente usa keyword matching semplice (potenziabile con embeddings). """ relevant = [] keywords = query.lower().split() async with self.indexer._lock: for entry in self.indexer.index: content = entry["content"].lower() score = sum(1 for kw in keywords if kw in content) if score > 0: entry_with_score = entry.copy() entry_with_score["score"] = score relevant.append(entry_with_score) # Ordina per score decrescente relevant.sort(key=lambda x: x["score"], reverse=True) return relevant[:10] # Ritorna i top 10 class GridRAG: """Interfaccia principale per il RAG della Grid.""" def __init__(self): self.indexer = GridIndexer() self.retriever = ContextRetriever(self.indexer) async def prepare_agent_context(self, goal: str) -> str: """ Prepara il contesto per l'agente unificando i dati RAG. """ context_items = await self.retriever.retrieve_relevant_context(goal) if not context_items: return "" context_str = "\n--- GRID RAG CONTEXT ---\n" for item in context_items: context_str += f"[{item['source']}] {item['content']}\n" context_str += "------------------------\n" return context_str # ── Singleton globale ────────────────────────────────────────────────────── _grid_rag_instance: Optional[GridRAG] = None def get_grid_rag() -> GridRAG: """Restituisce l'istanza globale del GridRAG.""" global _grid_rag_instance if _grid_rag_instance is None: _grid_rag_instance = GridRAG() return _grid_rag_instance