En este tutorial, creamos un sistema progresista de recuperación de agentes y gestació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 franco como FAISS, SentenceTransformers y Flan-T5. A medida que avanzamos, exploramos cómo el enrutamiento, la recuperación, la gestación y la autoevaluación se combinan para formar un proceso RAG estilo árbol de decisiones que imita el razonamiento agente del mundo actual. Mira el CÓDIGOS COMPLETOS aquí.
print("🔧 Setting up dependencies...")
import subprocess
import sys
def install_packages():
packages = ('sentence-transformers', 'transformers', 'torch', 'faiss-cpu', 'numpy', 'accelerate')
for package in packages:
print(f"Installing {package}...")
subprocess.check_call((sys.executable, '-m', 'pip', 'install', '-q', package))
try:
import faiss
except ImportError:
install_packages()
print("✓ All dependencies installed! Importing modules...n")
import torch
import numpy as np
from sentence_transformers import SentenceTransformer
from transformers import pipeline
import faiss
from typing import List, Dict, Tuple
import warnings
warnings.filterwarnings('ignore')
print("✓ All modules loaded successfully!n")Comenzamos instalando todas las dependencias necesarias, incluidos Transformers, FAISS y SentenceTransformers, para certificar una ejecución recinto sin problemas. Verificamos instalaciones e instalamos módulos esenciales como NumPy, PyTorch y FAISS para incrustación, recuperación y gestación. Confirmamos que todas las bibliotecas se cargan correctamente ayer de continuar con el proceso principal. Mira el CÓDIGOS COMPLETOS aquí.
class VectorStore:
def __init__(self, embedding_model="all-MiniLM-L6-v2"):
print(f"Loading embedding model: {embedding_model}...")
self.embedder = SentenceTransformer(embedding_model)
self.documents = ()
self.index = None
def add_documents(self, docs: List(str), sources: List(str)):
self.documents = ({"text": doc, "source": src} for doc, src in zip(docs, sources))
embeddings = self.embedder.encode(docs, show_progress_bar=False)
dimension = embeddings.shape(1)
self.index = faiss.IndexFlatL2(dimension)
self.index.add(embeddings.astype('float32'))
print(f"✓ Indexed {len(docs)} documentsn")
def search(self, query: str, k: int = 3) -> List(Dict):
query_vec = self.embedder.encode((query)).astype('float32')
distances, indices = self.index.search(query_vec, k)
return (self.documents(i) for i in indices(0))Diseñamos la clase VectorStore para acumular y recuperar documentos de forma valioso utilizando la búsqueda por similitud basada en FAISS. Incorporamos cada documento utilizando un maniquí transformador y creamos un índice para una recuperación rápida. Esto nos permite averiguar rápidamente el contexto más relevante para cualquier consulta entrante. Mira el CÓDIGOS COMPLETOS aquí.
class QueryRouter:
def __init__(self):
self.categories = {
'technical': ('how', 'implement', 'code', 'function', 'algorithm', 'debug'),
'factual': ('what', 'who', 'when', 'where', 'define', 'explain'),
'comparative': ('compare', 'difference', 'frente a', 'vs', 'better', 'which'),
'procedural': ('steps', 'process', 'guide', 'tutorial', 'how to')
}
def route(self, query: str) -> str:
query_lower = query.lower()
scores = {}
for category, keywords in self.categories.items():
score = sum(1 for kw in keywords if kw in query_lower)
scores(category) = score
best_category = max(scores, key=scores.get)
return best_category if scores(best_category) > 0 else 'factual'Introducimos la clase QueryRouter para clasificar consultas por intención, técnicas, fácticas, comparativas o de procedimiento. Utilizamos la concordancia de palabras esencia para determinar qué categoría se adapta mejor a la pregunta de entrada. Este paso de enrutamiento garantiza que la logística de recuperación se adapte dinámicamente a diferentes estilos de consulta. Mira el CÓDIGOS COMPLETOS aquí.
class AnswerGenerator:
def __init__(self, model_name="google/flan-t5-base"):
print(f"Loading generation model: {model_name}...")
self.generator = pipeline('text2text-generation', model=model_name, device=0 if torch.cuda.is_available() else -1, max_length=256)
device_type = "GPU" if torch.cuda.is_available() else "CPU"
print(f"✓ Generator ready (using {device_type})n")
def generate(self, query: str, context: List(Dict), query_type: str) -> str:
context_text = "nn".join((f"({doc('source')}): {doc('text')}" for doc in context))
Context:
{context_text}
Question: {query}
Answer:"""
answer = self.generator(prompt, max_length=200, do_sample=False)(0)('generated_text')
return answer.strip()
def self_check(self, query: str, answer: str, context: List(Dict)) -> Tuple(bool, str):
if len(answer) < 10:
return False, "Answer too short - needs more detail"
context_keywords = set()
for doc in context:
context_keywords.update(doc('text').lower().split()(:20))
answer_words = set(answer.lower().split())
overlap = len(context_keywords.intersection(answer_words))
if overlap < 2:
return False, "Answer not grounded in context - needs more evidence"
query_keywords = set(query.lower().split())
if len(query_keywords.intersection(answer_words)) < 1:
return False, "Answer doesn't address the query - rephrase needed"
return True, "Answer quality acceptable"Creamos la clase AnswerGenerator para manejar la creación de respuestas y la autoevaluación. Utilizando el maniquí 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. Mira el CÓDIGOS COMPLETOS aquí.
class AgenticRAG:
def __init__(self):
self.vector_store = VectorStore()
self.router = QueryRouter()
self.generator = AnswerGenerator()
self.max_iterations = 2
def add_knowledge(self, documents: List(str), sources: List(str)):
self.vector_store.add_documents(documents, sources)
def query(self, question: str, verbose: bool = True) -> Dict:
if verbose:
print(f"n{'='*60}")
print(f"🤔 Query: {question}")
print(f"{'='*60}")
query_type = self.router.route(question)
if verbose:
print(f"📍 Route: {query_type.upper()} query detected")
k_docs = {'technical': 2, 'comparative': 4, 'procedural': 3}.get(query_type, 3)
iteration = 0
answer_accepted = False
while iteration < self.max_iterations and not answer_accepted:
iteration += 1
if verbose:
print(f"n🔄 Iteration {iteration}")
context = self.vector_store.search(question, k=k_docs)
if verbose:
print(f"📚 Retrieved {len(context)} documents from sources:")
for doc in context:
print(f" - {doc('source')}")
answer = self.generator.generate(question, context, query_type)
if verbose:
print(f"💡 Generated answer: {answer(:100)}...")
answer_accepted, feedback = self.generator.self_check(question, answer, context)
if verbose:
status = "✓ ACCEPTED" if answer_accepted else "✗ REJECTED"
print(f"🔍 Self-check: {status}")
print(f" Feedback: {feedback}")
if not answer_accepted and iteration < self.max_iterations:
question = f"{question} (provide more specific details)"
k_docs += 1
return {'answer': answer, 'query_type': query_type, 'iterations': iteration, 'accepted': answer_accepted, 'sources': (doc('source') for doc in context)}Combinamos todos los componentes en el sistema AgenticRAG, que organiza el enrutamiento, la recuperación, la gestació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 feedback que progreso automáticamente el rendimiento. Mira el CÓDIGOS COMPLETOS aquí.
def main():
print("n" + "="*60)
print("🚀 AGENTIC RAG WITH ROUTING & SELF-CHECK")
print("="*60 + "n")
documents = (
"RAG (Retrieval-Augmented Generation) combines information retrieval with text generation. It retrieves relevant documents and uses them as context for generating accurate answers."
)
sources = ("Python Documentation", "ML Textbook", "Neural Networks Guide", "Deep Learning Paper", "Transformer Architecture", "RAG Research Paper")
rag = AgenticRAG()
rag.add_knowledge(documents, sources)
test_queries = ("What is Python?", "How does machine learning work?", "Compare neural networks and deep learning")
for query in test_queries:
result = rag.query(query, verbose=True)
print(f"n{'='*60}")
print(f"📊 FINAL RESULT:")
print(f" Answer: {result('answer')}")
print(f" Query Type: {result('query_type')}")
print(f" Iterations: {result('iterations')}")
print(f" Accepted: {result('accepted')}")
print(f"{'='*60}n")
if __name__ == "__main__":
main()Finalizamos la demostración cargando una pequeña cojín de conocimientos y ejecutando consultas de prueba a través del canal Agentic RAG. Observamos cómo el maniquí enruta, recupera y refina las respuestas paso a paso, imprimiendo resultados intermedios para maduro transparencia. Al final, confirmamos que nuestro sistema ofrece con éxito respuestas precisas y autovalidadas utilizando solamente cálculos locales.
En conclusión, creamos un ámbito Agentic RAG completamente utilitario 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 progreso a través de comentarios iterativos, todo adentro de un entorno recinto leve. A través de este entrenamiento, profundizamos nuestra comprensión de las arquitecturas RAG y todavía experimentamos cómo los componentes agentes pueden alterar sistemas de recuperación estáticos en agentes inteligentes que se mejoran a sí mismos.
Mira el CÓDIGOS COMPLETOS aquí. No dudes en consultar nuestra Página de GitHub para tutoriales, códigos y cuadernos. Encima, no dudes en seguirnos en Gorjeo y no olvides unirte a nuestro SubReddit de más de 100.000 ml y suscríbete a nuestro boletín. ¡Esperar! estas en telegrama? Ahora todavía puedes unirte a nosotros en Telegram.
Asif Razzaq es el director ejecutor de Marktechpost Media Inc.. Como emprendedor e ingeniero iluminado, Asif está comprometido a beneficiarse el potencial de la inteligencia químico para el correctamente social. Su esfuerzo más fresco es el dispersión de una plataforma de medios de inteligencia químico, Marktechpost, que se destaca por su cobertura en profundidad del estudios inevitable y las noticiero sobre estudios 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 conocido.