¿Cómo construir un sistema RAG de árbol de decisiones agente con enrutamiento inteligente de consultas, autoverificación y refinamiento iterativo?

En este tutorial, creamos un sistema avanzado de recuperación de agentes y generación aumentada (RAG) que va más allá de la simple respuesta a preguntas. Lo diseñamos para dirigir consultas de forma inteligente a las fuentes de conocimiento adecuadas, realizar autoverificaciones para evaluar la calidad de las respuestas y refinar las respuestas de forma iterativa para mejorar la precisión. Implementamos todo el sistema utilizando herramientas de código abierto como FAISS, SentenceTransformers y Flan-T5. A medida que avanzamos, exploramos cómo el enrutamiento, la recuperación, la generación y la autoevaluación se combinan para formar un proceso RAG estilo árbol de decisiones que imita el razonamiento agente del mundo real. Consulta los CÓDIGOS COMPLETOS aquí.

print(“🔧 Configurando dependencias…”) import subproceso import sys def install_packages(): paquetes = [‘sentence-transformers’, ‘transformers’, ‘torch’, ‘faiss-cpu’, ‘numpy’, ‘accelerate’]
para paquete en paquetes: print(f”Instalando {paquete}…”) subprocess.check_call([sys.executable, ‘-m’, ‘pip’, ‘install’, ‘-q’, package]) intente: importar faiss excepto ImportError: install_packages() print(“¡¡Todas las dependencias instaladas! Importando módulos…\n”) importar antorcha importar numpy como np de sentencia_transformers importar SentenceTransformer de transformadores importar tubería importar faiss de escribir importar Lista, Dict, Tupla importar advertencias advertencias.filterwarnings(‘ignorar’) imprimir(” ✓ Todos los módulos cargados exitosamente!\n”)

Comenzamos instalando todas las dependencias necesarias, incluidos Transformers, FAISS y SentenceTransformers, para garantizar una ejecución local sin problemas. Verificamos instalaciones e instalamos módulos esenciales como NumPy, PyTorch y FAISS para incrustación, recuperación y generación. Confirmamos que todas las bibliotecas se cargan correctamente antes de continuar con el proceso principal. Consulta los CÓDIGOS COMPLETOS aquí.

clase VectorStore: def __init__(self, embedding_model=”all-MiniLM-L6-v2″): print(f”Cargando modelo de incrustación: {embedding_model}…”) self.embedder = SentenceTransformer(embedding_model) self.documents = []
self.index = Ninguno def add_documents(self, docs: Lista[str]fuentes: Lista[str]): auto.documentos = [{“text”: doc, “source”: src} for doc, src in zip(docs, sources)]
incrustaciones = self.embedder.encode(docs, show_progress_bar=False) dimensión = incrustaciones.shape[1]
self.index = faiss.IndexFlatL2(dimensión) self.index.add(embeddings.astype(‘float32’)) print(f” ✓ Indexed {len(docs)} documentos\n”) def search(self, query: str, k: int = 3) -> Lista[Dict]: query_vec = self.embedder.encode([query]).astype(‘float32’) distancias, índices = self.index.search(query_vec, k) retorno [self.documents[i] para i en índices[0]]

Diseñamos la clase VectorStore para almacenar y recuperar documentos de manera eficiente utilizando la búsqueda por similitud basada en FAISS. Incorporamos cada documento utilizando un modelo transformador y creamos un índice para una recuperación rápida. Esto nos permite buscar rápidamente el contexto más relevante para cualquier consulta entrante. Consulta los CÓDIGOS COMPLETOS aquí.

clase QueryRouter: def __init__(self): self.categories = { ‘técnico’: [‘how’, ‘implement’, ‘code’, ‘function’, ‘algorithm’, ‘debug’]’fáctico’: [‘what’, ‘who’, ‘when’, ‘where’, ‘define’, ‘explain’]’comparativo’: [‘compare’, ‘difference’, ‘versus’, ‘vs’, ‘better’, ‘which’]’procesal’: [‘steps’, ‘process’, ‘guide’, ‘tutorial’, ‘how to’]
} def ruta(self, consulta: str) -> str: query_lower = query.lower() puntuaciones = {} para categoría, palabras clave en self.categories.items(): puntuación = suma(1 para kw en palabras clave si kw en query_lower) puntuacionesAgentic AI = puntuación mejor_categoría = max(puntuaciones, clave=puntuaciones.get) devuelve mejor_categoría si puntuaciones[best_category] > 0 más ‘fáctico’

