Главная / Статьи / Практические советы: Полное техническое руководство по инструментам RAG с открытым исходным кодом

rag tools agents llm architecture typescript

Практические советы: Полное техническое руководство по инструментам RAG с открытым исходным кодом

Пошаговое руководство по практическим рекомендациям: полное техническое руководство по инструментам RAG с открытым исходным кодом: контракты, проверки и готовые блоки кода для команд, использующих эту архитектуру.

3416 слов

В этом руководстве пошагово описывается путь от сырьевых материалов до готовой к работе системы для: «Полного технического руководства по инструментам RAG с открытым исходным кодом: Docling, LlamaIndex, LangChain, Haystack и RAGAS». Основное внимание уделяется практическим шагам, четкой проверке результатов и коду, который можно просто скопировать в репозиторий без необходимости догадываться о намерениях автора.

Создание систем генерации с усилением через поиск, готовых к использованию в производственных условиях, в 2026 году

Для этапа генерации с усилением на основе поиска, готового к использованию в производстве, необходимо определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Регистрируйте время выполнения и стоимость токенов или запросов рядом с функциональными результатами. Отображение стоимости заранее предотвращает неожиданные счета при переходе с демо-среды в общедоступные среды. Указывайте те фрагменты текста, которые на самом деле легли в основу ответа. Без цитат операторы не смогут отличить галлюцинации от пробелов в индексации.

Предварительные требования

На этапе предварительных требований необходимо определить входные данные, ответственного за выполнение шага и критерии завершения перед внесением изменений в код. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Конфигурацию следует хранить отдельно от кода приложения. Файлы среды, хранилища конфиденциальных данных и флаги функций должны находиться в одном месте, чтобы операторы могли их проверять, не читая весь код. Указывайте те участки текста, которые легли в основу ответа. Без цитат операторы не смогут отличить галлюцинации от пробелов в индексации.

1. Docling: Основа интеллектуальной обработки документов

На этапе 1 Docling Intelligent Document необходимо определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Необходимо задокументировать как успешный, так и восстановительный пути работы. Повторные попытки, проверки человеком и обработка неработоспособных сообщений являются частью продукта, а не этапом последующей доработки. Указывайте те фрагменты текста, которые легли в основу ответа. Без цитат операторы не смогут отличить галлюцинации от пробелов в индексации.

Установка и базовое использование

На этапе установки и базового использования необходимо определить входные данные, ответственного за выполнение шага и критерии завершения перед внесением изменений в код. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Лучше использовать небольшие, тестируемые модули вместо обширных скриптов. При сбое шага причина должна быть связана с конкретной областью ответственности, а не с запутанной структурой обработки. Указывайте те фрагменты текста, на которых основан ответ. Без цитат операторы не смогут отличить вымысел от проблем с индексацией.

pip install docling

from docling.document_converter import DocumentConverter

# Convert a PDF document to structured Markdown
converter = DocumentConverter()
result = converter.convert("sample_document.pdf")
structured_markdown = result.document.export_to_markdown()
print(structured_markdown)

Расширенная обработка документов с использованием OCR

Для усовершенствованной обработки документов с использованием этапов необходимо заранее определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Рассматривайте этот этап как контракт между входными данными и проверенными результатами. Укажите названия результатов работы, определите критерии успеха и не допускайте безответственного частичного завершения задачи. Цитируйте те фрагменты, которые фактически легли в основу ответа. Без цитат операторы не смогут отличить галлюцинации от пробелов в индексации.

from docling.document_converter import DocumentConverter
from docling.datamodel.pipeline_options import PipelineOptions

# Configure pipeline with OCR for scanned documents
pipeline_options = PipelineOptions(
    do_ocr=True,  # Enable OCR for scanned documents
    ocr_engine="tesseract",  # Use Tesseract OCR engine
    ocr_language="eng"  # English language
)
converter = DocumentConverter(pipeline_options=pipeline_options)
result = converter.convert("scanned_document.pdf")
# Extract structured data including tables and images
document = result.document
tables = document.tables
images = document.images
print(f"Found {len(tables)} tables and {len(images)} images")

Интеграция с другими фреймворками

На этапе интеграции с другими фреймворками необходимо определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Рядом с функциональными результатами следует записывать время выполнения и стоимость токенов или запросов. Отображение стоимости заранее предотвращает неожиданные счета при переходе с демо-среды в общедоступные среды. Указывайте те участки текста, которые фактически легли в основу ответа. Без цитат операторы не смогут отличить галлюцинации от пробелов в индексации.

