Notes pratiques : Partie 3 : RAG avancé — HyDE, auto-interrogation et flux de travail agents
Guide pratique pas à pas : Notes pratiques – Partie 3 : RAG expert — HyDE, requêtes autonomes et flux de travail agents : contrats, vérifications et emplacements pour du code à insérer destinés aux équipes qui développent des systèmes RAG.
Ce guide reconstitue le parcours allant des matières premières à un système fonctionnel pour : Partie 3 : RAG avancé — HyDE, requêtes autonomes et flux de travail agents. L’accent est mis sur des étapes opérationnelles, des vérifications explicites, ainsi que du code que vous pouvez intégrer directement dans un dépôt sans devoir deviner l’intention. Pour une vue d’ensemble, définissez les entrées, le responsable de l’étape et les critères de fin avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans avoir à deviner l’état caché. Enregistrez les temps d’exécution ainsi que le coût en tokens ou en requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite les factures inattendues lorsque le parcours passe d’un environnement de démonstration à des environnements partagés.
1. HyDE : Incrustations hypothétiques de documents
Lorsque vous travaillez sur 1. HyDE : Hypothetical Document Embeddings, notez d’abord les exigences : entrées requises, signal de succès, et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les administrateurs peuvent auditer sans devoir lire l’ensemble du système. Mesurez le taux de rappel sur un ensemble de questions fixe avant d’ajuster les prompts. Changer fréquemment les prompts ne résout que rarement un système de récupération insuffisant.
La réflexion
Lorsque vous travaillez sur The Insight, notez d’abord le contrat : les entrées requises, le signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Documentez ensemble le parcours normal et le parcours de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non traités font partie du produit, et non d’améliorations apportées ultérieurement. Mesurez le taux de rappel sur un ensemble fixe de questions avant d’ajuster les prompts. Un changement fréquent des prompts ne résout que rarement un système de récupération insuffisant.
Comment fonctionne HyDE
Lorsque vous travaillez sur « How HyDE Works », notez d’abord le contrat : les entrées requises, le signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Préférez des unités petites et testables à des scripts volumineux. Lorsqu’une étape échoue, l’échec doit pointer vers une seule responsabilité plutôt que vers un processus complexe et embrouillé. Évaluez le taux de rappel sur un ensemble de questions fixe avant d’ajuster les prompts. Un changement fréquent des prompts ne résout que rarement un système de récupération insuffisant.
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
Pourquoi HyDE fonctionne
Lorsque vous travaillez sur « Why HyDE Works », notez d’abord les conditions prévues : les entrées requises, le signal de succès, ainsi que ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Donnez des noms aux éléments générés, définez des vérifications de succès, et refusez les terminaisons partielles silencieuses. Évaluez le taux de rappel sur un ensemble de questions fixe avant d’ajuster les prompts. Le simple changement de prompts ne résout que rarement un système de récupération insuffisant.
Quand utiliser HyDE
Lorsque vous travaillez sur le guide « Quand utiliser HyDE », notez d’abord les éléments requis : les entrées nécessaires, le signal de succès, ainsi que ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Enregistrez les temps d’exécution ainsi que le coût des tokens ou des requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite les factures inattendues lorsque le système passe de l’environnement de démonstration à des environnements partagés. Mesurez le taux de rappel sur un ensemble fixe de questions avant d’ajuster les prompts. Changer fréquemment les prompts ne résout que rarement un système de récupération insuffisant.
HyDE contre les systèmes RAG traditionnels : exemple concret
Lorsque vous travaillez sur « HyDE vs Traditional RAG: Real Example », notez d’abord les exigences : entrées requises, signal de succès, et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Gardez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les administrateurs peuvent auditer sans devoir lire l’ensemble du système. Mesurez le taux de rappel sur un ensemble de questions fixe avant d’ajuster les prompts. Changer fréquemment les prompts ne résout que rarement un système de récupération insuffisant.
2. Compression contextuelle : précision plutôt que quantité
Lorsque vous travaillez sur la section 2. Compression contextuelle : précision avant quantité, notez d’abord les exigences : entrées requises, signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Documentez ensemble le parcours normal et le parcours de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non traités font partie intégrante du produit, et non d’améliorations apportées ultérieurement. Mesurez le taux de rappel sur un ensemble fixe de questions avant d’ajuster les prompts. Le remplacement des prompts résout rarement un système de récupération insuffisant.
Le problème
Lorsque vous travaillez sur le problème, notez d’abord les spécifications : entrées requises, signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Préférez des unités petites et testables plutôt que des scripts complexes. Lorsqu’une étape échoue, l’échec doit indiquer une seule responsabilité et non un processus embrouillé. Mesurez le taux de rappel sur un ensemble de questions fixe avant d’ajuster les prompts. Un changement fréquent des prompts ne résout que rarement un système de récupération insuffisant. Lorsque vous travaillez sur le problème, notez d’abord les spécifications : entrées requises, signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Enregistrez les temps d’exécution ainsi que le coût en tokens ou requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite les factures inattendues lorsque le système passe de l’environnement de démonstration à des environnements partagés.
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
Solution de compression contextuelle
La solution de compression contextuelle fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un transcript idéal, un cas d’échec et la note de réversion avant d’élargir le périmètre. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans devoir lire l’ensemble du système. Séparez la politique de segmentation des données de la politique de récupération. Modifier l’une ne doit pas obliger à réécrire l’autre lorsque les métriques de qualité évoluent.
Query + Chunk → Compressor LLM → Relevant Sentences Only
Stratégie de mise en œuvre
La stratégie de mise en œuvre fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un transcript idéal, un cas d’échec et la note de réversion avant d’élargir le périmètre. Documentez ensemble le parcours optimal et le parcours de récupération. Les tentatives répétées, les contrôles humains et le traitement des messages non livrés font partie intégrante du produit, et non d’une mise en forme ultérieure. Séparez la politique de segmentation de la politique de récupération : modifier l’une ne doit pas obliger à réécrire l’autre lorsque les métriques de qualité évoluent.
# 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
Avantages
Benefits fonctionne le mieux lorsqu’il est considéré comme une surface mesurable. Capturez un exemple idéal, un cas d’échec et la note de réversion avant d’élargir le périmètre. Préférez des unités petites et testables aux scripts complexes. Lorsqu’une étape échoue, l’échec doit pointer vers une seule responsabilité plutôt que vers un processus embrouillé. Séparez la politique de segmentation de la politique de récupération. Modifier l’une ne doit pas obliger à réécrire l’autre lorsque les métriques de qualité évoluent. Benefits fonctionne le mieux lorsqu’il est considéré comme une surface mesurable. Capturez un exemple idéal, un cas d’échec et la note de réversion avant d’élargir le périmètre. Enregistrez les temps d’exécution ainsi que le coût des tokens ou des requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite les factures inattendues lorsque le processus passe de l’environnement de démonstration à des environnements partagés.
3. Auto-interrogation : Laissez l’LLM décider
Pour 3. Auto-questionnement : Laissez le LLM décider, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier du code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans avoir à lire l’ensemble du graphe. Préférez des sorties structurées avec validation de schéma plutôt que du texte libre lorsque l’étape suivante consiste en du code ou une appel à outil.
Le concept
Pour le concept, définissez les entrées, le responsable de l’étape et les critères de fin avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Documentez ensemble le parcours idéal et le parcours de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non traités font partie du produit, et non d’une mise en forme ultérieure. Citez les passages qui ont réellement servi de base à la réponse. Sans citations, les opérateurs ne peuvent pas distinguer une hallucination d’un manque d’indexation.
Approche traditionnelle vs Approche auto-questionnement
Pour les approches traditionnelles par rapport aux approches d’autrepartie, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Préférez des unités petites et testables aux scripts complexes. Lorsqu’une étape échoue, l’échec doit indiquer une seule responsabilité plutôt qu’un processus embrouillé. Citez les passages qui ont réellement servi de base à la réponse. Sans citations, les opérateurs ne peuvent pas distinguer une hallucination d’un manque d’indexation. Pour les approches traditionnelles par rapport aux approches d’autrepartie, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Enregistrez les temps d’exécution ainsi que le coût en tokens ou en requêtes à côté des résultats fonctionnels. Une visibilité précoce du coût évite des factures inattendues lorsque le processus passe de l’environnement de démonstration à des environnements partagés.
# 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)
Pourquoi l’auto-questionnement est important
Lorsque vous travaillez sur le sujet « Pourquoi l’auto-questionnement est important », notez d’abord les éléments requis : les entrées nécessaires, le signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de garantir l’honnêteté des modifications ultérieures du code. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les administrateurs peuvent auditer sans devoir lire l’ensemble du système. Mesurez le taux de rappel sur un ensemble fixe de questions avant d’ajuster les prompts. Un changement fréquent des prompts ne résout que rarement un système de récupération insuffisant.
Exemple : L’auto-questionnement en pratique
Lorsque vous travaillez sur l’Exemple : L’autovérification en pratique, notez d’abord les exigences : les entrées requises, le signal de succès, ainsi que ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Documentez ensemble le parcours normal et les procédures de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non traités font partie intégrante du produit, et non d’améliorations apportées ultérieurement. Évaluez le taux de rappel sur un ensemble fixe de questions avant d’ajuster les prompts. Le remplacement des prompts résout rarement un système de récupération insuffisant.
{
"semantic_query": "articles about India",
"metadata_filters": {
"year": 2023,
"content_type": "article"
},
"search_type": "hybrid",
"top_k": 10
}
4. Agentic RAG : L’évolution ultime
Lorsque vous travaillez sur le chapitre 4, « Agentic RAG : L’évolution ultime », notez d’abord les éléments essentiels : les entrées requises, le signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Préférez des unités petites et testables plutôt que des scripts complexes. Lorsqu’une étape échoue, l’échec doit indiquer une seule responsabilité et non un processus embrouillé. Évaluez le taux de rappel sur un ensemble de questions fixe avant d’ajuster les prompts. Changer fréquemment les prompts ne résout que rarement un système de récupération insuffisant.
Qu’est-ce qui rend RAG « agent » ?
Lorsque vous travaillez sur le sujet « Qu’est-ce qui rend RAG “agent” ? », notez d’abord les conditions du contrat : entrées requises, signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Donnez des noms aux artefacts, définez des vérifications de succès et refusez les terminaisons partielles silencieuses. Mesurez le taux de rappel sur un ensemble fixe de questions avant d’ajuster les prompts. Le simple changement de prompts ne résout que rarement un système de récupération insuffisant.
Architecture RAG agente
Lorsque vous travaillez sur l’architecture Agentic RAG, notez d’abord le contrat : les entrées requises, le signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Enregistrez les temps d’exécution ainsi que le coût en tokens ou en requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite les factures inattendues lorsque le système passe de l’environnement de démonstration à des environnements partagés. Mesurez le taux de rappel sur un ensemble fixe de questions avant d’ajuster les prompts. Changer fréquemment les prompts ne résout que rarement un système de récupération insuffisant.
┌─────────────┐
│ User Query │
└──────┬──────┘
│
┌──────▼──────┐
│ Agent (LLM) │ ← Makes decisions
└──────┬──────┘
│
┌──────────────────┼──────────────────┐
│ │ │
┌────▼────┐ ┌─────▼─────┐ ┌─────▼──────┐
│ Tool 1 │ │ Tool 2 │ │ Tool 3 │
│ (RAG) │ │(Web Search│ │(Calculator)│
└────┬────┘ └─────┬─────┘ └─────┬──────┘
│ │ │
└──────────────────┼──────────────────┘
│
┌──────▼──────┐
│ Agent │ ← Synthesizes
└──────┬──────┘
│
┌──────▼──────┐
│ Final │
│ Answer │
└─────────────┘
Processus de prise de décision de l’agent
Lorsque vous travaillez sur le processus de prise de décision de l’agent, notez d’abord les éléments requis : les données d’entrée nécessaires, le signal de succès, ainsi que ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de garantir l’intégrité des modifications ultérieures du code. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans avoir à lire l’ensemble du système. Mesurez le taux de rappel sur un ensemble fixe de questions avant d’ajuster les prompts. Un changement fréquent des prompts ne résout que rarement un système de récupération insuffisant.
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
Fonctionnalités de l’Agentic RAG
Exemple de flux de travail agent
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
Avantages de l’Agentic RAG
Combinaison de toutes les techniques : le système RAG ultime
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
Quand utiliser chaque technique
HyDE
Compression contextuelle
Auto-interrogation
Agentic RAG
Mise en œuvre du code étape par étape
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()
Cas d’usage dans le monde réel
Cas d’usage 1 : Système de questions-réponses médicales
Cas d’usage 2 : Assistant de recherche juridique
Cas d’usage 3 : Bot de support client
La pile complète Expert RAG
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