mirror of
https://github.com/Alkatrazz24/Funk-lab.git
synced 2026-07-08 17:14:42 +02:00
ALI YESILKAYA
3ea4c3c706
Asa peut désormais AGIR sur le homelab quand on le demande explicitement, via un outil de la boucle agentique — mais jamais sans confirmation. - Outil admin_action(description) (contexte asa) : le LLM PROPOSE une action, n'exécute rien. brain.ask_with_tools gagne `confirm_tools` : un tel outil arrête la boucle et surface sa réponse (la question de confirmation). - _handle_agentic : stocke la proposition en pending par session ; au tour suivant « confirme » → agent.run_action → hermes-exec (hermes -z --yolo), « annule » → oubli. Réutilise le handshake + jeton du contexte agent. - admin_action n'est exposé que si _actions_available() (STT_ACTIONS_ENABLED + jeton) ; sinon retiré des schémas envoyés au modèle. - Factorisation du ctx_debug du visualiseur. 1 test unitaire (confirm_tools arrête la boucle). Serveur 0.9.0 ; doc stt.md + journal. Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
334 lines
14 KiB
Python
334 lines
14 KiB
Python
"""STT-server — API FastAPI pour les clients STT.
|
|
|
|
Endpoints :
|
|
GET /healthz → état du service
|
|
POST /v1/ask {text} → {reply} (requête AI, orchestrée vers LiteLLM)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import time
|
|
from contextlib import asynccontextmanager
|
|
|
|
import httpx
|
|
from fastapi import BackgroundTasks, FastAPI, HTTPException
|
|
from pydantic import BaseModel
|
|
|
|
from stt_server import __version__
|
|
from stt_server import brain
|
|
from stt_server.brain import ask as brain_ask
|
|
from stt_server.config import settings
|
|
from stt_server import agent
|
|
from stt_server import tools
|
|
from stt_server.contexts import CONTEXTS, assemble, get_context
|
|
from stt_server.knowledge import Knowledge
|
|
from stt_server.longterm import LongTermMemory
|
|
from stt_server.memory import SessionStore
|
|
from stt_server.sources import fetch_blocks, ghostfolio_phrase
|
|
|
|
log = logging.getLogger("stt_server")
|
|
# uvicorn ne configure que ses propres loggers : on attache notre handler en INFO
|
|
# pour que les lignes de timing (`ask … recall/gen/total`) sortent dans les logs du pod.
|
|
if not log.handlers:
|
|
log.setLevel(logging.INFO)
|
|
_h = logging.StreamHandler()
|
|
_h.setFormatter(logging.Formatter("%(levelname)s: %(name)s: %(message)s"))
|
|
log.addHandler(_h)
|
|
log.propagate = False
|
|
|
|
sessions = SessionStore()
|
|
longterm = LongTermMemory() if settings.memory_longterm else None
|
|
knowledge = Knowledge() if settings.docs_rag else None
|
|
|
|
# Contexte « agent » : action en attente de confirmation, par session.
|
|
pending_actions: dict[str, str] = {}
|
|
|
|
|
|
def _actions_available() -> bool:
|
|
return settings.actions_enabled and bool(settings.hermes_exec_token)
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
yield
|
|
# Fermeture propre des clients HTTP persistants (pooling).
|
|
await brain.aclose()
|
|
if longterm:
|
|
await longterm.aclose()
|
|
if knowledge:
|
|
await knowledge.aclose()
|
|
|
|
|
|
app = FastAPI(title="STT-server", version=__version__, lifespan=lifespan)
|
|
|
|
|
|
class AskRequest(BaseModel):
|
|
text: str
|
|
model: str | None = None # alias LiteLLM ; défaut serveur si absent
|
|
session_id: str | None = None # mémoire court-terme : fil de conversation
|
|
context: str | None = None # contexte présélectionné (funk/ghostfolio/…)
|
|
secrets: dict | None = None # jetons client (ex. {"ghostfolio_token": …})
|
|
|
|
|
|
class PortfolioRequest(BaseModel):
|
|
token: str | None = None # jeton Ghostfolio fourni par le client (sinon env)
|
|
|
|
|
|
class AskReply(BaseModel):
|
|
reply: str
|
|
model: str
|
|
context_id: str # contexte effectivement utilisé
|
|
context: dict | None = None # contexte assemblé (visualiseur HUD)
|
|
|
|
|
|
@app.get("/healthz")
|
|
async def healthz() -> dict:
|
|
return {"status": "ok", "version": __version__}
|
|
|
|
|
|
@app.get("/v1/models")
|
|
async def v1_models() -> dict:
|
|
return {"default": settings.model, "available": settings.allowed_models}
|
|
|
|
|
|
@app.get("/v1/contexts")
|
|
async def v1_contexts() -> dict:
|
|
"""Contextes présélectionnables (pour le sélecteur du HUD)."""
|
|
return {
|
|
"default": settings.default_context,
|
|
"contexts": [
|
|
{"id": c.id, "label": c.label, "icon": c.icon, "description": c.description}
|
|
for c in CONTEXTS.values()
|
|
# le contexte « agent » n'apparaît que si les actions sont activées + jeton
|
|
if c.id != "agent" or _actions_available()
|
|
],
|
|
}
|
|
|
|
|
|
@app.get("/v1/memory/health")
|
|
async def v1_memory_health() -> dict:
|
|
"""État de la mémoire long-terme (embeddings + Qdrant + collection), erreurs exposées."""
|
|
if not longterm:
|
|
return {"enabled": False}
|
|
return await longterm.health()
|
|
|
|
|
|
@app.post("/v1/portfolio")
|
|
async def v1_portfolio(req: PortfolioRequest) -> dict:
|
|
"""Valeur du portefeuille Ghostfolio (intent vocal client → serveur).
|
|
|
|
Le serveur fait l'appel Ghostfolio avec le jeton fourni par le client (sinon le
|
|
jeton serveur `STT_GHOSTFOLIO_TOKEN`). Un seul code Ghostfolio, lecture seule.
|
|
"""
|
|
async with httpx.AsyncClient() as c:
|
|
summary = await ghostfolio_phrase(c, req.token)
|
|
return {"summary": summary}
|
|
|
|
|
|
@app.post("/v1/reset")
|
|
async def v1_reset(req: AskRequest) -> dict:
|
|
if req.session_id:
|
|
sessions.reset(req.session_id)
|
|
return {"status": "reset"}
|
|
|
|
|
|
def _agent_context_debug(detail: dict) -> dict:
|
|
"""Structure de visualisation pour le contexte agent (transparence du handshake)."""
|
|
return {
|
|
"id": "agent", "label": "Agent (actions)", "icon": "🤖",
|
|
"system_prompt": "Flux agent : la demande est confirmée puis exécutée par "
|
|
"le vrai agent Hermes (hermes -z --yolo).",
|
|
"blocks": [{"source": "agent", "title": "Action", "text": str(detail)}],
|
|
"docs": [], "memories": [],
|
|
}
|
|
|
|
|
|
async def _handle_agent(req: AskRequest, text: str) -> AskReply:
|
|
"""Handshake de confirmation puis exécution via hermes-exec. Court-circuite le LLM."""
|
|
sid = req.session_id or "anon"
|
|
if not _actions_available():
|
|
return AskReply(
|
|
reply="Les actions sont désactivées côté serveur.",
|
|
model=settings.model, context_id="agent",
|
|
context=_agent_context_debug({"disabled": True}),
|
|
)
|
|
pending = pending_actions.get(sid)
|
|
if pending and agent.is_confirmation(text):
|
|
pending_actions.pop(sid, None)
|
|
out = await agent.run_action(pending)
|
|
log.info("agent EXEC sid=%s prompt=%r", sid, pending[:120])
|
|
return AskReply(reply=out, model=settings.model, context_id="agent",
|
|
context=_agent_context_debug({"executed": pending, "result": out[:300]}))
|
|
if pending and agent.is_cancellation(text):
|
|
pending_actions.pop(sid, None)
|
|
return AskReply(reply="C'est annulé.", model=settings.model, context_id="agent",
|
|
context=_agent_context_debug({"cancelled": pending}))
|
|
# nouvelle demande (ou remplacement) → on stocke et on demande confirmation
|
|
pending_actions[sid] = text
|
|
reply = f"Tu veux que j'exécute : « {text} » ? Dis « confirme » ou « annule »."
|
|
return AskReply(reply=reply, model=settings.model, context_id="agent",
|
|
context=_agent_context_debug({"pending": text}))
|
|
|
|
|
|
def _tool_blocks(trace: list[dict]) -> list[dict]:
|
|
"""Trace d'outils → blocs du visualiseur HUD (un bloc par appel)."""
|
|
return [
|
|
{
|
|
"source": f"tool:{t['name']}",
|
|
"title": f"🔧 {t['name']}({', '.join(f'{k}={v}' for k, v in t['args'].items())})",
|
|
"text": t["result"],
|
|
}
|
|
for t in trace
|
|
]
|
|
|
|
|
|
def _agentic_ctx_debug(ctx, memories: list[str], trace: list[dict]) -> dict:
|
|
return {
|
|
"id": ctx.id, "label": ctx.label, "icon": ctx.icon,
|
|
"system_prompt": ctx.system_prompt,
|
|
"blocks": _tool_blocks(trace), "docs": [], "memories": memories,
|
|
}
|
|
|
|
|
|
async def _handle_agentic(
|
|
req: "AskRequest", text: str, ctx, model: str, background: BackgroundTasks
|
|
) -> "AskReply":
|
|
"""Contexte à outils : recall mémoire (perso) + boucle de function-calling (état live).
|
|
|
|
La doc n'est plus injectée d'office (elle est devenue l'outil `search_docs`) ; on garde
|
|
le recall long-terme pour la personnalisation. Le trace des outils alimente le visualiseur.
|
|
Phase 3 : si une action admin a été proposée au tour précédent, ce tour la confirme/annule.
|
|
"""
|
|
t0 = time.perf_counter()
|
|
sid = req.session_id or "anon"
|
|
# Confirmation d'une action admin proposée au tour précédent (handshake, hors LLM).
|
|
pending = pending_actions.get(sid)
|
|
if pending and _actions_available():
|
|
if agent.is_confirmation(text):
|
|
pending_actions.pop(sid, None)
|
|
out = await agent.run_action(pending)
|
|
log.info("asa EXEC sid=%s prompt=%r", sid, pending[:120])
|
|
trace = [{"name": "admin_action", "args": {"description": pending}, "result": out}]
|
|
return AskReply(reply=out, model=model, context_id=ctx.id,
|
|
context=_agentic_ctx_debug(ctx, [], trace))
|
|
if agent.is_cancellation(text):
|
|
pending_actions.pop(sid, None)
|
|
return AskReply(reply="C'est annulé.", model=model, context_id=ctx.id,
|
|
context=_agentic_ctx_debug(ctx, [], []))
|
|
# Ni confirmation ni annulation → l'utilisateur est passé à autre chose : on oublie.
|
|
pending_actions.pop(sid, None)
|
|
history = sessions.history(req.session_id) if req.session_id else None
|
|
memories, qvec = await longterm.recall(text) if longterm else ([], None)
|
|
system = ctx.system_prompt
|
|
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
|
|
)
|
|
# admin_action n'est exposé que si les actions sont activées (opt-in + jeton).
|
|
tool_names = tuple(
|
|
t for t in ctx.tools if t != "admin_action" or _actions_available()
|
|
)
|
|
schemas = tools.schemas_for(tool_names)
|
|
t_recall = time.perf_counter()
|
|
async with httpx.AsyncClient() as client:
|
|
deps = tools.ToolDeps(client=client, knowledge=knowledge, qvec=qvec)
|
|
|
|
async def _dispatch(name: str, args: dict) -> str:
|
|
# admin_action : on n'exécute PAS — on mémorise la proposition et on demande
|
|
# confirmation (la boucle s'arrête via confirm_tools). Exécution = tour suivant.
|
|
if name == "admin_action":
|
|
desc = (args.get("description") or "").strip()
|
|
if not desc:
|
|
return "Aucune action décrite."
|
|
pending_actions[sid] = desc
|
|
return (f"Tu veux que j'exécute : « {desc} » ? "
|
|
"Dis « confirme » pour lancer, ou « annule ».")
|
|
return await tools.run(name, args, deps)
|
|
|
|
try:
|
|
reply, tool_trace = await brain.ask_with_tools(
|
|
text, system, schemas=schemas, dispatch=_dispatch,
|
|
model=model, history=history, confirm_tools=("admin_action",),
|
|
)
|
|
except httpx.HTTPError as e:
|
|
raise HTTPException(status_code=502, detail=f"upstream LiteLLM : {e}") from e
|
|
t_gen = time.perf_counter()
|
|
if req.session_id:
|
|
sessions.add(req.session_id, "user", text)
|
|
sessions.add(req.session_id, "assistant", reply)
|
|
if longterm:
|
|
background.add_task(longterm.store, req.session_id or "anon", text, qvec)
|
|
log.info(
|
|
"ask ctx=%s model=%s recall=%.0fms loop=%.0fms total=%.0fms mem=%d tools=%d",
|
|
ctx.id, model, (t_recall - t0) * 1000, (t_gen - t_recall) * 1000,
|
|
(t_gen - t0) * 1000, len(memories), len(tool_trace),
|
|
)
|
|
# Visualiseur HUD : chaque appel d'outil = un bloc (réutilise l'UI existante).
|
|
return AskReply(reply=reply, model=model, context_id=ctx.id,
|
|
context=_agentic_ctx_debug(ctx, memories, tool_trace))
|
|
|
|
|
|
@app.post("/v1/ask", response_model=AskReply)
|
|
async def v1_ask(req: AskRequest, background: BackgroundTasks) -> AskReply:
|
|
text = req.text.strip()
|
|
if not text:
|
|
raise HTTPException(status_code=400, detail="text vide")
|
|
model = req.model or settings.model
|
|
if model not in settings.allowed_models:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"modèle '{model}' non autorisé ; dispo : {settings.allowed_models}",
|
|
)
|
|
ctx = get_context(req.context)
|
|
# Contexte « agent » : flux dédié (confirmation 2 temps → hermes-exec), sans LLM.
|
|
if ctx.id == "agent":
|
|
return await _handle_agent(req, text)
|
|
# Contexte à outils (« asa ») : boucle de function-calling (le modèle sonde l'état live).
|
|
if ctx.tools:
|
|
return await _handle_agentic(req, text, ctx, model, background)
|
|
t0 = time.perf_counter()
|
|
history = sessions.history(req.session_id) if req.session_id else None
|
|
# recall : timeout serré, dégrade vite ; renvoie aussi le vecteur (réutilisé ci-dessous)
|
|
memories, qvec = await longterm.recall(text) if longterm else ([], None)
|
|
# RAG doc : seulement si le contexte le demande (réutilise qvec, même embedder nomic)
|
|
docs = []
|
|
if "docs" in ctx.sources and knowledge:
|
|
docs = await knowledge.search(text, qvec)
|
|
# Sources live du contexte (Ghostfolio / Prometheus / Alertmanager) — best-effort
|
|
blocks: list = []
|
|
live = tuple(s for s in ctx.sources if s != "docs")
|
|
if live:
|
|
async with httpx.AsyncClient() as c:
|
|
blocks = await fetch_blocks(c, live, secrets=req.secrets)
|
|
t_recall = time.perf_counter()
|
|
system, ctx_debug = assemble(ctx, blocks=blocks, docs=docs, memories=memories)
|
|
try:
|
|
reply = await brain_ask(text, system, model, history)
|
|
except httpx.HTTPError as e:
|
|
raise HTTPException(status_code=502, detail=f"upstream LiteLLM : {e}") from e
|
|
t_gen = time.perf_counter()
|
|
if req.session_id:
|
|
sessions.add(req.session_id, "user", text)
|
|
sessions.add(req.session_id, "assistant", reply)
|
|
# store : APRÈS la réponse (BackgroundTasks) → hors latence perçue, et on réutilise qvec
|
|
if longterm:
|
|
background.add_task(longterm.store, req.session_id or "anon", text, qvec)
|
|
log.info(
|
|
"ask ctx=%s model=%s recall+src=%.0fms gen=%.0fms total=%.0fms mem=%d docs=%d blocks=%d",
|
|
ctx.id, model,
|
|
(t_recall - t0) * 1000,
|
|
(t_gen - t_recall) * 1000,
|
|
(t_gen - t0) * 1000,
|
|
len(memories), len(docs), len(blocks),
|
|
)
|
|
return AskReply(reply=reply, model=model, context_id=ctx.id, context=ctx_debug)
|
|
|
|
|
|
def run() -> None:
|
|
"""Entrypoint `stt-server` (dev local). En prod : uvicorn via le conteneur."""
|
|
import uvicorn
|
|
|
|
uvicorn.run(app, host="0.0.0.0", port=8000) # noqa: S104 — service interne au cluster
|