Advanced RAG Patterns

123456789101112
from typing import List, Dict, Tuple
import numpy as np
from collections import defaultdict

def reciprocal_rank_fusion(
    results_lists: List[List[Tuple[str, float]]],
    k: int = 60,
    weights: List[float] = None
) -> List[Tuple[str, float]]:
    """
    Combine multiple ranked lists using RRF.
    

123456789101112
from openai import OpenAI
import numpy as np

client = OpenAI()

def hyde_retrieval(
    query: str,
    vector_store,
    num_hypothetical: int = 3,
    top_k: int = 10
) -> list:
    """

123456789101112
from typing import List, Dict, Tuple
import numpy as np

def reciprocal_rank_fusion(
    sparse_results: List[Tuple[str, float]],
    dense_results: List[Tuple[str, float]],
    k: int = 60,
    sparse_weight: float = 0.5
) -> List[Tuple[str, float]]:
    """
    Combine sparse and dense retrieval results using RRF.
    k parameter controls how much to favor top-ranked documents.

123456789101112
import cohere
from typing import List, Dict
import asyncio

class RerankerPipeline:
    def __init__(self, api_key: str):
        self.co = cohere.Client(api_key)
        self.cache = {}  # Simple cache for repeated queries
    
    async def retrieve_and_rerank(
        self,
        query: str,

123456789101112
from openai import OpenAI
import asyncio
from typing import List

client = OpenAI()

QUERY_EXPANSION_PROMPT = """Generate 4 alternative search queries for the following question.
Each alternative should:
1. Use different words but seek the same information
2. Include one that uses more technical terminology
3. Include one that uses simpler, everyday language
4. Include one that approaches the topic from a different angle

123456789101112
from openai import OpenAI
from typing import List, Dict, Optional
from dataclasses import dataclass

@dataclass
class HopResult:
    query: str
    documents: List[Dict]
    extracted_answer: str
    confidence: float

class MultiHopRetriever:

123456789101112
import asyncio
from dataclasses import dataclass
from typing import List, Dict, Tuple
import numpy as np
from openai import AsyncOpenAI
import asyncpg

@dataclass
class SearchResult:
    id: str
    content: str
    score: float

123456789101112
from anthropic import Anthropic
from pydantic import BaseModel
from typing import List
import json

class ExpandedQueries(BaseModel):
    original: str
    synonyms: List[str]  # Same meaning, different words
    specific: List[str]  # More narrow interpretations
    general: List[str]   # Broader interpretations
    related: List[str]   # Adjacent concepts

123456789101112
from dataclasses import dataclass
from typing import List, Optional, Tuple
from anthropic import Anthropic
import json

@dataclass
class RetrievalCritique:
    document_id: str
    relevance_score: int  # 1-5
    reasoning: str
    useful_excerpts: List[str]