¿Cómo construir un sistema avanzado de generación de recuperación de agente (RAG) con estrategia dinámica y recuperación inteligente?

En este tutorial, pasamos por la implementación de un sistema de generación de recuperación de agente (RAG). Lo diseñamos para que el agente haga más que recuperar documentos; Decide activamente cuándo se necesita recuperación, selecciona la mejor estrategia de recuperación y sintetiza las respuestas con conciencia contextual. Al combinar incrustaciones, indexación FAISS y un simulacro de LLM, creamos una demostración práctica de cómo la toma de decisiones de agente puede elevar la tubería de trapo estándar a algo más adaptativo e inteligente. Mira los códigos completos aquí.

import numpy como np import faiss de orente_transformers import sentencetransformer import json import re de escribir lista de importación, dict, cualquiera, opcional de dataClasses import fatacLass de enum import enum class mockllm: def generar (self, solicit: str, max_tokens: int = 150) -> str: indic_lower = aprop. Si alguno (palabra en apur_lower para palabras en [“specific”, “recent”, “data”, “facts”, “when”, “who”, “what”]): devolver “Recuperar: la consulta requiere información objetiva específica que debe recuperarse”. else: return “No_retrieve: esta es una pregunta general que puede responderse con el conocimiento existente”. ELIF “Elija Estrategia de recuperación” en Pront_lower: si “comparación” en Pront_lower o “versus” en la estrategia de retrat_lower: return “: multi_query: necesita recuperar información sobre múltiples entidades para comparar”. ELIF “reciente” en apur_lower o “Último” en aproped_lower: return “Estrategia: temporal – enfoque en la información reciente”. else: return “estrategia: semántica – búsqueda de similitud semántica estándar”. Elif “Synthese” en aprop_lower y “contexto:” en pronto: return “basado en la información recuperada, aquí hay una respuesta integral que combina múltiples fuentes y proporciona detalles específicos con un contexto adecuado”. Regrese “Esta es una respuesta simulada. En la práctica, use un verdadero LLM como el GPT de OpenAi o similar”. class RECRIEVALSTRATY (enum): semántico = “semántico” multi_query = “multi_query” temporal = “temporal” hybrid = “hybrid” @dataclass Documento de clase: id: contenido str: str metadata: dict: dict[str, Any]
Incrustación: opcional[np.ndarray] = Ninguno

Configuramos la base de nuestro sistema de trapo de agente. Definimos un simulacro de LLM para simular la toma de decisiones, crear una estrategia de recuperación enum y diseñar un documento de datos para que podamos estructurar y administrar nuestra base de conocimiento de manera eficiente. Mira los códigos completos aquí.

clase AgenticRagSystem: def __init __ (self, model_name: str = “all-minilm-l6-v2”): self.encoder = sentencetransformer (model_name) self.llm = mockllm () self.documents: list[Document] = []
self.index: opcional[faiss.Index] = Ninguno Def add_documents (self, documentos: lista[Dict[str, Any]]) -> Ninguno: print (f “procesamiento {len (documentos)} documentos …”) para i, doc en enumerado (documentos): doc_obj = document (id = doc.get (‘id’, str (i)), content = doc[‘content’]metadata = doc.get (‘metadata’, {})) self.documents.append (doc_obj) contenido = [doc.content for doc in self.documents]
incrustaciones = self.encoder.encode (contenido, show_progress_bar = true) para DOC, incrustando en zip (self.documents, incrustes): doc.embedding = incrusting dimension = incrustar.shape[1]
self.index = faiss.indexflatip (dimension) faiss.normalize_l2 (incrustes) self.index.add (increddings.astype (‘float32’)) imprime (f “base de conocimiento construida con {len (self.documents)} documentos”)

Construimos el núcleo de nuestro sistema de trapo de agente. Inicializamos el modelo de incrustación, configuramos el índice FAISS y agregamos documentos codificando sus contenidos en vectores, permitiendo una recuperación semántica rápida y precisa de nuestra base de conocimiento. Mira los códigos completos aquí.

