"""
Module de configuration centralise de l'assistant Mirza.
Fournit des dataclasses typees pour le parametrage audio, les modeles d'inference,
la synthese vocale, le prompt systeme, les profils d'assistant et la gestion de la memoire.
Supporte le chargement intelligent et la mise en cache depuis un fichier YAML (config.yaml).
"""
from __future__ import annotations
import os
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
import yaml
from mirza.core.prompts import (
ASSISTANT_AGENT_FINAL_ANSWER_PROMPT,
ASSISTANT_AGENT_INSTRUCTIONS_PROMPT,
ASSISTANT_AGENT_SYSTEM_PROMPT,
ASSISTANT_SYSTEM_PROMPT,
ASSISTANT_THINKING_FILLER_PHRASES,
CLINICAL_AGENT_FINAL_ANSWER_PROMPT,
CLINICAL_AGENT_INSTRUCTIONS_PROMPT,
CLINICAL_AGENT_SYSTEM_PROMPT,
CLINICAL_SYSTEM_PROMPT,
CLINICAL_THINKING_FILLER_PHRASES,
SYSTEM_PROMPT,
)
#: Plafond dur de la fenetre de contexte en tokens (prompt, RAG, historique et reponse).
SAFE_CONTEXT_TOKENS = 16000
[docs]
@dataclass
class AudioConfig:
"""Parametres acoustiques et detection vocale (Silero VAD)."""
sampleRate: int = 16000
chunkSamples: int = 512
silenceDurationThreshold: float = 0.85
speechThreshold: float = 0.45
preRollChunks: int = 16
initialWaitTimeout: float = 8.0
maxTotalDuration: float = 45.0
targetPeak: float = 0.95
[docs]
@dataclass
class SttConfig:
"""Configuration du moteur de transcription vocale (FastConformer Nemotron / Whisper)."""
provider: str = "nemotron"
sttModelRepo: str = "mlx-community/nemotron-3.5-asr-streaming-0.6b-8bit"
language: str = "fr"
[docs]
@dataclass
class LlmConfig:
"""Configuration du LLM et des parametres d'inference Metal GPU."""
provider: str = "mtplx"
llmModelRepo: str = "pooyakn73/Qwen3.8-9B-Distill-MTPLX-Q6G64"
prefillStepSize: int = 512
maxKvCacheTokens: int = SAFE_CONTEXT_TOKENS
kvBits: int = 8
bytesPerContextToken: int = 128_000
maxTokens: int = 1024
maxTokensAgent: int = 4096
maxTokensThinking: int = 5120
temperature: float = 0.5
temperatureAgent: float = 0.2
topP: float = 0.9
topK: int = 0
minP: float = 0.0
presencePenalty: float = 0.0
repetitionPenalty: float = 1.05
# Speculative Decoding (MTP)
enableSpeculative: bool = True
drafterModelRepo: str = "ulises-c/Qwen3.5-9B-MTP-8bit"
draftKind: str = "mtp"
draftBlockSize: int = 2
# Mode Reflexion (Chain-of-Thought / Thinking)
enableThinking: bool = False
reasoningBudget: int = 0
# MTPLX Server Settings
mtplxHost: str = "127.0.0.1"
mtplxPort: int = 8766
mtplxModelRepo: str = "pooyakn73/Qwen3.8-9B-Distill-MTPLX-Q6G64"
mtplxAutoStart: bool = True
mtplxReasoningEffort: str = "medium"
mtplxSessionId: str = ""
# API distante OpenRouter
openrouterModel: str = "anthropic/claude-sonnet-4.5"
# Serveur OptiQ
optiqHost: str = "127.0.0.1"
optiqPort: int = 8080
optiqAutoStart: bool = True
optiqMtp: bool = True
optiqMtpDepth: int = 2
[docs]
@dataclass
class TtsConfig:
"""Configuration du moteur de synthese vocale."""
provider: str = "kokoro"
voice: str = "ff_siwis"
voiceSpeedRate: str = "+10%"
voicePitch: str = "0Hz"
minClauseChars: int = 18
minCommaChars: int = 45
language: str = "fr-fr"
meloSpeed: float = 1.08
meloNoiseScale: float = 0.75
meloNoiseScaleW: float = 0.85
meloSdpRatio: float = 0.4
[docs]
@dataclass
class PromptConfig:
"""Configuration du prompt systeme et de l'historique."""
systemPrompt: str = SYSTEM_PROMPT
historyMaxTurns: int = 10
[docs]
@dataclass
class MemoryConfig:
"""
Configuration de la memoire episodique vectorielle (RAG SQLite-vec & Reranking FlashRank).
"""
enabled: bool = True
dbPath: str = "mirza/data/memory.db"
embeddingModel: str = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
rerankerModel: str = "ms-marco-TinyBERT-L-2-v2"
embeddingDim: int = 384
similarityThreshold: float = 0.65 # Distance cosinus max pour retention (ou score min reranker)
maxContextTokens: int = 900
topKRetrieval: int = 10
topKRerank: int = 4
autoArchive: bool = True
[docs]
@dataclass
class AgentPromptsConfig:
"""Gabarits de prompts pour le mode agent smolagents."""
systemPrompt: str = ""
instructionsPrompt: str = ""
finalAnswerPre: str = ""
finalAnswerPost: str = ""
[docs]
@dataclass
class AssistantProfile:
"""
Profil d'assistant (persona, domaine d'expertise, prompts, outils et mémoire).
"""
id: str
label: str
description: str = ""
systemPrompt: str = ""
fillerPhrases: tuple[str, ...] = field(default_factory=tuple)
memoryDbPath: str = "mirza/data/memory.db"
allowedTools: tuple[str, ...] | None = None
consultationPrompt: str = ""
agent: AgentPromptsConfig = field(default_factory=AgentPromptsConfig)
[docs]
def toDict(self) -> dict[str, Any]:
"""Convertit le profil en dictionnaire pour les API et le HUD."""
return {
"id": self.id,
"label": self.label,
"description": self.description,
"systemPrompt": self.systemPrompt,
"fillerPhrases": list(self.fillerPhrases),
"memoryDbPath": self.memoryDbPath,
"allowedTools": list(self.allowedTools) if self.allowedTools else None,
"consultationPrompt": self.consultationPrompt,
}
#: Profils d'assistant natifs intégrés par défaut
DEFAULT_ASSISTANT_PROFILES: dict[str, AssistantProfile] = {
"assistant": AssistantProfile(
id="assistant",
label="Assistant Général",
description="Assistant personnel polyvalent avec mémoire épisodique, recherche web et synthèse orale concise.",
systemPrompt=ASSISTANT_SYSTEM_PROMPT,
fillerPhrases=ASSISTANT_THINKING_FILLER_PHRASES,
memoryDbPath="mirza/data/memory_assistant.db",
allowedTools=(
"datetime_info",
"system_metrics",
"web_search",
"visit_webpage",
"recall_memory",
"save_memory",
),
agent=AgentPromptsConfig(
systemPrompt=ASSISTANT_AGENT_SYSTEM_PROMPT,
instructionsPrompt=ASSISTANT_AGENT_INSTRUCTIONS_PROMPT,
finalAnswerPre=ASSISTANT_AGENT_FINAL_ANSWER_PROMPT["pre_messages"],
finalAnswerPost=ASSISTANT_AGENT_FINAL_ANSWER_PROMPT["post_messages"],
),
),
"medical": AssistantProfile(
id="medical",
label="Assistant Médical & Clinique",
description="Assistant clinique expert pour dossiers patients, bilans et requêtes médicales.",
systemPrompt=CLINICAL_SYSTEM_PROMPT,
fillerPhrases=CLINICAL_THINKING_FILLER_PHRASES,
memoryDbPath="mirza/data/memory.db",
allowedTools=None, # Tous les outils permis
agent=AgentPromptsConfig(
systemPrompt=CLINICAL_AGENT_SYSTEM_PROMPT,
instructionsPrompt=CLINICAL_AGENT_INSTRUCTIONS_PROMPT,
finalAnswerPre=CLINICAL_AGENT_FINAL_ANSWER_PROMPT["pre_messages"],
finalAnswerPost=CLINICAL_AGENT_FINAL_ANSWER_PROMPT["post_messages"],
),
),
}
[docs]
@dataclass
class AppConfig:
"""
Configuration globale de l'application Mirza et des allocations materielles.
"""
vramLimitBytes: int = 16 * 1024 * 1024 * 1024
cacheLimit: int = 0
activeAssistantProfile: str = "assistant"
assistantProfiles: dict[str, AssistantProfile] = field(
default_factory=lambda: dict(DEFAULT_ASSISTANT_PROFILES)
)
audio: AudioConfig = field(default_factory=AudioConfig)
stt: SttConfig = field(default_factory=SttConfig)
llm: LlmConfig = field(default_factory=LlmConfig)
tts: TtsConfig = field(default_factory=TtsConfig)
prompt: PromptConfig = field(default_factory=PromptConfig)
memory: MemoryConfig = field(default_factory=MemoryConfig)
def __post_init__(self) -> None:
if not self.assistantProfiles:
self.assistantProfiles = dict(DEFAULT_ASSISTANT_PROFILES)
self.applyAssistantProfile(self.activeAssistantProfile)
[docs]
def getAssistantProfile(self, profileId: str | None = None) -> AssistantProfile:
"""Retourne le profil d'assistant demandé ou actif."""
targetId = profileId or self.activeAssistantProfile
if targetId in self.assistantProfiles:
return self.assistantProfiles[targetId]
if "assistant" in self.assistantProfiles:
return self.assistantProfiles["assistant"]
if self.assistantProfiles:
return next(iter(self.assistantProfiles.values()))
return DEFAULT_ASSISTANT_PROFILES["assistant"]
[docs]
def listAssistantProfiles(self) -> list[dict[str, Any]]:
"""Liste tous les profils d'assistant configurés."""
return [p.toDict() for p in self.assistantProfiles.values()]
[docs]
def applyAssistantProfile(self, profileId: str) -> None:
"""Applique les paramètres du profil d'assistant sélectionné."""
if profileId not in self.assistantProfiles:
return
self.activeAssistantProfile = profileId
prof = self.assistantProfiles[profileId]
if prof.systemPrompt:
self.prompt.systemPrompt = prof.systemPrompt
if prof.memoryDbPath:
self.memory.dbPath = prof.memoryDbPath
[docs]
@classmethod
def load(
cls,
configPath: str | Path | None = None,
profile: str | None = None,
reload: bool = False,
) -> AppConfig:
"""Charge la configuration depuis un fichier YAML avec cache en mémoire."""
return loadConfig(configPath=configPath, profile=profile, reload=reload)
_CACHED_APP_CONFIG: AppConfig | None = None
_CACHED_CONFIG_SOURCE: str | None = None
[docs]
def loadConfig(
configPath: str | Path | None = None,
profile: str | None = None,
reload: bool = False,
) -> AppConfig:
"""
Charge la configuration depuis un fichier YAML de façon intelligente.
Mise en cache en mémoire (singleton) : ne réouvre pas le fichier disque
à chaque appel ou tour de parole.
"""
global _CACHED_APP_CONFIG, _CACHED_CONFIG_SOURCE
if not reload and _CACHED_APP_CONFIG is not None and configPath is None and profile is None:
return _CACHED_APP_CONFIG
candidate_paths: list[Path] = []
if configPath:
candidate_paths.append(Path(configPath))
env_cfg = os.getenv("MIRZA_CONFIG")
if env_cfg:
candidate_paths.append(Path(env_cfg))
candidate_paths.extend([
Path("config.yaml"),
Path("config.yml"),
Path("mirza/config.yaml"),
])
resolved_path: Path | None = None
for p in candidate_paths:
if p.is_file():
resolved_path = p
break
raw_data: dict[str, Any] = {}
if resolved_path:
try:
with open(resolved_path, encoding="utf-8") as f:
raw_data = yaml.safe_load(f) or {}
_CACHED_CONFIG_SOURCE = str(resolved_path.resolve())
except Exception as exc:
import logging
logging.getLogger("mirza.config").warning(
"Impossible de charger le fichier YAML %s : %s. Valeurs par défaut utilisées.",
resolved_path,
exc,
)
cfg = AppConfig()
# Détermination du profil d'assistant actif
active_prof_id = (
profile
or os.getenv("MIRZA_PROFILE")
or raw_data.get("active_profile")
or raw_data.get("activeAssistantProfile")
)
# Profils d'assistant
if "profiles" in raw_data and isinstance(raw_data["profiles"], dict):
parsed_profiles: dict[str, AssistantProfile] = {}
for pid, pdata in raw_data["profiles"].items():
if not isinstance(pdata, dict):
continue
agent_data = pdata.get("agent", {}) or {}
fa_data = agent_data.get("final_answer_prompt", {}) or {}
if isinstance(fa_data, dict):
fa_pre = fa_data.get("pre_messages", "")
fa_post = fa_data.get("post_messages", "")
else:
fa_pre = str(fa_data)
fa_post = ""
allowed = pdata.get("allowed_tools")
allowed_tuple = tuple(allowed) if isinstance(allowed, (list, tuple)) else None
filler = pdata.get("filler_phrases", [])
filler_tuple = tuple(filler) if isinstance(filler, (list, tuple)) else ()
# Repli sur le gabarit par défaut si le prompt n'est pas spécifié dans le YAML
default_p = DEFAULT_ASSISTANT_PROFILES.get(pid)
sys_prompt = pdata.get("system_prompt")
if sys_prompt is None and default_p:
sys_prompt = default_p.systemPrompt
elif sys_prompt is None:
sys_prompt = ASSISTANT_SYSTEM_PROMPT
parsed_profiles[pid] = AssistantProfile(
id=pid,
label=pdata.get("label", pid.capitalize()),
description=pdata.get("description", ""),
systemPrompt=sys_prompt.strip(),
fillerPhrases=filler_tuple or (default_p.fillerPhrases if default_p else ()),
memoryDbPath=pdata.get("memory_db", f"mirza/data/memory_{pid}.db"),
allowedTools=allowed_tuple,
consultationPrompt=(
pdata.get("consultation_reformulate_prompt")
or pdata.get("consultation_prompt")
or (default_p.consultationPrompt if default_p else "")
).strip(),
agent=AgentPromptsConfig(
systemPrompt=agent_data.get("system_prompt", default_p.agent.systemPrompt if default_p else "").strip(),
instructionsPrompt=agent_data.get("instructions_prompt", default_p.agent.instructionsPrompt if default_p else "").strip(),
finalAnswerPre=fa_pre.strip() or (default_p.agent.finalAnswerPre if default_p else ""),
finalAnswerPost=fa_post.strip() or (default_p.agent.finalAnswerPost if default_p else ""),
),
)
if parsed_profiles:
cfg.assistantProfiles = parsed_profiles
# Matériel / VRAM
hw = raw_data.get("hardware", {})
if "vram_limit_gb" in hw:
cfg.vramLimitBytes = int(hw["vram_limit_gb"]) * 1024 * 1024 * 1024
if "cache_limit" in hw:
cfg.cacheLimit = int(hw["cache_limit"])
# Audio
a = raw_data.get("audio", {})
for k, v in a.items():
attr = "".join(word.capitalize() if i > 0 else word for i, word in enumerate(k.split("_")))
if hasattr(cfg.audio, attr):
setattr(cfg.audio, attr, type(getattr(cfg.audio, attr))(v))
# STT
s = raw_data.get("stt", {})
if "provider" in s:
cfg.stt.provider = s["provider"]
if "model_repo" in s:
cfg.stt.sttModelRepo = s["model_repo"]
if "language" in s:
cfg.stt.language = s["language"]
# LLM
llm_cfg = raw_data.get("llm", {})
if "provider" in llm_cfg:
cfg.llm.provider = llm_cfg["provider"]
if "model_repo" in llm_cfg:
cfg.llm.llmModelRepo = llm_cfg["model_repo"]
if "max_tokens" in llm_cfg:
cfg.llm.maxTokens = int(llm_cfg["max_tokens"])
if "max_tokens_agent" in llm_cfg:
cfg.llm.maxTokensAgent = int(llm_cfg["max_tokens_agent"])
if "max_tokens_thinking" in llm_cfg:
cfg.llm.maxTokensThinking = int(llm_cfg["max_tokens_thinking"])
if "temperature" in llm_cfg:
cfg.llm.temperature = float(llm_cfg["temperature"])
if "temperature_agent" in llm_cfg:
cfg.llm.temperatureAgent = float(llm_cfg["temperature_agent"])
if "top_p" in llm_cfg:
cfg.llm.topP = float(llm_cfg["top_p"])
if "top_k" in llm_cfg:
cfg.llm.topK = int(llm_cfg["top_k"])
if "enable_speculative" in llm_cfg:
cfg.llm.enableSpeculative = bool(llm_cfg["enable_speculative"])
if "drafter_repo" in llm_cfg:
cfg.llm.drafterModelRepo = llm_cfg["drafter_repo"]
if "enable_thinking" in llm_cfg:
cfg.llm.enableThinking = bool(llm_cfg["enable_thinking"])
if "reasoning_budget" in llm_cfg:
cfg.llm.reasoningBudget = int(llm_cfg["reasoning_budget"])
if "mtplx_host" in llm_cfg:
cfg.llm.mtplxHost = llm_cfg["mtplx_host"]
if "mtplx_port" in llm_cfg:
cfg.llm.mtplxPort = int(llm_cfg["mtplx_port"])
# TTS
t = raw_data.get("tts", {})
if "provider" in t:
cfg.tts.provider = t["provider"]
if "voice" in t:
cfg.tts.voice = t["voice"]
if "language" in t:
cfg.tts.language = t["language"]
if "voice_speed_rate" in t:
cfg.tts.voiceSpeedRate = t["voice_speed_rate"]
if "voice_pitch" in t:
cfg.tts.voicePitch = t["voice_pitch"]
# Memory
m = raw_data.get("memory", {})
if "enabled" in m:
cfg.memory.enabled = bool(m["enabled"])
if "embedding_model" in m:
cfg.memory.embeddingModel = m["embedding_model"]
if "reranker_model" in m:
cfg.memory.rerankerModel = m["reranker_model"]
if "similarity_threshold" in m:
cfg.memory.similarityThreshold = float(m["similarity_threshold"])
if "max_context_tokens" in m:
cfg.memory.maxContextTokens = int(m["max_context_tokens"])
# Application du profil actif
if active_prof_id:
cfg.applyAssistantProfile(active_prof_id)
_CACHED_APP_CONFIG = cfg
return cfg
[docs]
def getConfig() -> AppConfig:
"""Retourne l'instance de configuration mise en cache en mémoire (singleton)."""
if _CACHED_APP_CONFIG is not None:
return _CACHED_APP_CONFIG
return loadConfig()