Initial PWA assistant proof of concept
This commit is contained in:
237
app.py
Normal file
237
app.py
Normal file
@@ -0,0 +1,237 @@
|
||||
import base64
|
||||
import json
|
||||
import os
|
||||
import smtplib
|
||||
import tempfile
|
||||
from email.mime.multipart import MIMEMultipart
|
||||
from email.mime.text import MIMEText
|
||||
from pathlib import Path
|
||||
|
||||
from dotenv import load_dotenv
|
||||
from flask import Flask, jsonify, request, send_from_directory
|
||||
from groq import Groq
|
||||
|
||||
load_dotenv()
|
||||
|
||||
ROOT = Path(__file__).resolve().parent
|
||||
STATIC_DIR = ROOT / "static"
|
||||
CONTACTS_FILE = ROOT / "contacts.json"
|
||||
|
||||
GROQ_API_KEY = os.getenv("GROQ_API_KEY", "")
|
||||
GMAIL_EMAIL = os.getenv("GMAIL_EMAIL", "")
|
||||
GMAIL_PASSWORD = os.getenv("GMAIL_PASSWORD", "")
|
||||
|
||||
CHAT_MODEL = "llama-3.3-70b-versatile"
|
||||
STT_MODEL = "whisper-large-v3-turbo"
|
||||
TTS_MODEL = "canopylabs/orpheus-v1-english"
|
||||
TTS_VOICE = "troy"
|
||||
|
||||
app = Flask(__name__, static_folder=str(STATIC_DIR), static_url_path="")
|
||||
app.secret_key = os.getenv("APP_SECRET", "dev-secret-change-me")
|
||||
|
||||
client = Groq(api_key=GROQ_API_KEY)
|
||||
histories = {}
|
||||
|
||||
SYSTEM_PROMPT = """Tu es l'assistant personnel vocal de Laurent.
|
||||
Tu peux discuter normalement en francais et aider a envoyer des emails via Gmail.
|
||||
|
||||
Quand l'utilisateur veut envoyer un email, collecte les informations manquantes :
|
||||
destinataire, sujet, contenu.
|
||||
|
||||
Quand tu as toutes les informations pour envoyer un email, reponds uniquement avec ce JSON :
|
||||
{"action": "envoyer_email", "destinataire": "...", "sujet": "...", "corps": "..."}
|
||||
|
||||
Pour tout le reste, reponds en francais naturel, avec des phrases courtes.
|
||||
En mode voiture, sois bref : une ou deux phrases maximum."""
|
||||
|
||||
|
||||
def charger_contacts():
|
||||
try:
|
||||
return json.loads(CONTACTS_FILE.read_text(encoding="utf-8"))
|
||||
except FileNotFoundError:
|
||||
return {"laurent": "barontil@hotmail.com"}
|
||||
|
||||
|
||||
def construire_system_prompt():
|
||||
contacts = charger_contacts()
|
||||
lignes = [f"- {nom.title()} : {email}" for nom, email in contacts.items()]
|
||||
return SYSTEM_PROMPT + "\n\nContacts connus :\n" + "\n".join(lignes)
|
||||
|
||||
|
||||
def resoudre_contact(destinataire):
|
||||
contacts = charger_contacts()
|
||||
valeur = destinataire.strip()
|
||||
return contacts.get(valeur.lower(), valeur)
|
||||
|
||||
|
||||
def extraire_action_email(reponse):
|
||||
decodeur = json.JSONDecoder()
|
||||
for index, caractere in enumerate(reponse):
|
||||
if caractere != "{":
|
||||
continue
|
||||
try:
|
||||
data, _ = decodeur.raw_decode(reponse[index:])
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
if isinstance(data, dict) and data.get("action") == "envoyer_email":
|
||||
return data
|
||||
return None
|
||||
|
||||
|
||||
def envoyer_email(destinataire, sujet, corps):
|
||||
destinataire = resoudre_contact(destinataire)
|
||||
msg = MIMEMultipart()
|
||||
msg["From"] = GMAIL_EMAIL
|
||||
msg["To"] = destinataire
|
||||
msg["Subject"] = sujet
|
||||
msg.attach(MIMEText(corps, "plain", "utf-8"))
|
||||
|
||||
with smtplib.SMTP("smtp.gmail.com", 587) as serveur:
|
||||
serveur.starttls()
|
||||
serveur.login(GMAIL_EMAIL, GMAIL_PASSWORD)
|
||||
serveur.sendmail(GMAIL_EMAIL, destinataire, msg.as_string())
|
||||
|
||||
return destinataire
|
||||
|
||||
|
||||
def demander_au_llm(session_id, texte):
|
||||
historique = histories.setdefault(session_id, [])
|
||||
historique.append({"role": "user", "content": texte})
|
||||
del historique[:-20]
|
||||
|
||||
reponse = client.chat.completions.create(
|
||||
model=CHAT_MODEL,
|
||||
messages=[{"role": "system", "content": construire_system_prompt()}] + historique,
|
||||
temperature=0.45,
|
||||
max_tokens=900,
|
||||
)
|
||||
contenu = reponse.choices[0].message.content
|
||||
historique.append({"role": "assistant", "content": contenu})
|
||||
return contenu
|
||||
|
||||
|
||||
def transcrire_audio(fichier):
|
||||
suffix = Path(fichier.filename or "voice.webm").suffix or ".webm"
|
||||
with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
|
||||
fichier.save(tmp)
|
||||
chemin = tmp.name
|
||||
|
||||
try:
|
||||
with open(chemin, "rb") as audio:
|
||||
transcription = client.audio.transcriptions.create(
|
||||
model=STT_MODEL,
|
||||
file=audio,
|
||||
language="fr",
|
||||
response_format="text",
|
||||
)
|
||||
return transcription if isinstance(transcription, str) else transcription.text
|
||||
finally:
|
||||
try:
|
||||
os.remove(chemin)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
def generer_audio(texte):
|
||||
texte = texte.strip()
|
||||
if len(texte) > 900:
|
||||
texte = texte[:897] + "..."
|
||||
|
||||
reponse_audio = client.audio.speech.create(
|
||||
model=TTS_MODEL,
|
||||
voice=TTS_VOICE,
|
||||
input=texte,
|
||||
response_format="wav",
|
||||
speed=1.0,
|
||||
)
|
||||
|
||||
with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as tmp:
|
||||
chemin = tmp.name
|
||||
|
||||
try:
|
||||
reponse_audio.write_to_file(chemin)
|
||||
audio = Path(chemin).read_bytes()
|
||||
return "data:audio/wav;base64," + base64.b64encode(audio).decode("ascii")
|
||||
finally:
|
||||
try:
|
||||
os.remove(chemin)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
def traiter_message(session_id, texte, avec_audio=False):
|
||||
reponse = demander_au_llm(session_id, texte)
|
||||
action = extraire_action_email(reponse)
|
||||
email_sent = False
|
||||
|
||||
if action:
|
||||
destinataire = envoyer_email(action["destinataire"], action["sujet"], action["corps"])
|
||||
reponse = f"Email envoye a {destinataire}."
|
||||
email_sent = True
|
||||
|
||||
audio_url = None
|
||||
if avec_audio:
|
||||
audio_url = generer_audio(reponse)
|
||||
|
||||
return {
|
||||
"reply": reponse,
|
||||
"audioUrl": audio_url,
|
||||
"emailSent": email_sent,
|
||||
}
|
||||
|
||||
|
||||
@app.get("/")
|
||||
def index():
|
||||
return send_from_directory(STATIC_DIR, "index.html")
|
||||
|
||||
|
||||
@app.get("/health")
|
||||
def health():
|
||||
return jsonify({"ok": True})
|
||||
|
||||
|
||||
@app.post("/api/chat")
|
||||
def chat():
|
||||
data = request.get_json(force=True)
|
||||
texte = (data.get("text") or "").strip()
|
||||
session_id = data.get("sessionId") or "default"
|
||||
avec_audio = bool(data.get("withAudio"))
|
||||
|
||||
if not texte:
|
||||
return jsonify({"error": "Message vide"}), 400
|
||||
|
||||
try:
|
||||
return jsonify(traiter_message(session_id, texte, avec_audio))
|
||||
except Exception as e:
|
||||
print(f"Erreur /api/chat : {e}", flush=True)
|
||||
return jsonify({"error": str(e)}), 500
|
||||
|
||||
|
||||
@app.post("/api/voice")
|
||||
def voice():
|
||||
fichier = request.files.get("audio")
|
||||
session_id = request.form.get("sessionId") or "default"
|
||||
|
||||
if not fichier:
|
||||
return jsonify({"error": "Audio manquant"}), 400
|
||||
|
||||
try:
|
||||
texte = transcrire_audio(fichier).strip()
|
||||
resultat = traiter_message(session_id, texte, avec_audio=True)
|
||||
resultat["transcript"] = texte
|
||||
return jsonify(resultat)
|
||||
except Exception as e:
|
||||
print(f"Erreur /api/voice : {e}", flush=True)
|
||||
return jsonify({"error": str(e)}), 500
|
||||
|
||||
|
||||
@app.post("/api/reset")
|
||||
def reset():
|
||||
data = request.get_json(silent=True) or {}
|
||||
session_id = data.get("sessionId") or "default"
|
||||
histories.pop(session_id, None)
|
||||
return jsonify({"ok": True})
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
app.run(host="0.0.0.0", port=int(os.getenv("PORT", "8080")), debug=True)
|
||||
Reference in New Issue
Block a user