# Example: Integrating Docling with LangChain
from langchain_community.document_loaders import TextLoader
from docling.document_converter import DocumentConverter

def docling_to_langchain_docs(file_path):
    """Convert document using Docling and return as LangChain documents"""
    converter = DocumentConverter()
    result = converter.convert(file_path)
    content = result.document.export_to_markdown()
    # Create LangChain document
    loader = TextLoader(file_path)
    docs = loader.load()
    docs[0].page_content = content
    docs[0].metadata["source"] = file_path
    return docs
# Usage
langchain_docs = docling_to_langchain_docs("technical_manual.pdf")

2. LlamaIndex: архитектура с первичным поиском

Для этапа архитектуры LlamaIndex Retrieval-First размером 2 необходимо заранее определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Конфигурацию следует хранить отдельно от кода приложения. Файлы среды, хранилища секретов и флаги функций должны находиться в одном месте, чтобы операторы могли их проверять, не читая весь код. Указывайте те участки текста, которые фактически легли в основу ответа. Без цитат операторы не смогут отличить галлюцинации от пробелов в индексации.

Основная реализация RAG

На этапе базовой реализации RAG необходимо определить входные данные, ответственного за выполнение шага и критерии завершения перед внесением изменений в код. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Необходимо задокументировать как успешный, так и восстановительный сценарии работы. Повторные попытки, проверка человеком и обработка неработоспособных сообщений являются частью продукта, а не этапом последующей доработки. Указывайте те фрагменты текста, которые легли в основу ответа. Без цитат операторы не смогут отличить галлюцинации от пробелов в индексации.

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.llms.openai import OpenAI

# Load documents
documents = SimpleDirectoryReader("./data").load_data()
# Create index
index = VectorStoreIndex.from_documents(documents)
# Create query engine
query_engine = index.as_query_engine(
    similarity_top_k=3,
    response_mode="compact"
)
# Query the system
response = query_engine.query("What are the main features of the product?")
print(response)

Иерархическое разбиение на фрагменты с автоматическим объединением

На этапе иерархического разбиения на чанки с автоматическим объединением необходимо заранее определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не пытаясь угадать скрытое состояние. Лучше использовать небольшие, тестируемые единицы вместо обширных скриптов. При сбое шага он должен указывать на конкретную причину, а не на сложную взаимосвязь всех этапов. Указывайте те фрагменты текста, которые легли в основу ответа. Без цитат операторы не смогут отличить галлюцинации от проблем с индексацией.

from llama_index.core import VectorStoreIndex
from llama_index.core.node_parser import HierarchicalNodeParser, get_leaf_nodes
from llama_index.core.retrievers import AutoMergingRetriever
from llama_index.core.query_engine import RetrieverQueryEngine

# Create hierarchical nodes: 2048 -> 512 -> 128 token chunks
node_parser = HierarchicalNodeParser.from_defaults(
    chunk_sizes=[2048, 512, 128]
)
nodes = node_parser.get_nodes_from_documents(documents)
leaf_nodes = get_leaf_nodes(nodes)
# Build index on leaf nodes only
index = VectorStoreIndex(leaf_nodes)
index.storage_context.docstore.add_documents(nodes)
# Auto-merging retriever replaces small chunks with parent context when relevant
retriever = AutoMergingRetriever(
    index.as_retriever(similarity_top_k=3),
    index.storage_context,
    merge_batch_size=5,
)
query_engine = RetrieverQueryEngine(retriever)
response = query_engine.query("Explain the technical specifications in detail")
print(response)

Пользовательская оценка с LlamaIndex

На этапе пользовательской оценки с использованием LlamaIndex необходимо заранее определить входные данные, ответственного за выполнение шага и критерии завершения перед внесением изменений в код. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Рассматривайте этот этап как контракт между входными данными и проверенными результатами. Укажите названия создаваемых файлов, определите критерии успеха и не допускайте молчаливого частичного выполнения задачи. Указывайте конкретные фрагменты текста, на которых основан ответ. Без цитат операторы не смогут отличить галлюцинации от пробелов в индексации.

from llama_index.core.evaluation import FaithfulnessEvaluator, RelevancyEvaluator
from llama_index.core import Settings

