Практические заметки: Часть 3: Экспертные RAG — HyDE, самостоятельный поиск информации и агентные рабочие процессы
Пошаговое руководство по практическим заметкам: Часть 3 — Экспертные RAG-системы: HyDE, самостоятельные запросы и агентные рабочие процессы: контракты, проверки и готовые блоки кода для команд, разрабатывающих RAG-системы.
В этом руководстве показано, как пройти путь от сырья до рабочей системы для раздела 3: Экспертные RAG — HyDE, самостоятельные запросы и агентные рабочие процессы. Основное внимание уделяется практическим шагам, четким проверкам и коду, который можно просто добавить в репозиторий без необходимости угадывать его назначение. Для получения обзора определите входные данные, ответственного за шаг и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не пытаясь угадать скрытое состояние. Записывайте время выполнения и стоимость токенов или запросов рядом с функциональными результатами. Отображение стоимости заранее помогает избежать неожиданных счетов при переходе от демо-среды к общедоступным средам.
1. HyDE: Гипотетические вкладки документов
При работе над разделом 1. HyDE: Hypothetical Document Embeddings сначала запишите условия работы алгоритма: необходимые входные данные, сигнал о успешном выполнении и последствия частичной неудачи. Такой чек-лист поможет сохранять честность при последующих изменениях кода. Храните конфигурацию вне кода приложения. Файлы с настройками окружения, хранилища секретов и флаги функций должны находиться в одном месте, чтобы операторы могли их проверять, не читая весь код. Измеряйте показатель воспроизведения на фиксированном наборе вопросов перед настройкой подсказок. Частая смена подсказок редко помогает улучшить качество поиска.
Вывод
При работе над проектом The Insight сначала запишите условия работы: необходимые входные данные, сигнал успешного выполнения и действия при частичной неудаче. Такой чек-лист поможет сохранять честность при последующих изменениях кода. Документируйте одновременно успешный и восстановительный сценарии работы. Повторные попытки, проверки человеком и обработка неработоспособных сообщений являются частью продукта, а не элементами последующей доработки. Измеряйте уровень воспроизводимости ответов на фиксированном наборе вопросов перед настройкой подсказок. Частая смена подсказок редко помогает улучшить качество поиска.
Как работает HyDE
При изучении того, как работает HyDE, сначала запишите условия работы алгоритма: необходимые входные данные, сигнал о успешном выполнении и что происходит при частичной неудаче. Такой список поможет сохранять честность при последующих изменениях кода. Предпочитайте небольшие, тестируемые модули большим скриптам. Когда какой-то шаг не срабатывает, причина должна быть связана с конкретной функцией, а не с запутанной цепочкой операций. Оцените уровень воспроизведения результатов на фиксированном наборе вопросов перед настройкой подсказок. Частая смена подсказок редко помогает улучшить качество поиска.
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
Почему HyDE работает
При работе над материалом «Почему HyDE работает» сначала запишите условия работы алгоритма: необходимые входные данные, сигнал о успешном выполнении и последствия частичной неудачи. Такой список поможет сохранять честность при последующих изменениях кода. Рассматривайте этот этап как договор между входными данными и проверенными выходными результатами. Дайте названия соответствующим элементам, определите критерии успешности и не допускайте молчаливого частичного выполнения задачи. Оцените уровень воспроизводимости на фиксированном наборе вопросов перед настройкой подсказок. Частая смена подсказок редко помогает улучшить качество поиска.
Когда использовать HyDE
При работе над руководством «Когда использовать HyDE» сначала запишите условия работы алгоритма: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой чек-лист поможет избежать ошибок при последующих изменениях кода. Рядом с функциональными результатами занесите информацию о времени выполнения и стоимости токенов или запросов. Отслеживание затрат с самого начала предотвращает неожиданные расходы при переходе с демо-среды в общедоступные среды. Перед настройкой подсказок измерьте уровень воспроизводимости ответов на фиксированный набор вопросов. Частая смена подсказок редко помогает улучшить качество поиска информации.
HyDE против традиционных решений типа RAG: реальный пример
При работе над проектом «HyDE против традиционного RAG: реальный пример» сначала определите условия работы: необходимые входные данные, сигнал о успешном выполнении и что происходит при частичной неудаче. Такой чек-лист помогает сохранять честность при последующих изменениях кода. Храните конфигурацию вне кода приложения. Файлы с настройками окружения, хранилища секретов и флаги функций должны находиться в одном месте, чтобы операторы могли их проверять, не читая весь код. Измеряйте показатель воспроизведения информации на фиксированном наборе вопросов перед настройкой подсказок. Частая замена подсказок редко помогает улучшить качество поиска.
2. Компрессия контекста: точность важнее количества
При работе над разделом 2. «Контекстуальное сжатие: точность важнее количества» сначала запишите условия работы: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой список поможет сохранять честность при последующих изменениях кода. Документируйте одновременно успешный и восстановительный сценарии работы. Повторные попытки, проверки человеком и обработка неработоспособных сообщений являются частью продукта, а не элементами последующей доработки. Перед настройкой запросов измерьте уровень воспроизведения информации на фиксированном наборе вопросов. Частая смена запросов редко помогает улучшить качество поиска.
Проблема
При решении задачи сначала запишите условия работы: необходимые входные данные, сигнал о успешном выполнении и последствия частичной неудачи. Такой список помогает сохранять честность при последующих изменениях кода. Предпочитайте небольшие, тестируемые модули большим скриптам. Если какой-то шаг не сработает, причина должна быть связана с конкретной функцией, а не с запутанной цепочкой операций. Оцените уровень воспроизводимости на фиксированном наборе вопросов перед настройкой подсказок. Частая смена подсказок редко помогает улучшить качество поиска. При решении задачи сначала запишите условия работы: необходимые входные данные, сигнал о успешном выполнении и последствия частичной неудачи. Такой список помогает сохранять честность при последующих изменениях кода. Рядом с функциональными результатами записывайте время выполнения и стоимость в токенах или запросах. Отслеживание затрат с самого начала предотвращает неожиданные расходы при переходе с демо-среды в общедоступные среды.
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
Решение проблемы контекстной компрессии
Решение для контекстной компрессии работает наилучшим образом, когда рассматривается как измеримая поверхность. Соберите один идеальный пример транскрипции, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объёма работы. Храните конфигурацию вне кода приложения. Файлы среды, хранилища секретов и флаги функций должны находиться в одном месте, чтобы операторы могли их проверять, не читая весь структурный анализ. Разделяйте политику разбиения на части и политику извлечения данных. Изменение одной из них не должно приводить к переписыванию другой при изменении показателей качества.
Query + Chunk → Compressor LLM → Relevant Sentences Only
Стратегия реализации
Стратегия реализации работает наилучшим образом, когда рассматривается как измеримая структура. Соберите один идеальный пример работы, один случай сбоя и запись о возврате к предыдущему состоянию перед расширением объёма работ. Документируйте одновременно успешный сценарий выполнения и сценарий восстановления. Повторные попытки, проверки человеком и обработка неработающих сообщений являются частью продукта, а не этапом последующей доработки. Разделяйте политику разбиения на части и политику извлечения данных. Изменение одной из них не должно приводить к переписыванию другой при изменении показателей качества.
# 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
Преимущества
Подход Benefits работает наилучшим образом, когда его рассматривают как измеримую сферу. Соберите один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объёма работ. Предпочитайте небольшие, тестируемые единицы вместо обширных скриптов. Когда какой-то шаг сбивается, причина сбоя должна указывать на конкретную ответственность, а не на запутанную цепочку операций. Разделяйте политику разбиения на части и политику извлечения данных. Изменение одной из них не должно вынуждать переписывать другую при изменении показателей качества. Подход Benefits работает наилучшим образом, когда его рассматривают как измеримую сферу. Соберите один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объёма работ. Записывайте время выполнения и стоимость токенов или запросов рядом с функциональными результатами. Отслеживание затрат на раннем этапе предотвращает неожиданные расходы при переходе с демо-среды в общедоступные среды.
3. Автономный поиск: пусть решает LLM
3. Автономный запрос: пусть большая языковая модель принимает решение, определив входные данные, ответственного за шаг и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Храните конфигурацию вне кода приложения. Файлы среды, хранилища секретов и флаги функций должны находиться в одном месте, которое операторы могут проверять, не читая весь граф. Предпочитайте структурированные выходные данные с проверкой схемы вместо свободного текста, когда следующим шагом является написание кода или вызов инструмента.
Концепция
Для формулировки концепции необходимо заранее определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Необходимо одновременно задокументировать успешный сценарий выполнения и сценарий восстановления. Повторные попытки, проверки человеком и обработка неработоспособных сообщений являются частью продукта, а не этапом последующей доработки. Указывайте конкретные фрагменты текста, на которых основан ответ; без цитат операторы не смогут отличить галлюцинации от проблем с индексацией.
Традиционный подход vs самостоятельные запросы
При сравнении традиционного и самостоятельного способов выполнения запросов необходимо заранее определить входные данные, ответственного за шаг и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Лучше использовать небольшие, тестируемые единицы кода вместо обширных скриптов. При сбое шага причина должна быть связана с конкретной функцией, а не с запутанной структурой обработки. Указывайте те части текста, которые легли в основу ответа. Без цитат операторы не смогут отличить галлюцинации от проблем с индексацией. При сравнении традиционного и самостоятельного способов выполнения запросов необходимо заранее определить входные данные, ответственного за шаг и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Записывайте время выполнения, а также стоимость токенов или запросов рядом с функциональными результатами. Отображение стоимости заранее помогает избежать неожиданных расходов при переходе с демо-среды в общедоступные среды.
# 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)
Почему важно самостоятельное запросирование
При работе над темой «Почему важно самостоятельное запросирование» сначала запишите условия работы: необходимые входные данные, сигнал о успешном выполнении и последствия частичной неудачи. Такой список помогает сохранять честность при последующих изменениях кода. Храните конфигурацию вне кода приложения. Файлы с настройками окружения, хранилища секретов и флаги функций должны находиться в одном месте, чтобы операторы могли их проверять, не читая весь код. Измеряйте уровень воспроизведения ответов на фиксированном наборе вопросов перед настройкой подсказок. Частая смена подсказок редко помогает улучшить качество поиска.
Пример: самостоятельное запросирование в действии
При работе над примером «Самостоятельные запросы на практике» сначала запишите условия работы: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой список поможет сохранять честность при последующих изменениях кода. Документируйте одновременно успешный и восстановительный сценарии работы. Повторные попытки, проверки человеком и обработка неработоспособных сообщений являются частью продукта, а не элементами последующей доработки. Измеряйте уровень воспроизведения ответов на фиксированном наборе вопросов перед настройкой подсказок. Частая смена подсказок редко помогает улучшить качество поиска.
{
"semantic_query": "articles about India",
"metadata_filters": {
"year": 2023,
"content_type": "article"
},
"search_type": "hybrid",
"top_k": 10
}
4. Агентные RAG: конечная стадия развития
При работе над разделом 4. «Агентный RAG: последняя эволюция» сначала запишите условия работы: необходимые входные данные, сигнал успешного выполнения и что происходит при частичной неудаче. Такой чек-лист поможет сохранять честность при последующих изменениях кода. Лучше использовать небольшие, тестируемые единицы кода вместо обширных скриптов. Когда какой-то шаг не срабатывает, причина должна быть связана с конкретной функцией, а не с запутанной цепочкой операций. Оцените уровень воспроизведения информации на фиксированном наборе вопросов перед настройкой подсказок. Частая смена подсказок редко помогает улучшить качество поиска информации.
Что делает RAG «агентным»?
При работе над темой «Что делает RAG „агентным“?» сначала запишите условия соглашения: необходимые входные данные, сигнал о успешном выполнении и что происходит при частичной неудаче. Такой список помогает избегать ошибок при последующих изменениях кода. Рассматривайте этот этап как соглашение между входными данными и проверенными результатами. Дайте названия элементам, определите критерии успеха и не допускайте молчаливого частичного выполнения задачи. Оцените уровень воспроизводимости ответов на фиксированном наборе вопросов перед настройкой подсказок. Частая смена подсказок редко помогает улучшить качество поиска информации.
Архитектура агентного RAG
При работе над архитектурой Agentic RAG сначала запишите условия работы: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой список помогает сохранять честность при последующих изменениях кода. Рядом с функциональными результатами записывайте время выполнения и стоимость токенов или запросов. Отслеживание затрат с самого начала предотвращает неожиданные расходы при переходе с демо-среды в общедоступные среды. Измеряйте уровень воспроизводимости ответов на фиксированный набор вопросов перед настройкой подсказок. Частая смена подсказок редко помогает улучшить качество поиска информации.
┌─────────────┐
│ User Query │
└──────┬──────┘
│
┌──────▼──────┐
│ Agent (LLM) │ ← Makes decisions
└──────┬──────┘
│
┌──────────────────┼──────────────────┐
│ │ │
┌────▼────┐ ┌─────▼─────┐ ┌─────▼──────┐
│ Tool 1 │ │ Tool 2 │ │ Tool 3 │
│ (RAG) │ │(Web Search│ │(Calculator)│
└────┬────┘ └─────┬─────┘ └─────┬──────┘
│ │ │
└──────────────────┼──────────────────┘
│
┌──────▼──────┐
│ Agent │ ← Synthesizes
└──────┬──────┘
│
┌──────▼──────┐
│ Final │
│ Answer │
└─────────────┘
Процесс принятия решений агентом
При работе над процессом принятия решений агентом сначала запишите условия контракта: необходимые входные данные, сигнал успешного выполнения и действия при частичной неудаче. Такой чек-лист помогает сохранять честность при последующих изменениях кода. Храните конфигурацию вне кода приложения. Файлы среды, хранилища секретов и флаги функций должны находиться в одном месте, чтобы операторы могли их проверять, не читая весь код. Оцените уровень воспроизводимости на фиксированном наборе вопросов перед настройкой подсказок. Изменение подсказок редко помогает улучшить слабую систему поиска информации.
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
Особенности агентного RAG
Пример рабочего процесса агента
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
Преимущества агентного RAG
Сочетание всех техник: идеальная система 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
Когда использовать каждую технику
HyDE
Компрессия контекста
Самостоятельный поиск информации
Агентный RAG
Пошаговая реализация кода
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()
Практические применения в реальном мире
Сценарий использования 1: Система вопросов и ответов в медицине
Сценарий использования 2: Помощник для юридических исследований
Сценарий использования 3: Бот для обслуживания клиентов
Полная стек-сборка 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