Def decide_retrieval (self, consulta: str) -> bool: decision_prompt = f “” “Analice la siguiente consulta y decida si recuperar información: consulta:” {consulta} “Decide si recuperar información de la base de conocimiento. Considere si esto necesita hechos específicos, datos recientes o puede ser respondido generalmente. Responda con: recuperar: recuperar: [reason] o no_retrieve: [reason]
“” “Respuesta = self.llm.Generate (Decision_Prompt) debería_retrieve = Response.Startswith (” Recuperar: “) Imprimir (F” 🤖 Agente Decisión: {‘Recuperar’ si debería_ratar más ‘Respuesta directa’} “) Impresión (F” Razonamiento: {Response.split (‘:’, 1)[1].Strip () if ‘:’ en respuesta más respuesta} “) return debería ser_retrieve def elige_strategy (self, consulta: str) -> RECORTALSTRATYGY: Strategy_Prompt = f” “” Elija la mejor estrategia de recuperación para esta consulta: consultas: “ALTI Híbrido: enfoque combinado Elija estrategia de recuperación y explique por qué. Responder con: Estrategia: [strategy_name] – [reasoning]
“” “” Respuesta = Self.llm.Generate (estrategia_prompt) si “multi_query” en respuesta.lower (): estrategia = recuperación de retrato. print (f “🎯 Estrategia de recuperación: {estrategia.value}”) imprime (f “razonamiento: {respuesta.split (‘-‘, 1)[1].Strip () if ‘-‘ en respuesta más respuesta} “) Estrategia de retorno

Le damos a nuestro agente la capacidad de pensar antes de que obtenga. Primero determinamos si una consulta realmente requiere recuperación, luego seleccionamos la estrategia más adecuada: semántica, múltiple, temporal o híbrida. Esto nos permite dirigir el contexto correcto con un razonamiento claro e impreso para cada paso. Mira los códigos completos aquí.

Def Retrieve_Documents (self, consulta: STR, estrategia: recuperación de recuperación, k: int = 3) -> list[Document]: si no self.index: imprime (“❌ no hay base de conocimiento disponible”) return []

if estrategia == RECERVIRSTRATY.MULTI_QUERY: QUERIES = [query, f”advantages of {query}”, f”disadvantages of {query}”]
all_docs = []
Para Q en consultas: docs = self._semantic_search (q, k = 2) all_docs.extend (docs) Seen_ids = set () unique_docs = []
Para DOC en all_docs: si doc.id no en seen_ids: unique_docs.append (doc) Seen_ids.add (doc.id) return unique_docs[:k]

ELIF Strategy == RECINSTRATGY.Temporal: DOCS = SELF._SEMANTIC_SEARCH (QUERY, K = K*2) DOCS_WITH_DATES = [(doc, doc.metadata.get(‘date’, ‘1900-01-01’)) for doc in docs]
docs_with_dates.sort (key = lambda x: x[1]reverso = true) return [doc for doc, _ in docs_with_dates[:k]]else: return self._semantic_search (consulta, k = k) def _semantic_search (self, consulta: str, k: int) -> list[Document]: query_embedding = self.encoder.encode ([query]) faiss.normaly_l2 (query_embedding) puntajes, indices = self.index.search (query_embedding.astype (‘float32’), k) resultados = []
Para el puntaje, IDX en zip (puntajes[0]índices[0]): if idx str: si no recuperado_docs: return self.llm.Generate (f “Responda esta consulta: {Query}”) context = “\ n \ n” .Join ([f”Document {i+1}: {doc.content}”
for i, doc in enumerate(retrieved_docs)]) synthesis_promppt = f “” “consulta: {consulta} contexto: {context} Synthese una respuesta integral usando el contexto proporcionado. Sea específico y haga referencia a las fuentes de información cuando sea relevante.” “” Return Self.llm.Generate (synthesis_prompt, max_tokens = 200)

Implementamos cómo realmente buscamos y usamos conocimiento. Realizamos la búsqueda semántica, ramificamos en el reanimiento de múltiples cantidades o temporales cuando sea necesario, deduplicamos los resultados y luego sintetizamos una respuesta enfocada del contexto recuperado. Al hacerlo, mantenemos una recuperación eficiente, transparente y estrechamente alineada. Mira los códigos completos aquí.