Introducimos la clase QueryRouter para clasificar consultas por intención, técnicas, fácticas, comparativas o de procedimiento. Utilizamos la concordancia de palabras clave para determinar qué categoría se adapta mejor a la pregunta de entrada. Este paso de enrutamiento garantiza que la estrategia de recuperación se adapte dinámicamente a diferentes estilos de consulta. Consulta los CÓDIGOS COMPLETOS aquí.

clase AnswerGenerator: def __init__(self, model_name=”google/flan-t5-base”): print(f”Cargando modelo de generación: {model_name}…”) self.generator = pipeline(‘text2text-generación’, modelo=model_name, dispositivo=0 if torch.cuda.is_available() else -1, max_length=256) tipo_dispositivo = “GPU” if torch.cuda.is_available() else “CPU” print(f” ✓ Generador listo (usando {tipo_dispositivo})\n”) def generate(self, consulta: str, contexto: Lista[Dict]tipo_consulta: cadena) -> cadena: contexto_texto = “\n\n”.join([f”[{doc[‘source’]}]: {doc[‘text’]}” para doc en contexto]) Contexto: {context_text} Pregunta: {consulta} Respuesta:””” respuesta = self.generator(prompt, max_length=200, do_sample=False)[0][‘generated_text’]

devolver respuesta.strip() def self_check(self, consulta: str, respuesta: str, contexto: Lista[Dict]) -> Tupla[bool, str]: si len(respuesta) < 10: devuelve Falso, "Respuesta demasiado corta - necesita más detalles" context_keywords = set() para doc en contexto: context_keywords.update(doc['text'].inferior().split()[:20]) respuestas_palabras = set(answer.lower().split()) superposición = len(context_keywords.intersection(answer_words)) si superposición < 2: devuelve Falso, "Respuesta no basada en contexto - necesita más evidencia" query_keywords = set(query.lower().split()) si len(query_keywords.intersection(answer_words)) < 1: devuelve Falso, "La respuesta no aborda la consulta - reformular necesario" return Verdadero, "Calidad de respuesta aceptable"

Creamos la clase AnswerGenerator para manejar la creación de respuestas y la autoevaluación. Utilizando el modelo Flan-T5, generamos respuestas de texto basadas en documentos recuperados. Luego, realizamos una autoevaluación para evaluar la extensión de la respuesta, el contexto y la relevancia, asegurando que nuestro resultado sea significativo y preciso. Consulta los CÓDIGOS COMPLETOS aquí.

clase AgenticRAG: def __init__(self): self.vector_store = VectorStore() self.router = QueryRouter() self.generator = AnswerGenerator() self.max_iterations = 2 def add_knowledge(self, documentos: Lista[str]fuentes: Lista[str]): self.vector_store.add_documents(documentos, fuentes) def query(self, question: str, detallado: bool = True) -> Dict: if detallado: print(f”\n{‘=’*60}”) print(f”🤔 Consulta: {pregunta}”) print(f”{‘=’*60}”) query_type = self.router.route(pregunta) si es detallado: print(f”📍 Ruta: {query_type.upper()} consulta detectada”) k_docs = {‘technical’: 2, ‘comparative’: 4, ‘procedural’: 3}.get(query_type, 3) iteración = 0 respuesta_accepted = False mientras iteración < self.max_iterations y no respuesta_aceptada: iteración += 1 si es detallado: print(f"\n🔄 Iteración {iteración}") contexto = self.vector_store.search(question, k=k_docs) si es detallado: print(f"📚 Documentos recuperados {len(context)} de fuentes:") para doc en contexto: print(f" - {doc['source']}") respuesta = self.generator.generate(pregunta, contexto, tipo_consulta) si es detallado: print(f"💡 Respuesta generada: {respuesta[:100]}...") respuesta_aceptada, retroalimentación = self.generator.self_check(pregunta, respuesta, contexto) si detallado: estado = " ✓ ACEPTADO" si respuesta_aceptada else "✗ RECHAZADO" print(f"🔍 Autocomprobación: {status}") print(f" Comentarios: {feedback}") si no respuesta_aceptada e iteración < self.max_iterations: pregunta = f"{pregunta} (proporcione más específico detalles)" k_docs += 1 return {'respuesta': respuesta, 'query_type': query_type, 'iteraciones': iteración, 'aceptado': respuesta_aceptada, 'fuentes': [doc['source'] para doc en contexto]}

