app/graph.py
Rama main · réplica de lectura del repositorio original.
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()