"""
Gestionnaire central du mode agentique de Mirza, base sur smolagents (CodeAgent).
Gere l'activation et la desactivation a chaud du mode et des outils individuels.
"""
from __future__ import annotations
import re
import threading
from collections.abc import Callable
from typing import Any
from smolagents import CodeAgent
from ...core.logger import getMirzaLogger
from ...core.prompts import (
AGENT_FINAL_ANSWER_PROMPT,
AGENT_INSTRUCTIONS_PROMPT,
AGENT_SYSTEM_PROMPT,
)
from ...core.promptTokens import stripMemoTag
from ...core.textCleaner import TextCleaner
from .mirzaModelAdapter import MirzaModelAdapter
from .tools import (
clinical_records_stats,
datetime_info,
execute_clinical_sql,
recall_memory,
save_memory,
system_metrics,
visit_webpage,
web_search,
)
logger = getMirzaLogger("mirza.agent.manager")
[docs]
class AgentManager:
"""
Gestionnaire central du mode agentique, base sur smolagents (CodeAgent).
Parameters
----------
llmEngine : BaseLlm
Moteur d'inference LLM auquel l'agent est attache.
broadcaster : EventBroadcaster, optional
Diffuseur WebSocket notifiant le frontend des etats d'outils.
Defaut : None.
"""
[docs]
def __init__(self, llmEngine: Any, broadcaster: Any = None) -> None:
self.llmEngine = llmEngine
self.broadcaster = broadcaster
self.enabled = False
# Restriction d'outils propre au profil actif. None = tous permis.
# Distincte de l'interrupteur par outil : celui-ci est un choix de
# l'utilisateur dans la session, celle-la une consequence du modele.
self.allowedTools: frozenset[str] | None = None
self.lock = threading.Lock()
self.toolsRegistry: dict[str, dict[str, Any]] = {
"datetime_info": {
"id": "datetime_info",
"name": "Horloge & Calendrier",
"description": "Date, heure courante, jour de la semaine et fuseau horaire, localement ou pour tout pays/ville (ex: Ouganda, Tokyo, New York).",
"tool": datetime_info,
"enabled": True,
},
"system_metrics": {
"id": "system_metrics",
"name": "Telemetrie Materielle",
"description": "Surveillance en temps reel du CPU, de la RAM et de la VRAM Metal du Mac.",
"tool": system_metrics,
"enabled": True,
},
"web_search": {
"id": "web_search",
"name": "Recherche Web",
"description": "Recherche en direct sur Internet via DuckDuckGo.",
"tool": web_search,
"enabled": True,
},
"visit_webpage": {
"id": "visit_webpage",
"name": "Exploration de Pages Web",
"description": "Extraction et lecture du texte principal d'un lien URL.",
"tool": visit_webpage,
"enabled": True,
},
"recall_memory": {
"id": "recall_memory",
"name": "Memoire Episodique & Passe",
"description": "Recherche dans les souvenirs et faits passes de Mirza (rendez-vous, discussions, choix anterieurs).",
"tool": recall_memory,
"enabled": True,
},
"save_memory": {
"id": "save_memory",
"name": "Memorisation & RAG",
"description": "Enregistrement permanent d'une information, fait cle ou decouverte dans la memoire episodique / RAG de Mirza.",
"tool": save_memory,
"enabled": True,
},
"clinical_records_stats": {
"id": "clinical_records_stats",
"name": "Statistiques Cliniques & Comptage",
"description": "Analyse quantitative, denombrement, statistiques et recherche exhaustive sur les dossiers medicaux (ex: combien de patients, plus de X ordonnances, cas par pathologie).",
"tool": clinical_records_stats,
"enabled": True,
},
"execute_clinical_sql": {
"id": "execute_clinical_sql",
"name": "Requete SQL Clinique",
"description": "Execution de requetes SQL SELECT analytiques avancees sur la table 'memories' de la base clinique.",
"tool": execute_clinical_sql,
"enabled": True,
},
}
# La configuration est l'unique source de verite : un second plafond ici
# la remplacerait silencieusement (T-080).
is_remote = getattr(self.llmEngine, "isRemote", False) or getattr(getattr(self.llmEngine, "config", None), "provider", "") == "openrouter"
default_max_tokens = 1024 if is_remote else 512
agent_max_tokens = getattr(self.llmEngine.config, "maxTokensAgent", default_max_tokens)
# 0.2 et non 0.4 : l'agent doit tenir une forme stricte, un appel
# d'outil suivi de guillemets triples puis d'une balise fermee. Un
# modele petit s'en ecarte d'autant plus que le tirage est large, et il
# perd alors la balise : le travail n'est plus archive. La valeur est
# reglee par profil, le 4B etant le plus expose.
agent_temperature = float(
getattr(self.llmEngine.config, "temperatureAgent", 0.2)
)
self.modelAdapter = MirzaModelAdapter(
llmEngine=self.llmEngine,
max_tokens=agent_max_tokens,
temperature=agent_temperature,
)
dummy_agent = CodeAgent(tools=[], model=None)
self.promptTemplates = dict(dummy_agent.prompt_templates)
self.promptTemplates["system_prompt"] = AGENT_SYSTEM_PROMPT
self.promptTemplates["final_answer"] = dict(AGENT_FINAL_ANSWER_PROMPT)
self.instructionsPrompt = AGENT_INSTRUCTIONS_PROMPT
[docs]
def setAssistantProfilePrompts(
self,
systemPrompt: str | None = None,
instructionsPrompt: str | None = None,
finalAnswerPrompt: dict[str, str] | None = None,
allowedTools: tuple[str, ...] | None = None,
) -> None:
"""
Met a jour les gabarits de prompts et les restrictions d'outils de l'agent.
Parameters
----------
systemPrompt : str, optional
Nouveau prompt systeme de l'agent.
instructionsPrompt : str, optional
Nouvelles instructions d'execution de l'agent.
finalAnswerPrompt : dict, optional
Nouveau gabarit final_answer.
allowedTools : tuple[str, ...] | None, optional
Liste d'outils autorises ou None pour tout autoriser.
"""
with self.lock:
if systemPrompt:
self.promptTemplates["system_prompt"] = systemPrompt
if instructionsPrompt:
self.instructionsPrompt = instructionsPrompt
if finalAnswerPrompt:
self.promptTemplates["final_answer"] = dict(finalAnswerPrompt)
self.allowedTools = frozenset(allowedTools) if allowedTools else None
logger.info("Prompts et outils de l'agent mis a jour pour le profil actif.")
[docs]
def setLlmEngine(self, llmEngine) -> None:
"""
Rattache l'agent au moteur qui vient d'etre charge.
L'adaptateur avait capture le moteur du profil de demarrage : apres un
changement de modele, il pointait sur un daemon arrete, et les outils
echouaient alors que le dialogue normal fonctionnait.
Parameters
----------
llmEngine : object
Le moteur LLM nouvellement cree.
"""
self.llmEngine = llmEngine
self.modelAdapter.llmEngine = llmEngine
# Le budget de l'agent suit la configuration du nouveau moteur : pas de
# plafond secondaire qui masquerait le choix.
is_remote = getattr(llmEngine, "isRemote", False) or getattr(getattr(llmEngine, "config", None), "provider", "") == "openrouter"
default_max_tokens = 1024 if is_remote else 512
self.modelAdapter.max_tokens = getattr(
getattr(llmEngine, "config", None), "maxTokensAgent", default_max_tokens
)
logger.info("Agent rattache au moteur %s", type(llmEngine).__name__)
[docs]
def isEnabled(self) -> bool:
"""
Indique si le mode agentique est actif.
Returns
-------
bool
True si l'agent est active, False sinon.
"""
return self.enabled
[docs]
def setEnabled(self, enabled: bool) -> None:
"""
Active ou desactive le mode agentique.
Parameters
----------
enabled : bool
Nouvel etat du mode agentique.
"""
with self.lock:
self.enabled = bool(enabled)
if self.broadcaster:
self.broadcaster.broadcastEvent("agentStatus", self.getStatus())
logger.info("Mode agentique %s", "ACTIVE" if self.enabled else "DESACTIVE")
[docs]
def getStatus(self) -> dict[str, Any]:
"""
Retourne l'etat complet de l'agent et la liste de ses outils.
Returns
-------
dict[str, Any]
Dictionnaire decrivant le statut et les outils disponibles.
"""
tools_list = [
{
"id": data["id"],
"name": data["name"],
"description": data["description"],
"enabled": data["enabled"],
}
for data in self.toolsRegistry.values()
]
return {
"enabled": self.enabled,
"tools": tools_list,
}
def _getActiveTools(self) -> list[Any]:
"""
Liste les instances d'outils actuellement actives.
Returns
-------
list[Any]
Fonctions d'outils activees.
"""
return [
data["tool"] for data in self.toolsRegistry.values()
if data["enabled"]
and (
self.allowedTools is None
or data.get("id") in self.allowedTools
)
]
[docs]
def run(
self,
userPrompt: str,
conversationHistory: list[dict[str, str]] | None = None,
onThoughtChunk: Callable[[str], None] | None = None,
onTokenChunk: Callable[[str], None] | None = None,
) -> str:
"""
Execute la requete utilisateur via smolagents CodeAgent.
Parameters
----------
userPrompt : str
Instruction ou requete de l'utilisateur.
conversationHistory : list[dict[str, str]] | None, optional
Historique conversationnel recent.
onThoughtChunk : Callable[[str], None] | None, optional
Callback pour streamer les etapes de reflexion de l'agent.
onTokenChunk : Callable[[str], None] | None, optional
Callback pour streamer les fragments de reponse.
Returns
-------
str
Reponse finale textuelle de l'agent, balise ``result`` comprise
pour les documents.
"""
active_tools = self._getActiveTools()
task_prompt = userPrompt
if conversationHistory:
recent_turns = [
turn for turn in conversationHistory[-2:]
if turn.get("content")
]
if recent_turns:
dialogue_lines: list[str] = []
for turn in recent_turns:
role_label = "Utilisateur" if turn.get("role") == "user" else "Mirza"
snippet = stripMemoTag(turn.get("content", ""))
if not snippet:
continue
if len(snippet) > 150:
snippet = snippet[:150] + "..."
dialogue_lines.append(f"- {role_label} : {snippet}")
task_prompt = (
"Contexte de la conversation précédente :\n"
+ "\n".join(dialogue_lines)
+ f"\n\nDemande actuelle de l'utilisateur à résoudre avec les outils en français :\n{userPrompt}"
)
agent_instructions = self.instructionsPrompt
agent = CodeAgent(
tools=active_tools,
model=self.modelAdapter,
prompt_templates=self.promptTemplates,
instructions=agent_instructions,
# Deux unites de budget pour deux passes utiles : la premiere appelle
# l'outil, la seconde ecrit la reponse. A max_steps=1 l'appel d'outil
# consomme le seul budget et l'agent termine sur "Reached max steps",
# sans reponse.
max_steps=2,
verbosity_level=1,
additional_authorized_imports=["json", "sqlite3", "math", "re"],
)
if onThoughtChunk:
onThoughtChunk("[Agent Mirza] Analyse de la requete avec les outils...\n")
original_step = agent.step
def _logged_step(*args: Any, **kwargs: Any) -> Any:
step_log = original_step(*args, **kwargs)
try:
action_text = getattr(step_log, "action_output", None) or getattr(step_log, "tool_calls", None)
thought = getattr(step_log, "thought", None)
msg = ""
if thought:
msg += f"Raisonnement : {thought}\n"
if action_text:
msg += f"Action outil : {action_text}\n"
if msg and onThoughtChunk:
onThoughtChunk(f"[Agent] {msg}")
except Exception as exc:
logger.debug("Erreur logged_step: %s", exc)
return step_log
agent.step = _logged_step
self.modelAdapter.on_token_chunk = onTokenChunk
try:
result = agent.run(task_prompt)
final_text = str(result).strip() if result is not None else ""
if not final_text:
final_text = "L'operation s'est achevee sans retour textuel explicite."
final_text = TextCleaner.stripEmojis(final_text)
if final_text.startswith("[Source 1]") or "[Source 1]" in final_text[:60]:
cleaned_parts = []
for p in final_text.split("\n\n"):
p_clean = re.sub(r"^\[Source\s*\d+\]\s*", "", p.strip())
if p_clean and not any(p_clean.startswith(ign) for ign in ["View the profiles", "Join Facebook", "Voir vos relations"]):
cleaned_parts.append(p_clean)
if cleaned_parts:
final_text = " ".join(cleaned_parts[:2])
return final_text.strip()
except Exception as e:
err_msg = f"Erreur lors de l'execution de l'agent : {e}"
logger.error(err_msg, exc_info=True)
if onThoughtChunk:
onThoughtChunk(f"\n[Erreur Agent] {err_msg}\n")
return f"Desole, une erreur est survenue lors de l'execution de la tache agentique : {e}"
finally:
self.modelAdapter.on_token_chunk = None