You've shipped a RAG system. Users are querying it, answers are coming back, and somewhere between "good enough for demo day" and "production-grade intelligence," you've noticed something uncomfortable: the system's performance is frozen in amber. The chunks you created during your initial indexing run still live in your vector store, scored with whatever embedding model you were using three months ago, retrieved by the same naive top-k similarity search you set up in week one. Meanwhile, your users' actual queries have taught you nothing. Your system is getting older, but it isn't getting smarter.
This is the core tension of production RAG: the pipeline is inherently static, but the world it's supposed to serve is dynamic. Documents get updated, user intent drifts, retrieval patterns reveal which chunks are actually useful versus which ones merely look relevant at embedding time. The gap between a RAG system that works and one that continuously improves is not a matter of choosing a better base model — it's a matter of building feedback infrastructure around the entire pipeline. Every query is a free lesson. Most teams throw those lessons away.
By the end of this lesson, you'll have built the conceptual and practical foundation for a self-improving RAG pipeline: one that scores chunk quality automatically, captures retrieval signals from real usage, uses those signals to reweight and reorganize its index, and does all of this without requiring a human to babysit every failure mode.
What you'll learn:
You should be comfortable with:
You do not need prior experience building feedback systems or ML training pipelines. We'll build that intuition from scratch.
Before we build anything, we need to understand the failure modes clearly, because the architecture decisions we'll make are direct responses to specific problems.
The Chunking Lottery. When you first chunk a document corpus, you're making a bet about what queries will look like. Fixed-size chunking assumes every answer fits in 512 tokens. Sentence-boundary chunking assumes semantic meaning doesn't span paragraphs. Recursive chunking is better, but still blind to actual retrieval intent. Whatever strategy you chose, some chunks came out great — coherent, information-dense, contextually complete. Others came out terrible — mid-sentence splits, orphaned data tables, repetitive boilerplate. Your vector store holds both kinds, and it has no way to tell them apart.
The Embedding Drift Problem. Embedding models encode semantic meaning as it was understood at training time. When your documents use domain-specific language ("churn," "CAC," or "sprint velocity" mean very different things in SaaS versus general conversation), your embeddings may be systematically misaligning query intent with chunk content. This mismatch doesn't show up as an error — it shows up as subtly wrong answers that are hard to debug because the retrieved chunks are plausible.
The Relevance Signal Graveyard. Every time a user gets a good answer, every time they rephrase a query because the first result was off, every time a generated response includes a caveat like "I couldn't find specific information about..." — all of these are signals. They tell you which chunks are doing work and which are dead weight. Without infrastructure to capture these signals, you're flying blind, rebuilding your index periodically based on intuition rather than evidence.
The Cold Start / Warm Corpus Mismatch. Your initial index was built on day one, before you had any user traffic. Your corpus has evolved — documents added, updated, deprecated — but chunk-level metadata hasn't kept pace. Over time, your index becomes a palimpsest: new content layered on top of stale structure, with no consistent quality signal across the whole thing.
The first piece of self-improvement infrastructure is a scoring system that evaluates every chunk on dimensions that actually predict retrieval success. We're not trying to replace human judgment — we're trying to scale it.
A useful chunk quality score needs to capture at least four things:
Each of these can be computed programmatically with varying degrees of sophistication. Let's build a scorer that handles all four.
import re
import hashlib
from dataclasses import dataclass, field
from typing import Optional
import numpy as np
from openai import AsyncOpenAI
@dataclass
class ChunkQualityScore:
chunk_id: str
semantic_density: float # 0.0 - 1.0
answerability: float # 0.0 - 1.0
context_sufficiency: float # 0.0 - 1.0
redundancy_penalty: float # 0.0 - 1.0 (higher = more redundant)
composite_score: float # weighted combination
scoring_version: str = "1.0"
metadata: dict = field(default_factory=dict)
def is_indexable(self, threshold: float = 0.45) -> bool:
return self.composite_score >= threshold
class ChunkQualityScorer:
"""
Multi-dimensional chunk quality scorer. Designed to run at index time
and periodically as a background process against existing chunks.
"""
BOILERPLATE_PATTERNS = [
r"^\s*(page \d+|chapter \d+|table of contents)\s*$",
r"^\s*(all rights reserved|copyright \d{4})\s*$",
r"^\s*\.{3,}\s*\d+\s*$", # TOC dot leaders
r"^(https?://\S+\s*){3,}$", # chunks that are mostly URLs
]
CONTEXT_DEPENDENCY_PATTERNS = [
r"^(as mentioned|as discussed|as noted|as stated|as shown) (above|below|earlier|previously)",
r"^(this|these|those|the following|the above) (is|are|was|were|refers to|shows|demonstrates)",
r"^(see|refer to|as per) (the|figure|table|section|above|below)",
r"^(in addition|furthermore|moreover|consequently|therefore|thus),",
]
def __init__(
self,
openai_client: AsyncOpenAI,
embedding_model: str = "text-embedding-3-small",
redundancy_similarity_threshold: float = 0.92,
):
self.client = openai_client
self.embedding_model = embedding_model
self.redundancy_threshold = redundancy_similarity_threshold
self._boilerplate_re = [
re.compile(p, re.IGNORECASE | re.MULTILINE)
for p in self.BOILERPLATE_PATTERNS
]
self._context_dep_re = [
re.compile(p, re.IGNORECASE)
for p in self.CONTEXT_DEPENDENCY_PATTERNS
]
def score_semantic_density(self, text: str) -> float:
"""
Heuristic density scoring based on information-bearing tokens.
A production version might use a trained classifier here.
"""
if not text.strip():
return 0.0
words = text.split()
if len(words) < 10:
return 0.1 # Very short chunks rarely retrieve well
# Check for boilerplate patterns
for pattern in self._boilerplate_re:
if pattern.search(text):
return 0.05
# Ratio of unique meaningful words to total words
stopwords = {
"the", "a", "an", "is", "in", "it", "of", "to", "and",
"or", "but", "that", "this", "for", "with", "on", "at",
"from", "by", "as", "be", "was", "are", "were", "been"
}
meaningful_words = [
w.lower() for w in words
if w.isalpha() and w.lower() not in stopwords
]
unique_meaningful = set(meaningful_words)
lexical_diversity = len(unique_meaningful) / max(len(words), 1)
content_ratio = len(meaningful_words) / max(len(words), 1)
# Normalize to 0-1, tuned to typical document distributions
density = min(1.0, (lexical_diversity * 0.6 + content_ratio * 0.4) * 1.8)
return round(density, 4)
def score_context_sufficiency(self, text: str) -> float:
"""
Detects whether the chunk depends on context not present in the chunk itself.
"""
score = 1.0
first_sentence = text.strip().split('.')[0] if '.' in text else text[:200]
for pattern in self._context_dep_re:
if pattern.match(first_sentence):
score -= 0.35
break # one penalty per chunk; don't double-count
# Hanging references: pronouns at the start suggest prior context needed
if re.match(r"^(it|they|them|their|its|he|she|his|her)\s", first_sentence, re.IGNORECASE):
score -= 0.25
# Penalize very short chunks that can't be self-contained
word_count = len(text.split())
if word_count < 30:
score -= 0.20
return max(0.0, round(score, 4))
async def score_answerability(self, text: str, chunk_id: str) -> float:
"""
Uses an LLM to estimate whether a real question could be answered
from this chunk alone. This is the most expensive step — cache aggressively.
"""
prompt = f"""You are evaluating whether a text chunk from a document retrieval system
could serve as a useful answer to at least one specific, realistic question.
Respond with a JSON object containing:
- "can_answer": boolean (true if the chunk answers at least one clear question)
- "example_question": string (the question it answers, or null)
- "confidence": float between 0.0 and 1.0
TEXT CHUNK:
{text[:1200]}
Respond with only the JSON object."""
try:
response = await self.client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"},
temperature=0.1,
max_tokens=150,
)
import json
result = json.loads(response.choices[0].message.content)
return float(result.get("confidence", 0.5)) if result.get("can_answer") else 0.15
except Exception as e:
# Fail gracefully — don't let scoring errors block indexing
print(f"Answerability scoring failed for chunk {chunk_id}: {e}")
return 0.5 # Neutral default
def score_redundancy(
self,
chunk_embedding: list[float],
existing_embeddings: list[list[float]],
) -> float:
"""
Computes redundancy as max cosine similarity against existing chunk embeddings.
Returns a penalty value (higher = more redundant).
"""
if not existing_embeddings:
return 0.0
vec = np.array(chunk_embedding)
existing_matrix = np.array(existing_embeddings)
# Cosine similarity (assumes embeddings are already normalized)
similarities = existing_matrix @ vec / (
np.linalg.norm(existing_matrix, axis=1) * np.linalg.norm(vec) + 1e-9
)
max_similarity = float(np.max(similarities))
if max_similarity >= self.redundancy_threshold:
return min(1.0, (max_similarity - self.redundancy_threshold) / (1 - self.redundancy_threshold))
return 0.0
async def score_chunk(
self,
chunk_id: str,
text: str,
chunk_embedding: list[float],
existing_embeddings: list[list[float]],
weights: dict = None,
) -> ChunkQualityScore:
"""
Compute composite quality score for a single chunk.
"""
weights = weights or {
"semantic_density": 0.30,
"answerability": 0.35,
"context_sufficiency": 0.25,
"redundancy_penalty": 0.10,
}
density = self.score_semantic_density(text)
sufficiency = self.score_context_sufficiency(text)
answerability = await self.score_answerability(text, chunk_id)
redundancy = self.score_redundancy(chunk_embedding, existing_embeddings)
# Redundancy is a penalty, so we subtract it
composite = (
density * weights["semantic_density"]
+ answerability * weights["answerability"]
+ sufficiency * weights["context_sufficiency"]
- redundancy * weights["redundancy_penalty"]
)
composite = max(0.0, min(1.0, composite))
return ChunkQualityScore(
chunk_id=chunk_id,
semantic_density=density,
answerability=answerability,
context_sufficiency=sufficiency,
redundancy_penalty=redundancy,
composite_score=round(composite, 4),
)
A note on LLM-based answerability scoring: Calling GPT-4o-mini for every chunk during indexing has real cost implications at scale. For a corpus of 100,000 chunks, even at $0.15/1M tokens, it's manageable — but you should cache these scores and only re-score chunks when their content changes. Hash the chunk text and store it alongside the score.
Quality scores are most useful when they live inside your vector store as metadata, queryable at retrieval time. Here's the pattern for Qdrant:
from qdrant_client import QdrantClient
from qdrant_client.models import PointStruct, VectorParams, Distance
async def upsert_chunk_with_quality(
qdrant: QdrantClient,
collection_name: str,
chunk_id: str,
text: str,
embedding: list[float],
quality_score: ChunkQualityScore,
source_metadata: dict,
):
"""
Stores a chunk with its quality score embedded in Qdrant metadata.
This enables quality-aware filtering at retrieval time.
"""
payload = {
"text": text,
"source": source_metadata.get("source", "unknown"),
"document_id": source_metadata.get("document_id"),
"chunk_index": source_metadata.get("chunk_index"),
"quality": {
"composite_score": quality_score.composite_score,
"semantic_density": quality_score.semantic_density,
"answerability": quality_score.answerability,
"context_sufficiency": quality_score.context_sufficiency,
"redundancy_penalty": quality_score.redundancy_penalty,
"scoring_version": quality_score.scoring_version,
},
"retrieval_stats": {
"times_retrieved": 0,
"times_useful": 0,
"times_skipped": 0,
"last_retrieved_at": None,
},
"indexed_at": source_metadata.get("indexed_at"),
"content_hash": hashlib.sha256(text.encode()).hexdigest(),
}
qdrant.upsert(
collection_name=collection_name,
points=[PointStruct(id=chunk_id, vector=embedding, payload=payload)],
)
With quality scores in metadata, you can immediately start filtering at retrieval time:
from qdrant_client.models import Filter, FieldCondition, Range
def build_quality_filter(min_quality: float = 0.45) -> Filter:
"""Only retrieve chunks that meet a minimum quality threshold."""
return Filter(
must=[
FieldCondition(
key="quality.composite_score",
range=Range(gte=min_quality)
)
]
)
This single filter immediately raises the floor of your retrieval quality. You stop wasting context window space on boilerplate, orphaned fragments, and near-duplicates.
Quality scoring tells you what you think will retrieve well. Feedback signals tell you what actually does retrieve well. The gap between those two things is where your improvement loop lives.
Not all feedback signals are equal. A useful taxonomy for RAG systems:
Explicit signals (user-generated, high confidence):
Implicit signals (behavior-inferred, lower confidence but high volume):
System signals (pipeline-generated, highest volume):
from enum import Enum
from datetime import datetime, UTC
from typing import Optional
from pydantic import BaseModel
class FeedbackSignalType(str, Enum):
EXPLICIT_POSITIVE = "explicit_positive"
EXPLICIT_NEGATIVE = "explicit_negative"
QUERY_REFORMULATION = "query_reformulation"
SESSION_ABANDONMENT = "session_abandonment"
LLM_INSUFFICIENT = "llm_insufficient" # LLM said it couldn't answer
LLM_CONFIDENT = "llm_confident"
FAITHFULNESS_HIGH = "faithfulness_high"
FAITHFULNESS_LOW = "faithfulness_low"
class RetrievalFeedbackEvent(BaseModel):
event_id: str
session_id: str
query: str
query_embedding_hash: str # hash of the query embedding for clustering
retrieved_chunk_ids: list[str]
chunk_scores_at_retrieval: dict[str, float] # chunk_id -> similarity score
signal_type: FeedbackSignalType
signal_weight: float # normalized 0-1 signal strength
timestamp: datetime
response_text: Optional[str] = None
metadata: dict = {}
class Config:
use_enum_values = True
The instrumented retriever wraps your base retrieval logic, adds timing, and emits feedback events without blocking the response path:
import asyncio
import uuid
import hashlib
from datetime import datetime, UTC
from typing import Callable
class InstrumentedRetriever:
"""
A retriever wrapper that captures retrieval events and coordinates
feedback collection for downstream processing.
"""
def __init__(
self,
base_retriever,
feedback_store, # Your PostgreSQL, Redis, or event queue client
faithfulness_scorer: Optional[Callable] = None,
):
self.retriever = base_retriever
self.feedback_store = feedback_store
self.faithfulness_scorer = faithfulness_scorer
async def retrieve_and_instrument(
self,
query: str,
session_id: str,
top_k: int = 5,
quality_threshold: float = 0.45,
) -> tuple[list[dict], str]:
"""
Returns retrieved chunks and a retrieval_id for linking
subsequent feedback events to this specific retrieval.
"""
retrieval_id = str(uuid.uuid4())
query_embedding = await self._embed_query(query)
query_hash = hashlib.md5(
str(query_embedding[:10]).encode()
).hexdigest()
chunks = await self.retriever.retrieve(
query_embedding=query_embedding,
top_k=top_k,
filter=build_quality_filter(quality_threshold),
)
# Record which chunks were retrieved and at what similarity
chunk_scores = {c["chunk_id"]: c["similarity"] for c in chunks}
# Emit the retrieval event — do this asynchronously
# so it doesn't block the user-facing response
asyncio.create_task(
self._emit_retrieval_event(
retrieval_id=retrieval_id,
session_id=session_id,
query=query,
query_embedding_hash=query_hash,
chunk_scores=chunk_scores,
)
)
return chunks, retrieval_id
async def record_llm_signal(
self,
retrieval_id: str,
session_id: str,
query: str,
response: str,
chunk_ids: list[str],
chunk_scores: dict[str, float],
):
"""
Automatically scored signal based on LLM response analysis.
Run this after generation, before returning to user.
"""
insufficient_phrases = [
"i don't have enough information",
"i couldn't find",
"the provided context doesn't",
"based on available information, i cannot",
"i was unable to find specific",
]
response_lower = response.lower()
is_insufficient = any(phrase in response_lower for phrase in insufficient_phrases)
signal_type = (
FeedbackSignalType.LLM_INSUFFICIENT
if is_insufficient
else FeedbackSignalType.LLM_CONFIDENT
)
# Weight: insufficient signals are stronger than confident ones
# because false negatives (missing info) are more costly than
# false positives (having extra info available)
weight = 0.9 if is_insufficient else 0.4
event = RetrievalFeedbackEvent(
event_id=str(uuid.uuid4()),
session_id=session_id,
query=query,
query_embedding_hash="",
retrieved_chunk_ids=chunk_ids,
chunk_scores_at_retrieval=chunk_scores,
signal_type=signal_type,
signal_weight=weight,
timestamp=datetime.now(UTC),
response_text=response[:500],
)
await self.feedback_store.append_event(event)
async def record_explicit_feedback(
self,
retrieval_id: str,
session_id: str,
query: str,
chunk_ids: list[str],
chunk_scores: dict[str, float],
is_positive: bool,
):
"""Called when a user explicitly rates a response."""
event = RetrievalFeedbackEvent(
event_id=str(uuid.uuid4()),
session_id=session_id,
query=query,
query_embedding_hash="",
retrieved_chunk_ids=chunk_ids,
chunk_scores_at_retrieval=chunk_scores,
signal_type=(
FeedbackSignalType.EXPLICIT_POSITIVE
if is_positive
else FeedbackSignalType.EXPLICIT_NEGATIVE
),
signal_weight=1.0, # Explicit signals get full weight
timestamp=datetime.now(UTC),
)
await self.feedback_store.append_event(event)
async def _embed_query(self, query: str) -> list[float]:
# Delegate to your embedding client
raise NotImplementedError
async def _emit_retrieval_event(self, **kwargs):
# Internal — write to feedback store
raise NotImplementedError
Store feedback events in a relational database so you can aggregate, join, and analyze them efficiently. Here's a PostgreSQL schema:
CREATE TABLE retrieval_feedback_events (
event_id UUID PRIMARY KEY,
session_id TEXT NOT NULL,
query TEXT NOT NULL,
query_hash TEXT NOT NULL, -- For clustering similar queries
retrieved_chunk_ids TEXT[] NOT NULL,
chunk_scores JSONB NOT NULL, -- {chunk_id: similarity_score}
signal_type TEXT NOT NULL,
signal_weight FLOAT NOT NULL,
response_text TEXT,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
metadata JSONB DEFAULT '{}'
);
-- Materialized view: chunk-level aggregated signal
CREATE MATERIALIZED VIEW chunk_retrieval_stats AS
SELECT
chunk_id,
COUNT(*) as times_retrieved,
SUM(CASE WHEN signal_type IN ('explicit_positive', 'llm_confident', 'faithfulness_high')
THEN signal_weight ELSE 0 END) as positive_signal_sum,
SUM(CASE WHEN signal_type IN ('explicit_negative', 'llm_insufficient', 'faithfulness_low',
'query_reformulation', 'session_abandonment')
THEN signal_weight ELSE 0 END) as negative_signal_sum,
MAX(created_at) as last_retrieved_at
FROM retrieval_feedback_events,
UNNEST(retrieved_chunk_ids) AS chunk_id
GROUP BY chunk_id;
CREATE INDEX ON retrieval_feedback_events (query_hash);
CREATE INDEX ON retrieval_feedback_events (created_at);
CREATE INDEX ON retrieval_feedback_events USING GIN (retrieved_chunk_ids);
Now we connect the signals to action. The feedback loop has two modes: incremental (lightweight, runs frequently) and batch (heavier, runs periodically).
In incremental mode, you don't touch the index at all. Instead, you fuse the static quality score and embedding similarity with a dynamic signal score computed from the feedback store at query time:
class FeedbackAwareReranker:
"""
Fuses embedding similarity, chunk quality score, and accumulated
feedback signals into a final ranking score.
"""
def __init__(self, feedback_store, signal_window_days: int = 30):
self.feedback_store = feedback_store
self.signal_window_days = signal_window_days
async def rerank(
self,
query: str,
candidates: list[dict],
weights: dict = None,
) -> list[dict]:
"""
Candidates should already have 'similarity', 'quality_score', and 'chunk_id' fields.
"""
weights = weights or {
"similarity": 0.50,
"quality": 0.25,
"feedback_signal": 0.25,
}
chunk_ids = [c["chunk_id"] for c in candidates]
feedback_scores = await self._get_feedback_scores(chunk_ids)
scored = []
for chunk in candidates:
cid = chunk["chunk_id"]
feedback = feedback_scores.get(cid, {"score": 0.5, "confidence": 0.0})
# Blend feedback score toward neutral when confidence is low
# (i.e., when we have few observations for this chunk)
blended_feedback = (
feedback["score"] * feedback["confidence"]
+ 0.5 * (1 - feedback["confidence"])
)
final_score = (
chunk["similarity"] * weights["similarity"]
+ chunk["quality_score"] * weights["quality"]
+ blended_feedback * weights["feedback_signal"]
)
scored.append({**chunk, "final_score": round(final_score, 4)})
return sorted(scored, key=lambda x: x["final_score"], reverse=True)
async def _get_feedback_scores(
self, chunk_ids: list[str]
) -> dict[str, dict]:
"""
Fetches aggregated feedback for a list of chunk IDs.
Uses Wilson score interval for statistical robustness.
"""
raw_stats = await self.feedback_store.get_chunk_stats(
chunk_ids=chunk_ids,
window_days=self.signal_window_days,
)
result = {}
for cid, stats in raw_stats.items():
pos = stats["positive_signal_sum"]
neg = stats["negative_signal_sum"]
total = pos + neg
if total < 5:
# Not enough data — low confidence, neutral score
result[cid] = {"score": 0.5, "confidence": min(total / 5.0, 1.0)}
else:
# Wilson score interval lower bound (conservative estimate)
import math
z = 1.96 # 95% confidence
p_hat = pos / total
denominator = 1 + z**2 / total
center = (p_hat + z**2 / (2 * total)) / denominator
margin = (z * math.sqrt(p_hat*(1-p_hat)/total + z**2/(4*total**2))) / denominator
wilson_lower = center - margin
result[cid] = {
"score": round(wilson_lower, 4),
"confidence": min(1.0, total / 50.0), # Full confidence at 50+ observations
}
return result
The Wilson score interval is important here. It prevents you from massively upranking a chunk that got three positive signals and zero negative ones — three observations are not statistically meaningful. The Wilson lower bound stays near 0.5 until you accumulate enough data to justify a stronger adjustment.
Batch optimization jobs run as background processes — nightly or weekly — and make structural changes to the index. There are three primary jobs:
Job 1: Quality Score Refresh
Chunks that have accumulated significant negative feedback should have their quality scores recalibrated. This is also the right time to re-score any chunks that haven't been scored yet (documents added before you deployed the scorer).
class QualityScoreRefreshJob:
"""
Identifies chunks whose quality scores need updating and recomputes them.
Designed to run as a background job without disrupting live traffic.
"""
def __init__(self, qdrant, scorer, feedback_store, batch_size: int = 100):
self.qdrant = qdrant
self.scorer = scorer
self.feedback_store = feedback_store
self.batch_size = batch_size
async def run(self, collection_name: str, max_chunks: int = 5000):
"""
Priority order for rescoring:
1. Chunks with high negative feedback and no recent rescore
2. Chunks with old or missing quality scores
3. Chunks with content changes (hash mismatch)
"""
candidates = await self._identify_rescore_candidates(
collection_name, max_chunks
)
print(f"Quality refresh: {len(candidates)} chunks to rescore")
for i in range(0, len(candidates), self.batch_size):
batch = candidates[i : i + self.batch_size]
await self._rescore_batch(collection_name, batch)
await asyncio.sleep(0.1) # Yield — don't saturate the event loop
async def _identify_rescore_candidates(
self, collection_name: str, max_chunks: int
) -> list[str]:
"""
Queries Qdrant for chunks matching rescore criteria.
In a real implementation, use scroll API with appropriate filters.
"""
# Chunks with high negative feedback
high_negative = await self.feedback_store.get_chunks_with_negative_ratio(
threshold=0.6, min_observations=10, limit=max_chunks // 2
)
# Chunks with stale scores (scored more than 30 days ago)
# This would come from a Qdrant scroll with date filter on quality.scored_at
stale_scored = [] # Simplified for clarity
return list(set(high_negative + stale_scored))[:max_chunks]
async def _rescore_batch(self, collection_name: str, chunk_ids: list[str]):
# Fetch chunk text and embeddings from Qdrant
# Re-run scorer
# Upsert updated quality metadata
# This is straightforward but verbose — structure mirrors upsert_chunk_with_quality
pass
Job 2: Chunk Retirement and Replacement
Some chunks are beyond rescoring — they're genuinely bad and no amount of quality adjustment will fix them. The retirement job identifies these and either removes them or flags them for human review:
async def retire_low_quality_chunks(
qdrant,
feedback_store,
collection_name: str,
quality_floor: float = 0.20,
min_negative_signals: int = 20,
dry_run: bool = True,
) -> list[str]:
"""
Identifies chunks that meet ALL retirement criteria:
- Composite quality score below quality_floor
- More than min_negative_signals negative feedback events
- Retrieved at least 10 times (enough observations to be confident)
- Not the only chunk from its source document
(we don't want to completely remove a document from the index)
"""
candidates = await feedback_store.get_retirement_candidates(
quality_floor=quality_floor,
min_negative_signals=min_negative_signals,
min_retrievals=10,
)
# Safety check: don't retire chunks if they're the last chunk
# from their source document
safe_to_retire = []
for chunk_id in candidates:
doc_id = await _get_document_id(qdrant, collection_name, chunk_id)
sibling_count = await _count_document_chunks(qdrant, collection_name, doc_id)
if sibling_count > 1:
safe_to_retire.append(chunk_id)
if not dry_run:
qdrant.delete(
collection_name=collection_name,
points_selector=safe_to_retire,
)
return safe_to_retire
Always run retirement jobs in dry_run mode first. Log what would be deleted, review a sample manually, then enable the actual deletion. You should also write retired chunk IDs to a separate audit table before deletion — you'll want to know what was removed and when.
Job 3: Chunk Splitting for High-Performing Dense Chunks
This is the most sophisticated optimization: if a chunk is consistently retrieved but the LLM only uses part of it, it might be too large. Conversely, if a chunk is frequently retrieved alongside a specific sibling chunk, they might belong together.
class ChunkSplitSuggester:
"""
Analyzes usage patterns to suggest chunking improvements.
Does NOT automatically split — outputs suggestions for review.
"""
async def analyze(
self,
chunk_id: str,
chunk_text: str,
usage_stats: dict,
co_retrieval_patterns: dict,
) -> dict:
"""
co_retrieval_patterns: {chunk_id: frequency} of chunks retrieved
alongside this one.
"""
suggestions = []
# Long chunks that frequently co-retrieve with themselves
# (different queries hitting same chunk) might benefit from splitting
word_count = len(chunk_text.split())
if word_count > 400 and usage_stats.get("query_diversity_score", 0) > 0.7:
suggestions.append({
"type": "consider_split",
"reason": "High query diversity against long chunk suggests multiple topics",
"word_count": word_count,
})
# Chunks almost always retrieved together should be merged candidates
top_co_retrieved = sorted(
co_retrieval_patterns.items(), key=lambda x: x[1], reverse=True
)
if top_co_retrieved and top_co_retrieved[0][1] > 0.85:
suggestions.append({
"type": "consider_merge",
"reason": f"Retrieved together with {top_co_retrieved[0][0]} in >85% of cases",
"merge_candidate": top_co_retrieved[0][0],
})
return {"chunk_id": chunk_id, "suggestions": suggestions}
All of the above is useless if running it takes your RAG system offline. Production index optimization requires a deployment strategy that treats the index as a live system, not a batch artifact.
Maintain two collections in your vector store simultaneously: the "live" index serving traffic and a "staging" index being optimized. Swap them atomically when staging is ready:
class BlueGreenIndexManager:
"""
Manages a blue/green deployment pattern for vector index updates.
The active alias always points to whichever collection is currently live.
"""
def __init__(self, qdrant, collection_prefix: str = "rag_knowledge"):
self.qdrant = qdrant
self.prefix = collection_prefix
def get_active_collection(self) -> str:
"""Reads current active collection from a simple key-value store."""
# In practice, use Redis, a config file, or your DB
# Here we check which collection exists and is marked active
pass
async def prepare_staging(self, source_collection: str) -> str:
"""
Creates a new staging collection, copies all vectors from source,
then applies optimization passes.
"""
staging_name = f"{self.prefix}_staging_{uuid.uuid4().hex[:8]}"
# Qdrant supports collection cloning via snapshot API
# For other vector DBs, you'd copy points in batches
await self._clone_collection(source_collection, staging_name)
return staging_name
async def promote_staging(self, staging_name: str):
"""
Atomically swaps staging to live. Old live collection
is kept for a rollback window before deletion.
"""
old_live = self.get_active_collection()
new_archive_name = f"{self.prefix}_archive_{uuid.uuid4().hex[:8]}"
# Rename old live to archive (keep for rollback)
await self._rename_collection(old_live, new_archive_name)
# Rename staging to live
await self._rename_collection(staging_name, f"{self.prefix}_live")
# Schedule archive deletion after rollback window (e.g., 24h)
asyncio.create_task(
self._delete_after_delay(new_archive_name, delay_hours=24)
)
async def rollback(self, archive_name: str):
"""Emergency rollback to previous live collection."""
current_live = self.get_active_collection()
await self._rename_collection(archive_name, f"{self.prefix}_live")
await self._rename_collection(current_live, f"{self.prefix}_rolled_back")
For frequent metadata updates (quality score refreshes, retrieval stat updates) that don't require rebuilding embeddings, use a separate PostgreSQL table as your source of truth and query it as a sidecar to your vector store:
class HybridMetadataStore:
"""
Keeps fast-changing metadata (retrieval stats, feedback scores) in
PostgreSQL while storing stable data (embeddings, text) in Qdrant.
At retrieval time, merges both sources.
"""
async def get_enriched_chunks(
self, chunk_ids: list[str]
) -> dict[str, dict]:
"""Fetches base chunks from Qdrant and enriches with live PG metadata."""
qdrant_data = await self._fetch_from_qdrant(chunk_ids)
pg_metadata = await self._fetch_live_metadata(chunk_ids)
enriched = {}
for cid in chunk_ids:
base = qdrant_data.get(cid, {})
live = pg_metadata.get(cid, {})
enriched[cid] = {**base, **live} # Live metadata wins on conflict
return enriched
async def _fetch_live_metadata(self, chunk_ids: list[str]) -> dict:
# SELECT chunk_id, composite_score, times_retrieved, feedback_score
# FROM chunk_live_metadata WHERE chunk_id = ANY($1)
pass
Objective: Build and test a minimal feedback loop on a real document corpus.
You'll need:
qdrant-client, openai, asyncpg, numpy, pydanticdocker run -p 6333:6333 qdrant/qdrant)If you have an existing RAG index, extract a sample of 500 chunks, run them through the ChunkQualityScorer, and analyze the distribution. What percentage score below 0.45? What are the most common failure modes in your corpus?
async def audit_existing_index(qdrant, collection_name, scorer, sample_size=500):
# Scroll through collection, sample chunks, score them
# Output: histogram of composite scores, examples of low scorers
pass
Wrap your existing retriever with InstrumentedRetriever. For 48 hours, log all retrieval events and LLM signals without making any changes to retrieval behavior. This is your baseline.
At the end of 48 hours, query your feedback store:
SELECT
signal_type,
COUNT(*) as event_count,
AVG(signal_weight) as avg_weight
FROM retrieval_feedback_events
GROUP BY signal_type
ORDER BY event_count DESC;
Enable the build_quality_filter(min_quality=0.45) in your retriever. Run for another 48 hours and compare:
Enable FeedbackAwareReranker after you have at least 500 feedback events. The Wilson score confidence mechanism will ensure chunks without enough observations get neutral scores and aren't artificially promoted or demoted.
Measure the improvement in explicit positive feedback rate over the following week.
Mistake 1: Scoring all dimensions with equal weight regardless of corpus type. Technical documentation is information-dense and context-dependent — context sufficiency should get higher weight. FAQ-style content is naturally self-contained — answerability weight should dominate. Tune weights for your specific corpus.
Mistake 2: Over-retiring chunks based on insufficient feedback. If you retire a chunk after only 5 negative signals, you're being statistically irresponsible. The Wilson confidence mechanism protects against this in reranking, but your retirement job needs its own minimum observation threshold. A chunk that's been retrieved 3 times and got 2 negative signals is not worth retiring — it hasn't had enough exposure.
Mistake 3: Not differentiating between retrieval failure and generation failure. When a user gives explicit negative feedback, it might be because the retrieval was wrong (wrong chunks surfaced) or because the generation was wrong (right chunks, bad synthesis). If you attribute all negative signals to retrieval quality, you'll over-penalize chunks that were actually relevant but surrounded by a bad prompt. Use faithfulness scoring to separate the two.
Mistake 4: Letting the feedback loop create a popularity bias. Chunks about frequently-asked topics accumulate more positive signals and get systematically upranked, while chunks about rare but important topics get fewer observations and stay at neutral scores. This creates a feedback loop that makes common queries excellent and rare queries progressively worse. Combat this by including an "underexplored" bonus in your reranking for chunks with high quality scores but low retrieval counts — they may be undiscovered gems.
def underexplored_bonus(times_retrieved: int, quality_score: float) -> float:
"""Give a small boost to high-quality chunks that haven't been retrieved much."""
if times_retrieved < 10 and quality_score > 0.65:
return 0.05 * (1 - times_retrieved / 10)
return 0.0
Mistake 5: Running batch optimization jobs during peak traffic. Your index optimization jobs read and write heavily. Schedule them during your traffic nadir. If you have global users and no nadir exists, use the blue/green pattern to isolate optimization from live traffic completely.
Mistake 6: Not versioning your scoring logic.
When you update the ChunkQualityScorer with a new version, all historical scores become incomparable. Always include scoring_version in your stored scores and filter comparisons to chunks scored with the same version. Re-scoring is cheap compared to the confusion of mixing score distributions.
Mistake 7: Treating LLM self-reflection as ground truth. "I couldn't find enough information" from an LLM might mean the information genuinely isn't in your index, or it might mean your prompt is poorly calibrated, or it might be an artifact of the LLM being overly cautious. Weight these signals at 0.9 but don't treat them as absolute. Cross-reference with explicit feedback and reformulation rates.
When your corpus grows past ~1M chunks, several things change:
Redundancy scoring becomes expensive. Computing cosine similarity against a full embedding matrix during indexing doesn't scale. Switch to approximate nearest-neighbor approaches for redundancy detection — run a dedicated ANN query for each new chunk and flag candidates above 0.92 similarity. This is what your vector store is designed for.
Feedback aggregation becomes a hot query. As feedback events accumulate (millions per month for active deployments), real-time aggregation becomes expensive. Move to a pre-aggregated model: run an hourly job that materializes chunk-level stats, and serve the reranker from that materialized view rather than computing fresh on every request. The 1-hour staleness is acceptable for feedback signals.
Quality score drift monitoring. As your scoring version evolves, track the distribution of scores over time. Sudden shifts in mean quality score often indicate a corpus-level issue (bulk import of low-quality documents, embedding model version change) that needs human attention. Set up simple alerting: if mean composite quality drops more than 0.05 in 24 hours, alert the on-call engineer.
You've built the conceptual and practical foundation for a RAG system that improves from its own usage. The key architectural insight is that every retrieval event is a training signal, and the infrastructure to capture, process, and act on those signals is just as important as the initial indexing pipeline.
Here's what you've built:
Where to go from here:
Faithfulness scoring: Integrate a dedicated faithfulness evaluator (Ragas, TruLens, or a custom LLM judge) to separate retrieval quality from generation quality in your feedback signals. This is the highest-leverage next step.
Query clustering: You're already hashing query embeddings. The next step is clustering similar queries and analyzing which chunks serve which query clusters well. This enables query-type-specific optimization — your chunks for procedural questions ("how do I...") can be evaluated differently from conceptual questions ("what is...").
Adaptive chunking: Close the loop fully by automatically re-chunking documents based on the split/merge suggestions from ChunkSplitSuggester, then running the full re-indexing pipeline in staging.
Embedding model evaluation: The feedback infrastructure you've built is also a perfect evaluation harness for testing new embedding models. Run a new model in shadow mode — embed queries with both your current model and the candidate — and compare which produces better retrieval outcomes according to your feedback signals. No labeled dataset required.
Multi-index routing: For large corpora covering multiple domains, consider routing queries to domain-specific sub-indexes rather than searching everything. Your query clustering data will reveal the natural domain boundaries.
The difference between a RAG system that's frozen in time and one that compounds its own improvement is this feedback infrastructure. Build it early, instrument it thoroughly, and let the data tell you where to optimize.
Learning Path: Building with LLMs