Implement advanced RAG with multiple techniques:
Pipeline:
- Query → 2. Rewrite → 3. Decompose → 4. Multi-retrieve → 5. Fuse → 6. Rerank
Methods:
rewrite_query(query, context): Improve query- Expand acronyms
- Add context from history
- Fix typos
decompose_query(query): Split complex queriesmulti_strategy_retrieval(queries): Vector + keyword + hybridreciprocal_rank_fusion(lists, k): Combine results- RRF formula:
score = sum(1.0 / (rank + k))
- RRF formula:
rerank_results(query, results): Final rankingretrieve(query, top_k): Full pipeline
Fusion: Multiple retrieval strategies → RRF → reranking
Examples
Example 1:
Input:
class MockVS: pass; class MockLLM: pass; rag = AdvancedRAG(MockVS(), MockLLM()); rewritten = rag.rewrite_query('What is AI?'); isinstance(rewritten, str)Output:
TrueExplanation: Query rewritten to string
Starter Code
class AdvancedRAG:
"""
Advanced RAG with query rewriting, multi-retrieval, and result fusion.
"""
def __init__(self, vector_store, llm_client):
self.vector_store = vector_store
self.llm = llm_client
self.query_history = []
def rewrite_query(self, query, context=None):
"""
Rewrite query for better retrieval.
Handles ambiguous, complex, or conversational queries.
"""
# Your implementation here
pass
def decompose_query(self, query):
"""Decompose complex query into sub-queries"""
# Your implementation here
pass
def multi_strategy_retrieval(self, queries, strategies=None):
"""
Retrieve using multiple strategies: vector, keyword, hybrid.
"""
# Your implementation here
pass
def reciprocal_rank_fusion(self, results_lists, k=60):
"""
Fuse multiple result lists using RRF.
RRF score = sum(1 / (rank + k)) for each doc
"""
# Your implementation here
pass
def rerank_results(self, query, results):
"""Rerank results by relevance to query"""
# Your implementation here
pass
def retrieve(self, query, top_k=5):
"""End-to-end advanced retrieval"""
# Your implementation here
passPython3
ReadyLines: 1Characters: 0
Ready