# Initialize evaluators
faithfulness_evaluator = FaithfulnessEvaluator(llm=Settings.llm)
relevancy_evaluator = RelevancyEvaluator(llm=Settings.llm)
# Evaluate response
eval_result = faithfulness_evaluator.evaluate_response(
    query="What are the system requirements?",
    response=response,
    contexts=[node.text for node in response.source_nodes]
)
print(f"Faithfulness score: {eval_result.score}")
print(f"Relevancy score: {relevancy_evaluator.evaluate_response(query='What are the system requirements?', response=response, contexts=[node.text for node in response.source_nodes]).score}")

3. LangChain: фреймворк, ориентированный на рабочий процесс

На этапе использования фреймворка LangChain, ориентированного на рабочий процесс, необходимо заранее определить входные данные, ответственного за выполнение шага и критерии завершения перед внесением изменений в код. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии.

Базовая схема RAG

from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# Load and split documents
loader = PyPDFLoader("sample.pdf")
docs = loader.load()
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
splits = text_splitter.split_documents(docs)
# Create vector store
vectorstore = Chroma.from_documents(documents=splits, embedding=OpenAIEmbeddings())
# Create retriever
retriever = vectorstore.as_retriever()
# Create prompt template
template = """Answer the question based only on the following context:
{context}
Question: {question}
"""
prompt = ChatPromptTemplate.from_template(template)
# Create chain
llm = ChatOpenAI(model_name="gpt-4o", temperature=0)
rag_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)
# Execute
response = rag_chain.invoke("What are the key benefits mentioned in the document?")
print(response)

Многоэтапный рабочий процесс с использованием памяти

from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory

# Create chat history
class InMemoryHistory(BaseChatMessageHistory):
    def __init__(self):
        self.messages = []
    def add_user_message(self, message: str):
        self.messages.append(HumanMessage(content=message))
    def add_ai_message(self, message: str):
        self.messages.append(AIMessage(content=message))
    def clear(self):
        self.messages = []
# Store chat history
chat_histories = {}
def get_chat_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in chat_histories:
        chat_histories[session_id] = InMemoryHistory()
    return chat_histories[session_id]
# Create chain with memory
chain_with_memory = RunnableWithMessageHistory(
    rag_chain,
    get_chat_history,
    input_messages_key="question",
    history_messages_key="chat_history",
)
# Execute with memory
session_id = "user_123"
response = chain_with_memory.invoke(
    "What are the key benefits mentioned in the document?",
    config={"configurable": {"session_id": session_id}}
)
print(response)
# Follow-up question
follow_up_response = chain_with_memory.invoke(
    "Can you elaborate on the second benefit?",
    config={"configurable": {"session_id": session_id}}
)
print(follow_up_response)

4. Haystack: поисковая система производственного уровня

Основная настройка конвейера обработки

from haystack import Pipeline
from haystack.components.embedders import SentenceTransformersTextEmbedder
from haystack.components.retrievers import InMemoryBM25Retriever, InMemoryEmbeddingRetriever
from haystack.components.joiners import JoinDocuments
from haystack.components.generators import OpenAIGenerator
from haystack.components.preprocessors import DocumentCleaner, DocumentSplitter
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.utils import Secret

# Initialize components
document_store = InMemoryDocumentStore()
cleaner = DocumentCleaner()
splitter = DocumentSplitter(split_by="word", split_length=1000)
embedder = SentenceTransformersTextEmbedder(model="sentence-transformers/all-MiniLM-L6-v2")
bm25_retriever = InMemoryBM25Retriever(document_store=document_store)
embedding_retriever = InMemoryEmbeddingRetriever(document_store=document_store)
joiner = JoinDocuments(join_mode="concatenate")
generator = OpenAIGenerator(api_key=Secret.from_env_var("OPENAI_API_KEY"), model="gpt-4o")
# Create pipeline
pipeline = Pipeline()
pipeline.add_component("cleaner", cleaner)
pipeline.add_component("splitter", splitter)
pipeline.add_component("embedder", embedder)
pipeline.add_component("bm25_retriever", bm25_retriever)
pipeline.add_component("embedding_retriever", embedding_retriever)
pipeline.add_component("joiner", joiner)
pipeline.add_component("generator", generator)
# Connect components
pipeline.connect("cleaner", "splitter")
pipeline.connect("splitter", "embedder")
pipeline.connect("embedder", "embedding_retriever")
pipeline.connect("bm25_retriever", "joiner")
pipeline.connect("embedding_retriever", "joiner")
pipeline.connect("joiner", "generator")
# Add documents to store
from haystack.dataclasses import Document
documents = [
    Document(content="The system requires 8GB RAM minimum"),
    Document(content="Supports Windows, macOS, and Linux"),
    Document(content="Network bandwidth should be at least 10Mbps")
]
document_store.write_documents(documents)
# Run pipeline
result = pipeline.run({
    "cleaner": {"documents": documents},
    "bm25_retriever": {"query": "system requirements"},
    "embedding_retriever": {"query": "system requirements"}
})
print(result["generator"]["replies"][0])

