RAG în producție: arhitectură, scalare și monitorizare
Construirea unui prototip RAG care funcționează local este relativ simplă. Adu-l înăuntru producția, unde trebuie să gestioneze mii de interogări simultane, să răspundă în mai puțin de două secunde, menținerea calității înalte în timp și nu pierderea datelor și ceva complet diferit. Distanța dintre „funcționează pe laptopul meu” și „funcționează în producție pentru 10.000 de utilizatori” este uriaș și multe proiecte RAG eșuează chiar la acest pas.
În acest articol abordăm adevăratele provocări ale RAG în producție: arhitectură scalabilă, fragmentare optimă, reclasificare, management al actualizărilor la corpus, monitorizarea cu metrici specifice RAG și evaluarea automată a calitate cu cadre precum RAGAS. Nu e vorba de teorie: fiecare secțiune include Cod Python executabil și modele testate pe sisteme reale.
Ce vei învăța
- Arhitectură pregătită pentru producție a unui sistem RAG scalabil
- Strategii avansate de fragmentare (recursivă, semantică, fereastră de propoziție)
- Conducta de reclasificare încrucișată pentru a îmbunătăți acuratețea
- Gestionarea actualizărilor incrementale de corpus vectorial
- Monitorizare cu valori specifice RAG (fidelitate, relevanță, rechemare)
- Evaluare automată cu framework RAGAS
- Memorarea în cache inteligentă pentru a optimiza latența și costurile
- Gestionarea erorilor și degradarea grațioasă în producție
Prezentare generală a seriei
| # | Articol | Concentrează-te |
|---|---|---|
| 1 | RAG a explicat | Fundații și arhitectură |
| 2 | Înglobări și căutare semantică | BERT, SBERT, FAISS |
| 3 | Baza de date vectorială | Qdrant, Pinecone, Milvus |
| 4 | Recuperare hibridă | BM25 + căutare vectorială |
| 5 | RAG în producție (ești aici) | Scalare, monitorizare, evaluare |
| 6 | LangChain pentru RAG | Cadre și modele avansate |
| 7 | Gestionarea ferestrelor de context | Optimizați intrarea LLM |
| 8 | Sisteme multi-agenți | Orchestrare și coordonare |
| 9 | Inginerie promptă în producție | Șablon, versiune, testare |
| 10 | Grafice de cunoștințe pentru AI | Cunoștințe structurate în LLM |
1. Arhitectura unui sistem RAG gata de producție
Un sistem RAG în producție nu este o simplă conductă secvenţială: este un sistem distribuite cu componente specializate, fiecare cu cerințe de scalabilitate, toleranță și monitorizare diferite la erori. Înțelegeți arhitectura completă și primul pas pentru a construi ceva care să reziste în producție.
Arhitectura de producție RAG: Componente principale
ARCHITETTURA RAG PRODUCTION
┌─────────────────────────────────────────────────────────┐
│ API GATEWAY │
│ (Rate limiting, auth, routing) │
└──────────────────────┬──────────────────────────────────┘
│
┌──────────┴──────────┐
│ │
┌──────▼──────┐ ┌───────▼────────┐
│ QUERY │ │ INGESTION │
│ SERVICE │ │ SERVICE │
└──────┬──────┘ └───────┬────────┘
│ │
┌───────▼──────┐ ┌───────▼────────┐
│ RETRIEVAL │ │ DOCUMENT │
│ ENGINE │ │ PROCESSOR │
│ ┌─────────┐ │ │ ┌──────────┐ │
│ │Embedding│ │ │ │Chunking │ │
│ │Cache │ │ │ │Embedding │ │
│ └────┬────┘ │ │ │Indexing │ │
│ │ │ │ └──────────┘ │
│ ┌────▼────┐ │ └───────┬────────┘
│ │Vector │ │ │
│ │Search │ │ ┌───────▼────────┐
│ └────┬────┘ │ │ VECTOR DB │
│ │ │ │ (Qdrant/Pine) │
│ ┌────▼────┐ │ └────────────────┘
│ │Reranker │ │
│ └────┬────┘ │
└───────┼──────┘
│
┌───────▼──────┐ ┌────────────────┐
│ GENERATION │ │ CACHE │
│ SERVICE │◄──►│ (Redis/ │
│ (LLM) │ │ Semantic) │
└───────┬──────┘ └────────────────┘
│
┌───────▼──────┐ ┌────────────────┐
│ MONITORING │ │ EVALUATION │
│ SERVICE │ │ SERVICE │
│ (Prometheus)│ │ (RAGAS) │
└──────────────┘ └────────────────┘
1.1 Separarea preocupărilor: Ingestie vs Interogare
Modelul fundamental într-un sistem de producție RAG este separare între planul de asimilare și planul de interogare. Aceste două căi au cerințe foarte diferit:
Ingestie vs interogare: cerințe diferite
| Dimensiune | Calea de ingestie | QueryPath |
|---|---|---|
| Latența | Nu este critic (lot) | Critică (<2s p95) |
| Debit | Scăzut-mediu (documente) | Ridicat (mii solicitate/s) |
| CPU/GPU | Generare de încorporare (GPU) | Încorporarea interogărilor + reclasificare (GPU) |
| Erori | Reîncercați cu backoff | Fallback grațios |
| Scalare | Lot orizontal | Apatrid orizontal |
2. Strategii avansate de chunking
Chunking-ul este probabil cea mai neglijată variabilă din sistemele RAG, dar a făcut-o un impact uriaș asupra calității finale. O bucată care este prea mică pierde contextul; unul prea mare introduce zgomot și depășește contextul modelului de încorporare.
2.1 Divizor recursiv de text cu caractere
Cea mai eficientă metodă pentru majoritatea cazurilor de utilizare este împărțirea caracterelor recursive: împarte textul pe separatoare progresiv mai fin (paragrafe, propoziții, cuvinte) pentru a respecta structura firească a documentului.
Chunking recursiv avansat
from langchain.text_splitter import RecursiveCharacterTextSplitter
from typing import List, Dict, Any
import re
class AdvancedChunker:
"""Chunker avanzato con metadati e strategie multiple"""
def __init__(
self,
chunk_size: int = 512,
chunk_overlap: int = 64,
strategy: str = "recursive"
):
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
self.strategy = strategy
self.recursive_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
separators=["\n\n", "\n", ". ", "! ", "? ", " ", ""],
length_function=len,
is_separator_regex=False
)
def chunk_with_metadata(
self,
text: str,
doc_metadata: Dict[str, Any]
) -> List[Dict]:
"""Crea chunks con metadati completi per il retrieval"""
if self.strategy == "recursive":
chunks = self.recursive_splitter.split_text(text)
elif self.strategy == "semantic":
chunks = self._semantic_split(text)
elif self.strategy == "sentence_window":
chunks = self._sentence_window_split(text)
else:
raise ValueError(f"Strategia non supportata: {self.strategy}")
result = []
for i, chunk in enumerate(chunks):
chunk_meta = {
**doc_metadata,
"chunk_index": i,
"total_chunks": len(chunks),
"chunk_size": len(chunk),
"strategy": self.strategy,
# Snippet per il contesto nella risposta
"prev_chunk": chunks[i-1][:100] if i > 0 else None,
"next_chunk": chunks[i+1][:100] if i < len(chunks)-1 else None,
}
result.append({"text": chunk, "metadata": chunk_meta})
return result
def _sentence_window_split(
self,
text: str,
window_size: int = 3
) -> List[str]:
"""
Sentence Window: indicizza singole frasi ma recupera
le frasi vicine per mantenere il contesto.
Tecnica avanzata che migliora il recall mantenendo la precisione.
"""
# Split in frasi
sentences = re.split(r'(?<=[.!?])\s+', text)
sentences = [s.strip() for s in sentences if s.strip()]
chunks = []
for i, sentence in enumerate(sentences):
# Finestra di contesto: frase centrale + vicine
start = max(0, i - window_size // 2)
end = min(len(sentences), i + window_size // 2 + 1)
window = " ".join(sentences[start:end])
chunks.append(window)
return chunks
def _semantic_split(self, text: str) -> List[str]:
"""
Semantic split: divide in base ai cambi di topic
usando embeddings per individuare boundary semantici.
"""
from sentence_transformers import SentenceTransformer
import numpy as np
model = SentenceTransformer('all-MiniLM-L6-v2')
sentences = re.split(r'(?<=[.!?])\s+', text)
if len(sentences) < 3:
return [text]
# Calcola embeddings di ogni frase
embeddings = model.encode(sentences)
# Calcola similarità tra frasi consecutive
similarities = []
for i in range(len(embeddings) - 1):
sim = np.dot(embeddings[i], embeddings[i+1]) / (
np.linalg.norm(embeddings[i]) * np.linalg.norm(embeddings[i+1])
)
similarities.append(sim)
# Trova boundary semantici (bassa similarità = cambio di topic)
threshold = np.mean(similarities) - np.std(similarities)
boundaries = [0]
for i, sim in enumerate(similarities):
if sim < threshold:
boundaries.append(i + 1)
boundaries.append(len(sentences))
# Crea chunks dai segmenti
chunks = []
for i in range(len(boundaries) - 1):
segment = " ".join(sentences[boundaries[i]:boundaries[i+1]])
if len(segment) > self.chunk_size * 2:
# Chunk troppo grande: dividi ulteriormente
sub_chunks = self.recursive_splitter.split_text(segment)
chunks.extend(sub_chunks)
elif segment.strip():
chunks.append(segment)
return chunks
# Esempio di utilizzo
chunker = AdvancedChunker(chunk_size=512, chunk_overlap=64, strategy="recursive")
document = """
Cos'è il RAG?
Il Retrieval-Augmented Generation (RAG) è un'architettura che combina la ricerca su basi
di conoscenza con la generazione di testo degli LLM. Il principio fondamentale è semplice:
invece di lasciare che l'LLM risponda solo con la sua conoscenza interna (soggetta ad
allucinazioni), RAG prima recupera documenti rilevanti e poi li fornisce come contesto.
Come funziona il retrieval?
Il processo di retrieval avviene in due fasi. Prima, i documenti vengono convertiti in
embeddings vettoriali e memorizzati in un vector database. Poi, la query dell'utente viene
anch'essa convertita in un embedding e viene effettuata una ricerca di similarità per
trovare i documenti più rilevanti.
"""
chunks = chunker.chunk_with_metadata(
document,
doc_metadata={"source": "rag-intro.txt", "author": "AI Team"}
)
for chunk in chunks:
print(f"Chunk {chunk['metadata']['chunk_index']}: {len(chunk['text'])} chars")
print(f" {chunk['text'][:100]}...")
2.2 Împărțirea părinte-copil
Una dintre cele mai eficiente strategii pentru sistemele RAG avansate este Împărțirea părinte-copil: bucăți mici (copii) sunt indexate pentru precizie de recuperare, dar bucăți mari (părinte) sunt returnate pentru a furniza context suficient pentru LLM.
Împărțirea părinte-copil
from langchain.storage import InMemoryStore
from langchain.retrievers import ParentDocumentRetriever
from langchain.vectorstores import Qdrant
from langchain_openai import OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
# Splitter per chunks grandi (parent) - per il contesto
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000, chunk_overlap=200)
# Splitter per chunks piccoli (child) - per l'indicizzazione
child_splitter = RecursiveCharacterTextSplitter(chunk_size=400, chunk_overlap=40)
# Storage: chunks grandi in memory store, chunks piccoli in vector store
vectorstore = Qdrant.from_documents(
[], # inizialmente vuoto
OpenAIEmbeddings(),
url="http://localhost:6333",
collection_name="rag_child_chunks"
)
docstore = InMemoryStore() # o Redis per persistenza
# Parent-Child Retriever
retriever = ParentDocumentRetriever(
vectorstore=vectorstore,
docstore=docstore,
child_splitter=child_splitter,
parent_splitter=parent_splitter,
)
# Come funziona:
# 1. add_documents() divide i documenti in parent e child chunks
# 2. I child chunks vengono indicizzati nel vector store
# 3. I parent chunks vengono salvati nel docstore con un ID univoco
# 4. I child chunks hanno un metadato "doc_id" che punta al parent
# Retrieval:
# 1. Cerca i child chunks più simili alla query
# 2. Recupera i parent chunks corrispondenti
# Risultato: massima precisione (child) + massimo contesto (parent)
3. Reclasificare: Îmbunătățiți acuratețea regăsirii
Il reclasificare este o tehnică care îmbunătățește semnificativ calitatea de regăsire prin aplicarea unui al doilea model, mai precis (dar mai lent) la rezultate initialele. Fluxul tipic este: preluați 50-100 de candidați cu căutare rapidă vectorială, Apoi reordonați cu codificare încrucișată precisă, În cele din urmă obține top-k-urile.
3.1 Bi-Encoder vs Cross-Encoder
Diferența dintre cele două abordări este fundamentală:
- Bi-encoder (recuperare): codificați interogările și documentele separat în vectori. Foarte rapid în regăsire deoarece documentele sunt precalculate, dar mai puțin precis pentru că nu vede interogarea în timp ce codifică documentele.
- Cross-encoder (reclasificare): preia interogarea + documentul concatenat și produce un scor de relevanță. Mult mai precis, dar nu la fel de scalabil ca regăsirea, deoarece trebuie să proceseze fiecare pereche interogare-document în timp real.
Reranking Pipeline cu Cross-Encoder
from sentence_transformers import CrossEncoder
from typing import List, Tuple
import time
class RerankingRetriever:
"""Retriever con reranking a due stadi"""
def __init__(
self,
bi_encoder, # Sentence Transformer per retrieval veloce
vector_index, # FAISS o vector DB
cross_encoder_name: str = "cross-encoder/ms-marco-MiniLM-L-6-v2"
):
self.bi_encoder = bi_encoder
self.index = vector_index
self.cross_encoder = CrossEncoder(cross_encoder_name)
self.documents = []
def retrieve_and_rerank(
self,
query: str,
initial_k: int = 50,
final_k: int = 5
) -> List[Tuple[str, float]]:
"""
Pipeline completa:
1. Retrieval veloce con bi-encoder (top-50)
2. Reranking preciso con cross-encoder (top-5)
"""
t0 = time.time()
# STAGE 1: Retrieval veloce
query_emb = self.bi_encoder.encode([query], normalize_embeddings=True)
scores, indices = self.index.search(query_emb.astype('float32'), initial_k)
candidates = [
(self.documents[i], float(s))
for s, i in zip(scores[0], indices[0])
if i != -1
]
t1 = time.time()
print(f"Retrieval: {(t1-t0)*1000:.1f}ms - {len(candidates)} candidati")
# STAGE 2: Reranking preciso con cross-encoder
if not candidates:
return []
# Prepara coppie (query, documento) per il cross-encoder
cross_pairs = [(query, doc) for doc, _ in candidates]
cross_scores = self.cross_encoder.predict(cross_pairs)
# Combina e riordina
reranked = sorted(
zip([doc for doc, _ in candidates], cross_scores),
key=lambda x: x[1],
reverse=True
)
t2 = time.time()
print(f"Reranking: {(t2-t1)*1000:.1f}ms")
return reranked[:final_k]
def reciprocal_rank_fusion(
self,
results_list: List[List[Tuple[str, float]]],
k: int = 60
) -> List[Tuple[str, float]]:
"""
Reciprocal Rank Fusion: combina risultati da multiple strategie
di retrieval (es. dense + sparse) in un singolo ranking.
Formula: score = sum(1 / (k + rank_i)) per ogni lista i
"""
doc_scores = {}
for results in results_list:
for rank, (doc, _) in enumerate(results):
if doc not in doc_scores:
doc_scores[doc] = 0.0
doc_scores[doc] += 1.0 / (k + rank + 1)
return sorted(doc_scores.items(), key=lambda x: x[1], reverse=True)
Modele Cross-Encoder pentru reclasificare
| Model | Viteză | calitate | Utilizare recomandată |
|---|---|---|---|
| cross-encoder/ms-marco-MiniLM-L-6-v2 | Ridicat | Bun | Producția, latența importante |
| cross-encoder/ms-marco-electra-base | Medie | Excelent | Echilibrul bun |
| BAAI/bge-reranker-large | Scăzut | Excelent | Calitate maximă, latență non-critică |
| API-ul Cohere Rerank | API-uri | Excelent | Prototipare, buget disponibil |
4. Cache inteligentă pentru latență și cost
Într-un sistem RAG de producție, un procent mare de interogări sunt similare sau identice (de exemplu, Întrebări frecvente, întrebări frecvente). The cache semantică merge mai departe cache exactă (cache-hit numai pentru interogări identice) și reutilizați rezultatele per interogare similar semantic, reducând dramatic costurile de inferență LLM.
Cache semantic cu Redis și Embeddings
import redis
import numpy as np
import json
import hashlib
from sentence_transformers import SentenceTransformer
from typing import Optional, Tuple
import time
class SemanticCache:
"""Cache semantica che riusa risposte per query simili"""
def __init__(
self,
redis_url: str = "redis://localhost:6379",
embedding_model: str = "all-MiniLM-L6-v2",
similarity_threshold: float = 0.95,
ttl_seconds: int = 3600 # 1 ora
):
self.redis = redis.from_url(redis_url)
self.model = SentenceTransformer(embedding_model)
self.threshold = similarity_threshold
self.ttl = ttl_seconds
def _get_cache_key(self, text: str) -> str:
return f"rag_cache:{hashlib.md5(text.encode()).hexdigest()}"
def get(self, query: str) -> Optional[Tuple[str, float]]:
"""
Cerca in cache:
1. Prima cerca match esatto (O(1))
2. Poi cerca match semantico (O(n) - ottimizzabile con FAISS)
Ritorna (risposta, similarity_score) o None
"""
# Cache esatto
exact_key = self._get_cache_key(query)
cached = self.redis.get(exact_key)
if cached:
data = json.loads(cached)
return data['response'], 1.0
# Cache semantico
query_emb = self.model.encode([query], normalize_embeddings=True)[0]
# Ottimizzazione: usa Redis SCAN per iterare sulle chiavi cached
# In produzione, usa un indice FAISS separato per il lookup semantico
best_score = 0.0
best_response = None
for key in self.redis.scan_iter("rag_cache:*"):
cached = self.redis.get(key)
if not cached:
continue
data = json.loads(cached)
cached_emb = np.array(data['embedding'])
sim = float(np.dot(query_emb, cached_emb))
if sim > best_score:
best_score = sim
best_response = data['response']
if best_score >= self.threshold:
return best_response, best_score
return None
def set(self, query: str, response: str):
"""Salva la risposta in cache con l'embedding della query"""
query_emb = self.model.encode([query], normalize_embeddings=True)[0]
key = self._get_cache_key(query)
data = {
'query': query,
'response': response,
'embedding': query_emb.tolist(),
'timestamp': time.time()
}
self.redis.setex(key, self.ttl, json.dumps(data))
def get_stats(self) -> dict:
"""Statistiche del cache"""
keys = list(self.redis.scan_iter("rag_cache:*"))
return {
'total_cached': len(keys),
'memory_bytes': sum(
self.redis.memory_usage(k) or 0 for k in keys[:100]
)
}
# Integrazione nel RAG pipeline
class CachedRAGPipeline:
def __init__(self, rag_pipeline, cache: SemanticCache):
self.rag = rag_pipeline
self.cache = cache
self.hits = 0
self.misses = 0
def query(self, question: str) -> dict:
# 1. Prova cache
cached = self.cache.get(question)
if cached:
response, sim = cached
self.hits += 1
return {
"answer": response,
"cached": True,
"similarity": sim,
"latency_ms": 1 # quasi istantaneo
}
# 2. Full RAG pipeline
t0 = time.time()
response = self.rag.generate(question)
latency = (time.time() - t0) * 1000
# 3. Salva in cache per future query simili
self.cache.set(question, response)
self.misses += 1
return {
"answer": response,
"cached": False,
"similarity": None,
"latency_ms": latency
}
@property
def hit_rate(self) -> float:
total = self.hits + self.misses
return self.hits / total if total > 0 else 0.0
5. Monitorizare și observabilitate
Un sistem RAG în producție trebuie să fie observabil la toate nivelurile. Nu este suficient monitorizați latența HTTP: trebuie să măsurați calitatea recuperării, rata halucinații, satisfacția utilizatorilor și costurile pe interogare.
5.1 Metrici de infrastructură
Prometheus Metrics pentru RAG
from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time
from functools import wraps
# Metriche infrastrutturali
rag_queries_total = Counter(
'rag_queries_total',
'Numero totale di query RAG',
['status', 'cached']
)
rag_query_duration = Histogram(
'rag_query_duration_seconds',
'Durata delle query RAG',
['component'],
buckets=[0.05, 0.1, 0.25, 0.5, 1.0, 2.0, 5.0]
)
rag_retrieval_chunks = Histogram(
'rag_retrieval_chunks',
'Numero di chunks recuperati per query',
buckets=[1, 3, 5, 10, 20, 50]
)
rag_retrieval_score = Histogram(
'rag_retrieval_score',
'Score di rilevanza del top-1 chunk',
buckets=[0.1, 0.3, 0.5, 0.7, 0.8, 0.9, 0.95, 1.0]
)
rag_cache_hit_rate = Gauge(
'rag_cache_hit_rate',
'Tasso di hit del semantic cache'
)
rag_llm_tokens_total = Counter(
'rag_llm_tokens_total',
'Totale token LLM utilizzati',
['type'] # 'prompt' o 'completion'
)
# Metriche qualità (aggiornate da evaluation service)
rag_faithfulness_score = Gauge(
'rag_faithfulness_score',
'Score medio di faithfulness (risposta supportata da contesto)'
)
rag_answer_relevance = Gauge(
'rag_answer_relevance',
'Score medio di rilevanza della risposta'
)
# Decorator per timing automatico
def track_timing(component: str):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
start = time.time()
try:
result = func(*args, **kwargs)
rag_queries_total.labels(status='success', cached='false').inc()
return result
except Exception as e:
rag_queries_total.labels(status='error', cached='false').inc()
raise
finally:
duration = time.time() - start
rag_query_duration.labels(component=component).observe(duration)
return wrapper
return decorator
# Esempio utilizzo
class MonitoredRAGService:
@track_timing('full_query')
def query(self, question: str) -> dict:
# Retrieval
with rag_query_duration.labels(component='retrieval').time():
chunks = self.retrieve(question)
rag_retrieval_chunks.observe(len(chunks))
if chunks:
rag_retrieval_score.observe(chunks[0][1]) # score top-1
# Generation
with rag_query_duration.labels(component='generation').time():
response = self.generate(question, chunks)
return response
5.2 Înregistrare structurată pentru RAG
Înregistrare structurată pentru RAG cu urmărire
import structlog
import uuid
from contextlib import contextmanager
# Configura structlog per output JSON
structlog.configure(
processors=[
structlog.processors.TimeStamper(fmt="iso"),
structlog.stdlib.add_log_level,
structlog.processors.JSONRenderer()
]
)
logger = structlog.get_logger()
class TracedRAGPipeline:
"""RAG pipeline con tracing distribuito"""
def query(self, question: str, user_id: str = None) -> dict:
trace_id = str(uuid.uuid4())
log = logger.bind(
trace_id=trace_id,
user_id=user_id,
question_hash=hashlib.md5(question.encode()).hexdigest()[:8]
)
log.info("rag_query_start", question_length=len(question))
# Retrieval
t0 = time.time()
try:
chunks = self.retrieve(question)
retrieval_time = time.time() - t0
log.info(
"rag_retrieval_complete",
chunks_retrieved=len(chunks),
top_score=chunks[0][1] if chunks else 0,
retrieval_ms=retrieval_time * 1000
)
except Exception as e:
log.error("rag_retrieval_error", error=str(e))
raise
# Generation
t1 = time.time()
try:
response = self.generate(question, chunks)
generation_time = time.time() - t1
log.info(
"rag_generation_complete",
response_length=len(response),
generation_ms=generation_time * 1000,
total_ms=(time.time() - t0) * 1000
)
except Exception as e:
log.error("rag_generation_error", error=str(e))
raise
return {
"answer": response,
"trace_id": trace_id,
"chunks_used": len(chunks),
"total_ms": (time.time() - t0) * 1000
}
6. Evaluare automată cu RAGAS
RAGAS (RAG Assessment) este cel mai răspândit cadru de evaluare sisteme automate RAG. Măsoară patru dimensiuni fundamentale ale calității:
- Fidelitate: Răspunsul este susținut de contextul preluat? (anti-halucinatii)
- Relevanța răspunsului: Este răspunsul relevant pentru întrebare?
- Reamintire context: Contextul preluat conține informațiile necesare?
- Acuratețea contextului: Bucățile recuperate sunt toate relevante?
Evaluarea RAG cu RAGAS
from ragas import evaluate
from ragas.metrics import (
faithfulness,
answer_relevancy,
context_recall,
context_precision,
answer_correctness
)
from datasets import Dataset
from typing import List, Dict
import pandas as pd
class RAGEvaluator:
"""Valutazione automatica sistema RAG con RAGAS"""
def __init__(self, rag_pipeline):
self.rag = rag_pipeline
def create_evaluation_dataset(
self,
test_questions: List[str],
ground_truths: List[str]
) -> Dataset:
"""
Crea dataset di valutazione processando le domande con il RAG.
Ogni row contiene: domanda, risposta, contesti recuperati, verita.
"""
questions = []
answers = []
contexts = []
for question, gt in zip(test_questions, ground_truths):
# Recupera contesti
chunks = self.rag.retrieve(question, top_k=5)
context_list = [chunk for chunk, _ in chunks]
# Genera risposta
answer = self.rag.generate(question)
questions.append(question)
answers.append(answer)
contexts.append(context_list)
return Dataset.from_dict({
"question": questions,
"answer": answers,
"contexts": contexts,
"ground_truth": ground_truths
})
def evaluate_pipeline(
self,
test_questions: List[str],
ground_truths: List[str]
) -> pd.DataFrame:
"""
Valuta il pipeline RAG e ritorna un report dettagliato.
"""
dataset = self.create_evaluation_dataset(test_questions, ground_truths)
results = evaluate(
dataset,
metrics=[
faithfulness, # 0-1: risposta supportata da contesto?
answer_relevancy, # 0-1: risposta pertinente alla domanda?
context_recall, # 0-1: contesto copre la ground truth?
context_precision, # 0-1: contesto è tutto pertinente?
answer_correctness # 0-1: risposta corretta vs ground truth?
]
)
# Report
df = results.to_pandas()
print("\n=== RAGAS Evaluation Report ===")
print(f"Faithfulness: {df['faithfulness'].mean():.3f}")
print(f"Answer Relevancy: {df['answer_relevancy'].mean():.3f}")
print(f"Context Recall: {df['context_recall'].mean():.3f}")
print(f"Context Precision:{df['context_precision'].mean():.3f}")
print(f"Answer Correct.: {df['answer_correctness'].mean():.3f}")
# Identifica casi problematici
low_faith = df[df['faithfulness'] < 0.5]
if len(low_faith) > 0:
print(f"\nATTENZIONE: {len(low_faith)} domande con faithfulness bassa:")
for _, row in low_faith.iterrows():
print(f" Q: {row['question'][:80]}...")
return df
def continuous_evaluation(self, sample_rate: float = 0.05):
"""
Valutazione continua in produzione: campiona il 5% delle query
e le valuta automaticamente per rilevare degradazione.
"""
import random
def evaluate_sample(question: str, answer: str, contexts: List[str]):
if random.random() > sample_rate:
return
# Stima qualità senza ground truth usando LLM-as-judge
from openai import OpenAI
client = OpenAI()
judge_prompt = f"""Valuta la qualità di questa risposta RAG.
Domanda: {question}
Contesti recuperati: {" | ".join(contexts[:2])}
Risposta generata: {answer}
Valuta su scala 1-5:
1. La risposta è supportata dai contesti? (faithfulness)
2. La risposta è pertinente alla domanda? (relevance)
Rispondi SOLO con JSON: {"faithfulness": X, "relevance": X}"""
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": judge_prompt}],
temperature=0
)
try:
import json
scores = json.loads(response.choices[0].message.content)
# Aggiorna metriche Prometheus
rag_faithfulness_score.set(scores['faithfulness'] / 5.0)
rag_answer_relevance.set(scores['relevance'] / 5.0)
except:
pass
return evaluate_sample
# Test set di esempio
test_questions = [
"Cos'è il RAG e come funziona?",
"Qual è la differenza tra BERT e Sentence Transformers?",
"Come si sceglie un vector database per la produzione?"
]
ground_truths = [
"RAG (Retrieval-Augmented Generation) combina la ricerca su basi di conoscenza con la generazione LLM per ridurre le allucinazioni.",
"BERT produce embeddings contestuali per token, mentre Sentence Transformers è ottimizzato per produrre embeddings a livello di frase per la similarity search.",
"La scelta dipende da scala, requisiti di latenza, budget e se serve hosting gestito o self-hosted."
]
7. Gestionarea actualizărilor Corpus
Un sistem RAG în producție trebuie să gestioneze un corpus care se modifică în timp: documente noi sunt adăugate, documentele învechite urmează să fie eliminate, documentele existente sunt actualizat. Aceasta este problema cu managementul corpusului.
Actualizări incrementale ale indexului cu Qdrant
from qdrant_client import QdrantClient
from qdrant_client.models import (
PointStruct, UpdateStatus, Filter, FieldCondition, MatchValue
)
from sentence_transformers import SentenceTransformer
import hashlib
import time
from typing import List, Dict, Optional
class IncrementalRAGIndex:
"""Gestione aggiornamenti incrementali del corpus RAG"""
def __init__(
self,
collection_name: str = "rag_corpus",
embedding_model: str = "all-MiniLM-L6-v2"
):
self.client = QdrantClient(url="http://localhost:6333")
self.model = SentenceTransformer(embedding_model)
self.collection = collection_name
def _doc_hash(self, text: str) -> str:
"""Hash deterministico per deduplicazione"""
return hashlib.sha256(text.encode()).hexdigest()[:16]
def upsert_document(
self,
doc_id: str,
text: str,
metadata: Dict,
chunk_size: int = 512
) -> int:
"""
Upsert (insert o update) di un documento.
Se il documento è già presente e non modificato, skippa.
"""
# Calcola hash per deduplication
content_hash = self._doc_hash(text)
# Controlla se già presente e non modificato
existing = self.client.scroll(
collection_name=self.collection,
scroll_filter=Filter(
must=[FieldCondition(
key="doc_id",
match=MatchValue(value=doc_id)
)]
),
limit=1
)
if existing[0] and existing[0][0].payload.get('content_hash') == content_hash:
return 0 # Nessun aggiornamento necessario
# Rimuovi vecchi chunks del documento
self.client.delete(
collection_name=self.collection,
points_selector=Filter(
must=[FieldCondition(
key="doc_id",
match=MatchValue(value=doc_id)
)]
)
)
# Crea nuovi chunks
words = text.split()
chunk_words = chunk_size // 5 # ~5 chars/word
chunks = [
' '.join(words[i:i+chunk_words])
for i in range(0, len(words), chunk_words - 10) # 10 word overlap
if words[i:i+chunk_words]
]
if not chunks:
return 0
# Genera embeddings
embeddings = self.model.encode(chunks, normalize_embeddings=True)
# Crea points
points = []
for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)):
point_id = int(hashlib.md5(f"{doc_id}_{i}".encode()).hexdigest()[:8], 16)
points.append(PointStruct(
id=point_id,
vector=embedding.tolist(),
payload={
**metadata,
"doc_id": doc_id,
"chunk_index": i,
"content_hash": content_hash,
"text": chunk,
"updated_at": time.time()
}
))
# Upsert in Qdrant
self.client.upsert(
collection_name=self.collection,
points=points
)
return len(chunks)
def delete_document(self, doc_id: str):
"""Rimuove tutti i chunks di un documento"""
self.client.delete(
collection_name=self.collection,
points_selector=Filter(
must=[FieldCondition(
key="doc_id",
match=MatchValue(value=doc_id)
)]
)
)
def get_corpus_stats(self) -> Dict:
"""Statistiche sul corpus"""
info = self.client.get_collection(self.collection)
return {
"total_chunks": info.points_count,
"index_status": info.status,
"vectors_config": str(info.config.params.vectors)
}
8. Cele mai bune practici și anti-modele în producție
Lista de verificare RAG pentru producție
- Îmbunătățire: utilizați dimensiunea bucăților 400-600 de jetoane cu 10-15% suprapunere; testați diferite strategii pe corpul dumneavoastră specific
- Reclasificare: implementați întotdeauna un codificator încrucișat pentru interogările în care precizia este critică; acceptă latență suplimentară (100-300 ms)
- Memorarea în cache: memoria cache semantică cu pragul 0,92-0,97 reduce costurile LLM cu 30-60% pe corpus asemănător FAQ
- Monitorizare: urmărirea fidelității, relevanța răspunsului, latența de recuperare, costul simbolului LLM pe interogare
- Evaluare: mențineți un set de teste de aur (100-200 de întrebări cu adevărul de bază) și evaluați la fiecare implementare
- Alternative: dacă recuperarea nu găsește nimic relevant (scor superior < 0,5), declarați „Nu știu” în loc să halucinați
- Versiune: Șabloane de încorporare a versiunii și reindexați atunci când schimbați; menține indici paraleli în timpul migrării
Anti-modele de evitat în producție
- Fără monitorizare a calității: monitorizarea doar a latenței și a timpului de funcționare nu este suficientă. RAG-ul poate „funcționa” din punct de vedere tehnic, dar oferă răspunsuri greșite.
- Corpus neactualizat: un sistem RAG pe documentație învechită este mai rău decât nu RAG: răspunde cu încredere la informații incorecte.
- Top-k fix: adaptați numărul de bucăți preluate la lungimea interogării. Interogările complexe necesită mai mult context.
- Ignorarea latenței de încorporare: Generarea încorporarii unei interogări durează 10-50 ms. Înmulțiți cu 1000 req/s și devine blocajul.
- LLM ca judecător absolut: modelul generativ poate „halucina” în afara contextului chiar și cu RAG. Implementați balustrade pentru răspunsuri.
Concluzii
Aducerea unui sistem RAG în producție necesită mult mai mult decât o simplă conductă secvenţial. Am văzut cum arhitectura pregătită pentru producție separă planurile asimilarea și interogările, modul în care fragmentarea avansată are un impact direct asupra calității, cum reclasificarea încrucișată îmbunătățește acuratețea recuperării și cum monitorizarea cu RAGAS vă permite să urmăriți calitatea în timp.
Punctele cheie:
- Separați întotdeauna asimilarea și calea de interogare - au cerințe radical diferite
- Investește în chunking: este cea mai impactantă variabilă și adesea trecută cu vederea
- Implementați reclasificarea intercoderului pentru cazurile de utilizare în care precizia este critică
- Utilizați memoria cache semantică pentru a reduce costurile și latența la interogările repetitive
- Măsurați fidelitatea și relevanța răspunsurilor, nu doar latența și timpul de funcționare
- Mențineți un set de teste de aur și evaluați la fiecare implementare cu RAGAS
În următorul articol vom explora LangChain pentru RAG: cadrul cel mai popular pentru construirea de aplicații LLM, cu accent pe modele avansate, cum ar fi RAG conversațional, recuperarea multi-hop și apelarea instrumentelor.
Seria continuă
- Articolul 1: RAG Explained - Fundamente
- Articolul 2: Înglobări și căutare semantică
- Articolul 3: Baza de date vectorială - Qdrant vs Pinecone
- Articolul 4: Recuperare hibridă: BM25 + Vector
- Articolul 5: RAG în producție (actual)
- Articolul 6: LangChain pentru RAG
Aflați mai multe cu: pgvector pentru RAG pe PostgreSQL e MLOps: Servire model în producție.