Def Query (self, Query: str) -> str: print (f “\ n🔍 Processing Query: ‘{Query}'”) print (“=” * 50) Si no self.decide_retrieval (consulta): print (“\ n📝 respuesta directa …”) return print (f “\ n📚 Recuperando documentos utilizando {estrategia de estrategia Recupered_docs: imprime (“\ n📄 contexto recuperado:”) para i, doc en enumerate (recuperado_docs[:2]1): print (f “{i}. {Doc.content[:100]} … “) Respuesta de retorno

Entramos todas las partes en una sola tubería. Cuando ejecutamos una consulta, primero determinamos si la recuperación es necesaria, luego seleccione la estrategia apropiada, obtenga documentos en consecuencia y finalmente sintetizamos una respuesta al tiempo que muestra el contexto recuperado para la transparencia. Esto hace que el sistema se sienta más agente y explicable. Mira los códigos completos aquí.

Def create_sample_knowledge_base (): return [
{
“id”: “ai_1”,
“content”: “Artificial Intelligence (AI) refers to computer systems that can perform tasks requiring human intelligence”,
“metadata”: {“topic”: “AI basics”, “date”: “2024-01-15”}
},
{
“id”: “ml_1”,
“content”: “ML is a subset of AI.”,
“metadata”: {“topic”: “Machine Learning”, “date”: “2024-02-10”}
},
{
“id”: “rag_1”,
“content”: “Retrieval-Augmented Generation (RAG) combines the power of large language models with external knowledge retrieval to provide more accurate and up-to-date responses.”,
“metadata”: {“topic”: “RAG”, “date”: “2024-03-05”}
},
{
“id”: “agents_1”,
“content”: “AI agents”,
“metadata”: {“topic”: “AI Agents”, “date”: “2024-03-20”}
}
]

if __name__ == “__main__”: imprime (“🚀 Inicializando el sistema de trapo de agente …”) rag_system = agenticragSystem () docs = create_sample_knowledge_base () rag_system.add_documents (docs) demo_queries = [
“What is artificial intelligence?”,
“How are you today?”,
“Compare AI and Machine Learning”,
]

para consulta en demo_queries: respuesta = rag_system.query (consulta) print (f “\ n💬 Respuesta final: {Respuesta}”) print (“\ n” + “=”*80) print (“\ n✅ tutorial de rag agente completo!”) print (“\ nkey características demostradas:”) imprime (“• decisiones de recuperación de retrato de agente”) print (“• selección de la estrategia dinámica”) Enfoques de recuperación “) Impresión (” • Proceso de razonamiento transparente “)

Envolvemos todo en una demostración ejecutable. Creamos una pequeña base de conocimiento de documentos relacionados con la IA, inicializamos el sistema de Rag Agentic y ejecutamos consultas de muestra que resalten varios comportamientos, incluida la recuperación, la respuesta directa y la comparación. Este bloque final une todo el tutorial y muestra el razonamiento del agente en acción.

En conclusión, vemos cómo las decisiones de recuperación impulsadas por el agente, la selección de la estrategia dinámica y el razonamiento transparente se unen para formar un flujo de trabajo de trapo de agente avanzado. Ahora tenemos un sistema de trabajo que destaca el potencial de agregar agencia a RAG, haciendo que la recuperación de la información sea más inteligente, más dirigida y más humana en su adaptabilidad. Esta base nos permite extender el sistema con LLM reales, bases de conocimiento más grandes y estrategias más sofisticadas en futuras iteraciones.

Mira los códigos completos aquí. No dude en consultar nuestra página de GitHub para obtener tutoriales, códigos y cuadernos. Además, no dude en seguirnos en Twitter y no olvide unirse a nuestro subreddit de 100k+ ml y suscribirse a nuestro boletín.

Asif Razzaq es el CEO de MarktechPost Media Inc .. Como empresario e ingeniero visionario, ASIF se compromete 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 profunda de noticias de aprendizaje automático y de aprendizaje profundo que es técnicamente sólido y fácilmente comprensible por una audiencia amplia. La plataforma cuenta con más de 2 millones de vistas mensuales, ilustrando su popularidad entre el público.

🔥[Recommended Read] NVIDIA AI Open-Sources Vipe (motor de pose de video): una herramienta de anotación de video 3D potente y versátil para AI espacial