Реализация гибридного поиска

from haystack import Pipeline
from haystack.components.embedders import SentenceTransformersTextEmbedder
from haystack.components.retrievers import InMemoryBM25Retriever, InMemoryEmbeddingRetriever
from haystack.components.rankers import TransformersRanker
from haystack.components.joiners import JoinDocuments
from haystack.components.generators import OpenAIGenerator
from haystack.document_stores.in_memory import InMemoryDocumentStore

# Create hybrid search pipeline
pipeline = Pipeline()
# Add components
document_store = InMemoryDocumentStore()
bm25_retriever = InMemoryBM25Retriever(document_store=document_store)
embedding_retriever = InMemoryEmbeddingRetriever(document_store=document_store)
ranker = TransformersRanker(model_name_or_path="BAAI/bge-reranker-base")
joiner = JoinDocuments(join_mode="concatenate")
generator = OpenAIGenerator(model="gpt-4o")
# Add components to pipeline
pipeline.add_component("bm25_retriever", bm25_retriever)
pipeline.add_component("embedding_retriever", embedding_retriever)
pipeline.add_component("ranker", ranker)
pipeline.add_component("joiner", joiner)
pipeline.add_component("generator", generator)
# Connect components
pipeline.connect("bm25_retriever", "ranker.query")
pipeline.connect("embedding_retriever", "ranker.documents")
pipeline.connect("ranker", "joiner")
pipeline.connect("joiner", "generator")
# Run hybrid search
result = pipeline.run({
    "bm25_retriever": {"query": "system requirements"},
    "embedding_retriever": {"query": "system requirements"}
})
print(result["generator"]["replies"][0])

5. RAGAS: платформа для оценки

Основная настройка процедуры оценки

import pandas as pd
from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
    context_relevancy,
    answer_similarity
)
from datasets import Dataset

# Create evaluation dataset
data = {
    "question": ["What are the system requirements?", "How does the authentication work?"],
    "answer": ["The system requires 8GB RAM minimum", "Authentication uses OAuth 2.0"],
    "contexts": [
        ["The system requires 8GB RAM minimum", "Supports Windows, macOS, and Linux"],
        ["Authentication uses OAuth 2.0", "Multi-factor authentication is optional"]
    ],
    "ground_truths": [
        ["The system requires 8GB RAM minimum"],
        ["Authentication uses OAuth 2.0 with JWT tokens"]
    ]
}
dataset = Dataset.from_dict(data)
# Evaluate
result = evaluate(
    dataset,
    metrics=[
        faithfulness,
        answer_relevancy,
        context_precision,
        context_recall,
        context_relevancy,
        answer_similarity
    ]
)
print(result)

Персонализированная оценка с использованием метрик без ссылок на эталоны

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
    context_relevancy,
    answer_similarity,
    answer_correctness
)
from datasets import Dataset

# Create dataset without ground truth (reference-free evaluation)
data = {
    "question": ["What are the system requirements?", "How does the authentication work?"],
    "answer": ["The system requires 8GB RAM minimum", "Authentication uses OAuth 2.0"],
    "contexts": [
        ["The system requires 8GB RAM minimum", "Supports Windows, macOS, and Linux"],
        ["Authentication uses OAuth 2.0", "Multi-factor authentication is optional"]
    ]
}
dataset = Dataset.from_dict(data)
# Evaluate without ground truth
result = evaluate(
    dataset,
    metrics=[
        faithfulness,
        answer_relevancy,
        context_precision,
        context_recall,
        context_relevancy,
        answer_similarity
    ]
)
print(result)

Интеграция с существующими системами RAG

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision
from datasets import Dataset
import json

