from fastapi import FastAPI from fastapi.responses import HTMLResponse, StreamingResponse from pydantic import BaseModel from openai import OpenAI from langchain_huggingface import HuggingFaceEmbeddings from langchain_chroma import Chroma import os import json import re import time import io import asyncio import edge_tts import requests app = FastAPI(title="Valdeuck Maître Deuck V13") # ========================= # VOICE — voix masculine française Microsoft # ========================= TTS_VOICE = "fr-FR-RemyMultilingualNeural" # FREE CHAT MODELS (for valdeuck.fr) # model="HuggingFaceH4/zephyr-7b-gemma-v0.1", # Modèle Chat natif et gratuit # model="meta-llama/Llama-3.2-3B-Instruct", # Modèle Chat natif # ========================= # HUGGING FACE INFERENCE PROVIDERS # ========================= # Dans les Secrets du Space, définir HF_TOKEN. # HF_CHAT_MODEL peut être changé sans modifier le code. # :preferred demande au routeur HF le premier fournisseur disponible. HF_TOKEN_VALUE = ( os.getenv("HF_TOKEN") or os.getenv("HUGGINGFACEHUB_API_TOKEN") or "" ).strip() #• Qwen/Qwen2.5-Coder-7B-Instruct (Le choix idéal : très intelligent, rapide, gère parfaitement le français) #• deepseek-ai/DeepSeek-R1-Distill-Qwen-7B (Excellent modèle de raisonnement logique basé sur l'architecture ouverte de DeepSeek) #• Qwen/Qwen2.5-Coder-3B-Instruct (Ultra-léger, s'exécute instantanément sur les serveurs CPU de Hugging Face) ############################# Ne fonctionne plus sur huggingface, version avec abonnement uniquement #HF_CHAT_MODEL = os.getenv( # "HF_CHAT_MODEL", # "deepseek-ai/DeepSeek-R1-Distill-Qwen-7B", #) # #if not HF_TOKEN_VALUE: # print("ERREUR : aucun jeton HF_TOKEN / HUGGINGFACEHUB_API_TOKEN détecté.") #else: # print("Jeton Hugging Face détecté :", True) # print("Modèle configuré :", HF_CHAT_MODEL) # #client = OpenAI( # base_url="https://router.huggingface.co/v1", # api_key=HF_TOKEN_VALUE or "missing-token" #) # # ========================================================================================== # CONFIGURATION DE L'IA (Contournement gratuit via Pollinations AI) # via interface (bibliothèque) openai versus bibliothgèque standard : exclusif avec standard # ========================================================================================== # On utilise 'qwen' qui est performant, gratuit et gère parfaitement le français #HF_CHAT_MODEL = os.getenv("HF_CHAT_MODEL", "qwen") #print("Jeton d'authentification : Non requis (API Libre)") #print("Modèle configuré :", HF_CHAT_MODEL) # CORRECTION : On utilise la racine de l'URL sans le /v1 pour éviter l'erreur 405 #client = OpenAI( # base_url="https://pollinations.ai", # api_key="not-needed" #) #=========================================================================================== #################################################################### # DEBUG - Liste des modeles disponibles aucune garantie de gratuité #try: # models = client.models.list() # print("=== MODELES DISPONIBLES VIA LE ROUTEUR HF ===") # for model in models.data: # print(model.id) #except Exception as e: # print("ERREUR LISTE DES MODELES :", repr(e)) ##################################################################### ########################## Si Utilisation bibliothèque standard vs openai : exclusif avec openai HF_CHAT_MODEL = os.getenv("HF_CHAT_MODEL", "openai-fast") print("Jeton d'authentification : Non requis (API Libre)") print("Modèle configuré :", HF_CHAT_MODEL) # Utilisation de l'endpoint OpenAI officiel de Pollinations client = OpenAI( base_url="https://text.pollinations.ai/openai", api_key="not-needed" ) # ========================= # VECTOR DB # ========================= embeddings = HuggingFaceEmbeddings( model_name="intfloat/multilingual-e5-base" ) beer_db = Chroma( persist_directory="vectorstore_beers", embedding_function=embeddings ) knowledge_db = Chroma( persist_directory="vectorstore_knowledge", embedding_function=embeddings ) beer_retriever = beer_db.as_retriever(search_kwargs={"k": 6}) knowledge_retriever = knowledge_db.as_retriever(search_kwargs={"k": 3}) # ========================= # REQUEST # ========================= class Query(BaseModel): query: str # ========================= # CLEAN # ========================= def clean(t): if not t: return "" return re.sub(r"\s+", " ", t).strip() # ========================= # NETTOYAGE STRICT ANTI-ASTÉRISQUE ET ACCENT US # ========================= def clean_tts_text(text: str) -> str: if not text: return "" # 1. Supprime définitivement la syntaxe Markdown (astérisques, tirets, accents graves) t = text.replace("**", "").replace("*", "").replace("`", "") # 2. Remplacer les noms anglais par leur équivalent phonétique français # évite que le moteur vocal Remy ne bascule sur un accent américain forcé phonetic_map = { "Bongo Fury": "Bongo Fiouri", "Concrete Jungle": "Concrète Jeunguel", "China Girl": "Tchaïna Geurl", "Red Hot Kampot Pepper": "Red Hot Kampotte Pèpeur", "Elder Foot Beat": "Elder Foute Bite", "Melody Nelson": "Mélodi Nelson", "White Rabbit": "Ouaïte Rabitte", "IPA": "I P A", "Stout": "Stoute", "Sour": "Safeur", "Craft Beer": "Crafte Bière" } for eng, fr in phonetic_map.items(): t = re.sub(r'\b' + re.escape(eng) + r'\b', fr, t, flags=re.IGNORECASE) return t # ========================= # TTS ENDPOINT SÉCURISÉ CONTRE LES ERREURS RÉSEAU # Génère un fichier audio MP3 côté serveur avec edge-tts # ========================= @app.get("/tts") async def tts(text: str): try: # On applique le filtre anti-astérisque et phonétique avant l'envoi de la voix text_clean = clean_tts_text(text) if not text_clean.strip(): return StreamingResponse(io.BytesIO(), media_type="audio/mpeg") communicate = edge_tts.Communicate(text_clean, TTS_VOICE) buf = io.BytesIO() # Flux sécurisé asynchrone par morceaux (evite les coupures de proxy) async for chunk in communicate.stream(): if chunk["type"] == "audio": buf.write(chunk["data"]) buf.seek(0) # Validation du contenu généré if buf.getbuffer().nbytes == 0: return StreamingResponse(io.BytesIO(), media_type="audio/mpeg") return StreamingResponse(buf, media_type="audio/mpeg") except Exception as e: # Capture l'erreur réseau proprement sans faire crasher l'interface utilisateur print(f"❌ Erreur réseau interceptée sur le module Voix (edge-tts): {e}") return StreamingResponse(io.BytesIO(), media_type="audio/mpeg") # ========================= # UI # ========================= @app.get("/", response_class=HTMLResponse) def home(): return """ Maître Deuck
Bincherie Fabrik — Salon-de-Provence
Maître Deuck
Bonjour ! Je suis Maître Deuck, l'expert brassicole de la Valdeuck.
Comment puis-je vous aider ?
""" # ========================= # HEALTH - # ========================= @app.get("/health") def health(): return { "status": "ok", "beer": beer_db._collection.count(), "knowledge": knowledge_db._collection.count(), "tts_voice": TTS_VOICE, "time": time.time() } # ========================= # ASK # ========================= # Si openai sans Pollinations #@app.post("/ask") #def ask(q: Query): # try: # beer_docs = beer_retriever.invoke(q.query) # know_docs = knowledge_retriever.invoke(q.query) # context = "\n".join( # clean(d.page_content) for d in beer_docs + know_docs # ) # system = """Tu es Maître Deuck, expert de la Valdeuck. #Règles : réponds en 2-3 phrases maximum. Sois direct et précis. #Si l'info est absente du contexte : "Information non disponible." #Pas de marques externes. Français uniquement. #Remplace systématiquement "alc" et "d'alc" par "alcool" et "d'alcool" dans tes réponses.""" # user = f"CONTEXTE :\n{context}\n\nQUESTION : {q.query}" # if not HF_TOKEN_VALUE: # raise RuntimeError( # "Jeton absent : configurez HF_TOKEN dans les Secrets du Space." # ) # res = client.chat.completions.create( # model=HF_CHAT_MODEL, # messages=[ # {"role": "system", "content": system}, # {"role": "user", "content": user} # ], # temperature=0.2, # max_tokens=300 # ) # answer = (res.choices[0].message.content or "").strip() # if not answer: # raise RuntimeError("Le fournisseur a renvoyé une réponse vide.") # answer = answer.replace("vol d'alc/", "vol d'alcool/") # answer = answer.replace("% alc", "% alcool") # answer = answer.replace("% d'alc", "% d'alcool") # answer = answer.replace(" alc.", " alcool") # answer = answer.replace(" alc/", " alcool/") # answer = answer.replace("Valdeuck Bincherie Fabrik", "Valdeuck") # return {"response": answer} # except Exception as e: # print(f"❌ Erreur critique dans /ask ({type(e).__name__}) : {str(e)}") # return { # "response": ( # "Maître Deuck est momentanément indisponible. " # "Réessayez dans quelques instants." # ) # } @app.post("/ask") def ask(q: Query): try: beer_docs = beer_retriever.invoke(q.query) know_docs = knowledge_retriever.invoke(q.query) context = "\n".join( clean(d.page_content) for d in beer_docs + know_docs ) system = """Tu es Maître Deuck, expert de la Valdeuck. Règles : réponds en 2-3 phrases maximum. Sois direct et précis. Si l'info est absente du contexte : "Information non disponible." Pas de marques externes. Français uniquement. Remplace systématiquement "alc" et "d'alc" par "alcool" et "d'alcool" dans tes réponses.""" user = f"CONTEXTE :\n{context}\n\nQUESTION : {q.query}" # Appel direct à Pollinations.ai via son endpoint text compatible OpenAI url = "https://text.pollinations.ai/" payload = { "messages": [ {"role": "system", "content": system}, {"role": "user", "content": user} ], "model": HF_CHAT_MODEL, "json": False } print(f"Envoi de la requête à Pollinations pour : {q.query}") res = requests.post(url, json=payload, timeout=30) if res.status_code == 200: answer = res.text.strip() print("Réponse reçue avec succès !") else: raise Exception(f"Erreur serveur Pollinations code {res.status_code}: {res.text}") if not answer: raise RuntimeError("Le fournisseur a renvoyé une réponse vide.") # Post-traitements de nettoyage answer = answer.replace("vol d'alc/", "vol d'alcool/") answer = answer.replace("% alc", "% alcool") answer = answer.replace("% d'alc", "% d'alcool") answer = answer.replace(" alc.", " alcool") answer = answer.replace(" alc/", " alcool/") answer = answer.replace("Valdeuck Bincherie Fabrik", "Valdeuck") return {"response": answer} except Exception as e: print(f"❌ Erreur critique dans /ask ({type(e).__name__}) : {str(e)}") return { "response": ( "Maître Deuck est momentanément indisponible. " "Réessayez dans quelques instants." ) }