sam/family_home_lab
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

dsh: Option 1 gateway — home-shaped chat requests (announce, HA, shopping, calendar, music, messaging) route raw to Jervis; normal chat untouched

Browse Source
This commit is contained in:
Sam Rolfe
2026-09-20 06:55:04 +10:00
parent 1d3e2b6ab7
commit 1535e14c6f
1 changed files with 33 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

90
dsh/app.py
View File

@@ -42,13 +42,7 @@ SYSTEM_PROMPT = os.getenv(
"DSH_SYSTEM_PROMPT",
f"You are {USER_NAME}'s helpful DeepSeek assistant on the family home lab. "
"Be clear, safe and concise. Never expose server file contents or system "
"secrets. If asked something unsafe, decline politely.\n\n"
"You can speak aloud through the house speakers using the speak function "
"when the user asks to announce/say something over them (e.g. 'announce "
"X', 'tell Finn that...', 'say dinner is ready', or requests a voice "
"profile like Donald Trump or HAL 9000). Pass the exact message as text "
"and the requested voice when given. Only fall back to writing the message "
"in chat if the tool is unavailable.",
"secrets. If asked something unsafe, decline politely.",
)
# --- vision (image ingest) -> OpenRouter gpt-5 (works; OmniRoute auto/best-vision
@@ -147,62 +141,43 @@ _SUM_HEAD = (
)
_SPEAK_TOOLS = [{"type": "function", "function": {
"name": "speak",
"description": "Speak text aloud through the house speakers via the Jervis voice agent (Snapcast). Use when the user wants an announcement, wants someone told something out loud, or asks for a voice profile.",
"parameters": {"type": "object", "properties": {
"text": {"type": "string", "description": "The exact text to say"},
"voice": {"type": "string", "description": "Optional voice profile name e.g. 'Donald Trump', 'HAL 9000', 'Picard'. Omit for the default voice."}
}, "required": ["text"]}
}}]
def _looks_speakish(msg: str) -> bool:
def _looks_homeish(msg: str) -> bool:
"""Broad gate: hand the message to Jervis when it talks about the home
(announcements, HA devices, shopping/calendar/timers, music, messaging,
fish, time/weather). Jervis decides answer-vs-act and speaks confirmations."""
m = (msg or "").lower()
return any(h in m for h in (
# speak / announce
"announce", "speak", "speaker", "shout", "out loud",
"say on ", "say over ", "say through ", "over the speakers",
"on the speakers", "through the speakers",
"tell finn", "tell jo", "tell harry", "tell sam", "tell the family",
"say on ", "say over ", "say through ", "on the speakers",
"over the speakers", "through the speakers",
# home assistant actions
"turn on ", "turn off ", "switch on ", "switch off ",
"lights", "lamp", "dim ", "scene", "thermostat", "climate",
"aircon", "heating", "curtains", "blinds", "garage",
"feed the fish",
# lists / calendar / reminders
"shopping", "bunnings", "add to my calendar", "calendar",
"remind me", "reminder", "set a timer", "timer for ",
# music
"play some music", "play music", "play a song", "play ",
"pause the music", "music ", "spotify", "mopidy", "volume",
# messaging
"text ", "whatsapp", "message ", "notify ", "email ",
))
async def _tool_speak(hist: list, model: str, headers: dict) -> str | None:
"""Tool-decide pass: offer the model a 'speak' function. Returns a display
string if an announcement was made, else None (no speak requested)."""
payload = {"model": model, "stream": False, "temperature": 0.4,
"tools": _SPEAK_TOOLS, "tool_choice": "auto",
"messages": [{"role": "system", "content": SYSTEM_PROMPT}] + hist}
async def _jervis_act(text: str) -> str | None:
"""Forward a raw home command to Jervis (/voice). Returns Jervis's spoken
reply for the chat bubble, or None if Jervis is unreachable/errored."""
try:
async with httpx.AsyncClient(timeout=120) as client:
r = await client.post(f"{LLM_BASE.rstrip('/')}/chat/completions",
json=payload, headers=headers)
r.raise_for_status()
msg = r.json()["choices"][0]["message"]
r = await client.post(VOICE_AGENT_URL, json={"text": (text or "")[:500]})
if r.status_code != 200:
return None
return (r.json().get("reply") or "").strip()[:500] or None
except Exception:
return None
for c in msg.get("tool_calls") or []:
fn = c.get("function") or {}
if fn.get("name") != "speak":
continue
try:
args = json.loads(fn.get("arguments") or "{}")
except Exception:
args = {}
text = (args.get("text") or "").strip()[:400]
if not text:
continue
voice = (args.get("voice") or "").strip()
command = (f"Speak in the {voice} voice: {text}"
if voice and voice.lower() not in ("default",) else text)
try:
async with httpx.AsyncClient(timeout=90) as client:
rr = await client.post(VOICE_AGENT_URL, json={"text": command})
ok = rr.status_code == 200
except Exception as exc:
return f"⚠️ Couldn't reach Jervis to speak: {exc}"
return f"🔊 {text}" if ok else "⚠️ Jervis couldn't speak it just now."
return None
s = re.sub(r"(?is)<(script|style|head|noscript|svg)[^>]*>.*?</\1>", " ", s)
s = re.sub(r"(?s)<[^>]+>", " ", s)
@@ -370,10 +345,11 @@ async def chat(message: Annotated[str, Form()],
if base == LLM_BASE:
# OmniRoute requires a session id for opencode-go models (else 400 "MissingSessionID")
headers["x-opencode-session"] = f"dsh-{os.getpid()}-{uuid4().hex[:8]}"
# Speak pass: only when the user sounds like they want an announcement.
# Non-streaming tool-decide, then stream a short confirmation.
if _looks_speakish(message):
spoke = await _tool_speak(hist, active_model, headers)
# Home pass: announce/HA/shopping/calendar/music/messaging requests
# go straight to Jervis (full HA tool loop, speaks confirmations).
# Jervis decides answer-vs-act; dsh just relays its reply.
if _looks_homeish(message):
spoke = await _jervis_act(message)
if spoke:
hist.append({"role": "assistant", "content": spoke})
save_session(sid, hist, session.get("model"))