def evaluate_rag_system(rag_system, test_questions, expected_answers=None):
    """Evaluate a RAG system against test questions"""
    results = {
        "question": [],
        "answer": [],
        "contexts": [],
        "ground_truths": [] if expected_answers else None
    }
    for i, question in enumerate(test_questions):
        # Get answer from RAG system
        answer = rag_system(question)
        # Get contexts used by RAG system (assuming it returns contexts)
        contexts = rag_system.get_contexts(question) if hasattr(rag_system, 'get_contexts') else []
        results["question"].append(question)
        results["answer"].append(answer)
        results["contexts"].append(contexts)
        if expected_answers:
            results["ground_truths"].append([expected_answers[i]])
    # Create dataset
    dataset = Dataset.from_dict(results)
    # Evaluate
    metrics = [faithfulness, answer_relevancy, context_precision]
    if expected_answers:
        metrics.append(context_recall)
    evaluation_result = evaluate(dataset, metrics=metrics)
    return evaluation_result
# Example usage
# Assuming you have a RAG system object
# evaluation = evaluate_rag_system(your_rag_system, test_questions, expected_answers)
# print(evaluation)

Сравнительный анализ: сочетание инструментов в производственных архитектурах

Аспекты реализации и лучшие практики

1. Лучшие практики обработки документов

# Best practices for document processing with Docling
from docling.document_converter import DocumentConverter
from docling.datamodel.pipeline_options import PipelineOptions

def optimize_docling_processing():
    """Optimize Docling for different document types"""
    # For scanned documents with OCR
    pipeline_options_ocr = PipelineOptions(
        do_ocr=True,
        ocr_engine="tesseract",
        ocr_language="eng"
    )
    # For clean digital documents
    pipeline_options_clean = PipelineOptions(
        do_ocr=False,
        extract_images=False  # Disable image extraction for text-only processing
    )
    # For documents with complex layouts
    pipeline_options_layout = PipelineOptions(
        do_ocr=True,
        ocr_engine="easyocr",  # More accurate but slower
        extract_tables=True,
        extract_images=True
    )
    return {
        "ocr": pipeline_options_ocr,
        "clean": pipeline_options_clean,
        "layout": pipeline_options_layout
    }
# Usage
optimizations = optimize_docling_processing()
converter = DocumentConverter(pipeline_options=optimizations["ocr"])

2. Оптимизация стратегии разбиения на части

# Advanced chunking strategies for different content types
from llama_index.core.node_parser import (
    SentenceSplitter,
    SemanticSplitterNodeParser,
    HierarchicalNodeParser
)

def create_optimal_chunking_strategy(content_type="general"):
    """Create optimal chunking strategy based on content type"""
    if content_type == "technical":
        # Technical documents need smaller chunks for precision
        return SentenceSplitter(
            chunk_size=512,
            chunk_overlap=64,
            paragraph_separator="\n\n",
            sentence_separator="\\n"
        )
    elif content_type == "legal":
        # Legal documents need to preserve entire clauses
        return SemanticSplitterNodeParser(
            buffer_size=1,
            breakpoint_percentile_threshold=95,
            embed_model="sentence-transformers/all-MiniLM-L6-v2"
        )
    elif content_type == "research":
        # Research papers benefit from hierarchical chunking
        return HierarchicalNodeParser.from_defaults(
            chunk_sizes=[2048, 512, 128]
        )
    else:
        # General purpose chunking
        return SentenceSplitter(
            chunk_size=1024,
            chunk_overlap=128
        )
# Usage
chunking_strategy = create_optimal_chunking_strategy("technical")

3. Аспекты внедрения в производство

# Production deployment configuration for Haystack
from haystack import Pipeline
from haystack.components.embedders import SentenceTransformersTextEmbedder
from haystack.components.retrievers import InMemoryEmbeddingRetriever
from haystack.components.generators import OpenAIGenerator
from haystack.document_stores.in_memory import InMemoryDocumentStore
import os

