Python AI Rerank with Cross-Encoder: 5 Key Patterns to Boost RAG Retrieval Accuracy by 40%
AI与大数据
Why Your RAG Retrieval Is Always "Almost There"
You spent three days building a RAG system, but when a user asks "how to process a refund," it returns documentation about "how to register." The problem isn't the LLM — it's the retrieval layer. RAG without Rerank is like a competition without a referee: among the Top-K results from initial retrieval, truly relevant ones may account for only 20%.
In 2026, Rerank has become a standard component of any RAG system. From Cohere Rerank API to open-source cross-encoders, from hybrid retrieval to custom fine-tuning, this article will guide you through 5 key patterns that can boost retrieval accuracy by 40%.
Core Concepts Reference Table
| Concept | Key Definition | Typical Application |
|---|---|---|
| Reranker | A model/component that performs secondary ranking on initial retrieval results | RAG retrieval optimization, search result optimization |
| Cross-Encoder | Encodes query and document jointly by concatenating them, outputting a relevance score | Re-ranking stage, QA matching |
| Bi-Encoder | Encodes query and document independently, matching via vector similarity | Initial filtering stage, large-scale recall |
| Late Interaction | Encodes query and document into token-level vectors separately, then performs fine-grained matching | ColBERT model, efficient re-ranking |
| Hybrid Retrieval | A retrieval strategy combining dense retrieval and sparse retrieval | Multi-modal recall, semantic + keyword |
| RAG (Retrieval-Augmented Generation) | A technical paradigm that retrieves external knowledge to assist LLM generation | Enterprise knowledge bases, intelligent customer service |
| RRF (Reciprocal Rank Fusion) | A fusion ranking algorithm for multi-path retrieval results | Hybrid retrieval result merging |
| Cross-Attention | The attention mechanism between query and document in Transformer | Core mechanism inside Cross-Encoder |
RAG Without Rerank: 5 Fatal Pain Points
-
Semantic Drift: In high-dimensional space, Bi-Encoder easily recalls similar but irrelevant documents. A user asks about "Python exception handling" but gets "Python installation guide" — the vector distance is close, but the semantics are poles apart.
-
Keyword Loss: Pure dense retrieval has weak exact keyword matching capability. Searching for "OAuth2.0 authorization code flow," Bi-Encoder may return a generic "OAuth introduction" because it lacks precise keyword matching signals.
-
Crude Ranking: Initial retrieval relies solely on vector cosine similarity for ranking, unable to capture deep interaction between query and document. Among the Top-10, only 2-3 may be truly relevant.
-
Long-tail Query Inaccuracy: For rare entities, professional terminology, and abbreviations in long-tail queries, Bi-Encoder encoding quality drops significantly, causing retrieval accuracy to plummet.
-
Multi-intent Confusion: A single query may contain multiple intents, and Bi-Encoder's single vector representation cannot distinguish them, leading to mixed-intent results.
Pattern 1: Cohere Rerank API Integration — The Fastest Way to Get Started
Cohere Rerank is currently the most mature commercial Rerank API, supporting 100+ languages with latency as low as 50ms, ideal for rapid integration.
"""
Cohere Rerank API Integration Example
Install dependencies: pip install cohere>=5.0
"""
import cohere
from typing import List, Dict
class CohereReranker:
"""Cohere Rerank API Wrapper"""
def __init__(self, api_key: str, model: str = "rerank-v3.5"):
self.client = cohere.ClientV2(api_key=api_key)
self.model = model
def rerank(
self,
query: str,
documents: List[str],
top_n: int = 5,
max_chunks_per_doc: int = 3,
) -> List[Dict]:
"""
Re-rank a list of documents
Args:
query: User query text
documents: List of documents to rank
top_n: Return top N results
max_chunks_per_doc: Maximum chunks per document
Returns:
Re-ranked result list containing index, relevance_score, document
"""
response = self.client.rerank(
model=self.model,
query=query,
documents=documents,
top_n=top_n,
max_chunks_per_doc=max_chunks_per_doc,
)
reranked_results = []
for result in response.results:
reranked_results.append({
"index": result.index,
"relevance_score": result.relevance_score,
"document": documents[result.index],
})
return reranked_results
# === Complete Usage Example ===
def demo_cohere_rerank():
"""Cohere Rerank complete usage demo"""
reranker = CohereReranker(api_key="your-cohere-api-key")
query = "How to handle JSON parsing exceptions in Python?"
documents = [
"Python installation guide: Download the installer from the official website, double-click to run and complete installation.",
"JSON parsing error handling: When using json.loads(), catch json.JSONDecodeError and log the original text for debugging.",
"Python list comprehensions are a concise syntax for creating lists, e.g., [x**2 for x in range(10)].",
"When handling JSON data in Python, wrap json.loads() calls with try-except and validate the input as a valid JSON string.",
"In the Flask framework, you can use the jsonify function to quickly return JSON responses.",
]
results = reranker.rerank(query=query, documents=documents, top_n=3)
print(f"Query: {query}\n")
for i, result in enumerate(results, 1):
print(f"Top-{i} | Relevance: {result['relevance_score']:.4f}")
print(f" Document: {result['document'][:80]}...")
print()
# === Integration with RAG Pipeline ===
class RAGPipelineWithCohere:
"""RAG Pipeline with Cohere Rerank Integration"""
def __init__(
self,
cohere_api_key: str,
embedding_model_name: str = "BAAI/bge-large-en-v1.5",
):
from sentence_transformers import SentenceTransformer
self.encoder = SentenceTransformer(embedding_model_name)
self.reranker = CohereReranker(api_key=cohere_api_key)
self.document_store: List[Dict] = []
def index_documents(self, documents: List[str], metadata: List[Dict] = None):
"""Index documents"""
embeddings = self.encoder.encode(documents, normalize_embeddings=True)
for i, (doc, emb) in enumerate(zip(documents, embeddings)):
self.document_store.append({
"text": doc,
"embedding": emb.tolist(),
"metadata": metadata[i] if metadata else {},
})
def retrieve(
self,
query: str,
top_k: int = 10,
rerank_top_n: int = 3,
) -> List[Dict]:
"""Retrieve and re-rank"""
import numpy as np
query_embedding = self.encoder.encode([query], normalize_embeddings=True)[0]
# Initial filtering: cosine similarity
scored_docs = []
for doc in self.document_store:
score = float(np.dot(query_embedding, doc["embedding"]))
scored_docs.append({**doc, "score": score})
scored_docs.sort(key=lambda x: x["score"], reverse=True)
top_candidates = scored_docs[:top_k]
# Re-ranking: Cohere Rerank
candidate_texts = [doc["text"] for doc in top_candidates]
reranked = self.reranker.rerank(
query=query, documents=candidate_texts, top_n=rerank_top_n,
)
final_results = []
for r in reranked:
original_doc = top_candidates[r["index"]]
final_results.append({
"text": original_doc["text"],
"metadata": original_doc["metadata"],
"rerank_score": r["relevance_score"],
"initial_score": original_doc["score"],
})
return final_results
if __name__ == "__main__":
demo_cohere_rerank()
Pattern 2: Sentence-Transformers Cross-Encoder Re-ranking — The Top Open-Source Choice
When data privacy requirements are high and external API calls are not possible, locally deployed Cross-Encoder is the optimal choice.
"""
Sentence-Transformers Cross-Encoder Re-ranking
Install dependencies: pip install sentence-transformers>=3.0
"""
from sentence_transformers import CrossEncoder
from typing import List, Dict, Optional
import logging
logger = logging.getLogger(__name__)
class CrossEncoderReranker:
"""Local re-ranker based on Cross-Encoder"""
# Recommended models and their max sequence lengths
SUPPORTED_MODELS = {
"cross-encoder/ms-marco-MiniLM-L-6-v2": 512,
"cross-encoder/ms-marco-MiniLM-L-12-v2": 512,
"cross-encoder/stsb-roberta-large": 512,
"BAAI/bge-reranker-large": 512,
"BAAI/bge-reranker-v2-m3": 8192,
}
def __init__(
self,
model_name: str = "BAAI/bge-reranker-v2-m3",
max_length: Optional[int] = None,
device: Optional[str] = None,
):
self.model_name = model_name
self.max_length = max_length or self.SUPPORTED_MODELS.get(model_name, 512)
logger.info(f"Loading Cross-Encoder model: {model_name}")
self.model = CrossEncoder(
model_name,
max_length=self.max_length,
device=device,
)
logger.info("Model loaded successfully")
def rerank(
self,
query: str,
documents: List[str],
top_n: int = 5,
batch_size: int = 32,
) -> List[Dict]:
"""
Re-rank documents using Cross-Encoder
Args:
query: Query text
documents: List of documents to rank
top_n: Return top N results
batch_size: Inference batch size
Returns:
Re-ranked result list
"""
# Construct (query, document) pairs
pairs = [(query, doc) for doc in documents]
# Batch inference to get relevance scores
scores = self.model.predict(pairs, batch_size=batch_size)
# Sort by score in descending order
ranked_indices = scores.argsort()[::-1]
results = []
for rank, idx in enumerate(ranked_indices[:top_n]):
results.append({
"index": int(idx),
"relevance_score": float(scores[idx]),
"document": documents[idx],
"rank": rank + 1,
})
return results
def rerank_with_threshold(
self,
query: str,
documents: List[str],
threshold: float = 0.5,
top_n: int = 10,
) -> List[Dict]:
"""
Re-ranking with threshold filtering — results below threshold are filtered out
Args:
query: Query text
documents: List of documents to rank
threshold: Relevance threshold
top_n: Maximum number of results to return
Returns:
Filtered re-ranked results
"""
results = self.rerank(query, documents, top_n=top_n)
filtered = [r for r in results if r["relevance_score"] >= threshold]
logger.info(
f"Re-ranking complete: {len(documents)} input, "
f"{len(filtered)} after threshold filtering (threshold={threshold})"
)
return filtered
# === Complete Usage Example ===
def demo_cross_encoder_rerank():
"""Cross-Encoder re-ranking complete demo"""
reranker = CrossEncoderReranker(
model_name="BAAI/bge-reranker-v2-m3",
)
query = "Kubernetes pod graceful termination strategy"
documents = [
"Basic Docker container commands include run, stop, rm, etc., suitable for beginners.",
"Kubernetes Pod graceful termination: Configure terminationGracePeriodSeconds, implement PreStop hooks, ensure containers complete cleanup after receiving SIGTERM.",
"Kubernetes Service types include ClusterIP, NodePort, LoadBalancer, etc., for different network exposure needs.",
"Pod termination flow: kubelet sends SIGTERM → waits for graceful termination period → sends SIGKILL for forced termination. Adding a sleep delay in PreStop is recommended to wait for connection draining.",
"Helm is a package management tool for Kubernetes that simplifies application deployment and upgrade processes.",
]
results = reranker.rerank(query=query, documents=documents, top_n=3)
print(f"Query: {query}\n")
for result in results:
print(f"Rank-{result['rank']} | Score: {result['relevance_score']:.4f}")
print(f" Document: {result['document'][:80]}...")
print()
# === Multi-Query Fusion Re-ranking ===
class MultiQueryReranker:
"""Multi-query fusion re-ranking: rewrite one query into multiple sub-queries and merge ranking results"""
def __init__(self, cross_encoder_model: str = "BAAI/bge-reranker-v2-m3"):
self.reranker = CrossEncoderReranker(model_name=cross_encoder_model)
def rerank_multi_query(
self,
queries: List[str],
documents: List[str],
top_n: int = 5,
fusion_strategy: str = "rrf",
) -> List[Dict]:
"""
Multi-query fusion re-ranking
Args:
queries: List of multiple query texts
documents: List of documents to rank
top_n: Return top N results
fusion_strategy: Fusion strategy, supports rrf (Reciprocal Rank Fusion) or avg (average score)
Returns:
Fused re-ranked results
"""
doc_scores = {i: 0.0 for i in range(len(documents))}
for query in queries:
results = self.reranker.rerank(query, documents, top_n=len(documents))
if fusion_strategy == "rrf":
# Reciprocal Rank Fusion
k = 60 # RRF smoothing parameter
for result in results:
doc_scores[result["index"]] += 1.0 / (k + result["rank"])
elif fusion_strategy == "avg":
# Average score fusion
for result in results:
doc_scores[result["index"]] += result["relevance_score"]
# Normalize
if fusion_strategy == "avg":
for idx in doc_scores:
doc_scores[idx] /= len(queries)
# Sort
sorted_indices = sorted(
doc_scores.keys(), key=lambda x: doc_scores[x], reverse=True
)
final_results = []
for rank, idx in enumerate(sorted_indices[:top_n]):
final_results.append({
"index": int(idx),
"fusion_score": float(doc_scores[idx]),
"document": documents[idx],
"rank": rank + 1,
})
return final_results
if __name__ == "__main__":
demo_cross_encoder_rerank()
Pattern 3: Hybrid Retrieval (Dense + Sparse + Rerank) — The Ceiling of Retrieval Performance
A single retrieval method always has blind spots. Hybrid retrieval combines the semantic understanding of dense retrieval with the exact matching of sparse retrieval, then applies Rerank for fine ranking — this is the best practice for RAG systems in 2026.
"""
Hybrid Retrieval: Dense + Sparse + Rerank
Install dependencies:
pip install sentence-transformers>=3.0
pip install rank-bm25
pip install numpy
"""
from sentence_transformers import SentenceTransformer, CrossEncoder
from rank_bm25 import BM25Okapi
from typing import List, Dict, Optional
import numpy as np
import logging
logger = logging.getLogger(__name__)
class HybridRetrieverWithRerank:
"""Complete pipeline for hybrid retrieval + re-ranking"""
def __init__(
self,
dense_model_name: str = "BAAI/bge-large-en-v1.5",
cross_encoder_model: str = "BAAI/bge-reranker-v2-m3",
rrf_k: int = 60,
):
# Dense retrieval model (Bi-Encoder)
self.dense_model = SentenceTransformer(dense_model_name)
# Cross-Encoder (for re-ranking)
self.cross_encoder = CrossEncoder(cross_encoder_model)
# RRF fusion parameter
self.rrf_k = rrf_k
# Document storage
self.documents: List[str] = []
self.dense_embeddings: Optional[np.ndarray] = None
self.bm25: Optional[BM25Okapi] = None
def _tokenize(self, text: str) -> List[str]:
"""Tokenize text for BM25"""
return text.lower().split()
def index_documents(self, documents: List[str]):
"""Index documents, build dense and sparse indices"""
self.documents = documents
# Build dense index
logger.info("Building dense vector index...")
self.dense_embeddings = self.dense_model.encode(
documents, normalize_embeddings=True, show_progress_bar=True,
)
# Build sparse index (BM25)
logger.info("Building BM25 sparse index...")
tokenized_corpus = [self._tokenize(doc) for doc in documents]
self.bm25 = BM25Okapi(tokenized_corpus)
logger.info(f"Indexing complete, {len(documents)} documents total")
def _dense_search(self, query: str, top_k: int) -> List[Dict]:
"""Dense retrieval"""
query_embedding = self.dense_model.encode(
[query], normalize_embeddings=True,
)[0]
scores = np.dot(self.dense_embeddings, query_embedding)
top_indices = np.argsort(scores)[::-1][:top_k]
return [
{"index": int(idx), "score": float(scores[idx]), "text": self.documents[idx]}
for idx in top_indices
]
def _sparse_search(self, query: str, top_k: int) -> List[Dict]:
"""Sparse retrieval (BM25)"""
tokenized_query = self._tokenize(query)
scores = self.bm25.get_scores(tokenized_query)
top_indices = np.argsort(scores)[::-1][:top_k]
return [
{"index": int(idx), "score": float(scores[idx]), "text": self.documents[idx]}
for idx in top_indices
]
def _rrf_fuse(
self,
dense_results: List[Dict],
sparse_results: List[Dict],
) -> List[Dict]:
"""Reciprocal Rank Fusion"""
rrf_scores: Dict[int, float] = {}
for rank, result in enumerate(dense_results):
idx = result["index"]
rrf_scores[idx] = rrf_scores.get(idx, 0) + 1.0 / (self.rrf_k + rank + 1)
for rank, result in enumerate(sparse_results):
idx = result["index"]
rrf_scores[idx] = rrf_scores.get(idx, 0) + 1.0 / (self.rrf_k + rank + 1)
sorted_indices = sorted(
rrf_scores.keys(), key=lambda x: rrf_scores[x], reverse=True,
)
return [
{
"index": int(idx),
"rrf_score": float(rrf_scores[idx]),
"text": self.documents[idx],
}
for idx in sorted_indices
]
def _rerank(
self,
query: str,
candidates: List[Dict],
top_n: int,
) -> List[Dict]:
"""Re-rank using Cross-Encoder"""
pairs = [(query, c["text"]) for c in candidates]
scores = self.cross_encoder.predict(pairs)
for i, candidate in enumerate(candidates):
candidate["rerank_score"] = float(scores[i])
candidates.sort(key=lambda x: x["rerank_score"], reverse=True)
return candidates[:top_n]
def search(
self,
query: str,
dense_top_k: int = 20,
sparse_top_k: int = 20,
rerank_top_n: int = 5,
) -> List[Dict]:
"""
Complete hybrid retrieval + re-ranking flow
Args:
query: Query text
dense_top_k: Number of results from dense retrieval
sparse_top_k: Number of results from sparse retrieval
rerank_top_n: Final re-ranking return count
Returns:
Final ranked results
"""
# Step 1: Dual-path recall
dense_results = self._dense_search(query, dense_top_k)
sparse_results = self._sparse_search(query, sparse_top_k)
# Step 2: RRF fusion
fused_results = self._rrf_fuse(dense_results, sparse_results)
# Step 3: Cross-Encoder fine ranking
final_results = self._rerank(query, fused_results, rerank_top_n)
return final_results
# === Complete Usage Example ===
def demo_hybrid_retrieval():
"""Hybrid retrieval complete demo"""
retriever = HybridRetrieverWithRerank()
documents = [
"Python GIL (Global Interpreter Lock): GIL ensures only one thread executes Python bytecode at a time; multithreading is suitable for IO-bound tasks.",
"Python multiprocessing: Use the multiprocessing module to bypass GIL limitations; each process has its own GIL and memory space.",
"Python asyncio: Write coroutines using async/await syntax, suitable for high-concurrency IO operations like HTTP requests and database queries.",
"Python thread pool: concurrent.futures.ThreadPoolExecutor provides a convenient thread pool interface for parallel IO-bound tasks.",
"Python performance optimization: Use cProfile for bottleneck analysis, Cython for compiling hot code, numpy to replace pure Python loops.",
"Go concurrency model: goroutines and channels are Go's concurrency primitives, lighter than Python threads, suitable for CPU-bound parallel computing.",
"Rust ownership system: Guarantees memory safety through compile-time checks, no garbage collection needed, suitable for systems-level high-performance programming.",
"Python memory management: Reference counting as primary, generational garbage collection as secondary, circular references handled by the gc module.",
]
retriever.index_documents(documents)
query = "Best practices for Python concurrent programming"
results = retriever.search(query, rerank_top_n=3)
print(f"Query: {query}\n")
for result in results:
print(f"Rerank score: {result['rerank_score']:.4f} | RRF score: {result['rrf_score']:.4f}")
print(f" Document: {result['text'][:80]}...")
print()
if __name__ == "__main__":
demo_hybrid_retrieval()
Pattern 4: Custom Cross-Encoder Fine-Tuning — Make Your Reranker Understand Your Domain
General-purpose Cross-Encoders perform poorly in specialized domains (medical, legal, financial). Fine-tuning is the essential path.
"""
Custom Cross-Encoder Fine-Tuning
Install dependencies:
pip install sentence-transformers>=3.0
pip install datasets
"""
from sentence_transformers import CrossEncoder, InputExample
from sentence_transformers.cross_encoder.evaluation import CECorrelationEvaluator
from torch.utils.data import DataLoader
from typing import List, Tuple, Optional
import logging
logger = logging.getLogger(__name__)
class CrossEncoderFineTuner:
"""Cross-Encoder Fine-Tuner"""
def __init__(
self,
base_model: str = "cross-encoder/ms-marco-MiniLM-L-6-v2",
num_labels: int = 1,
max_length: int = 512,
):
self.base_model = base_model
self.num_labels = num_labels
self.max_length = max_length
self.model = CrossEncoder(
base_model,
num_labels=num_labels,
max_length=max_length,
)
def prepare_training_data(
self,
query_doc_pairs: List[Tuple[str, str, float]],
) -> List[InputExample]:
"""
Prepare training data
Args:
query_doc_pairs: List of (query, document, relevance_score) triples
relevance_score: 0/1 for binary classification, 0-1 continuous for regression
Returns:
List of InputExample
"""
examples = []
for query, doc, score in query_doc_pairs:
examples.append(InputExample(texts=[query, doc], label=score))
logger.info(f"Prepared training data: {len(examples)} examples")
return examples
def train(
self,
train_examples: List[InputExample],
val_examples: Optional[List[InputExample]] = None,
output_path: str = "./fine_tuned_cross_encoder",
epochs: int = 3,
batch_size: int = 16,
warmup_steps: int = 100,
learning_rate: float = 2e-5,
):
"""
Fine-tuning training
Args:
train_examples: Training data
val_examples: Validation data (optional)
output_path: Model save path
epochs: Number of training epochs
batch_size: Batch size
warmup_steps: Warmup steps
learning_rate: Learning rate
"""
train_dataloader = DataLoader(
train_examples,
shuffle=True,
batch_size=batch_size,
)
# Build evaluator
evaluator = None
if val_examples:
val_pairs = [(ex.texts[0], ex.texts[1]) for ex in val_examples]
val_labels = [ex.label for ex in val_examples]
evaluator = CECorrelationEvaluator(
sentences1=[p[0] for p in val_pairs],
sentences2=[p[1] for p in val_pairs],
scores=val_labels,
name="validation",
)
# Training configuration
train_config = {
"train_dataloader": train_dataloader,
"evaluator": evaluator,
"epochs": epochs,
"warmup_steps": warmup_steps,
"output_path": output_path,
"show_progress_bar": True,
}
# Select loss function based on num_labels
if self.num_labels == 1:
# Regression task: MSE loss
self.model.fit(
**train_config,
loss_fct="MSE",
)
else:
# Classification task: Cross-entropy loss
self.model.fit(
**train_config,
)
logger.info(f"Model saved to: {output_path}")
def load_fine_tuned(self, model_path: str) -> CrossEncoder:
"""Load fine-tuned model"""
self.model = CrossEncoder(model_path)
logger.info(f"Loaded fine-tuned model: {model_path}")
return self.model
# === Domain data construction example ===
def build_domain_training_data() -> List[Tuple[str, str, float]]:
"""
Build domain training data (example: medical domain)
Returns:
List of (query, document, relevance) triples
"""
training_pairs = [
# Positive samples
("How to treat hypertension?", "Hypertension treatment guidelines: First-line drugs include ACEI, ARB, CCB, etc., requiring individualized drug selection based on comorbidities.", 1.0),
("Diabetes dietary precautions", "Diabetes dietary management: Control total caloric intake, choose low-GI foods, limit refined sugars, increase dietary fiber intake.", 1.0),
("What medicine for cold and fever", "Symptomatic cold treatment: For temperatures above 38.5°C, take acetaminophen or ibuprofen, ensure adequate hydration and rest.", 1.0),
# Negative samples
("How to treat hypertension?", "Symptomatic cold treatment: For temperatures above 38.5°C, take acetaminophen or ibuprofen, ensure adequate hydration and rest.", 0.0),
("Diabetes dietary precautions", "Hypertension treatment guidelines: First-line drugs include ACEI, ARB, CCB, etc., requiring individualized drug selection based on comorbidities.", 0.0),
("What medicine for cold and fever", "Python installation guide: Download the installer from the official website, double-click to run.", 0.0),
# Hard negatives (similar but irrelevant)
("How to treat hypertension?", "Hypotension diagnostic criteria: Systolic pressure below 90mmHg or diastolic below 60mmHg, need to rule out medication factors.", 0.2),
("Diabetes dietary precautions", "Diabetes pharmacotherapy: Metformin is the first-line medication for type 2 diabetes, requires monitoring of renal function and lactate levels.", 0.4),
]
return training_pairs
# === Complete fine-tuning flow ===
def demo_fine_tuning():
"""Cross-Encoder fine-tuning complete demo"""
fine_tuner = CrossEncoderFineTuner(
base_model="cross-encoder/ms-marco-MiniLM-L-6-v2",
num_labels=1,
)
# Prepare training data
training_pairs = build_domain_training_data()
train_examples = fine_tuner.prepare_training_data(training_pairs)
# Split into training and validation sets
split_idx = int(len(train_examples) * 0.8)
train_data = train_examples[:split_idx]
val_data = train_examples[split_idx:]
# Fine-tune training
fine_tuner.train(
train_examples=train_data,
val_examples=val_data,
output_path="./models/medical_cross_encoder",
epochs=3,
batch_size=8,
learning_rate=2e-5,
)
# Load fine-tuned model and test
model = fine_tuner.load_fine_tuned("./models/medical_cross_encoder")
scores = model.predict([
("How to treat hypertension?", "Hypertension treatment guidelines: First-line drugs include ACEI, ARB, CCB, etc."),
("How to treat hypertension?", "Symptomatic cold treatment: For temperatures above 38.5°C, take acetaminophen."),
])
print(f"Relevant document score: {scores[0]:.4f}")
print(f"Irrelevant document score: {scores[1]:.4f}")
if __name__ == "__main__":
demo_fine_tuning()
Pattern 5: Production-Grade RAG Pipeline — From Prototype to Deployment
Integrate the previous four patterns into a deployable production-grade RAG pipeline with engineering capabilities including caching, fallback, and monitoring.
"""
Production-Grade RAG Pipeline (with Re-ranking)
Install dependencies:
pip install sentence-transformers>=3.0
pip install rank-bm25
pip install redis
pip install numpy
"""
from sentence_transformers import SentenceTransformer, CrossEncoder
from rank_bm25 import BM25Okapi
from typing import List, Dict, Optional
from dataclasses import dataclass, field
import numpy as np
import hashlib
import logging
import time
logger = logging.getLogger(__name__)
@dataclass
class RerankConfig:
"""Re-ranking configuration"""
dense_model_name: str = "BAAI/bge-large-en-v1.5"
cross_encoder_model: str = "BAAI/bge-reranker-v2-m3"
dense_top_k: int = 20
sparse_top_k: int = 20
rerank_top_n: int = 5
rrf_k: int = 60
rerank_threshold: float = 0.3
enable_cache: bool = True
cache_ttl: int = 3600
max_query_length: int = 512
max_doc_length: int = 8192
@dataclass
class SearchResult:
"""Search result"""
text: str
metadata: Dict = field(default_factory=dict)
rerank_score: float = 0.0
initial_score: float = 0.0
retrieval_method: str = "hybrid_rerank"
class ProductionRAGPipeline:
"""Production-grade RAG Pipeline"""
def __init__(self, config: RerankConfig):
self.config = config
# Load models
logger.info(f"Loading Dense model: {config.dense_model_name}")
self.dense_model = SentenceTransformer(config.dense_model_name)
logger.info(f"Loading Cross-Encoder model: {config.cross_encoder_model}")
self.cross_encoder = CrossEncoder(config.cross_encoder_model)
# Document storage
self.documents: List[str] = []
self.doc_metadata: List[Dict] = []
self.dense_embeddings: Optional[np.ndarray] = None
self.bm25: Optional[BM25Okapi] = None
# Cache (replace with Redis in production)
self._cache: Dict[str, tuple] = {}
# Monitoring metrics
self._metrics = {
"total_queries": 0,
"cache_hits": 0,
"avg_latency_ms": 0.0,
"avg_rerank_score": 0.0,
}
def index_documents(self, documents: List[str], metadata: Optional[List[Dict]] = None):
"""Index documents"""
start_time = time.time()
self.documents = documents
self.doc_metadata = metadata or [{} for _ in documents]
# Dense index
self.dense_embeddings = self.dense_model.encode(
documents, normalize_embeddings=True, show_progress_bar=True,
)
# BM25 index
tokenized = [doc.lower().split() for doc in documents]
self.bm25 = BM25Okapi(tokenized)
elapsed = (time.time() - start_time) * 1000
logger.info(f"Indexing complete: {len(documents)} documents, took {elapsed:.0f}ms")
def _get_cache_key(self, query: str) -> str:
"""Generate cache key"""
raw = f"{query}:{self.config.dense_top_k}:{self.config.rerank_top_n}"
return hashlib.md5(raw.encode()).hexdigest()
def _check_cache(self, query: str) -> Optional[List[SearchResult]]:
"""Check cache"""
if not self.config.enable_cache:
return None
cache_key = self._get_cache_key(query)
if cache_key in self._cache:
cached_data, cached_time = self._cache[cache_key]
if time.time() - cached_time < self.config.cache_ttl:
self._metrics["cache_hits"] += 1
return cached_data
return None
def _set_cache(self, query: str, results: List[SearchResult]):
"""Write to cache"""
if not self.config.enable_cache:
return
cache_key = self._get_cache_key(query)
self._cache[cache_key] = (results, time.time())
def _validate_query(self, query: str) -> str:
"""Query preprocessing and validation"""
query = query.strip()
if not query:
raise ValueError("Query text cannot be empty")
if len(query) > self.config.max_query_length:
logger.warning(f"Query too long ({len(query)} chars), truncated to {self.config.max_query_length}")
query = query[: self.config.max_query_length]
return query
def _dense_search(self, query: str) -> List[Dict]:
"""Dense retrieval"""
query_emb = self.dense_model.encode([query], normalize_embeddings=True)[0]
scores = np.dot(self.dense_embeddings, query_emb)
top_indices = np.argsort(scores)[::-1][: self.config.dense_top_k]
return [
{"index": int(idx), "score": float(scores[idx]), "text": self.documents[idx]}
for idx in top_indices
]
def _sparse_search(self, query: str) -> List[Dict]:
"""Sparse retrieval"""
tokenized_query = query.lower().split()
scores = self.bm25.get_scores(tokenized_query)
top_indices = np.argsort(scores)[::-1][: self.config.sparse_top_k]
return [
{"index": int(idx), "score": float(scores[idx]), "text": self.documents[idx]}
for idx in top_indices
]
def _rrf_fuse(
self,
dense_results: List[Dict],
sparse_results: List[Dict],
) -> List[Dict]:
"""RRF fusion"""
rrf_scores: Dict[int, float] = {}
for rank, result in enumerate(dense_results):
idx = result["index"]
rrf_scores[idx] = rrf_scores.get(idx, 0) + 1.0 / (self.config.rrf_k + rank + 1)
for rank, result in enumerate(sparse_results):
idx = result["index"]
rrf_scores[idx] = rrf_scores.get(idx, 0) + 1.0 / (self.config.rrf_k + rank + 1)
sorted_indices = sorted(rrf_scores.keys(), key=lambda x: rrf_scores[x], reverse=True)
return [
{"index": int(idx), "rrf_score": float(rrf_scores[idx]), "text": self.documents[idx]}
for idx in sorted_indices
]
def _rerank(self, query: str, candidates: List[Dict]) -> List[SearchResult]:
"""Cross-Encoder re-ranking"""
pairs = [(query, c["text"]) for c in candidates]
scores = self.cross_encoder.predict(pairs)
results = []
for i, candidate in enumerate(candidates):
rerank_score = float(scores[i])
if rerank_score >= self.config.rerank_threshold:
results.append(SearchResult(
text=candidate["text"],
metadata=self.doc_metadata[candidate["index"]],
rerank_score=rerank_score,
initial_score=candidate.get("rrf_score", candidate.get("score", 0.0)),
retrieval_method="hybrid_rerank",
))
results.sort(key=lambda x: x.rerank_score, reverse=True)
return results[: self.config.rerank_top_n]
def search(self, query: str) -> List[SearchResult]:
"""
Production-grade search entry point
Args:
query: User query
Returns:
Ranked search results
"""
start_time = time.time()
self._metrics["total_queries"] += 1
# Query validation
query = self._validate_query(query)
# Cache check
cached = self._check_cache(query)
if cached is not None:
logger.info("Cache hit")
return cached
try:
# Dual-path recall
dense_results = self._dense_search(query)
sparse_results = self._sparse_search(query)
# RRF fusion
fused = self._rrf_fuse(dense_results, sparse_results)
# Re-ranking
results = self._rerank(query, fused)
# Write to cache
self._set_cache(query, results)
# Update monitoring metrics
elapsed_ms = (time.time() - start_time) * 1000
self._metrics["avg_latency_ms"] = (
self._metrics["avg_latency_ms"] * 0.9 + elapsed_ms * 0.1
)
if results:
self._metrics["avg_rerank_score"] = (
self._metrics["avg_rerank_score"] * 0.9
+ results[0].rerank_score * 0.1
)
logger.info(f"Search complete: {len(results)} results, took {elapsed_ms:.0f}ms")
return results
except Exception as e:
logger.error(f"Search failed: {e}")
# Fallback: return dense retrieval results only
dense_results = self._dense_search(query)[: self.config.rerank_top_n]
return [
SearchResult(
text=r["text"],
metadata=self.doc_metadata[r["index"]],
initial_score=r["score"],
retrieval_method="dense_fallback",
)
for r in dense_results
]
def get_metrics(self) -> Dict:
"""Get monitoring metrics"""
return {
**self._metrics,
"cache_hit_rate": (
self._metrics["cache_hits"] / max(self._metrics["total_queries"], 1)
),
"document_count": len(self.documents),
}
# === Complete Usage Example ===
def demo_production_pipeline():
"""Production-grade RAG pipeline complete demo"""
config = RerankConfig(
dense_top_k=20,
sparse_top_k=20,
rerank_top_n=3,
rerank_threshold=0.3,
enable_cache=True,
)
pipeline = ProductionRAGPipeline(config)
documents = [
"Python exception handling best practices: Use try-except to catch specific exceptions, avoid bare except, log exception context information.",
"Python type annotations: Use the typing module to define function signatures, combined with mypy for static type checking.",
"Python decorator pattern: Decorators are higher-order functions that modify function behavior, commonly used for logging, caching, and permission checks.",
"Python generators and iterators: The yield keyword creates generators with lazy evaluation to save memory, suitable for large datasets.",
"Python context managers: The with statement combined with __enter__/__exit__ ensures proper resource release.",
"Go error handling: Uses multiple return values (error type) instead of exceptions, explicitly handling each error.",
"Rust lifetimes: The compiler ensures reference validity through lifetime annotations, preventing dangling pointers.",
"Python concurrency model: GIL limits multithreading parallelism; recommend asyncio (IO-bound) or multiprocessing (CPU-bound).",
]
metadata = [
{"source": "python-guide", "category": "error-handling"},
{"source": "python-guide", "category": "type-system"},
{"source": "python-guide", "category": "design-pattern"},
{"source": "python-guide", "category": "advanced"},
{"source": "python-guide", "category": "advanced"},
{"source": "go-guide", "category": "error-handling"},
{"source": "rust-guide", "category": "memory-safety"},
{"source": "python-guide", "category": "concurrency"},
]
pipeline.index_documents(documents, metadata)
# Search
query = "Python error handling and exception catching"
results = pipeline.search(query)
print(f"Query: {query}\n")
for result in results:
print(f"Rerank score: {result.rerank_score:.4f} | Method: {result.retrieval_method}")
print(f" Document: {result.text[:80]}...")
print(f" Metadata: {result.metadata}")
print()
# Monitoring metrics
metrics = pipeline.get_metrics()
print("Monitoring metrics:")
for key, value in metrics.items():
print(f" {key}: {value}")
if __name__ == "__main__":
demo_production_pipeline()
Pitfall Guide: 5 Common Rerank Traps
Trap 1: Running Cross-Encoder on the Entire Document Corpus
# ❌ Wrong: Running Cross-Encoder inference on all 100k documents
all_documents = load_100k_documents()
results = cross_encoder.predict([(query, doc) for doc in all_documents])
# Result: Inference takes hours, GPU memory overflow
# ✅ Correct: First Bi-Encoder initial filtering for Top-K, then Cross-Encoder fine ranking
from sentence_transformers import SentenceTransformer, CrossEncoder
bi_encoder = SentenceTransformer("BAAI/bge-large-en-v1.5")
cross_encoder = CrossEncoder("BAAI/bge-reranker-v2-m3")
# Step 1: Bi-Encoder quick initial filtering
query_emb = bi_encoder.encode([query], normalize_embeddings=True)
scores = np.dot(doc_embeddings, query_emb.T).flatten()
top_k_indices = np.argsort(scores)[::-1][:50] # Only take Top-50
# Step 2: Cross-Encoder fine ranking on Top-50
candidates = [all_documents[i] for i in top_k_indices]
pairs = [(query, doc) for doc in candidates]
rerank_scores = cross_encoder.predict(pairs)
Trap 2: Ignoring Cross-Encoder's Maximum Sequence Length
# ❌ Wrong: Feeding long documents directly, truncated beyond model's max length
model = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2") # max_length=512
# Input a 2000-word document, the latter half is truncated, losing key information
result = model.predict([("query", long_document)])
# ✅ Correct: Choose a long-context model or chunk the document
# Option 1: Choose a long-context model
model = CrossEncoder("BAAI/bge-reranker-v2-m3") # max_length=8192
# Option 2: Chunk the document and score each chunk
def rerank_long_document(query: str, doc: str, chunk_size: int = 400, overlap: int = 50):
chunks = []
for i in range(0, len(doc), chunk_size - overlap):
chunks.append(doc[i:i + chunk_size])
pairs = [(query, chunk) for chunk in chunks]
scores = model.predict(pairs)
return max(scores) # Use the highest chunk score as the document score
Trap 3: BM25 Tokenizer Mismatch with Language
# ❌ Wrong: Using an English tokenizer (split by spaces) for Chinese documents
from rank_bm25 import BM25Okapi
corpus = ["Python异常处理是编程的基础技能", "机器学习模型训练需要大量数据"]
tokenized = [doc.split() for doc in corpus] # Chinese split by spaces, each sentence becomes one token
bm25 = BM25Okapi(tokenized) # BM25 completely ineffective
# ✅ Correct: Use a Chinese tokenizer
import jieba
tokenized = [list(jieba.cut(doc)) for doc in corpus]
# Result: [['Python', '异常', '处理', '是', '编程', '的', '基础', '技能'], ...]
bm25 = BM25Okapi(tokenized)
Trap 4: Improper Rerank Threshold Settings
# ❌ Wrong: No threshold set, returning all results (including irrelevant ones)
results = reranker.rerank(query, documents, top_n=10)
# Even if all documents are irrelevant, 10 results are still returned
# ✅ Correct: Set a reasonable threshold to filter low-score results
def smart_rerank(query, documents, top_n=10, min_threshold=0.3, max_threshold=0.7):
results = reranker.rerank(query, documents, top_n=top_n)
# Dynamic threshold: 60% of the highest score as threshold, but not below min_threshold
if results:
dynamic_threshold = max(min_threshold, results[0]["relevance_score"] * 0.6)
dynamic_threshold = min(dynamic_threshold, max_threshold)
filtered = [r for r in results if r["relevance_score"] >= dynamic_threshold]
return filtered if filtered else [results[0]] # Return at least 1 result
return []
Trap 5: Cache Not Accounting for Document Updates
# ❌ Wrong: Cache never expires, returns stale results after document updates
cache = {}
def search(query):
if query in cache:
return cache[query] # Documents updated, but cache not invalidated
results = do_search(query)
cache[query] = results
return results
# ✅ Correct: Versioned cache strategy
class VersionedCache:
def __init__(self, ttl_seconds: int = 3600):
self._cache: Dict[str, tuple] = {}
self._doc_version: int = 0
self._ttl = ttl_seconds
def invalidate_on_update(self):
"""Call when documents are updated to invalidate all cache"""
self._doc_version += 1
def get(self, key: str) -> Optional[list]:
if key in self._cache:
cached_data, cached_version, cached_time = self._cache[key]
# Invalidate if version mismatch or TTL expired
if (cached_version == self._doc_version
and time.time() - cached_time < self._ttl):
return cached_data
del self._cache[key]
return None
def set(self, key: str, value: list):
self._cache[key] = (value, self._doc_version, time.time())
Error Troubleshooting Reference Table
| Error Symptom | Possible Cause | Troubleshooting Steps | Solution |
|---|---|---|---|
| Cross-Encoder inference OOM | Too many batch input pairs | Check batch_size and document length | Reduce batch_size or batch inference |
| Accuracy drops after Rerank | Model-domain mismatch | Evaluate model performance with domain data | Switch to domain model or fine-tune |
| BM25 returns empty results | Tokenizer mismatch | Print tokenization results to check | Use jieba for Chinese, MeCab for Japanese |
| Cohere API timeout | Network or quota issues | Check API Key and network | Increase timeout, implement retry mechanism |
| Dense retrieval returns similar scores | Vectors not normalized | Check normalize parameter during encode | Set normalize_embeddings=True |
| RRF fusion results worsen | Low overlap between retrieval paths | Analyze recall rates for each path | Adjust top_k and rrf_k parameters |
| Fine-tuned model overfits | Too little training data or uneven distribution | Check training set size and label distribution | Add data, use early stopping, add regularization |
| Long document Rerank scores abnormal | Truncated beyond max_length | Check document length and model max_length | Use long-context model or chunking strategy |
| Extremely low cache hit rate | Cache key includes random factors | Check cache key generation logic | Cache key should only include query and config params |
| Low GPU utilization | CPU-GPU data transfer bottleneck | Monitor GPU utilization | Increase batch_size, use DataLoader |
Advanced Optimization: 5 Key Strategies to Improve Rerank Effectiveness
1. Query Rewriting
Before Rerank, use an LLM to rewrite the user's original query into more precise retrieval queries:
def rewrite_query_with_llm(original_query: str, llm_client) -> List[str]:
"""Use LLM to rewrite query, generating multiple sub-queries"""
prompt = f"""Rewrite the following user query into 3 more precise retrieval queries, one per line:
Original query: {original_query}
Requirements: Add implicit context, eliminate ambiguity, preserve core intent."""
response = llm_client.chat(prompt)
sub_queries = [q.strip() for q in response.strip().split("\n") if q.strip()]
return [original_query] + sub_queries # Keep original query
2. Adaptive Top-K Strategy
Dynamically adjust the initial filtering count based on query complexity:
def adaptive_top_k(query: str, base_top_k: int = 20) -> int:
"""Adaptively adjust Top-K based on query characteristics"""
# Short queries (keyword-type) need more candidates
if len(query) <= 10:
return base_top_k * 2
# Long queries (descriptive) have clearer semantics, fewer candidates needed
elif len(query) >= 50:
return base_top_k
else:
return int(base_top_k * 1.5)
3. Tiered Re-ranking
First coarse ranking with a lightweight model, then fine ranking with a heavier model:
def tiered_rerank(query, documents, top_n=5):
"""Tiered re-ranking: lightweight model coarse ranking → heavy model fine ranking"""
# Tier 1: Lightweight Cross-Encoder coarse ranking (MiniLM, fast inference)
light_reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")
light_results = light_reranker.rank(query, documents, top_k=20)
# Tier 2: Heavy Cross-Encoder fine ranking (Large, more accurate)
heavy_reranker = CrossEncoder("BAAI/bge-reranker-large")
candidates = [documents[r["corpus_id"]] for r in light_results]
final_results = heavy_reranker.rank(query, candidates, top_k=top_n)
return final_results
4. Multi-Model Ensemble
def ensemble_rerank(query, documents, models, top_n=5):
"""Multi-model ensemble re-ranking"""
all_scores = {}
for model_name, weight in models:
model = CrossEncoder(model_name)
pairs = [(query, doc) for doc in documents]
scores = model.predict(pairs)
for i, score in enumerate(scores):
if i not in all_scores:
all_scores[i] = 0.0
all_scores[i] += float(score) * weight
sorted_indices = sorted(all_scores.keys(), key=lambda x: all_scores[x], reverse=True)
return [documents[i] for i in sorted_indices[:top_n]]
5. Negative Feedback Learning
def collect_hard_negatives(
query: str,
documents: List[str],
reranker: CrossEncoderReranker,
user_feedback: Dict[int, bool],
) -> List[Tuple[str, str, float]]:
"""Collect user negative feedback as hard negative samples"""
training_pairs = []
results = reranker.rerank(query, documents, top_n=len(documents))
for result in results:
idx = result["index"]
is_relevant = user_feedback.get(idx, None)
if is_relevant is not None:
training_pairs.append((
query,
documents[idx],
1.0 if is_relevant else 0.0,
))
return training_pairs
Rerank Approach Comparison
| Approach | Latency | Accuracy | Cost | Deployment | Use Case |
|---|---|---|---|---|---|
| Cohere Rerank API | 50-200ms | ★★★★☆ | Pay-per-use | Cloud API | Quick integration, multilingual |
| BGE-Reranker-v2-m3 | 20-100ms | ★★★★★ | GPU inference | Local deployment | High precision, long documents |
| MiniLM-L-6-v2 | 5-30ms | ★★★☆☆ | CPU capable | Local deployment | Low latency, resource-constrained |
| ColBERT Late Interaction | 30-80ms | ★★★★☆ | GPU inference | Local deployment | Fine-grained matching |
| Custom Fine-tuned Model | Varies by base | ★★★★★ | Training + inference | Local deployment | Specialized domains |
| Hybrid Retrieval + Rerank | 100-300ms | ★★★★★ | GPU inference | Local deployment | Production best practice |
💡 Selection Guide: Use Cohere Rerank for quick validation, BGE-Reranker for cost-effectiveness, fine-tuning is essential for specialized domains, and hybrid retrieval + Rerank is recommended for production environments.
Summary
Rerank is not an optional enhancement for RAG — it's a requirement. A RAG system without Rerank is like a car without brakes — it can move, but it can't stop precisely. From the 5-minute integration of Cohere Rerank API to the production-grade pipeline of hybrid retrieval + Cross-Encoder, the 5 key patterns cover the complete path from prototype to deployment. Remember: Bi-Encoder for recall first, Cross-Encoder for fine ranking second — this is the golden rule of RAG retrieval in 2026.
Recommended Online Tools
When building a Rerank pipeline, these online tools can boost your development efficiency:
- 📄 JSON Formatter — Format JSON results returned by Rerank API, quickly inspect response structure
- 🌐 cURL to Code — Convert Cohere Rerank API cURL examples to Python code with one click
- 🔒 Hash Calculator — Generate document fingerprints for cache keys, enabling efficient cache management
Try these browser-local tools — no sign-up required →
#Rerank#交叉编码器#RAG检索增强#Python AI#2026#AI与大数据