"""
Orchestrateur central du cycle conversationnel multimodal de Mirza.
Coordonne les evenements microphone, le VAD neuronal, la transcription ASR,
le dialogue LLM, le filtrage temps reel bi-canal, la memoire RAG hybride,
et la restitution vocale TTS non bloquante.
"""
from __future__ import annotations
import datetime
import json
import queue
import random
import re
import string
import threading
import time
from collections.abc import Callable
from dataclasses import dataclass
from typing import Any
import numpy as np
from ..audio.audioProcessor import AudioProcessor
from ..audio.deviceManager import DeviceManager
from ..audio.vadDetector import SileroVadDetector
from ..config import SAFE_CONTEXT_TOKENS, AppConfig
from ..core.episodicMemory import EpisodicMemoryStore
from ..core.logger import getMirzaLogger
from ..core.memoryManager import MemoryManager
from ..core.modelProfiles import (
DEFAULT_PROFILE_ID,
ModelProfile,
getProfile,
listProfiles,
profileFromRepo,
)
from ..core.modelSampling import recommendedSampling
from ..core.prompts import THINKING_FILLER_PHRASES
from ..core.promptTokens import (
ARCHIVE_ANNOUNCEMENT,
stripArchivePhrases,
stripMemoTag,
)
from ..core.textCleaner import FinalAnswerStreamParser, StreamingTextFilter, TextCleaner
from ..engine.agent.agentManager import AgentManager
from ..engine.factory import EngineFactory
from ..engine.interfaces.baseLlm import BaseLlm
from ..engine.interfaces.baseStt import BaseStt
from ..engine.interfaces.baseTts import BaseTts
from .agentOutput import (
AGENT_ERROR_PREFIXES,
needsArchiveAnnouncement,
splitAgentOutput,
splitSentences,
splitSpokenForSynthesis,
spokenFallback,
)
from .audioStreamPipeline import AudioStreamPipeline
from .clauseSplitter import ClauseSplitter
from .contextManager import ConversationContextManager
logger = getMirzaLogger("mirza.pipeline.orchestrator")
RE_DOC_TAGS = re.compile(r"</?(?:result|code|code_view)[^>]*>")
# L'agent renvoie son document dans un bloc de code : sans ce nettoyage
# l'enveloppe partait en memoire et l'archive etait reinjectee telle quelle.
RE_CODE_FENCE = re.compile(r"^\s*```[a-zA-Z0-9_+-]*\s*\n|\n?\s*```\s*$")
RE_FINAL_ANSWER_CALL = re.compile(
r"\bfinal_answer\s*\(\s*(?:f?\"\"\"|'''|f?\")?", re.MULTILINE)
def _cleanArchivedDocument(document: str) -> str:
"""
Ne garder que le contenu du document, sans l'enveloppe de l'agent.
La parenthese fermante n'est retiree que si l'appel ``final_answer(...)`` a
reellement ete trouve : une parenthese de fin de ligne peut legitimement
appartenir au texte, comme dans un lien Markdown ``[W](x)``.
Parameters
----------
document : str
Document tel que produit par l'agent.
Returns
-------
str
Le texte du document, sans ses balises, son bloc de code ni l'appel
``final_answer(...)`` qui l'encadrait.
"""
clean = RE_DOC_TAGS.sub("", document or "")
clean = RE_CODE_FENCE.sub("", clean)
withoutCall = RE_FINAL_ANSWER_CALL.sub("", clean, count=1)
if withoutCall == clean:
return clean.strip()
tail = withoutCall.rstrip()
if tail.endswith(")"):
tail = tail[:-1].rstrip()
return tail.rstrip("\"'").rstrip()
#: Correspondance entre les cles de generation_config.json et les champs de
#: configuration Mirza. Seules ces cles sont reprises du depot.
_SAMPLING_FIELDS = {
"temperature": "temperature",
"top_p": "topP",
"top_k": "topK",
"min_p": "minP",
"presence_penalty": "presencePenalty",
"repetition_penalty": "repetitionPenalty",
}
[docs]
@dataclass(frozen=True)
class TurnModes:
"""
The execution modes that apply to a single conversational turn.
Agent mode and reflection mode are mutually exclusive by construction.
"""
agentActive: bool
"""Whether the smolagents CodeAgent handles the turn."""
thinking: bool
"""Whether the engine streams a chain of thought before answering."""
reasoningBudget: int
"""Thought-token budget for this turn, always 0 outside reflection mode."""
[docs]
def showsThinking(self) -> bool:
"""Whether the HUD should display the THINKING state for this turn."""
return self.agentActive or self.reasoningBudget > 0
@property
def streamsThoughtsUpFront(self) -> bool:
"""Whether the stream starts inside a thought block, with no opening tag."""
return self.thinking and self.reasoningBudget > 0 and not self.agentActive
[docs]
@classmethod
def resolve(cls, agentActive: bool, thinkingEnabled: bool, budget: int) -> TurnModes:
"""
Build the modes for a turn from the raw engine flags.
Parameters
----------
agentActive : bool
Whether agent mode is on.
thinkingEnabled : bool
The engine's ``enableThinking`` flag.
budget : int
The engine's ``reasoningBudget``, in tokens.
Returns
-------
TurnModes
The resolved modes. Agent mode forces both other fields to 0.
"""
if agentActive:
return cls(agentActive=True, thinking=False, reasoningBudget=0)
if not thinkingEnabled:
return cls(agentActive=False, thinking=False, reasoningBudget=0)
return cls(agentActive=False, thinking=True, reasoningBudget=max(0, budget))
[docs]
@dataclass
class TurnOutcome:
"""
What the agent branch produced during a turn.
Mutable by design: it is filled in progressively as the agent runs and the
answer is arbitrated. An explicit holder keeps the surrounding code free of
``'name' in locals()`` probing.
"""
agentText: str = ""
"""The raw text returned by the agent, empty in direct (non-agent) mode."""
document: str | None = None
"""The rich document diverted to the Docs panel, if any."""
spoken: str = ""
"""The spoken synthesis, empty when the caller must fall back."""
ragContext: str | None = None
"""The RAG memory context injected during this turn, if any."""
[docs]
class ConversationOrchestrator:
"""
Orchestrateur backend principal du cycle conversationnel temps reel.
Gere l'ecoute vocale (VAD), la saisie textuelle directe, le streaming LLM et
le TTS.
"""
[docs]
def __init__(
self,
config: AppConfig | None = None,
sttEngine: BaseStt | None = None,
llmEngine: BaseLlm | None = None,
ttsEngine: BaseTts | None = None,
textOnly: bool = False,
):
"""
Initialise l'orchestrateur de conversation et ses sous-systemes.
Parameters
----------
config : AppConfig, optional
Configuration globale, defaut: AppConfig().
sttEngine : BaseStt, optional
Moteur ASR injecte, sinon instancie via EngineFactory.
llmEngine : BaseLlm, optional
Moteur de langage injecte, sinon instancie via EngineFactory.
ttsEngine : BaseTts, optional
Moteur TTS injecte, sinon instancie via EngineFactory.
textOnly : bool, optional
Desactive STT et TTS pour un fonctionnement allonge en memoire,
defaut: False.
"""
self.config = config or AppConfig()
self.textOnly = textOnly
self.isRunning = False
self.isInterrupted = False
self.currentState: str = "IDLE"
self.generationLock = threading.Lock()
self._engineBusy = False
# Seul un choix explicite de l'utilisateur declenche une reflexion : un
# modele qui raisonne de lui-meme ne decide pas a sa place.
self._thinkingOverride: bool = False
#: Budget de reflexion retenu au dernier changement de modele, pour que
#: le niveau choisi survive a un changement de profil.
self._reasoningBudget: int = 0
promptCfg = getattr(self.config, "prompt", None)
maxTurns = getattr(promptCfg, "historyMaxTurns", 8) if promptCfg else 8
self.contextManager = ConversationContextManager(maxHistoryTurns=maxTurns)
self.conversationHistory = self.contextManager.history
MemoryManager.initializeEnvironment(
vramLimitBytes=self.config.vramLimitBytes,
cacheLimit=self.config.cacheLimit,
mtplxPort=int(getattr(self.config.llm, "mtplxPort", 0) or 0),
purgeThresholdGb=self.config.vramLimitBytes / (1024**3) * 0.8,
)
self.deviceManager = DeviceManager()
self.micId, self.micName, self.speakerId, self.speakerName = self.deviceManager.detectDefaultDevices()
self.vadDetector = SileroVadDetector(
sampleRate=self.config.audio.sampleRate,
speechThreshold=self.config.audio.speechThreshold,
)
self.sttEngine: BaseStt = sttEngine or EngineFactory.createSttEngine(
provider=self.config.stt.provider,
modelRepo=self.config.stt.sttModelRepo,
)
# Profil actif par defaut
self.activeProfileId: str = DEFAULT_PROFILE_ID
self._cachedLocalEngine: BaseLlm | None = None
self._cachedLocalProfileId: str | None = None
self._cachedRemoteEngine: BaseLlm | None = None
self.agentManager: AgentManager | None = None
if llmEngine is None:
self._writeProfileConfig(getProfile(self.activeProfileId))
self.llmEngine: BaseLlm = llmEngine or EngineFactory.createLlmEngine(
modelConfig=self.config.llm,
provider=self.config.llm.provider,
)
self.ttsEngine: BaseTts = ttsEngine or EngineFactory.createTtsEngine(
ttsConfig=self.config.tts,
provider=self.config.tts.provider,
)
self.audioPipeline = AudioStreamPipeline(
ttsEngine=self.ttsEngine,
speakerId=self.speakerId,
targetPeak=self.config.audio.targetPeak,
textOnly=self.textOnly,
onSpeechAudioGenerated=lambda b, cid: self.onSpeechAudioGenerated(b, cid) if self.onSpeechAudioGenerated else None,
onStateChange=self._notifyState,
)
# Callbacks d'integration frontend (widget desktop, WebSocket, GUI)
self.onStateChange: Callable[[str], None] | None = None
self.onAudioLevel: Callable[[float, float], None] | None = None # (rms, speechProb)
self.onTranscription: Callable[[str], None] | None = None
self.onTokenStream: Callable[[str], None] | None = None
self.onThoughtStream: Callable[[str], None] | None = None
self.onAssistantResponseComplete: Callable[[str], None] | None = None
self.onMetrics: Callable[[dict], None] | None = None
self.onSpeechAudioGenerated: Callable[[bytes, str | None], None] | None = None
self.onAgentDoc: Callable[[str], None] | None = None
self.onCodeGenerating: Callable[[bool], None] | None = None
self.onMemoryUpdated: Callable[[dict], None] | None = None
self.onActiveContextUpdated: Callable[[dict], None] | None = None
self.onLog: Callable[[str, str, str], None] | None = None
self.activeAssistantProfileId: str = getattr(
self.config, "activeAssistantProfile", "assistant"
)
currentAssistantProfile = (
self.config.getAssistantProfile(self.activeAssistantProfileId)
if hasattr(self.config, "getAssistantProfile")
else None
)
self.thinkingFillerPhrases: tuple[str, ...] = (
currentAssistantProfile.fillerPhrases
if currentAssistantProfile and currentAssistantProfile.fillerPhrases
else THINKING_FILLER_PHRASES
)
self.llmEngine.setReasoningBudget(self.config.llm.reasoningBudget)
self.agentManager = AgentManager(llmEngine=self.llmEngine)
if currentAssistantProfile and hasattr(self.agentManager, "setAssistantProfilePrompts"):
self.agentManager.setAssistantProfilePrompts(
systemPrompt=currentAssistantProfile.agent.systemPrompt,
instructionsPrompt=currentAssistantProfile.agent.instructionsPrompt,
finalAnswerPrompt={
"pre_messages": currentAssistantProfile.agent.finalAnswerPre,
"post_messages": currentAssistantProfile.agent.finalAnswerPost,
},
allowedTools=currentAssistantProfile.allowedTools,
)
else:
# Application des restrictions d'outils du profil de modèle actif
self.agentManager.setAllowedTools(getProfile(self.activeProfileId).allowedTools)
self.memoryStore: EpisodicMemoryStore | None = None
if getattr(self.config, "memory", None) and self.config.memory.enabled:
self.memoryStore = EpisodicMemoryStore(
dbPath=self.config.memory.dbPath,
embeddingModel=self.config.memory.embeddingModel,
embeddingDim=self.config.memory.embeddingDim,
similarityThreshold=0.10,
maxContextTokens=self.config.memory.maxContextTokens,
topKRetrieval=self.config.memory.topKRetrieval,
topKRerank=self.config.memory.topKRerank,
)
from ..engine.agent.tools import setActiveMemoryStore
setActiveMemoryStore(self.memoryStore)
activeProfile = None
try:
activeProfile = getProfile(self.activeProfileId)
except KeyError:
pass
profileSupportsMemory = activeProfile.useMemory if activeProfile else True
self.memoryRetrievalEnabled: bool = (
getattr(self.config.memory, "enabled", True) and profileSupportsMemory
)
self.lastInjectedRagContext: str | None = None
self.sessionRagHistory: list[dict[str, Any]] = []
[docs]
def setMemoryRetrieval(self, enabled: bool) -> dict:
"""
Active ou desactive le retrieval semantique RAG passif a chaque tour.
Parameters
----------
enabled : bool
True pour activer l'injection semantique automatique, False pour desactiver.
Returns
-------
dict
Le statut de la memoire conversationnelle.
"""
activeProfile = None
profileId = self.getActiveModelProfileId() or self.activeProfileId
if profileId:
try:
activeProfile = getProfile(profileId)
except KeyError:
pass
if enabled and activeProfile and not activeProfile.useMemory:
logger.warning(
"Activation de la memoire refusee : le profil actif '%s' interdit l'usage memoire (confidentialite).",
profileId,
)
self.memoryRetrievalEnabled = False
else:
self.memoryRetrievalEnabled = bool(enabled)
return self.getMemoryStatus()
[docs]
def getMemoryStatus(self) -> dict:
"""
Renvoie le statut de la memoire conversationnelle RAG.
Returns
-------
dict
L'etat d'activation, le nombre d'enregistrements et l'auto-archivage.
"""
count = self.memoryStore.count() if self.memoryStore else 0
activeProfile = None
profileId = self.getActiveModelProfileId() or self.activeProfileId
if profileId:
try:
activeProfile = getProfile(profileId)
except KeyError:
pass
profileSupportsMemory = activeProfile.useMemory if activeProfile else True
effectiveEnabled = bool(
getattr(self, "memoryRetrievalEnabled", True) and profileSupportsMemory
)
return {
"enabled": effectiveEnabled,
"count": count,
"autoArchive": bool(getattr(self.config.memory, "autoArchive", True)),
"supported": profileSupportsMemory,
}
def _sanitizeHistory(self) -> None:
"""Nettoie l'historique actif pour purger les balises result et memo."""
self.contextManager.sanitizeInPlace()
[docs]
def setAgentMode(self, enabled: bool) -> dict:
"""
Active ou desactive le mode agentique CodeAgent.
Desactive aussi le mode reflexion (BrainBoost) des que l'agent est active.
Parameters
----------
enabled : bool
True pour activer l'agent smolagents, False pour le mode direct.
Returns
-------
dict
Le statut actualise du gestionnaire d'agent.
"""
self.agentManager.setEnabled(enabled)
if enabled:
self.setThinkingMode(False)
self.setReasoningBudget(0)
self._sanitizeHistory()
return self.agentManager.getStatus()
[docs]
def getAgentStatus(self) -> dict:
"""
Renvoie l'etat courant de l'agent et de ses outils.
Returns
-------
dict
Statut detaille de l'agent et des outils.
"""
return self.agentManager.getStatus()
[docs]
def setAudioDevices(self, micId: int | None = None, speakerId: int | None = None) -> dict:
"""
Modifie dynamiquement les peripheriques audio d'entree et de sortie.
Parameters
----------
micId : int, optional
Identifiant du microphone souhaite.
speakerId : int, optional
Identifiant du haut-parleur souhaite.
Returns
-------
dict
Peripheriques audio actuellement configures.
"""
if micId is not None:
self.deviceManager.setMicDevice(micId)
self.micId = self.deviceManager.micId
self.micName = self.deviceManager.micName
if speakerId is not None:
self.deviceManager.setSpeakerDevice(speakerId)
self.speakerId = self.deviceManager.speakerId
self.speakerName = self.deviceManager.speakerName
self.audioPipeline.speakerId = self.speakerId
return {
"micId": self.micId,
"micName": self.micName,
"speakerId": self.speakerId,
"speakerName": self.speakerName,
}
[docs]
def getAudioDevices(self) -> dict:
"""
Recupere la liste des peripheriques audio disponibles et les peripheriques actifs.
Returns
-------
dict
Peripheriques audio disponibles et actifs.
"""
devices = self.deviceManager.getAvailableDevices()
devices["activeMicId"] = self.micId
devices["activeMicName"] = self.micName
devices["activeSpeakerId"] = self.speakerId
devices["activeSpeakerName"] = self.speakerName
return devices
[docs]
def loadModels(self) -> None:
"""
Precharge les modeles en VRAM et effectue un warm-up initial.
"""
self._notifyState("CHARGEMENT_MODELES")
print(f"[LLM ({self.config.llm.provider.upper()})] Chargement du modele...")
self.llmEngine.loadModel(systemPrompt=self.config.prompt.systemPrompt)
if not self.textOnly:
print(f"[STT ({self.config.stt.provider.upper()})] Prechargement et calibration...")
self.sttEngine.warmUp()
self.ttsEngine.warmUp()
print(f"[TTS ({self.config.tts.provider.upper()})] Moteur pret ({self.config.tts.voice}, {self.config.tts.voiceSpeedRate})...")
print(f"[Audio] Micro : {self.micName} (ID: {self.micId}) | Sortie : {self.speakerName} (ID: {self.speakerId})")
else:
print("[Mode Texte Pur] STT et TTS non initialisés (économie RAM: ~1.1 Go).")
if self.memoryStore:
self.memoryStore.warmUp()
print("[Systeme] Tous les modules backend sont operationnels.\n")
if self.config.llm.provider == "mlx" or (self.config.stt.provider == "mlx" and not self.textOnly):
MemoryManager.displayMemoryProfile("Apres chargement initial")
self._notifyState("IDLE")
def _notifyState(self, state: str) -> None:
"""
Notifie les ecouteurs du changement d'etat de l'orchestrateur.
Parameters
----------
state : str
Nouvel etat (ex: 'IDLE', 'LISTENING', 'PROCESSING', 'SPEAKING', 'SHUTDOWN').
"""
self.currentState = state
if self.onStateChange:
self.onStateChange(state)
[docs]
def interrupt(self) -> None:
"""Interrompt immediatement la generation ou la lecture audio (barge-in)."""
self.isInterrupted = True
self.audioPipeline.interrupt()
[docs]
def emergencyStop(self) -> dict:
"""
Arret d'urgence : tue le processus du modele.
L'interruption ordinaire arrete la consommation du flux, mais la
generation continue cote serveur et la memoire reste occupee. Ici le
processus est termine, ce qui coupe la generation la ou elle se fait et
rend la memoire. Le moteur est marque decharge : la prochaine question
le recharge normalement.
Returns
-------
dict
L'etat obtenu, avec le nom du moteur arrete.
"""
self.isInterrupted = True
try:
self.audioPipeline.interrupt()
except Exception as exc:
logger.debug("Arret audio failed: %s", exc)
if getattr(self, "_cachedLocalEngine", None) is not None:
cached = self._cachedLocalEngine
supervisor = getattr(cached, "supervisor", None)
if supervisor is not None:
try:
supervisor.stop()
except Exception as exc:
logger.warning("Arret d'urgence moteur local en cache : %s", exc)
try:
cached.unload()
except Exception as exc:
logger.debug("Dechargement du moteur en cache interrompu: %s", exc)
self._cachedLocalEngine = None
self._cachedLocalProfileId = None
engine = self.llmEngine
supervisor = getattr(engine, "supervisor", None)
stopped = "aucun (moteur in-process)"
if supervisor is not None:
stopped = getattr(supervisor, "label", type(engine).__name__)
try:
supervisor.stop()
except Exception as exc:
logger.warning("Arret d'urgence : moteur non arrete : %s", exc)
stopped = f"{stopped} (echec : {exc})"
try:
engine.unload()
except Exception as exc:
logger.debug("Dechargement du moteur interrompu par l'arret d'urgence: %s", exc)
self.isLoaded = False
logger.warning("ARRET D'URGENCE : moteur %s arrete", stopped)
return {"status": "emergency_stop", "engine": stopped}
def _ttsWorker(self, textQueue: queue.Queue, audioQueue: queue.Queue) -> None:
"""Worker de synthese vocale delegue a audioPipeline."""
self.audioPipeline.ttsWorker(textQueue, audioQueue)
def _playbackWorker(
self,
audioQueue: queue.Queue,
onFirstAudioPlayed: Callable[[], None] | None = None,
isClientOrigin: bool = False,
clientTargetId: str | None = None,
) -> None:
"""Worker de lecture sonore et de diffusion reseau delegue a audioPipeline."""
self.audioPipeline.playbackWorker(
audioQueue=audioQueue,
onFirstAudioPlayed=onFirstAudioPlayed,
isClientOrigin=isClientOrigin,
clientTargetId=clientTargetId,
)
[docs]
def listModelProfiles(self) -> list[dict]:
"""
Renvoie les profils de modele selectionnables.
Returns
-------
list[dict]
Un profil par entree du catalogue, marque actif pour celui en cours.
"""
active = self.getActiveModelProfileId()
return [
{**entry, "active": entry["id"] == active}
for entry in listProfiles()
if entry.get("selectable", True)
]
[docs]
def getActiveModelProfileId(self) -> str | None:
"""
Identifie le profil correspondant au modele actuellement charge.
Returns
-------
str | None
L'identifiant du profil, ou None si le modele charge n'est pas
du catalogue.
"""
if getattr(self.llmEngine, "isRemote", False) is True:
return "openrouter_nemotron"
engineRepo = getattr(self.llmEngine, "modelRepo", None)
if engineRepo:
return profileFromRepo(engineRepo)
fallback = getattr(self.config.llm, "llmModelRepo", "") or getattr(self.config.llm, "mtplxModelRepo", "")
return profileFromRepo(fallback)
[docs]
def switchModel(self, profileId: str) -> dict:
"""
Bascule vers un autre profil de modele.
Pour les modeles locaux, la memoire est liberee avant de charger le nouveau
modele afin de ne jamais conserver deux modeles lourds en VRAM.
Pour le modele distant (OpenRouter), le modele local reste conserve en
RAM pour permettre un basculement instantane sans rechargement.
Parameters
----------
profileId : str
Identifiant du profil cible.
Returns
-------
dict
L'etat des moteurs apres bascule, avec ``"error"`` en cas d'echec.
La bascule est refusee si une generation est en cours.
"""
try:
profile = getProfile(profileId)
except KeyError as exc:
return {**self.getEngineStatus(), "error": str(exc)}
if not profile.selectable:
return {**self.getEngineStatus(), "error": f"Profil non selectionnable : {profile.note}"}
if profile.id == self.getActiveModelProfileId() and self.getEngineStatus()["llmLoaded"]:
return {**self.getEngineStatus(), "profileId": profile.id, "switched": False}
if not self.generationLock.acquire(blocking=False):
return {**self.getEngineStatus(), "error": "generation_in_progress"}
# Etat a restaurer si la bascule echoue. Sans cela, un echec laissait le
# moteur casse en place : le profil distant avait deja ete ecrit dans la
# configuration et le moteur local mis en cache, donc le serveur ne
# savait plus quel modele servait. Le retour au local ne passait plus
# par la branche qui restaure le moteur, et le modele local semblait
# alors casse lui aussi.
previousProfileId = self.activeProfileId
previousEngine = self.llmEngine
try:
self._engineBusy = True
self.isInterrupted = True
self.audioPipeline.interrupt()
isTargetRemote = profile.provider == "openrouter"
isCurrentRemote = getattr(self.llmEngine, "isRemote", False) is True
if isTargetRemote:
# Bascule vers OpenRouter : conserver le moteur local en cache sans le decharger
print(f"[Modeles] Bascule a chaud vers {profile.label} (conservation VRAM)...")
if not isCurrentRemote:
self._cachedLocalEngine = self.llmEngine
self._cachedLocalProfileId = self.getActiveModelProfileId() or self.activeProfileId
self.activeProfileId = profile.id
self._writeProfileConfig(profile)
if getattr(self, "_cachedRemoteEngine", None) is not None:
self.llmEngine = self._cachedRemoteEngine
else:
self.llmEngine = EngineFactory.createLlmEngine(
modelConfig=self.config.llm,
provider=profile.provider,
)
self.agentManager.setLlmEngine(self.llmEngine)
self.agentManager.setAllowedTools(profile.allowedTools)
self.memoryRetrievalEnabled = profile.useMemory
if not profile.useMemory:
self.lastInjectedRagContext = None
if getattr(self, "sessionRagHistory", None) is not None:
self.sessionRagHistory.clear()
if not getattr(self.llmEngine, "isLoaded", False):
self.llmEngine.loadModel(systemPrompt=self.config.prompt.systemPrompt)
elif (
getattr(self, "_cachedLocalEngine", None) is not None
and profile.id == getattr(self, "_cachedLocalProfileId", None)
):
# Retour instantane vers le modele local conserve en RAM
print(f"[Modeles] Retour instantane vers {profile.label} (restauration 0 ms)...")
self._cachedRemoteEngine = self.llmEngine
self.llmEngine = self._cachedLocalEngine
self._cachedLocalEngine = None
self._cachedLocalProfileId = None
self.activeProfileId = profile.id
self._writeProfileConfig(profile)
self.agentManager.setLlmEngine(self.llmEngine)
self.agentManager.setAllowedTools(profile.allowedTools)
self.memoryRetrievalEnabled = profile.useMemory
if not profile.useMemory:
self.lastInjectedRagContext = None
if getattr(self, "sessionRagHistory", None) is not None:
self.sessionRagHistory.clear()
if not getattr(self.llmEngine, "isLoaded", False):
self.llmEngine.loadModel(systemPrompt=self.config.prompt.systemPrompt)
else:
# Bascule vers un autre modele local : liberation du LLM precedent requise
self._notifyState("CHARGEMENT_MODELES")
self._releaseEngines()
self._engineBusy = False
self.activeProfileId = profile.id
self._applyProfile(profile)
self.loadModels()
if not profile.useMemory:
self.lastInjectedRagContext = None
if getattr(self, "sessionRagHistory", None) is not None:
self.sessionRagHistory.clear()
except Exception as exc:
logger.error("Echec de la bascule de modele vers %s : %s", profileId, exc, exc_info=True)
# Restaure l'etat d'avant la bascule. Le moteur distant a deja
# remplace le local dans llmEngine et ecrit sa configuration quand
# il echoue a se charger ; sans cette remise en place, le serveur
# tourne avec un moteur qui ne peut rien servir et un profil qui
# ne correspond plus a ce qui est en memoire.
self.activeProfileId = previousProfileId
self.llmEngine = previousEngine
self._writeProfileConfig(getProfile(previousProfileId))
self.agentManager.setLlmEngine(self.llmEngine)
return {**self.getEngineStatus(), "error": str(exc)}
finally:
self._engineBusy = False
self.generationLock.release()
self.isInterrupted = False
self._notifyState("IDLE")
if getattr(self, "onMemoryUpdated", None):
try:
self.onMemoryUpdated(self.getMemoryStatus())
except Exception as exc:
logger.warning("Notification de statut memoire echouee : %s", exc)
if getattr(self, "onActiveContextUpdated", None):
try:
self.onActiveContextUpdated(self.getActiveContextInfo())
except Exception as exc:
logger.debug("Erreur broadcast onActiveContextUpdated: %s", exc)
return {**self.getEngineStatus(), "profileId": profile.id, "switched": True}
[docs]
def listAssistantProfiles(self) -> list[dict[str, Any]]:
"""
Liste l'ensemble des profils d'assistant configurés.
Returns
-------
list[dict[str, Any]]
Description de chaque profil d'assistant.
"""
if hasattr(self.config, "listAssistantProfiles"):
return self.config.listAssistantProfiles()
return []
[docs]
def getActiveAssistantProfileId(self) -> str:
"""
Retourne l'identifiant du profil d'assistant actuellement actif.
Returns
-------
str
Identifiant du profil (ex: "assistant", "medical").
"""
return getattr(self, "activeAssistantProfileId", "assistant")
[docs]
def switchAssistantProfile(self, profileId: str) -> dict[str, Any]:
"""
Bascule a chaud vers un autre profil d'assistant (ex: assistant, medical).
Met a jour le prompt systeme, les prompts et outils de l'agent, les phrases
d'attente vocale et la base de donnees memoire.
Parameters
----------
profileId : str
Identifiant du profil cible.
Returns
-------
dict[str, Any]
Resultat de la bascule.
"""
if not hasattr(self.config, "assistantProfiles") or profileId not in self.config.assistantProfiles:
valid = list(self.config.assistantProfiles.keys()) if hasattr(self.config, "assistantProfiles") else []
return {
"status": "error",
"error": f"Profil d'assistant inconnu : {profileId!r}. Disponibles : {valid}",
"activeAssistantProfile": self.getActiveAssistantProfileId(),
}
prof = self.config.getAssistantProfile(profileId)
self.activeAssistantProfileId = profileId
self.config.applyAssistantProfile(profileId)
# 1. Phrases d'attente / reflexion vocale
self.thinkingFillerPhrases = prof.fillerPhrases or THINKING_FILLER_PHRASES
# 2. Prompts et restrictions d'outils de l'agent
if self.agentManager:
self.agentManager.setAssistantProfilePrompts(
systemPrompt=prof.agent.systemPrompt,
instructionsPrompt=prof.agent.instructionsPrompt,
finalAnswerPrompt={
"pre_messages": prof.agent.finalAnswerPre,
"post_messages": prof.agent.finalAnswerPost,
},
allowedTools=prof.allowedTools,
)
# 3. Base de donnees de memoire episodique
if prof.memoryDbPath and self.memoryStore and getattr(self.memoryStore, "dbPath", "") != prof.memoryDbPath:
try:
self.memoryStore = EpisodicMemoryStore(
dbPath=prof.memoryDbPath,
embeddingModel=self.config.memory.embeddingModel,
embeddingDim=self.config.memory.embeddingDim,
similarityThreshold=0.10,
maxContextTokens=self.config.memory.maxContextTokens,
topKRetrieval=self.config.memory.topKRetrieval,
topKRerank=self.config.memory.topKRerank,
)
from ..engine.agent.tools import setActiveMemoryStore
setActiveMemoryStore(self.memoryStore)
logger.info("Magasin de memoire bascule vers : %s", prof.memoryDbPath)
except Exception as exc:
logger.warning("Echec de la bascule de memoire vers %s : %s", prof.memoryDbPath, exc)
# 4. Le systemPrompt est passe dynamiquement a formatPrompt lors de chaque tour de generation (0ms reload)
logger.debug("System prompt actif mis a jour pour le profil %s", profileId)
logger.info("Profil d'assistant bascule vers : %s (%s)", prof.label, profileId)
return {
"status": "ok",
"activeAssistantProfile": profileId,
"profile": prof.toDict(),
}
def _releaseEngines(self) -> None:
"""
Libere tous les poids resident et restitue la memoire unifiee.
Volontairement tolerant : un moteur qui echoue a l'arret ne doit pas
empecher les autres d'etre liberes.
"""
if getattr(self, "_cachedLocalEngine", None) is not None:
try:
self._cachedLocalEngine.unload()
except Exception as exc:
logger.warning("Liberation du moteur local en cache impossible : %s", exc)
self._cachedLocalEngine = None
self._cachedLocalProfileId = None
for name, engine in (("TTS", self.ttsEngine), ("STT", self.sttEngine), ("LLM", self.llmEngine)):
try:
engine.unload()
except Exception as exc:
logger.warning("Liberation du moteur %s impossible : %s", name, exc)
# Forcee : les poids viennent d'etre liberes, la memoire doit revenir.
MemoryManager.purgeMemory(force=True)
def _writeProfileConfig(self, profile: ModelProfile) -> None:
"""
Ecrit les reglages du profil dans la configuration du moteur.
Separe de :meth:`_applyProfile` parce qu'elle doit pouvoir tourner
avant que le moteur et l'agent existent, au demarrage.
Parameters
----------
profile : ModelProfile
Le profil a ecrire dans la configuration.
"""
llmConfig = self.config.llm
llmConfig.provider = profile.provider
# Application de l'echantillonnage recommande du modele
for key, value in recommendedSampling(profile.modelRepo).items():
field = _SAMPLING_FIELDS.get(key)
if field and hasattr(llmConfig, field):
setattr(llmConfig, field, value)
if profile.kvBits:
llmConfig.kvBits = profile.kvBits
if profile.provider == "mtplx":
llmConfig.mtplxModelRepo = profile.modelRepo
llmConfig.drafterModelRepo = profile.drafterRepo
elif profile.provider == "openrouter":
llmConfig.openrouterModel = profile.modelRepo
llmConfig.llmModelRepo = profile.modelRepo
llmConfig.draftModelRepo = profile.drafterRepo
llmConfig.enableSpeculative = bool(profile.drafterRepo)
llmConfig.draftKind = profile.draftKind
llmConfig.draftBlockSize = profile.draftBlockSize
llmConfig.maxKvCacheTokens = min(profile.contextWindow, SAFE_CONTEXT_TOKENS)
llmConfig.optiqMtp = profile.mtpDepth > 0
llmConfig.optiqMtpDepth = profile.mtpDepth or 2
thinking = bool(self._thinkingOverride) if self._thinkingOverride is not None else profile.thinkingByDefault
if not profile.thinkingByDefault and self._thinkingOverride is None:
thinking = False
llmConfig.enableThinking = thinking
llmConfig.reasoningBudget = self._reasoningBudget if thinking else 0
llmConfig.openrouterModel = (
profile.modelRepo if profile.provider == "openrouter" else ""
)
llmConfig.maxTokens = profile.maxOutputTokens or 1024
llmConfig.maxTokensAgent = profile.maxAgentTokens or 512
if profile.provider == "openrouter":
llmConfig.optiqMtp = False
llmConfig.drafterModelRepo = None
llmConfig.kvBits = 8
# La memoire et les outils suivent le profil, pas la session : passer au
# modele distant ne doit pas laisser partir les souvenirs, et y revenir
# doit les rendre.
self.memoryRetrievalEnabled = profile.useMemory
# Au demarrage, cette methode s'execute avant que l'agent existe : les
# restrictions d'outils sont reappliquees des sa construction, qui lit
# le profil actif.
if self.agentManager is not None:
self.agentManager.setAllowedTools(profile.allowedTools)
def _applyProfile(self, profile: ModelProfile) -> None:
"""
Ecrit le profil, puis reconstruit le moteur qui doit le servir.
Parameters
----------
profile : ModelProfile
Le profil a appliquer.
"""
self._writeProfileConfig(profile)
llmConfig = self.config.llm
self.llmEngine = EngineFactory.createLlmEngine(
modelConfig=llmConfig,
provider=profile.provider,
)
# L'agent garde une reference vers le moteur qu'il avait a sa creation :
# sans ce rattachement il continue d'appeler le daemon du profil de
# demarrage, arrete, et repond "erreur de communication".
self.agentManager.setLlmEngine(self.llmEngine)
try:
self.llmEngine.loadModel(systemPrompt=self.config.prompt.systemPrompt)
except Exception as loadErr:
logger.warning("Erreur chargement/prefill modele lors de l'application du profil %s : %s", profile.id, loadErr)
logger.info("Profil de modele applique : %s (%s)", profile.id, profile.modelRepo)
[docs]
def getEngineStatus(self) -> dict:
"""
Renvoie l'etat de chargement de chaque moteur.
Returns
-------
dict
Les cles ``llmLoaded``, ``sttLoaded``, ``ttsLoaded``, ``allLoaded``
et ``anyLoading``.
"""
# textOnly ne precharge ni STT ni TTS : ils sont rapportes charges pour
# que le HUD n'alerte pas sur un dechargement qui ne libererait rien.
llmLoaded = bool(getattr(self.llmEngine, "isLoaded", True))
return {
"llmLoaded": llmLoaded,
"sttLoaded": True if self.textOnly else bool(getattr(self.sttEngine, "isLoaded", True)),
"ttsLoaded": True if self.textOnly else bool(getattr(self.ttsEngine, "isLoaded", True)),
"allLoaded": llmLoaded,
"anyLoading": self._engineBusy,
}
[docs]
def setEngineLoaded(self, loaded: bool) -> dict:
"""
Decharge ou recharge l'integralite des modeles a la volee.
Toute requete est refusee pendant une generation : decharger les poids
sous un flux de tokens en cours produirait une panne segfault.
Parameters
----------
loaded : bool
True pour recharger, False pour tout decharger.
Returns
-------
dict
L'etat des moteurs apres l'operation, avec ``"error"`` en cas d'echec.
"""
if not self.generationLock.acquire(blocking=False):
return {**self.getEngineStatus(), "error": "generation_in_progress"}
try:
self._engineBusy = True
self.isInterrupted = True
self.audioPipeline.interrupt()
if loaded:
self._notifyState("CHARGEMENT_MODELES")
print("[Moteurs] Rechargement des modeles en cours...")
self.loadModels()
else:
self._notifyState("IDLE")
print("[Moteurs] Dechargement des modeles en cours...")
if getattr(self, "_cachedLocalEngine", None) is not None:
try:
self._cachedLocalEngine.unload()
except Exception as exc:
logger.warning("Dechargement du moteur local en cache impossible : %s", exc)
self._cachedLocalEngine = None
self._cachedLocalProfileId = None
for name, engine in (("TTS", self.ttsEngine), ("STT", self.sttEngine), ("LLM", self.llmEngine)):
try:
engine.unload()
except Exception as exc:
logger.warning("Dechargement du moteur %s impossible : %s", name, exc)
MemoryManager.purgeMemory()
print("[Moteurs] Modeles decharges.")
except Exception as exc:
logger.error("Echec du basculement des modeles : %s", exc, exc_info=True)
return {**self.getEngineStatus(), "error": str(exc)}
finally:
self._engineBusy = False
self.generationLock.release()
self.isInterrupted = False
self._notifyState("IDLE")
status = self.getEngineStatus()
self.lastInjectedRagContext = None
return status
def _resolveTurnModes(self) -> TurnModes:
"""
Resolve the execution modes for the current turn.
Returns
-------
TurnModes
Snapshot of agent mode, reflection mode and the thought budget.
"""
return TurnModes.resolve(
agentActive=self.agentManager.isEnabled(),
thinkingEnabled=bool(getattr(self.llmEngine, "enableThinking", False)),
budget=int(getattr(self.llmEngine, "reasoningBudget", 0) or 0),
)
def _buildTurnHistory(
self,
outcome: TurnOutcome,
streamedText: str,
resultBuffer: str,
agentActive: bool,
) -> tuple[str, str]:
"""
Build the two texts kept for a completed turn.
Parameters
----------
outcome : TurnOutcome
What the agent branch produced, if it ran.
streamedText : str
Everything streamed as speech during the turn.
resultBuffer : str
Document content captured by the streaming filter.
agentActive : bool
Whether the turn went through the agent.
Returns
-------
tuple[str, str]
The text shown in the chat, and the (shorter) text kept in history.
"""
if agentActive and outcome.agentText:
raw_display = outcome.agentText.strip()
# Nettoyer les eventuels artefacts XML d'outils et de safety
raw_display = re.sub(r"<tool_call>[\s\S]*?</tool_call>", "", raw_display)
raw_display = re.sub(r"(?:User|Response)\s+Safety\s*:\s*\w+\s*", "", raw_display).strip()
display = raw_display
synthesis = outcome.spoken
if not synthesis:
docMatch = re.search(r"<(?:result|code|code_view)\b|```", display, re.IGNORECASE)
synthesis = (
display[:docMatch.start()].strip() if docMatch else display
)
synthesis = TextCleaner.cleanForTts(stripArchivePhrases(synthesis))
sentences = splitSentences(synthesis)
historyText = " ".join(sentences[:2]) if sentences else synthesis[:200]
else:
if resultBuffer.strip():
display = (
f"{streamedText.strip()}\n\n{resultBuffer.strip()}".strip()
if streamedText.strip()
else resultBuffer.strip()
)
# Conserver explicitement la balise <result> dans l'historique conversationnel
# afin que le LLM apprenne in-context à toujours encapsuler ses documents
historyText = (
f"{streamedText.strip()}\n<result>\n{resultBuffer.strip()}\n</result>".strip()
if streamedText.strip()
else f"<result>\n{resultBuffer.strip()}\n</result>"
)
else:
display = streamedText.strip()
historyText = display
display = TextCleaner.stripEmojis(display).strip()
historyText = TextCleaner.stripEmojis(historyText).strip()
return display, historyText
def _archiveTurnInBackground(
self,
userText: str,
displayResponse: str,
document: str | None,
) -> None:
"""
Archive a completed turn in the episodic store without blocking the turn.
Runs on a daemon thread so the orchestrator stays available for the next
request. Any failure is logged and swallowed: losing a memory must never
break the conversation that produced it.
Parameters
----------
userText : str
The user input, as spoken.
displayResponse : str
The assistant response, as displayed.
document : str | None
The diverted document, archived separately when substantial.
"""
if not self.memoryStore or not getattr(self.config.memory, "autoArchive", True):
return
# Pour le profil médical, aucun document ni tour conversationnel ordinaire n'est archivé
# automatiquement en mémoire épisodique pour éviter d'induire le LLM en erreur.
if self.getActiveAssistantProfileId() == "medical":
return
def _archive() -> None:
try:
cleanUser = stripMemoTag(userText)
# Archivage du tour de conversation (assistant et utilisateur)
self.memoryStore.storeTurn(
userInput=cleanUser,
assistantResponse=stripMemoTag(displayResponse),
)
# Un document agent substantiel merite sa propre entree, sauf
# s'il figure deja dans la reponse archivee ci-dessus : sinon
# le meme texte est stocke sous deux cles et chaque rappel
# renvoie la paire.
cleanDoc = _cleanArchivedDocument(document) if document else ""
alreadyArchived = (
cleanDoc
and cleanDoc[:120].lower() in stripMemoTag(displayResponse).lower()
)
if len(cleanDoc) >= 80 and not alreadyArchived:
self.memoryStore.storeTurn(
userInput=f"Document clinique ou de recherche extrait : {cleanUser}",
assistantResponse=cleanDoc,
summary=f"Recherche documentaire : {cleanUser[:60]}",
metadata='{"source": "agent_doc"}',
)
if self.onMemoryUpdated:
self.onMemoryUpdated(self.getMemoryStatus())
except Exception as exc:
logger.warning("Erreur archivage tour en memoire episodique : %s", exc)
threading.Thread(target=_archive, daemon=True).start()
def _signalCodeGenerating(self, active: bool) -> None:
"""
Tell the HUD whether the agent is currently producing a document.
Parameters
----------
active : bool
True while the agent is running, False once it has finished.
"""
if not self.onCodeGenerating:
return
try:
self.onCodeGenerating(active)
except Exception as exc:
print(f"[Orchestrator] Erreur callback onCodeGenerating: {exc}")
def _runAgentTurn(
self,
userText: str,
outcome: TurnOutcome,
textQueue: queue.Queue,
consume: Callable[[str], None],
onThought: Callable[[str], None],
streamedSoFar: Callable[[], str],
documentAlreadySent: bool,
) -> None:
"""
Run one turn through the smolagents CodeAgent and speak the result.
Parameters
----------
userText : str
The user input.
outcome : TurnOutcome
Filled in with the agent's text, document and spoken synthesis.
textQueue : queue.Queue
Queue feeding the TTS worker.
consume : Callable[[str], None]
Callback that filters, records and speaks a text chunk.
onThought : Callable[[str], None]
Callback streaming the agent's reasoning steps.
streamedSoFar : Callable[[], str]
Returns the speech streamed so far this turn.
documentAlreadySent : bool
Whether the streaming filter already delivered a document, in which
case the document is not sent twice.
"""
# Le mode agent n'injecte pas la memoire episodique dans le prompt :
# l'agent indexe ses propres resultats via AgentManager.
self.lastInjectedRagContext = None
print("\n[Agent] Mode Agentique smolagents actif. Analyse et résolution avec outils...")
print("Mirza : ", end="")
parser = FinalAnswerStreamParser(on_speech_chunk=consume, on_thought_chunk=onThought)
# Phrase de remplissage immediatee pour eviter tout silence.
textQueue.put(random.choice(self.thinkingFillerPhrases or THINKING_FILLER_PHRASES))
self._signalCodeGenerating(True)
try:
agentText = self.agentManager.run(
userPrompt=userText,
conversationHistory=self.conversationHistory,
onThoughtChunk=onThought,
onTokenChunk=parser.process_chunk,
)
finally:
self._signalCodeGenerating(False)
parser.finalize()
# Arbitrage document / parole : fonction pure, regles dans agentOutput.py.
outcome.agentText = agentText
outcome.document = splitAgentOutput(agentText).document
outcome.spoken = splitAgentOutput(agentText).spoken
if outcome.document and not documentAlreadySent and self.onAgentDoc:
self.onAgentDoc(outcome.document)
isError = agentText.startswith(AGENT_ERROR_PREFIXES)
if parser.hasStreamedSpeech() and not isError:
# La parole est deja partie en direct : seule l'annonce d'archive
# peut manquer.
if outcome.document and needsArchiveAnnouncement(streamedSoFar()):
consume(f" {ARCHIVE_ANNOUNCEMENT}")
return
# Aucune parole en direct, ou une erreur : on rejoue la synthese.
chunks = splitSpokenForSynthesis(outcome.spoken) if outcome.spoken else spokenFallback(agentText, isError)
for chunk in chunks:
consume(chunk)
def _runDirectTurn(
self,
userText: str,
modes: TurnModes,
outcome: TurnOutcome,
textQueue: queue.Queue,
consume: Callable[[str], None],
) -> None:
"""
Run one turn straight through the LLM, with optional memory injection.
Parameters
----------
userText : str
The user input.
modes : TurnModes
The resolved turn modes.
outcome : TurnOutcome
Turn outcome collecting turn metadata (such as RAG context).
textQueue : queue.Queue
Queue feeding the TTS worker.
consume : Callable[[str], None]
Callback that filters, records and speaks a text chunk.
"""
print("\n[Streaming] Reponse de Mirza en direct...")
print("Mirza : ", end="")
augmentedInput = userText
if self.memoryStore and self.memoryRetrievalEnabled:
memoryContext = self.memoryStore.formatMemoryContext(query=userText)
if memoryContext:
outcome.ragContext = memoryContext
self.lastInjectedRagContext = memoryContext
ragTokens = len(memoryContext) // 4
self.sessionRagHistory.append({
"turn": len(self.conversationHistory) // 2 + 1,
"query": userText,
"ragContext": memoryContext,
"ragTokens": ragTokens,
"timestamp": time.time(),
})
print(f"\n[Memoire] Contexte episodique pertinent retrouve et injecte (~{ragTokens} tokens).")
if self.onLog:
self.onLog("MEMORY", f"Contexte RAG injecté : ~{ragTokens} tokens", "INFO")
augmentedInput = f"{memoryContext}\n{userText}"
# Calcul du budget de tokens pour la fenetre glissante
answerTokens = (
self.config.llm.maxTokensAgent
if self.agentManager.isEnabled()
else self.config.llm.maxTokens
)
reservedGeneration = self.llmEngine.resolveMaxTokens(
answerTokens, self.config.llm.maxTokensThinking
)
reservedTokens = (
len(self.config.prompt.systemPrompt) + len(augmentedInput)
) // 4 + reservedGeneration
# Application de la limite de tokens pour les modeles locaux
if not getattr(self.llmEngine, "isRemote", False):
self.contextManager.enforceTokenLimit(
self.config.llm.maxKvCacheTokens,
reservedTokens=reservedTokens,
)
if modes.streamsThoughtsUpFront:
# Phrase de remplissage pour que le delai de reflexion ne s'entende pas.
textQueue.put(random.choice(self.thinkingFillerPhrases or THINKING_FILLER_PHRASES))
prompt = self.llmEngine.formatPrompt(
conversationHistory=self.conversationHistory,
currentUserInput=augmentedInput,
systemPrompt=self.config.prompt.systemPrompt,
maxTurns=self.config.prompt.historyMaxTurns,
)
for chunk in self.llmEngine.streamResponse(prompt):
if self.isInterrupted:
break
consume(chunk)
def _emitTurnMetrics(
self,
speechEndTime: float,
llmStartTime: float,
llmEndTime: float,
sttDurationMs: float,
firstAudioTime: float | None,
) -> None:
"""
Publish the latency and throughput metrics of a completed turn.
Round-trip latency is measured to the moment the first audio sample was
handed to the sound card, or to the end of generation when no audio was
played. Everything is contained: a metrics failure must never fail the
turn.
Parameters
----------
speechEndTime : float
``time.time()`` when the user stopped speaking.
llmStartTime : float
``time.time()`` when generation started.
llmEndTime : float
``time.time()`` when generation and playback finished.
sttDurationMs : float
Transcription duration in milliseconds.
firstAudioTime : float | None
``time.time()`` when the first audio sample was played, if any.
"""
try:
ttftMs = getattr(self.llmEngine, "lastTtftMs", 0.0) or 0.0
tokensPerSec = getattr(self.llmEngine, "currentSpeedTokPerSec", 0.0) or 0.0
tokenCount = getattr(self.llmEngine, "lastTokenCount", 0) or 0
kvTokens = getattr(self.llmEngine, "estimatedKvTokens", 0) or 0
maxKvTokens = self.config.llm.maxKvCacheTokens
endReference = firstAudioTime if firstAudioTime is not None else llmEndTime
roundtripLatencyMs = (endReference - speechEndTime) * 1000.0
if getattr(self.llmEngine, "isRemote", False):
# Aucun cache local : afficher une occupation de cache sur un
# modele distant inventerait une mesure. L'aiguille reste vide.
kvPercentage = 0.0
kvTokens = 0
else:
kvPercentage = min(100.0, round((kvTokens / float(maxKvTokens)) * 100, 1))
turnMetrics = {
"sttDurationMs": round(sttDurationMs, 1),
"ttftMs": round(ttftMs, 1),
"llmTotalTimeMs": round((llmEndTime - llmStartTime) * 1000.0, 1),
"tokensPerSec": round(tokensPerSec, 1),
"tokenCount": tokenCount,
"kvCacheTokens": kvTokens,
"kvCacheMax": maxKvTokens,
"kvCachePercentage": kvPercentage,
"roundtripLatencyMs": round(roundtripLatencyMs, 1),
}
if self.onMetrics:
self.onMetrics(turnMetrics)
print(
f"\n[Metriques] TTFT: {ttftMs:.0f}ms | Vitesse: {tokensPerSec:.1f} tok/s | "
f"KV: {kvTokens}/{maxKvTokens} ({kvPercentage}%) | Latence totale: {roundtripLatencyMs:.0f}ms\n"
)
except Exception as exc:
logger.warning("Erreur lors du calcul des metriques de tour : %s", exc)
def _generateAndSpeak(
self,
userText: str,
speechEndTime: float,
sttDurationMs: float = 0.0,
isClientOrigin: bool = False,
clientTargetId: str | None = None,
) -> bool:
"""
Stream la reponse LLM et dispatche le TTS pour un tour complet.
Verrouillee par ``generationLock`` pour interdire toute concurrence sur
le GPU Metal.
Parameters
----------
userText : str
La requete de l'utilisateur, deja transcribee.
speechEndTime : float
``time.time()`` a la fin de la parole de l'utilisateur.
sttDurationMs : float, optional
Duree de transcription en millisecondes, defaut: 0.0.
isClientOrigin : bool, optional
True si le tour provient d'un client distant, defaut: False.
clientTargetId : str | None, optional
Identifiant du client WebSocket destinataire de l'audio.
Returns
-------
bool
True si le tour s'est deroule, False s'il a ete rejete ou en erreur.
"""
if not self.generationLock.acquire(blocking=False):
print("[Generation] Rejet : une génération est déjà active sur le GPU Metal.")
return False
try:
self.isInterrupted = False
self.audioPipeline.resetInterrupt()
self._sanitizeHistory()
modes = self._resolveTurnModes()
self._notifyState("THINKING" if modes.showsThinking() else "GENERATING")
print(f"\nVous : {userText}")
textQueue = queue.Queue()
audioQueue = queue.Queue()
firstAudioPlaybackTime = [None]
def onFirstAudio():
"""Record when the first audio sample reached the sound card."""
if firstAudioPlaybackTime[0] is None:
firstAudioPlaybackTime[0] = time.time()
ttsThread = threading.Thread(target=self._ttsWorker, args=(textQueue, audioQueue), daemon=True)
ttsThread.start()
playbackThread = threading.Thread(
target=self._playbackWorker,
args=(audioQueue, onFirstAudio, isClientOrigin, clientTargetId),
daemon=True,
)
playbackThread.start()
def _handleThought(tChunk: str):
if self.onThoughtStream:
self.onThoughtStream(tChunk)
docSentToClient = False
def _handleResultDoc(docText: str):
nonlocal docSentToClient
cleanedDoc = stripArchivePhrases(docText)
if self.onAgentDoc and cleanedDoc:
docSentToClient = True
self.onAgentDoc(cleanedDoc)
fullAssistantResponse = ""
splitter = ClauseSplitter()
outcome = TurnOutcome()
def _handleDocActive(active: bool):
if self.onCodeGenerating:
try:
self.onCodeGenerating(active)
except Exception as e:
print(f"[Orchestrator] Erreur callback onCodeGenerating: {e}")
textFilter = StreamingTextFilter(
onThought=_handleThought,
onResultDoc=_handleResultDoc,
onDocActive=_handleDocActive,
maxThoughtTokens=modes.reasoningBudget,
initialThinkingActive=modes.streamsThoughtsUpFront,
)
llmStartTime = time.time()
def _consumeChunk(textChunk: str):
nonlocal fullAssistantResponse
if self.isInterrupted:
return
# Filtre de streaming : retire les blocs de reflexion, les
# balises documentaires et le balisage orphelin.
filteredChunk = textFilter.processChunk(textChunk)
if not filteredChunk:
return
fullAssistantResponse += filteredChunk
print(filteredChunk, end="", flush=True)
if self.onTokenStream:
self.onTokenStream(filteredChunk)
clause = splitter.feed(filteredChunk)
textToSpeak = TextCleaner.cleanForTts(clause)
if len(textToSpeak) >= 3:
textQueue.put(textToSpeak)
if modes.agentActive:
self._runAgentTurn(
userText=userText,
outcome=outcome,
textQueue=textQueue,
consume=_consumeChunk,
onThought=_handleThought,
streamedSoFar=lambda: fullAssistantResponse,
documentAlreadySent=docSentToClient,
)
else:
self._runDirectTurn(
userText=userText,
modes=modes,
outcome=outcome,
textQueue=textQueue,
consume=_consumeChunk,
)
finalChunk = textFilter.finalize()
if finalChunk:
fullAssistantResponse += finalChunk
splitter.feed(finalChunk)
print(finalChunk, end="", flush=True)
if self.onTokenStream:
self.onTokenStream(finalChunk)
# Garde-fou anti-blocage : repli vocal si le modele s'est arrete sans texte.
if not fullAssistantResponse.strip() and not self.isInterrupted:
if textFilter.resultBuffer.strip():
if getattr(textFilter, "currentDocTag", None) == "code" or "```" in textFilter.resultBuffer:
summary = "Voici le document demandé. Je suis en train de générer un compte rendu."
else:
cleanRes = TextCleaner.cleanForTts(textFilter.resultBuffer)
sentences = [s.strip() for s in re.split(r'(?<=[.!?])\s+', cleanRes) if s.strip()]
summary = " ".join(sentences[:2]) if sentences else cleanRes[:150]
if not any(k in summary.lower() for k in ("générer un compte", "archivé", "archiver")):
summary = f"{summary.rstrip('.')} Je suis en train de générer un compte rendu."
fullAssistantResponse = summary
splitter.feed(summary)
if self.onTokenStream:
self.onTokenStream(summary)
else:
fallbackMsg = "J'ai terminé l'analyse. Que souhaitez-vous approfondir ?"
fullAssistantResponse = fallbackMsg
splitter.feed(fallbackMsg)
print(fallbackMsg, flush=True)
if self.onTokenStream:
self.onTokenStream(fallbackMsg)
residual = splitter.flush()
if residual.strip() and not self.isInterrupted:
textToSpeak = TextCleaner.cleanForTts(residual)
if len(textToSpeak) >= 1:
textQueue.put(textToSpeak)
textQueue.put(None)
ttsThread.join()
playbackThread.join()
llmEndTime = time.time()
displayResponse, historyAssistantContent = self._buildTurnHistory(
outcome=outcome,
streamedText=fullAssistantResponse,
resultBuffer=textFilter.resultBuffer,
agentActive=modes.agentActive,
)
if historyAssistantContent:
userEntry: dict[str, Any] = {
"role": "user",
"content": stripMemoTag(userText),
"ragInjected": bool(outcome.ragContext),
"ragContext": outcome.ragContext,
"ragTokens": len(outcome.ragContext) // 4 if outcome.ragContext else 0,
}
self.conversationHistory.append(userEntry)
self.conversationHistory.append({"role": "assistant", "content": historyAssistantContent})
self._archiveTurnInBackground(
userText=userText,
displayResponse=displayResponse,
document=outcome.document,
)
if self.onAssistantResponseComplete:
self.onAssistantResponseComplete(displayResponse)
# Apres un tour agent, purger le cache KV et reaffecter les tokens
# estimes au dialogue reel.
if modes.agentActive and outcome.agentText:
try:
self.llmEngine.clearCache()
except Exception as clearErr:
logger.debug("Erreur lors de la reinitialisation du cache KV apres tour agent : %s", clearErr)
sysTokens = max(1, len(self.config.prompt.systemPrompt) // 4)
histTokens = sum(len(m.get("content", "")) // 4 for m in self.conversationHistory)
ragTokens = len(self.lastInjectedRagContext) // 4 if self.lastInjectedRagContext else 0
self.llmEngine.estimatedKvTokens = sysTokens + histTokens + ragTokens
self._emitTurnMetrics(
speechEndTime=speechEndTime,
llmStartTime=llmStartTime,
llmEndTime=llmEndTime,
sttDurationMs=sttDurationMs,
firstAudioTime=firstAudioPlaybackTime[0],
)
if self.onActiveContextUpdated:
try:
self.onActiveContextUpdated(self.getActiveContextInfo())
except Exception as ctxErr:
logger.debug("Erreur broadcast onActiveContextUpdated: %s", ctxErr)
return True
except Exception as e:
import traceback
print(f"[Generation] ERREUR dans _generateAndSpeak : {e}")
traceback.print_exc()
if self.onAssistantResponseComplete:
self.onAssistantResponseComplete("Désolé, une anomalie temporaire est survenue.")
return False
finally:
self._notifyState("IDLE")
self.generationLock.release()
# Mode automatique : ne se declenche que sous vraie pression
# memoire. Purger a chaque tour jetait les buffers Metal que le
# tour suivant reutilise.
MemoryManager.purgeMemory()
[docs]
def processClientAudio(self, audioBytes: bytes, clientTargetId: str | None = None) -> bool:
"""
Traite un flux audio envoye directement par un client distant.
Parameters
----------
audioBytes : bytes
Octets binaires bruts de l'enregistrement audio au format WAV ou PCM.
clientTargetId : str, optional
Identifiant unique du client WebSocket demandeur.
Returns
-------
bool
True si la transcription et la synthese vocale ont abouti, False sinon.
"""
if not audioBytes or len(audioBytes) == 0:
self._notifyState("IDLE")
return False
# Regle Half-Duplex : ignorer tout paquet audio si Mirza parle ou infere.
if self.currentState in ("THINKING", "SPEAKING", "TRANSCRIBING"):
print("[Audio Client] Paquet audio ignoré : Mirza est déjà active (Half-Duplex).")
self._notifyState(self.currentState)
return False
self.isInterrupted = False
self.audioPipeline.resetInterrupt()
self._notifyState("TRANSCRIBING")
t0Stt = time.time()
try:
audioArray, _sampleRate = AudioProcessor.decodeAudioBytes(audioBytes)
if len(audioArray) == 0:
print("[Audio Client] Flux audio vide ou non decode.")
self._notifyState("IDLE")
return False
audioDurationSec = len(audioArray) / 16000.0 if len(audioArray) > 0 else 0.0
peakLevel = float(np.max(np.abs(audioArray))) if len(audioArray) > 0 else 0.0
sttLang = getattr(self.config.stt, "language", "en")
transcribedText = self.sttEngine.transcribe(audioArray, language=sttLang)
sttDurationMs = (time.time() - t0Stt) * 1000.0
if not transcribedText or len(transcribedText.strip()) < 2:
print(f"[Audio Client] Aucun texte intelligible detecte ({audioDurationSec:.2f}s, peak={peakLevel:.3f}).")
self._notifyState("IDLE")
return False
print(f"[Audio Client] Transcrit ({audioDurationSec:.2f}s, peak={peakLevel:.3f}) : \"{transcribedText}\" (en {sttDurationMs:.1f}ms)")
if self.onTranscription:
self.onTranscription(transcribedText)
return self._generateAndSpeak(
userText=transcribedText,
speechEndTime=time.time(),
sttDurationMs=sttDurationMs,
isClientOrigin=True,
clientTargetId=clientTargetId,
)
except Exception as e:
print(f"[Audio Client] Erreur lors du decodage de l'audio client: {e}")
self._notifyState("IDLE")
return False
[docs]
def processTextInput(self, textInput: str, clientTargetId: str | None = None) -> bool:
"""
Traite directement une invite textuelle sans capture microphonique.
Parameters
----------
textInput : str
Texte de la question ou commande soumise par l'utilisateur.
clientTargetId : str, optional
Identifiant du client WebSocket cible.
Returns
-------
bool
True si la reponse a ete generee avec succes, False si l'entree est vide.
"""
if not textInput or len(textInput.strip()) == 0:
return False
self.isInterrupted = False
self.audioPipeline.resetInterrupt()
if self.onTranscription:
self.onTranscription(textInput)
return self._generateAndSpeak(
userText=textInput,
speechEndTime=time.time(),
sttDurationMs=0.0,
isClientOrigin=True,
clientTargetId=clientTargetId,
)
[docs]
def processSingleTurn(self) -> bool:
"""
Execute un tour conversationnel complet : VAD, STT, streaming LLM, TTS.
Returns
-------
bool
True si le tour a abouti, False si aucun signal vocal n'a ete detecte.
"""
self.isInterrupted = False
self.audioPipeline.resetInterrupt()
self._notifyState("LISTENING")
print("\n[1/3 - Ecoute] En attente de votre voix (Silero VAD 450ms)...")
audioArray = AudioProcessor.recordSpeechWithVad(
micId=self.micId,
vadDetector=self.vadDetector,
audioConfig=self.config.audio,
onAudioLevel=self.onAudioLevel,
stopSignal=lambda: not self.isRunning or self.isInterrupted,
)
turnSpeechEndTime = time.time()
if audioArray is None:
print("[Ecoute] Aucun signal vocal detecte (temps d'attente ecoule ou micro muet).")
self._notifyState("IDLE")
return False
self._notifyState("TRANSCRIBING")
t0Stt = time.time()
sttLang = getattr(self.config.stt, "language", "en")
transcribedText = self.sttEngine.transcribe(audioArray, language=sttLang)
sttDurationMs = (time.time() - t0Stt) * 1000.0
if not transcribedText or len(transcribedText.strip()) < 2:
print("[Transcription] Aucun texte intelligible detecte.")
self._notifyState("IDLE")
return False
print(f"[2/3 - Transcription] {transcribedText} (en {sttDurationMs:.1f}ms)")
if self.onTranscription:
self.onTranscription(transcribedText)
return self._generateAndSpeak(userText=transcribedText, speechEndTime=turnSpeechEndTime, sttDurationMs=sttDurationMs)
[docs]
def startSession(self, maxTurns: int = 10) -> None:
"""
Demarre la boucle de conversation interactive.
Parameters
----------
maxTurns : int, optional
Nombre maximal de tours de conversation (defaut: 10).
"""
self.isRunning = True
print("=== SESSION CONVERSATIONNELLE MIRZA DEMARREE ===")
print(f"STT : {self.config.stt.provider.upper()} | LLM : {self.config.llm.provider.upper()} | TTS : {self.config.tts.provider.upper()}")
print("Parlez librement. Pour arreter, appuyez sur Ctrl+C.\n")
turnCount = 0
try:
while self.isRunning and turnCount < maxTurns:
turnCount += 1
print(f"--- Tour {turnCount}/{maxTurns} ---")
success = self.processSingleTurn()
if not success:
time.sleep(0.4)
else:
time.sleep(0.2)
except KeyboardInterrupt:
print("\n[Session] Arret demande par l'utilisateur.")
finally:
self.isRunning = False
self._notifyState("IDLE")
print("=== FIN DE LA SESSION CONVERSATIONNELLE ===")
[docs]
def resetConversation(self) -> None:
"""Reinitialise l'historique, vide le cache KV et purge la memoire Metal."""
self.interrupt()
self.contextManager.clear()
self.conversationHistory = self.contextManager.history
self.lastInjectedRagContext = None
self.sessionRagHistory.clear()
self.llmEngine.estimatedKvTokens = 0
self.llmEngine.clearCache()
MemoryManager.purgeMemory()
self._notifyState("IDLE")
# Mise a zero des metriques KV
if self.onMetrics:
self.onMetrics({
"sttDurationMs": 0.0,
"ttftMs": 0.0,
"llmTotalTimeMs": 0.0,
"tokensPerSec": 0.0,
"tokenCount": 0,
"kvCacheTokens": 0,
"kvCacheMax": self.config.llm.maxKvCacheTokens,
"kvCachePercentage": 0.0,
"roundtripLatencyMs": 0.0,
})
if self.onActiveContextUpdated:
try:
self.onActiveContextUpdated(self.getActiveContextInfo())
except Exception as ctxErr:
logger.debug("Erreur broadcast onActiveContextUpdated: %s", ctxErr)
self.llmEngine.setReasoningBudget(0)
logger.info("Historique conversationnel et KV Cache reinitialises avec succes")
[docs]
def setThinkingMode(self, enabled: bool) -> None:
"""
Active ou desactive le mode reflexion au niveau du moteur LLM.
Parameters
----------
enabled : bool
True pour activer BrainBoost (CoT), False pour desactiver.
"""
if enabled and self.agentManager.isEnabled():
self.agentManager.setEnabled(False)
self._thinkingOverride = bool(enabled)
if enabled and self._reasoningBudget <= 0:
# Un bouton presse sans niveau choisi prend le defaut declare par
# le moteur, pas un nombre ecrit ici : 2048 etait ecrit en dur et
# ne suivait pas les paliers quand ils ont bouge.
self._reasoningBudget = int(
getattr(self.llmEngine, "defaultReasoningBudget", 1024)
)
self.llmEngine.setThinkingMode(enabled)
logger.info("Mode reflexion : %s", "ACTIVE" if enabled else "DESACTIVE")
[docs]
def setReasoningBudget(self, budget: int) -> None:
"""
Definit a chaud le budget de tokens alloue a la reflexion.
Parameters
----------
budget : int
Nombre de tokens alloues au raisonnement interne (0 a 2048).
"""
if budget > 0 and self.agentManager.isEnabled():
self.agentManager.setEnabled(False)
# Un budget nul *est* le choix de l'utilisateur de ne pas reflechir :
# cela enregistre le meme override qu'un bouton desactive.
self._thinkingOverride = budget > 0
self._reasoningBudget = max(0, budget)
self.llmEngine.setReasoningBudget(budget)
logger.info("Budget de reflexion configure a %d tokens", budget)
[docs]
def getActiveContextInfo(self) -> dict:
"""
Extrait et synthetise l'etat du contexte actif charge en memoire vive (KV).
Returns
-------
dict
Decompte des tokens par section, occupation du cache KV et historique.
"""
kvTokens = self.llmEngine.currentKvTokens or self.llmEngine.estimatedKvTokens
maxKv = getattr(self.config.llm, "maxKvCacheTokens", 16384) or 16384
return self.contextManager.getActiveContextInfo(
systemPrompt=self.config.prompt.systemPrompt,
lastRagContext=self.lastInjectedRagContext,
estimatedKvTokens=kvTokens,
maxKvCacheTokens=maxKv,
ragHistory=self.sessionRagHistory,
)
[docs]
def listMemories(
self,
query: str | None = None,
limit: int = 50,
offset: int = 0,
) -> list[dict]:
"""
Interroge la memoire episodique SQLite et retourne la liste des souvenirs.
Parameters
----------
query : str | None, optional
Terme textuel a rechercher dans les souvenirs, defaut: None.
limit : int, optional
Nombre maximum d'elements retournes, defaut: 50.
offset : int, optional
Index de pagination, defaut: 0.
Returns
-------
list[dict]
Les dictionnaires de souvenirs trouves.
"""
if not self.memoryStore:
return []
return self.memoryStore.listMemories(query=query, limit=limit, offset=offset)
[docs]
def deleteMemory(self, memoryId: int) -> bool:
"""
Supprime un souvenir de la base SQLite et retourne le statut.
Parameters
----------
memoryId : int
Identifiant numerique du souvenir a effacer.
Returns
-------
bool
True si le souvenir a ete supprime, False sinon.
"""
if not self.memoryStore:
return False
return self.memoryStore.deleteMemory(memoryId)
[docs]
def ingestConsultationNote(self, rawText: str) -> dict[str, Any]:
"""
Reformule une note brute ou une dictée de consultation médicale au format
standardisé de la patientèle, puis l'injecte dans la base de données épisodique.
N'utilise aucun TTS et n'émet aucun flux vocal.
Parameters
----------
rawText : str
Observations orales ou notes brutes saisies par le médecin.
Returns
-------
dict[str, Any]
Détails de la consultation insérée (success, memoryId, formattedNote, docId, etc.).
"""
if not self.memoryStore:
return {"success": False, "error": "memory_store_unavailable"}
rawText = (rawText or "").strip()
if not rawText:
return {"success": False, "error": "empty_input"}
todayStr = datetime.date.today().isoformat()
# 1. Tentative d'identification du patient et de son identifiant PAT dans la base
suggestedPatId = None
patientNom = ""
mNom = re.search(r"(?:patient[e]?\s+|m\.\s+|mme\s+)?([A-ZÀ-Ý][a-zà-ÿ]+(?:\s+[A-ZÀ-Ý][a-zà-ÿ]+)?)", rawText)
if mNom:
cand = mNom.group(1).strip()
if cand.lower() not in {"consultation", "dossier", "visite", "examen", "suivi", "notes", "note", "pour", "avec", "sous", "traité", "traitée"}:
patientNom = cand
if self.memoryStore:
try:
with self.memoryStore.backend.connection() as conn:
# Recherche d'un patient existant avec correspondance des tokens de nom
if patientNom:
cur = conn.execute(
"SELECT DISTINCT json_extract(metadata, '$.patient_id'), json_extract(metadata, '$.patient_nom') "
"FROM memories WHERE json_extract(metadata, '$.patient_nom') IS NOT NULL;"
)
cand_tokens = [t.lower() for t in re.split(r"[, -]+", patientNom) if len(t) > 2]
for pid, pnom in cur.fetchall():
if not pid or not pnom:
continue
pnom_tokens = [t.lower() for t in re.split(r"[, -]+", pnom) if len(t) > 2]
if cand_tokens and all(ct in pnom_tokens for ct in cand_tokens):
suggestedPatId = pid
patientNom = pnom
break
# Si nouveau patient ou non trouvé, allouer le prochain PAT-xxxx libre au-dessus de 2100
if not suggestedPatId:
cur_max = conn.execute(
"SELECT MAX(CAST(SUBSTR(json_extract(metadata, '$.patient_id'), 5) AS INTEGER)) "
"FROM memories WHERE json_extract(metadata, '$.patient_id') LIKE 'PAT-%';"
)
row_max = cur_max.fetchone()
max_num = row_max[0] if (row_max and row_max[0]) else 2199
next_num = max(2200, max_num + 1)
suggestedPatId = f"PAT-{next_num:04d}"
except Exception as e:
logger.debug("Erreur lookup patient existant: %s", e)
if not suggestedPatId:
suggestedPatId = "PAT-2200"
if not patientNom:
patientNom = "Patient"
# 2. Récupération du prompt de reformulation depuis le profil médical
medProfile = self.config.assistantProfiles.get("medical")
reformulatePrompt = ""
if medProfile and getattr(medProfile, "consultationPrompt", None):
reformulatePrompt = medProfile.consultationPrompt
if not reformulatePrompt:
reformulatePrompt = (
"Tu es un secrétaire médical et transcripteur clinique expert.\n"
"À partir des notes brutes ou de la dictée vocale du médecin, rédige la fiche de consultation clinique structurée, synthétique et professionnelle, respectant scrupuleusement la nomenclature des 700 dossiers patients du cabinet.\n\n"
"STRUCTURE EXACTE DE LA FICHE :\n"
"Patient : {patient_nom} ({patient_id}). Date : {date_today}. Type : consultation. Praticien : Dr. Lemaire.\n"
"Observations cliniques : Synthèse clinique claire (motif, antécédents, constantes TA / FC / poids, examen physique, auscultation).\n"
"Prescriptions : Traitements mentionnés avec posologie exacte (traitement de fond, nouvelle ordonnance, ou bilans complémentaires).\n"
"Actes médicaux : Codes des actes réalisés (ex: CS, T.A. ou CS, ECG001 ou CS).\n\n"
"RÈGLES STRICTES :\n"
"1. Toute thérapeutique, médicament ou posologie (ex: Thyroïdax 3000 mg 5 fois par jour) DOIT impérativement être inscrit dans 'Prescriptions :'. Ne JAMAIS laisser cette ligne vide ou avec un point. Si aucun traitement, écris 'Prescriptions : Aucune prescription'.\n"
"2. Remplis directement la date ({date_today}) et le praticien (Dr. Lemaire).\n"
"3. N'utilise JAMAIS de symboles '<' ou '>' ni de placeholders (<à compléter>, etc.)."
)
reformulatePrompt = reformulatePrompt.replace("{date_today}", todayStr)
reformulatePrompt = reformulatePrompt.replace("{today_date}", todayStr)
reformulatePrompt = reformulatePrompt.replace("{patient_id}", suggestedPatId)
reformulatePrompt = reformulatePrompt.replace("{suggested_id}", suggestedPatId)
reformulatePrompt = reformulatePrompt.replace("{patient_nom}", patientNom)
# 3. Exécution du LLM pour formater la note
try:
formattedPrompt = self.llmEngine.formatPrompt(
conversationHistory=[],
currentUserInput=f"Note clinique brute à formater :\n{rawText}",
systemPrompt=reformulatePrompt,
maxTurns=1,
)
responseTokens: list[str] = []
for token in self.llmEngine.streamResponse(formattedPrompt):
responseTokens.append(token)
formattedNote = "".join(responseTokens).strip()
formattedNote = re.sub(r"<think>[\s\S]*?</think>", "", formattedNote).strip()
formattedNote = re.sub(r"</?result>", "", formattedNote).strip()
except Exception as exc:
logger.warning("Erreur génération reformulation consultation : %s", exc)
formattedNote = rawText
if not formattedNote:
formattedNote = rawText
# 4. Auto-healing et assainissement strict des métadonnées
patMatch = re.search(r"Patient\s*:\s*([^(\n.]+?)(?:\s*\(([^)\n]+)\))?(?:\.|$|\s*Date)", formattedNote, re.IGNORECASE)
finalNom = patientNom
finalPatId = suggestedPatId
if patMatch:
foundNom = patMatch.group(1).strip()
foundId = (patMatch.group(2) or "").strip()
if foundNom and foundNom.lower() not in {"patient", "<nom, prénom>", "inconnu", "patient non précisé"}:
finalNom = foundNom
if foundId and "<" not in foundId and foundId.lower() not in {"id", "id_patient", "pat-xxxx"}:
finalPatId = foundId
dateMatch = re.search(r"Date\s*:\s*([0-9]{4}-[0-9]{2}-[0-9]{2})", formattedNote, re.IGNORECASE)
finalDate = dateMatch.group(1) if dateMatch else todayStr
pratMatch = re.search(r"Praticien\s*:\s*Dr\.?\s*([A-Za-zÀ-ÿ]+)", formattedNote, re.IGNORECASE)
finalPrat = "Dr. Lemaire"
if pratMatch:
pName = pratMatch.group(1).strip()
if pName.lower() not in {"nom", "en", "charge", "médecin", "praticien"}:
finalPrat = f"Dr. {pName}"
# Traitement des Prescriptions
prescMatch = re.search(r"Prescriptions\s*:\s*([^\n]+)", formattedNote, re.IGNORECASE)
prescContent = prescMatch.group(1).strip() if prescMatch else ""
prescClean = re.sub(r"<[^>]*>", "", prescContent).strip().rstrip(".").strip()
if not prescClean or prescClean.lower() in {"aucune", "aucun", "none", ""}:
extractedMeds: list[str] = []
medDosages = re.findall(
r"([A-Za-zÀ-ÿ0-9-]+\s+[0-9]+(?:\s*[.,]\s*[0-9]+)?\s*(?:mg|µg|g|ml)\s*(?:\([^)]+\))?)",
f"{rawText} {formattedNote}",
re.IGNORECASE,
)
for md in medDosages:
mClean = md.strip()
if mClean and mClean not in extractedMeds:
extractedMeds.append(mClean)
bilanMatch = re.search(r"(bilan\s+[^\n.,;]+)", rawText, re.IGNORECASE)
if bilanMatch:
bClean = bilanMatch.group(1).strip()
if bClean not in extractedMeds:
extractedMeds.append(bClean)
if extractedMeds:
prescClean = ", ".join(extractedMeds)
elif "aucune" in prescContent.lower():
prescClean = "Aucune prescription"
else:
prescClean = "Aucune prescription"
if prescClean and prescClean != "Aucune prescription":
prescClean = prescClean[0].upper() + prescClean[1:]
# Observations cliniques
obsMatch = re.search(r"Observations cliniques\s*:\s*([\s\S]+?)(?=\n\s*(?:Prescriptions|Actes)|$)", formattedNote, re.IGNORECASE)
finalObs = obsMatch.group(1).strip() if obsMatch else rawText
finalObs = re.sub(r"<([^>]+)>", r"\1", finalObs).strip()
# Actes médicaux
actesMatch = re.search(r"Actes médicaux\s*:\s*([^\n]+)", formattedNote, re.IGNORECASE)
finalActes = actesMatch.group(1).strip() if actesMatch else "CS"
finalActes = re.sub(r"<[^>]*>", "", finalActes).strip().rstrip(".").strip() or "CS"
# Reconstruction CANONIQUE standardisée
formattedNote = (
f"Patient : {finalNom} ({finalPatId}). Date : {finalDate}. Type : consultation. Praticien : {finalPrat}.\n\n"
f"Observations cliniques : {finalObs}\n\n"
f"Prescriptions : {prescClean}\n\n"
f"Actes médicaux : {finalActes}"
)
dateCompact = finalDate.replace("-", "")
salt = "".join(random.choices(string.digits, k=2))
docId = f"CONS-{dateCompact}-{salt}"
docTitle = f"Dossier médical patient {finalNom} ({finalPatId}) - consultation du {finalDate} [Réf: {docId}]"
obsSnippet = finalObs.split(".")[0].strip() if "." in finalObs else finalObs[:120]
summary = f"consultation {finalNom} {finalDate}: {obsSnippet[:120]}"
metaDict = {
"doc_id": docId,
"patient_id": finalPatId,
"patient_nom": finalNom,
"type_document": "consultation",
"date": finalDate,
"source": "consultation_directe",
}
metaJson = json.dumps(metaDict, ensure_ascii=False)
# 5. Enregistrement direct dans episodicMemory
memId = self.memoryStore.storeTurn(
userInput=docTitle,
assistantResponse=formattedNote,
summary=summary,
metadata=metaJson,
)
if self.onMemoryUpdated:
try:
self.onMemoryUpdated(self.getMemoryStatus())
except Exception as e:
logger.debug("onMemoryUpdated callback error: %s", e)
return {
"success": True,
"memoryId": memId,
"docId": docId,
"patientNom": finalNom,
"patientId": finalPatId,
"date": finalDate,
"formattedNote": formattedNote,
}
[docs]
def shutdown(self) -> None:
"""Arrete l'orchestrateur et libere les ressources allouees."""
self.isRunning = False
self.interrupt()
self.llmEngine.unload()
self.conversationHistory = []
MemoryManager.purgeMemory()
self._notifyState("SHUTDOWN")
print("[Systeme] Mirza arrete et ressources liberees.")