Building RAG on AWS

123456789101112
import boto3
import json
from langchain.text_splitter import RecursiveCharacterTextSplitter
from hashlib import sha256

bedrock = boto3.client('bedrock-runtime')
s3 = boto3.client('s3')

def chunk_document(text: str, metadata: dict) -> list[dict]:
    """Split document into overlapping chunks with metadata."""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,

123456789101112
from opensearchpy import OpenSearch, RequestsHttpConnection
from requests_aws4auth import AWS4Auth
import boto3

def hybrid_search(
    query_text: str,
    query_embedding: list[float],
    index_name: str,
    k: int = 10,
    vector_weight: float = 0.7,
    filters: dict = None
) -> list[dict]:

123456789101112
import boto3
import json
from typing import List, Dict
from dataclasses import dataclass
from concurrent.futures import ThreadPoolExecutor, as_completed
import backoff

@dataclass
class EmbeddingResult:
    chunk_id: str
    embedding: List[float]
    token_count: int

123456789101112
from opensearchpy import OpenSearch, RequestsHttpConnection
from requests_aws4auth import AWS4Auth
import boto3

class HybridSearchClient:
    def __init__(self, collection_endpoint: str, index_name: str):
        credentials = boto3.Session().get_credentials()
        self.auth = AWS4Auth(
            credentials.access_key,
            credentials.secret_key,
            'us-east-1',
            'aoss',

123456789101112
import boto3
import json
from opensearchpy import OpenSearch, RequestsHttpConnection
from requests_aws4auth import AWS4Auth

class RAGQueryHandler:
    def __init__(self):
        self.bedrock = boto3.client('bedrock-runtime')
        self.opensearch = self._init_opensearch()
        self.embedding_model = 'amazon.titan-embed-text-v1'
        self.generation_model = 'anthropic.claude-3-sonnet-20240229-v1:0'
    

123456789101112
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
from rouge_score import rouge_scorer
import boto3
import json

class RAGEvaluator:
    def __init__(self):
        self.bedrock = boto3.client('bedrock-runtime')
        self.rouge = rouge_scorer.RougeScorer(['rougeL'], use_stemmer=True)
    
    def evaluate_retrieval(self, retrieved_docs: list, relevant_docs: list, k: int = 5) -> dict: