""" RAG Chat API - Gustave Eiffel Hackathon 2026 ============================================= Version améliorée : - Chunking fixe conservé - Recherche vectorielle conservée - Recherche BM25 locale ajoutée - Reranking déterministe sobre ajouté - Query rewriting conditionnel par LLM - Expansion déterministe des acronymes actuariels - Seuil de distance adaptatif - Top-K augmenté raisonnablement pour améliorer l'accuracy Architecture: User Query → Conditional LLM Query Rewriting → Deterministic Query Expansion → Multi-query Retrieval → Embeddings → Hybrid Vector + BM25 Search → Cheap Deterministic Reranking → Adaptive Distance Filtering → Neighbor Chunk Expansion → Context Retrieval → LLM Generation → Response """ import os import json import logging import math import re import time import unicodedata from collections import Counter, defaultdict from pathlib import Path from typing import Optional # Must be set before chromadb is imported so the module never registers its # posthog telemetry hook. os.environ.setdefault("ANONYMIZED_TELEMETRY", "False") import requests as http_requests import gradio as gr from fastapi import FastAPI, HTTPException from fastapi.responses import JSONResponse from pydantic import BaseModel import chromadb from chromadb.config import Settings from langchain_text_splitters import RecursiveCharacterTextSplitter from pypdf import PdfReader from llm import call_llm as call_llm_with_metrics logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Suppress non-fatal chromadb telemetry errors logging.getLogger("chromadb.telemetry.product.posthog").setLevel(logging.CRITICAL) # --------------------------------------------------------------------------- # Configuration # --------------------------------------------------------------------------- # Resolve the data directory: /data when running inside HF Spaces bucket mount, # ./data for local development. DATA_DIR = Path("/data") if Path("/data").is_dir() else Path("./data") CHROMA_PERSIST_DIR = str(DATA_DIR / "chroma_db") TRAIN_DOCS_DIR = Path("./train_data") COLLECTION_NAME = "rag_documents" # Chunking fixe de base CHUNK_SIZE = 512 CHUNK_OVERLAP = 50 # Ancien top_k par défaut TOP_K_RESULTS = 4 # --------------------------------------------------------------------------- # Retrieval Configuration # --------------------------------------------------------------------------- # Nombre de candidats récupérés par la recherche vectorielle. # On reste volontairement bas pour limiter la latence. RETRIEVAL_CANDIDATES = 25 # Nombre de candidats récupérés par BM25. # BM25 est local et ne consomme pas de tokens, mais on limite quand même # pour garder un reranking rapide. BM25_CANDIDATES = 25 # Nombre maximum de candidats hybrides rerankés après déduplication. RERANK_CANDIDATES_LIMIT = 40 # Poids du reranker déterministe. # Aucun modèle supplémentaire : coût API nul, CO2 quasi nul côté LLM. RERANK_VECTOR_WEIGHT = 0.45 RERANK_BM25_WEIGHT = 0.35 RERANK_TERM_COVERAGE_WEIGHT = 0.20 # Seuil minimal du score reranké pour accepter un chunk lexicalement pertinent. # Garde-fou contre les chunks récupérés par mots-clés trop faibles. MIN_RERANK_SCORE = 0.08 MIN_TERM_COVERAGE = 0.10 # Seuil de distance par défaut. # Plus le seuil est bas, plus on est strict. DISTANCE_THRESHOLD = 0.68 # Nombre maximum de chunks envoyés au LLM. # 5 augmente un peu les tokens, mais améliore le contexte disponible pour répondre. MAX_CONTEXT_CHUNKS = 5 # Si True, quand aucun chunk ne passe les filtres hybrides, # on envoie quand même le meilleur chunk. # En évaluation hackathon, on préfère tenter avec le meilleur chunk plutôt que répondre vide. FALLBACK_TO_BEST_CHUNK = True # Nombre de chunks voisins ajoutés autour des meilleurs chunks. # Cela augmente un peu les tokens, mais améliore souvent les réponses # quand l'information est coupée entre deux chunks. NEIGHBOR_CHUNK_WINDOW = 1 MAX_CONTEXT_CHUNKS_AFTER_NEIGHBORS = 7 # --------------------------------------------------------------------------- # Query Rewriting Configuration # --------------------------------------------------------------------------- # Active/désactive le query rewriting LLM. QUERY_REWRITE_ENABLED = True # On ne reformule que les questions courtes/ambiguës pour limiter coût, tokens et CO2. QUERY_REWRITE_MAX_WORDS = 12 # Nombre maximal de tokens générés par le LLM pour la reformulation. QUERY_REWRITE_MAX_TOKENS = 80 # --------------------------------------------------------------------------- # Load Config # --------------------------------------------------------------------------- _CONFIG_PATH = DATA_DIR / "config.json" if not _CONFIG_PATH.exists(): _CONFIG_PATH = Path(__file__).parent / "config.json" logger.warning( f"No config.json found in {DATA_DIR} — falling back to root config.json. " "Copy config.json to the data directory and fill in your values for production use." ) with open(_CONFIG_PATH, encoding="utf-8") as _f: _config = json.load(_f) # Embedding model EMBEDDING_ENDPOINT_URL = _config["embedding"]["endpoint_url"] EMBEDDING_MODEL_NAME = _config["embedding"]["model"] # LLM LLM_ENDPOINT_URL = _config["llm"]["endpoint_url"] LLM_MODEL_NAME = _config["llm"]["model"] LLM_MAX_TOKENS = _config["llm"].get("max_completion_tokens", 700) # Pour un RAG évalué sur l'exactitude, une température basse limite les réponses inventives. LLM_TEMPERATURE = _config["llm"].get("temperature", 0.1) LLM_TOP_P = _config["llm"].get("top_p", 1.0) # Azure API key from environment variable AZURE_API_KEY = os.environ.get("AZURE_API_KEY") if not AZURE_API_KEY: logger.warning("AZURE_API_KEY is not set — LLM and embedding calls will fail.") # Prompt template _PROMPT_TEMPLATE_PATH = Path(__file__).parent / "prompts" / "rag_prompt.txt" RAG_PROMPT_TEMPLATE = _PROMPT_TEMPLATE_PATH.read_text(encoding="utf-8") logger.info(f"Embedding model configured: {EMBEDDING_MODEL_NAME} via Azure OpenAI") # --------------------------------------------------------------------------- # Initialize Vector Store # --------------------------------------------------------------------------- logger.info(f"Initializing ChromaDB at: {CHROMA_PERSIST_DIR}") chroma_client = chromadb.PersistentClient( path=CHROMA_PERSIST_DIR, settings=Settings(anonymized_telemetry=False), ) collection = chroma_client.get_or_create_collection( name=COLLECTION_NAME, metadata={"hnsw:space": "cosine"}, ) logger.info(f"ChromaDB collection '{COLLECTION_NAME}' ready. Documents: {collection.count()}") logger.info(f"LLM configured: {LLM_MODEL_NAME} via {LLM_ENDPOINT_URL}") # Cache global du BM25. # Il est reconstruit seulement quand le nombre de chunks change. BM25_INDEX_CACHE = { "count": -1, "ids": [], "documents": [], "metadatas": [], "doc_tokens": [], "doc_lengths": [], "avgdl": 0.0, "idf": {}, "inverted_index": {}, } # --------------------------------------------------------------------------- # Helper Functions # --------------------------------------------------------------------------- def extract_text_from_pdf(pdf_path: Path) -> str: """ Extract text content from a PDF file using pypdf. """ reader = PdfReader(str(pdf_path)) pages_text = [] for page_num, page in enumerate(reader.pages, start=1): text = page.extract_text() if text and text.strip(): pages_text.append(f"[Page {page_num}]\n{text.strip()}") full_text = "\n\n".join(pages_text) logger.info( f"Extracted {len(reader.pages)} pages from PDF: " f"{pdf_path.name} ({len(full_text)} chars)" ) return full_text def chunk_text(text: str, source: str = "unknown") -> list[dict]: """ Chunking fixe de base. """ splitter = RecursiveCharacterTextSplitter( chunk_size=CHUNK_SIZE, chunk_overlap=CHUNK_OVERLAP, separators=["\n\n", "\n", ". ", " ", ""], ) chunks = splitter.split_text(text) return [ { "text": chunk, "source": source, "chunk_index": i, } for i, chunk in enumerate(chunks) if chunk.strip() ] def generate_embeddings(texts: list[str]) -> list[list[float]]: """ Generate vector embeddings via the Azure OpenAI /embeddings endpoint. """ headers = { "api-key": AZURE_API_KEY, "Content-Type": "application/json", } payload = { "input": texts, "model": EMBEDDING_MODEL_NAME, } try: resp = http_requests.post( EMBEDDING_ENDPOINT_URL, headers=headers, json=payload, timeout=120, ) resp.raise_for_status() data = resp.json() return [item["embedding"] for item in data["data"]] except http_requests.exceptions.HTTPError as e: logger.error(f"Embedding API call failed: {e} — {resp.text}") raise HTTPException( status_code=503, detail=f"Embedding service unavailable: {str(e)}", ) except (http_requests.exceptions.JSONDecodeError, ValueError) as e: logger.error( f"Embedding API returned non-JSON response " f"(status {resp.status_code}): {repr(resp.text)}" ) raise HTTPException( status_code=502, detail="Embedding service returned an invalid response", ) except (KeyError, IndexError) as e: logger.error(f"Unexpected embedding response format: {e} — body: {resp.text}") raise HTTPException( status_code=502, detail="Unexpected response from embedding service", ) def add_documents_to_vectorstore(documents: list[dict]) -> int: """ Save document embeddings to the ChromaDB vector store. """ if not documents: return 0 texts = [doc["text"] for doc in documents] embeddings = generate_embeddings(texts) existing_count = collection.count() ids = [ f"doc_{existing_count + i}" for i in range(len(documents)) ] metadatas = [ { "source": doc.get("source", "unknown"), "chunk_index": doc.get("chunk_index", i), } for i, doc in enumerate(documents) ] collection.add( ids=ids, embeddings=embeddings, documents=texts, metadatas=metadatas, ) logger.info(f"Added {len(documents)} chunks to vector store. Total: {collection.count()}") return len(documents) # --------------------------------------------------------------------------- # Lightweight BM25 + Deterministic Reranker # --------------------------------------------------------------------------- SEARCH_STOPWORDS = { "a", "au", "aux", "avec", "ce", "ces", "cette", "dans", "de", "des", "du", "elle", "en", "et", "est", "il", "ils", "je", "la", "le", "les", "leur", "leurs", "mais", "ou", "où", "par", "pas", "pour", "que", "qui", "sur", "un", "une", "se", "sa", "son", "ses", "the", "of", "and", "to", "in", "is", "for", "on", "with", "what", "how", "why", } def normalize_for_search(text: str) -> str: """ Normalise un texte pour la recherche lexicale : - minuscules ; - suppression des accents ; - conservation des lettres/chiffres. """ text = unicodedata.normalize("NFKD", text.lower()) text = "".join(ch for ch in text if not unicodedata.combining(ch)) return text def tokenize_for_search(text: str) -> list[str]: """ Tokenisation sobre pour BM25 et le reranking. Aucun modèle NLP externe n'est chargé. """ normalized = normalize_for_search(text) tokens = re.findall(r"[a-z0-9]+", normalized) return [ token for token in tokens if token not in SEARCH_STOPWORDS and (len(token) >= 2 or token.isdigit()) ] def build_bm25_index_if_needed() -> dict: """ Construit un index BM25 local à partir des chunks déjà présents dans ChromaDB. Sobriété : - aucun appel API ; - aucun embedding supplémentaire ; - reconstruction uniquement si le nombre de chunks change. """ current_count = collection.count() if BM25_INDEX_CACHE["count"] == current_count: return BM25_INDEX_CACHE logger.info(f"Rebuilding BM25 index for {current_count} chunks.") if current_count == 0: BM25_INDEX_CACHE.update({ "count": 0, "ids": [], "documents": [], "metadatas": [], "doc_tokens": [], "doc_lengths": [], "avgdl": 0.0, "idf": {}, "inverted_index": {}, }) return BM25_INDEX_CACHE stored = collection.get(include=["documents", "metadatas"]) ids = stored.get("ids", []) documents = stored.get("documents", []) or [] metadatas = stored.get("metadatas", []) or [] doc_tokens = [tokenize_for_search(doc or "") for doc in documents] doc_lengths = [len(tokens) for tokens in doc_tokens] avgdl = sum(doc_lengths) / max(len(doc_lengths), 1) doc_freq = Counter() inverted_index = defaultdict(list) for doc_idx, tokens in enumerate(doc_tokens): counts = Counter(tokens) for term, freq in counts.items(): doc_freq[term] += 1 inverted_index[term].append((doc_idx, freq)) total_docs = len(documents) idf = { term: math.log(1 + (total_docs - freq + 0.5) / (freq + 0.5)) for term, freq in doc_freq.items() } BM25_INDEX_CACHE.update({ "count": current_count, "ids": ids, "documents": documents, "metadatas": metadatas, "doc_tokens": doc_tokens, "doc_lengths": doc_lengths, "avgdl": avgdl, "idf": idf, "inverted_index": dict(inverted_index), }) return BM25_INDEX_CACHE def bm25_search(query: str, top_n: int = BM25_CANDIDATES) -> list[dict]: """ Recherche BM25 locale. BM25 favorise les correspondances exactes de termes, utile pour : BEL, SCR, TVOG, GLM, noms de méthodes, formules, etc. """ index = build_bm25_index_if_needed() if index["count"] == 0: return [] query_terms = tokenize_for_search(query) if not query_terms: return [] k1 = 1.2 b = 0.75 scores = defaultdict(float) unique_terms = set(query_terms) for term in unique_terms: postings = index["inverted_index"].get(term, []) term_idf = index["idf"].get(term, 0.0) for doc_idx, freq in postings: dl = index["doc_lengths"][doc_idx] avgdl = index["avgdl"] or 1.0 denom = freq + k1 * (1 - b + b * dl / avgdl) scores[doc_idx] += term_idf * (freq * (k1 + 1)) / max(denom, 1e-9) ranked = sorted(scores.items(), key=lambda item: item[1], reverse=True)[:top_n] results = [] for doc_idx, score in ranked: metadata = index["metadatas"][doc_idx] or {} results.append({ "id": index["ids"][doc_idx], "text": index["documents"][doc_idx], "source": metadata.get("source", "unknown"), "chunk_index": metadata.get("chunk_index"), "distance": None, "similarity_score": 0.0, "bm25_score": score, "retrieval_methods": {"bm25"}, }) return results def minmax_normalize(values: list[float]) -> list[float]: """ Normalisation simple entre 0 et 1 pour fusionner des scores hétérogènes. """ if not values: return [] min_value = min(values) max_value = max(values) if max_value <= min_value: return [1.0 if max_value > 0 else 0.0 for _ in values] return [ (value - min_value) / (max_value - min_value) for value in values ] def term_coverage_score(query: str, text: str) -> float: """ Part des termes importants de la requête retrouvés dans le chunk. Sert de petit garde-fou lexical dans le reranking. """ query_terms = set(tokenize_for_search(query)) if not query_terms: return 0.0 text_terms = set(tokenize_for_search(text)) return len(query_terms & text_terms) / len(query_terms) def merge_candidates(vector_candidates: list[dict], bm25_candidates: list[dict]) -> list[dict]: """ Fusionne les candidats vectoriels et BM25 sans doublons. Clé principale : source + chunk_index. Fallback : hash du texte si les métadonnées sont absentes. """ merged = {} for candidate in vector_candidates + bm25_candidates: key = ( candidate.get("source"), candidate.get("chunk_index"), ) if key[1] is None: key = (candidate.get("source"), hash(candidate.get("text", ""))) if key not in merged: merged[key] = { **candidate, "bm25_score": candidate.get("bm25_score", 0.0), "retrieval_methods": set(candidate.get("retrieval_methods", set())), } else: existing = merged[key] candidate_distance = candidate.get("distance") existing_distance = existing.get("distance") # En multi-query, le même chunk peut être retrouvé plusieurs fois. # On garde la meilleure distance vectorielle, pas la dernière rencontrée. if candidate_distance is not None and ( existing_distance is None or candidate_distance < existing_distance ): existing["distance"] = candidate_distance existing["similarity_score"] = candidate.get( "similarity_score", existing.get("similarity_score", 0.0), ) existing["bm25_score"] = max( existing.get("bm25_score", 0.0), candidate.get("bm25_score", 0.0), ) existing["retrieval_methods"].update(candidate.get("retrieval_methods", set())) return list(merged.values()) def rerank_candidates(candidates: list[dict], query: str, effective_threshold: float) -> list[dict]: """ Reranker déterministe et sobre. Il ne charge aucun cross-encoder et ne fait aucun appel LLM. Score final = vectoriel + BM25 + couverture lexicale. """ if not candidates: return [] vector_scores = [ max(0.0, candidate.get("similarity_score", 0.0) or 0.0) for candidate in candidates ] bm25_scores = [ max(0.0, candidate.get("bm25_score", 0.0) or 0.0) for candidate in candidates ] coverage_scores = [ term_coverage_score(query, candidate.get("text", "")) for candidate in candidates ] vector_norm = minmax_normalize(vector_scores) bm25_norm = minmax_normalize(bm25_scores) reranked = [] for idx, candidate in enumerate(candidates): rerank_score = ( RERANK_VECTOR_WEIGHT * vector_norm[idx] + RERANK_BM25_WEIGHT * bm25_norm[idx] + RERANK_TERM_COVERAGE_WEIGHT * coverage_scores[idx] ) distance = candidate.get("distance") vector_passed = distance is not None and distance <= effective_threshold lexical_passed = ( bm25_norm[idx] > 0 and coverage_scores[idx] >= MIN_TERM_COVERAGE and rerank_score >= MIN_RERANK_SCORE ) candidate["rerank_score"] = rerank_score candidate["bm25_normalized_score"] = bm25_norm[idx] candidate["term_coverage"] = coverage_scores[idx] candidate["passed_threshold"] = vector_passed or lexical_passed candidate["retrieval_method"] = "+".join(sorted(candidate.get("retrieval_methods", []))) # Valeurs de compatibilité pour l'affichage/API. if candidate.get("distance") is None: candidate["distance"] = 1.0 if candidate.get("similarity_score") is None: candidate["similarity_score"] = 0.0 reranked.append(candidate) reranked.sort(key=lambda x: x["rerank_score"], reverse=True) return reranked[:RERANK_CANDIDATES_LIMIT] def add_neighbor_chunks( contexts: list[dict], window: int = NEIGHBOR_CHUNK_WINDOW, max_chunks: int = MAX_CONTEXT_CHUNKS_AFTER_NEIGHBORS, ) -> list[dict]: """ Ajoute les chunks voisins des meilleurs chunks sélectionnés. Objectif : améliorer l'accuracy quand la réponse est répartie sur deux chunks consécutifs, par exemple une définition en fin de chunk et une formule au début du chunk suivant. """ if not contexts or window <= 0 or max_chunks <= len(contexts): return contexts[:max_chunks] index = build_bm25_index_if_needed() if index["count"] == 0: return contexts[:max_chunks] chunk_lookup = {} for doc_idx, metadata in enumerate(index["metadatas"]): metadata = metadata or {} source = metadata.get("source", "unknown") chunk_index = metadata.get("chunk_index") if chunk_index is not None: chunk_lookup[(source, chunk_index)] = doc_idx expanded_contexts = [] seen_keys = set() def add_context(ctx: dict): key = (ctx.get("source"), ctx.get("chunk_index"), hash(ctx.get("text", ""))) if key in seen_keys: return expanded_contexts.append(ctx) seen_keys.add(key) for ctx in contexts: source = ctx.get("source") chunk_index = ctx.get("chunk_index") if chunk_index is None: add_context(ctx) if len(expanded_contexts) >= max_chunks: break continue # On met le chunk central en premier, puis les voisins proches. ordered_neighbor_indexes = [chunk_index] for offset in range(1, window + 1): ordered_neighbor_indexes.extend([chunk_index - offset, chunk_index + offset]) for neighbor_index in ordered_neighbor_indexes: if len(expanded_contexts) >= max_chunks: break lookup_key = (source, neighbor_index) if lookup_key not in chunk_lookup: continue doc_idx = chunk_lookup[lookup_key] metadata = index["metadatas"][doc_idx] or {} neighbor_ctx = { **ctx, "text": index["documents"][doc_idx], "source": metadata.get("source", source), "chunk_index": metadata.get("chunk_index", neighbor_index), "retrieval_method": ( ctx.get("retrieval_method", "retrieval") if neighbor_index == chunk_index else f"{ctx.get('retrieval_method', 'retrieval')}+neighbor" ), } add_context(neighbor_ctx) if len(expanded_contexts) >= max_chunks: break return expanded_contexts[:max_chunks] # --------------------------------------------------------------------------- # Query Rewriting and Query Expansion # --------------------------------------------------------------------------- def clean_query_words(query: str) -> list[str]: """ Nettoyage simple d'une question pour détecter mots/acronymes. """ q = query.lower().strip() cleaned = ( q.replace("?", " ") .replace(",", " ") .replace(".", " ") .replace(";", " ") .replace(":", " ") .replace("'", " ") .replace('"', " ") .replace("(", " ") .replace(")", " ") ) return cleaned.split() def should_rewrite_query(query: str) -> bool: """ Détermine si la question doit être reformulée par LLM. On limite volontairement le rewriting aux questions courtes ou ambiguës pour éviter d'ajouter un appel LLM inutile à chaque requête. """ if not QUERY_REWRITE_ENABLED: return False words = clean_query_words(query) if not words: return False technical_terms = { "bel", "scr", "mcr", "glm", "mrh", "var", "tvar", "ifrs", "alm", "tvog" } # Acronyme seul ou question très courte avec acronyme. if len(words) <= QUERY_REWRITE_MAX_WORDS and any(w in technical_terms for w in words): return True # Question très courte, potentiellement ambiguë. if len(words) <= 3: return True return False def rewrite_query_with_llm(query: str) -> tuple[str, dict]: """ Reformule la question utilisateur pour améliorer la recherche vectorielle. Important : - La reformulation sert uniquement au retrieval. - La question originale reste utilisée dans le prompt final. - La fonction retourne aussi les métriques du rewriting. """ default_info = { "query_rewrite_used": False, "original_query": query, "rewritten_query": query, "query_rewrite_prompt_tokens": 0, "query_rewrite_completion_tokens": 0, "query_rewrite_total_tokens": 0, "query_rewrite_co2_grams": None, "query_rewrite_energy_kwh": None, } if not should_rewrite_query(query): return query, default_info rewrite_prompt = f""" Tu reformules une question pour améliorer une recherche dans un corpus de mémoires d'actuariat. Règles : - Ne réponds pas à la question. - Ne rajoute pas d'information inventée. - Explicite seulement les acronymes actuariels évidents si présents : BEL = Best Estimate Liability, SCR = Solvency Capital Requirement, MCR = Minimum Capital Requirement, GLM = modèle linéaire généralisé, MRH = multirisque habitation, VaR = Value at Risk, TVaR = Tail Value at Risk, ALM = Asset Liability Management, TVOG = Time Value of Options and Guarantees. - Retourne une seule question reformulée, en français. - Maximum 25 mots. - Ne retourne pas de JSON. Question originale : {query} Question reformulée : """.strip() try: rewrite_result = call_llm_with_metrics( rewrite_prompt, endpoint_url=LLM_ENDPOINT_URL, api_key=AZURE_API_KEY, model=LLM_MODEL_NAME, max_completion_tokens=QUERY_REWRITE_MAX_TOKENS, temperature=0, top_p=1, ) rewritten_query = rewrite_result["content"].strip() rewritten_query = rewritten_query.strip('"').strip("'").strip() if not rewritten_query: return query, default_info tokens = rewrite_result.get("tokens", {}) info = { "query_rewrite_used": True, "original_query": query, "rewritten_query": rewritten_query, "query_rewrite_prompt_tokens": tokens.get("prompt", 0), "query_rewrite_completion_tokens": tokens.get("completion", 0), "query_rewrite_total_tokens": tokens.get("total", 0), "query_rewrite_co2_grams": rewrite_result.get("co2_grams"), "query_rewrite_energy_kwh": rewrite_result.get("energy_kwh"), } logger.info( f"Query rewritten: original='{query}' | rewritten='{rewritten_query}'" ) return rewritten_query, info except Exception as e: logger.error(f"Query rewriting failed: {e}") return query, default_info def expand_query(query: str) -> str: """ Enrichit les acronymes actuariels pour améliorer la recherche vectorielle. Important : - Cela ne change pas la question envoyée au LLM final. - Cela change seulement la requête utilisée pour chercher les chunks. - Pas besoin de refaire l'ingestion. """ q = query.lower().strip() expansions = { "bel": "Best Estimate Liability Best Estimate provision technique assurance vie solvabilité", "scr": "Solvency Capital Requirement capital de solvabilité Solvabilité II", "mcr": "Minimum Capital Requirement minimum capital requis Solvabilité II", "glm": "modèle linéaire généralisé GLM tarification fréquence sévérité sinistres", "mrh": "multirisque habitation assurance habitation sinistres habitation", "var": "Value at Risk VaR quantile risque capital économique", "tvar": "Tail Value at Risk TVaR risque extrême capital économique", "ifrs": "IFRS 17 norme comptable assurance contrats d'assurance", "alm": "Asset Liability Management gestion actif passif", "tvog": "Time Value of Options and Guarantees valeur temps des options et garanties Solvabilité II", } words = set(clean_query_words(q)) expanded = query for term, expansion in expansions.items(): if term in words or q == term: expanded += " " + expansion return expanded def build_retrieval_queries(query: str) -> tuple[list[str], dict]: """ Construit plusieurs requêtes utilisées pour le retrieval. Cela augmente un peu le coût d'embedding, mais améliore le recall : - question originale ; - question reformulée ; - question originale enrichie ; - question reformulée enrichie. """ rewritten_query, rewrite_info = rewrite_query_with_llm(query) raw_queries = [ query, rewritten_query, expand_query(query), expand_query(rewritten_query), ] retrieval_queries = [] seen = set() for candidate_query in raw_queries: candidate_query = candidate_query.strip() key = candidate_query.lower() if candidate_query and key not in seen: retrieval_queries.append(candidate_query) seen.add(key) rewrite_info["retrieval_queries"] = retrieval_queries rewrite_info["retrieval_query"] = " | ".join(retrieval_queries) return retrieval_queries, rewrite_info def build_retrieval_query(query: str) -> tuple[str, dict]: """ Wrapper conservé pour compatibilité éventuelle avec d'anciens appels. La pipeline principale utilise maintenant build_retrieval_queries(). """ retrieval_queries, rewrite_info = build_retrieval_queries(query) return retrieval_queries[-1], rewrite_info def get_distance_threshold(query: str) -> float: """ Seuil adaptatif selon le type de question. Idée : - Requêtes très courtes ou acronymes : seuil plus permissif. - Requêtes normales : seuil standard. """ q = query.lower().strip() technical_terms = { "bel", "scr", "mcr", "glm", "mrh", "var", "tvar", "ifrs", "alm", "tvog" } words = clean_query_words(query) # Acronyme seul : BEL, SCR, GLM... if q in technical_terms: return 0.70 # Question courte contenant un terme technique if len(words) <= 3 and any(word in technical_terms for word in words): return 0.70 # Question très courte : seuil un peu plus permissif if len(words) <= 5: return 0.65 # Cas général return DISTANCE_THRESHOLD # --------------------------------------------------------------------------- # Retrieval # --------------------------------------------------------------------------- def retrieve_relevant_context( query: str, top_k: int = TOP_K_RESULTS, ) -> tuple[list[dict], dict]: """ Retrieve relevant document chunks with hybrid retrieval, multi-query search, cheap reranking and neighbor expansion. Améliorations accuracy : - plusieurs requêtes de retrieval ; - plus de candidats vectoriels et BM25 ; - fusion/déduplication ; - reranking déterministe ; - seuil adaptatif plus permissif ; - ajout des chunks voisins. """ if collection.count() == 0: return [], { "query_rewrite_used": False, "original_query": query, "rewritten_query": query, "retrieval_query": query, "retrieval_queries": [query], "effective_threshold": None, "query_rewrite_prompt_tokens": 0, "query_rewrite_completion_tokens": 0, "query_rewrite_total_tokens": 0, "query_rewrite_co2_grams": None, "query_rewrite_energy_kwh": None, } retrieval_queries, rewrite_info = build_retrieval_queries(query) effective_threshold = get_distance_threshold(query) # On embed plusieurs requêtes d'un coup : un seul appel API embeddings, # mais plusieurs vecteurs pour améliorer le recall. query_embeddings = generate_embeddings(retrieval_queries) requested_top_k = top_k or TOP_K_RESULTS max_contexts_before_neighbors = min( max(requested_top_k, TOP_K_RESULTS), MAX_CONTEXT_CHUNKS, ) vector_candidate_count = min( max(RETRIEVAL_CANDIDATES, max_contexts_before_neighbors), collection.count(), ) vector_candidates = [] for retrieval_query, query_embedding in zip(retrieval_queries, query_embeddings): vector_results = collection.query( query_embeddings=[query_embedding], n_results=vector_candidate_count, include=["documents", "metadatas", "distances"], ) for i in range(len(vector_results["documents"][0])): distance = vector_results["distances"][0][i] metadata = vector_results["metadatas"][0][i] or {} vector_candidates.append({ "text": vector_results["documents"][0][i], "source": metadata.get("source", "unknown"), "chunk_index": metadata.get("chunk_index"), "distance": distance, "similarity_score": 1 - distance, "bm25_score": 0.0, "retrieval_methods": {"vector"}, }) bm25_candidates = [] for retrieval_query in retrieval_queries: bm25_candidates.extend( bm25_search(retrieval_query, top_n=BM25_CANDIDATES) ) merged_candidates = merge_candidates(vector_candidates, bm25_candidates) # Pour la couverture lexicale, on évite de pénaliser avec toutes les expansions. # Les expansions servent à chercher ; le ranking lexical se base surtout sur # la question originale et sa reformulation. rerank_query = " ".join([ query, rewrite_info.get("rewritten_query", ""), ]).strip() reranked_candidates = rerank_candidates( merged_candidates, query=rerank_query or query, effective_threshold=effective_threshold, ) filtered_contexts = [ ctx for ctx in reranked_candidates if ctx.get("passed_threshold") ] if not filtered_contexts and FALLBACK_TO_BEST_CHUNK and reranked_candidates: logger.info( f"No chunk passed hybrid filters. Falling back to best reranked chunk " f"with rerank_score={reranked_candidates[0]['rerank_score']:.4f}." ) filtered_contexts = [reranked_candidates[0]] selected_contexts = filtered_contexts[:max_contexts_before_neighbors] selected_contexts = add_neighbor_chunks( selected_contexts, window=NEIGHBOR_CHUNK_WINDOW, max_chunks=MAX_CONTEXT_CHUNKS_AFTER_NEIGHBORS, ) rewrite_info["effective_threshold"] = effective_threshold rewrite_info["vector_candidates"] = len(vector_candidates) rewrite_info["bm25_candidates"] = len(bm25_candidates) rewrite_info["merged_candidates"] = len(merged_candidates) rewrite_info["reranked_candidates"] = len(reranked_candidates) rewrite_info["passed_threshold"] = len(filtered_contexts) rewrite_info["selected_contexts"] = len(selected_contexts) logger.info( f"Hybrid multi-query retrieval: " f"queries={len(retrieval_queries)}, " f"vector={len(vector_candidates)}, " f"bm25={len(bm25_candidates)}, " f"merged={len(merged_candidates)}, " f"passed={len(filtered_contexts)}, " f"selected_with_neighbors={len(selected_contexts)}, " f"threshold={effective_threshold}, " f"rewrite_used={rewrite_info.get('query_rewrite_used')}, " f"query='{query[:80]}'" ) return selected_contexts, rewrite_info def build_rag_prompt(query: str, contexts: list[dict]) -> str: """ Construct the RAG prompt by combining retrieved context with the user question. La distance n'est pas ajoutée dans le prompt pour économiser quelques tokens. Elle reste disponible dans les sources retournées par l'API. """ context_text = "\n\n".join( f"[Source: {ctx['source']}]\n{ctx['text']}" for ctx in contexts ) prompt = RAG_PROMPT_TEMPLATE.format( context=context_text, question=query, ) return prompt def rag_query(query: str, top_k: int = TOP_K_RESULTS) -> dict: """ End-to-end RAG pipeline. """ start_time = time.perf_counter() contexts, retrieval_info = retrieve_relevant_context(query, top_k=top_k) if not contexts: elapsed_ms = round((time.perf_counter() - start_time) * 1000, 2) if collection.count() == 0: answer = "No documents have been ingested yet. Please upload documents first." explanation = "No documents found in the vector store to retrieve context from." else: effective_threshold = retrieval_info.get("effective_threshold") answer = ( "I could not find sufficiently relevant context in the ingested documents " "to answer this question reliably." ) explanation = ( f"Documents exist in the vector store, but no retrieved chunk passed " f"the adaptive distance threshold of {effective_threshold}. " "This avoids sending weak or irrelevant context to the LLM." ) rewrite_tokens = retrieval_info.get("query_rewrite_total_tokens", 0) return { "answer": answer, "sources": [], "explanation": explanation, "total_token": rewrite_tokens, "prompt_tokens": retrieval_info.get("query_rewrite_prompt_tokens", 0), "completion_tokens": retrieval_info.get("query_rewrite_completion_tokens", 0), "cached_tokens": 0, "query_rewrite_used": retrieval_info.get("query_rewrite_used", False), "rewritten_query": retrieval_info.get("rewritten_query", query), "retrieval_query": retrieval_info.get("retrieval_query", query), "query_rewrite_total_tokens": retrieval_info.get("query_rewrite_total_tokens", 0), "co2_grams": retrieval_info.get("query_rewrite_co2_grams"), "energy_kwh": retrieval_info.get("query_rewrite_energy_kwh"), "run_time_in_ms": elapsed_ms, } prompt = build_rag_prompt(query, contexts) llm_result = call_llm_with_metrics( prompt, endpoint_url=LLM_ENDPOINT_URL, api_key=AZURE_API_KEY, model=LLM_MODEL_NAME, max_completion_tokens=LLM_MAX_TOKENS, temperature=LLM_TEMPERATURE, top_p=LLM_TOP_P, ) raw_content = llm_result["content"] tokens = llm_result["tokens"] rewrite_total_tokens = retrieval_info.get("query_rewrite_total_tokens", 0) answer_total_tokens = tokens["total"] total_token = rewrite_total_tokens + answer_total_tokens # Parse structured JSON response from LLM json_str = raw_content.strip() if json_str.startswith("```"): json_str = json_str.split("\n", 1)[-1] json_str = json_str.rsplit("```", 1)[0].strip() try: parsed = json.loads(json_str) answer = parsed["answer"] explanation = parsed["explanation"] except (json.JSONDecodeError, KeyError): answer = raw_content explanation = "LLM did not return a structured explanation." elapsed_ms = round((time.perf_counter() - start_time) * 1000, 2) # Addition approximative des métriques CO2/énergie du rewriting et de la réponse finale. rewrite_co2 = retrieval_info.get("query_rewrite_co2_grams") final_co2 = llm_result.get("co2_grams") if isinstance(rewrite_co2, (int, float)) and isinstance(final_co2, (int, float)): total_co2 = rewrite_co2 + final_co2 else: total_co2 = final_co2 rewrite_energy = retrieval_info.get("query_rewrite_energy_kwh") final_energy = llm_result.get("energy_kwh") if isinstance(rewrite_energy, (int, float)) and isinstance(final_energy, (int, float)): total_energy = rewrite_energy + final_energy else: total_energy = final_energy return { "answer": answer, "sources": [ { "source": ctx["source"], "score": ctx["similarity_score"], "distance": ctx["distance"], "bm25_score": ctx.get("bm25_score", 0.0), "bm25_normalized_score": ctx.get("bm25_normalized_score", 0.0), "rerank_score": ctx.get("rerank_score", 0.0), "term_coverage": ctx.get("term_coverage", 0.0), "retrieval_method": ctx.get("retrieval_method", "vector"), "passed_threshold": ctx["passed_threshold"], "chunk_index": ctx.get("chunk_index"), "ref_text": ctx["text"], } for ctx in contexts ], "explanation": explanation, # Tokens totaux = rewriting éventuel + génération finale. "total_token": total_token, "prompt_tokens": tokens["prompt"] + retrieval_info.get("query_rewrite_prompt_tokens", 0), "completion_tokens": tokens["completion"] + retrieval_info.get("query_rewrite_completion_tokens", 0), "cached_tokens": tokens["cached"], # Détail du rewriting "query_rewrite_used": retrieval_info.get("query_rewrite_used", False), "rewritten_query": retrieval_info.get("rewritten_query", query), "retrieval_query": retrieval_info.get("retrieval_query", query), "query_rewrite_total_tokens": retrieval_info.get("query_rewrite_total_tokens", 0), "answer_generation_total_tokens": answer_total_tokens, # Métriques CO2/énergie totales approximatives "co2_grams": total_co2, "energy_kwh": total_energy, "run_time_in_ms": elapsed_ms, } # --------------------------------------------------------------------------- # Ingest Train Documents # --------------------------------------------------------------------------- def ingest_train_documents(): """ Load and embed training documents into the vector store. """ if collection.count() > 0: logger.info("Vector store already has documents, skipping ingestion.") return if not TRAIN_DOCS_DIR.exists(): logger.warning(f"No train_data directory found at: {TRAIN_DOCS_DIR}") return total_chunks = 0 for file_path in TRAIN_DOCS_DIR.rglob("*.txt"): logger.info(f"Ingesting text file: {file_path.name}") text = file_path.read_text(encoding="utf-8", errors="ignore") chunks = chunk_text(text, source=file_path.name) total_chunks += add_documents_to_vectorstore(chunks) for file_path in TRAIN_DOCS_DIR.rglob("*.pdf"): logger.info(f"Ingesting PDF file: {file_path.name}") text = extract_text_from_pdf(file_path) if text.strip(): chunks = chunk_text(text, source=file_path.name) total_chunks += add_documents_to_vectorstore(chunks) else: logger.warning(f"No extractable text found in: {file_path.name}") logger.info( f"Train document ingestion complete. " f"Chunks added: {total_chunks}. Total chunks: {collection.count()}" ) # --------------------------------------------------------------------------- # FastAPI Application # --------------------------------------------------------------------------- app = FastAPI( title="RAG Chat API - Gustave Eiffel Hackathon 2026", description="A RAG system with /query endpoint for evaluation", version="1.0.0", ) class QueryRequest(BaseModel): """ Request schema for the /query endpoint. """ query: str top_k: Optional[int] = TOP_K_RESULTS class IngestRequest(BaseModel): """ Request schema for the /ingest endpoint. """ text: str source: str = "user_upload" @app.post("/query") async def query_endpoint(request: QueryRequest): """ RAG Query Endpoint. """ logger.info(f"Query received: {request.query}") result = rag_query(request.query, top_k=request.top_k) return JSONResponse(content=result) @app.post("/ingest") async def ingest_endpoint(request: IngestRequest): """ Document Ingestion Endpoint. """ chunks = chunk_text(request.text, source=request.source) count = add_documents_to_vectorstore(chunks) return JSONResponse(content={ "status": "success", "chunks_added": count, "total_chunks": collection.count(), }) @app.get("/health") async def health_check(): """ Health check endpoint. """ return { "status": "healthy", "documents_in_store": collection.count(), "embedding_model": EMBEDDING_MODEL_NAME, "llm_model": LLM_MODEL_NAME, "retrieval_strategy": "multi_query_hybrid_vector_bm25_plus_deterministic_reranker_plus_neighbors", "query_rewrite_enabled": QUERY_REWRITE_ENABLED, "query_rewrite_max_words": QUERY_REWRITE_MAX_WORDS, "retrieval_candidates": RETRIEVAL_CANDIDATES, "bm25_candidates": BM25_CANDIDATES, "rerank_candidates_limit": RERANK_CANDIDATES_LIMIT, "rerank_weights": { "vector": RERANK_VECTOR_WEIGHT, "bm25": RERANK_BM25_WEIGHT, "term_coverage": RERANK_TERM_COVERAGE_WEIGHT, }, "default_distance_threshold": DISTANCE_THRESHOLD, "max_context_chunks": MAX_CONTEXT_CHUNKS, "fallback_to_best_chunk": FALLBACK_TO_BEST_CHUNK, "neighbor_chunk_window": NEIGHBOR_CHUNK_WINDOW, "max_context_chunks_after_neighbors": MAX_CONTEXT_CHUNKS_AFTER_NEIGHBORS, } # --------------------------------------------------------------------------- # Gradio UI # --------------------------------------------------------------------------- def gradio_query(question: str) -> tuple[str, str, str, str, str]: """ Handle queries from the Gradio chat interface. """ if not question.strip(): return "Please enter a question.", "", "", "", "" result = rag_query(question) sources_text = "\n".join( ( f" - {s['source']} " f"(method: {s.get('retrieval_method', 'vector')}, " f"distance: {s.get('distance', 0):.4f}, " f"vector: {s.get('score', 0):.4f}, " f"bm25: {s.get('bm25_score', 0):.4f}, " f"rerank: {s.get('rerank_score', 0):.4f})" ) for s in result["sources"] ) rewrite_info = "" if result.get("query_rewrite_used"): rewrite_info = ( f"\n\n🔎 Rewritten query used for retrieval:\n" f"{result.get('rewritten_query')}" ) answer = ( f"{result['answer']}\n\n📚 Sources:\n{sources_text}{rewrite_info}" if result["sources"] else f"{result['answer']}{rewrite_info}" ) explanation = result.get("explanation", "") token_info = str(result.get("total_token", 0)) co2_value = result.get("co2_grams") co2_info = f"{co2_value:.4f} g" if isinstance(co2_value, (int, float)) else "N/A" run_time = f"{result.get('run_time_in_ms', 0)} ms" return answer, explanation, token_info, co2_info, run_time def gradio_ingest(text: str, source_name: str) -> str: """ Handle document ingestion from the Gradio UI. """ if not text.strip(): return "Please provide text to ingest." chunks = chunk_text(text, source=source_name or "user_upload") count = add_documents_to_vectorstore(chunks) return ( f"✅ Ingested {count} chunks. " f"Total documents in store: {collection.count()}" ) with gr.Blocks(title="RAG Chat API - Gustave Eiffel Hackathon") as demo: gr.Markdown(""" # 🗼 RAG Chat API - Gustave Eiffel Hackathon 2026 This application demonstrates a complete **Retrieval-Augmented Generation (RAG)** system. **Current improvements:** - Conditional LLM query rewriting for short or ambiguous questions - Deterministic query expansion for actuarial acronyms such as BEL, SCR, GLM, MRH, TVOG - Multi-query retrieval to improve recall - Hybrid vector + BM25 search with more candidates - Cheap deterministic reranking, without extra LLM call - Adaptive distance threshold - Neighbor chunks added around selected passages for better context **API Endpoint:** Use `POST /query` with `{"query": "your question"}` for programmatic access. --- """) with gr.Tab("💬 Chat"): gr.Markdown("Ask questions about the ingested documents.") with gr.Row(): query_input = gr.Textbox( label="Your Question", placeholder="e.g., BEL, SCR, GLM, ou Comment le SCR est-il modélisé en assurance vie ?", lines=2, ) query_button = gr.Button("Ask", variant="primary") query_output = gr.Textbox(label="Answer", lines=8, interactive=False) query_explanation = gr.Textbox(label="Explanation", lines=3, interactive=False) with gr.Row(): query_tokens = gr.Textbox(label="Total Tokens", interactive=False) query_co2 = gr.Textbox(label="CO2 Emission", interactive=False) query_runtime = gr.Textbox(label="Run Time", interactive=False) query_button.click( fn=gradio_query, inputs=query_input, outputs=[ query_output, query_explanation, query_tokens, query_co2, query_runtime, ], ) with gr.Tab("📄 Ingest Documents"): gr.Markdown("Add new documents to the knowledge base.") doc_text = gr.Textbox( label="Document Text", placeholder="Paste your document text here...", lines=10, ) doc_source = gr.Textbox( label="Source Name", placeholder="e.g., my_document.txt", value="user_upload", ) ingest_button = gr.Button("Ingest Document", variant="primary") ingest_output = gr.Textbox(label="Status", interactive=False) ingest_button.click( fn=gradio_ingest, inputs=[doc_text, doc_source], outputs=ingest_output, ) with gr.Tab("ℹ️ API Info"): gr.Markdown(""" ## API Endpoints ### POST /query ```json { "query": "BEL", "top_k": 3 } ``` **Response:** ```json { "answer": "...", "sources": [ { "source": "document.pdf", "score": 0.82, "distance": 0.18 } ], "query_rewrite_used": true, "rewritten_query": "Qu'est-ce que le Best Estimate Liability dans les mémoires d'actuariat ?" } ``` ### POST /ingest ```json { "text": "Your document text here...", "source": "document_name.txt" } ``` ### GET /health Returns system health, document count and retrieval configuration. """) app = gr.mount_gradio_app(app, demo, path="/") # --------------------------------------------------------------------------- # Entry Point # --------------------------------------------------------------------------- if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=7860)