Notas prácticas: Parte 3: RAG avanzado — HyDE, autocompletado de consultas y flujos de trabajo agentes
Guía práctica paso a paso de Notas prácticas: Parte 3: RAG experto — HyDE, consultas automáticas y flujos de trabajo agentes: contratos, verificaciones y espacios para código listo para usar para equipos que desarrollan sistemas RAG.
Esta guía reconstruye el proceso desde las materias primas hasta un sistema funcional para: Parte 3: RAG avanzado — HyDE, consultas automáticas y flujos de trabajo agentes. El enfoque está en pasos operativos, verificaciones explícitas y código que se puede incorporar directamente a un repositorio sin necesidad de adivinar la intención. Para tener una visión general, defina las entradas, el responsable del paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Registre los tiempos de ejecución y el costo en tokens o consultas junto con los resultados funcionales. La visibilidad temprana del costo evita facturas inesperadas cuando el proceso pasa de una demostración a entornos compartidos.
1. HyDE: Incrustaciones hipotéticas de documentos
Al trabajar en 1. HyDE: Hypothetical Document Embeddings, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Guarde la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de datos secretos y las banderas de funcionalidad deben estar en un lugar donde los operadores puedan auditarlos sin tener que leer todo el sistema. Mida la tasa de recuperación en un conjunto fijo de preguntas antes de ajustar los prompts. El cambio constante de prompts rara vez soluciona un sistema de recuperación deficiente.
La idea clave
Al trabajar en The Insight, anote primero el contrato: los datos de entrada requeridos, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Documente junto con ello el camino óptimo y el camino de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras posteriores. Mida la capacidad de recuperación con un conjunto fijo de preguntas antes de ajustar los prompts. El cambio constante de prompts rara vez soluciona un sistema de recuperación deficiente.
Cómo funciona HyDE
Al estudiar cómo funciona HyDE, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación ayuda a mantener honestos los cambios posteriores en el código. Prefiera unidades pequeñas y probables sobre scripts extensos. Cuando un paso falla, el fallo debe apuntar a una única responsabilidad y no a un proceso complicado. Mida el rendimiento de recuperación con un conjunto fijo de preguntas antes de ajustar los prompts. El cambio constante de prompts rara vez soluciona un sistema de recuperación deficiente.
Step 1: User Query
"What are ISRO's major achievements?"
Step 2: Generate Hypothetical Answer (using LLM)
"ISRO has achieved several milestones including reaching Mars orbit
in 2014, successfully landing Chandrayaan-3 on the Moon's south pole
in 2023, and launching satellites for multiple countries at low cost..."
Step 3: Embed the Hypothetical Answer
[0.23, -0.45, 0.67, ...] ← This lives in document space!
Step 4: Retrieve Documents Similar to Hypothetical Answer
Now we're comparing document-to-document, not query-to-document
Step 5: Generate Final Answer
Use retrieved docs + original query → Better answer
Por qué funciona HyDE
Al trabajar en “Why HyDE Works”, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Considere esta etapa como un contrato entre las entradas y las salidas validadas. Asigne nombres a los artefactos, defina comprobaciones de éxito y rechace las completaciones parciales silenciosas. Mida la capacidad de recuperación con un conjunto fijo de preguntas antes de ajustar los prompts. El cambio constante de prompts rara vez soluciona un sistema de recuperación deficiente.
Cuándo usar HyDE
Al trabajar en “When to Use HyDE”, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Registre los tiempos y el costo de tokens o consultas junto con los resultados funcionales. Tener visibilidad del costo desde el principio evita facturas inesperadas cuando el proceso pasa de la fase de demostración a entornos compartidos. Mida el rendimiento en un conjunto fijo de preguntas antes de ajustar los prompts. El cambio constante de prompts rara vez soluciona un sistema de recuperación deficiente.
HyDE vs RAG tradicional: Ejemplo real
Al trabajar en “HyDE vs Traditional RAG: Ejemplo Real”, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Guarde la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de secretos y las banderas de funcionalidad deben estar en un lugar donde los operadores puedan auditarlos sin tener que leer todo el sistema. Mida la capacidad de recuperación con un conjunto fijo de preguntas antes de ajustar los prompts. El cambio constante de prompts rara vez soluciona un sistema de recuperación deficiente.
2. Compresión contextual: Precisión sobre cantidad
Al trabajar en la sección 2. Compresión contextual: precisión sobre cantidad, anote primero el contrato: entradas requeridas, señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Documente junto con ello el camino óptimo y el camino de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no de mejoras posteriores. Mida el rendimiento en un conjunto fijo de preguntas antes de ajustar los prompts. El cambio constante de prompts rara vez soluciona un sistema de recuperación deficiente.
El problema
Al abordar “El Problema”, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Prefiera unidades pequeñas y verificables a scripts extensos. Cuando un paso falla, el fallo debe apuntar a una única responsabilidad y no a un proceso complicado. Mida el rendimiento en un conjunto fijo de preguntas antes de ajustar los prompts. El cambio constante de prompts rara vez soluciona un sistema de recuperación deficiente. Al abordar “El Problema”, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Registre los tiempos de ejecución y el costo en tokens o consultas junto con los resultados funcionales. Tener visibilidad del costo desde el principio evita facturas inesperadas cuando se pasa de entornos de demostración a entornos compartidos.
India's capital city is New Delhi, which is located in the northern
part of the country. The city serves as the center of the Government
of India and houses important governmental buildings including the
Parliament House, Rashtrapati Bhavan, and various ministry buildings.
New Delhi was inaugurated in 1931 and became the capital of India
after independence in 1947. The previous capital was Calcutta, now
known as Kolkata. The decision to move the capital was made by the
British colonial government in 1911...
India's capital city is New Delhi
Solución de Compresión Contextual
La solución de compresión contextual funciona mejor cuando se trata como una superficie medible. Capture un transcripte ejemplar, un caso de fallo y la nota de reversión antes de ampliar el alcance. Mantenga la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de datos secretos y las banderas de funcionalidad deben estar en un lugar donde los operadores puedan auditarlos sin tener que leer todo el sistema. Separe la política de fragmentación de la política de recuperación. Cambiar una no debe obligar a reescribir la otra cuando cambian las métricas de calidad.
Query + Chunk → Compressor LLM → Relevant Sentences Only
Estrategia de implementación
La estrategia de implementación funciona mejor cuando se trata como un elemento medible. Capture una transcripción ejemplar, un caso de fallo y la nota de reversión antes de ampliar el alcance. Documente tanto el camino óptimo como el de recuperación juntos. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras posteriores. Separe la política de fragmentación de la política de recuperación; cambiar una no debe obligar a reescribir la otra cuando cambian las métricas de calidad.
# Traditional: Send full chunks
context = chunk1 + chunk2 + chunk3 # 1500 tokens
# Compressed: Extract relevant parts
for chunk in chunks:
compressed = compressor.extract_relevant(query, chunk)
context.append(compressed) # 300 tokens total
Beneficios
Benefits funciona mejor cuando se trata como una superficie medible. Capture un registro ideal, un caso de fallo y la nota de reversión antes de ampliar el alcance. Prefiera unidades pequeñas y verificables en lugar de scripts extensos. Cuando un paso falla, el fallo debe apuntar a una sola responsabilidad y no a un proceso complicado. Separe la política de fragmentación de la política de recuperación. Cambiar una no debe obligar a reescribir la otra cuando cambian las métricas de calidad. Benefits funciona mejor cuando se trata como una superficie medible. Capture un registro ideal, un caso de fallo y la nota de reversión antes de ampliar el alcance. Registre los tiempos y el costo de tokens o consultas junto con los resultados funcionales. Tener visibilidad del costo desde el principio evita facturas inesperadas cuando el proceso pasa de la versión de demostración a entornos compartidos.
3. Autopreguntado: Deje que el LLM decida
Para el punto 3: Autopreguntado. Deje que el LLM decida; defina las entradas, el responsable de la tarea y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la tarea a partir de un punto de control conocido, sin tener que adivinar el estado oculto. Mantenga la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de datos confidenciales y las banderas de funcionalidad deben estar en un único lugar que los operadores puedan auditar sin necesidad de leer todo el sistema.
El concepto
Para el concepto, defina las entradas, el responsable de la etapa y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la etapa desde un punto de control conocido sin tener que adivinar el estado oculto. Documente tanto la ruta óptima como la ruta de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras posteriores. Cite los pasajes que realmente sustentan la respuesta. Sin citaciones, los operadores no pueden distinguir entre alucinaciones y brechas en el indexado.
Tradicional vs. de autoconsulta
Para los enfoques tradicionales frente a los de autoconsulta, defina las entradas, el responsable de la tarea y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la tarea a partir de un punto de control conocido sin tener que adivinar el estado oculto. Prefiera unidades pequeñas y verificables en lugar de scripts extensos. Cuando una tarea falla, el error debe indicar una única responsabilidad y no un proceso complicado. Cite los pasajes que realmente sustentan la respuesta. Sin citas, los operadores no pueden distinguir entre alucinaciones y fallos en el indexado. Para los enfoques tradicionales frente a los de autoconsulta, defina las entradas, el responsable de la tarea y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la tarea a partir de un punto de control conocido sin tener que adivinar el estado oculto. Registre los tiempos de ejecución y el costo en tokens o consultas junto con los resultados funcionales. Tener visibilidad del costo desde el principio evita facturas inesperadas cuando la tarea pasa de un entorno de demostración a uno compartido.
# Developer hardcodes everything
query = "Find recent cricket matches"
top_k = 5
filters = {"category": "sports"}
results = retriever.run(query, top_k, filters)
# LLM decides everything
query = "Find recent cricket matches"
# LLM analyzes and decides:
# - Extract metadata: {"sport": "cricket", "recency": "2024"}
# - Set top_k: 10 (wants comprehensive results)
# - Use hybrid search (keyword "matches" + semantic)
results = self_querying_retriever.run(query)
Por qué es importante la autoconsulta
Al trabajar en el tema de por qué es importante la autoconsulta, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Guarde la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de secretos y las banderas de funcionalidad deben encontrarse en un lugar donde los operadores puedan auditarlos sin tener que leer todo el sistema. Mida la capacidad de recuperación con un conjunto fijo de preguntas antes de ajustar los prompts. El cambio constante de prompts rara vez soluciona un sistema de recuperación deficiente.
Ejemplo: Autoconsulta en acción
Al trabajar en el Ejemplo: Autopreguntas en acción, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Documente junto con ello el camino óptimo y el camino de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no de mejoras posteriores. Mida la capacidad de recuperación con un conjunto fijo de preguntas antes de ajustar los prompts. El cambio constante de prompts rara vez soluciona un sistema de recuperación deficiente.
{
"semantic_query": "articles about India",
"metadata_filters": {
"year": 2023,
"content_type": "article"
},
"search_type": "hybrid",
"top_k": 10
}
4. RAG agente: La evolución definitiva
Al trabajar en 4. Agentic RAG: La evolución definitiva, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Prefiera unidades pequeñas y probables sobre scripts extensos. Cuando un paso falla, el fallo debe apuntar a una única responsabilidad y no a un proceso complicado. Mida la capacidad de recuperación con un conjunto fijo de preguntas antes de ajustar los prompts. El cambio constante de prompts rara vez soluciona un sistema de recuperación deficiente.
¿Qué hace que RAG sea “agente”?
Al trabajar en “¿Qué hace que RAG sea ‘agente’?”, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Considere esta etapa como un contrato entre las entradas y los resultados validados. Asigne nombres a los artefactos, defina comprobaciones de éxito y rechace las completaciones parciales silenciosas. Mida la capacidad de recuperación con un conjunto fijo de preguntas antes de ajustar los prompts. El cambio constante de prompts rara vez soluciona un sistema de recuperación deficiente.
Arquitectura RAG agente
Al trabajar con la arquitectura Agentic RAG, primero escribe el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Registra los tiempos y el costo en tokens o consultas junto con los resultados funcionales. Ver la inversión desde el principio evita facturas inesperadas cuando se pasa de entornos de demostración a entornos compartidos. Mide el rendimiento en la recuperación de información con un conjunto fijo de preguntas antes de ajustar los prompts. Cambiar constantemente los prompts rara vez soluciona un sistema de recuperación deficiente.
┌─────────────┐
│ User Query │
└──────┬──────┘
│
┌──────▼──────┐
│ Agent (LLM) │ ← Makes decisions
└──────┬──────┘
│
┌──────────────────┼──────────────────┐
│ │ │
┌────▼────┐ ┌─────▼─────┐ ┌─────▼──────┐
│ Tool 1 │ │ Tool 2 │ │ Tool 3 │
│ (RAG) │ │(Web Search│ │(Calculator)│
└────┬────┘ └─────┬─────┘ └─────┬──────┘
│ │ │
└──────────────────┼──────────────────┘
│
┌──────▼──────┐
│ Agent │ ← Synthesizes
└──────┬──────┘
│
┌──────▼──────┐
│ Final │
│ Answer │
└─────────────┘
Proceso de toma de decisiones del agente
Al trabajar en el proceso de toma de decisiones del agente, anote primero el contrato: los datos de entrada necesarios, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Guarde la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de datos secretos y las banderas de funcionalidad deben encontrarse en un lugar donde los operadores puedan auditarlos sin tener que leer todo el sistema. Mida la capacidad de recuperación con un conjunto fijo de preguntas antes de ajustar los prompts. El cambio constante de prompts rara vez soluciona un sistema de recuperación deficiente.
Step 1: Analyze Query
- Requires comparison between two countries
- Need current statistics
- Two separate pieces of info needed
Step 2: Plan Actions
- Tool 1: Search knowledge base for India population
- Tool 2: Search knowledge base for China population
- Tool 3: If not found, use web search as fallback
Step 3: Execute
→ Search knowledge base for India: Found
→ Search knowledge base for China: Not found
→ Fallback to web search for China: Found
Step 4: Synthesize
Combine information from both sources into coherent answer
Características de Agentic RAG
Ejemplo de flujo de trabajo agente
User: "Tell me about ISRO"
Agent: [Uses RAG tool] → Provides answer from knowledge base
User: "What about NASA?"
Agent: [Uses RAG tool] → Not found in knowledge base
[Falls back to web search] → Retrieves info from web
→ Provides answer with source attribution
User: "Compare their budgets"
Agent: [Analyzes] → Needs both ISRO and NASA budget data
[Retrieves from both sources]
[Uses calculator tool for comparison]
→ Provides detailed comparison
Ventajas de Agentic RAG
Combinación de todas las técnicas: el sistema RAG definitivo
User Query: "What are recent achievements in India's space program?"
↓
1. Self-Querying
LLM analyzes: Needs recent info, space domain
Filters: {topic: "space", recency: "2023-2024"}
↓
2. HyDE Generation
"ISRO achieved remarkable milestones in 2023-2024, including
successful Moon landings and satellite launches..."
↓
3. Hybrid Retrieval + HyDE
Retrieve using both: original query + hypothetical answer
Get top 20 documents
↓
4. Contextual Compression
Extract only sentences about recent achievements
Reduce 20 chunks (10k tokens) → 5 compressed chunks (2k tokens)
↓
5. Agentic Decision
Agent: "Retrieved info looks good, but let me verify with web search"
→ Quick web search for latest news
→ Combines both sources
↓
6. Final Answer
Comprehensive, accurate, up-to-date response with source attribution
Cuándo usar cada técnica
HyDE
Compresión contextual
Autoconsulta
Agentic RAG
Implementación paso a paso del código
import os
from pathlib import Path
import json
from typing import List, Dict, Any, Optional
from haystack import Pipeline, Document, component
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.components.retrievers.in_memory import (
InMemoryBM25Retriever,
InMemoryEmbeddingRetriever
)
from haystack.components.embedders import (
SentenceTransformersTextEmbedder,
SentenceTransformersDocumentEmbedder
)
from haystack.components.writers import DocumentWriter
from haystack.components.preprocessors import DocumentSplitter, DocumentCleaner
from haystack.components.converters import TextFileToDocument
from haystack.components.builders.prompt_builder import PromptBuilder
from haystack.components.generators import OpenAIGenerator
from haystack.components.joiners import DocumentJoiner
from haystack.components.rankers import SentenceTransformersSimilarityRanker
from haystack.components.routers import ConditionalRouter
from haystack.utils import Secret
from haystack.dataclasses import ChatMessage
# ============================================================================
# CONFIGURATION
# ============================================================================
# GROQ_API_KEY = "your-groq-api-key-here"
os.environ["GROQ_API_KEY"] = "gsk_!!!"
# Model configurations
EMBEDDING_MODEL = "sentence-transformers/all-MiniLM-L6-v2"
RERANKER_MODEL = "cross-encoder/ms-marco-MiniLM-L-6-v2"
GROQ_MODEL = "llama-3.3-70b-versatile"
# Retrieval parameters
BM25_TOP_K = 10
EMBEDDING_TOP_K = 10
RERANKER_TOP_K = 5
# ============================================================================
# READING DATA FILE
# ============================================================================
def load_documents_with_metadata(file_path: str) -> list[Document]:
documents = []
with open(file_path, "r", encoding="utf-8") as f:
for line in f:
data = json.loads(line)
documents.append(
Document(
content=data["content"],
meta=data.get("meta", {})
)
)
return documents
# ============================================================================
# COMPONENT 1: HyDE (HYPOTHETICAL DOCUMENT EMBEDDINGS)
# ============================================================================
@component
class HyDEGenerator:
"""Generate hypothetical documents for HyDE retrieval"""
def __init__(self, llm: OpenAIGenerator):
self.llm = llm
self.template = """Generate a detailed paragraph that would perfectly answer
the following question. Write as if you're providing the ideal answer from a knowledge base.
Question: {query}
Ideal Answer Paragraph:"""
@component.output_types(hypothetical_doc=str)
def run(self, query: str) -> Dict[str, str]:
"""Generate hypothetical document"""
prompt = self.template.format(query=query)
result = self.llm.run(prompt=prompt)
hypothetical_doc = result["replies"][0]
return {"hypothetical_doc": hypothetical_doc}
# ============================================================================
# COMPONENT 2: CONTEXTUAL COMPRESSOR
# ============================================================================
@component
class ContextualCompressor:
"""Extract only relevant sentences from retrieved documents"""
def __init__(self, llm: OpenAIGenerator):
self.llm = llm
self.template = """Given the following document chunk and query, extract ONLY
the sentences that are directly relevant to answering the query.
Return only the relevant sentences, nothing else.
Query: {query}
Document Chunk:
{chunk}
Relevant Sentences:"""
@component.output_types(compressed_documents=List[Document])
def run(self, query: str, documents: List[Document]) -> Dict[str, List[Document]]:
"""Compress documents by extracting relevant content"""
compressed_docs = []
for doc in documents:
prompt = self.template.format(query=query, chunk=doc.content)
result = self.llm.run(prompt=prompt)
compressed_content = result["replies"][0]
# Create new document with compressed content
compressed_doc = Document(
content=compressed_content,
meta=doc.meta,
score=doc.score if hasattr(doc, 'score') else None
)
compressed_docs.append(compressed_doc)
return {"compressed_documents": compressed_docs}
# ============================================================================
# COMPONENT 3: SELF-QUERYING ANALYZER
# ============================================================================
@component
class SelfQueryAnalyzer:
"""Analyze query and extract metadata filters automatically"""
def __init__(self, llm: OpenAIGenerator):
self.llm = llm
self.template = """Analyze the following query and extract:
1. The core semantic query (cleaned, focused version)
2. Any metadata filters that should be applied
Available metadata fields:
- category: geography, politics, economy, sports, science, culture
- year: any year (e.g., 2023, 2024)
- topic: overview, capital, gdp, cricket, space, entertainment, language
- source: any source type
Query: {query}
Respond in this exact format:
SEMANTIC_QUERY: [your semantic query here]
FILTERS: category=value,year=value (or FILTERS: none if no filters apply)"""
@component.output_types(semantic_query=str, filters=Dict[str, Any])
def run(self, query: str) -> Dict[str, Any]:
"""Analyze query and extract filters"""
prompt = self.template.format(query=query)
result = self.llm.run(prompt=prompt)
response = result["replies"][0]
# Parse response
lines = response.strip().split('\n')
semantic_query = query # default
filters = {}
for line in lines:
if line.startswith("SEMANTIC_QUERY:"):
semantic_query = line.replace("SEMANTIC_QUERY:", "").strip()
elif line.startswith("FILTERS:"):
filters_str = line.replace("FILTERS:", "").strip()
if filters_str.lower() != "none":
# Parse filters
for filter_pair in filters_str.split(','):
if '=' in filter_pair:
key, value = filter_pair.split('=')
key = key.strip()
value = value.strip()
# Try to convert year to int
if key == "year":
try:
value = int(value)
except:
pass
filters[key] = value
return {
"semantic_query": semantic_query,
"filters": filters
}
# ============================================================================
# COMPONENT 4: ANSWER QUALITY CHECKER (FOR AGENTIC ROUTING)
# ============================================================================
@component
class AnswerQualityChecker:
"""Check if answer is satisfactory or needs web search fallback"""
@component.output_types(quality_score=str, route=str)
def run(self, answer: str, query: str) -> Dict[str, str]:
"""Check answer quality"""
# Simple heuristic - in production, use an LLM
if "I don't have" in answer or "cannot answer" in answer or len(answer) < 50:
return {"quality_score": "low", "route": "web_search"}
else:
return {"quality_score": "high", "route": "final_answer"}
# ============================================================================
# INDEXING WITH METADATA
# ============================================================================
def index_documents_with_metadata(document_store, file_path):
documents = load_documents_with_metadata(file_path)
embedder = SentenceTransformersDocumentEmbedder(model=EMBEDDING_MODEL)
embedder.warm_up()
docs_with_embeddings = embedder.run(documents)
document_store.write_documents(docs_with_embeddings["documents"])
print(f" Indexed {len(documents)} documents with metadata")
@component
class ReplySelector:
"""Select the primary reply from LLM output"""
@component.output_types(answer=str)
def run(self, replies: List[str]) -> Dict[str, str]:
if not replies:
return {"answer": ""}
return {"answer": replies[0]}
# ============================================================================
# BUILD EXPERT RAG PIPELINE WITH ALL TECHNIQUES
# ============================================================================
def build_expert_rag_pipeline(document_store):
"""Build expert RAG pipeline with HyDE, compression, and agentic routing"""
# Initialize LLMs
main_llm = OpenAIGenerator(
api_key=Secret.from_env_var("GROQ_API_KEY"),
api_base_url="https://api.groq.com/openai/v1",
model=GROQ_MODEL,
generation_kwargs={"max_tokens": 512, "temperature": 0.1}
)
hyde_llm = OpenAIGenerator(
api_key=Secret.from_env_var("GROQ_API_KEY"),
api_base_url="https://api.groq.com/openai/v1",
model=GROQ_MODEL,
generation_kwargs={"max_tokens": 300, "temperature": 0.1}
)
compressor_llm = OpenAIGenerator(
api_key=Secret.from_env_var("GROQ_API_KEY"),
api_base_url="https://api.groq.com/openai/v1",
model=GROQ_MODEL,
generation_kwargs={"max_tokens": 200, "temperature": 0.1}
)
# Initialize components
pipeline = Pipeline()
pipeline.add_component("reply_selector", ReplySelector())
# Self-querying
pipeline.add_component("self_query", SelfQueryAnalyzer(main_llm))
# HyDE generation
pipeline.add_component("hyde_generator", HyDEGenerator(hyde_llm))
# Embedders
pipeline.add_component(
"text_embedder",
SentenceTransformersTextEmbedder(model=EMBEDDING_MODEL)
)
pipeline.add_component(
"hyde_embedder",
SentenceTransformersTextEmbedder(model=EMBEDDING_MODEL)
)
# Dual retrievers
pipeline.add_component(
"bm25_retriever",
InMemoryBM25Retriever(document_store=document_store, top_k=BM25_TOP_K)
)
pipeline.add_component(
"semantic_retriever",
InMemoryEmbeddingRetriever(document_store=document_store, top_k=EMBEDDING_TOP_K)
)
# Document processing
pipeline.add_component("document_joiner", DocumentJoiner())
pipeline.add_component(
"ranker",
SentenceTransformersSimilarityRanker(model=RERANKER_MODEL, top_k=RERANKER_TOP_K)
)
# Contextual compression
pipeline.add_component("compressor", ContextualCompressor(compressor_llm))
# Answer generation
answer_template = """Answer the question based on the provided context.
If the context doesn't contain enough information, say so clearly.
Context:
{% for doc in documents %}
{{ doc.content }}
{% endfor %}
Question: {{ question }}
Answer:"""
pipeline.add_component("prompt_builder", PromptBuilder(template=answer_template))
pipeline.add_component("answer_generator", main_llm)
# Quality checker for agentic routing
pipeline.add_component("quality_checker", AnswerQualityChecker())
# Connect components
# Self-querying → retrieval
pipeline.connect("self_query.semantic_query", "bm25_retriever.query")
pipeline.connect("self_query.semantic_query", "text_embedder.text")
# HyDE pathway
pipeline.connect("self_query.semantic_query", "hyde_generator.query")
pipeline.connect("hyde_generator.hypothetical_doc", "hyde_embedder.text")
# Retrievers
pipeline.connect("text_embedder.embedding", "semantic_retriever.query_embedding")
# Join and rank
pipeline.connect("bm25_retriever.documents", "document_joiner.documents")
pipeline.connect("semantic_retriever.documents", "document_joiner.documents")
pipeline.connect("document_joiner.documents", "ranker.documents")
# Compression
pipeline.connect("ranker.documents", "compressor.documents")
# Answer generation
pipeline.connect("compressor.compressed_documents", "prompt_builder.documents")
pipeline.connect("prompt_builder", "answer_generator")
# Quality checking
pipeline.connect("answer_generator.replies", "reply_selector.replies")
pipeline.connect("reply_selector.answer", "quality_checker.answer")
return pipeline
# ============================================================================
# CONVERSATION CONTEXT MANAGER (FROM PART 2)
# ============================================================================
class ConversationContext:
"""Manages conversation history"""
def __init__(self, max_history=5):
self.history = []
self.max_history = max_history
def add_exchange(self, question: str, answer: str):
self.history.append({"question": question, "answer": answer})
if len(self.history) > self.max_history:
self.history = self.history[-self.max_history:]
def get_history_text(self) -> str:
if not self.history:
return ""
return "\n".join([f"Q: {e['question']}\nA: {e['answer']}" for e in self.history])
# ============================================================================
# EXPERT QUERY PROCESSOR
# ============================================================================
def process_expert_query(
pipeline,
query: str,
context: ConversationContext,
show_details=True
):
"""Process query with full expert RAG pipeline"""
if show_details:
print(f"\n{'='*70}")
print(f" Expert RAG Processing")
print(f"{'='*70}")
print(f"\n Original Query: {query}")
# Run the pipeline
try:
result = pipeline.run(
{
"self_query": {"query": query},
"ranker": {"query": query},
"compressor": {"query": query},
"prompt_builder": {"question": query},
"quality_checker": {"query": query}
},
include_outputs_from=["reply_selector", "quality_checker", "compressor"]
)
if show_details:
# Show self-querying results
if "self_query" in result:
print(f"\n Self-Query Analysis:")
print(f" Semantic Query: {result['self_query'].get('semantic_query', 'N/A')}")
print(f" Filters: {result['self_query'].get('filters', {})}")
# Show HyDE results
if "hyde_generator" in result:
hyde_doc = result['hyde_generator']['hypothetical_doc']
print(f"\n HyDE Hypothetical Document:")
print(f" {hyde_doc[:200]}...")
# Show compression results
if "compressor" in result:
print(f"\n Contextual Compression:")
compressed_docs = result['compressor']['compressed_documents']
print(f" Compressed {len(compressed_docs)} documents")
for i, doc in enumerate(compressed_docs[:2], 1):
print(f"\n Doc {i}: {doc.content[:150]}...")
# Show quality check
if "quality_checker" in result:
quality = result['quality_checker']
print(f"\n Answer Quality: {quality.get('quality_score', 'N/A')}")
print(f" Route Decision: {quality.get('route', 'N/A')}")
# Get final answer
answer = result["reply_selector"]["answer"]
if show_details:
print(f"\n{'─'*70}")
print(f" Final Answer:")
print(f"{answer}")
print(f"{'─'*70}")
# Update context
context.add_exchange(query, answer)
return answer, result
except Exception as e:
print(f"\n Error: {e}")
return f"Error processing query: {e}", {}
# ============================================================================
# MAIN EXECUTION
# ============================================================================
def main():
"""Main execution function"""
print("="*70)
print("EXPERT RAG SYSTEM - HAYSTACK + GROQ")
print(" Part 3: HyDE + Compression + Self-Querying + Agentic RAG")
print("="*70)
# Step 1: Initialize document store
print("\nStep 1: Initializing Document Store")
print("-"*70)
document_store = InMemoryDocumentStore()
# Step 2: Index documents with metadata
print("\nStep 2: Indexing Documents with Metadata")
print("-"*70)
index_documents_with_metadata(document_store, "/content/india_info.json")
# Step 3: Build expert RAG pipeline
print("\n Step 3: Building Expert RAG Pipeline")
print("-"*70)
pipeline = build_expert_rag_pipeline(document_store)
print("Expert RAG pipeline ready with:")
# Step 4: Initialize conversation context
print("\nStep 4: Initializing Conversation Context")
print("-"*70)
conversation = ConversationContext()
print("Conversation manager ready")
# Step 5: Test queries
print("\n" + "="*70)
print("TESTING EXPERT RAG")
print("="*70)
test_questions = [
"What are ISRO's major space achievements?",
"Tell me about India's economy",
"Find information about cricket from recent years",
"What makes the space program cost-effective?", # Follow-up with context
]
for i, question in enumerate(test_questions, 1):
print(f"\n\n{'#'*70}")
print(f"TEST QUERY {i}")
print(f"{'#'*70}")
answer, result = process_expert_query(
pipeline,
question,
conversation,
show_details=True
)
if __name__ == "__main__":
main()
Casos de uso en el mundo real
Caso de uso 1: Sistema de preguntas y respuestas médicas
Caso de uso 2: Asistente para investigación legal
Caso de uso 3: Bot de soporte al cliente
La solución completa Expert RAG Stack
Layer 1: Data Ingestion
- PDF/TXT/HTML converters
- Hierarchical chunking (better than fixed-size)
- Metadata extraction
Layer 2: Storage
- Vector DB (embeddings)
- Graph DB (relationships)
- SQL DB (metadata)
Layer 3: Retrieval
- HyDE generation
- Hybrid search (BM25 + Semantic)
- Self-querying with metadata
- Multi-hop retrieval
Layer 4: Processing
- Contextual compression
- Reranking
- Deduplication
Layer 5: Generation
- Agentic orchestration
- Tool usage
- Multi-source synthesis
- Source attribution
Layer 6: Monitoring
- Latency tracking
- Quality metrics
- Cost monitoring
- Error logging