Advanced RAG: Sentence-Window Retrieval and Cohere Re-ranking

!pip install -qU langchain sentence-transformers faiss-cpu nltk cohere openai

import nltk
nltk.download('punkt')

import os
import re
from typing import List, Dict, Any
from langchain.docstore.document import Document
from langchain.text_splitter import TextSplitter
from nltk.tokenize import sent_tokenize

# Let's define a custom LangChain-compatible Document Node
# to hold our sentence and its windowed context.
class SentenceWindowNode:
    def __init__(self, text: str, metadata: Dict[str, Any]):
        self.page_content = text
        self.metadata = metadata

    def __repr__(self):
        return f"SentenceWindowNode(text='{self.page_content}', metadata={self.metadata})"

class SentenceWindowNodeParser(TextSplitter):
    """Implementation of splitting text into nodes of sentences.

    This class is designed to create nodes for a sentence-window retrieval
    strategy. Each node contains a single sentence for embedding, but its
    metadata includes the surrounding sentences (the 'window').
    """

    def __init__(self, window_size: int = 3, **kwargs: Any):
        """Initialize with a window size.
        
        Args:
            window_size: The number of sentences on each side of the central
                         sentence to include in the context window. The total
                         window size will be (2 * window_size + 1).
        """
        super().__init__(**kwargs)
        self.window_size = window_size

    def split_text(self, text: str) -> List[SentenceWindowNode]:
        """Split text into a list of SentenceWindowNode objects."""
        sentences = sent_tokenize(text)
        nodes = []
        for i, sentence in enumerate(sentences):
            # Determine the start and end of the window
            start_index = max(0, i - self.window_size)
            end_index = min(len(sentences), i + self.window_size + 1)
            
            # Create the window of sentences
            window_sentences = sentences[start_index:end_index]
            window_text = " ".join(window_sentences)
            
            # Create the node with the single sentence as the text to embed
            # and the window as metadata.
            node = SentenceWindowNode(
                text=sentence,
                metadata={
                    'window': window_text,
                    'original_document': text, # Or some document ID
                    'sentence_index': i
                }
            )
            nodes.append(node)
        return nodes

# --- Example Usage ---

# Let's create a more complex sample document
sample_document_text = """
Generative AI is transforming industries. Its core component is the Large Language Model (LLM), which is trained on vast amounts of text data. 
One of the most popular architectures for LLMs is the Transformer, introduced by Google in 2017. The Transformer architecture relies on a mechanism called self-attention, which allows the model to weigh the importance of different words in the input sequence. This is a departure from previous architectures like RNNs and LSTMs which processed text sequentially.
However, deploying these models presents challenges. Latency and computational cost are significant hurdles for real-time applications. Techniques like quantization and knowledge distillation are employed to create smaller, more efficient models. Fine-tuning is another crucial step to adapt a pre-trained LLM for a specific task, such as customer support or code generation. The future of AI will likely involve multi-modal models that can process not just text, but also images, audio, and video.
"""

# Instantiate our parser with a window size of 1 (1 sentence before, 1 after)
parser = SentenceWindowNodeParser(window_size=1)

# Process the document
nodes = parser.split_text(sample_document_text)

# Let's inspect a node from the middle of the document
print("--- Inspecting Node 3 ---")
node_3 = nodes[3]
print(f"Text to be embedded: \n'{node_3.page_content}'\n")
print(f"Context window (for LLM): \n'{node_3.metadata['window']}'\n")

# Let's inspect a node at the edge
print("--- Inspecting Node 0 (Edge Case) ---")
node_0 = nodes[0]
print(f"Text to be embedded: \n'{node_0.page_content}'\n")
print(f"Context window (for LLM): \n'{node_0.metadata['window']}'\n")

--- Inspecting Node 3 ---
Text to be embedded: 
'The Transformer architecture relies on a mechanism called self-attention, which allows the model to weigh the importance of different words in the input sequence.'

Context window (for LLM): 
'One of the most popular architectures for LLMs is the Transformer, introduced by Google in 2017. The Transformer architecture relies on a mechanism called self-attention, which allows the model to weigh the importance of different words in the input sequence. This is a departure from previous architectures like RNNs and LSTMs which processed text sequentially.'

--- Inspecting Node 0 (Edge Case) ---
Text to be embedded: 
'Generative AI is transforming industries.'

Context window (for LLM): 
'Generative AI is transforming industries. Its core component is the Large Language Model (LLM), which is trained on vast amounts of text data.'

from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import FAISS

# Use a high-quality, open-source embedding model
model_name = "BAAI/bge-small-en-v1.5"
embedding_model = HuggingFaceEmbeddings(model_name=model_name)

# Extract just the sentence texts for embedding
sentence_texts = [node.page_content for node in nodes]

# Create the FAISS index from the sentence texts
# This will embed the sentences and create the index
vectorstore = FAISS.from_texts(texts=sentence_texts, embedding=embedding_model)

# We need a way to map the index in the vector store back to our original nodes.
# FAISS.from_texts doesn't directly support storing complex objects, so we'll
# manage a mapping ourselves. The default FAISS implementation in LangChain
# stores the original texts in a `docstore`, which we can leverage.

# For a more robust production system, you would use a vector database that
# supports storing rich metadata directly, like Pinecone, Weaviate, or ChromaDB.
# Here, we'll create a simple lookup dictionary for clarity.

id_to_node_map = {i: node for i, node in enumerate(nodes)}

# Let's test the retrieval
query = "What is the core mechanism of the Transformer architecture?"

