main.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""
Lyra V1 - Assistant Vocal DevOps Local
Pipeline: User Input -> LLM -> Tool Confirmation -> MCP Execution -> Response
Usage:
python main.py # Mode texte (defaut)
python main.py --vocal # Mode vocal (STT/TTS)
python main.py --model qwen2.5-coder:14b
"""
import os
import re
import sys
# Configurer LD_LIBRARY_PATH pour les libs CUDA (faster-whisper)
# Doit etre fait AVANT l'import de faster_whisper
def _setup_cuda_libs():
"""Configure le chemin des bibliotheques CUDA pour faster-whisper."""
venv_path = os.path.dirname(os.path.dirname(sys.executable))
cuda_paths = [
os.path.join(venv_path, "lib/python3.12/site-packages/nvidia/cublas/lib"),
os.path.join(venv_path, "lib/python3.12/site-packages/nvidia/cudnn/lib"),
os.path.join(venv_path, "lib64/python3.12/site-packages/nvidia/cublas/lib"),
os.path.join(venv_path, "lib64/python3.12/site-packages/nvidia/cudnn/lib"),
]
existing_paths = [p for p in cuda_paths if os.path.exists(p)]
if existing_paths:
current_ld_path = os.environ.get("LD_LIBRARY_PATH", "")
new_paths = ":".join(existing_paths)
os.environ["LD_LIBRARY_PATH"] = f"{new_paths}:{current_ld_path}" if current_ld_path else new_paths
_setup_cuda_libs()
import argparse
import subprocess
import time
from pathlib import Path
from typing import TYPE_CHECKING, Optional
if TYPE_CHECKING: # annotations seulement : l'import reel reste paresseux (dependances lourdes)
from modules.audio import VoiceInterface
# Ajouter le dossier modules au path
sys.path.insert(0, str(Path(__file__).parent))
from lyra.core.paths import backup_manager_dir, kvm_dir
from modules import ui
from modules.llm import LLMResponse, OllamaClient
from modules.mcp import MCPClient, MCPManager, get_all_default_tools
from modules.n8n import (
N8nClient,
N8nConfig,
get_async_executor,
get_async_workflow,
send_discord_notification,
should_use_async,
)
from modules.ui import setup_readline
_VM_NAME_RE = re.compile(r'^[a-zA-Z0-9_\-]{1,64}$')
_BACKUP_TYPE_RE = re.compile(r'^(timeshift|borg|rsync)$')
def _validate_vm_name(name: str) -> bool:
"""Valide un nom de VM pour eviter les injections shell."""
return bool(name and _VM_NAME_RE.match(name))
def _validate_backup_type(btype: str) -> bool:
"""Valide un type de backup pour eviter les injections shell."""
return bool(btype and _BACKUP_TYPE_RE.match(btype))
# =============================================================================
# Phase 5: Mode Performance - Outils sans confirmation
# =============================================================================
# Source unique de verite (lyra/core/constants.py) : les listes recopiees ici
# divergeaient (vm_exec absent, cast_* absents) -- audit 2026-09-19.
from lyra.core.constants import PERFORMANCE_TOOLS, is_dangerous_tool # noqa: E402
# Outils en lecture seule (pas de confirmation en mode performance)
QUICK_READ_TOOLS = {
"tv.get_state",
"hue.get_all_lights",
}
def load_system_prompt() -> str:
"""Charge le system prompt depuis le fichier externe."""
prompt_path = Path(__file__).parent / "prompts" / "system_prompt.txt"
try:
with open(prompt_path, "r", encoding="utf-8") as f:
return f.read()
except FileNotFoundError:
ui.print_warning(f"System prompt non trouve: {prompt_path}")
return "Tu es Lyra, un assistant DevOps. Reponds en francais."
# Charger le system prompt au demarrage
SYSTEM_PROMPT = load_system_prompt()
# Outils necessitant un Read-First (verification d'etat avant action)
READ_FIRST_TOOLS = {
"vm_start", "vm.start",
"vm_stop", "vm.stop",
"vm_destroy", "vm.destroy",
"vm_snapshot", "vm.snapshot",
"vm_clone", "vm.clone",
}
# Outils de lecture (pas besoin de Read-First)
READ_ONLY_TOOLS = {
"vm_status",
"backup_status",
"backup_list",
"backup_verify",
"help",
}
class Lyra:
"""Assistant DevOps avec Human-in-the-Loop."""
def __init__(self, model: str = "qwen2.5-coder:14b", vocal: bool = False, performance_mode: bool = False):
self.llm = OllamaClient(model=model)
self.mcp = MCPClient() # Client legacy (fedora-agents)
self.mcp_manager: Optional[MCPManager] = None # Multi-MCP (Phase 5)
self.tools = get_all_default_tools() # Tous les outils (VM + TV + Hue)
self._last_ctrl_c = 0 # Timestamp du dernier Ctrl+C
self.vocal = vocal
self.voice: Optional["VoiceInterface"] = None
self.n8n: Optional[N8nClient] = None
self.discord_webhook_url: Optional[str] = None
# Configuration et modes (Phase 5)
self.config: dict = {}
self.active_mode: str = "default"
# Charger la config
self._load_config()
# Override mode si argument --performance
if performance_mode:
self.active_mode = "performance"
if vocal:
self._init_voice()
def _load_config(self):
"""Charge la configuration depuis config.yaml."""
try:
import yaml
config_path = Path(__file__).parent / "config.yaml"
with open(config_path) as f:
self.config = yaml.safe_load(f)
# Mode actif (Phase 5)
self.active_mode = self.config.get("active_mode", "default")
# Config n8n
n8n_cfg = self.config.get("n8n", {})
if n8n_cfg.get("enabled", False):
self.n8n = N8nClient(N8nConfig(
base_url=n8n_cfg.get("base_url", "http://localhost:5678"),
enabled=True,
webhooks=n8n_cfg.get("webhooks", {})
))
if self.n8n.is_available():
ui.print_success("n8n connecte (operations async disponibles)")
else:
ui.print_warning("n8n configure mais non accessible")
self.n8n = None
# Config Discord (pour notifications fallback)
discord_cfg = self.config.get("discord", {})
if discord_cfg.get("enabled", False):
webhook_url = discord_cfg.get("webhook_url", "")
# Resoudre les variables d'environnement
if webhook_url.startswith("${") and webhook_url.endswith("}"):
env_var = webhook_url[2:-1]
self.discord_webhook_url = os.environ.get(env_var)
else:
self.discord_webhook_url = webhook_url
if self.discord_webhook_url:
ui.print_info("Discord notifications activees")
# Initialiser MCPManager si serveurs configures (Phase 5)
servers_cfg = self.config.get("mcp", {}).get("servers", {})
enabled_servers = [k for k, v in servers_cfg.items() if v.get("enabled", True)]
if enabled_servers:
try:
self.mcp_manager = MCPManager(self.config)
ui.print_info(f"MCPManager: {len(enabled_servers)} serveurs ({', '.join(enabled_servers)})")
except Exception as e:
ui.print_warning(f"MCPManager init error: {e}")
except Exception as e:
ui.print_warning(f"Erreur chargement config: {e}")
def _init_voice(self):
"""Initialise l'interface vocale."""
try:
import yaml
from modules.audio import AudioConfig, VoiceInterface
# Charger la config depuis config.yaml
config_path = Path(__file__).parent / "config.yaml"
with open(config_path) as f:
cfg = yaml.safe_load(f)
audio_cfg = cfg.get("audio", {})
stt_cfg = cfg.get("stt", {})
tts_cfg = cfg.get("tts", {})
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)...")
self.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=config
)
ui.print_success(f"Interface vocale prete (device: {config.input_device})")
except Exception as e:
ui.print_error(f"Erreur initialisation vocale: {e}")
ui.print_warning("Basculement en mode texte")
self.vocal = False
self.voice = None
def _should_skip_confirmation(self, tool_name: str) -> bool:
"""Determine si on peut skip la confirmation (mode performance).
Args:
tool_name: Nom de l'outil (avec ou sans prefixe)
Returns:
True si on peut executer sans confirmation
"""
# Toujours confirmer les outils dangereux
if is_dangerous_tool(tool_name):
return False
# Mode performance: skip pour les outils autorises
if self.active_mode == "performance":
if tool_name in PERFORMANCE_TOOLS or tool_name in QUICK_READ_TOOLS:
return True
# Mode default: toujours confirmer
return False
def _get_mode_config(self) -> dict:
"""Retourne la configuration du mode actif."""
modes = self.config.get("modes", {})
return modes.get(self.active_mode, modes.get("default", {}))
def set_mode(self, mode: str) -> bool:
"""Change le mode actif.
Args:
mode: "default" ou "performance"
Returns:
True si le mode a change
"""
if mode in ("default", "performance"):
self.active_mode = mode
return True
return False
def process_input(self, user_input: str) -> Optional[str]:
"""Traite une entree utilisateur.
Returns:
La reponse textuelle (pour le mode vocal) ou None
"""
response_text = None
# Commandes speciales
if user_input.lower() in ("help", "?", "aide"):
ui.print_help()
return "Voici l'aide." if self.vocal else None
if user_input.lower() in ("clear", "cls", "reset"):
self.llm.clear_history()
ui.print_success("Historique efface.")
return "Historique efface." if self.vocal else None
if user_input.lower() in ("clearscreen", "clr"):
ui.clear_screen()
return None
# Commandes de mode (Phase 5)
if user_input.lower() == "mode":
ui.print_mode_status(self.active_mode, len(PERFORMANCE_TOOLS))
return f"Mode actuel: {self.active_mode}" if self.vocal else None
if user_input.lower() == "mode performance":
self.set_mode("performance")
ui.print_success("Mode performance active - execution instantanee")
return "Mode performance active" if self.vocal else None
if user_input.lower() in ("mode default", "mode normal"):
self.set_mode("default")
ui.print_success("Mode default active - confirmations actives")
return "Mode normal active" if self.vocal else None
if user_input.lower() in ("quit", "exit", "q", "bye", "stop", "arrete"):
ui.print_info("Au revoir !")
if self.vocal and self.voice:
self.voice.speak("Au revoir !")
sys.exit(0)
# Envoyer au LLM
ui.print_thinking()
try:
response = self.llm.chat(
message=user_input,
tools=self.tools,
system_prompt=SYSTEM_PROMPT
)
except Exception as e:
ui.clear_thinking()
ui.print_error(f"Erreur LLM: {e}")
return f"Erreur: {e}" if self.vocal else None
ui.clear_thinking()
# Si c'est un ou plusieurs tool calls
if response.is_tool_call and response.tool_calls:
if len(response.tool_calls) > 1:
response_text = self._handle_multiple_tool_calls(response.tool_calls)
else:
response_text = self._handle_tool_call(response)
else:
# Reponse normale
ui.print_assistant(response.content)
response_text = response.content
return response_text
def _handle_tool_call(self, response: LLMResponse) -> Optional[str]:
"""Gere un appel d'outil avec confirmation.
Returns:
La reponse textuelle (resume LLM) ou None
"""
tool_call = response.tool_call
mode_config = self._get_mode_config()
# Mode performance: execution rapide si autorise
if self._should_skip_confirmation(tool_call.name):
return self._execute_quick(tool_call.name, tool_call.arguments)
# Read-First: pour les actions sur VM, verifier l'etat d'abord
vm_state = None
if mode_config.get("read_first", True) and tool_call.name in READ_FIRST_TOOLS:
vm_name = tool_call.arguments.get("vm_name") or tool_call.arguments.get("source_vm") or tool_call.arguments.get("source_vm_name")
if vm_name:
vm_state = self._read_first_check(vm_name)
# Verification intelligente: vm_clone necessite que la VM soit arretee
if tool_call.name in ("vm_clone", "vm.clone") and vm_state and vm_state.get("running"):
return self._handle_clone_running_vm(tool_call, vm_state)
# Afficher l'action proposee (avec l'etat VM si disponible)
ui.print_tool_call(tool_call.name, tool_call.arguments, vm_state=vm_state)
# En mode vocal, annoncer l'action de maniere plus naturelle
if self.vocal and self.voice and mode_config.get("tts_response", True):
action_desc = self._describe_action(tool_call.name, tool_call.arguments, vm_state)
self.voice.speak(action_desc)
# Demander confirmation (vocale ou clavier selon le mode)
confirmed = ui.confirm_action(
tool_call.name,
tool_call.arguments,
vocal_mode=self.vocal,
voice=self.voice
)
if not confirmed:
ui.print_warning("Action annulee.")
return "Action annulee." if self.vocal else None
# Verifier si c'est une operation async (n8n)
if self.n8n and should_use_async(tool_call.name, tool_call.arguments):
# Normaliser les arguments avant l'execution async
normalized_args = self._normalize_arguments(tool_call.name, tool_call.arguments)
return self._handle_async_operation(tool_call.name, normalized_args)
# Executer l'outil MCP (synchrone)
ui.print_info(f"Execution de {tool_call.name}...")
try:
result = self._call_mcp_tool(tool_call.name, tool_call.arguments)
except Exception as e:
ui.print_error(f"Erreur MCP: {e}")
return f"Erreur lors de l'execution: {e}" if self.vocal else None
# Afficher le resultat
if result.success:
ui.print_tool_result(result.content, success=True)
# Demander au LLM de resumer (si mode verbose)
if mode_config.get("verbose", True):
self.llm.add_tool_result(tool_call.name, result.content)
try:
summary = self.llm.chat(
message="", # Le message est deja dans l'historique
tools=None, # Pas de tools pour le resume
system_prompt=None
)
ui.print_assistant(summary.content)
return summary.content
except Exception:
pass
return result.content if self.vocal else None
else:
error_msg = result.error or "Erreur inconnue"
ui.print_tool_result(error_msg, success=False)
return f"Erreur: {error_msg}" if self.vocal else None
def _execute_quick(self, tool_name: str, arguments: dict) -> Optional[str]:
"""Execute un outil en mode rapide (performance mode).
Args:
tool_name: Nom de l'outil
arguments: Arguments
Returns:
Resultat minimal ou None
"""
try:
result = self._call_mcp_tool(tool_name, arguments)
if result.success:
ui.print_quick_result(tool_name, success=True, use_beep=True)
return "OK" if self.vocal else None
else:
ui.print_quick_result(tool_name, success=False, use_beep=True)
return f"Erreur: {result.error}" if self.vocal else None
except Exception as e:
ui.print_error(f"Erreur: {e}")
return f"Erreur: {e}" if self.vocal else None
def _normalize_arguments(self, tool_name: str, arguments: dict) -> dict:
"""Normalise les arguments pour corriger les variations du LLM.
Le LLM peut generer des noms de parametres legerement differents.
Cette fonction les corrige pour correspondre a l'API MCP.
Args:
tool_name: Nom de l'outil (normalise, ex: vm_clone)
arguments: Arguments originaux
Returns:
Arguments normalises
"""
args = arguments.copy()
# vm_clone: normaliser les variations de noms de parametres
if tool_name in ("vm_clone", "vm.clone"):
# source_vm_name -> source_vm
if "source_vm_name" in args and "source_vm" not in args:
args["source_vm"] = args.pop("source_vm_name")
# target_vm_name / target_vm / clone_name / name -> new_vm_name
if "new_vm_name" not in args:
for alt in ("target_vm_name", "target_vm", "clone_name", "name", "new_name"):
if alt in args:
args["new_vm_name"] = args.pop(alt)
break
# vm_copy: src -> source, dst -> dest
if tool_name in ("vm_copy", "vm.copy"):
if "src" in args and "source" not in args:
args["source"] = args.pop("src")
if "dst" in args and "dest" not in args:
args["dest"] = args.pop("dst")
return args
def _call_mcp_tool(self, tool_name: str, arguments: dict):
"""Appelle un outil MCP via le client approprie.
Utilise MCPManager si disponible (multi-serveurs),
sinon fallback sur le client legacy.
Args:
tool_name: Nom de l'outil (avec ou sans prefixe)
arguments: Arguments de l'outil
Returns:
MCPResult
"""
# Normaliser les arguments (corriger variations LLM)
arguments = self._normalize_arguments(tool_name, arguments)
# Si l'outil a un prefixe et MCPManager est disponible
if self.mcp_manager and "." in tool_name:
return self.mcp_manager.call_tool(tool_name, arguments)
# Sinon utiliser le client legacy
# Retirer le prefixe "fedora." si present
clean_name = tool_name.split(".")[-1] if "." in tool_name else tool_name
return self.mcp.call_tool(clean_name, arguments)
def _read_first_check(self, vm_name: str) -> Optional[dict]:
"""Verifie l'etat d'une VM avant une action (Read-First).
Returns:
Dict avec l'etat de la VM ou None si erreur
"""
try:
ui.print_info(f"Verification de l'etat de {vm_name}...")
result = self.mcp.call_tool("vm_status", {"vm_name": vm_name})
if result.success:
# Parser le resultat pour extraire l'etat
content = result.content
state = {
"vm_name": vm_name,
"raw": content,
"running": "running" in content.lower() or "en cours" in content.lower(),
"ip": self._extract_ip(content),
}
return state
except Exception as e:
ui.print_warning(f"Read-First: impossible de verifier {vm_name}: {e}")
return None
def _extract_ip(self, content: str) -> Optional[str]:
"""Extrait une adresse IP du contenu."""
import re
match = re.search(r'(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})', content)
return match.group(1) if match else None
def _handle_clone_running_vm(self, tool_call, vm_state: dict) -> Optional[str]:
"""Gere le cas ou l'utilisateur veut cloner une VM en cours d'execution.
Propose d'arreter la VM, de la cloner, puis de la redemarrer.
Args:
tool_call: ToolCall original (vm_clone)
vm_state: Etat actuel de la VM
Returns:
Resume des actions ou None
"""
source_vm = tool_call.arguments.get("source_vm") or tool_call.arguments.get("source_vm_name") or ""
new_vm = tool_call.arguments.get("new_vm_name") or tool_call.arguments.get("target_vm_name") or ""
# Afficher l'avertissement
print()
ui.print_warning(f"La VM '{source_vm}' est en cours d'execution!")
print()
print(" Pour cloner une VM, elle doit etre arretee.")
print()
print(" Je propose d'executer ces actions:")
print(f" 1. Arreter {source_vm}")
print(f" 2. Cloner {source_vm} -> {new_vm}")
print(f" 3. Redemarrer {source_vm}")
print()
# Demander confirmation
try:
choice = input(" Executer ? [O/n/s] (O=oui, n=non, s=sans redemarrer) : ").strip().lower()
except (KeyboardInterrupt, EOFError):
print()
ui.print_warning("Action annulee.")
return "Action annulee." if self.vocal else None
if choice in ("n", "non"):
ui.print_warning("Action annulee.")
return "Action annulee." if self.vocal else None
restart_after = choice not in ("s", "sans")
# 1. Arreter la VM
ui.print_info(f"Arret de {source_vm}...")
stop_result = self.mcp.call_tool("vm_stop", {"vm_name": source_vm, "wait": True})
if not stop_result.success:
ui.print_error(f"Echec de l'arret: {stop_result.error}")
return f"Erreur: {stop_result.error}" if self.vocal else None
ui.print_success(f"{source_vm} arretee.")
# 2. Cloner la VM (normaliser les arguments)
clone_args = self._normalize_arguments("vm_clone", tool_call.arguments)
ui.print_info(f"Clonage de {source_vm} vers {new_vm}...")
# Verifier si c'est async
if self.n8n and should_use_async("vm_clone", clone_args):
result = self._handle_async_operation("vm_clone", clone_args)
if restart_after:
ui.print_info(f"Note: {source_vm} sera redemarre une fois le clone termine.")
return result
clone_result = self.mcp.call_tool("vm_clone", clone_args)
if not clone_result.success:
ui.print_error(f"Echec du clonage: {clone_result.error}")
# Redemarrer quand meme la VM source en cas d'echec
if restart_after:
ui.print_info(f"Redemarrage de {source_vm}...")
self.mcp.call_tool("vm_start", {"vm_name": source_vm})
return f"Erreur: {clone_result.error}" if self.vocal else None
ui.print_success(f"Clone {new_vm} cree.")
# 3. Redemarrer la VM source (si demande)
if restart_after:
ui.print_info(f"Redemarrage de {source_vm}...")
start_result = self.mcp.call_tool("vm_start", {"vm_name": source_vm, "wait_ssh": True})
if start_result.success:
ui.print_success(f"{source_vm} redemarre.")
else:
ui.print_warning(f"Echec du redemarrage: {start_result.error}")
summary = f"VM {source_vm} clonee vers {new_vm}"
if restart_after:
summary += f", {source_vm} redemarre"
ui.print_success(summary)
return summary if self.vocal else None
def _describe_action(self, tool_name: str, arguments: dict, vm_state: Optional[dict]) -> str:
"""Genere une description vocale de l'action proposee."""
vm_name = arguments.get("vm_name") or arguments.get("source_vm") or arguments.get("source_vm_name") or ""
source_vm = arguments.get("source_vm") or arguments.get("source_vm_name") or ""
new_vm = arguments.get("new_vm_name") or arguments.get("target_vm_name") or arguments.get("target_vm") or arguments.get("clone_name") or ""
descriptions = {
"vm_start": f"Je vais demarrer la VM {vm_name}",
"vm_stop": f"Je vais arreter la VM {vm_name}",
"vm_destroy": f"Attention, je vais supprimer definitivement la VM {vm_name}",
"vm_snapshot": f"Je vais gerer un snapshot de {vm_name}",
"vm_exec": f"Je vais executer une commande sur {vm_name}",
"vm_clone": f"Je vais cloner la VM {source_vm} vers {new_vm}",
"backup_create": "Je vais creer une sauvegarde",
"backup_restore": "Attention, je vais restaurer une sauvegarde, cela ecrasera les donnees actuelles",
"backup_clean": "Je vais nettoyer les anciennes sauvegardes",
}
desc = descriptions.get(tool_name, f"Je vais executer {tool_name}")
# Ajouter l'etat actuel si disponible
if vm_state:
status = "en cours d'execution" if vm_state["running"] else "arretee"
if vm_state.get("ip"):
desc += f". Elle est actuellement {status} avec l'IP {vm_state['ip']}"
else:
desc += f". Elle est actuellement {status}"
desc += ". Confirmes-tu?"
return desc
def _handle_multiple_tool_calls(self, tool_calls: list) -> Optional[str]:
"""Gere plusieurs tool calls comme une todo list.
Args:
tool_calls: Liste de ToolCall a executer
Returns:
Resume des actions executees
"""
# Afficher la todo list
print()
ui.print_info(f"Todo list: {len(tool_calls)} actions proposees")
print("=" * 50)
for i, tc in enumerate(tool_calls, 1):
vm_name = tc.arguments.get("vm_name") or tc.arguments.get("source_vm") or tc.arguments.get("source_vm_name") or "?"
print(f" [{i}] {tc.name} -> {vm_name}")
print("=" * 50)
print()
# Demander confirmation globale
try:
choice = input("Executer ? [T]out / [1] par 1 / [n]on : ").strip().lower()
except (KeyboardInterrupt, EOFError):
print()
ui.print_warning("Annule.")
return "Actions annulees." if self.vocal else None
if choice in ("n", "non", "no"):
ui.print_warning("Actions annulees.")
return "Actions annulees." if self.vocal else None
execute_all = choice in ("t", "tout", "all", "o", "oui", "")
results = []
for i, tc in enumerate(tool_calls, 1):
vm_name = tc.arguments.get("vm_name") or tc.arguments.get("source_vm") or tc.arguments.get("source_vm_name") or "?"
print()
ui.print_info(f"[{i}/{len(tool_calls)}] {tc.name} -> {vm_name}")
if not execute_all:
# Demander confirmation pour chaque action
try:
confirm = input(" Executer ? [o/n] : ").strip().lower()
except (KeyboardInterrupt, EOFError):
print()
ui.print_warning("Arret de la todo list.")
break
if confirm not in ("o", "oui", "y", "yes", ""):
ui.print_warning(f" -> {tc.name} ignore")
results.append(f"{vm_name}: ignore")
continue
# Executer l'action
try:
# Verifier si c'est une operation async
if self.n8n and should_use_async(tc.name, tc.arguments):
self._handle_async_operation(tc.name, tc.arguments)
results.append(f"{vm_name}: lance en arriere-plan")
else:
# Utiliser _call_mcp_tool pour router vers le bon serveur (Phase 5.4)
result = self._call_mcp_tool(tc.name, tc.arguments)
if result.success:
ui.print_success(" -> OK")
results.append(f"{vm_name}: OK")
else:
ui.print_error(f" -> Erreur: {result.error}")
results.append(f"{vm_name}: erreur")
except Exception as e:
ui.print_error(f" -> Exception: {e}")
results.append(f"{vm_name}: exception")
# Resume
print()
summary = f"Todo list terminee: {len(results)}/{len(tool_calls)} actions"
ui.print_success(summary)
return summary if self.vocal else None
def _handle_async_operation(self, tool_name: str, arguments: dict) -> Optional[str]:
"""Gere une operation longue via n8n (async) ou fallback subprocess.
Args:
tool_name: Nom de l'outil MCP
arguments: Arguments de l'outil
Returns:
Message de confirmation
"""
workflow = get_async_workflow(tool_name)
if not workflow:
ui.print_error(f"Pas de workflow pour {tool_name}")
return None
ui.print_info(f"Lancement async via n8n ({workflow})...")
# Preparer les donnees pour le webhook
data = dict(arguments)
data["tool"] = tool_name
data["timestamp"] = time.strftime("%Y-%m-%d %H:%M:%S")
result = self.n8n.trigger_workflow(workflow, data)
if result.success:
msg = "Operation lancee en arriere-plan via n8n. Tu recevras une notification Discord."
ui.print_success(msg)
if self.vocal and self.voice:
self.voice.speak("C'est lance. Tu recevras une notification Discord.")
return msg
else:
# Fallback: execution en arriere-plan via subprocess
# Afficher la raison de l'echec n8n pour debug
ui.print_warning(f"n8n: {result.message}")
if result.error and result.error != result.message:
ui.print_warning(f" Detail: {result.error}")
ui.print_info("Lancement en arriere-plan (subprocess)...")
# Capturer discord_webhook_url pour le callback
discord_url = self.discord_webhook_url
def on_complete(name, args, success, stdout, stderr):
"""Callback quand l'operation est terminee."""
# Effacer la ligne courante et revenir au debut
print("\r\033[K", end="", flush=True)
if success:
ui.print_success(f"[ASYNC] Operation {name} terminee avec succes!")
else:
ui.print_error(f"[ASYNC] Operation {name} echouee: {stderr[:100] if stderr else 'erreur inconnue'}")
# Reafficher le prompt
print("\nToi: ", end="", flush=True)
# Envoyer notification Discord si configure
if discord_url:
footer = "Lyra DevOps Assistant - Mode Fallback (n8n indisponible)"
verify_output = ""
verify_success = False
# Executer verification deep selon le type d'operation
if name in ("vm_clone", "vm.clone") and success:
# Pour vm_clone: attendre IP, verifier, eteindre
# Note: la VM est deja demarree par le script clone (--start)
new_vm = args.get("new_vm_name") or args.get("target_vm_name", "")
if not new_vm:
verify_output = f"Erreur: new_vm_name/target_vm_name non trouve dans args: {args}"
verify_success = False
elif not _validate_vm_name(new_vm):
verify_output = f"Nom de VM invalide: {new_vm!r}"
verify_success = False
else:
try:
# 1. Attendre que la VM ait une IP (max 120s)
# La VM est deja demarree par le script clone
ip_found = False
vm_ip = ""
for attempt in range(40): # 40 tentatives x 3s = 120s max
time.sleep(3)
# Utiliser virsh domifaddr pour obtenir l'IP
ip_check = subprocess.run(
["virsh", "domifaddr", new_vm],
capture_output=True, text=True
)
# Chercher une IP dans la sortie (format: vnet0 52:54:00:xx:xx:xx ipv4 192.168.122.xxx/24)
ip_match = re.search(r'(\d+\.\d+\.\d+\.\d+)', ip_check.stdout)
if ip_match:
vm_ip = ip_match.group(1)
ip_found = True
break
if ip_found:
# 2. Attendre que SSH soit disponible (host key accept)
ssh_ready = False
for ssh_attempt in range(20): # 60s max
ssh_check = subprocess.run(
["ssh", "-o", "StrictHostKeyChecking=accept-new",
"-o", "ConnectTimeout=3",
f"amineutron@{vm_ip}", "echo OK"],
capture_output=True, text=True, timeout=10
)
if "OK" in ssh_check.stdout:
ssh_ready = True
break
time.sleep(3)
if ssh_ready:
# 3. Lancer verify-vm-clone.sh --verbose
verify_result = subprocess.run(
[str(kvm_dir() / "verify-vm-clone.sh"),
"--vm", new_vm, "--verbose"],
capture_output=True, text=True, timeout=180
)
verify_output = verify_result.stdout
if verify_result.stderr:
verify_output += f"\n{verify_result.stderr}"
# Nettoyer codes ANSI pour verifier le status
clean_check = re.sub(r'\x1b\[[0-9;]*m', '', verify_output)
# Le script retourne 1 pour warnings, on considere OK si output contient "PRET"
verify_success = "PRET" in clean_check or "PRÊT" in clean_check or verify_result.returncode == 0
else:
verify_output = f"SSH non disponible apres 60s (IP: {vm_ip})"
verify_success = False
else:
# Verifier l'etat de la VM pour debug
vm_state = subprocess.run(["virsh", "domstate", new_vm], capture_output=True, text=True)
verify_output = f"Timeout: la VM {new_vm} n'a pas obtenu d'IP apres 120s\nEtat VM: {vm_state.stdout.strip()}"
verify_success = False
# 3. Eteindre la VM
subprocess.run(["virsh", "shutdown", new_vm], capture_output=True, timeout=30)
except Exception as e:
verify_output = f"Erreur verification: {e}"
verify_success = False
# Tenter d'eteindre la VM en cas d'erreur
subprocess.run(["virsh", "destroy", new_vm], capture_output=True)
elif name in ("backup_create", "backup.create") and success:
# Lancer backup verify --deep
backup_type = args.get("type", "timeshift")
if not _validate_backup_type(backup_type):
verify_output = f"Type de backup invalide: {backup_type!r}"
verify_success = False
else:
try:
verify_result = subprocess.run(
[str(backup_manager_dir() / "backup-verify.sh"),
backup_type, "--deep"],
capture_output=True, text=True, timeout=300
)
verify_output = verify_result.stdout
verify_success = verify_result.returncode == 0
except Exception as e:
verify_output = f"Erreur verification: {e}"
verify_success = False
elif name in ("backup_restore", "backup.restore") and success:
# Lancer backup verify --deep apres restauration
backup_type = args.get("type", "timeshift")
if not _validate_backup_type(backup_type):
verify_output = f"Type de backup invalide: {backup_type!r}"
verify_success = False
else:
try:
verify_result = subprocess.run(
[str(backup_manager_dir() / "backup-verify.sh"),
backup_type, "--deep"],
capture_output=True, text=True, timeout=300
)
verify_output = verify_result.stdout
verify_success = verify_result.returncode == 0
except Exception as e:
verify_output = f"Erreur verification: {e}"
verify_success = False
# Couleur finale basee sur operation + verification
final_success = success and (verify_success if verify_output else True)
color = 3066993 if final_success else 15158332 # Vert ou Rouge
# Premier message: Verification Deep (si disponible)
if verify_output and len(verify_output.strip()) > 0:
# Nettoyer les codes ANSI et messages comm pour Discord
import re
clean_output = re.sub(r'\x1b\[[0-9;]*m', '', verify_output)
# Filtrer les lignes d'erreur comm (tri de fichiers)
clean_output = '\n'.join(line for line in clean_output.split('\n') if not line.startswith('comm:'))
output_text = clean_output[:3500] if len(clean_output) > 3500 else clean_output
verify_title = {
"vm_clone": "Verification VM Clone",
"vm.clone": "Verification VM Clone",
"backup_create": "Verification Backup",
"backup.create": "Verification Backup",
"backup_restore": "Verification Restauration",
"backup.restore": "Verification Restauration",
}.get(name, f"Verification {name}")
send_discord_notification(
discord_url,
title=verify_title,
description=f"```\n{output_text}\n```",
color=color,
footer=footer
)
# Attendre pour eviter rate limit Discord
time.sleep(2)
# Deuxieme message: Resume synthetise
if name in ("vm_clone", "vm.clone"):
title = "Clone VM - Resume"
# Extraire les scores du rapport de verification
func_score = "N/A"
ident_score = "N/A"
if verify_output:
import re
# Nettoyer les codes ANSI avant extraction
clean_verify = re.sub(r'\x1b\[[0-9;]*m', '', verify_output)
# Format: "FONCTIONNEL: 96/100 (27/28)"
func_match = re.search(r'FONCTIONNEL:\s+(\d+)/100', clean_verify)
if func_match:
func_score = f"{func_match.group(1)}/100"
# Format: "IDENTITÉ: 95/100 (21/22)"
ident_match = re.search(r'IDENTIT[EÉ]:\s+(\d+)/100', clean_verify)
if ident_match:
ident_score = f"{ident_match.group(1)}/100"
fields = [
{"name": "Source", "value": args.get("source_vm") or args.get("source_vm_name", "?"), "inline": True},
{"name": "Clone", "value": args.get("new_vm_name") or args.get("target_vm_name", "?"), "inline": True},
{"name": "Clonage", "value": "OK" if success else "ECHEC", "inline": True},
{"name": "Fonctionnel", "value": func_score, "inline": True},
{"name": "Conformite Hote", "value": ident_score, "inline": True},
{"name": "Status Final", "value": "PRET" if final_success else "ERREUR", "inline": False},
]
elif name in ("backup_create", "backup.create"):
title = "Backup Create - Resume"
fields = [
{"name": "Type", "value": args.get("type", "timeshift"), "inline": True},
{"name": "Creation", "value": "OK" if success else "ECHEC", "inline": True},
{"name": "Verification", "value": "OK" if verify_success else ("ECHEC" if verify_output else "N/A"), "inline": True},
{"name": "Status Final", "value": "TOUT EST OK" if final_success else "ERREUR", "inline": False},
]
elif name in ("backup_restore", "backup.restore"):
title = "Backup Restore - Resume"
fields = [
{"name": "Type", "value": args.get("type", "timeshift"), "inline": True},
{"name": "Identifier", "value": args.get("identifier", "?"), "inline": True},
{"name": "Restauration", "value": "OK" if success else "ECHEC", "inline": True},
{"name": "Verification", "value": "OK" if verify_success else ("ECHEC" if verify_output else "N/A"), "inline": True},
{"name": "Status Final", "value": "TOUT EST OK" if final_success else "ERREUR", "inline": False},
]
else:
title = f"Operation {name} - Resume"
fields = [
{"name": "Status", "value": "OK" if success else "ERREUR", "inline": True}
]
send_discord_notification(
discord_url,
title=title,
fields=fields,
color=color,
footer=footer
)
executor = get_async_executor(on_complete)
fallback_result = executor.execute_async(tool_name, arguments)
if fallback_result.success:
msg = "Operation lancee en arriere-plan (mode local). Tu seras notifie dans le terminal."
ui.print_success(msg)
if self.vocal and self.voice:
self.voice.speak("C'est lance en arriere-plan.")
return msg
else:
ui.print_error(f"Erreur fallback: {fallback_result.message}")
return None
def run(self) -> None:
"""Boucle principale."""
# Configurer readline (Ctrl+L pour clear)
setup_readline()
ui.print_banner()
if self.vocal:
ui.print_info("Mode VOCAL actif - Parlez apres le bip")
ui.print_info("Dites 'stop' ou 'arrete' pour quitter")
else:
ui.print_info("Tapez 'help' pour l'aide, 'quit' pour quitter.")
ui.print_info(f"Modele: {self.llm.model}")
# Afficher le mode actif (Phase 5)
if self.active_mode == "performance":
ui.print_mode_status("performance", len(PERFORMANCE_TOOLS))
# Verifier la connexion MCP
try:
tools = self.mcp.list_tools()
if tools:
ui.print_success(f"MCP connecte: {len(tools)} outils disponibles")
else:
ui.print_warning("MCP: utilisation des outils par defaut")
except Exception as e:
ui.print_warning(f"MCP: {e} - utilisation des outils par defaut")
print()
# Choisir la boucle appropriee
if self.vocal and self.voice:
self._run_vocal_loop()
else:
self._run_text_loop()
def _run_text_loop(self) -> None:
"""Boucle principale en mode texte."""
while True:
try:
user_input = ui.get_user_input()
if not user_input:
continue
self.process_input(user_input)
except KeyboardInterrupt:
now = time.time()
# Double Ctrl+C en moins de 1.5 secondes = quitter
if now - self._last_ctrl_c < 1.5:
print()
ui.print_info("Au revoir !")
sys.exit(0)
else:
self._last_ctrl_c = now
print()
ui.print_info("Ctrl+C again to quit, Ctrl+L to clear.")
except Exception as e:
ui.print_error(f"Erreur: {e}")
def _run_vocal_loop(self) -> None:
"""Boucle principale en mode vocal."""
# Annoncer le demarrage
self.voice.speak("Lyra est prete.")
print()
ui.print_info("Parle apres le bip. La barre indique le niveau audio.")
print()
while True:
try:
# Ecouter (avec bip) - l'indicateur de niveau s'affiche pendant l'ecoute
user_input = self.voice.listen(with_beep=True)
if not user_input:
print(" (silence - pas de parole detectee)")
continue
# Afficher ce qui a ete compris
print(f" Toi: \"{user_input}\"")
# Traiter et obtenir la reponse
response = self.process_input(user_input)
# Prononcer la reponse
if response and self.voice:
self.voice.speak(response)
print() # Ligne vide avant la prochaine ecoute
except KeyboardInterrupt:
now = time.time()
if now - self._last_ctrl_c < 1.5:
print()
ui.print_info("Au revoir !")
try:
if self.voice:
self.voice.speak("Au revoir !")
except KeyboardInterrupt:
pass # Ignorer si interrompu pendant le TTS
sys.exit(0)
else:
self._last_ctrl_c = now
print()
ui.print_info("Ctrl+C again to quit")
except Exception as e:
ui.print_error(f"Erreur: {e}")
if self.voice:
self.voice.speak(f"Une erreur s'est produite: {e}")
def main():
"""Point d'entree principal."""
parser = argparse.ArgumentParser(
description="Lyra - Assistant DevOps Local avec Human-in-the-Loop"
)
parser.add_argument(
"-m", "--model",
default="qwen2.5-coder:14b",
help="Modele Ollama a utiliser (default: qwen2.5-coder:14b)"
)
parser.add_argument(
"-v", "--vocal",
action="store_true",
help="Activer le mode vocal (STT/TTS)"
)
parser.add_argument(
"-p", "--performance",
action="store_true",
help="Mode performance - execution instantanee pour domotique (Phase 5)"
)
parser.add_argument(
"--no-banner",
action="store_true",
help="Ne pas afficher la banniere"
)
parser.add_argument(
"--list-devices",
action="store_true",
help="Lister les peripheriques audio disponibles"
)
args = parser.parse_args()
# Liste des peripheriques audio
if args.list_devices:
import sounddevice as sd
print("Peripheriques audio disponibles:")
print(sd.query_devices())
sys.exit(0)
lyra = Lyra(model=args.model, vocal=args.vocal, performance_mode=args.performance)
lyra.run()
if __name__ == "__main__":
main()