main_rag.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""
Lyra RAG - Point d'entree principal.
Version 2.0 avec architecture RAG hybride et dual models.
Usage:
python main_rag.py # Mode interactif
python main_rag.py --vocal # Mode vocal
python main_rag.py -p # Mode performance
python main_rag.py "demarre preprod-09" # One-shot
python main_rag.py "status des VMs" -v # One-shot verbose
python main_rag.py -y "allume la lumiere" # One-shot auto-confirme
Architecture:
USER INPUT -> RAG -> EPHAISTOS -> [clarification?] -> LYRA -> HESTIA -> LYRA -> OUTPUT
"""
from __future__ import annotations
import argparse
import sys
import threading
from pathlib import Path
from typing import TYPE_CHECKING, Optional
if TYPE_CHECKING: # annotations seulement : l'import reel reste paresseux (dependances lourdes)
from lyra.core.pipeline import Pipeline
# Ajouter le package lyra au path
sys.path.insert(0, str(Path(__file__).parent))
import yaml
from lyra.core.config import RAGConfig
from lyra.core.constants import ( # noqa: F401
DANGEROUS_TOOLS,
PERFORMANCE_TOOLS,
is_dangerous_tool,
is_dangerous_with_annotations,
)
# QueryType et PipelineResult importes depuis types.py (pas de torch/sentence_transformers)
# Pipeline importe dans main() apres le banner pour ne pas bloquer le demarrage
from lyra.core.types import PipelineResult, QueryType
from lyra.core.workflows.context import ExecContext, UIContext
from lyra.core.workflows.vm_clone_exec import handle_vm_clone_with_stop
from lyra.core.workflows.vm_snapshot_exec import handle_snapshot_restore_with_safety
from lyra.hestia.background_tasks import BackgroundTaskManager
from lyra.models.model_manager import ModelManager
from lyra.rag.session_memory import SessionMemory
from lyra.utils.error_log import (
is_execution_error as _is_execution_error_fn,
)
from lyra.utils.error_log import (
lyra_error_message as _lyra_error_message_fn,
)
from lyra.utils.error_log import (
write_error_log,
)
# Import du module UI existant pour la compatibilite
from modules import ui
from modules.n8n import send_discord_notification
def _lyra_error_message(log_path: Path) -> str:
return _lyra_error_message_fn(log_path)
def _is_execution_error(exec_result) -> bool:
return _is_execution_error_fn(exec_result)
def _handle_error_log(tool_name: str, arguments: dict, exec_result, vocal: bool = False, voice=None) -> None:
"""Ecrit le log d'erreur et affiche le message LYRA friendly."""
log_path = write_error_log(tool_name, arguments, exec_result)
err_msg = _lyra_error_message(log_path)
print(f"\n{ui.Colors.RED}{err_msg}{ui.Colors.RESET}")
if vocal and voice:
voice.speak("ca a pas marche, j'ai logué l'erreur.")
# Helper pour accéder au pipeline réel (V2 ou Enhanced)
def get_actual_pipeline(pipeline):
"""Retourne le pipeline V2 sous-jacent si EnhancedPipeline, sinon le pipeline lui-même."""
return pipeline._pipeline_v2 if hasattr(pipeline, '_pipeline_v2') else pipeline
# Mapping nom symbolique -> code ANSI pour UIContext.colored
_COLOR_MAP: dict[str, str] = {
"yellow": ui.Colors.YELLOW,
"cyan": ui.Colors.CYAN,
"green": ui.Colors.GREEN,
"red": ui.Colors.RED,
"magenta": ui.Colors.MAGENTA,
"blue": ui.Colors.BLUE,
"bold": ui.Colors.BOLD,
"dim": ui.Colors.DIM,
"white": ui.Colors.WHITE,
}
def _colored_by_name(text: str, color: str) -> str:
"""Wrapper ui.colored acceptant les noms symboliques ('yellow') ou les codes ANSI directs."""
return ui.colored(text, _COLOR_MAP.get(color.lower(), color))
def _try_fast_path_rules(query: str):
"""Delegue a lyra.core.fastpath (partage avec le demon)."""
from lyra.core.fastpath import try_fast_path_rules
return try_fast_path_rules(query)
# DANGEROUS_TOOLS et PERFORMANCE_TOOLS importes depuis lyra.core.constants
# Outils avec operations async (longues) qui necessitent notification Discord
ASYNC_TOOLS = [
"vm_clone", "vm_clone_system",
"backup_create", "backup_restore",
"vm_snapshot", # Restauration snapshot
"vm_export", # Export + virt-sysprep (10-20 min)
"vm_import", # Import + copie disque (5-10 min)
]
# Variable globale pour webhook Discord
DISCORD_WEBHOOK_URL = None
def load_config(config_path: str = "config.yaml") -> RAGConfig:
"""Charge la configuration."""
path = Path(config_path)
if not path.exists():
ui.print_error(f"Config file not found: {config_path}")
sys.exit(1)
return RAGConfig.from_yaml(path)
def check_models(config: RAGConfig) -> bool:
"""Verifie que les modeles sont disponibles."""
ui.print_info("Verification des modeles Ollama...")
manager = ModelManager(config)
available = manager.check_models_available()
if not available.get("ephaistos"):
ui.print_error(f"Modele EPHAISTOS non trouve: {config.models.ephaistos.name}")
print(f" Installez-le avec: ollama pull {config.models.ephaistos.name}")
return False
if not available.get("lyra"):
ui.print_error(f"Modele LYRA non trouve: {config.models.lyra.name}")
print(f" Installez-le avec: ollama pull {config.models.lyra.name}")
return False
ui.print_success(f"EPHAISTOS: {config.models.ephaistos.name}")
ui.print_success(f"LYRA: {config.models.lyra.name}")
return True
def check_devices(cfg: dict) -> None:
"""Ping rapide des peripheriques domotiques au demarrage (non-bloquant).
Affiche un avertissement si un device est injoignable, mais ne bloque pas.
"""
import socket
import urllib.request
devices: list[tuple[str, str]] = [] # (label, check_type:address)
tv_host = cfg.get("tv", {}).get("host")
if tv_host:
devices.append((f"TV [{tv_host}]", f"http:{tv_host}:1925"))
hue_ip = cfg.get("hue", {}).get("bridge_ip")
if hue_ip:
devices.append((f"Hue [{hue_ip}]", f"http:{hue_ip}:80"))
denon_host = cfg.get("denon", {}).get("host")
denon_port = cfg.get("denon", {}).get("port", 23)
if denon_host:
devices.append((f"Denon [{denon_host}]", f"tcp:{denon_host}:{denon_port}"))
if not devices:
return
for label, check in devices:
try:
proto, host, port_str = check.split(":")
port = int(port_str)
if proto == "tcp":
s = socket.create_connection((host, port), timeout=1.5)
s.close()
else:
urllib.request.urlopen(f"http://{host}:{port}/", timeout=1.5)
ui.print_success(f" {label}")
except Exception:
ui.print_warning(f" {label} (injoignable)")
def print_banner(mode="v2", ephaistos_model="Qwen", lyra_model="Llama"):
"""Affiche la banniere avec le mode actif et les vrais noms de modeles.
Args:
mode: "enhanced" (RAG Enhanced), "legacy" (V1), ou "v2" (standard)
ephaistos_model: Nom du modele EPHAISTOS (depuis config)
lyra_model: Nom du modele LYRA (depuis config)
"""
if mode == "enhanced":
title = "LYRA RAG ENHANCED"
subtitle = "3-Tier RAG + Slang + Context + Feedback"
elif mode == "legacy":
title = "LYRA Legacy"
subtitle = "Assistant DevOps Classique"
else: # v2
title = "LYRA RAG"
subtitle = "Assistant DevOps avec RAG + Dual Models"
def _model_line(prefix: str, model: str, role: str) -> str:
"""Construit une ligne modele qui tient en 49 chars."""
full = f"{prefix}{model} - {role}"
if len(full) <= 49:
return full
# Essayer sans le role
short = f"{prefix}{model}"
if len(short) <= 49:
return short
# Tronquer le nom du modele
max_model = 49 - len(prefix)
return f"{prefix}{model[:max_model - 1]}~"
eph_line = _model_line("EPHAISTOS: ", ephaistos_model, "Analyse & Validation")
lyra_line = _model_line("LYRA: ", lyra_model, "Dialogue & Personnalite")
banner = f"""
╔═══════════════════════════════════════════════════╗
║ {title:^49s} ║
║ {subtitle:^49s} ║
╠═══════════════════════════════════════════════════╣
║ {eph_line:<49s} ║
║ {lyra_line:<49s} ║
║ HESTIA - Execution MCP ║
╚═══════════════════════════════════════════════════╝
"""
print(ui.colored(banner, ui.Colors.CYAN))
def should_skip_confirmation(tool_name: str, mode: str, dangerous: Optional[bool] = None) -> bool:
"""Determine si on peut skip la confirmation.
Args:
tool_name: Nom de l'outil
mode: Mode actif (default ou performance)
Returns:
True si on peut executer sans confirmation
"""
# Toujours confirmer les outils dangereux (nom prefixe ou non). `dangerous` :
# verdict deja calcule par l'appelant, annotation destructiveHint comprise.
if dangerous if dangerous is not None else is_dangerous_tool(tool_name):
return False
# Mode performance: skip pour les outils domotique
if mode == "performance" and tool_name in PERFORMANCE_TOOLS:
return True
return False
def _get_lyra_rag_step_message(step: str, data: dict) -> Optional[str]:
"""Genere un message LYRA pour une etape du RAG 3-tier (M1 verbose).
Delegue a LyraVoice.get_rag_step_message() qui gere les templates.
Args:
step: Nom de l'etape
data: Donnees de l'etape (score, server, tool, ...)
Returns:
Message a afficher ou None
"""
try:
from lyra.models.lyra_voice import LyraVoice
return LyraVoice.get_rag_step_message(step, data)
except Exception:
return None
def _handle_correction_intelligente(
pipeline,
tool_name: str,
arguments: dict,
rag_results: list = None
) -> Optional[tuple]:
"""Gere la correction intelligente quand l'user dit que la reponse est mauvaise (M3).
Demande a l'user ce qui ne va pas (serveur / outil / params),
affiche les alternatives disponibles, et retourne le nouveau choix.
Args:
pipeline: Pipeline RAG
tool_name: Nom de l'outil propose
arguments: Arguments proposes
rag_results: Resultats RAG pour afficher les alternatives
Returns:
(new_tool_name, new_arguments) ou None si annule
"""
actual_pipeline = get_actual_pipeline(pipeline)
server_prefix = tool_name.split(".")[0] if "." in tool_name else ""
short_tool = tool_name.split(".")[-1] if "." in tool_name else tool_name
ui.print_lyra_tag("c'est quoi qui va pas ?")
print()
print(" 1. Le serveur MCP (ex: FEDORA au lieu de HUE)")
print(" 2. L'outil (ex: vm_start au lieu de vm_clone)")
print(" 3. Les parametres (ex: mauvais nom de VM)")
print()
choice = input(" [1/2/3] ou [n] pour annuler : ").strip().lower()
if choice in ("n", "non", ""):
return None
if choice == "1":
# Afficher les serveurs disponibles
try:
servers = list(actual_pipeline._hestia.mcp_manager.clients.keys())
except Exception:
servers = ["FEDORA", "HUE", "TV", "CATT", "DENON"]
print()
print(ui.colored("Serveurs disponibles :", ui.Colors.YELLOW))
for i, s in enumerate(servers, 1):
marker = " <-- actuel" if s.lower() == server_prefix.lower() else ""
print(f" {i}. {s}{marker}")
print()
srv_choice = input(" Numero du bon serveur (ou nom) : ").strip()
if srv_choice.isdigit() and 1 <= int(srv_choice) <= len(servers):
new_server = servers[int(srv_choice) - 1]
elif srv_choice:
new_server = srv_choice.upper()
else:
return None
# Relancer le pipeline avec le serveur force
ui.print_lyra_tag(f"ok je cherche dans {new_server}...")
# Retourner None pour relancer via l'input utilisateur avec contexte
return None
elif choice == "2":
# Afficher les outils du meme serveur
print()
print(ui.colored(f"Outils disponibles sur {server_prefix or 'tous serveurs'} :", ui.Colors.YELLOW))
try:
# Recuperer les outils du serveur depuis HESTIA
all_tools = []
mcp_mgr = actual_pipeline._hestia.mcp_manager
for srv_name, client in mcp_mgr.clients.items():
if not server_prefix or srv_name.lower() == server_prefix.lower():
if hasattr(client, 'tools') and client.tools:
for t in client.tools:
all_tools.append((srv_name, t.get('name', '')))
if not all_tools:
# Fallback: afficher les outils du resultat RAG si disponible
print(" (liste non disponible)")
return None
# Afficher au max 12 outils pour lisibilite
display_tools = all_tools[:12]
for i, (srv, t) in enumerate(display_tools, 1):
short = t.split(".")[-1] if "." in t else t
marker = " <-- actuel" if t == tool_name or short == short_tool else ""
print(f" {i:2}. {short}{marker}")
if len(all_tools) > 12:
print(f" ... et {len(all_tools) - 12} autres")
except Exception:
print(" (liste non disponible, tape le nom directement)")
print()
tool_choice = input(" Numero ou nom de l'outil : ").strip()
new_tool = None
if tool_choice.isdigit():
idx = int(tool_choice) - 1
if 0 <= idx < len(all_tools):
srv, new_tool = all_tools[idx]
new_tool = f"{srv}.{new_tool}" if "." not in new_tool else new_tool
elif tool_choice:
new_tool = tool_choice
if new_tool:
ui.print_lyra_tag(f"ok, je garde les memes params pour {new_tool}. On y va ?")
confirm = input(" [O/n] : ").strip().lower()
if confirm not in ("n", "non"):
return (new_tool, arguments)
return None
elif choice == "3":
# Afficher les params de l'outil actuel
print()
print(ui.colored(f"Parametres de {short_tool} :", ui.Colors.YELLOW))
# Params actuels
print()
print(" Valeurs actuelles :")
for key, value in arguments.items():
print(f" {key} = {value}")
# Essayer de recuperer les params depuis les specs RAG
print()
print(" Modifie les valeurs (entree vide = garder) :")
new_arguments = {}
for key, value in arguments.items():
new_val = input(f" {key} [{value}] : ").strip()
new_arguments[key] = new_val if new_val else value
ui.print_lyra_tag(f"ok, {short_tool} avec les nouveaux params. On y va ?")
for k, v in new_arguments.items():
print(f" {k} = {v}")
print()
confirm = input(" [O/n] : ").strip().lower()
if confirm not in ("n", "non"):
return (tool_name, new_arguments)
return None
return None
def _generate_context_confirmation_message(result, tool_name: str, arguments: dict) -> str:
"""Génère un message de confirmation expliquant le contexte injecté.
Args:
result: Résultat du pipeline avec contexte injecté
tool_name: Nom de l'outil MCP
arguments: Arguments de l'outil
Returns:
str: Message de confirmation LYRA friendly
"""
# Extraire contexte du enriched_query
# Format: "[ctx: last_mcp=fedora.vm_start, last_server=FEDORA, last_vm=preprod-09]"
enriched_query = result.enriched_query or ""
context_parts = {}
if "[ctx:" in enriched_query:
# Extraire la partie contexte
ctx_str = enriched_query.split("[ctx:")[1].split("]")[0]
# Parser les paires key=value
for part in ctx_str.split(","):
if "=" in part:
key, value = part.strip().split("=", 1)
context_parts[key] = value
# Construire message selon contexte
last_mcp = context_parts.get("last_mcp", "")
last_vm = context_parts.get("last_vm", "")
# Action formatée
action_desc = tool_name.split(".")[-1].replace("_", " ")
# Construire message contextuel
msg_parts = []
if last_vm:
msg_parts.append(f"sur la VM **{last_vm}**")
if last_mcp:
last_action = last_mcp.split(".")[-1].replace("_", " ")
msg_parts.append(f"(dernier outil: {last_action})")
if msg_parts:
context_desc = " ".join(msg_parts)
message = (
f"J'ai détecté une ambiguïté. "
f"D'après le contexte {context_desc}, "
f"je vais exécuter : **{action_desc}**. "
f"C'est bien ça ?"
)
else:
# Fallback si pas de contexte clair
message = (
f"J'ai détecté une ambiguïté. "
f"Je propose d'exécuter : **{action_desc}**. "
f"Confirmes-tu ?"
)
return message
def handle_action(
pipeline: Pipeline,
result: PipelineResult,
mode: str,
vocal: bool = False,
voice=None,
task_manager: Optional[BackgroundTaskManager] = None,
webhook_url: Optional[str] = None
) -> str:
"""Gere une action avec confirmation adaptee au niveau de confiance (M2/M3).
Args:
pipeline: Instance du pipeline RAG
result: Resultat du pipeline avec tool_call
mode: Mode actif
vocal: Mode vocal actif
voice: Instance VoiceInterface
task_manager: Gestionnaire de taches en arriere-plan (optionnel)
webhook_url: URL webhook Discord (optionnel)
Returns:
Message de reponse
"""
tool_name = result.tool_call["name"]
arguments = result.tool_call["arguments"]
# Cas special: clone avec arret de VM / restauration de snapshot
if tool_name == "vm_clone_with_stop" or (
"vm_snapshot" in tool_name and arguments.get("action") == "revert"
):
actual = get_actual_pipeline(pipeline)
exec_ctx = ExecContext(
ui=UIContext(
print_info=ui.print_info,
print_success=ui.print_success,
print_error=ui.print_error,
print_warning=ui.print_warning,
print_tool_result=ui.print_tool_result,
colored=_colored_by_name,
confirm_action=ui.confirm_action,
ask_input=input,
println=print,
),
execute_action=actual.execute_action,
hestia_execute=actual._hestia.execute,
vocal=vocal,
voice=voice,
notify_discord=_send_discord_if_async,
)
if tool_name == "vm_clone_with_stop":
return handle_vm_clone_with_stop(arguments, exec_ctx)
return handle_snapshot_restore_with_safety(arguments, exec_ctx)
# Context Injector: Confirmation explicite si contexte injecte (ambiguite detectee)
context_injected = hasattr(result, 'context_injected') and result.context_injected
requires_confirmation = hasattr(result, 'requires_confirmation') and result.requires_confirmation
if context_injected and hasattr(result, 'enriched_query'):
context_msg = _generate_context_confirmation_message(result, tool_name, arguments)
print(f"\n{ui.Colors.YELLOW}[Context Injector]{ui.Colors.RESET}")
print(f"{context_msg}\n")
if vocal and voice:
voice.speak(context_msg)
# Mode performance: execution rapide si autorise ET pas de contexte injecte
dangerous = is_dangerous_with_annotations(
tool_name, getattr(get_actual_pipeline(pipeline), "_hestia", None))
if should_skip_confirmation(tool_name, mode, dangerous) and not requires_confirmation:
ui.print_info(f"Execution rapide: {tool_name}")
exec_result = get_actual_pipeline(pipeline).execute_action(tool_name, arguments)
has_error = _is_execution_error(exec_result)
if has_error:
_handle_error_log(tool_name, arguments, exec_result, vocal, voice)
ui.print_quick_result(tool_name, success=not has_error, use_beep=True)
_send_discord_if_async(tool_name, arguments, exec_result)
return exec_result.response
# --- M2 : Confirmation adaptee selon niveau de confiance ---
confidence_level = getattr(result, 'confidence_level', 'high') or 'high'
short_tool = tool_name.split(".")[-1].replace("_", " ") if tool_name else "?"
target = arguments.get("vm_name") or arguments.get("source_vm") or ""
# Afficher l'action proposee (toujours)
ui.print_tool_call(tool_name, arguments)
if confidence_level == "high":
# HIGH (80-100%): confirmation courte et directe
if target:
prompt_msg = f"{short_tool} sur {target}. C'est bon ?"
else:
prompt_msg = f"je vais {short_tool}. C'est bon ?"
ui.print_lyra_tag(prompt_msg)
if vocal and voice:
voice.speak(prompt_msg)
elif confidence_level == "medium":
# MEDIUM (50-80%): verif etat MCP + confirmation du bon serveur
server_prefix = tool_name.split(".")[0].upper() if "." in tool_name else "?"
if target:
status_hint = ""
# Tentative de recuperation de l'etat si VM
if "vm" in tool_name.lower() and target:
try:
actual_pl = get_actual_pipeline(pipeline)
status_result = actual_pl.execute_action("fedora.vm_status", {"vm_name": target})
if status_result.execution_result and status_result.execution_result.success:
raw = status_result.execution_result.result or ""
# Extraire etat court depuis le resultat
if "running" in raw.lower() or "en cours" in raw.lower():
status_hint = " [etat: demarre]"
elif "shut off" in raw.lower() or "arret" in raw.lower():
status_hint = " [etat: arretee]"
else:
status_hint = ""
except Exception:
status_hint = ""
prompt_msg = (
f"je pense {short_tool} sur {target}{status_hint}. "
f"C'est bien via {server_prefix} ? Confirme ?"
)
else:
prompt_msg = f"je pense {short_tool} via {server_prefix}. C'est bien ca ?"
ui.print_lyra_tag(prompt_msg)
if vocal and voice:
voice.speak(prompt_msg)
else:
# LOW (0-50%): LYRA exprime son incertitude
rag_results = getattr(result, '_rag_candidates', [])
if rag_results and len(rag_results) >= 2:
opt1 = rag_results[0].get('metadata', {}).get('tool_name', short_tool) if rag_results else short_tool
opt2 = rag_results[1].get('metadata', {}).get('tool_name', '?') if len(rag_results) > 1 else '?'
opt1 = opt1.split(".")[-1].replace("_", " ")
opt2 = opt2.split(".")[-1].replace("_", " ")
prompt_msg = (
f"franchement je suis pas super sure... "
f"tu parles de {opt1} ou de {opt2} ? "
f"Je propose {short_tool} mais dis-moi si c'est pas ca."
)
else:
prompt_msg = (
f"je suis pas certaine, j'ai propose {short_tool} "
f"mais ca me semble ambigu. Tu confirmes ?"
)
ui.print_lyra_tag(prompt_msg)
if vocal and voice:
voice.speak(prompt_msg)
# Demander confirmation standard
confirmed = ui.confirm_action(
tool_name,
arguments,
vocal_mode=vocal,
voice=voice
)
# --- M3 : Correction intelligente si "modifier" ---
if confirmed == "modify":
correction = _handle_correction_intelligente(pipeline, tool_name, arguments)
if correction is None:
ui.print_warning("Action annulee.")
return "Action annulee."
new_tool_name, new_arguments = correction
tool_name = new_tool_name
arguments = new_arguments
# Re-confirmer avec les nouvelles valeurs
ui.print_tool_call(tool_name, arguments)
confirmed = ui.confirm_action(
tool_name,
arguments,
vocal_mode=vocal,
voice=voice
)
if not confirmed or confirmed == "modify":
ui.print_warning("Action annulee.")
return "Action annulee."
# Cas: Utilisateur annule
if not confirmed:
ui.print_warning("Action annulee.")
return "Action annulee."
# Verifier si operation async (longue)
# EnhancedPipeline wrappe Pipeline V2, accéder via _pipeline_v2
hestia = pipeline._pipeline_v2._hestia if hasattr(pipeline, '_pipeline_v2') else pipeline._hestia
if task_manager and hestia.is_async_tool(tool_name):
async_info = hestia.get_async_info(tool_name)
# Validation concurrence: verifier si VM deja utilisee
vm_name = arguments.get("vm_name") or arguments.get("source_vm")
if vm_name:
existing_task = task_manager.get_vm_in_use(vm_name)
if existing_task:
# Afficher warning
from datetime import datetime
print()
print(ui.colored(f"⚠️ La VM {vm_name} est deja utilisee par:", ui.Colors.YELLOW + ui.Colors.BOLD))
print(f" - Tache: {existing_task.description}")
elapsed = (datetime.now() - existing_task.started_at).total_seconds()
print(f" - En cours depuis: {int(elapsed)}s")
print()
# Demander confirmation
print("Options:")
print(" 1. Continuer quand meme (risque de conflit)")
print(" 2. Annuler")
print()
choice = input("Choix ? [1/2] ").strip()
if choice != "1":
ui.print_warning("Operation annulee.")
return "Operation annulee (VM en cours d'utilisation)."
# Utilisateur prend le risque
print()
ui.print_warning("⚠️ Attention: Operations concurrentes sur la meme VM!")
print()
# Generer message friendly de LYRA
lyra_inst = pipeline._pipeline_v2._lyra if hasattr(pipeline, '_pipeline_v2') else pipeline._lyra
friendly_msg = lyra_inst.generate_async_message(
tool_name=tool_name,
estimated_time=async_info["estimated_time"],
description=async_info["description"]
)
print()
print(friendly_msg)
print()
# Lancer en subprocess
task_id = task_manager.launch_task(
tool_name=tool_name,
arguments=arguments,
description=async_info["description"],
estimated_time=async_info["estimated_time"],
webhook_url=webhook_url
)
msg = f"Operation lancee en arriere-plan (task: {task_id[-6:]})"
ui.print_success(msg)
return msg
# Executer via HESTIA (synchrone)
ui.print_info(f"Execution de {tool_name}...")
exec_result = get_actual_pipeline(pipeline).execute_action(tool_name, arguments)
has_error = _is_execution_error(exec_result)
if exec_result.executed:
ui.print_tool_result(exec_result.response, success=not has_error,
raw_error=exec_result.error if has_error else None)
else:
ui.print_tool_result(exec_result.response, success=False,
raw_error=exec_result.error)
has_error = True
if has_error:
_handle_error_log(tool_name, arguments, exec_result, vocal, voice)
# Envoyer notification Discord si operation async
_send_discord_if_async(tool_name, arguments, exec_result)
return exec_result.response
def _send_discord_if_async(tool_name: str, arguments: dict, exec_result) -> bool:
"""Envoie une notification Discord si l'outil est async.
Args:
tool_name: Nom de l'outil execute
arguments: Arguments de l'outil
exec_result: Resultat d'execution
Returns:
True si notification envoyee
"""
global DISCORD_WEBHOOK_URL
# Extraire le nom court (sans prefixe serveur)
short_name = tool_name.split(".")[-1] if "." in tool_name else tool_name
# Verifier si outil async
if short_name not in ASYNC_TOOLS or not DISCORD_WEBHOOK_URL:
return False
# Exception: vm_snapshot list n'est pas async (opération rapide de lecture)
if short_name == "vm_snapshot" and arguments.get("action") == "list":
return False
# Construire le titre et les fields
title_map = {
"vm_clone": "🔄 Clone VM",
"vm_clone_system": "🔄 Clone System VM",
"backup_create": "💾 Backup Create",
"backup_restore": "📦 Backup Restore",
"vm_snapshot": "📸 Snapshot VM"
}
title = title_map.get(short_name, f"✨ {short_name}")
is_success = exec_result.execution_result and exec_result.execution_result.success if hasattr(exec_result, 'execution_result') else False
color = 3066993 if is_success else 15158332 # Vert ou rouge
fields = [
{"name": "Outil", "value": short_name, "inline": True},
{"name": "Status", "value": "✅ OK" if is_success else "❌ ERREUR", "inline": True},
]
# Ajouter les arguments pertinents
for key, value in arguments.items():
if value and key not in ("start", "verbose"): # Ignorer les flags
fields.append({"name": key, "value": str(value), "inline": True})
# Description (resultat ou erreur)
if is_success:
# Tronquer le resultat si trop long
description = exec_result.response[:500]
if len(exec_result.response) > 500:
description += "..."
else:
description = f"❌ {exec_result.error or 'Erreur inconnue'}"
# Envoyer
return send_discord_notification(
webhook_url=DISCORD_WEBHOOK_URL,
title=title,
description=f"```\n{description}\n```" if is_success else description,
fields=fields,
color=color,
footer="Lyra RAG v2.0 - DevOps Assistant"
)
def run_one_shot(
request: str,
pipeline,
mode: str,
verbose: bool = False,
yes: bool = False,
task_manager=None,
webhook_url: str = "",
fast_analysis=None,
) -> int:
"""Execute une requete en mode one-shot et retourne le code de sortie.
Args:
request: La requete utilisateur
pipeline: Instance du pipeline RAG
mode: Mode actif (default ou performance)
verbose: Afficher les etapes intermediaires du pipeline
yes: Auto-confirmer les actions (sauf outils dangereux)
task_manager: Gestionnaire de taches en arriere-plan
webhook_url: URL webhook Discord
Returns:
Code de sortie: 0=ok, 1=erreur, 2=annule, 3=args manquants
"""
def on_rag_step_verbose(step: str, data: dict):
score = data.get("score", 0.0)
if score > 0.80:
color = ui.Colors.GREEN
elif score >= 0.50:
color = ui.Colors.YELLOW
else:
color = ui.Colors.RED
if step == "before_llm":
level = data.get("confidence_level", "medium")
tool = data.get("tool", "?").split(".")[-1].replace("_", " ")
score_str = f"{score:.2f}" if score else "?"
print(f" {color}[rag] {tool} score={score_str} ({level}){ui.Colors.RESET}", flush=True)
else:
tool = data.get("tool", "") or data.get("server", "")
if tool:
score_str = f"{score:.2f}" if score else "?"
print(f" {color}[rag] {step}: {tool} score={score_str}{ui.Colors.RESET}", flush=True)
def on_progress_verbose(step: str, message: str):
if step == "acknowledgement":
print(f" {ui.Colors.CYAN}[lyra] {message}{ui.Colors.RESET}", flush=True)
# Traiter la requete via le pipeline
# Fast path : analyse pre-calculee par regles, skip IntentClassifier + RAG + EPHAISTOS
if fast_analysis is not None and hasattr(pipeline, 'process_precomputed'):
result = pipeline.process_precomputed(request, fast_analysis)
else:
# rag_step_callback seulement pour EnhancedPipeline (Pipeline V2 ne l'accepte pas)
process_kwargs = {"callback": on_progress_verbose if verbose else None}
if hasattr(pipeline, '_pipeline_v2'):
process_kwargs["rag_step_callback"] = on_rag_step_verbose if verbose else None
result = pipeline.process(request, **process_kwargs)
# Afficher le type d'intention si verbose
if verbose:
qtype = result.query_type.value if hasattr(result.query_type, 'value') else str(result.query_type)
print(f" {ui.Colors.CYAN}[intent] {qtype}{ui.Colors.RESET}")
for warning in getattr(result, "warnings", []):
ui.print_warning(warning)
# Cas 1: Question de connaissance -> reponse directe
if result.query_type == QueryType.KNOWLEDGE:
ui.print_lyra(result.response)
return 0
# Cas 2: Args manquants -> impossible en one-shot
if result.pending_args:
ui.print_lyra_tag(result.response)
ui.print_warning(f"One-shot: arguments manquants ({', '.join(result.pending_args)})")
ui.print_warning("Relancez avec la requete complete ou utilisez le mode interactif.")
return 3
# Cas 3: Action prete -> confirmation et execution
if result.tool_call:
tool_name = result.tool_call["name"]
arguments = result.tool_call["arguments"]
if verbose:
print(f" {ui.Colors.CYAN}[action] {tool_name}({arguments}){ui.Colors.RESET}")
# Afficher la reponse de LYRA
ui.print_lyra(result.response)
# Determiner si confirmation necessaire
# Nom prefixe ("fedora.vm_destroy") : "in DANGEROUS_TOOLS" etait toujours
# faux et -y executait vm_destroy sans confirmation (audit 2026-09-19).
is_dangerous = is_dangerous_with_annotations(
tool_name, getattr(get_actual_pipeline(pipeline), "_hestia", None))
skip_confirm = should_skip_confirmation(tool_name, mode, is_dangerous)
if not skip_confirm and not (yes and not is_dangerous):
# Afficher l'action et demander confirmation
ui.print_tool_call(tool_name, arguments)
if is_dangerous:
print(
f"\n{ui.Colors.RED}ATTENTION action dangereuse: {tool_name}. "
f"Confirmer ? [o/N]{ui.Colors.RESET} ",
end="", flush=True
)
else:
short = tool_name.split(".")[-1].replace("_", " ")
tag = f'{ui.Colors.BG_ORANGE}{ui.Colors.FG_DARK} Lyra {ui.Colors.RESET}'
print(f'\n{tag} {short}. C\'est bon ? [O/n] ', end="", flush=True)
try:
answer = input().strip().lower()
except (EOFError, KeyboardInterrupt):
print()
ui.print_warning("Annule.")
return 2
confirmed = answer in ("o", "oui") if is_dangerous else answer in ("", "o", "oui")
if not confirmed:
ui.print_warning("Action annulee.")
return 2
elif verbose:
ui.print_info(f"Execution auto: {tool_name}")
# Verifier si operation async (longue)
hestia = pipeline._pipeline_v2._hestia if hasattr(pipeline, '_pipeline_v2') else pipeline._hestia
if task_manager and hestia.is_async_tool(tool_name):
async_info = hestia.get_async_info(tool_name)
task_id = task_manager.launch_task(
tool_name=tool_name,
arguments=arguments,
description=async_info["description"],
estimated_time=async_info["estimated_time"],
webhook_url=webhook_url,
)
ui.print_success(f"Operation lancee en arriere-plan (task: {task_id[-6:]})")
return 0
# Executer via HESTIA (synchrone)
# skip_lyra_format=True en fast path : evite l'appel LLM format_result
if verbose:
ui.print_info(f"Execution: {tool_name}...")
exec_result = get_actual_pipeline(pipeline).execute_action(
tool_name, arguments,
skip_lyra_format=(fast_analysis is not None)
)
has_error = _is_execution_error(exec_result)
if exec_result.executed:
ui.print_tool_result(exec_result.response, success=not has_error,
raw_error=exec_result.error if has_error else None)
else:
ui.print_tool_result(exec_result.response, success=False,
raw_error=exec_result.error)
has_error = True
if has_error:
_handle_error_log(tool_name, arguments, exec_result)
return 1
_send_discord_if_async(tool_name, arguments, exec_result)
return 0
# Cas 4: Pas de match -> reponse directe
ui.print_lyra(result.response)
return 0
def main():
"""Point d'entree principal."""
global DISCORD_WEBHOOK_URL
parser = argparse.ArgumentParser(
description="Lyra RAG - Assistant DevOps",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=(
"Exemples:\n"
" python main_rag.py # Mode interactif\n"
" python main_rag.py \"demarre preprod-09\" # One-shot\n"
" python main_rag.py \"status des VMs\" -v # One-shot verbose\n"
" python main_rag.py -y \"allume la lumiere\" # One-shot auto-confirme\n"
" python main_rag.py -p \"status vms\" -v # Performance + verbose\n"
)
)
parser.add_argument("request", nargs="?", default=None, metavar="REQUETE",
help="Requete one-shot (mode interactif si absent)")
parser.add_argument("--vocal", action="store_true", help="Mode vocal (STT/TTS)")
parser.add_argument("-p", "--performance", action="store_true", help="Mode performance")
parser.add_argument("--config", "-c", default="config.yaml", help="Fichier de config")
parser.add_argument("--check", action="store_true", help="Verifier les modeles et quitter")
parser.add_argument("--rag-enhanced", action="store_true", help="Activer RAG Enhanced")
parser.add_argument("--debug", "-d", action="store_true", help="Mode debug pipeline Enhanced")
parser.add_argument("--verbose", "-v", action="store_true",
help="Verbose one-shot: affiche les etapes du pipeline")
parser.add_argument("--yes", "-y", action="store_true",
help="Auto-confirmer les actions (sauf outils dangereux)")
args = parser.parse_args()
# Mode one-shot: requete passee en argument
one_shot = args.request is not None
# Charger la config (avant le bandeau pour afficher les vrais noms de modeles)
config = load_config(args.config)
# Fast path one-shot : detection IMMEDIATE avant check_models et init.
# Si une regle statique matche -> on skipera check_models + SentenceTransformer + ChromaDB + LLMs.
_fast_analysis = None
if one_shot:
_fast_analysis = _try_fast_path_rules(args.request)
if _fast_analysis is not None:
tool_short = _fast_analysis.tool.split(".")[-1].replace("_", " ") if _fast_analysis.tool else "?"
print(f" -> {tool_short}", flush=True)
# Bandeau : compact en one-shot, complet en interactif
if one_shot:
print(f"{ui.Colors.CYAN}LYRA{ui.Colors.RESET} {args.request}", flush=True)
else:
banner_mode = "enhanced" if args.rag_enhanced else "v2"
print_banner(
mode=banner_mode,
ephaistos_model=config.models.ephaistos.name,
lyra_model=config.models.lyra.name,
)
# Import lourd apres le banner : sentence_transformers + torch (~1-2s)
from lyra.core.pipeline import Pipeline # noqa: E402
# Charger le webhook Discord depuis config.yaml
global DISCORD_WEBHOOK_URL
try:
with open(args.config) as f:
cfg = yaml.safe_load(f)
from lyra.core.config import discord_webhook
DISCORD_WEBHOOK_URL = discord_webhook(args.config) # config.yaml, sinon secrets.yaml
if DISCORD_WEBHOOK_URL:
ui.print_info("Discord notifications activees")
except Exception as e:
cfg = {}
ui.print_warning(f"Config non chargee: {e}")
# Health check devices (interactif uniquement, non-bloquant)
if not one_shot:
ui.print_info("Peripheriques domotiques :")
check_devices(cfg)
# Verifier les modeles en one-shot uniquement (interactif: fait dans le thread bg)
if one_shot and _fast_analysis is None:
if not check_models(config):
sys.exit(1)
if args.check:
ui.print_success("Verification terminee.")
sys.exit(0)
# Mode : -p force performance, sinon reglage persiste (~/.lyra/settings.json)
from lyra.core.settings import UserSettings
user_settings = UserSettings()
mode = "performance" if args.performance else user_settings.active_mode()
# Interface vocale (optionnel)
vocal = args.vocal
voice = None
if vocal:
try:
from modules.audio import AudioConfig, VoiceInterface
with open("config.yaml") as f:
cfg = yaml.safe_load(f)
audio_cfg = cfg.get("audio", {})
stt_cfg = cfg.get("stt", {})
# Section tts de config.yaml + overrides utilisateur (/setting)
tts_cfg = user_settings.merged_tts(cfg.get("tts", {}))
audio_config = AudioConfig(
sample_rate=audio_cfg.get("sample_rate", 16000),
channels=audio_cfg.get("channels", 1),
silence_threshold=audio_cfg.get("silence_threshold", 0.01),
silence_duration=audio_cfg.get("silence_duration", 1.0),
input_device=audio_cfg.get("input_device"),
output_device=audio_cfg.get("output_device")
)
ui.print_info("Chargement du modele STT (Whisper)...")
voice = VoiceInterface(
stt_model=stt_cfg.get("model", "base"),
tts_model=f"models/{tts_cfg.get('model', 'fr_FR-upmc-medium')}.onnx",
language=stt_cfg.get("language", "fr"),
device=stt_cfg.get("device", "cuda"),
audio_config=audio_config,
tts_speaker_id=int(tts_cfg.get("speaker_id", 0)),
tts_length_scale=float(tts_cfg.get("length_scale", 1.0))
)
ui.print_success("Interface vocale prete")
except Exception as e:
ui.print_warning(f"Mode vocal indisponible: {e}")
vocal = False
voice = None
# Initialiser le pipeline
if args.rag_enhanced:
if not one_shot:
ui.print_info("Pipeline RAG Enhanced...")
from lyra.rag_enhanced import EnhancedPipeline
from lyra.rag_enhanced.config import RAGEnhancedConfig
# Charger la config RAG Enhanced depuis YAML
with open(args.config) as f:
cfg = yaml.safe_load(f)
enhanced_config = RAGEnhancedConfig.from_dict(cfg.get("rag_enhanced", {}))
pipeline = EnhancedPipeline(
config=config,
enhanced_config=enhanced_config,
enabled=enhanced_config.enabled,
tts_mode=vocal
)
else:
if not one_shot:
ui.print_info("Pipeline RAG V2...")
pipeline = Pipeline(config, tts_mode=vocal) # tts_mode=True si vocal
# Initialisation en arriere-plan avec progression visible
_init_done = threading.Event()
_checks_done = threading.Event() # set apres check_models (prints finis, barre peut s'afficher)
_init_error = [None]
def _bg_init():
try:
if _fast_analysis is not None:
_checks_done.set()
pipeline.initialize_fast()
else:
# En mode interactif: verifier les modeles ici (non-bloquant pour l'UI)
if not one_shot and not check_models(config):
_init_error[0] = Exception("Modele Ollama introuvable (voir messages ci-dessus)")
_checks_done.set()
return
_checks_done.set() # check_models a fini de printer — la barre peut apparaitre
pipeline.initialize()
# Warm-up Ollama en mode interactif : charge EPHAISTOS + LYRA en VRAM
# pendant que l'utilisateur lit le banner (premiere requete sans cold-load)
if not one_shot:
pipeline.preload_models()
except Exception as e:
_init_error[0] = e
_checks_done.set()
finally:
_init_done.set()
threading.Thread(target=_bg_init, daemon=True).start()
# one-shot: attendre la fin de l'init avant tout
if one_shot:
if _fast_analysis is not None:
print(" connexion MCP...", end="", flush=True)
else:
print(" chargement...", end="", flush=True)
_init_done.wait()
print(" pret", flush=True)
if _init_error[0]:
ui.print_error(f"Erreur initialisation: {_init_error[0]}")
sys.exit(1)
if hasattr(pipeline, '_pipeline_v2'):
servers = list(pipeline._pipeline_v2._hestia.mcp_manager.clients.keys())
else:
servers = list(pipeline._hestia.mcp_manager.clients.keys())
if args.verbose:
ui.print_success(f"MCP: {', '.join(servers)} ({len(servers)} serveurs)")
_servers_shown = False # pour afficher les serveurs au premier tour interactif
# Session memory
session = SessionMemory(max_turns=config.session.max_turns)
# Background task manager
task_manager = BackgroundTaskManager()
# Restaurer les taches des sessions precedentes
_restored = task_manager.restore_from_registry()
if _restored > 0:
print()
ui.print_info(f"{_restored} tache(s) en cours detectee(s) depuis la session precedente.")
# Mode one-shot: executer la requete et quitter
if one_shot:
sys.exit(run_one_shot(
request=args.request,
pipeline=pipeline,
mode=mode,
verbose=args.verbose,
yes=args.yes,
task_manager=task_manager,
webhook_url=DISCORD_WEBHOOK_URL,
fast_analysis=_fast_analysis,
))
# Mode interactif: juste une ligne avant le prompt
if args.debug and args.rag_enhanced:
ui.print_warning("Mode DEBUG actif")
if vocal:
ui.print_info("Mode vocal actif")
# Callback pour feedback progressif (Phase 4 - LYRA Immersive)
def on_progress(step: str, message: str):
"""Callback pour affichage progressif de LYRA.
Args:
step: Type de step ("acknowledgement", "progress", "result")
message: Message à afficher
"""
if step == "acknowledgement":
# Afficher l'acknowledgement immédiat en cyan (bleu clair)
print(f"{ui.Colors.CYAN}{message}{ui.Colors.RESET}")
# TTS immédiat si mode vocal
if vocal and voice:
voice.speak(message)
# Callback verbose RAG (M1) - messages LYRA correles au score
def on_rag_step(step: str, data: dict):
"""Callback appele a chaque etape du pipeline RAG.
Affiche un message LYRA adapte au score intermediaire.
Gere les etapes RAG (registry_done, capabilities_done) et pre-LLM (before_llm).
Args:
step: Nom de l'etape
data: Donnees de l'etape (score, server, tool, confidence_level...)
"""
if step == "rule_matched":
# Regle statique matchee : RAG skipee, message direct
import random as _random
tool = data.get("tool_short", data.get("tool", "?").split(".")[-1].replace("_", " "))
server = data.get("server", "?")
msg = _random.choice([
f"{server}: {tool}",
f"direct -> {tool}",
f"j'ai ca: {server} -> {tool}",
])
ui.print_lyra_tag(msg)
if vocal and voice:
voice.speak(msg)
return
if step == "before_llm":
# Message avant l'appel EPHAISTOS
level = data.get("confidence_level", "medium")
tool = data.get("tool", "?").split(".")[-1].replace("_", " ")
if level == "high":
msg = f"EPHAISTOS verifie {tool}..."
elif level == "medium":
msg = f"EPHAISTOS analyse {tool}..."
else:
msg = "EPHAISTOS cherche parmi les options..."
ui.print_lyra_tag(msg)
if vocal and voice:
voice.speak(msg)
return
# Etapes RAG (registry_done, capabilities_done, etc.)
msg = _get_lyra_rag_step_message(step, data)
if msg:
ui.print_lyra_tag(msg)
if vocal and voice:
voice.speak(msg)
# Menu /setting : machine a etats geree hors pipeline (les reponses "1",
# "2"... ne doivent pas passer par le RAG). Les callbacks hot-swap ne sont
# branches qu'en mode vocal ; sinon les reglages s'appliquent a la
# prochaine session vocale.
from lyra.core.settings_menu import SettingsCallbacks, SettingsMenu
def _set_mode(new_mode: str):
nonlocal mode
mode = new_mode
settings_menu = SettingsMenu(
settings=user_settings,
models_dir=Path(__file__).parent / "models",
callbacks=SettingsCallbacks(
apply_voice=(lambda p, sid: voice.tts.set_voice(p, sid)) if voice else None,
apply_speed=(lambda ls: voice.tts.set_speed(ls)) if voice else None,
preview=(lambda text: voice.speak(text)) if voice else None,
get_mode=lambda: mode,
set_mode=_set_mode,
),
)
# Boucle principale
while True:
try:
# Afficher les notifications de succes (les erreurs restent dans le bandeau)
completed_notifs = [n for n in task_manager.get_completed_notifications() if n.get("success")]
if completed_notifs:
ui.print_completed_task_notifications(completed_notifs)
# Nettoyer les taches completees anciennes (>5min)
task_manager.cleanup_completed(max_age_seconds=300)
# Input (vocal ou texte)
# live_input affiche le bandeau et le rafraichit toutes les secondes
if vocal and voice:
# Attendre la fin de l'init avant d'ecouter (evite pipeline.process sur init echouee)
if not _servers_shown:
if not _init_done.is_set():
_init_done.wait()
if _init_error[0]:
ui.print_error(f"Erreur initialisation: {_init_error[0]}")
break
# En mode vocal, afficher le bandeau separement (pas de live refresh)
active_tasks = task_manager.get_active_tasks()
if active_tasks:
ui.print_background_tasks(active_tasks, task_manager)
ui.print_info("Ecoute...")
user_input = voice.listen()
if user_input:
print(f">>> {user_input}")
else:
# Premier tour: attendre que check_models ait fini de printer
# avant de dessiner la barre (evite corruption de la barre)
if not _servers_shown and not _checks_done.is_set():
_checks_done.wait()
_loading = _init_done if not _servers_shown else None
user_input = ui.live_input(">>> ", task_manager, loading=_loading, mode=mode).strip()
# Afficher les notifications de succes creees PENDANT live_input
new_notifs = [n for n in task_manager.get_completed_notifications() if n.get("success")]
if new_notifs:
ui.print_completed_task_notifications(new_notifs)
# Entree vide : valide seulement si une action est en attente
# (necessite le pipeline initialise ; pendant l'init, on ignore)
if not user_input:
if not _init_done.is_set():
continue
session_obj = pipeline._pipeline_v2._session if hasattr(pipeline, '_pipeline_v2') else pipeline._session
if not session_obj.get_pending_action():
continue
# Effacer les notifications apres le premier message utilisateur
task_manager.clear_notifications()
# --- Commandes internes rapides : traitees AVANT l'attente d'init ---
# (elles n'ont pas besoin du pipeline RAG ; /setting repond ainsi
# immediatement meme pendant le chargement des modeles)
# Menu /setting ouvert : il consomme l'entree jusqu'a fermeture
if settings_menu.active:
response = settings_menu.handle(user_input)
ui.print_lyra(response)
if vocal and voice and response.startswith("[+]"):
voice.speak(response.splitlines()[0].lstrip("[+] "))
continue
# Commandes internes (acceptees avec ou sans prefixe "/")
cmd = user_input.lower().strip().lstrip("/")
if cmd in ("setting", "settings", "reglages", "parametres"):
if sys.stdin.isatty():
# TUI interactif : fleches + Entree, Echap/Ctrl+C pour revenir
from lyra.core.settings_tui import run_settings_tui
run_settings_tui(settings_menu, println=ui.print_lyra)
else:
# Fallback texte (pipe, one-shot, tests)
ui.print_lyra(settings_menu.open())
continue
if cmd in ("quit", "stop", "exit", "q"):
msg = "Au revoir!"
print(f"\n{msg}")
if vocal and voice:
voice.speak(msg)
break
if cmd in ("clearscreen", "cls"):
ui.clear_screen()
continue
if cmd == "mode":
ui.print_info(f"Mode actuel: {mode}")
continue
if cmd == "mode performance":
mode = "performance"
ui.print_success("Mode performance active")
continue
if cmd in ("mode default", "mode normal"):
mode = "default"
ui.print_success("Mode default active")
continue
# --- Fin des commandes rapides : la suite necessite le pipeline ---
# Attendre fin init si l'utilisateur a soumis avant la fin du chargement
if not _servers_shown:
if not _init_done.is_set():
_init_done.wait()
if _init_error[0]:
ui.print_error(f"Erreur initialisation: {_init_error[0]}")
break
if hasattr(pipeline, '_pipeline_v2'):
_srv_list = list(pipeline._pipeline_v2._hestia.mcp_manager.clients.keys())
else:
_srv_list = list(pipeline._hestia.mcp_manager.clients.keys())
ui.print_success(f"MCP: {', '.join(_srv_list)} ({len(_srv_list)} serveurs)")
_servers_shown = True
if cmd == "clear":
session.clear()
# EnhancedPipeline wrappe Pipeline V2, accéder via _pipeline_v2
session_obj = pipeline._pipeline_v2._session if hasattr(pipeline, '_pipeline_v2') else pipeline._session
session_obj.clear()
ui.print_info("Session effacee.")
continue
if cmd == "help":
print("""
Commandes internes (avec ou sans prefixe "/"):
quit, stop, exit - Quitter Lyra
clear - Effacer la session
clearscreen - Effacer l'ecran
/setting - Reglages (voix TTS, vitesse, mode)
mode - Afficher le mode actif
mode performance - Activer le mode performance
mode default - Activer le mode default
help - Afficher cette aide
outils - Lister tous les outils MCP
Exemples:
"demarre preprod-09"
"status des VMs"
"clone test-vm en test-vm-2"
"qu'est-ce que vm_clone ?"
""")
continue
if cmd in ("outils", "tools", "liste outils"):
# Lister tous les outils MCP via le pipeline
result = pipeline.process("quels outils disponibles ?")
ui.print_lyra(result.response)
continue
# Traiter la requete via le pipeline RAG (avec callbacks progressif + RAG verbose)
# En mode enhanced: les messages M1 remplacent l'indicateur "Reflexion en cours..."
if not args.rag_enhanced:
ui.print_thinking()
result = pipeline.process(
user_input,
callback=on_progress,
rag_step_callback=on_rag_step if args.rag_enhanced else None
)
if not args.rag_enhanced:
ui.clear_thinking()
for warning in getattr(result, "warnings", []):
ui.print_warning(warning)
# Mode DEBUG: Afficher details pipeline Enhanced
if args.debug and args.rag_enhanced and hasattr(result, 'normalized_query'):
print("\n" + "="*80)
print("🔍 DEBUG - Pipeline Enhanced")
print("="*80)
# Étape 1: Slang Normalization
if result.normalized_query and result.normalized_query != user_input:
print("[1] Slang Normalization:")
print(f" Input: '{user_input}'")
print(f" Output: '{result.normalized_query}'")
else:
print("[1] Slang Normalization: (inchangé)")
# Étape 2: Synonym Expansion
if result.expanded_query and result.expanded_query != (result.normalized_query or user_input):
orig_tokens = len((result.normalized_query or user_input).split())
exp_tokens = len(result.expanded_query.split())
added = exp_tokens - orig_tokens
print(f"[2] Synonym Expansion: +{added} tokens")
print(f" '{result.expanded_query[:80]}...'")
else:
print("[2] Synonym Expansion: (inchangé)")
# Étape 3: RAG Retrieval
if hasattr(result, 'rag_score') and result.rag_score is not None:
print("[3] RAG Retrieval:")
print(f" Score: {result.rag_score:.3f}")
if hasattr(result, 'rag_source') and result.rag_source:
print(f" Source: {result.rag_source}")
# Étape 4: Confidence Cascade
if hasattr(result, 'cascade_action') and result.cascade_action:
print("[4] Confidence Cascade:")
print(f" Action: {result.cascade_action}")
if result.rag_score:
if result.rag_score > 0.85:
print(" Level: HIGH (>0.85)")
elif result.rag_score >= 0.60:
print(" Level: MEDIUM (0.60-0.85)")
else:
print(" Level: LOW (<0.60)")
# Étape 5: Context Injection
if hasattr(result, 'should_inject_context'):
if result.should_inject_context:
print("[5] Context Injection: OUI")
else:
print("[5] Context Injection: NON")
# Étape 6: Tool Call Final
if result.tool_call:
print("[6] Tool Final:")
print(f" Name: {result.tool_call.get('name', 'N/A')}")
if result.tool_call.get('arguments'):
print(f" Args: {result.tool_call['arguments']}")
# Métriques de performance
if hasattr(result, 'metrics') and result.metrics:
print("\n📊 Performance Metrics:")
total = 0
for key, value in result.metrics.items():
if '_ms' in key and isinstance(value, (int, float)):
print(f" {key:25s}: {value:6.2f}ms")
total += value
if total > 0:
print(f" {'TOTAL':25s}: {total:6.2f}ms")
print("="*80 + "\n")
# Cas 1: Question de connaissance -> reponse directe
if result.query_type == QueryType.KNOWLEDGE:
ui.print_lyra(result.response)
if vocal and voice:
voice.speak(result.response)
# Cas 2: Action avec args manquants -> clarification
elif result.pending_args:
ui.print_lyra(result.response)
if vocal and voice:
voice.speak(result.response)
# Cas 3: Action prete -> confirmation et execution
elif result.tool_call:
# Afficher la reponse de LYRA (confirmation)
ui.print_lyra(result.response)
if vocal and voice:
voice.speak(result.response)
# Gerer l'execution avec confirmation
response = handle_action(
pipeline, result, mode, vocal, voice,
task_manager=task_manager,
webhook_url=DISCORD_WEBHOOK_URL
)
# Synthese vocale du resultat
if vocal and voice and response:
voice.speak(response)
# Cas 4: Pas de match -> reponse directe
else:
ui.print_lyra(result.response)
if vocal and voice:
voice.speak(result.response)
# Ajouter au contexte de session
session.add_turn(
user_input=user_input,
assistant_response=result.response,
tool_call=result.tool_call
)
except KeyboardInterrupt:
print("\n")
ui.print_info("Interruption. Au revoir!")
break
except Exception as e:
ui.print_error(f"Erreur: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
main()