admin/openbot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
9edba16a4eecbe46b806cd9f43ee03832b59de4d
openbot/app.py
laurentbarontini 9edba16a4e Improve router
2026-04-27 17:06:30 +02:00

1204 lines
43 KiB
Python
Raw Blame History

import base64
import json
import os
import re
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
from memory import (
add_memory,
add_repository,
add_task,
close_task,
deactivate_client,
execute_read_query,
get_memory_snapshot,
init_db,
list_clients,
list_open_tasks,
search_memory,
upsert_client,
upsert_project,
)
load_dotenv()
ROOT = Path(__file__).resolve().parent
STATIC_DIR = ROOT / "static"
CONTACTS_FILE = Path(os.getenv("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"
TTS_PROVIDER = os.getenv("TTS_PROVIDER", "browser").lower()
app = Flask(__name__, static_folder=str(STATIC_DIR), static_url_path="")
app.secret_key = os.getenv("APP_SECRET", "dev-secret-change-me")
init_db()
client = Groq(api_key=GROQ_API_KEY)
histories = {}
session_contexts = {}
SYSTEM_PROMPT = """Tu es l'assistant personnel vocal de Laurent.
Tu peux discuter normalement en francais et aider Laurent avec sa memoire professionnelle.
Le routage des actions, des lectures SQL et des confirmations est gere avant toi.
Si tu recois une question generale, reponds naturellement avec le contexte fourni.
Si une information manque pour repondre correctement, demande une precision.
Pour tout le reste, reponds en francais naturel, avec des phrases courtes.
N'utilise pas d'anglais sauf si l'utilisateur le demande explicitement.
En mode voiture, sois bref : une ou deux phrases maximum."""
ROUTER_PROMPT = """Tu es le routeur d'intention de l'assistant personnel de Laurent.
Ton role est de comparer le message utilisateur avec la liste des actions disponibles et le contexte fourni.
Reponds uniquement avec un JSON valide, sans markdown.
Format obligatoire :
{
"type": "action" | "answer" | "clarification",
"confidence": 0.0,
"action": null | {"action": "..."} | {"actions": [{"action": "..."}]},
"reply": "",
"question": ""
}
Actions disponibles :
- envoyer_email: {"action":"envoyer_email","destinataire":"...","sujet":"...","corps":"..."}
- ajouter_contact: {"action":"ajouter_contact","nom":"...","email":"..."}
- ajouter_client: {"action":"ajouter_client","nom":"...","notes":"...","statut":"actif"}
- supprimer_client: {"action":"supprimer_client","nom":"..."}
- ajouter_projet: {"action":"ajouter_projet","nom":"...","client":"...","statut":"...","resume":"...","prochaine_action":"..."}
- ajouter_tache: {"action":"ajouter_tache","titre":"...","client":"...","projet":"...","details":"...","echeance":"..."}
- terminer_tache: {"action":"terminer_tache","id":123}
- ajouter_repo: {"action":"ajouter_repo","nom":"...","projet":"...","chemin":"...","url":"...","branche":"..."}
- ajouter_note: {"action":"ajouter_note","contenu":"...","sujet":"...","client":"...","projet":"...","type":"note"}
- interroger_base: {"action":"interroger_base","sql":"SELECT ...","question":"..."}
Schema SQLite disponible pour interroger_base :
- clients(id, name, status, notes, created_at, updated_at)
- projects(id, client_id, name, status, summary, next_action, created_at, updated_at)
- tasks(id, client_id, project_id, title, details, due_at, done, created_at, updated_at)
- repositories(id, project_id, name, local_path, remote_url, main_branch, last_indexed_at, created_at, updated_at)
- memories(id, client_id, project_id, kind, subject, content, created_at)
Relations :
- projects.client_id -> clients.id
- tasks.client_id -> clients.id
- tasks.project_id -> projects.id
- repositories.project_id -> projects.id
- memories.client_id -> clients.id
- memories.project_id -> projects.id
Regles :
- Choisis une action seulement si l'intention et les arguments sont clairs.
- Si plusieurs actions sont necessaires, utilise {"actions":[...]}.
- Si l'utilisateur demande une modification mais que la cible est ambigue, type="clarification".
- Si l'utilisateur pose une question generale et que le contexte suffit, type="answer" et reponds directement.
- Pour toute question de lecture ou d'analyse sur la memoire professionnelle, prefere interroger_base avec un SELECT.
- Pour interroger_base, SQL doit etre une seule requete SELECT, sans point-virgule, sans PRAGMA, sans ecriture, avec LIMIT si le resultat peut etre long.
- Si aucune action ne correspond, type="answer".
- confidence >= 0.80 pour executer une action sans confirmation.
- confidence entre 0.45 et 0.79 : type="clarification".
- N'invente pas d'information absente du contexte. Si la demande porte sur une categorie inconnue, demande une precision.
- Pour "nouveaux clients", ne suppose pas que tous les clients sont nouveaux. Demande si Laurent veut les derniers ajoutes, tous les clients, ou une notion precise de nouveau.
- Pour supprimer/terminer une tache, utilise l'id si le contexte de conversation contient les dernieres taches affichees.
- Pour retirer/enlever/supprimer un client de la liste, utilise supprimer_client.
- Pour "quel client a le plus de taches", utilise interroger_base avec COUNT, GROUP BY, ORDER BY.
- Pour "quelles sont mes taches", "mes taches a faire", "taches pour mes clients", utilise interroger_base sur tasks avec clients/projects.
- Le dernier message est prioritaire, mais tu dois interpreter les pronoms et references implicites avec toute la conversation fournie.
- Si l'utilisateur dit "lui", "le", "la", "ca", "ça", "ce client", "ce projet" ou "cette tache", retrouve l'objet vise dans l'historique recent avant de choisir l'action.
- Si tu ne peux pas identifier clairement l'objet vise, type="clarification".
- Les questions de suivi courtes doivent etre resolues avec l'historique recent avant de choisir interroger_base, une action directe ou une clarification.
"""
def charger_contacts():
try:
return json.loads(CONTACTS_FILE.read_text(encoding="utf-8"))
except FileNotFoundError:
return {"laurent": "barontil@hotmail.com"}
def sauvegarder_contacts(contacts):
CONTACTS_FILE.parent.mkdir(parents=True, exist_ok=True)
CONTACTS_FILE.write_text(
json.dumps(contacts, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
def construire_system_prompt(session_id=None):
contacts = charger_contacts()
lignes = [f"- {nom.title()} : {email}" for nom, email in contacts.items()]
return (
SYSTEM_PROMPT
+ "\n\nContacts connus :\n"
+ "\n".join(lignes)
+ "\n\nMemoire professionnelle actuelle :\n"
+ construire_contexte_memoire(session_id)
)
def construire_contexte_memoire(session_id=None):
snapshot = get_memory_snapshot()
contexte = session_contexts.get(session_id or "", {})
lignes = []
clients_mem = snapshot["clients"]
if clients_mem:
lignes.append("Clients connus et taches ouvertes :")
for client_mem in clients_mem:
lignes.append(f"- {client_mem['name']} : {client_mem['open_task_count']} tache(s) ouverte(s)")
else:
lignes.append("Aucun client connu.")
if snapshot["tasks"]:
lignes.append("Taches ouvertes recentes :")
for tache in snapshot["tasks"][:15]:
sujet = tache.get("client_name") or tache.get("project_name") or "general"
lignes.append(f"- #{tache['id']} {sujet} : {tache['title']}")
if snapshot["projects"]:
lignes.append("Projets recents :")
for projet in snapshot["projects"][:10]:
client_nom = projet.get("client_name") or "sans client"
statut = projet.get("status") or "actif"
lignes.append(f"- {projet['name']} ({client_nom}) : {statut}")
if snapshot["repositories"]:
lignes.append("Depots Git associes :")
for repo in snapshot["repositories"][:10]:
cible = repo.get("project_name") or "sans projet"
chemin = repo.get("local_path") or repo.get("remote_url") or "chemin non renseigne"
lignes.append(f"- {repo['name']} -> {cible} : {chemin}")
if contexte:
lignes.append("Contexte de la conversation en cours :")
if contexte.get("last_client"):
lignes.append(f"- dernier client mentionne : {contexte['last_client']}")
if contexte.get("last_project"):
lignes.append(f"- dernier projet mentionne : {contexte['last_project']}")
if contexte.get("tasks"):
ids = ", ".join(f"#{tache['id']} {tache['title']}" for tache in contexte["tasks"][:5])
lignes.append(f"- dernieres taches affichees : {ids}")
lignes.append(
"Utilise ce contexte pour repondre aux questions globales. "
"Pour modifier la memoire, reponds avec une action JSON."
)
return "\n".join(lignes)
def resoudre_contact(destinataire):
contacts = charger_contacts()
valeur = destinataire.strip()
return contacts.get(valeur.lower(), valeur)
def extraire_action(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") in ACTIONS or isinstance(data.get("actions"), list)):
return data
return None
ACTIONS = {
"envoyer_email",
"ajouter_contact",
"ajouter_client",
"supprimer_client",
"ajouter_projet",
"ajouter_tache",
"terminer_tache",
"ajouter_repo",
"ajouter_note",
"chercher_memoire",
"lister_clients",
"lister_taches",
"analyser_taches",
"interroger_base",
}
def ajouter_contact(nom, email):
contacts = charger_contacts()
nom_normalise = " ".join(nom.strip().lower().split())
email_normalise = email.strip()
if not nom_normalise or "@" not in email_normalise:
raise ValueError("Contact incomplet")
contacts[nom_normalise] = email_normalise
sauvegarder_contacts(contacts)
return nom_normalise, email_normalise
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(session_id)}] + historique,
temperature=0.45,
max_tokens=900,
)
contenu = reponse.choices[0].message.content
historique.append({"role": "assistant", "content": contenu})
return contenu
def router_intention(session_id, texte):
historique = histories.get(session_id, [])[-20:]
messages = [
{"role": "system", "content": ROUTER_PROMPT},
{
"role": "system",
"content": (
"Contexte disponible pour router la demande :\n"
+ construire_contexte_routeur(session_id)
),
},
] + historique + [{"role": "user", "content": texte}]
reponse = client.chat.completions.create(
model=CHAT_MODEL,
messages=messages,
temperature=0.0,
max_tokens=700,
)
contenu = reponse.choices[0].message.content or ""
data = extraire_json_routeur(contenu)
if not data:
return {
"type": "answer",
"confidence": 0.0,
"action": None,
"reply": contenu.strip() or "Je n'ai pas compris. Tu peux reformuler ?",
"question": "",
}
return normaliser_decision_routeur(data)
def construire_contexte_routeur(session_id=None):
contacts = charger_contacts()
contacts_connus = "\n".join(f"- {nom.title()} : {email}" for nom, email in contacts.items())
return (
"Contacts email connus :\n"
+ (contacts_connus or "- aucun")
+ "\n\nMemoire professionnelle actuelle :\n"
+ construire_contexte_memoire(session_id)
)
def extraire_json_routeur(contenu):
decodeur = json.JSONDecoder()
for index, caractere in enumerate(contenu):
if caractere != "{":
continue
try:
data, _ = decodeur.raw_decode(contenu[index:])
except json.JSONDecodeError:
continue
if isinstance(data, dict):
return data
return None
def normaliser_decision_routeur(data):
type_decision = data.get("type") or "answer"
if type_decision not in {"action", "answer", "clarification"}:
type_decision = "answer"
try:
confidence = float(data.get("confidence", 0))
except (TypeError, ValueError):
confidence = 0.0
confidence = max(0.0, min(1.0, confidence))
action = data.get("action")
if isinstance(action, dict) and not action_valide(action):
action = None
if type_decision == "action" and (confidence < 0.80 or not action):
type_decision = "clarification"
return {
"type": type_decision,
"confidence": confidence,
"action": action,
"reply": str(data.get("reply") or "").strip(),
"question": str(data.get("question") or "").strip(),
}
def action_valide(action):
if action.get("action") in ACTIONS:
return True
actions = action.get("actions")
if isinstance(actions, list) and actions:
return all(isinstance(item, dict) and item.get("action") in ACTIONS for item in actions)
return False
def detecter_action_directe(texte, session_id=None):
texte_nettoye = texte.strip()
texte_min = texte_nettoye.lower()
clients_a_ajouter = extraire_liste_clients(texte_nettoye)
if clients_a_ajouter:
return {
"actions": [
{
"action": "ajouter_client",
"nom": nom,
"notes": "Ajoute depuis une liste fournie par Laurent.",
"statut": "actif",
}
for nom in clients_a_ajouter
]
}
suppression_client = detecter_suppression_client(texte_nettoye)
if suppression_client:
return suppression_client
analyse_taches = detecter_analyse_taches(texte_nettoye)
if analyse_taches:
return analyse_taches
tache_client = detecter_tache_client(texte_nettoye)
if tache_client:
return tache_client
recherche_taches = detecter_recherche_taches(texte_nettoye)
if recherche_taches:
return recherche_taches
suppression_tache = detecter_suppression_tache(texte_nettoye, session_id)
if suppression_tache:
return suppression_tache
recherche_match = re.search(
r"(?:qu(?:'| )?est[- ]?ce que tu sais sur|que sais[- ]?tu sur|cherche dans la memoire|memoire sur)\s+(.+)",
texte_nettoye,
flags=re.IGNORECASE,
)
if recherche_match:
return {
"action": "chercher_memoire",
"requete": recherche_match.group(1).strip(" ?.,;:"),
}
if detecter_liste_clients_demandee(texte_min):
return {"action": "lister_clients"}
if re.search(r"\b(liste|affiche|montre)\b.*\b(projets?|taches?|repos?|depots?)\b", texte_min):
return {"action": "chercher_memoire", "requete": ""}
repo_match = re.search(
r"(?:associe|ajoute|note)\s+(?:le\s+)?repo\s+(\S+)(?:\s+au\s+projet\s+(.+))?",
texte_nettoye,
flags=re.IGNORECASE,
)
if repo_match:
chemin_ou_url = repo_match.group(1).strip(" .,;:")
projet = (repo_match.group(2) or "").strip(" .,;:")
return {
"action": "ajouter_repo",
"chemin": chemin_ou_url if not chemin_ou_url.startswith(("http://", "https://", "git@")) else "",
"url": chemin_ou_url if chemin_ou_url.startswith(("http://", "https://", "git@")) else "",
"projet": projet,
}
client_match = re.search(
r"\b([A-Z0-9][A-Z0-9&.\- ]{2,})\s+est\s+(?:un|une)\s+nouveau\s+client\b",
texte_nettoye,
flags=re.IGNORECASE,
)
if not client_match:
return None
nom_client = client_match.group(1).strip(" .,;:")
actions = [
{
"action": "ajouter_client",
"nom": nom_client,
"notes": texte_nettoye,
"statut": "actif",
}
]
rappel_match = re.search(
r"(?:je\s+dois\s+|il\s+faut\s+|note\s+(?:quelque\s+part\s+)?que\s+je\s+dois\s+)(rappeler\s+.+)",
texte_nettoye,
flags=re.IGNORECASE,
)
if rappel_match:
actions.append(
{
"action": "ajouter_tache",
"titre": rappel_match.group(1).strip(" .,;:"),
"client": nom_client,
"details": texte_nettoye,
}
)
return {"actions": actions}
def detecter_tache_client(texte):
if not re.search(r"\b(note|notes|noter|tache|doit|dois|a faire|à faire)\b", texte, flags=re.IGNORECASE):
return None
match = re.search(
r"\bpour\s+(?:mon\s+client\s+|le\s+client\s+)?([A-Z0-9][A-Z0-9&.\- ]{1,40}?)[,\s]+(?:on\s+)?(?:doit|dois|devra|faut)\s+(.+?)(?:\s+tu\s+peux\s+noter.*|\s+note.*|\s+notes.*|\s*$)",
texte,
flags=re.IGNORECASE,
)
if not match:
return None
client_nom = nettoyer_nom_client(match.group(1))
titre = nettoyer_titre_tache(match.group(2))
if not client_nom or not titre:
return None
return {
"action": "ajouter_tache",
"titre": titre,
"client": client_nom,
"details": texte,
}
def detecter_suppression_client(texte):
if not re.search(r"\b(enleve|enlève|supprime|retire|efface|desactive|désactive)\b", texte, flags=re.IGNORECASE):
return None
if re.search(r"\btaches?\b|\bprojets?\b|\brepos?\b|\bdepots?\b|\bdépôts?\b", texte, flags=re.IGNORECASE):
return None
match = re.search(
r"\b(?:enleve|enlève|supprime|retire|efface|desactive|désactive)\s+(?:le\s+client\s+|client\s+)?(.+?)(?:\s+(?:de|dans)\s+(?:la\s+)?liste.*)?$",
texte,
flags=re.IGNORECASE,
)
if not match:
return None
nom = nettoyer_nom_client(match.group(1))
if not nom:
return None
return {"action": "supprimer_client", "nom": nom}
def detecter_recherche_taches(texte):
if re.search(
r"\b(pour\s+quel\s+client|quel\s+client|plus\s+de\s+taches?|plus\s+de\s+tâches?|combien|classement|top)\b",
texte,
flags=re.IGNORECASE,
):
return None
match = re.search(
r"\b(?:que\s+dois[- ]?je\s+faire|quoi\s+faire|taches?|a faire|à faire)\b.*\b(?:pour|sur)\s+(?:mon\s+client\s+|le\s+client\s+)?([A-Z0-9][A-Z0-9&.\- ]{1,40})",
texte,
flags=re.IGNORECASE,
)
if not match:
return None
client_nom = nettoyer_nom_client(match.group(1))
if not client_nom:
return None
return {"action": "chercher_memoire", "requete": client_nom}
def detecter_analyse_taches(texte):
if not re.search(r"\bclients?\b", texte, flags=re.IGNORECASE):
return None
if not re.search(r"\b(taches?|tâches?|a faire|à faire|reste)\b", texte, flags=re.IGNORECASE):
return None
if re.search(
r"\b(plus\s+de|le\s+plus|maximum|max|prioritaire|charge|chargé|chargee|chargée)\b",
texte,
flags=re.IGNORECASE,
):
return {"action": "analyser_taches", "type": "client_plus_taches"}
return None
def detecter_suppression_tache(texte, session_id):
if not re.search(r"\b(enleve|enlève|supprime|retire|efface|termine|marque.*faite)\b", texte, flags=re.IGNORECASE):
return None
if not re.search(r"\btaches?\b", texte, flags=re.IGNORECASE):
return None
contexte = session_contexts.get(session_id or "", {})
taches = contexte.get("tasks") or []
if not taches:
return None
client_match = re.search(
r"\b(?:pour|sur)\s+(?:mon\s+client\s+|le\s+client\s+)?([A-Z0-9][A-Z0-9&.\- ]{1,40})",
texte,
flags=re.IGNORECASE,
)
client_nom = nettoyer_nom_client(client_match.group(1)) if client_match else contexte.get("last_client", "")
candidates = [
tache for tache in taches
if not client_nom or (tache.get("client_name") or "").lower() == client_nom.lower()
]
if not candidates:
candidates = taches
if re.search(r"\b(derniere|dernière|dernier|derniere affichee|dernière affichée)\b", texte, flags=re.IGNORECASE):
task_id = candidates[-1].get("id")
else:
task_id = candidates[0].get("id") if len(candidates) == 1 else None
if not task_id:
return None
return {"action": "terminer_tache", "id": task_id}
def detecter_liste_clients_demandee(texte_min):
if not re.search(r"\bclients?\b", texte_min):
return False
return bool(
re.search(r"\b(liste|affiche|montre|donne)\b.*\bclients?\b", texte_min)
or re.search(r"\bclients?\b.*\b(liste|affiche|montre|donne)\b", texte_min)
or re.search(r"\b(quels?|qui)\s+sont\b.*\bclients?\b", texte_min)
or re.search(r"\btous\s+mes\s+clients?\b", texte_min)
or re.search(r"\bmes\s+clients?\b", texte_min)
)
def extraire_liste_clients(texte):
if not re.search(r"\bclients?\b", texte, flags=re.IGNORECASE):
return []
if not re.search(r"\b(voici|liste|note|notes|noter|memorise|ajoute|ajoutes)\b", texte, flags=re.IGNORECASE):
return []
lignes = [ligne.strip(" -\t\r.,;") for ligne in texte.splitlines()]
candidats = []
apres_entete = False
for ligne in lignes:
if not ligne:
continue
ligne_min = ligne.lower()
if "client" in ligne_min:
apres_entete = True
apres_deux_points = ligne.split(":", 1)[1].strip() if ":" in ligne else ""
if apres_deux_points:
candidats.extend(separer_noms_clients(apres_deux_points))
continue
if not apres_entete:
continue
if re.search(r"\b(note|notes|noter|quelque part|merci|stp|s'il te plait)\b", ligne_min):
continue
candidats.extend(separer_noms_clients(ligne))
noms = []
deja_vus = set()
for candidat in candidats:
nom = nettoyer_nom_client(candidat)
if not nom:
continue
cle = nom.lower()
if cle not in deja_vus:
noms.append(nom)
deja_vus.add(cle)
return noms
def separer_noms_clients(texte):
morceaux = re.split(r"[,;/]|\s+\bet\b\s+", texte)
return [morceau.strip() for morceau in morceaux if morceau.strip()]
def nettoyer_nom_client(nom):
nom = nom.strip(" -\t\r.,;:")
nom = re.sub(
r"\s+\b(je|j'ai|j ai|stp|svp|s'il te plait|s'il te plaît|merci)\b.*$",
"",
nom,
flags=re.IGNORECASE,
).strip(" -\t\r.,;:")
if not nom or len(nom) < 2:
return ""
if re.search(r"\b(liste|clients?|voici|note|notes|noter|quelque|part)\b", nom, flags=re.IGNORECASE):
return ""
if len(nom.split()) > 4:
return ""
return nom
def nettoyer_titre_tache(titre):
titre = titre.strip(" -\t\r.,;:?")
titre = re.sub(r"^de\s+", "", titre, flags=re.IGNORECASE)
titre = re.sub(r"\s+", " ", titre)
return titre[:1].upper() + titre[1:] if titre else ""
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 executer_action(action):
nom_action = action.get("action")
if nom_action == "envoyer_email":
destinataire = envoyer_email(action["destinataire"], action["sujet"], action["corps"])
return f"Email envoye a {destinataire}.", {"emailSent": True}
if nom_action == "ajouter_contact":
nom, email = ajouter_contact(action["nom"], action["email"])
return f"Contact ajoute : {nom.title()}.", {"contactSaved": True}
if nom_action == "ajouter_client":
client_mem = upsert_client(
action.get("nom", ""),
notes=action.get("notes", ""),
status=action.get("statut", "actif"),
)
return f"C'est note : {client_mem['name']} est dans tes clients.", {"memorySaved": True}
if nom_action == "supprimer_client":
client_mem = deactivate_client(action.get("nom", ""))
nom = action.get("nom", "").strip()
if not client_mem:
return f"Je n'ai pas trouve le client {nom} dans ta liste.", {"memoryRead": True}
return f"C'est fait : {client_mem['name']} n'apparaitra plus dans tes clients actifs.", {"memorySaved": True}
if nom_action == "ajouter_projet":
projet = upsert_project(
action.get("nom", ""),
client_name=action.get("client", ""),
status=action.get("statut", "actif"),
summary=action.get("resume", ""),
next_action=action.get("prochaine_action", ""),
)
return f"C'est note pour le projet {projet['name']}.", {"memorySaved": True}
if nom_action == "ajouter_tache":
tache = add_task(
action.get("titre", ""),
client_name=action.get("client", ""),
project_name=action.get("projet", ""),
details=action.get("details", ""),
due_at=action.get("echeance", ""),
)
return f"Tache ajoutee : {tache['title']}.", {"memorySaved": True}
if nom_action == "terminer_tache":
tache = close_task(action.get("id"))
if not tache:
return "Je n'ai pas retrouve cette tache ouverte.", {"memoryRead": True}
contexte = tache.get("client_name") or tache.get("project_name") or "memoire professionnelle"
return f"C'est fait : j'ai retire la tache {tache['title']} pour {contexte}.", {"memorySaved": True}
if nom_action == "ajouter_repo":
repo = add_repository(
action.get("nom", ""),
project_name=action.get("projet", ""),
local_path=action.get("chemin", ""),
remote_url=action.get("url", ""),
main_branch=action.get("branche", ""),
)
return f"Depot associe : {repo['name']}.", {"memorySaved": True}
if nom_action == "ajouter_note":
note = add_memory(
action.get("contenu", ""),
kind=action.get("type", "note"),
subject=action.get("sujet", ""),
client_name=action.get("client", ""),
project_name=action.get("projet", ""),
)
sujet = note["subject"] or "memoire professionnelle"
return f"C'est note dans {sujet}.", {"memorySaved": True}
if nom_action == "chercher_memoire":
resultat = search_memory(action.get("requete", ""))
return formater_memoire(resultat), {"memoryRead": True, "memoryResult": resultat, "memoryQuery": action.get("requete", "")}
if nom_action == "lister_clients":
clients_mem = list_clients()
return formater_liste_clients(clients_mem), {
"memoryRead": True,
"memoryResult": {"clients": clients_mem, "projects": [], "tasks": [], "repositories": [], "memories": []},
"memoryQuery": "",
}
if nom_action == "lister_taches":
taches = list_open_tasks()
return formater_liste_taches(taches), {
"memoryRead": True,
"memoryResult": {"clients": [], "projects": [], "tasks": taches, "repositories": [], "memories": []},
"memoryQuery": "",
}
if nom_action == "analyser_taches":
if action.get("type") == "client_plus_taches":
snapshot = get_memory_snapshot(limit=100)
reponse, client_nom = formater_client_plus_taches(snapshot)
resultat_client = search_memory(client_nom) if client_nom else None
return reponse, {
"memoryRead": True,
"memoryResult": resultat_client or {
"clients": snapshot["clients"],
"projects": snapshot["projects"],
"tasks": snapshot["tasks"],
"repositories": snapshot["repositories"],
"memories": [],
},
"memoryQuery": "",
}
if nom_action == "interroger_base":
sql = action.get("sql", "")
question = action.get("question") or action.get("requete") or ""
rows = execute_read_query(sql)
reponse = resumer_resultat_sql(question, sql, rows)
return reponse, {
"memoryRead": True,
"memoryResult": {"clients": [], "projects": [], "tasks": rows, "repositories": [], "memories": []},
"memoryQuery": question,
}
return "", {}
def executer_actions(action_data):
actions = action_data.get("actions")
if not isinstance(actions, list):
actions = [action_data]
reponses = []
flags = {
"emailSent": False,
"contactSaved": False,
"memorySaved": False,
"memoryRead": False,
"memoryResult": None,
"memoryQuery": "",
}
actions_executees = []
for action in actions:
if not isinstance(action, dict):
continue
reponse, action_flags = executer_action(action)
actions_executees.append(action)
if reponse:
reponses.append(reponse)
for cle in ("emailSent", "contactSaved", "memorySaved", "memoryRead"):
flags[cle] = flags[cle] or bool(action_flags.get(cle))
if action_flags.get("memoryResult"):
flags["memoryResult"] = action_flags["memoryResult"]
flags["memoryQuery"] = action_flags.get("memoryQuery", "")
if actions_executees and all(action.get("action") == "ajouter_client" for action in actions_executees):
noms = [action.get("nom", "").strip() for action in actions_executees if action.get("nom", "").strip()]
if len(noms) > 1:
return f"C'est note : {len(noms)} clients ajoutes ({', '.join(noms)}).", flags
return " ".join(reponses), flags
def mettre_a_jour_contexte(session_id, texte, reponse, flags):
ajouter_echange_historique(session_id, texte, reponse)
resultat = flags.get("memoryResult")
if not resultat:
return
contexte = session_contexts.setdefault(session_id, {})
if flags.get("memoryQuery"):
contexte["last_query"] = flags["memoryQuery"]
if resultat.get("clients"):
contexte["last_client"] = resultat["clients"][0].get("name", "")
elif flags.get("memoryQuery"):
contexte["last_client"] = flags["memoryQuery"]
elif resultat.get("tasks") and resultat["tasks"][0].get("client_name"):
contexte["last_client"] = resultat["tasks"][0].get("client_name", "")
if resultat.get("projects"):
contexte["last_project"] = resultat["projects"][0].get("name", "")
elif resultat.get("tasks") and resultat["tasks"][0].get("project_name"):
contexte["last_project"] = resultat["tasks"][0].get("project_name", "")
contexte["tasks"] = resultat.get("tasks", [])
def ajouter_echange_historique(session_id, texte, reponse):
historique = histories.setdefault(session_id, [])
if not historique or historique[-1].get("content") != texte:
historique.append({"role": "user", "content": texte})
historique.append({"role": "assistant", "content": reponse})
del historique[:-20]
def formater_memoire(resultat):
lignes = []
for client_mem in resultat["clients"][:5]:
detail = f"Client : {client_mem['name']}"
if client_mem.get("notes"):
detail += f" - {client_mem['notes']}"
lignes.append(detail)
for projet in resultat["projects"][:5]:
detail = f"Projet : {projet['name']}"
if projet.get("client_name"):
detail += f" ({projet['client_name']})"
if projet.get("status"):
detail += f" - statut : {projet['status']}"
if projet.get("next_action"):
detail += f" - prochaine action : {projet['next_action']}"
lignes.append(detail)
for tache in resultat["tasks"][:5]:
contexte = tache.get("client_name") or tache.get("project_name") or "general"
lignes.append(f"Tache : {tache['title']} ({contexte})")
for repo in resultat["repositories"][:5]:
chemin = repo.get("local_path") or repo.get("remote_url") or "chemin non renseigne"
lignes.append(f"Depot : {repo['name']} - {chemin}")
for note in resultat["memories"][:5]:
sujet = note.get("subject") or note.get("client_name") or note.get("project_name") or "note"
lignes.append(f"Note : {sujet} - {note['content']}")
if not lignes:
return "Je n'ai rien trouve dans la memoire professionnelle."
return "Voici ce que j'ai trouve :\n" + "\n".join(f"- {ligne}" for ligne in lignes[:10])
def formater_liste_clients(clients_mem):
if not clients_mem:
return "Je n'ai encore aucun client dans la memoire professionnelle."
noms = [client_mem["name"] for client_mem in clients_mem]
return "Voici tes clients :\n" + "\n".join(f"- {nom}" for nom in noms)
def formater_liste_taches(taches):
if not taches:
return "Tu n'as aucune tache ouverte dans la memoire professionnelle."
groupes = {}
for tache in taches:
cle = tache.get("client_name") or tache.get("project_name") or "General"
groupes.setdefault(cle, []).append(tache)
lignes = ["Voici tes taches ouvertes :"]
for sujet, items in groupes.items():
lignes.append(f"- {sujet} :")
for tache in items:
lignes.append(f" - #{tache['id']} {tache['title']}")
return "\n".join(lignes)
def resumer_resultat_sql(question, sql, rows):
if not rows:
return "Je n'ai rien trouve dans la memoire professionnelle."
if not GROQ_API_KEY:
return formater_lignes_sql(rows)
payload = json.dumps(rows[:50], ensure_ascii=False, indent=2)
messages = [
{
"role": "system",
"content": (
"Tu transformes un resultat SQL en reponse courte en francais pour Laurent. "
"Ne mentionne pas SQL. Ne rajoute rien qui n'est pas dans les lignes."
),
},
{
"role": "user",
"content": (
f"Question utilisateur: {question or 'question non fournie'}\n"
f"Requete executee: {sql}\n"
f"Resultats JSON:\n{payload}"
),
},
]
reponse = client.chat.completions.create(
model=CHAT_MODEL,
messages=messages,
temperature=0.2,
max_tokens=500,
)
return (reponse.choices[0].message.content or "").strip() or formater_lignes_sql(rows)
def formater_lignes_sql(rows):
lignes = ["Voici ce que j'ai trouve :"]
for row in rows[:20]:
valeurs = [f"{cle}: {valeur}" for cle, valeur in row.items()]
lignes.append("- " + ", ".join(valeurs))
return "\n".join(lignes)
def formater_client_plus_taches(snapshot):
clients_avec_taches = [
client_mem for client_mem in snapshot["clients"]
if int(client_mem.get("open_task_count") or 0) > 0
]
if not clients_avec_taches:
return "Tu n'as aucune tache ouverte dans la memoire professionnelle.", ""
max_count = max(int(client_mem["open_task_count"]) for client_mem in clients_avec_taches)
premiers = [
client_mem["name"] for client_mem in clients_avec_taches
if int(client_mem["open_task_count"]) == max_count
]
lignes = [
f"{client_mem['name']} : {client_mem['open_task_count']} tache(s)"
for client_mem in sorted(
clients_avec_taches,
key=lambda item: (-int(item["open_task_count"]), item["name"].lower()),
)
]
if len(premiers) == 1:
intro = f"Le client avec le plus de taches ouvertes est {premiers[0]} avec {max_count} tache(s)."
else:
intro = f"Les clients avec le plus de taches ouvertes sont {', '.join(premiers)} avec {max_count} tache(s) chacun."
return intro + "\n" + "\n".join(f"- {ligne}" for ligne in lignes), premiers[0]
def traiter_message(session_id, texte, avec_audio=False):
action_confirmee = recuperer_action_confirmee(session_id, texte)
if action_confirmee:
reponse, flags = executer_actions(action_confirmee)
mettre_a_jour_contexte(session_id, texte, reponse, flags)
audio_url = generer_audio(reponse) if avec_audio and TTS_PROVIDER == "groq" else None
return {
"reply": reponse,
"audioUrl": audio_url,
"emailSent": flags["emailSent"],
"contactSaved": flags["contactSaved"],
"memorySaved": flags["memorySaved"],
"memoryRead": flags["memoryRead"],
}
decision = router_intention(session_id, texte)
flags = {
"emailSent": False,
"contactSaved": False,
"memorySaved": False,
"memoryRead": False,
"memoryResult": None,
"memoryQuery": "",
}
if decision["type"] == "action":
reponse, flags = executer_actions(decision["action"])
mettre_a_jour_contexte(session_id, texte, reponse, flags)
elif decision["type"] == "clarification":
reponse = decision["question"] or decision["reply"] or "Tu peux préciser ce que tu veux que je fasse ?"
if decision.get("action"):
session_contexts.setdefault(session_id, {})["pending_action"] = decision["action"]
ajouter_echange_historique(session_id, texte, reponse)
else:
reponse = decision["reply"]
if not reponse:
reponse = demander_au_llm(session_id, texte)
ajouter_echange_historique(session_id, texte, reponse)
audio_url = None
if avec_audio and TTS_PROVIDER == "groq":
audio_url = generer_audio(reponse)
return {
"reply": reponse,
"audioUrl": audio_url,
"emailSent": flags["emailSent"],
"contactSaved": flags["contactSaved"],
"memorySaved": flags["memorySaved"],
"memoryRead": flags["memoryRead"],
}
def detecter_lecture_simple(texte):
texte_min = texte.strip().lower()
if re.search(r"\b(taches?|tâches?)\b", texte_min) and re.search(r"\b(quelles?|liste|affiche|montre|mes|a faire|à faire|ouvertes?)\b", texte_min):
if not re.search(r"\b(ajoute|note|notes|supprime|enleve|enlève|retire|termine|marque)\b", texte_min):
return {"action": "lister_taches"}
if re.search(r"\bclients?\b", texte_min) and re.search(r"\b(qui|quels?|liste|affiche|montre|mes|tous)\b", texte_min):
if not re.search(r"\b(ajoute|note|notes|supprime|enleve|enlève|retire|desactive|désactive|nouveaux?)\b", texte_min):
return {"action": "lister_clients"}
return None
def detecter_suivi_contexte(session_id, texte):
contexte = session_contexts.get(session_id, {})
dernier_client = contexte.get("last_client")
if not dernier_client:
return None
texte_min = texte.strip().lower()
if re.search(r"\b(lui|ce client|pour lui)\b", texte_min) and re.search(r"\b(que dois|quoi faire|a faire|à faire|tache|tâche)\b", texte_min):
return {"action": "chercher_memoire", "requete": dernier_client}
if re.search(r"\b(c'est quoi|quelle est|quelles sont|montre|detail|détail)\b", texte_min) and re.search(r"\b(tache|tâche|taches|tâches)\b", texte_min):
return {"action": "chercher_memoire", "requete": dernier_client}
return None
def recuperer_action_confirmee(session_id, texte):
contexte = session_contexts.get(session_id, {})
pending_action = contexte.get("pending_action")
if not pending_action:
return None
texte_min = texte.strip().lower()
if re.fullmatch(r"(oui|yes|ok|d'accord|vas-y|confirme|c'est ca|c'est ça|exact|tout a fait|tout à fait)[\s.!?]*", texte_min):
contexte.pop("pending_action", None)
return pending_action
if re.fullmatch(r"(non|no|annule|stop|laisse tomber|pas ca|pas ça)[\s.!?]*", texte_min):
contexte.pop("pending_action", None)
return None
return None
@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"
avec_audio = request.form.get("withAudio", "false").lower() == "true"
if not fichier:
return jsonify({"error": "Audio manquant"}), 400
try:
texte = transcrire_audio(fichier).strip()
resultat = traiter_message(session_id, texte, avec_audio=avec_audio)
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)
session_contexts.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=os.getenv("FLASK_DEBUG", "0") == "1",
)
Reference in New Issue View Git Blame Copy Permalink