Uwagi praktyczne: Część 3: Ekspert RAG — HyDE, samodzielne zadawanie pytań i przepływy pracy agentowe
Krok po kroku przewodnik po Notatkach praktycznych: Część 3: Ekspert RAG — HyDE, samodzielne zadawanie pytań i przepływy pracy agentowe: umowy, weryfikacje oraz gotowe miejsca na kod dla zespołów tworzących systemy RAG.
To przewodnik pokazuje, jak odtworzyć ścieżkę od surowców do działającego systemu w ramach: Część 3: Ekspert RAG — HyDE, samodzielne wyszukiwanie informacji i przepływy pracy oparte na agentach. Skupiamy się na krokach realizowalnych w praktyce, wyraźnych sprawdzeniach oraz kodzie, który można bez problemu dodać do repozytorium, nie musząc zgadywać intencji. Aby uzyskać ogólny obraz, zdefiniuj wprowadzenia, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Zapisuj czas trwania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega nieoczekiwanym rachunkom, gdy ścieżka przechodzi z środowiska demonstracyjnego do współdzielonych środowisk.
1. HyDE: Hipotetyczne embeddingi dokumentów
Gdy pracujesz nad rozdziałem 1. HyDE: Hypothetical Document Embeddings, najpierw zapisz specyfikację: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Przechowuj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny haseł oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całej struktury. Zmierz stopę odzyskiwania informacji na ustalonej grupie pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko poprawiają słabą skuteczność wyszukiwania.
Wnioski
Gdy pracujesz nad The Insight, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zdokumentuj razem ścieżkę prawidłowego działania oraz ścieżkę naprawczą. Próby ponownych działań, kontrola przez ludzi oraz obsługa wiadomości nieodebranych stanowią część produktu, a nie elementy dodawane później w celu udoskonalenia. Zmierz stopień przywoływania informacji na podstawie ustalonego zestawu pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko naprawiają słabe mechanizmy wyszukiwania.
Jak działa HyDE
Gdy studiujesz, jak działa HyDE, najpierw zapisz specyfikację: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Wolno preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakiś krok się nie powiedzie, błąd powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Zmierz stopień przywoływania informacji na ustalonej grupie pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko naprawiają słabe mechanizmy wyszukiwania.
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
Dlaczego HyDE działa
Gdy pracujesz nad książką „Why HyDE Works”, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Traktuj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzuć przypadkowe, częściowe ukończenie zadań. Zmierz stopień przywoływalności na ustalonej serii pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko poprawiają słabą skuteczność wyszukiwania.
Kiedy używać HyDE
Gdy pracujesz nad „When to Use HyDE”, najpierw zapisz specyfikację: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zapisz czas wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega nieoczekiwanym rachunkom, gdy ścieżka przechodzi z środowiska demonstracyjnego do współdzielonych środowisk. Zmierz dokładność odzyskiwania informacji na ustalonej grupie pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko poprawiają słabe możliwości wyszukiwania.
HyDE kontra tradycyjne RAG: rzeczywisty przykład
Gdy pracujesz nad tematem HyDE kontra tradycyjny RAG: Przykład z życia, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Trzymaj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całej struktury. Zmierz stopień przywoływania informacji na ustalonej serii pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko naprawiają słabe mechanizmy wyszukiwania.
2. Kompresja kontekstowa: precyzja ponad ilość
Gdy pracujesz nad rozdziałem 2. „Kompresja kontekstowa: precyzja ponad ilość”, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zdokumentuj razem ścieżkę prawidłowego działania oraz ścieżkę naprawczą. Próby ponownych działań, kontrola przez ludzi oraz obsługa wiadomości nieodebranych stanowią część produktu, a nie elementy dopiero późniejszej optymalizacji. Zmierz stopień odzyskiwania informacji na ustalonej grupie pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko naprawiają słabe możliwości wyszukiwania.
Problem
Gdy pracujesz nad rozwiązaniem problemu, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Wolno preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakiś krok się nie powiedzie, błąd powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany proces. Zmierz stopień przywoływania informacji na ustalonej serii pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko naprawiają słabe mechanizmy wyszukiwania. Gdy pracujesz nad rozwiązaniem problemu, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zapisz czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy przechodzi się z środowiska demonstracyjnego do wspólnych środowisk.
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
Rozwiązanie kompresji kontekstowej
Rozwiązanie kompresji kontekstowej działa najlepiej, gdy traktuje się je jako mierzalną powierzchnię. Zapisz jeden idealny przepis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres. Przechowuj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całej struktury. Oddziel zasadę dzielenia na fragmenty od zasady pobierania danych. Zmiana jednej z nich nie powinna zmuszać do przepisywania drugiej, gdy zmieniają się metryki jakości.
Query + Chunk → Compressor LLM → Relevant Sentences Only
Strategia wdrożenia
Strategia wdrożenia działa najlepiej, gdy jest traktowana jako mierzalna struktura. Zanim rozszerzysz zakres, zapisz jeden idealny przypadek działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian. Zdokumentuj zarówno prawidłowy przebieg działania, jak i ścieżkę naprawczą. Próby ponownych działań, kontrola ludzka oraz obsługa wiadomości nieodebranych stanowią część produktu, a nie elementy dodawane później. Oddziel zasady dzielenia na fragmenty od zasad wyszukiwania. Zmiana jednych nie powinna zmuszać do przepisywania drugich, gdy zmieniają się metryki jakości.
# 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
Korzyści
Benefits funkcjonuje najlepiej, gdy traktuje się je jako coś mierzalnego. Zanim rozszerzysz zakres, zapisz jeden idealny przykład działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian. Wolij małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakiś krok zawiedzie, awaria powinna wskazywać na konkretną odpowiedzialność, a nie na skomplikowany proces. Rozdziel politykę dzielenia na części od polityki pobierania danych. Zmiana jednej z nich nie powinna zmuszać do przepisywania drugiej, gdy zmieniają się metryki jakości. Benefits funkcjonuje najlepiej, gdy traktuje się je jako coś mierzalnego. Zanim rozszerzysz zakres, zapisz jeden idealny przykład działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian. Zapisuj czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy przechodzi się z środowiska demonstracyjnego do współdzielonych środowisk.
3. Samodzielne pytania: Niech LLM decyduje
Dla punktu 3. Samodzielne zadawanie pytań: Niech LLM podejmie decyzję – zdefiniuj dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Konfigurację należy przechowywać poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całej struktury. W przypadku, gdy następnym krokiem jest kod lub wywołanie narzędzia, należy preferować ustrukturyzowane wyniki z walidacją schematu zamiast tekstu w formie swobodnej.
Koncepcja
Dla tego koncepcji należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy udokumentować zarówno prawidłowy przebieg procesu, jak i ścieżkę naprawczą. Próby ponownych działań, kontrola przez ludzi oraz obsługa wiadomości błędnych stanowią część produktu, a nie elementy dodawane później. Należy podać fragmenty tekstu, które faktycznie stanowią podstawę odpowiedzi. Bez tych odniesień operatorzy nie będą w stanie odróżnić halucynacji od luki w indeksowaniu.
Tradycyjne podejście vs samodzielne wyszukiwanie
Dla podejścia tradycyjnego w porównaniu z samodzielnym wyszukiwaniem, należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy preferować małe, łatwe do przetestowania jednostki zamiast rozbudowanych skryptów. Gdy dany krok zawiedzie, powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Należy podawać fragmenty tekstu, które faktycznie stanowią podstawę odpowiedzi. Bez tych odniesień operatorzy nie będą w stanie odróżnić halucynacji od braku danych w indeksie. Dla podejścia tradycyjnego w porównaniu z samodzielnym wyszukiwaniem, należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy rejestrować czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy proces przechodzi z środowiska demonstracyjnego do wspólnych środowisk.
# 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)
Dlaczego samo-zadawanie pytań jest ważne
Gdy zajmujesz się tematem „Dlaczego samo-zadawanie pytań jest ważne”, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Trzymaj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całej struktury. Zmierz stopę przywoływania informacji na ustalonej serii pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko naprawiają słabe mechanizmy wyszukiwania.
Przykład: samo-zadawanie pytań w praktyce
Gdy pracujesz nad Przykładem: Samodzielne zadawanie pytań w praktyce, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zdokumentuj zarówno ścieżkę prawidłowego działania, jak i ścieżkę naprawczą. Próby ponownych działań, kontrola przez ludzi oraz obsługa wiadomości nieodebranych stanowią część produktu, a nie elementy dodawane później w celu udoskonalenia. Zmierz stopień odzyskiwania informacji na ustalonej grupie pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko naprawiają słabe mechanizmy wyszukiwania informacji.
{
"semantic_query": "articles about India",
"metadata_filters": {
"year": 2023,
"content_type": "article"
},
"search_type": "hybrid",
"top_k": 10
}
4. Agentic RAG: Ostateczna ewolucja
Gdy pracujesz nad rozdziałem 4. Agentic RAG: The Ultimate Evolution, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Wolij małe, testowalne jednostki od rozbudowanych skryptów. Gdy jakiś krok się nie powiedzie, błąd powinien wskazywać na jedną konkretne odpowiedzialność, a nie na skomplikowany łańcuch operacji. Zmierz stopień przywoływania informacji na ustalonej serii pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko naprawiają słabe mechanizmy wyszukiwania informacji.
Czym charakteryzuje się RAG jako „agentywny”?
Gdy pracujesz nad tematem „Co sprawia, że RAG jest ‘agentem’?”, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Traktuj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzuć przypadkowe, częściowe ukończenie zadań. Zmierz stopień przywoływania informacji na ustalonej serii pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko poprawiają słabą efektywność wyszukiwania informacji.
Architektura agentowego RAG
Gdy pracujesz nad architekturą Agentic RAG, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zapisz czas trwania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega nieoczekiwanym rachunkom, gdy przechodzi się od wersji demonstracyjnej do środowisk współdzielonych. Zmierz stopień odzyskiwania informacji na ustalonej serii pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko poprawiają słabą efektywność wyszukiwania.
┌─────────────┐
│ User Query │
└──────┬──────┘
│
┌──────▼──────┐
│ Agent (LLM) │ ← Makes decisions
└──────┬──────┘
│
┌──────────────────┼──────────────────┐
│ │ │
┌────▼────┐ ┌─────▼─────┐ ┌─────▼──────┐
│ Tool 1 │ │ Tool 2 │ │ Tool 3 │
│ (RAG) │ │(Web Search│ │(Calculator)│
└────┬────┘ └─────┬─────┘ └─────┬──────┘
│ │ │
└──────────────────┼──────────────────┘
│
┌──────▼──────┐
│ Agent │ ← Synthesizes
└──────┬──────┘
│
┌──────▼──────┐
│ Final │
│ Answer │
└─────────────┘
Proces podejmowania decyzji przez agenta
Gdy pracujesz nad procesem podejmowania decyzji przez agenta, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Przechowuj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całej struktury. Zmierz stopę odzyskiwania informacji na ustalonej serii pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko naprawiają słabe mechanizmy wyszukiwania.
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
Cechy Agentic RAG
Przykład przepływu pracy agenta
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
Korzyści z Agentic RAG
Łączenie wszystkich technik: ostateczny system RAG
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
Kiedy stosować każdą technikę
HyDE
Kompresja kontekstowa
Samozapytanie
Agentic RAG
Krok po kroku: implementacja kodu
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()
Zastosowania w praktyce
Zastosowanie 1: System pytań i odpowiedzi w medycynie
Zastosowanie 2: Asystent do badań prawnych
Zastosowanie 3: Bot obsługi klienta
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
Co dalej:
Wniosek
Zasoby
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