Intermédiaire
🧠 Fondamentaux
20 XP
0 personnes ont réussi
Pipeline de chunking avec Documents LangChain
En RAG, le pipeline classique c'est : charger les documents, les decouper en chunks, les vectoriser, puis les indexer. On va assembler les étapes de chargement et decoupe.
Écris une fonction prepare_documents(raw_texts, sources, chunk_size, chunk_overlap) qui : 1. Crée un Document LangChain pour chaque texte avec sa source 2. Decoupe tous les documents avec RécursiveCharacterTextSplitter 3. Ajoute dans les metadonnées de chaque chunk : - 'chunk_index' : son index dans la liste finale - 'total_chunks' : le nombre total de chunks 4. Renvoie la liste des chunks
raw_texts et sources sont deux listes de meme longueur.
from langchain_core.documents import Document
texts = ['Bonjour le monde. ' * 30, 'Python est genial. ' * 20]
sources = ['doc1.txt', 'doc2.txt']
chunks = prepare_documents(texts, sources, 80, 10)
assert len(chunks) > 2, 'Les textes doivent etre decoupes en plusieurs chunks'
assert all(isinstance(c, Document) for c in chunks), 'Chaque chunk doit etre un Document'
assert 'source' in chunks[0].metadata, 'Chaque chunk doit avoir une source'
assert chunks[0].metadata['chunk_index'] == 0, 'Le premier chunk doit avoir chunk_index 0'
assert chunks[-1].metadata['chunk_index'] == len(chunks) - 1, 'Le dernier chunk doit avoir le bon index'
assert chunks[0].metadata['total_chunks'] == len(chunks), 'total_chunks doit valoir le nombre de chunks'
sources_found = {c.metadata['source'] for c in chunks}
assert sources_found == {'doc1.txt', 'doc2.txt'}, 'Les deux sources doivent etre presentes'
Indices (3 disponibles)
Solution officielle
from langchain_core.documents import Document
from langchain_text_splitters import RécursiveCharacterTextSplitter
def prepare_documents(raw_texts, sources, chunk_size, chunk_overlap):
documents = [
Document(page_content=text, metadata={'source': source})
for text, source in zip(raw_texts, sources)
]
splitter = RécursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap
)
chunks = splitter.split_documents(documents)
total = len(chunks)
for i, chunk in enumerate(chunks):
chunk.metadata['chunk_index'] = i
chunk.metadata['total_chunks'] = total
return chunks