"""
Memoire episodique et semantique de Mirza : RAG local, hybride, reranked RRF.
Combine SQLite (relationnel + vectoriel sqlite-vec), FastEmbed et la fusion de
rangs reciproques (Reciprocal Rank Fusion) pour un enrichissement contextuel de
moins de 4 ms d'overhead par tour.
"""
from __future__ import annotations
import re
from typing import Any
from .logger import getMirzaLogger
from .memory.embeddingEngine import EmbeddingEngine
from .memory.sqliteBackend import SQLiteMemoryBackend
from .promptTokens import MEMO_TAG_CLOSE, MEMO_TAG_OPEN, stripMemoTag
logger = getMirzaLogger("mirza.memory")
[docs]
class EpisodicMemoryStore:
"""
Gestionnaire de base de donnees vectorielle et relationnelle locale pour Mirza.
Indexe des souvenirs episodiques (tours de conversation, faits cles, resumes)
et recupere les contextes les plus pertinents par reranking hybride
(dense + BM25, fusion RRF).
Parameters
----------
dbPath : str, optional
Chemin vers le fichier de base SQLite. Defaut : "mirza/data/memory.db".
embeddingModel : str, optional
Identifiant HuggingFace du modele d'embedding. Defaut :
"sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2".
embeddingDim : int, optional
Dimension des vecteurs d'embedding dense. Defaut : 384.
similarityThreshold : float, optional
Score cosinus minimum (1 - dist) pour juger un souvenir pertinent.
Defaut : 0.35.
maxContextTokens : int, optional
Nombre maximal approximatif de tokens injectes dans le prompt.
Defaut : 600.
topKRetrieval : int, optional
Nombre de candidats extraits par source (dense et sparse). Defaut : 8.
topKRerank : int, optional
Nombre final de souvenirs retenus apres fusion RRF. Defaut : 2.
"""
[docs]
def __init__(
self,
dbPath: str = "mirza/data/memory.db",
embeddingModel: str = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2",
embeddingDim: int = 384,
similarityThreshold: float = 0.10,
maxContextTokens: int = 600,
topKRetrieval: int = 8,
topKRerank: int = 2,
) -> None:
self.dbPath = dbPath
self.backend = SQLiteMemoryBackend(dbPath=dbPath, embeddingDim=embeddingDim)
self.engine = EmbeddingEngine(modelName=embeddingModel, dim=embeddingDim)
self.similarityThreshold = similarityThreshold
self.maxContextTokens = maxContextTokens
self.topKRetrieval = topKRetrieval
self.topKRerank = topKRerank
[docs]
def warmUp(self) -> None:
"""
Prechauffe le modele d'embedding ONNX pour eliminer tout surcout au premier tour.
"""
self.engine.warmUp()
@staticmethod
def _isConversationalOrTrivial(query: str) -> bool:
"""
Detecte une salutation, un remerciement ou une phrase de test.
Parameters
----------
query : str
La requete a classifier.
Returns
-------
bool
True si aucun rappel RAG passif n'a de sens ici.
"""
clean = re.sub(r"[^\w\s]", "", query.lower()).strip()
if len(clean) < 3:
return True
trivial_patterns = [
r"^(bonjour|salut|hello|bonsoir|coucou|hey)\b",
r"^(merci|merci beaucoup|thanks|thx)\b",
r"^(au revoir|bonne nuit|a bientot|a plus|bye)\b",
r"^(comment vas[- ]tu|comment tu vas|ca va|comment ca va)\b",
r"^(ceci est un test|test micro|test audio|test de fonctionnement|un deux un deux)\b",
r"^(qui es[- ]tu|tu es qui|presente[- ]toi|que sais[- ]tu faire)\b",
]
return any(re.search(pat, clean) for pat in trivial_patterns)
[docs]
def storeTurn(
self,
userInput: str,
assistantResponse: str,
summary: str | None = None,
metadata: str | None = None,
) -> int | None:
"""
Archive un tour de conversation complet dans la memoire a long terme.
Parameters
----------
userInput : str
Texte de l'enonce soumis par l'utilisateur.
assistantResponse : str
Reponse textuelle generee par l'assistant.
summary : str | None, optional
Resume concis du tour. Defaut : None.
metadata : str | None, optional
Metadonnees contextuelles additionnelles. Defaut : None.
Returns
-------
int | None
Identifiant entier de la ligne inseree ou None en cas d'echec.
"""
if not userInput or not assistantResponse:
return None
cleanUser = stripMemoTag(userInput)
cleanAssistant = stripMemoTag(assistantResponse)
if len(cleanUser) < 2 and len(cleanAssistant) < 2:
return None
# Salutations et tests triviaux n'ont rien a archiver
if self._isConversationalOrTrivial(cleanUser):
logger.debug("Echange conversationnel trivial detecte, non archive en memoire")
return None
if self._isRefusalResponse(cleanAssistant):
logger.debug("Reponse de refus detectee, non enregistree en memoire")
return None
combinedText = f"Question : {cleanUser}\nReponse : {cleanAssistant}"
vec = self.engine.embedText(combinedText)
blob = self.engine.serializeVector(vec) if vec is not None else None
memId = self.backend.insertTurn(
userInput=cleanUser,
assistantResponse=cleanAssistant,
summary=summary,
metadata=metadata,
embeddingBytes=blob,
)
if memId is not None:
logger.debug("Tour de conversation archive (ID: %d)", memId)
return memId
[docs]
def retrieveRelevantContext(
self,
query: str,
topK: int | None = None,
) -> list[dict[str, Any]]:
"""
Recherche semantique hybride (Dense cosine distance sqlite-vec + Lexical BM25 FTS5).
Parameters
----------
query : str
Requete ou invite textuelle pour la recherche semantique.
topK : int | None, optional
Nombre de candidats extraits par canal. Defaut : None, ce qui
utilise ``topKRetrieval``.
Returns
-------
list[dict[str, Any]]
Liste des souvenirs ordonnes par score RRF decroissant.
"""
if not query or len(query.strip()) < 3:
return []
cleanQuery = query.strip()
# Requetes de pure conversation ou de test
if self._isConversationalOrTrivial(cleanQuery):
return []
kRet = topK or self.topKRetrieval
kRank = self.topKRerank
denseMatches: list[tuple[int, float]] = []
vec = self.engine.embedText(cleanQuery)
if vec is not None:
rawBlob = self.engine.serializeVector(vec)
rawDense = self.backend.searchVec(rawBlob, limit=kRet)
for memId, dist in rawDense:
# sqlite-vec avec distance_metric=cosine : distance = 1 - cos(u, v)
cosSim = max(-1.0, min(1.0, 1.0 - dist))
if cosSim >= self.similarityThreshold:
denseMatches.append((memId, cosSim))
sparseMatches = self.backend.searchFts(cleanQuery, limit=kRet)
# Aucun canal n'a produit de candidat : ne rien injecter
if not denseMatches and not sparseMatches:
return []
fused = self.engine.rrfFusion(denseMatches, sparseMatches)
if not fused:
return []
denseDocIds = {memId for memId, _ in denseMatches}
# Pertinence : appui dense ou concordance lexicale FTS5
validFused: list[tuple[int, float]] = []
for memId, score in fused:
if memId in denseDocIds or (sparseMatches and memId in sparseMatches):
validFused.append((memId, score))
if not validFused:
return []
topIds = [item[0] for item in validFused[:kRank * 2]]
docs = self.backend.getMemoriesByIds(topIds)
# Priorité absolue aux dossiers médicaux authentifiés ('dossier_medical' ou 'consultation_directe')
clinical_items: list[tuple[int, float]] = []
other_items: list[tuple[int, float]] = []
for memId, score in validFused[:kRank * 2]:
d = docs.get(memId)
if not d:
continue
meta = d.get("metadata") or ""
if "dossier_medical" in meta or "consultation_directe" in meta or "patient_id" in meta:
clinical_items.append((memId, score))
else:
other_items.append((memId, score))
ordered_candidates = (clinical_items + other_items)[:kRank]
results: list[dict[str, Any]] = []
for memId, score in ordered_candidates:
if memId in docs:
entry = dict(docs[memId])
entry["rrfScore"] = score
results.append(entry)
return results
[docs]
def reindexAll(self) -> int:
"""
Recalcule tous les vecteurs pour realigner la memoire sur le modele
d'embedding courant.
Returns
-------
int
Nombre de souvenirs reindexes.
"""
return self.backend.reindexAll(self.engine)
@staticmethod
def _isRefusalResponse(resp: str) -> bool:
"""
Detecte si la reponse de l'assistant correspond a un refus ou une absence de donnees.
Parameters
----------
resp : str
Texte de la reponse generee par l'assistant.
Returns
-------
bool
True si la reponse indique un refus de stocker ou une absence de memoire.
"""
respLower = resp.lower()
refusalTriggers = [
"ne peux pas accéder",
"ne peux pas stocker",
"ne peux pas connaître",
"pas accès à vos informations",
"strictement confidentiel",
"je n'ai pas accès",
"je n'ai pas de mémoire",
"en tant qu'intelligence artificielle",
"en tant qu'ia",
"don't have access",
"do not have access",
"i cannot access",
"as an ai",
"as a language model",
"i don't remember",
"i have no memory",
]
return any(trig in respLower for trig in refusalTriggers)
def _detectAndComputeAggregates(self, query: str) -> str | None:
"""
Calcule des statistiques exactes et exhaustives sur la base SQLite pour
les questions de denombrement, de seuil ou de recherche dans la patientele.
"""
qLower = query.lower()
countKeywords = [
"combien", "nombre de", "combien de", "combien y a-t-il",
"plus de", "ordonnance", "ordonnances", "qui a plus de",
"sur combien", "j'ai de patients", "jai de patients",
"est-ce qu un patient", "est-ce qu'un patient", "liste tous", "tous les patients",
"combien de cas", "combien de patient",
"âge moyen", "age moyen", "moyenne d'âge", "moyenne d age", "moyenne d'age",
"variance d'âge", "variance d'age", "variance d age", "écart-type", "ecart-type",
"médiane d'âge", "mediane d'age", "répartition par sexe", "hommes et femmes",
"pyramide des âges", "démographie"
]
if not any(kw in qLower for kw in countKeywords):
return None
# Dictionnaire des concepts cliniques majeurs avec regles d'inclusion et exclusion
CLINICAL_CONCEPTS: dict[str, dict[str, list[str]]] = {
"diabète": {
"keywords": ["diabète", "diabete", "diabétique", "diabetique", "glycémie", "glycemie"],
"triggers": ["diabète de type 2", "diabete de type 2", "diabétique", "diabetique", "metformine"],
"negations": ["sans diabète", "pas de diabète"],
},
"apnée": {
"keywords": ["apnée", "apnee", "ppc", "machine à pression", "machine a pression"],
"triggers": ["apnée du sommeil", "apnee du sommeil", "ppc", "machine à pression", "pression positive continue", "iah"],
"negations": [],
},
"grippe": {
"keywords": ["grippe"],
"triggers": ["grippe", "vaccination antigrippale", "vaccin grippe"],
"negations": [],
},
"hypertension": {
"keywords": ["hypertension", "hta", "tension"],
"triggers": ["hta", "hypertension", "amlodipine", "ramipril", "irbesartan", "perindopril", "périndopril", "bisoprolol"],
"negations": ["pas d'hta", "normotendu"],
},
}
try:
with self.backend.connection() as conn:
cur = conn.cursor()
# Cas 0 : Analyse démographique globale (âge, moyenne, variance, sexe) sur l'ensemble des 40 patients
demoKeywords = [
"âge moyen", "age moyen", "moyenne d'âge", "moyenne d age", "moyenne d'age",
"variance d'âge", "variance d'age", "variance d age", "variance", "écart-type", "ecart-type",
"médiane", "mediane", "démographie", "demographie"
]
if any(dk in qLower for dk in demoKeywords):
sql_demo = """
SELECT
json_extract(metadata, '$.patient_id') AS pid,
MAX(json_extract(metadata, '$.patient_nom')) AS pnom,
MAX(json_extract(metadata, '$.age')) AS age,
MAX(json_extract(metadata, '$.sexe')) AS sexe
FROM memories
WHERE metadata IS NOT NULL
GROUP BY json_extract(metadata, '$.patient_id')
ORDER BY pnom ASC;
"""
cur.execute(sql_demo)
demo_rows = cur.fetchall()
if demo_rows:
import math
ages = [r[2] for r in demo_rows if r[2] is not None]
n = len(ages)
if n > 0:
mean_a = sum(ages) / n
var_a = sum((x - mean_a) ** 2 for x in ages) / (n - 1) if n > 1 else 0.0
std_a = math.sqrt(var_a)
sorted_a = sorted(ages)
med_a = (sorted_a[n // 2 - 1] + sorted_a[n // 2]) / 2 if n % 2 == 0 else sorted_a[n // 2]
nb_f = sum(1 for r in demo_rows if r[3] == "F")
nb_m = sum(1 for r in demo_rows if r[3] == "M")
lines_table = ["| Patient | ID | Âge | Sexe |", "|---|---|---|---|"]
for r in demo_rows:
lines_table.append(f"| {r[1]} | {r[0]} | {r[2]} ans | {r[3]} |")
table_str = "\n".join(lines_table)
return (
f"[ANALYSE DÉMOGRAPHIQUE EXHAUSTIVE DE LA PATIENTÈLE (100% DES {len(demo_rows)} PATIENTS DU CABINET) :\n"
f"- Effectif total de la patientèle : {len(demo_rows)} patients\n"
f"- Répartition par sexe : {nb_f} femmes, {nb_m} hommes\n"
f"- Âge moyen exact : {mean_a:.1f} ans\n"
f"- Médiane d'âge : {med_a:.1f} ans\n"
f"- Variance d'âge exacte : {var_a:.1f} ans²\n"
f"- Écart-type : {std_a:.1f} ans (étendue de {min(ages)} à {max(ages)} ans)\n\n"
f"Tableau complet des 40 patients à inclure impérativement dans <result> :\n{table_str}\n\n"
f"Consigne stricte pour la réponse : Utilise impérativement ces statistiques certifiées portant sur l'intégralité des {len(demo_rows)} patients (Âge moyen: {mean_a:.1f} ans, Variance: {var_a:.1f} ans²). N'affirme JAMAIS que tu n'as accès qu'à 3 ou 4 patients : la patientèle complète compte exactement {len(demo_rows)} patients.]"
)
# Cas 1 : Question avec seuil (ex: 'plus de 3 ordonnances de metformine')
minMatch = re.search(r"plus de (\d+)", qLower)
if minMatch:
threshold = int(minMatch.group(1))
meds = ["metformine", "ramipril", "atorvastatine", "kardegic", "amoxicilline", "paracetamol", "doliprane", "amlodipine"]
targetMed = next((m for m in meds if m in qLower), None)
if targetMed:
sql = """
SELECT
MAX(json_extract(metadata, '$.patient_nom')) AS patient,
json_extract(metadata, '$.patient_id') AS pid,
COUNT(*) AS occurrences
FROM memories
WHERE metadata IS NOT NULL AND assistant_response LIKE ?
GROUP BY json_extract(metadata, '$.patient_id')
HAVING occurrences > ?
ORDER BY occurrences DESC;
"""
cur.execute(sql, (f"%{targetMed}%", threshold))
rows = cur.fetchall()
if not rows:
return f"[Données statistiques certifiées : Aucun patient n'a plus de {threshold} ordonnance(s) de {targetMed}.]"
lines = [f"[Données statistiques certifiées : {len(rows)} patient(s) avec plus de {threshold} ordonnances de {targetMed} :]"]
for pat, pid, count in rows:
lines.append(f"- {pat} ({pid}) : {count} ordonnances")
return "\n".join(lines)
# Cas 2 : Question de cohorte clinique globale (ex: 'combien de patients diabétiques')
matchedConcept = None
for cName, cData in CLINICAL_CONCEPTS.items():
if any(kw in qLower for kw in cData["keywords"]):
matchedConcept = cData
break
if matchedConcept:
cur.execute("""
SELECT
json_extract(metadata, '$.patient_id') AS pid,
MAX(json_extract(metadata, '$.patient_nom')) AS pnom,
GROUP_CONCAT(assistant_response, ' ') AS full_text
FROM memories
WHERE metadata IS NOT NULL
GROUP BY json_extract(metadata, '$.patient_id');
""")
patients = cur.fetchall()
totalPatients = len(patients)
positives = []
for pid, pnom, text in patients:
tLow = text.lower()
hasTrigger = any(trig in tLow for trig in matchedConcept["triggers"])
hasNegOnly = any(neg in tLow for neg in matchedConcept["negations"]) and not any(trig in tLow for trig in matchedConcept["triggers"][:3])
if hasTrigger and not hasNegOnly:
positives.append((pnom, pid))
pNames = ", ".join(f"{pnom} ({pid})" for pnom, pid in positives)
return (
f"[RÉSULTAT STATISTIQUE EXHAUSTIF BASE PATIENTÈLE (100% DES {totalPatients} DOSSIERS ANALYSÉS) :\n"
f"- Nombre total de patients dans le cabinet : {totalPatients}\n"
f"- Nombre exact de patients concernés : {len(positives)} sur {totalPatients}\n"
f"- Liste complète des patients concernés : {pNames}\n"
f"Consigne stricte pour la réponse : Utilise impérativement ce décompte exact ({len(positives)} patients sur {totalPatients}). N'affirme JAMAIS que tu n'as accès qu'à 4 dossiers : l'analyse a scanné l'intégralité des {totalPatients} dossiers de la patientèle.]"
)
except Exception as exc:
logger.debug("Erreur calcul agrégats statistiques : %s", exc)
return None
[docs]
def formatMemoryContext(self, query: str) -> str:
"""
Recupere les contextes pertinents et les formate pour injection dans le prompt.
Parameters
----------
query : str
Entree utilisateur courante.
Returns
-------
str
Bloc formate sous [Memo context : ...] ou chaine vide si aucun souvenir pertinent.
"""
recalled = self.retrieveRelevantContext(query)
exactStats = self._detectAndComputeAggregates(query)
if not recalled and not exactStats:
return ""
lines = [MEMO_TAG_OPEN]
if exactStats:
lines.append(exactStats)
currentTokens = 0
for item in recalled:
# Refus archives par accident
memId = item.get("id")
tag = f"[Mem #{memId}] " if memId is not None else ""
entry = f"- {tag}User : {item['userInput']} | Mirza : {item['assistantResponse']}"
approxTokens = len(entry) // 4
if currentTokens + approxTokens > self.maxContextTokens:
if currentTokens == 0:
maxChars = self.maxContextTokens * 4
lines.append(entry[:maxChars].rstrip() + "...")
break
lines.append(entry)
currentTokens += approxTokens
if len(lines) <= 1:
return ""
lines.append(MEMO_TAG_CLOSE)
return "\n".join(lines)
[docs]
def count(self) -> int:
"""
Retourne le nombre total de souvenirs archives.
Returns
-------
int
Nombre d'enregistrements en base.
"""
return self.backend.count()
[docs]
def listMemories(
self,
query: str | None = None,
limit: int = 50,
offset: int = 0,
) -> list[dict[str, Any]]:
"""
Liste les souvenirs stockes avec pagination.
Parameters
----------
query : str | None, optional
Filtre de recherche plein texte. Defaut : None.
limit : int, optional
Nombre d'elements par page. Defaut : 50.
offset : int, optional
Indice de depart. Defaut : 0.
Returns
-------
list[dict[str, Any]]
Liste des souvenirs trouves.
"""
return self.backend.listMemories(query=query, limit=limit, offset=offset)
[docs]
def deleteMemory(self, memoryId: int) -> bool:
"""
Supprime un souvenir specifique par son identifiant.
Parameters
----------
memoryId : int
Identifiant du souvenir a supprimer.
Returns
-------
bool
True si la suppression a reussi, False sinon.
"""
return self.backend.deleteMemory(memoryId)
# Alias de retrocompatibilite
EpisodicMemory = EpisodicMemoryStore