def create_production_pipeline():
    """Create production-ready Haystack pipeline"""
    # Use environment variables for sensitive data
    api_key = os.getenv("OPENAI_API_KEY")
    model_name = os.getenv("LLM_MODEL_NAME", "gpt-4o")
    # Initialize components with production settings
    document_store = InMemoryDocumentStore()
    embedder = SentenceTransformersTextEmbedder(
        model="sentence-transformers/all-MiniLM-L6-v2",
        batch_size=32  # Optimize for throughput
    )
    retriever = InMemoryEmbeddingRetriever(
        document_store=document_store,
        top_k=5  # Return more results for ranking
    )
    generator = OpenAIGenerator(
        api_key=api_key,
        model=model_name,
        max_tokens=1024,
        temperature=0.7  # Balance creativity and accuracy
    )
    # Create pipeline
    pipeline = Pipeline()
    pipeline.add_component("embedder", embedder)
    pipeline.add_component("retriever", retriever)
    pipeline.add_component("generator", generator)
    # Connect components
    pipeline.connect("embedder", "retriever")
    pipeline.connect("retriever", "generator")
    return pipeline
# Add monitoring and logging
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def monitored_pipeline_run(pipeline, query):
    """Run pipeline with monitoring"""
    logger.info(f"Processing query: {query}")
    try:
        result = pipeline.run({"retriever": {"query": query}})
        logger.info("Query processed successfully")
        return result
    except Exception as e:
        logger.error(f"Error processing query: {e}")
        raise

4. Оценка и непрерывное совершенствование

# Continuous evaluation and improvement loop
from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall
)
from datasets import Dataset
import time

class RAGEvaluationSystem:
    def __init__(self):
        self.evaluation_history = []
        self.improvement_plan = {}
    def run_evaluation_cycle(self, rag_system, test_set, previous_results=None):
        """Run evaluation cycle and generate improvement plan"""
        # Evaluate current system
        evaluation_result = self.evaluate_system(rag_system, test_set)
        # Compare with previous results
        if previous_results:
            improvement = self.calculate_improvement(evaluation_result, previous_results)
            self.generate_improvement_plan(improvement, evaluation_result)
        # Record results
        self.evaluation_history.append({
            "timestamp": time.time(),
            "results": evaluation_result,
            "improvement_plan": self.improvement_plan.copy()
        })
        return evaluation_result
    def evaluate_system(self, rag_system, test_set):
        """Evaluate RAG system"""
        data = {
            "question": test_set["questions"],
            "answer": [],
            "contexts": [],
            "ground_truths": test_set["answers"]
        }
        # Generate answers
        for question in test_set["questions"]:
            answer = rag_system(question)
            contexts = rag_system.get_contexts(question) if hasattr(rag_system, 'get_contexts') else []
            data["answer"].append(answer)
            data["contexts"].append(contexts)
        # Create dataset
        dataset = Dataset.from_dict(data)
        # Evaluate
        metrics = [
            faithfulness,
            answer_relevancy,
            context_precision,
            context_recall
        ]
        result = evaluate(dataset, metrics=metrics)
        return result
    def calculate_improvement(self, current, previous):
        """Calculate improvement between evaluation cycles"""
        improvement = {}
        for metric in current.keys():
            if metric in previous:
                improvement[metric] = current[metric] - previous[metric]
        return improvement
    def generate_improvement_plan(self, improvement, current_results):
        """Generate improvement plan based on evaluation results"""
        # Identify areas needing improvement
        if improvement.get("faithfulness", 0) < 0.1:
            self.improvement_plan["faithfulness"] = "Improve retrieval quality by adjusting chunk size or using better embedding model"
        if improvement.get("answer_relevancy", 0) < 0.1:
            self.improvement_plan["answer_relevancy"] = "Improve prompt engineering or use more sophisticated answer generation techniques"
        if improvement.get("context_precision", 0) < 0.1:
            self.improvement_plan["context_precision"] = "Implement re-ranking or hybrid search to improve context selection"
        if improvement.get("context_recall", 0) < 0.1:
            self.improvement_plan["context_recall"] = "Increase top-k parameter or implement query expansion techniques"
# Example usage
evaluator = RAGEvaluationSystem()
# Define test set
test_set = {
    "questions": ["What are the system requirements?", "How does authentication work?"],
    "answers": [["The system requires 8GB RAM minimum"], ["Authentication uses OAuth 2.0"]]
}
# Run evaluation
evaluation_result = evaluator.run_evaluation_cycle(your_rag_system, test_set)
print(evaluation_result)
print("Improvement Plan:", evaluator.improvement_plan)

Заключение

Список литературы и рекомендуемая литература

Чек-лист для работы

Все статьи Прислать бриф