Combinamos todos los componentes en el sistema AgenticRAG, que organiza el enrutamiento, la recuperación, la generación y el control de calidad. El sistema refina iterativamente sus respuestas basándose en comentarios de autoevaluación, ajustando la consulta o ampliando el contexto cuando sea necesario. Esto crea un RAG de árbol de decisiones basado en retroalimentación que mejora automáticamente el rendimiento. Consulta los CÓDIGOS COMPLETOS aquí.

def main(): print(“\n” + “=”*60) print(“🚀 TRAPO AGÉNTICO CON ENRUTAMIENTO Y AUTOCOMPROBACIÓN”) print(“=”*60 + “\n”) documentos = [
“RAG (Retrieval-Augmented Generation) combines information retrieval with text generation. It retrieves relevant documents and uses them as context for generating accurate answers.”
]
fuentes = [“Python Documentation”, “ML Textbook”, “Neural Networks Guide”, “Deep Learning Paper”, “Transformer Architecture”, “RAG Research Paper”]
rag = AgenticRAG() rag.add_knowledge(documentos, fuentes) test_queries = [“What is Python?”, “How does machine learning work?”, “Compare neural networks and deep learning”]
para consulta en test_queries: resultado = rag.query(query, detallado=True) print(f”\n{‘=’*60}”) print(f”📊 RESULTADO FINAL:”) print(f” Respuesta: {resultado[‘answer’]}”) print(f” Tipo de consulta: {resultado[‘query_type’]}”) print(f” Iteraciones: {resultado[‘iterations’]}”) print(f” Aceptado: {resultado[‘accepted’]}”) print(f”{‘=’*60}\n”) if __name__ == “__main__”: principal()

Finalizamos la demostración cargando una pequeña base de conocimientos y ejecutando consultas de prueba a través del canal Agentic RAG. Observamos cómo el modelo enruta, recupera y refina las respuestas paso a paso, imprimiendo resultados intermedios para mayor transparencia. Al final, confirmamos que nuestro sistema ofrece con éxito respuestas precisas y autovalidadas utilizando únicamente cálculos locales.

En conclusión, creamos un marco Agentic RAG completamente funcional que recupera, razona y refina sus respuestas de forma autónoma. Somos testigos de cómo el sistema enruta dinámicamente diferentes tipos de consultas, evalúa sus propias respuestas y las mejora a través de comentarios iterativos, todo dentro de un entorno local liviano. A través de este ejercicio, profundizamos nuestra comprensión de las arquitecturas RAG y también experimentamos cómo los componentes agentes pueden transformar sistemas de recuperación estáticos en agentes inteligentes que se mejoran a sí mismos.

Consulta los CÓDIGOS COMPLETOS aquí. No dude en consultar nuestra página de GitHub para tutoriales, códigos y cuadernos. Además, no dude en seguirnos en Twitter y no olvide unirse a nuestro SubReddit de más de 100.000 ML y suscribirse a nuestro boletín. ¡Esperar! estas en telegrama? Ahora también puedes unirte a nosotros en Telegram.

Asif Razzaq es el director ejecutivo de Marktechpost Media Inc.. Como empresario e ingeniero visionario, Asif está comprometido a aprovechar el potencial de la inteligencia artificial para el bien social. Su esfuerzo más reciente es el lanzamiento de una plataforma de medios de inteligencia artificial, Marktechpost, que se destaca por su cobertura en profundidad del aprendizaje automático y las noticias sobre aprendizaje profundo que es técnicamente sólida y fácilmente comprensible para una amplia audiencia. La plataforma cuenta con más de 2 millones de visitas mensuales, lo que ilustra su popularidad entre el público.

🙌 Siga MARKTECHPOST: agréguenos como fuente preferida en Google.