Notes pratiques : Le guide technique ultime sur les outils RAG open source
Guide pratique pas à pas : Le guide technique ultime sur les outils RAG open source – contrats, vérifications et emplacements de code prêts à l’emploi pour les équipes qui utilisent ce modèle.
Ce guide reconstitue le parcours allant des matières premières jusqu’à un système opérationnel pour : Le Guide technique ultime sur les outils RAG open source : Docling, LlamaIndex, LangChain, Haystack et RAGAS. L’accent est mis sur des étapes concrètes, des vérifications explicites, ainsi que du code que vous pouvez intégrer directement dans un dépôt sans avoir à deviner l’intention derrière chaque action.
Création de systèmes de génération augmentée par la récupération prêts pour le terrain en 2026
Pour l’étape de génération améliorée par la récupération prête à l’emploi dans le cadre de la production de bâtiments, définissez les entrées, le responsable de l’étape et les critères de fin avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Enregistrez les temps d’exécution ainsi que le coût des tokens ou des requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite les factures inattendues lorsque le processus passe d’un environnement de démonstration à des environnements partagés. Citez les passages qui ont réellement servi de base à la réponse. Sans citations, les opérateurs ne peuvent pas distinguer une hallucination d’un manque d’indexation.
Prérequis
Pendant l’étape des prérequis, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans avoir à lire l’ensemble du système. Citez les passages qui ont réellement servi de base à la réponse. Sans citations, les opérateurs ne peuvent pas distinguer une hallucination d’un manque d’indexation.
1. Docling : Fondement du traitement intelligent des documents
Pour l’étape 1 du Docling Intelligent Document, définissez les entrées, le responsable de l’étape et les critères de fin avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Documentez ensemble le parcours normal et le parcours de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non traités font partie du produit, et non d’une mise en forme ultérieure. Citez les passages qui ont réellement servi de base à la réponse. Sans citations, les opérateurs ne peuvent pas distinguer une hallucination d’un manque d’indexation.
Installation et utilisation de base
Pendant l’étape d’installation et d’utilisation de base, définissez les entrées, le responsable de l’étape ainsi que les critères d’arrêt avant de modifier du code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Préférez des unités petites et testables à des scripts complexes. Lorsqu’une étape échoue, l’échec doit indiquer une seule responsabilité plutôt qu’un processus embrouillé. Citez les passages qui ont réellement servi de base à la réponse. Sans citations, les opérateurs ne peuvent pas distinguer une hallucination d’un manque d’indexation.
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)
Traitement avancé de documents avec OCR
Pour le traitement avancé de documents par étapes, définissez les entrées, le responsable de chaque étape ainsi que les critères de fin avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu, sans avoir à deviner l’état caché. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Donnez des noms aux artefacts, définites des vérifications de succès et refusez toute complétion partielle silencieuse. Citez les passages qui justifient réellement la réponse. Sans citations, les opérateurs ne peuvent pas distinguer une hallucination d’un manque d’indexation.
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")
Intégration avec d’autres frameworks
Pendant l’étape d’intégration avec d’autres frameworks, définissez les entrées, le responsable de l’étape et les critères de fin avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Enregistrez les temps d’exécution ainsi que le coût des tokens ou des requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite les factures inattendues lorsque le processus passe d’un environnement de démonstration à des environnements partagés. Citez les passages qui ont réellement servi de base à la réponse. Sans citations, les opérateurs ne peuvent pas distinguer une hallucination d’un manque dans l’indexation.
# 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 : Architecture basée sur la récupération d’informations
Pour la phase 2 de l’architecture Retrieval-First de LlamaIndex, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les stocks de secrets et les flags fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans avoir à lire l’ensemble du graphe. Citez les passages qui ont réellement servi de base à la réponse. Sans citations, les opérateurs ne peuvent pas distinguer une hallucination d’un manque d’indexation.
Mise en œuvre de base de RAG
Pour la phase de mise en œuvre de base de RAG, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Documentez conjointement le parcours normal et le parcours de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non traités font partie du produit, et non d’une amélioration ultérieure. Citez les passages qui ont réellement servi de base à la réponse. Sans citations, les opérateurs ne peuvent pas distinguer une hallucination d’un manque d’indexation.
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)
Chunking hiérarchique avec fusion automatique
Pour l’étape de segmentation hiérarchique avec fusion automatique, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Préférez des unités petites et testables aux scripts complexes. Lorsqu’une étape échoue, l’échec doit indiquer une seule responsabilité plutôt qu’un processus embrouillé. Citez les passages qui ont réellement servi de base à la réponse. Sans citations, les opérateurs ne peuvent pas distinguer une hallucination d’un manque d’indexation.
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)
Évaluation personnalisée avec LlamaIndex
Pour l’étape d’évaluation personnalisée avec LlamaIndex, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Donnez des noms aux artefacts, définites des vérifications de succès et refusez les complétions partielles silencieuses. Citez les passages qui ont réellement servi de base à la réponse. Sans citations, les opérateurs ne peuvent pas distinguer une hallucination d’un manque dans l’indexation.
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 : cadre axé sur le flux de travail
Pour l’étape du cadre LangChain axé sur le flux de travail, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché.
Pipeline RAG de base
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)
Flux de travail en plusieurs étapes avec mémoire
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 : Recherche de niveau professionnel
Mise en place de base du 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])
Mise en œuvre d’une recherche hybride
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 : Cadre d’évaluation
Mise en place de base de l’évaluation
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)
Évaluation personnalisée avec des métriques sans référence
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)
Intégration avec des systèmes RAG existants
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)
Analyse comparative : combinaison d’outils dans des architectures de production
Modèle architectural recommandé
# 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)
Considérations de mise en œuvre et bonnes pratiques
1. Bonnes pratiques pour le traitement des documents
# 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. Optimisation de la stratégie de segmentation
# 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. Considérations relatives au déploiement en production
# 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. Évaluation et amélioration continue
# 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)