RAG (Retrieval-Augmented Generation)

# Pipeline RAG complet : du scraping à la génération
from dataclasses import dataclass
from typing import List
import numpy as np

# Simuler les imports (dans la vraie vie: openai, chromadb, etc.)
# from openai import OpenAI
# import chromadb

@dataclass
class Document:
    content: str
    metadata: dict
    embedding: List[float] = None

class ScrapingRAGPipeline:
    """Pipeline RAG alimenté par web scraping"""
    
    def __init__(self, embedding_model="text-embedding-3-small"):
        self.embedding_model = embedding_model
        self.documents = []
        # self.client = OpenAI()
        # self.vectordb = chromadb.Client()
    
    def ingest_scraped_content(self, scraped_data: List[dict]):
        """Ingère le contenu scrapé dans la base vectorielle"""
        for item in scraped_data:
            # Chunking sémantique
            chunks = self._semantic_chunk(item['content'])
            
            for i, chunk in enumerate(chunks):
                doc = Document(
                    content=chunk,
                    metadata={
                        'source_url': item['url'],
                        'title': item.get('title', ''),
                        'scraped_at': item.get('scraped_at'),
                        'chunk_index': i
                    }
                )
                
                # Générer l'embedding
                doc.embedding = self._embed(chunk)
                self.documents.append(doc)
        
        # Stocker dans la vector DB
        self._store_in_vectordb()
    
    def _semantic_chunk(self, text: str, max_tokens: int = 500) -> List[str]:
        """Découpe le texte en chunks sémantiques"""
        # Simplification : découper par paragraphes
        paragraphs = text.split('

')
        
        chunks = []
        current_chunk = []
        current_length = 0
        
        for para in paragraphs:
            para_length = len(para.split())  # Approximation tokens
            
            if current_length + para_length > max_tokens:
                if current_chunk:
                    chunks.append('

'.join(current_chunk))
                current_chunk = [para]
                current_length = para_length
            else:
                current_chunk.append(para)
                current_length += para_length
        
        if current_chunk:
            chunks.append('

'.join(current_chunk))
        
        return chunks
    
    def _embed(self, text: str) -> List[float]:
        """Génère l'embedding d'un texte"""
        # En production: 
        # response = self.client.embeddings.create(
        #     model=self.embedding_model,
        #     input=text
        # )
        # return response.data[0].embedding
        
        # Placeholder : embedding aléatoire 1536 dims
        return list(np.random.rand(1536))
    
    def retrieve(self, query: str, top_k: int = 5) -> List[Document]:
        """Récupère les documents pertinents"""
        query_embedding = self._embed(query)
        
        # Calcul de similarité cosinus
        similarities = []
        for doc in self.documents:
            sim = np.dot(query_embedding, doc.embedding) / (
                np.linalg.norm(query_embedding) * np.linalg.norm(doc.embedding)
            )
            similarities.append((sim, doc))
        
        # Tri et top-k
        similarities.sort(key=lambda x: x[0], reverse=True)
        return [doc for _, doc in similarities[:top_k]]
    
    def generate(self, query: str) -> str:
        """Génère une réponse RAG"""
        # Récupérer le contexte
        relevant_docs = self.retrieve(query)
        
        # Construire le prompt
        context = "

---

".join([
            f"Source: {doc.metadata['source_url']}
{doc.content}"
            for doc in relevant_docs
        ])
        
        prompt = f"""Réponds à la question en te basant uniquement sur le contexte fourni.
Si l'information n'est pas dans le contexte, dis-le.

Contexte:
{context}

Question: {query}

Réponse:"""
        
        # En production:
        # response = self.client.chat.completions.create(
        #     model="gpt-4-turbo",
        #     messages=[{"role": "user", "content": prompt}]
        # )
        # return response.choices[0].message.content
        
        return f"[RAG Response based on {len(relevant_docs)} sources]"