Wskazówki praktyczne: Ostateczny przewodnik techniczny po narzędziach RAG o otwartym kodzie źródłowym
Krok po kroku praktyczne wskazówki: Ostateczny przewodnik techniczny po narzędziach RAG open-source: umowy, sprawdzania oraz gotowe miejsca na kod dla zespołów wdrażających ten wzorzec.
To przewodnictwo pokazuje, jak przejść od surowców do gotowego systemu dla: Ostatecznego przewodnika technicznego po narzędziach RAG open-source: Docling, LlamaIndex, LangChain, Haystack i RAGAS. Skupia się na krokach operacyjnych, wyraźnych sprawdzeniach oraz kodzie, który można bez problemu umieścić w repozytorium, nie musząc zgadywać intencji.
Budowanie systemów generacji wzbogaconej o wyszukiwanie gotowych do użycia w produkcji w 2026 roku
Dla etapu generowania w budynkach z użyciem wzmacniania poprzez wyszukiwanie, przed zmianą kodu należy zdefiniować dane wejściowe, osobę odpowiedzialną za ten krok oraz kryteria zakończenia. Operatorzy powinni móc ponownie uruchomić ten 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 przechodzi się z środowiska demonstracyjnego do współdzielonych środowisk. Należy podawać fragmenty tekstu, które faktycznie stanowiły podstawę odpowiedzi. Bez tych odniesień operatorzy nie mogą odróżnić halucynacji od luki w indeksowaniu.
Wymagania wstępne
W fazie Wstępnych wymagań należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą 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ć, nie musząc czytać całej struktury. Należy podawać konkretne fragmenty tekstu, na których opiera się odpowiedź. Bez tych odniesień operatorzy nie będą w stanie odróżnić halucynacji od braku danych w indeksie.
1. Docling: Podstawa inteligentnej obróbki dokumentów
W fazie 1 Docling Intelligent Document należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten 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 dopracowywane 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.
Instalacja i podstawowe zasady użycia
W fazie instalacji i podstawowego użytkowania 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, testowalne jednostki zamiast rozbudowanych skryptów. Gdy dany krok zawiedzie, powód awarii powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowaną strukturę procesu. Należy podawać fragmenty tekstu, które faktycznie stanowią podstawę odpowiedzi. Bez tych odniesień operatorzy nie będą w stanie odróżnić fałszywych informacji od braków w indeksowaniu.
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)
Zaawansowana obróbka dokumentów z wykorzystaniem OCR
Dla zaawansowanej obróbki dokumentów z etapami 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 znanej punktacji kontrolnej, bez konieczności zgadywania ukrytego stanu. Traktuj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzuć ciche, częściowe ukończenie zadania. Wskazuj fragmenty tekstu, które faktycznie stanowią podstawę odpowiedzi. Bez tych odniesień operatorzy nie będą w stanie odróżnić halucynacji od luki w indeksowaniu.
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")
Integracja z innymi frameworkami
W fazie integracji z innymi frameworkami należy określić dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić ten 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ółdzielonych środowisk. Należy podawać fragmenty tekstu, które faktycznie stanowiły podstawę odpowiedzi. Bez tych odniesień operatorzy nie są w stanie odróżnić halucynacji od luki w indeksowaniu.
# 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: Architektura oparta na wyszukiwaniu
Dla etapu architektury LlamaIndex Retrieval-First należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten 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ć, nie musząc czytać całej struktury. Należy podawać fragmenty tekstu, które faktycznie stanowiły podstawę odpowiedzi. Bez tych odniesień operatorzy nie będą w stanie odróżnić halucynacji od braku danych w indeksie.
Bazowa implementacja RAG
W fazie podstawowej implementacji RAG należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy udokumentować zarówno prawidłowy przebieg działania, 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 dopiero późniejszej optymalizacji. Należy wskazać fragmenty tekstu, które faktycznie posłużyły jako podstawa odpowiedzi. Bez tych odniesień operatorzy nie będą w stanie odróżnić halucynacji od luki w indeksowaniu.
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)
Hierarchiczne dzielenie na fragmenty z automatycznym łączeniem
W etapie hierarchicznego dzielenia na fragmenty z automatycznym łączeniem należy zdefiniować dane wejściowe, osobę odpowiedzialną za ten krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić ten 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 krok się nie powiedzie, przyczyna błędu powinna wskazywać na konkretną odpowiedzialność, a nie na skomplikowany proces. Należy podawać fragmenty tekstu, które faktycznie stanowią podstawę odpowiedzi. Bez tych odniesień operatorzy nie będą w stanie odróżnić halucynacji od luki w indeksowaniu.
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)
Spersonalizowana ocena za pomocą LlamaIndex
Dla etapu Custom Evaluation with LlamaIndex należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Traktuj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nazwij poszczególne elementy, zdefiniuj kryteria sukcesu i odrzuć ciche, częściowe ukończenie zadania. Wymień fragmenty tekstu, które faktycznie stanowiły podstawę odpowiedzi. Bez tych odniesień operatorzy nie będą w stanie odróżnić halucynacji od luki w indeksowaniu.
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: Framework skupiony na procesach pracy
Dla etapu 3 LangChain Workflow-Centric Framework należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu.
Bazowy pipeline 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)
Wieloetapowy proces pracy z pamięcią
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: wyszukiwanie na poziomie produkcyjnym
Bazowe ustawienia pipeline’u
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])
Wdrożenie hybrydowego wyszukiwania
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: ramy oceny
Bazowe ustawienia oceny
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)
Szablonowa ocena z metrykami bez odniesień
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)
Integracja z istniejącymi systemami 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)
Analiza porównawcza: łączenie narzędzi w architekturach produkcyjnych
Zalecany wzorzec architektury
# Complete production architecture combining all tools
from docling.document_converter import DocumentConverter
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from langchain_core.prompts import ChatPromptTemplate
from haystack import Pipeline
from ragas import evaluate
from datasets import Dataset
class ProductionRAGSystem:
def __init__(self):
self.docling_converter = DocumentConverter()
self.llamaindex_index = None
self.langchain_chain = None
self.haystack_pipeline = None
self.evaluation_metrics = []
def preprocess_documents(self, document_paths):
"""Use Docling for intelligent document processing"""
processed_docs = []
for path in document_paths:
result = self.docling_converter.convert(path)
markdown_content = result.document.export_to_markdown()
processed_docs.append({
"content": markdown_content,
"metadata": {"source": path}
})
return processed_docs
def build_llamaindex_index(self, documents):
"""Build index using LlamaIndex for efficient retrieval"""
from llama_index.core import VectorStoreIndex
from llama_index.core.node_parser import SentenceSplitter
# Split documents
splitter = SentenceSplitter(chunk_size=512, chunk_overlap=50)
nodes = splitter.get_nodes_from_documents(documents)
# Create index
self.llamaindex_index = VectorStoreIndex(nodes)
return self.llamaindex_index
def create_langchain_chain(self, index):
"""Create LangChain chain for complex workflows"""
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
# Create retriever
retriever = index.as_retriever(similarity_top_k=3)
# Create prompt
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)
self.langchain_chain = (
{"context": retriever, "question": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)
return self.langchain_chain
def setup_haystack_pipeline(self, documents):
"""Set up Haystack pipeline for production deployment"""
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
# Initialize components
document_store = InMemoryDocumentStore()
embedder = SentenceTransformersTextEmbedder(model="sentence-transformers/all-MiniLM-L6-v2")
retriever = InMemoryEmbeddingRetriever(document_store=document_store)
generator = OpenAIGenerator(model="gpt-4o")
# 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")
# Add documents
document_store.write_documents(documents)
self.haystack_pipeline = pipeline
return pipeline
def evaluate_system(self, test_questions, test_answers=None):
"""Evaluate system using RAGAS"""
data = {
"question": test_questions,
"answer": [],
"contexts": []
}
if test_answers:
data["ground_truths"] = []
# Generate answers
for question in test_questions:
answer = self.langchain_chain.invoke(question)
contexts = self.llamaindex_index.as_retriever().invoke(question)
data["answer"].append(answer)
data["contexts"].append([ctx.text for ctx in contexts])
if test_answers:
data["ground_truths"].append([test_answers[test_questions.index(question)]])
# Create dataset
dataset = Dataset.from_dict(data)
# Evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision
metrics = [faithfulness, answer_relevancy, context_precision]
if test_answers:
from ragas.metrics import context_recall
metrics.append(context_recall)
result = evaluate(dataset, metrics=metrics)
self.evaluation_metrics.append(result)
return result
# Example usage
rag_system = ProductionRAGSystem()
# 1. Preprocess documents with Docling
processed_docs = rag_system.preprocess_documents(["doc1.pdf", "doc2.pdf"])
# 2. Build index with LlamaIndex
index = rag_system.build_llamaindex_index(processed_docs)
# 3. Create LangChain chain
chain = rag_system.create_langchain_chain(index)
# 4. Set up Haystack pipeline for production
haystack_pipeline = rag_system.setup_haystack_pipeline(processed_docs)
# 5. Evaluate system
test_questions = ["What are the system requirements?", "How does authentication work?"]
test_answers = ["The system requires 8GB RAM minimum", "Authentication uses OAuth 2.0"]
evaluation = rag_system.evaluate_system(test_questions, test_answers)
print(evaluation)
Kwestie związane z wdrożeniem i najlepsze praktyki
1. Najlepsze praktyki przetwarzania dokumentów
# 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>Optymalizacja strategii dzielenia na fragmenty
# 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. Kwestie związane z wdrożeniem produkcyjnym
# 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. Ocena i ciągłe doskonalenie
# 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)