Create app.py

app.py

@@ -0,0 +1,170 @@
+import os
+import sys
+import pandas as pd
+import numpy as np
+import faiss
+import gradio as gr
+from sentence_transformers import SentenceTransformer
+from huggingface_hub import InferenceClient
+from datasets import load_dataset
+import json
+
+
+DATASET_REPO = "LCA/HACKATHON_PARTS" 
+
+dataset = load_dataset(DATASET_REPO, split="train")
+df = dataset.to_pandas()
+
+descriptions = df['DESIGNATION'].tolist()
+codes = df["CODE"].astype(str).tolist()
+
+
+# --- Embedding model ---
+embedding_model = SentenceTransformer("all-MiniLM-L6-v2")
+
+#--- Load or compute embeddings + FAISS index ---
+#For start, test perf without caching this
+if os.path.exists("embeddings.npy") and os.path.exists("faiss.index"):
+    embeddings = np.load("embeddings.npy")
+    index = faiss.read_index("faiss.index")
+else:
+    embeddings = embedding_model.encode(descriptions, convert_to_numpy=True)
+    faiss.normalize_L2(embeddings)
+    index = faiss.IndexFlatIP(embeddings.shape[1])
+    index.add(embeddings) 
+     # Save embeddings and index for future use
+    np.save("embeddings.npy", embeddings)
+    faiss.write_index(index, "faiss.index")
+
+# --- Inference API client ---
+# client = InferenceClient("HuggingFaceH4/zephyr-7b-beta", token=os.getenv("HF_TOKEN"))
+
+def rechercher_article(articleSource):
+   
+    source = articleSource["designation"]
+    query_embedding = embedding_model.encode([source], convert_to_numpy=True)
+    faiss.normalize_L2(query_embedding)
+    # Recherche du/des voisin(s) le(s) plus proche(s)
+    similarity_scores, indices = index.search(query_embedding, k=1)
+    # Gérer la qualité du retour avec un seuil de similarité
+    threshold = 0.7  # à ajuster selon vos tests
+    if similarity_scores[0][0] < threshold:
+        print(f"Score de similarité trop faible ({similarity_scores[0][0]:.2f}) pour '{source}'")
+        return "UNKNOWN"
+
+    
+    article = {}
+    article["code"] = codes[indices[0][0]]
+    article["designation"] = descriptions[indices[0][0]]
+    article["source"] = source
+    article["quantite"] = articleSource.get("quantite", None)
+    print(f"Code trouvé pour '{source}': {article['code']} / {article['designation']}")
+    
+   
+    return article
+
+def respond(message):
+
+    # Prompt par défaut
+    custom_prompt = """Tu es un analyseur de messages expert.
+    Ta mission est de déterminer dans le messages fourni quels sont les articles qui sont demandés et pour quelle quantité.   
+    La réponse est au format json et donne 2 informations par article identifié : la désignation et le nombre
+    La désignation est formé du type d'article et des caractéristiques comme la matière ou les dimensions
+
+    Ne retourne que le JSON.
+
+    """
+    # query_embedding = embedding_model.encode([message], convert_to_numpy=True)
+    # faiss.normalize_L2(query_embedding)
+    # _, indices = index.search(query_embedding, k=5)
+    # context = "\n".join([f"{codes[i]}: {descriptions[i]}" for i in indices[0]])
+
+    # Utilise le prompt personnalisé
+    
+
+    # message = custom_prompt.format(message=message)
+
+    messages = [{"role": "system", "content": custom_prompt}]
+    messages +=  [{"role": "user", "content": message}]
+
+    # full_response = client.text_generation(message)
+    client = InferenceClient("HuggingFaceH4/zephyr-7b-beta", token=os.getenv("HF_TOKEN"))
+    # client = InferenceClient(
+    #     "deepseek-ai/DeepSeek-R1-Distill-Qwen-32B",
+    #     token=os.getenv("HF_TOKEN"),
+    #     #provider="auto"  # or choose a supported provider from the error message
+    # )
+    
+
+    full_response = ""
+    for chunk in client.chat_completion(
+        messages,
+        max_tokens=512,
+        stream=True,
+        temperature=0.1,
+        top_p=0.8,
+    ):
+        token = chunk.choices[0].delta.content
+        if token:
+            full_response += token
+            # yield full_response.replace("\n", "\n\n")
+
+    # If you expect a JSON response, you can try to parse it here
+    # import json
+    # try:
+    order = {}
+    try:
+        data = json.loads(full_response)
+        articles = []
+        for article in data.get("articles", []):
+            found_article = rechercher_article(article)
+            if found_article != "UNKNOWN":
+                articles.append(found_article)
+        order["articles"] = articles
+        # Ajouter les champs destinataire et delai avec des valeurs figées
+        order["destinataire"] = {
+            "societe": "Société Exemple",
+            "nom": "Dupont",
+            "prenom": "Jean",
+            "email": "jean.dupont@exemple.com"
+        }
+        order["delai"] = "2024-07-15"
+    except Exception as e:
+        print("Could not parse articles:", e)
+        order = {}
+
+    return order
+    
+
+with gr.Blocks() as demo:
+    gr.Markdown("# Part identification Assistant")
+    #prompt_box = gr.Textbox(label="Prompt système", value=DEFAULT_PROMPT, lines=8)
+    #temperature_slider = gr.Slider(label="Température", minimum=0.0, maximum=1.0, value=0.1, step=0.01)
+    #top_p_slider = gr.Slider(label="Top-p", minimum=0.0, maximum=1.0, value=0.8, step=0.01)
+    message_box = gr.Textbox(label="Votre question")
+    response_box = gr.Textbox(label="Réponse de l'assistant", interactive=False, lines=30)
+    send_btn = gr.Button("Envoyer")
+
+    def chat(message):
+        history = []  # ou récupère l'historique si tu veux le gérer
+        gen = respond(message)
+        # full_response = ""
+        # for response in gen:
+        #     full_response = full_response + response
+        # On renvoie la dernière réponse et le contexte utilisé
+        # Il faut recalculer le contexte ici pour l'afficher
+        # query_embedding = embedding_model.encode([message], convert_to_numpy=True)
+        # faiss.normalize_L2(query_embedding)
+        # _, indices = index.search(query_embedding, k=5)
+        # context = "\n".join([f"{codes[i]}: {descriptions[i]}" for i in indices[0]])
+        return json.dumps(gen, indent=2, ensure_ascii=False)
+
+    send_btn.click(
+        chat,
+        inputs=[message_box],
+        outputs=[response_box]
+    )
+
+if __name__ == "__main__":
+    demo.launch(share=True)
+