e0959e6fd7
Mounts agents router and schema init, adds VM123 checks, chat copilot, Desk UI module, isolated docker-compose staging on ports 8180/8192, and full spec documentation without touching production ports. Co-authored-by: Cursor <cursoragent@cursor.com>
87 lines
3 KiB
Python
87 lines
3 KiB
Python
"""Ollama VM123 + fallback — Spec 029 T0/T1."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
|
|
import httpx
|
|
|
|
OLLAMA_BASE_URL = os.getenv("OLLAMA_BASE_URL", "http://10.10.10.123:11434").rstrip("/")
|
|
AGENTIC_LLM_MODEL = os.getenv("AGENTIC_LLM_MODEL", "qwen2.5:7b-instruct")
|
|
AGENTIC_EMBED_MODEL = os.getenv("AGENTIC_EMBED_MODEL", "nomic-embed-text")
|
|
AGENTIC_LLM_ENABLED = os.getenv("AGENTIC_LLM_ENABLED", "false").lower() in ("1", "true", "yes")
|
|
|
|
|
|
def ollama_available() -> bool:
|
|
try:
|
|
with httpx.Client(timeout=3.0) as c:
|
|
return c.get(f"{OLLAMA_BASE_URL}/api/tags").status_code == 200
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def _chat(prompt: str, *, system: str | None = None, max_tokens: int = 800) -> tuple[str, str]:
|
|
if not AGENTIC_LLM_ENABLED or not ollama_available():
|
|
return ("", "t0")
|
|
messages = []
|
|
if system:
|
|
messages.append({"role": "system", "content": system})
|
|
messages.append({"role": "user", "content": prompt})
|
|
try:
|
|
with httpx.Client(timeout=120.0) as c:
|
|
r = c.post(
|
|
f"{OLLAMA_BASE_URL}/api/chat",
|
|
json={"model": AGENTIC_LLM_MODEL, "messages": messages, "stream": False},
|
|
)
|
|
if r.status_code == 200:
|
|
txt = (r.json().get("message") or {}).get("content", "").strip()
|
|
if txt:
|
|
return txt, AGENTIC_LLM_MODEL
|
|
except Exception:
|
|
pass
|
|
return ("", "t0-fallback")
|
|
|
|
|
|
def advise_human_action(
|
|
*, finding_title: str, finding_detail: str, kb_snippets: list[str] | None = None
|
|
) -> tuple[str, str]:
|
|
prompt = (
|
|
"Advisor Agentic Ops Ligbox. Português BR, máx 6 frases. O que fazer AGORA?\n"
|
|
f"Problema: {finding_title}\nDetalhe: {finding_detail}\n"
|
|
f"KB: {'---'.join(kb_snippets or [])[:2500] or 'N/A'}"
|
|
)
|
|
txt, model = _chat(prompt)
|
|
if txt:
|
|
return txt, model
|
|
return (f"Investigar manualmente: {finding_title}", "t0")
|
|
|
|
|
|
def chat_context(
|
|
*,
|
|
question: str,
|
|
kb_snippets: list[str] | None = None,
|
|
findings_summary: str | None = None,
|
|
user_role: str = "technician",
|
|
) -> tuple[str, str]:
|
|
"""T1 — resposta contextual para janela Desk / bot interno."""
|
|
system = (
|
|
"És o copiloto Agentic Ops da Ligbox (VM112 wizard, VM122 Desk, VM123 finance). "
|
|
"Responde em português BR, objectivo, com passos acionáveis. "
|
|
"Nunca inventes credenciais. Se não souberes, diz o que verificar."
|
|
)
|
|
ctx = []
|
|
if findings_summary:
|
|
ctx.append(f"Findings abertos:\n{findings_summary[:2000]}")
|
|
if kb_snippets:
|
|
ctx.append("KB:\n" + "\n---\n".join(kb_snippets[:6])[:4000])
|
|
prompt = (
|
|
f"Perfil utilizador: {user_role}\n"
|
|
f"Contexto ops:\n{chr(10).join(ctx) or 'N/A'}\n\n"
|
|
f"Pergunta: {question}"
|
|
)
|
|
txt, model = _chat(prompt, system=system)
|
|
if txt:
|
|
return txt, model
|
|
return (
|
|
"Modo T0 activo — LLM indisponível. Consulte findings e audit log no painel Agentic Ops.",
|
|
"t0",
|
|
)
|