from typing import TypedDict from langchain_core.messages import HumanMessage, SystemMessage, ToolMessage from langchain_core.tools import tool import logging import time import httpx from langchain_ollama import ChatOllama from langgraph.graph import StateGraph, END from .config import get_settings from .memory import Memory from .rag import retrieve_context from . import tools as t from . import watch as w from .tools import record_llm_latency from . import actions logger = logging.getLogger("vektor.graph") SYSTEM = """Tu es Vektor, l'assistant personnel de Thomas. Identité technique : tu es le modèle Qwen 2.5 14B (Alibaba Cloud) auto-hébergé via Ollama. Si on te demande qui tu es, quel modèle/pour qui tu as été fait, réponds cette vérité — ne dis JAMAIS Anthropic, OpenAI, Claude ou GPT (les petits modèles hallucinent souvent cette identité, ne le fais pas). Réponds en français suisse. Ne prétends jamais avoir effectué une action non confirmée. La documentation est un contexte, pas une preuve de l'état actuel. Les actions d'écriture suivent un flux strict : proposition puis confirmation OUI explicite, jamais d'exécution directe. Ne promets jamais d'exécuter une action toi-même. Ne révèle jamais de secret, token, mot de passe ou clé privée. Des outils de consultation en lecture seule sont disponibles. RÈGLE D'ACTION : quand la question porte sur l'état présent de l'infrastructure (état, status, comment marche X, est-ce que Y tourne, quel modèle, RAM, disque, DNS, backups...), APPELLE DIRECTEMENT l'outil pertinent sans demander la permission — c'est ta façon normale de répondre, pas une action à confirmer. Seules les ACTIONS D'ÉCRITURE (redémarrer, arrêter, modifier) exigent une confirmation OUI. RÈGLE MULTI-TÂCHES : si le message contient PLUSIEURS questions ou demandes (ex. « vérifie X et dis-moi si Y »), traite-les TOUTES, une par une, en appelant chaque outil nécessaire (tu peux enchaîner plusieurs tours d'outils avant de répondre), puis structure ta réponse point par point. Ne réponds jamais à une seule partie de la demande. Après un appel d'outil, réponds à partir de son résultat : cite les valeurs telles quelles, sans les inventer. Si aucun outil ni la documentation ne répondent à la question, dis-le en une phrase, puis réponds quand même avec ton sens général (le homelab de Thomas : Proxmox, Jellyfin, arr-stack, Pi-hole, VPS) plutôt que de refuser sèchement. """ class State(TypedDict): text: str history: list[dict[str, str]] context: str live_result: str response: str # ── Outils lecture seule exposés au LLM (tool-calling natif) ────────────── # Aucune action d'écriture ici : la whitelist + double confirmation OUI # reste un chemin déterministe en amont, hors de portée du modèle. @tool async def etat_proxmox() -> str: """État du nœud Proxmox : CPU, charge, RAM, swap, uptime.""" return await t.pve_summary() @tool async def liste_conteneurs() -> str: """Liste des conteneurs LXC Proxmox : état, RAM, uptime de chaque CT.""" return await t.pve_lxc_status() @tool async def etat_stockage() -> str: """Remplissage des stockages Proxmox avec alerte au-delà de 90 %.""" return await t.pve_storage_status() @tool async def inventaire_docker() -> str: """Inventaire des conteneurs Docker du homelab (état, image).""" return await t.docker_inventory() @tool async def statut_service(service: str) -> str: """Vérifie qu'un service surveillé répond (jellyfin, sonarr, radarr...).""" return await t.check_service(service) @tool async def rapport_seeds() -> str: """Seeding qBittorrent : top torrents, ratio, espace staging récupérable.""" return await t.seeds_report() @tool async def derniers_backups() -> str: """Dernières sauvegardes vzdump visibles par l'API Proxmox (âge, taille).""" return await w.backups_report() @tool async def etat_dns() -> str: """Sonde les résolveurs DNS configurés (DoH) sur un domaine de test.""" return await w.dns_report() @tool async def etat_monitoring() -> str: """État des checks de monitoring (Healthchecks.io) : ok, en retard, down.""" return await w.monitoring_report() @tool async def recherche_bibliotheque(query: str) -> str: """Cherche un film ou une série dans la bibliothèque Jellyfin. À appeler pour toute demande du type « ai-je le film X », « est-ce que la série Y est sur Jellyfin ». Accepte un titre (partiel ok).""" return await t.jellyfin_search(query) @tool async def info_externes() -> str: """Informations externes utiles au diagnostic : statut global Cloudflare (incident en cours ?), IP publique du foyer (dérive home-ip-monitor), dérive d'horloge NTP du serveur. Sources publiques sans clé.""" from . import extinfo return await extinfo.external_report() READONLY_TOOLS = [etat_proxmox, liste_conteneurs, etat_stockage, inventaire_docker, statut_service, rapport_seeds, derniers_backups, etat_dns, etat_monitoring, recherche_bibliotheque, info_externes] TOOLS_BY_NAME = {tool_item.name: tool_item for tool_item in READONLY_TOOLS} async def _invoke_with_tools(llm, messages: list, max_rounds: int = 3) -> str: """Inférence avec outils en plusieurs passes (max 3 par défaut). Chaque passe exécute TOUS les outils demandés puis réinjecte les résultats : le modèle peut enchaîner les vérifications (d'abord un état de service, puis une recherche bibliothèque...) avant de répondre — indispensable pour les messages à plusieurs tâches.""" llm_with_tools = llm.bind_tools(READONLY_TOOLS) for _round in range(max_rounds): answer = await llm_with_tools.ainvoke(messages) tool_calls = getattr(answer, "tool_calls", None) if not tool_calls: return answer.content messages.append(answer) for call in tool_calls: selected = TOOLS_BY_NAME.get(call["name"]) if selected is None: result = f"Outil inconnu ou non autorisé : {call['name']}" else: try: result = str(await selected.ainvoke(call.get("args") or {})) except Exception as exc: # un outil en panne ne doit pas casser la réponse result = f"Outil indisponible ({exc.__class__.__name__})." messages.append(ToolMessage(content=result, tool_call_id=call["id"])) # Plafond de passes atteint : réponse construite sur ce qui est déjà collecté final = await llm.ainvoke(messages) return final.content async def _pick_llm(base_url: str, model: str) -> ChatOllama: """Construit le LLM pour l'endpoint demandé (pas d'attribut timeout sur ChatOllama : le plafond effectif est le timeout HTTP du canal appelant).""" # keep_alive borné à 5m : règle RAM-GUARD (17/09) — le serveur MC partage # le VPS, un modèle résident long le met en OOM (cf. incidents 17/09 et # 01-02/10). Première question d'une salve ~15-40 s, les suivantes ~15 s. return ChatOllama(base_url=base_url, model=model, temperature=0.1, keep_alive="5m") async def select_llm() -> tuple[ChatOllama, str]: """Choisit l'endpoint d'inférence : primaire s'il répond, fallback (OLLAMA_FALLBACK_URL, ex. machine locale via VPN) sinon. Sonde légère /api/tags (3 s). Fallback vide ou injoignable → primaire : une machine éteinte ne produit jamais d'erreur visible.""" settings = get_settings() fallback_url = settings.ollama_fallback_url.strip() if fallback_url: try: async with httpx.AsyncClient(timeout=3) as client: response = await client.get(f"{fallback_url.rstrip('/')}/api/tags") if response.status_code == 200: return await _pick_llm(fallback_url, settings.ollama_model), "fallback" except httpx.HTTPError: pass return await _pick_llm(settings.ollama_base_url, settings.ollama_model), "primaire" async def run_agent(memory: Memory, text: str, history: list[dict[str, str]], user_key: str = "default") -> str: llm, source = await select_llm() if source == "fallback": logger.info("LLM : endpoint fallback (machine locale) utilisé") # 1. Confirmation en attente ? (le OUI n'exécute QUE l'action proposée) confirmation = await actions.confirm_pending(text, user_key) if confirmation is not None: return confirmation # 2. Intention d'action whitelistée ? -> classe autonome = exécution # immédiate (réversible/informationnelle : snapshots, inventaires, scan) ; # classe critique = proposition, jamais d'exécution sans OUI explicite. detected = actions.detect_action(text) if detected: action = detected[1] if actions.is_autonomous(action): result = await actions.execute_autonomous(action) prefix = ( "🩺 Diagnostic exécuté (lecture seule) :\n" if action.startswith("diagnose_") else "🤖 Action autonome exécutée (réversible, sans confirmation) :\n" ) return f"{prefix}{result}" already = actions._pending_notice(user_key) if already: return already return actions.propose(action, user_key) # 3. Routeur déterministe : donnée infra évidente -> réponse live sans LLM live_result = await t.infra_live_report(text) if live_result: return live_result # 4. LLM avec outils lecture seule : le modèle décide s'il doit consulter context = await retrieve_context(memory, text) messages = [SystemMessage(content=SYSTEM)] messages.extend(HumanMessage(content=item["content"]) for item in history[-8:]) messages.append(HumanMessage(content=( f"Documentation pertinente :\n{context}\n\n" f"Demande actuelle : {text}" ))) inf_start = time.perf_counter() try: return await _invoke_with_tools(llm, messages) finally: # Latence d'inférence réelle, affichée par la fiche /model record_llm_latency(time.perf_counter() - inf_start) def build_graph(): graph = StateGraph(State) graph.add_node("answer", lambda state: state) graph.set_entry_point("answer") graph.add_edge("answer", END) return graph.compile()