# Retrieve the indices and scores of the most similar sentence embeddings
retrieved_docs = vectorstore.similarity_search_with_score(query, k=2)

print("--- Retrieval Results (Sentence Level) ---")
for doc, score in retrieved_docs:
    print(f"Score: {score:.4f}")
    print(f"Retrieved Sentence: '{doc.page_content}'")
    print("-" * 20)

# Now, let's get the full context windows for the top result
if retrieved_docs:
    top_retrieval_index = vectorstore.index_to_docstore_id[0] # This is a bit of a hack for FAISS in langchain
    # A better way is to get the index from the search results if the library supports it.
    # Let's simulate getting the index from the result.
    # Find the index of the first retrieved document's content in our original list
    first_retrieved_sentence = retrieved_docs[0][0].page_content
    original_index = sentence_texts.index(first_retrieved_sentence)

    # Use our map to get the full node
    retrieved_node = id_to_node_map[original_index]
    
    print("\n--- Context Provided to LLM for Top Result ---")
    print(retrieved_node.metadata['window'])

import cohere

# It's best to set this as an environment variable
# os.environ["COHERE_API_KEY"] = "YOUR_API_KEY"
cohere_api_key = os.getenv("COHERE_API_KEY")

if not cohere_api_key:
    raise ValueError("COHERE_API_KEY environment variable not set.")

co = cohere.Client(cohere_api_key)

def retrieve_and_rerank(query: str, vector_store: FAISS, node_map: Dict, top_k: int = 10, top_n: int = 3) -> List[Dict]:
    """Retrieves documents, re-ranks them, and returns the top N results with their context windows."""
    
    # Stage 1: Initial Retrieval from Vector Store
    # Retrieve a larger set of candidate documents
    initial_retrievals = vector_store.similarity_search(query, k=top_k)
    initial_sentences = [doc.page_content for doc in initial_retrievals]

    if not initial_sentences:
        return []

    # Stage 2: Re-ranking with Cohere
    print(f"--- Re-ranking {len(initial_sentences)} initial results... ---")
    reranked_results = co.rerank(
        model='rerank-english-v2.0',
        query=query,
        documents=initial_sentences,
        top_n=top_n
    )

    # Stage 3: Map re-ranked results back to our original nodes to get the full context
    final_results = []
    original_sentence_list = [node.page_content for node in node_map.values()]

    for result in reranked_results.results:
        # Find the original node that corresponds to the re-ranked sentence
        original_index = result.index
        retrieved_sentence = initial_sentences[original_index]
        
        # Find this sentence in our master list to get its global index
        try:
            global_index = original_sentence_list.index(retrieved_sentence)
            node = node_map[global_index]
            final_results.append({
                'node': node,
                'relevance_score': result.relevance_score
            })
        except ValueError:
            print(f"Warning: Could not find sentence '{retrieved_sentence}' in original node map.")
            continue

    return final_results

# --- Example Usage with a More Nuanced Query ---

# This query is tricky. "Efficiency" is mentioned with quantization, but the
# core challenge is latency/cost. A simple vector search might over-index on the
# word "efficiency" and miss the broader context.
nuanced_query = "How are LLM deployment challenges related to model efficiency?"

# Execute the full pipeline
final_ranked_nodes = retrieve_and_rerank(nuanced_query, vectorstore, id_to_node_map)

print("\n--- Final Top 3 Re-ranked Results ---")
for i, result in enumerate(final_ranked_nodes):
    print(f"Rank {i+1} (Score: {result['relevance_score']:.4f}):")
    print(f"Context Window: '{result['node'].metadata['window']}'")
    print("-" * 20)

import openai

# Set up your OpenAI API key
# os.environ["OPENAI_API_KEY"] = "YOUR_API_KEY"
openai.api_key = os.getenv("OPENAI_API_KEY")

if not openai.api_key:
    raise ValueError("OPENAI_API_KEY environment variable not set.")

def generate_final_answer(query: str, ranked_nodes: List[Dict]):
    """Generates a final answer using the top-ranked context."""
    if not ranked_nodes:
        return "I could not find a relevant answer in the provided documents."

    # Combine the context windows from the top N results
    combined_context = "\n\n---\n\n".join([res['node'].metadata['window'] for res in ranked_nodes])
    
    # Create the prompt for the LLM
    prompt = f"""
    You are a helpful AI assistant. Answer the user's query based on the following context.
    
    Context:
    {combined_context}
    
    Query: {nuanced_query}
    
    Answer:
    """

    try:
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=[
                {"role": "system", "content": "You are an expert Q&A assistant."},
                {"role": "user", "content": prompt}
            ],
            temperature=0.0
        )
        return response.choices[0].message['content'].strip()
    except Exception as e:
        return f"An error occurred during generation: {e}"

# --- Putting it all together ---

# 1. We have our nuanced query
# nuanced_query = "How are LLM deployment challenges related to model efficiency?"

# 2. We already ran our advanced RAG pipeline to get the best nodes
# final_ranked_nodes = retrieve_and_rerank(nuanced_query, vectorstore, id_to_node_map)

# 3. Now, generate the final answer
final_answer = generate_final_answer(nuanced_query, final_ranked_nodes)

print("\n=== FINAL GENERATED ANSWER ===\n")
print(final_answer)

=== FINAL GENERATED ANSWER ===

LLM deployment challenges are directly related to model efficiency. The primary hurdles for using these models in real-time applications are latency and high computational costs. To improve efficiency and overcome these challenges, techniques such as quantization and knowledge distillation are used to create smaller, more optimized models.