Praktische Hinweise: Der umfassende technische Leitfaden zu Open-Source-RAG-Tools
Schritt-für-Schritt-Anleitung zu den Praktischen Notizen: Der umfassende technische Leitfaden für Open-Source-RAG-Tools – Verträge, Überprüfungen sowie Code-Slots für Teams, die dieses Muster einsetzen.
Dieser Leitfaden zeigt Schritt für Schritt den Weg von Rohstoffen bis zu einem funktionsfähigen System für: Der ultimative technische Leitfaden zu Open-Source RAG-Tools: Docling, LlamaIndex, LangChain, Haystack & RAGAS. Der Schwerpunkt liegt auf umsetzbaren Schritten, klaren Überprüfungen sowie Code, den man ohne Rückschluss auf die Absicht direkt in ein Repository einfügen kann.
Erstellung von für die Produktion bereiten Retrieval-Augmented Generation-Systemen im Jahr 2026
Für die für die Gebäudeproduktion bereite Phase der retrieval-augmented Generation müssen vor dem Ändern des Codes die Eingaben, der Verantwortliche für diesen Schritt sowie die Abbruchkriterien definiert werden. Die Bediener sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Erfassen Sie die Laufzeiten sowie die Kosten für Token oder Abfragen zusammen mit den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Weg von einer Demo-Umgebung in gemeinsam genutzte Umgebungen wechselt. Zitieren Sie die Passagen, die tatsächlich der Grundlage für die Antwort waren. Ohne Zitate können die Bediener nicht zwischen Halluzinationen und Lücken im Indexing unterscheiden.
Voraussetzungen
In der Voraussetzungsphase sollten Eingabedaten, Verantwortliche für die jeweiligen Schritte sowie Abbruchkriterien definiert werden, bevor Code geändert wird. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Die Konfiguration sollte außerhalb des Anwendungscode gespeichert werden. Umgebungsdateien, Geheimdatenspeicher sowie Feature-Flags sollten an einem Ort zusammengefasst sein, den die Operator überprüfen können, ohne den gesamten Codeverlauf durchlesen zu müssen. Zitieren Sie die Passagen, auf denen die Antwort tatsächlich beruht. Ohne Zitate können die Operator nicht zwischen Halluzinationen und Lücken in der Indizierung unterscheiden.
1. Docling: Grundlage für intelligente Dokumentverarbeitung
Zur Phase 1 von Docling Intelligent Document sollten die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien vor dem Codeändern definiert werden. Die Bediener sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Dokumentieren Sie sowohl den erfolgreichen Ablauf als auch den Notfallweg gemeinsam. Wiederholungsversuche, menschliche Überprüfungen sowie die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen. Zitieren Sie die Passagen, die tatsächlich die Antwort begründen. Ohne Zitate können die Bediener nicht zwischen Halluzinationen und Lücken in der Indizierung unterscheiden.
Installation und Grundlegende Verwendung
Zur Phase der Installation und grundlegenden Nutzung sollten Eingabedaten, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden, bevor Code geändert wird. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckten Zuständen schließen zu müssen. Es sind kleinere, testbare Einheiten vorzuziehen statt umfangreicher Skripte. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Ablaufverfahren. Zitieren Sie die Passagen, die tatsächlich die Antwort untermauern. Ohne Zitate können die Operator nicht zwischen Halluzinationen und Lücken in der Indizierung unterscheiden.
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)
Fortgeschrittene Dokumentverarbeitung mit OCR
Für die fortgeschrittene Dokumentverarbeitung in Phasen sollten Eingaben, der Verantwortliche für die jeweilige Schritt und die Abbruchkriterien vor dem Ändern des Codes definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Betrachten Sie diese Phase als Vertrag zwischen den Eingaben und den validierten Ausgaben. Benennen Sie die Erzeugnisse, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab. Zitieren Sie die Passagen, die tatsächlich die Antwort begründen. Ohne Zitate können die Operator nicht zwischen Halluzinationen und Lücken in der Indizierung unterscheiden.
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")
Integration mit anderen Frameworks
In der Phase der Integration mit anderen Frameworks sollten vor dem Ändern des Codes die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckten Zuständen schließen zu müssen. Erfassen Sie die Laufzeiten sowie die Kosten für Token oder Abfragen zusammen mit den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Prozess von einer Demo-Umgebung in gemeinsam genutzte Umgebungen übergeht. Zitieren Sie die Passagen, die tatsächlich die Antwort untermauern. Ohne Zitate können die Operator nicht zwischen Halluzinationen und Lücken in der Indizierung unterscheiden.
# 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: Retrieval-First-Architektur
Für die Phase der 2-LlamaIndex Retrieval-First-Architektur sollten Eingaben, Verantwortliche für die jeweiligen Schritte sowie Abbruchkriterien vor dem Ändern des Codes definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Die Konfiguration sollte außerhalb des Anwendungscode gespeichert werden. Umgebungsdateien, Geheimdatenspeicher sowie Feature-Flags sollten an einem Ort zusammengefasst sein, den die Operator überprüfen können, ohne den gesamten Codeverlauf durchlesen zu müssen. Zitieren Sie die Passagen, die tatsächlich der Grundlage für die Antwort waren. Ohne Zitate können die Operator nicht zwischen Halluzinationen und Lücken im Indexing unterscheiden.
Basische RAG-Implementierung
Zur Grundlegenden RAG-Implementierungsphase sollten die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden, bevor der Code geändert wird. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Dokumentieren Sie gemeinsam den erfolgreichen Ablauf sowie den Notfallweg. Wiederholungsversuche, menschliche Überprüfungen und die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen. Zitieren Sie die Passagen, die tatsächlich die Grundlage für die Antwort lieferten. Ohne Zitate können die Operator keine Halluzinationen von Lücken in der Indizierung unterscheiden.
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)
Hierarchisches Chunking mit automatischem Zusammenführen
Zur Phase des hierarchischen Chunkings mit automatischem Zusammenführen sollten die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien vor dem Ändern des Codes definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Es sollten kleine, testbare Einheiten vorzugsweise gegenüber umfangreichen Skripten verwendet werden. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Ablaufverfahren. Zitieren Sie die Passagen, die tatsächlich die Antwort untermauern. Ohne Zitate können die Operator nicht zwischen Halluzinationen und Lücken in der Indizierung unterscheiden.
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)
Kundenspezifische Bewertung mit LlamaIndex
Für die Phase der kundenspezifischen Bewertung mit LlamaIndex sollten vor dem Ändern des Codes die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Betrachten Sie diese Phase als Vertrag zwischen den Eingaben und den validierten Ausgaben. Benennen Sie die Artefakte, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab. Zitieren Sie die Passagen, die tatsächlich die Antwort begründen. Ohne Zitate können die Operator nicht zwischen Halluzinationen und Lücken in der Indizierung unterscheiden.
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 mit Fokus auf Workflows
Für die Phase des 3-fachigen LangChain Frameworks mit Fokus auf Workflows sollten vor dem Ändern des Codes die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen.
Basierender RAG-Pipeline
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)
Mehrschrittiger Workflow mit Speicherfunktion
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: Suchfunktionen in Produktionsumgebungen
Basische Einrichtung der Pipeline
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])
Implementierung einer hybriden Suchfunktion
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: Bewertungsrahmenwerk
Basische Einrichtung der Bewertung
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)
Kundenspezifische Bewertung mit metriken ohne Referenzen
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)
Integration in bestehende RAG-Systeme
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)
Vergleichsanalyse: Kombination von Tools in Produktionsarchitekturen
Empfohlenes Architekturmuster
# 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)
Implementierungsaspekte und bewährte Praktiken
1. Bewährte Praktiken bei der Dokumentverarbeitung
# 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. Optimierung der Strategie zur Aufteilung in Blöcke
# 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. Überlegungen zur Produktionseinrichtung
# 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. Bewertung und kontinuierliche Verbesserung
# 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)