Spaces:
Paused
Paused
File size: 2,976 Bytes
f1c3ff2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 | """Help chat: try optional Ollama, always fall back to FAQ matcher (CPU-safe)."""
from __future__ import annotations
import os
from typing import Any, Dict, List, Optional
import httpx
from sensorflow.help.matcher import format_fallback_answer, match_faq
def _llm_endpoints() -> List[Dict[str, str]]:
endpoints: List[Dict[str, str]] = []
url = os.environ.get("SENSORFLOW_LLM_URL")
if url:
endpoints.append(
{
"url": url if url.endswith("/api/chat") else url.rstrip("/") + "/api/chat",
"model": os.environ.get("SENSORFLOW_LLM_MODEL", "gemma3:latest"),
}
)
# Local Ollama last — optional enrichment only
endpoints.append({"url": "http://localhost:11434/api/chat", "model": "gemma3:latest"})
return endpoints
def _build_prompt(question: str, context_blocks: List[str], page_id: Optional[str]) -> str:
ctx = "\n\n".join(context_blocks[:4])
page_line = f"Current UI page id: {page_id}\n" if page_id else ""
return (
"You are the Sensorflow Studio in-app help assistant. Answer briefly and accurately "
"using ONLY the context below. If the context is insufficient, say what the user "
"should open in the Help menu. Do not invent APIs or metrics.\n\n"
f"{page_line}"
f"CONTEXT:\n{ctx}\n\n"
f"USER QUESTION: {question}\n\n"
"Answer in plain language (2–6 short sentences). Mention relevant page names when useful."
)
def answer_help_question(question: str, page_id: Optional[str] = None) -> Dict[str, Any]:
q = (question or "").strip()
if not q:
return {
"answer": "Ask a question about a Sensorflow Studio page or feature.",
"sources": [],
"provider": "faq_offline",
"page_hint": page_id,
}
matches = match_faq(q, page_id=page_id, top_k=4)
offline = format_fallback_answer(q, matches, page_id=page_id)
context_blocks = [f"[{doc.kind}] {doc.title}: {doc.text[:700]}" for doc, _ in matches]
if not context_blocks:
return offline
prompt = _build_prompt(q, context_blocks, page_id)
for ep in _llm_endpoints():
try:
res = httpx.post(
ep["url"],
json={
"model": ep["model"],
"messages": [{"role": "user", "content": prompt}],
"stream": False,
},
timeout=8.0,
)
if res.status_code == 200:
text = (res.json().get("message") or {}).get("content") or ""
if text.strip():
return {
"answer": text.strip(),
"sources": offline["sources"],
"provider": ep["url"],
"page_hint": offline.get("page_hint"),
}
except Exception:
continue
return offline
|