Chunking strategie pro RAG
Chunking je klíčová technika pro úspěšné RAG (Retrieval-Augmented Generation) systémy, která určuje, jak efektivně rozdělíme text na menší části. Správná volba chunking strategie zásadně ovlivňuje kvalitu vyhledávání relevantních informací a následné generování odpovědí pomocí jazykových modelů.
Chunking strategie pro RAG: Klíč k efektivnímu vyhledávání
Retrieval-Augmented Generation (RAG) se stal standardem pro vytváření AI aplikací, které potřebují pracovat s velkými objemy dat. Úspěch RAG systému však zásadně závisí na kvalitě chunking strategie – způsobu, jakým rozdělujeme dokumenty na menší části pro embedding a následné vyhledávání.
Proč je chunking kritický?
Embedding modely mají omezení na délku vstupu (obvykle 512-8192 tokenů) a jejich výkon klesá s rostoucí délkou textu. Špatně navržený chunking může vést k:
- Ztrátě kontextu mezi souvisejícími informacemi
- Neefektivnímu vyhledávání relevantních pasáží
- Rozdrobení sémanticky souvisejících bloků
- Vysoké latenci a nákladům na inference
Základní chunking strategie
Fixed-size chunking
Nejjednodušší přístup rozděluje text na bloky pevné velikosti s volitelným překryvem:
def fixed_size_chunking(text, chunk_size=500, overlap=50):
chunks = []
start = 0
while start < len(text):
end = start + chunk_size
chunk = text[start:end]
chunks.append(chunk)
start = end - overlap
return chunks
# Použití
text = "Váš dlouhý dokument..."
chunks = fixed_size_chunking(text, chunk_size=1000, overlap=100)
Výhody: Jednoduchost, predictabilní velikost chunks. Nevýhody: Může rozdělit věty nebo odstavce uprostřed, ignoruje strukturu dokumentu.
Semantic chunking
Pokročilejší přístup využívá NLP techniky pro zachování sémantické integrity:
import spacy
from sentence_transformers import SentenceTransformer
import numpy as np
class SemanticChunker:
def __init__(self):
self.nlp = spacy.load("cs_core_news_sm") # Český model
self.embedding_model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')
def chunk_by_similarity(self, text, similarity_threshold=0.7, max_chunk_size=1000):
doc = self.nlp(text)
sentences = [sent.text.strip() for sent in doc.sents]
if len(sentences) <= 1:
return [text]
embeddings = self.embedding_model.encode(sentences)
chunks = []
current_chunk = [sentences[0]]
for i in range(1, len(sentences)):
# Vypočítej podobnost s předchozí větou
similarity = np.dot(embeddings[i-1], embeddings[i]) / (
np.linalg.norm(embeddings[i-1]) * np.linalg.norm(embeddings[i])
)
# Zkontroluj velikost chunku
current_text = " ".join(current_chunk + [sentences[i]])
if similarity > similarity_threshold and len(current_text) < max_chunk_size:
current_chunk.append(sentences[i])
else:
chunks.append(" ".join(current_chunk))
current_chunk = [sentences[i]]
if current_chunk:
chunks.append(" ".join(current_chunk))
return chunks
Structure-aware chunking
Pro strukturované dokumenty (HTML, Markdown, PDF) je efektivní respektovat hierarchii obsahu:
from bs4 import BeautifulSoup
import re
class StructureChunker:
def __init__(self, max_chunk_size=1000):
self.max_chunk_size = max_chunk_size
def chunk_html(self, html_content):
soup = BeautifulSoup(html_content, 'html.parser')
chunks = []
# Rozděluj podle hlavních sekcí
sections = soup.find_all(['h1', 'h2', 'h3', 'section', 'article'])
for section in sections:
section_text = self._extract_section_content(section)
if len(section_text) > self.max_chunk_size:
# Rekurzivně rozděluj větší sekce
sub_chunks = self._split_large_section(section_text)
chunks.extend(sub_chunks)
else:
chunks.append({
'text': section_text,
'metadata': {
'tag': section.name,
'heading': section.get_text()[:100] if section.name.startswith('h') else None
}
})
return chunks
def _extract_section_content(self, element):
# Získej text včetně všech potomků až po další heading
content = []
current = element
while current and current.next_sibling:
current = current.next_sibling
if hasattr(current, 'name') and current.name and current.name.startswith('h'):
break
if hasattr(current, 'get_text'):
content.append(current.get_text())
return " ".join(content).strip()
Hybridní přístupy
V praxi nejlepší results dosahujeme kombinací více strategií:
class HybridChunker:
def __init__(self):
self.semantic_chunker = SemanticChunker()
self.structure_chunker = StructureChunker()
def chunk_document(self, content, doc_type='text'):
if doc_type == 'html':
# Nejprve structure-aware chunking
structural_chunks = self.structure_chunker.chunk_html(content)
final_chunks = []
for chunk in structural_chunks:
# Pak semantic chunking pro větší bloky
if len(chunk['text']) > 1200:
semantic_chunks = self.semantic_chunker.chunk_by_similarity(
chunk['text'], max_chunk_size=1000
)
for i, sem_chunk in enumerate(semantic_chunks):
final_chunks.append({
'text': sem_chunk,
'metadata': {
**chunk['metadata'],
'sub_chunk': i
}
})
else:
final_chunks.append(chunk)
return final_chunks
else:
# Pro plain text používej pouze semantic chunking
return self.semantic_chunker.chunk_by_similarity(content)
Optimalizace pro různé typy obsahu
Různé typy dokumentů vyžadují specifické přístupy:
- Technická dokumentace: Respektuj sekce, code bloky a hierarchii
- Právní dokumenty: Zachovej číslování paragrafů a odkazy
- Vědecké články: Udržuj together abstrakty, metodiky a závěry
- Chatboty: Krátké chunks s vysokým překryvem pro přesné odpovědi
Evaluace chunking strategie
Pro měření kvality chunking strategie používejme metriky:
def evaluate_chunking_strategy(chunks, queries, ground_truth):
from sklearn.metrics.pairwise import cosine_similarity
# Embed chunks a queries
chunk_embeddings = embedding_model.encode([c['text'] for c in chunks])
query_embeddings = embedding_model.encode(queries)
metrics = {
'avg_chunk_size': np.mean([len(c['text']) for c in chunks]),
'chunk_size_variance': np.var([len(c['text']) for c in chunks]),
'retrieval_accuracy': 0
}
# Eval retrieval accuracy
correct_retrievals = 0
for i, query in enumerate(queries):
similarities = cosine_similarity([query_embeddings[i]], chunk_embeddings)[0]
top_chunk_idx = np.argmax(similarities)
if chunks[top_chunk_idx]['id'] in ground_truth[i]:
correct_retrievals += 1
metrics['retrieval_accuracy'] = correct_retrievals / len(queries)
return metrics
Production tips
Pro nasazení do produkce doporučujeme:
- Cachování embeddings pro častěji používané chunks
- Asynchronní processing pro velké dokumenty
- Monitoring metrik jako chunk retrieval rate a response relevance
- A/B testování různých chunking strategií
- Periodické re-chunking při změně embedding modelu
# Async chunking pro production
import asyncio
from concurrent.futures import ThreadPoolExecutor
class ProductionChunker:
def __init__(self):
self.executor = ThreadPoolExecutor(max_workers=4)
self.chunker = HybridChunker()
self.cache = {}
async def chunk_document_async(self, doc_id, content, doc_type='text'):
if doc_id in self.cache:
return self.cache[doc_id]
loop = asyncio.get_event_loop()
chunks = await loop.run_in_executor(
self.executor,
self.chunker.chunk_document,
content,
doc_type
)
self.cache[doc_id] = chunks
return chunks
Shrnutí
Kvalitní chunking strategie je foundation každého úspěšného RAG systému. Kombinace semantic awareness, strukturální integrity a optimalizace pro konkrétní use case vede k výrazně lepším výsledkům než jednoduché fixed-size chunking. Investice času do návrhu a testování chunking pipeline se vyplatí ve formě vyšší relevance responses a lepší user experience. Nezapomeňte pravidelně měřit a optimalizovat svou strategii na základě real-world dat.