"""Contextes présélectionnables d'Asa — profils de domaine + assemblage du prompt. Chaque profil définit un system prompt et les **sources live** à injecter dans le contexte (Ghostfolio, Prometheus, Alertmanager, doc Funk). Le client choisit un contexte par requête (`context` dans /v1/ask) ; le serveur assemble le prompt et renvoie le contexte assemblé pour le **visualiseur** du HUD. Ajouter un contexte = une entrée dans CONTEXTS (id, label, icône, prompt, sources). """ from __future__ import annotations from dataclasses import dataclass from stt_server.config import settings # Préambule commun (ton, langue, concision) — chaque prompt de contexte le complète. _BASE = ( "Tu es Hermes, l'assistant vocal du homelab Funk. Réponds toujours en français, " "de façon concise (2-3 phrases maximum), sans markdown ni listes." ) @dataclass(frozen=True) class Context: id: str label: str icon: str description: str system_prompt: str sources: tuple[str, ...] = () # "docs","ghostfolio","alerts","cluster","metrics" tools: tuple[str, ...] = () # outils function-calling (cf. tools.py) → boucle agentique # Le contexte « funk » garde le prompt grounding-strict historique (env override possible). CONTEXTS: dict[str, Context] = { # Contexte « asa » : boucle d'outils (function calling). Asa décide elle-même quand # sonder l'état live du homelab ou chercher dans la doc, puis répond. C'est le contexte # par défaut côté production (STT_DEFAULT_CONTEXT=asa) — « gpu-01 tourne bien ? » marche # sans présélection. Les outils sont en LECTURE SEULE (volet actions = contexte « agent »). "asa": Context( id="asa", label="Asa · agent", icon="✨", description="Asa répond en s'appuyant sur des outils (doc + état live du homelab).", system_prompt=_BASE + " Tu disposes d'outils pour obtenir l'état RÉEL et LIVE du " "homelab (santé des machines, état du cluster, état du réseau, métriques Prometheus), " "chercher dans la documentation Funk, et faire une recherche sur INTERNET (web_search). Dès qu'une " "question porte sur l'état courant, un fait précis du homelab, ou une info externe / " "récente, APPELLE l'outil adéquat puis réponds à partir de son résultat — n'invente " "jamais. Si les outils ne donnent pas l'information, dis-le simplement. Pour le " "bavardage général, réponds directement sans outil. Si l'utilisateur demande " "explicitement d'AGIR (redémarrer, relancer, modifier…), utilise l'outil admin_action " "(s'il est disponible) — il proposera l'action et demandera confirmation, sans rien " "exécuter de toi-même.", tools=("search_docs", "host_health", "cluster_status", "network_status", "prometheus_query", "web_search", "admin_action"), ), "funk": Context( id="funk", label="Funk · cluster", icon="🛠️", description="Questions sur le homelab depuis la documentation.", system_prompt=settings.system_prompt, sources=("docs",), ), "ghostfolio": Context( id="ghostfolio", label="Ghostfolio", icon="💰", description="Ton portefeuille et tes investissements.", system_prompt=_BASE + " Tu aides sur le portefeuille d'investissement de " "l'utilisateur. Appuie-toi sur les données Ghostfolio fournies ci-dessous ; " "donne des montants précis quand ils y figurent. Si une donnée n'est pas " "fournie, dis-le simplement.", sources=("ghostfolio",), ), "grafana": Context( id="grafana", label="Grafana · métriques", icon="📊", description="Métriques et observabilité du homelab.", system_prompt=_BASE + " Tu réponds sur les métriques et l'observabilité du " "homelab à partir des données fournies ci-dessous (Prometheus). Reste factuel.", sources=("metrics",), ), "alerting": Context( id="alerting", label="Alerting", icon="🚨", description="Alertes actives et leur signification.", system_prompt=_BASE + " Tu expliques et tries les alertes actives fournies " "ci-dessous : gravité, cause probable, quoi vérifier. S'il n'y a aucune " "alerte, rassure brièvement.", sources=("alerts",), ), "cluster": Context( id="cluster", label="Gestion cluster", icon="🚢", description="État et gestion du cluster Kubernetes Funk (lecture seule).", system_prompt=_BASE + " Tu réponds sur l'état et la gestion du cluster " "Kubernetes Funk à partir de l'état fourni ci-dessous et de la documentation. " "Tu fais de la LECTURE SEULE : tu n'exécutes aucune action.", sources=("cluster", "docs"), ), "reseau": Context( id="reseau", label="Réseau", icon="🌐", description="Surveillance réseau active : joignabilité, internet, DNS, latence.", system_prompt=_BASE + " Tu réponds sur l'état du RÉSEAU du homelab à partir des " "sondes actives fournies ci-dessous (joignabilité des hôtes et de la passerelle, " "accès internet, DNS *.lab.local, latence LAN, services injoignables). Reste " "factuel ; si une sonde est en échec, précise laquelle.", sources=("network",), ), # Contexte « agent » : court-circuite le LLM — la demande est confirmée puis # exécutée par le vrai agent Hermes (hermes-exec). Géré entièrement dans app.py # (handshake), pas via assemble(). Exposé seulement si actions_enabled + jeton. "agent": Context( id="agent", label="Agent (actions)", icon="🤖", description="Demander à Hermes d'AGIR sur le homelab (confirmation requise).", system_prompt="", # non utilisé : flux dédié sources=(), ), } # Titres des blocs injectés (visualiseur + prompt) _SOURCE_TITLES = { "ghostfolio": "Portefeuille Ghostfolio (données live)", "metrics": "Métriques Prometheus (live)", "alerts": "Alertes actives (Alertmanager)", "cluster": "État du cluster (live)", "network": "Réseau — sondes actives (live)", "docs": "Documentation Funk", } def get_context(context_id: str | None) -> Context: return CONTEXTS.get(context_id or "", CONTEXTS[settings.default_context]) def assemble( ctx: Context, *, blocks: list[tuple[str, str]], docs: list[str], memories: list[str], ) -> tuple[str, dict]: """Construit le system prompt final + la structure de visualisation. `blocks` : (source_id, texte) des sources live déjà récupérées. Renvoie (system_prompt, debug) où debug alimente le visualiseur du HUD. """ system = ctx.system_prompt debug_blocks: list[dict] = [] for src, text in blocks: if not text: continue title = _SOURCE_TITLES.get(src, src) system += f"\n\n=== {title} ===\n{text}" debug_blocks.append({"source": src, "title": title, "text": text}) if "docs" in ctx.sources and docs: extraits = "\n\n".join(docs) system += ( "\n\n=== Documentation Funk (réponds À PARTIR de ces extraits ; " "si l'info n'y est pas, dis-le) ===\n" + extraits ) if memories: souvenirs = "\n".join(f"- {m}" for m in memories) system += ( "\n\nÉléments de mémoire long-terme (peuvent aider, ignore si hors-sujet) :\n" + souvenirs ) debug = { "id": ctx.id, "label": ctx.label, "icon": ctx.icon, "system_prompt": ctx.system_prompt, "blocks": debug_blocks, "docs": docs, "memories": memories, } return system, debug