Upload 2 files

app/database.py

@@ -21,6 +21,9 @@ from dotenv import load_dotenv
 import pyarrow as pa
 import pyarrow.compute as pc
 
+import logging
+logging.basicConfig(level=logging.DEBUG)
+
 # Initialisations
 load_dotenv()
 HF_TOKEN = os.getenv('API_HF_TOKEN')
@@ -161,10 +164,13 @@ def fetch_articles_by_tags(tags: List[str]) -> List[Dict]:
     except Exception as e:
         return {"status": "error", "code": type(e).__name__, "message": str(e)}
 
-def fetch_query_results(query: str, k_model: int = 10, k_cross: int = 5) -> Dict[str, Any]:
+def fetch_query_results(query: str, k_model: int = 10,
+                        k_cross: int = 5, use_rerank: bool = True
+                        ) -> Dict[str, Any]:
     """
-    Exécute une requête de recherche sémantique avec FAISS, puis rerank avec un cross-encoder
-    et retourne les meilleurs passages enrichis avec des métadonnées provenant de DuckDB.
+    Exécute une requête de recherche sémantique avec FAISS, puis (optionnellement)
+    rerank avec un cross-encoder et retourne les meilleurs passages enrichis avec
+    des métadonnées provenant de DuckDB.
 
     Paramètres
     ----------
@@ -174,6 +180,8 @@ def fetch_query_results(query: str, k_model: int = 10, k_cross: int = 5) -> Dict
         Nombre de résultats les plus proches à récupérer depuis l'index FAISS.
     k_cross : int, optionnel (défaut = 5)
         Nombre de résultats finaux à conserver après reranking avec le cross-encoder.
+    use_rerank : bool, optionnel (défaut = True)
+        Si False, on désactive complètement le cross-encoder et le rerank.
 
     Retour
     ------
@@ -206,18 +214,26 @@ def fetch_query_results(query: str, k_model: int = 10, k_cross: int = 5) -> Dict
         distance_map = dict(zip(faiss_ids_list, distances_list))
         df["distance"] = df["faiss_id"].map(distance_map)
 
-        # Cross-encoder
-        df["chunk_text"] = df["chunk_text"].str.replace(r'\s+', ' ', regex=True).str.strip()
-        top_passages = df["chunk_text"].tolist()
-        cross_input = [(query, p) for p in top_passages]
-        cross_scores = cross_encoder.predict(cross_input)
-
-        # Rerank
-        df["cross_score"] = cross_scores
-        df = df.sort_values(by="cross_score", ascending=False)
-
-        # Garder top k_cross
-        df_top = df.head(k_cross)
+        if use_rerank:
+            status_dbg = "ok_rerank"
+            # Cross-encoder
+            df["chunk_text"] = df["chunk_text"].str.replace(r'\s+', ' ', regex=True).str.strip()
+            top_passages = df["chunk_text"].tolist()
+            cross_input = [(query, p) for p in top_passages]
+            cross_scores = cross_encoder.predict(cross_input)
+
+            # Rerank
+            df["cross_score"] = cross_scores
+            df = df.sort_values(by="cross_score", ascending=False)
+
+            # Garder top k_cross
+            df_top = df.head(k_cross)
+        else:
+            status_dbg = "ok_no_rerank"
+            df = df.sort_values(by="distance", ascending=False)
+            df["cross_score"] = df["distance"]
+            # Garder top k_model
+            df_top = df.head(k_model)
 
         # Enregistrer dans DuckDB
         con.register("faiss_tmp", df_top)
@@ -250,7 +266,7 @@ def fetch_query_results(query: str, k_model: int = 10, k_cross: int = 5) -> Dict
         # Liste finale de dictionnaires
         list_result = duck_res.to_dict(orient="records")
 
-        return {"status": "ok", "result": list_result}
+        return {"status": status_dbg, "result": list_result}
     except Exception as e:
         return {"status": "error", "code": type(e).__name__, "message": str(e)}
 

app/main.py

@@ -1,189 +1,226 @@
-from fastapi import FastAPI, Query
-from typing import List, Dict, Any
-from app import database
-from fastapi.middleware.cors import CORSMiddleware
-from fastapi.responses import HTMLResponse
-
-from pydantic import BaseModel
-from transformers import AutoTokenizer, AutoModelForCausalLM
-import torch
-from app.templates.prompt_mistral_rag import RAG_PROMPT_TEMPLATE
-
-app = FastAPI(
-    title="Articles API",
-    description="API pour récupérer articles et tags depuis SQLite",
-    version="1.0"
-)
-
-# Chargement du modèle génératif
-MODEL_NAME = "mistralai/Mistral-7B-Instruct-v0.3"
-tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
-model = AutoModelForCausalLM.from_pretrained(MODEL_NAME,
-                                             torch_dtype=torch.float16,
-                                             device_map="auto"
-                                            )
-
-# CORS pour permettre l'accès depuis le navigateur
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],  # autorise toutes les origines
-    allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"],
-)
-
-@app.get("/", response_class=HTMLResponse)
-def home():
-    return """
-    <html>
-        <head><title>Page d'accueil</title></head>
-        <body>
-            <h1>Welcome on the API search articles !</h1>
-        </body>
-    </html>
-    """
-
-@app.get("/get_tags")
-def get_tags():
-    """
-    Récupère la liste de tous les tags disponibles via l'API.
-
-    Returns:
-        Dict: Un dictionnaire contenant soit la liste des tags, soit les informations d'erreur.
-            - Si succès :
-                {
-                    "status": "ok",
-                    "tags": List[str]  # Liste des noms de tags triés par ordre alphabétique
-                }
-            - En cas d'erreur :
-                {
-                    "status": "error",
-                    "code": str,      # Nom de l'exception
-                    "message": str    # Message de l'exception
-                }
-
-    Notes:
-        - L'appel de cet endpoint effectue un accès à la base de données via la fonction `fetch_tags`.
-        - En cas de problème avec la base de données, un message d'erreur détaillé est retourné.
-    """
-    try:
-        dict_result = database.fetch_tags()
-        if dict_result["status"] == "ok":
-            return {"status": "ok", "tags": dict_result["result"]}
-        else:
-            return dict_result
-    except Exception as e:
-        return {"status": "error", "code": type(e).__name__, "message": str(e)}
-
-@app.get("/get_articles_with_tags")
-def get_articles_with_tags(tags: List[str] = Query(..., description="Liste des tags à filtrer")):
-    """
-    Récupère les articles associés à une ou plusieurs tags spécifiés.
-
-    Args:
-        tags (List[str]): Liste des noms de tags utilisés pour filtrer les articles.
-                          Doit contenir au moins un tag.
-
-    Returns:
-        Dict: Un dictionnaire contenant soit les articles correspondants, soit les informations d'erreur.
-            - Si succès :
-                {
-                    "status": "ok",
-                    "tags": List[str],        # Tags utilisés pour filtrer
-                    "articles": List[Dict]    # Liste des articles correspondants
-                }
-                Chaque article est un dictionnaire contenant :
-                    - 'article_id': int, ID de l'article
-                    - 'article_title': str, Titre de l'article
-                    - 'article_url': str, URL de l'article
-            - En cas d'erreur :
-                {
-                    "status": "error",
-                    "code": str,             # Code d'erreur ou nom de l'exception
-                    "message": str           # Message d'erreur
-                }
-
-    Notes:
-        - Si la liste `tags` est vide, la fonction retourne une erreur avec le code "no_tags".
-        - L'appel de cet endpoint utilise la fonction `fetch_articles_by_tags` pour récupérer les articles.
-    """
-    try:
-        dict_result = database.fetch_articles_by_tags(tags)
-        if dict_result["status"] == "ok":
-            return {"status": "ok",
-                    "tags": tags,
-                    "articles": dict_result["result"]}
-        else:
-            return dict_result
-    except Exception as e:
-        return {"status": "error", "code": type(e).__name__, "message": str(e)}
-
-
-@app.get("/get_query_results")
-def get_query_results(query: str = Query(..., description="Requête de recherche textuelle"),
-                      k_model: int = Query(10, description="Nombre de candidats retournés par FAISS"),
-                      k_cross: int = Query(5, description="Nombre de résultats conservés après reranking")
-                      ) -> Dict[str, Any]:
-    """
-    Récupère les résultats d'une requête en utilisant deux modèles de recherche.
-
-    Args:
-        query (str): La requête utilisateur pour laquelle récupérer les résultats.
-        k_model (int, optional): Nombre de résultats à retourner pour le modèle principal. Par défaut à 10.
-        k_cross (int, optional): Nombre de résultats à retourner pour le modèle croisé. Par défaut à 5.
-
-    Returns:
-        Dict[str, Any]: Un dictionnaire contenant soit les résultats de la requête, soit les informations d'erreur.
-
-    Notes:
-        - L'appel de cet endpoint utilise la fonction `fetch_query_result` pour obtenir les résultats.
-        - En cas de problème lors du traitement de la requête, un message d'erreur détaillé est retourné.
-    """
-    try:
-        dict_result = database.fetch_query_results(query, k_model, k_cross)
-        if dict_result["status"] == "ok":
-            return {"status": "ok",
-                    "results": dict_result["result"]}
-        else:
-            return dict_result
-    except Exception as e:
-        return {"status": "error", "code": type(e).__name__, "message": str(e)}
-
-# 🔹 Exemple de modèle d'entrée utilisateur
-class QueryRequest(BaseModel):
-    question: str
-
-@app.post("/ask")
-async def ask_question(request: QueryRequest):
-    try:
-        user_query = request.question.strip()
-        dict_result = database.fetch_query_results(user_query, k_model=10, k_cross=5)
-        if dict_result["status"] == "ok":
-            list_chunks = [resp['chunk_text'] for resp in dict_result['result']]
-            if not list_chunks:
-                answer = ("Je ne dispose pas d’informations sur ce sujet. "
-                          "Je peux uniquement répondre à des questions sur les articles " \
-                          "du jeu de données.")
-            else:
-                # Construction du prompt
-                prompt = RAG_PROMPT_TEMPLATE.format(
-                    context="\n".join(list_chunks),
-                    question=user_query
-                )
-                print("*** Prompt : ", prompt)
-                # Génération de la réponse
-                inputs = tokenizer(prompt, return_tensors="pt").to(model.device)
-                outputs = model.generate(**inputs, max_new_tokens=500)
-                generated_tokens = outputs[0][inputs["input_ids"].shape[-1]:]  # uniquement la partie générée
-                answer = tokenizer.decode(generated_tokens, skip_special_tokens=True).strip()
-            return {"status": "ok",
-                    "results": dict_result["result"],
-                    "answer": answer}
-        else:
-            answer = f"Une erreur est survenue lors de la récupération des informations : \
-                       {dict_result['code']} - {dict_result['message']}."
-            return {"status": "error", "answer": answer}
-    except Exception as e:
-        answer = f"Une erreur est survenue lors de la récupération des informations : \
-                       {type(e).__name__} - {str(e)}."
+from fastapi import FastAPI, Query
+from typing import List, Dict, Any
+from app import database
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import HTMLResponse
+
+from pydantic import BaseModel
+from transformers import AutoTokenizer, AutoModelForCausalLM
+import torch
+from app.templates.prompt_mistral_rag import RAG_PROMPT_TEMPLATE
+
+app = FastAPI(
+    title="Articles API",
+    description="API pour récupérer articles et tags depuis SQLite",
+    version="1.0"
+)
+
+# Chargement du modèle génératif
+MODEL_NAME = "mistralai/Mistral-7B-Instruct-v0.3"
+tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
+model = AutoModelForCausalLM.from_pretrained(MODEL_NAME,
+                                             torch_dtype=torch.float16,
+                                             device_map="auto"
+                                            )
+
+# CORS pour permettre l'accès depuis le navigateur
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],  # autorise toutes les origines
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+@app.get("/", response_class=HTMLResponse)
+def home():
+    return """
+    <html>
+        <head><title>Page d'accueil</title></head>
+        <body>
+            <h1>Welcome on the API search articles !</h1>
+        </body>
+    </html>
+    """
+
+@app.get("/get_tags")
+def get_tags():
+    """
+    Récupère la liste de tous les tags disponibles via l'API.
+
+    Returns:
+        Dict: Un dictionnaire contenant soit la liste des tags, soit les informations d'erreur.
+            - Si succès :
+                {
+                    "status": "ok",
+                    "tags": List[str]  # Liste des noms de tags triés par ordre alphabétique
+                }
+            - En cas d'erreur :
+                {
+                    "status": "error",
+                    "code": str,      # Nom de l'exception
+                    "message": str    # Message de l'exception
+                }
+
+    Notes:
+        - L'appel de cet endpoint effectue un accès à la base de données via la fonction `fetch_tags`.
+        - En cas de problème avec la base de données, un message d'erreur détaillé est retourné.
+    """
+    try:
+        dict_result = database.fetch_tags()
+        if dict_result["status"] == "ok":
+            return {"status": "ok", "tags": dict_result["result"]}
+        else:
+            return dict_result
+    except Exception as e:
+        return {"status": "error", "code": type(e).__name__, "message": str(e)}
+
+@app.get("/get_articles_with_tags")
+def get_articles_with_tags(tags: List[str] = Query(..., description="Liste des tags à filtrer")):
+    """
+    Récupère les articles associés à une ou plusieurs tags spécifiés.
+
+    Args:
+        tags (List[str]): Liste des noms de tags utilisés pour filtrer les articles.
+                          Doit contenir au moins un tag.
+
+    Returns:
+        Dict: Un dictionnaire contenant soit les articles correspondants, soit les informations d'erreur.
+            - Si succès :
+                {
+                    "status": "ok",
+                    "tags": List[str],        # Tags utilisés pour filtrer
+                    "articles": List[Dict]    # Liste des articles correspondants
+                }
+                Chaque article est un dictionnaire contenant :
+                    - 'article_id': int, ID de l'article
+                    - 'article_title': str, Titre de l'article
+                    - 'article_url': str, URL de l'article
+            - En cas d'erreur :
+                {
+                    "status": "error",
+                    "code": str,             # Code d'erreur ou nom de l'exception
+                    "message": str           # Message d'erreur
+                }
+
+    Notes:
+        - Si la liste `tags` est vide, la fonction retourne une erreur avec le code "no_tags".
+        - L'appel de cet endpoint utilise la fonction `fetch_articles_by_tags` pour récupérer les articles.
+    """
+    try:
+        dict_result = database.fetch_articles_by_tags(tags)
+        if dict_result["status"] == "ok":
+            return {"status": "ok",
+                    "tags": tags,
+                    "articles": dict_result["result"]}
+        else:
+            return dict_result
+    except Exception as e:
+        return {"status": "error", "code": type(e).__name__, "message": str(e)}
+
+
+@app.get("/get_query_results")
+def get_query_results(query: str = Query(..., description="Requête de recherche textuelle"),
+                      k_model: int = Query(10, description="Nombre de candidats retournés par FAISS"),
+                      k_cross: int = Query(5, description="Nombre de résultats conservés après reranking"),
+                      use_rerank: bool = Query(True, description="Indique si le reranking avec cross-encoder doit être utilisé")
+                      ) -> Dict[str, Any]:
+    """
+    Récupère les résultats d'une requête en utilisant deux modèles de recherche.
+
+    Args:
+        query (str): La requête utilisateur pour laquelle récupérer les résultats.
+        k_model (int, optionel): Nombre de résultats à retourner pour le modèle principal. Par défaut à 10.
+        k_cross (int, optionel): Nombre de résultats à retourner pour le modèle croisé. Par défaut à 5.
+        use_rerank (bool, optionnel): Indique si le reranking avec cross-encoder doit être utilisé. Par défaut à True.
+        Si False, on désactive complètement le cross-encoder et le rerank.
+    Returns:
+        Dict[str, Any]: Un dictionnaire contenant soit les résultats de la requête, soit les informations d'erreur.
+
+    Notes:
+        - L'appel de cet endpoint utilise la fonction `fetch_query_result` pour obtenir les résultats.
+        - En cas de problème lors du traitement de la requête, un message d'erreur détaillé est retourné.
+    """
+    try:
+        dict_result = database.fetch_query_results(query, k_model, k_cross, use_rerank)
+        if dict_result["status"] == "ok":
+            return {"status": "ok",
+                    "results": dict_result["result"]}
+        else:
+            return dict_result
+    except Exception as e:
+        return {"status": "error", "code": type(e).__name__, "message": str(e)}
+
+#
+class QueryRequest(BaseModel):
+    question: str
+    use_rerank: bool = True
+
+@app.post("/get_answer")
+async def ask_question(request: QueryRequest):
+    """
+    Traite une question utilisateur en effectuant une recherche dans la base de données
+    puis en générant une réponse à l’aide du modèle de langage (RAG).
+
+    Le fonctionnement se déroule en trois étapes :
+    1. Extraction et nettoyage de la question utilisateur.
+    2. Recherche des passages pertinents dans la base de données (`fetch_query_results`).
+    3. Génération d’une réponse fondée sur les morceaux de texte récupérés (RAG).
+
+    Paramètres
+    ----------
+    request : QueryRequest
+        Objet contenant la question utilisateur sous forme de chaîne de caractères.
+
+    Retour
+    ------
+    dict
+        Un dictionnaire contenant :
+        - "status" : "ok" si la requête a réussi, "error" en cas d'échec.
+        - "results" : liste des chunks retournés par la base de données (présent seulement si status = "ok").
+        - "answer" : réponse générée par le modèle ou message d’erreur.
+
+    Exceptions
+    ----------
+    Toute exception survenant durant l’exécution est interceptée
+    et retournée sous forme d’un message d’erreur dans la clé "answer".
+
+    Notes
+    -----
+    - Si aucun chunk pertinent n'est trouvé, la fonction renvoie un message indiquant
+      que seules les questions relatives aux articles du jeu de données peuvent être traitées.
+    - La génération de la réponse utilise un template RAG et produit jusqu’à 500 tokens.
+    """
+    try:
+        user_query = request.question.strip()
+        use_rerank = request.use_rerank
+        dict_result = database.fetch_query_results(user_query, k_model=10,
+                                                   k_cross=5, use_rerank=use_rerank)
+        if dict_result["status"] == "ok":
+            list_chunks = [resp['chunk_text'] for resp in dict_result['result']]
+            if not list_chunks:
+                answer = ("Je ne dispose pas d’informations sur ce sujet. "
+                          "Je peux uniquement répondre à des questions sur les articles " \
+                          "du jeu de données.")
+            else:
+                # Construction du prompt
+                prompt = RAG_PROMPT_TEMPLATE.format(
+                    context="\n".join(list_chunks),
+                    question=user_query
+                )
+                # Génération de la réponse
+                inputs = tokenizer(prompt, return_tensors="pt").to(model.device)
+                outputs = model.generate(**inputs, max_new_tokens=500)
+                generated_tokens = outputs[0][inputs["input_ids"].shape[-1]:]  # uniquement la partie générée
+                answer = tokenizer.decode(generated_tokens, skip_special_tokens=True).strip()
+            return {"status": "ok",
+                    "results": dict_result["result"],
+                    "answer": answer}
+        else:
+            answer = f"Une erreur est survenue lors de la récupération des informations : \
+                       {dict_result['code']} - {dict_result['message']}."
+            return {"status": "error", "answer": answer}
+    except Exception as e:
+        answer = f"Une erreur est survenue lors de la récupération des informations : \
+                       {type(e).__name__} - {str(e)}."
         return {"status": "error", "answer": answer}