This repository has no description
23 kB
529 lines
1"""hermes-voice: talk to Hermes through OpenAI's full-duplex gpt-live-1.
2
3The browser holds a WebRTC session with gpt-live-1, which listens and speaks at
4the same time and chats on its own. Whenever the user asks for real work, the
5voice model emits a client delegation; the page hands the recent transcript to
6this server, which runs it as an ordinary Hermes turn on the API server
7(/v1/chat/completions, one X-Hermes-Session-Id per conversation) and streams
8the reply back. The page then gives the answer to the voice model, which says
9it aloud.
10
11This server exists so neither secret reaches the browser: it exchanges the
12page's SDP offer for a Live session with the OpenAI key, and it calls Hermes
13with the API server's bearer token. That token is root-equivalent for the
14Hermes host, so whoever can use this server can drive the agent: every API
15route requires the page's own access token (--access-token-file), and serving
16beyond loopback without one is refused.
17
18Vendor contract: https://developers.openai.com/api/docs/guides/live-delegation
19"""
20
21import argparse
22import hmac
23import json
24import logging
25import os
26import re
27import socket
28import time
29import urllib.error
30import urllib.request
31from http import HTTPStatus
32from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
33from pathlib import Path
34
35log = logging.getLogger("hermes-voice")
36
37OPENAI_BASE_URL = "https://api.openai.com/v1"
38DEFAULT_MODEL = "gpt-live-1"
39DEFAULT_VOICE = "marin"
40# gpt-live-1 built-in voices (live-conversations guide voice table, plus the
41# realtime defaults marin and cedar).
42BUILT_IN_VOICES = (
43 "marin", "cedar", "quartz", "ripple", "vesper", "willow", "stone", "gleam",
44 "meridian", "bossa", "tempo", "beacon", "delta", "cinder",
45)
46# gpt-live-1 bills every second a session is open, silence included; the page
47# closes the session after this much inactivity.
48DEFAULT_IDLE_SECONDS = 90
49MAX_BODY = 256 * 1024
50# Hermes turns run tools for minutes; the API server streams tool progress, but
51# a single long tool call can be silent for a while.
52HERMES_TIMEOUT = 900
53SESSION_ID_RE = re.compile(r"^[A-Za-z0-9_-]{1,64}$")
54# Hermes run ids are "chatcmpl-<hex>", approval request ids hex.
55RUN_ID_RE = re.compile(r"^[A-Za-z0-9_.:-]{1,256}$")
56
57# Frontend persona, adapted from Hermes's own (tools/voice_live.py): role,
58# style and a labelled delegation policy. The backend carries the real
59# instructions, tools and memory.
60PERSONA = """You are Hermes, lass's warm, upbeat and genuinely helpful voice assistant. Sound friendly and glad to help: positive and encouraging, never curt, annoyed, bored or sarcastic, even when delivering bad news or when the user is short with you. Talk fast: speak at a rapid, energetic pace, like someone who talks quickly by nature, with no slow drawn-out words and no long pauses between sentences. Keep replies short; a brief warm acknowledgment before the facts is good.
61
62Backchannel policy: Use moderate backchannels. Acknowledge naturally without competing with the main response.
63
64Interruption policy: Stop speaking when the user interrupts. Listen to what they say.
65
66Delegation policy:
67Backend tools:
68- Hermes agent: a full AI agent with tools. It can run commands on lass's machines, read and edit files, manage the calendar, inspect the coding agents in herdr, browse the web, search, and remember things across sessions. It is the one who actually does work and knows facts.
69
70Delegate to the backend when:
71- The user asks a question that needs facts, current information, or careful reasoning.
72- The user asks you to do, check, find, make, fix, run or remember anything.
73- A correction changes work already requested.
74
75Do not delegate to the backend when:
76- The user greets you, makes small talk, or asks you to repeat a result already provided.
77- You need a brief clarification to understand the request.
78
79Delegate before giving an answer that depends on backend work. Do not guess the result while waiting; say briefly and cheerfully that you are on it, then wait for the result.
80
81Approval policy:
82Hermes sometimes needs the user's approval before it runs a command. When you are told so, explain in plain words what Hermes wants to do and ask whether to approve or deny it. When the user answers, delegate to the backend so the answer reaches Hermes. Never decide an approval yourself."""
83
84# Ephemeral system prompt for every delegated Hermes turn (Hermes layers system
85# messages on top of its core prompt without storing them).
86BACKEND_NOTE = """This turn is a delegation from a live spoken conversation. The user message is a speech transcript: it may contain mis-hearings, phonetic spellings of technical terms, hesitations and later corrections; use the latest intent. Your reply will be spoken aloud by a friendly voice model that paraphrases it: answer in warm, plain conversational sentences, keep it short (a few sentences unless the user asked for detail), no markdown, no lists, no code blocks, no URLs read out character by character. Do the work with your tools as usual; only the final facts need to be spoken. Report an action as done only after it actually succeeded."""
87
88
89def glossary_text(glossary):
90 """Glossary entries as prompt lines: `term` or `term (sounds like "…")`."""
91 lines = []
92 for entry in glossary:
93 if isinstance(entry, str):
94 lines.append(f"- {entry}")
95 else:
96 term = entry["term"]
97 spoken = entry.get("spoken")
98 lines.append(f'- {term} (sounds like "{spoken}")' if spoken else f"- {term}")
99 return "\n".join(lines)
100
101
102class Config:
103 def __init__(self, raw, openai_key, hermes_url, hermes_key):
104 self.model = raw.get("model", DEFAULT_MODEL)
105 self.voice = raw.get("voice", DEFAULT_VOICE)
106 self.idle_seconds = int(raw.get("idle_seconds", DEFAULT_IDLE_SECONDS))
107 self.glossary = raw.get("glossary", [])
108 self.extra_instructions = raw.get("instructions", "").strip()
109 self.openai_key = openai_key
110 self.hermes_url = hermes_url.rstrip("/")
111 self.hermes_key = hermes_key
112
113 def live_instructions(self):
114 parts = [PERSONA]
115 if self.glossary:
116 parts.append(
117 "Technical terms the user says often. Understand them when spoken and "
118 "say them like this:\n" + glossary_text(self.glossary)
119 )
120 if self.extra_instructions:
121 parts.append(self.extra_instructions)
122 return "\n\n".join(parts)
123
124 def backend_note(self):
125 if not self.glossary:
126 return BACKEND_NOTE
127 return (
128 BACKEND_NOTE
129 + "\n\nTerms the transcript may contain as phonetic misspellings:\n"
130 + glossary_text(self.glossary)
131 )
132
133 def voices(self):
134 """Voices the page may pick: the built-ins plus the configured one
135 (which may be a custom voice id)."""
136 return list(dict.fromkeys([self.voice, *BUILT_IN_VOICES]))
137
138
139def voice_param(voice):
140 # Custom voices are objects ({"id": "voice_…"}); built-ins are names.
141 if voice.startswith("voice_"):
142 return {"id": voice}
143 return voice
144
145
146def live_history(lines):
147 """Seed history for a new Live session from the page's transcript lines
148 (spoken dialogue only; Hermes answers were spoken through the voice)."""
149 items = []
150 dialogue = [line for line in lines if line.get("speaker") in ("user", "assistant")]
151 for line in dialogue[-24:]:
152 text = str(line.get("text", "")).strip()[:1200]
153 if not text:
154 continue
155 if line["speaker"] == "assistant":
156 content = {"type": "output_text", "text": text}
157 else:
158 content = {"type": "input_text", "text": text}
159 items.append({"type": "message", "role": line["speaker"], "content": [content]})
160 return items
161
162
163def create_live_session(config, sdp, history, voice):
164 session = {
165 "model": config.model,
166 "instructions": config.live_instructions(),
167 "audio": {"output": {"voice": voice_param(voice)}},
168 "delegation": {"type": "client"},
169 # The page is an untrusted frontend: it may only answer delegations and
170 # hang up, not rewrite the session.
171 "client": {
172 "data_channel": {
173 "allowed_client_events": [
174 "session.commentary.append",
175 "session.thinking.append",
176 "session.close",
177 ]
178 }
179 },
180 }
181 if history:
182 session["input"] = history
183 body = json.dumps(
184 {"session": session, "transport": {"type": "webrtc", "sdp": sdp}}
185 ).encode()
186 request = urllib.request.Request(
187 f"{OPENAI_BASE_URL}/live/sessions",
188 data=body,
189 method="POST",
190 headers={
191 "Authorization": f"Bearer {config.openai_key}",
192 "Content-Type": "application/json",
193 },
194 )
195 with urllib.request.urlopen(request, timeout=30) as response:
196 answer = json.load(response)
197 return {"session_id": answer["session"]["id"], "sdp": answer["transport"]["sdp"]}
198
199
200def transcript_message(lines):
201 rendered = []
202 for line in lines:
203 text = str(line.get("text", "")).strip()
204 if text:
205 who = "Assistant" if line.get("speaker") == "assistant" else "User"
206 rendered.append(f"{who}: {text}")
207 return "Spoken conversation since the last request, newest last:\n" + "\n".join(
208 rendered
209 )
210
211
212def iter_sse(response):
213 """Yield (event, data) pairs from a text/event-stream response."""
214 event, data = None, []
215 for raw in response:
216 line = raw.decode("utf-8", "replace").rstrip("\r\n")
217 if not line:
218 if data:
219 yield event, "\n".join(data)
220 event, data = None, []
221 elif line.startswith("event:"):
222 event = line[6:].strip()
223 elif line.startswith("data:"):
224 data.append(line[5:].lstrip())
225 if data:
226 yield event, "\n".join(data)
227
228
229def hermes_turn(config, hermes_session, lines):
230 """Run one Hermes turn; yield NDJSON-ready dicts (tool, delta, done)."""
231 body = json.dumps(
232 {
233 "messages": [
234 {"role": "system", "content": config.backend_note()},
235 {"role": "user", "content": transcript_message(lines)},
236 ],
237 "stream": True,
238 }
239 ).encode()
240 request = urllib.request.Request(
241 f"{config.hermes_url}/v1/chat/completions",
242 data=body,
243 method="POST",
244 headers={
245 "Authorization": f"Bearer {config.hermes_key}",
246 "Content-Type": "application/json",
247 "X-Hermes-Session-Id": hermes_session,
248 },
249 )
250 with urllib.request.urlopen(request, timeout=HERMES_TIMEOUT) as response:
251 for event, data in iter_sse(response):
252 if data == "[DONE]":
253 return
254 payload = json.loads(data)
255 if event == "hermes.tool.progress":
256 yield {
257 "type": "tool",
258 "status": payload.get("status"),
259 "label": payload.get("label") or payload.get("tool"),
260 }
261 continue
262 if event == "approval.request":
263 # Hermes blocks the turn until POST /v1/runs/{run_id}/approval
264 # answers (or its own approval timeout denies).
265 yield {
266 "type": "approval",
267 "run_id": payload.get("run_id"),
268 "request_id": payload.get("request_id"),
269 "command": payload.get("command"),
270 "description": payload.get("description"),
271 }
272 continue
273 if event:
274 continue
275 for choice in payload.get("choices", []):
276 text = choice.get("delta", {}).get("content")
277 if text:
278 yield {"type": "delta", "text": text}
279 if choice.get("finish_reason"):
280 done = {"type": "done", "finish_reason": choice["finish_reason"]}
281 if "error" in payload:
282 done["error"] = payload["error"].get("message")
283 yield done
284
285
286def hermes_approve(config, run_id, request_id, choice):
287 """Answer a pending approval of a streaming Hermes turn."""
288 answer = {"choice": choice}
289 if request_id:
290 answer["request_id"] = request_id
291 request = urllib.request.Request(
292 f"{config.hermes_url}/v1/runs/{run_id}/approval",
293 data=json.dumps(answer).encode(),
294 method="POST",
295 headers={
296 "Authorization": f"Bearer {config.hermes_key}",
297 "Content-Type": "application/json",
298 },
299 )
300 with urllib.request.urlopen(request, timeout=30) as response:
301 return json.load(response)
302
303
304class Handler(BaseHTTPRequestHandler):
305 server_version = "hermes-voice"
306 config: Config
307 static: Path
308 access_token: "str | None" = None
309
310 def authorized(self):
311 """Check the page's bearer token; answer 401 when it is wrong."""
312 if self.access_token is None:
313 return True
314 header = self.headers.get("Authorization", "")
315 scheme, _, token = header.partition(" ")
316 if scheme == "Bearer" and hmac.compare_digest(
317 token.encode(), self.access_token.encode()
318 ):
319 return True
320 self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "access token required"})
321 return False
322
323 def log_message(self, format, *args):
324 log.info("%s %s", self.address_string(), format % args)
325
326 def send_json(self, status, payload):
327 body = json.dumps(payload).encode()
328 self.send_response(status)
329 self.send_header("Content-Type", "application/json")
330 self.send_header("Content-Length", str(len(body)))
331 self.send_header("Cache-Control", "no-store")
332 self.end_headers()
333 self.wfile.write(body)
334
335 def read_json(self):
336 length = int(self.headers.get("Content-Length") or 0)
337 if length <= 0 or length > MAX_BODY:
338 raise ValueError("bad body length")
339 return json.loads(self.rfile.read(length))
340
341 def do_GET(self):
342 if self.path == "/config":
343 if not self.authorized():
344 return
345 self.send_json(
346 HTTPStatus.OK,
347 {
348 "model": self.config.model,
349 "voice": self.config.voice,
350 "voices": self.config.voices(),
351 "idle_seconds": self.config.idle_seconds,
352 },
353 )
354 return
355 names = {
356 "/": ("index.html", "text/html; charset=utf-8"),
357 "/app.js": ("app.js", "text/javascript; charset=utf-8"),
358 "/style.css": ("style.css", "text/css; charset=utf-8"),
359 }
360 if self.path not in names:
361 self.send_error(HTTPStatus.NOT_FOUND)
362 return
363 name, content_type = names[self.path]
364 body = (self.static / name).read_bytes()
365 self.send_response(HTTPStatus.OK)
366 self.send_header("Content-Type", content_type)
367 self.send_header("Content-Length", str(len(body)))
368 self.send_header("Cache-Control", "no-store")
369 self.end_headers()
370 self.wfile.write(body)
371
372 def do_POST(self):
373 if not self.authorized():
374 return
375 try:
376 request = self.read_json()
377 except ValueError:
378 self.send_json(HTTPStatus.BAD_REQUEST, {"error": "invalid JSON body"})
379 return
380 if self.path == "/session":
381 self.handle_session(request)
382 elif self.path == "/delegate":
383 self.handle_delegate(request)
384 elif self.path == "/approve":
385 self.handle_approve(request)
386 else:
387 self.send_error(HTTPStatus.NOT_FOUND)
388
389 def handle_session(self, request):
390 # The vendor's SDP parser needs the offer byte-exact (trailing CRLF too).
391 sdp = request.get("sdp")
392 if not isinstance(sdp, str) or not sdp.strip():
393 self.send_json(HTTPStatus.BAD_REQUEST, {"error": "sdp offer required"})
394 return
395 try:
396 voice = request.get("voice") or self.config.voice
397 if voice not in self.config.voices():
398 self.send_json(HTTPStatus.BAD_REQUEST, {"error": f"unknown voice {voice!r}"})
399 return
400 result = create_live_session(
401 self.config, sdp, live_history(request.get("history") or []), voice
402 )
403 except urllib.error.HTTPError as error:
404 detail = error.read().decode("utf-8", "replace")[:800]
405 log.warning("live session creation failed: %s %s", error.code, detail)
406 self.send_json(
407 HTTPStatus.BAD_GATEWAY,
408 {"error": f"OpenAI rejected the session ({error.code}): {detail}"},
409 )
410 return
411 except (urllib.error.URLError, OSError, KeyError, ValueError) as error:
412 log.warning("live session creation failed: %s", error)
413 self.send_json(HTTPStatus.BAD_GATEWAY, {"error": f"OpenAI: {error}"})
414 return
415 self.send_json(HTTPStatus.OK, result)
416
417 def handle_delegate(self, request):
418 hermes_session = request.get("hermes_session")
419 lines = request.get("lines")
420 if not isinstance(hermes_session, str) or not SESSION_ID_RE.match(hermes_session):
421 self.send_json(HTTPStatus.BAD_REQUEST, {"error": "invalid hermes_session"})
422 return
423 if not isinstance(lines, list) or not lines:
424 self.send_json(HTTPStatus.BAD_REQUEST, {"error": "lines required"})
425 return
426 self.send_response(HTTPStatus.OK)
427 self.send_header("Content-Type", "application/x-ndjson")
428 self.send_header("Cache-Control", "no-store")
429 self.send_header("X-Accel-Buffering", "no")
430 self.end_headers()
431 try:
432 try:
433 started = time.monotonic()
434 tools = 0
435 for item in hermes_turn(self.config, hermes_session, lines):
436 if item["type"] == "tool" and item["status"] == "running":
437 tools += 1
438 self.write_line(item)
439 log.info(
440 "hermes turn took %.1fs (%d tool calls)", time.monotonic() - started, tools
441 )
442 except urllib.error.HTTPError as error:
443 detail = error.read().decode("utf-8", "replace")[:400]
444 self.write_line({"type": "error", "message": f"Hermes {error.code}: {detail}"})
445 except (urllib.error.URLError, OSError, ValueError) as error:
446 self.write_line({"type": "error", "message": f"Hermes unreachable: {error}"})
447 except (BrokenPipeError, ConnectionResetError):
448 # The page abandoned this delegation (a newer one superseded it).
449 # Leaving the with-block closed the upstream stream, which makes
450 # Hermes interrupt the turn.
451 log.info("delegation for %s abandoned by the page", hermes_session)
452
453 def handle_approve(self, request):
454 run_id = request.get("run_id")
455 request_id = request.get("request_id")
456 choice = request.get("choice")
457 # No "always": a permanent allowlist entry is not a voice decision.
458 if (
459 choice not in ("once", "session", "deny")
460 or not isinstance(run_id, str)
461 or not RUN_ID_RE.match(run_id)
462 or not (
463 request_id is None
464 or (isinstance(request_id, str) and RUN_ID_RE.match(request_id))
465 )
466 ):
467 self.send_json(HTTPStatus.BAD_REQUEST, {"error": "invalid approval answer"})
468 return
469 try:
470 result = hermes_approve(self.config, run_id, request_id, choice)
471 except urllib.error.HTTPError as error:
472 detail = error.read().decode("utf-8", "replace")[:400]
473 self.send_json(HTTPStatus.BAD_GATEWAY, {"error": f"Hermes {error.code}: {detail}"})
474 return
475 except (urllib.error.URLError, OSError, ValueError) as error:
476 self.send_json(HTTPStatus.BAD_GATEWAY, {"error": f"Hermes unreachable: {error}"})
477 return
478 log.info("approval %s for run %s: %s", choice, run_id, result.get("resolved"))
479 self.send_json(HTTPStatus.OK, result)
480
481 def write_line(self, item):
482 self.wfile.write(json.dumps(item).encode() + b"\n")
483 self.wfile.flush()
484
485
486def main():
487 parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
488 parser.add_argument("--listen", default="127.0.0.1:8790", help="host:port to serve on")
489 parser.add_argument("--hermes-url", required=True, help="Hermes API server base URL")
490 parser.add_argument("--hermes-key-file", required=True, help="file holding API_SERVER_KEY")
491 parser.add_argument("--openai-key-file", required=True, help="file holding the OpenAI API key")
492 parser.add_argument("--access-token-file", help="file holding the token the page must present")
493 parser.add_argument("--config", help="JSON file: voice, model, idle_seconds, glossary, instructions")
494 parser.add_argument(
495 "--static",
496 default=os.environ.get("HERMES_VOICE_STATIC"),
497 help="directory with index.html, app.js, style.css (default: $HERMES_VOICE_STATIC)",
498 )
499 args = parser.parse_args()
500 logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
501
502 host, _, port = args.listen.rpartition(":")
503 host = host.strip("[]")
504 if args.access_token_file is None and host not in ("127.0.0.1", "::1", "localhost"):
505 parser.error("--access-token-file is required when not listening on loopback")
506
507 raw = json.loads(Path(args.config).read_text()) if args.config else {}
508 config = Config(
509 raw,
510 openai_key=Path(args.openai_key_file).read_text().strip(),
511 hermes_url=args.hermes_url,
512 hermes_key=Path(args.hermes_key_file).read_text().strip(),
513 )
514 Handler.config = config
515 Handler.static = Path(args.static)
516 if args.access_token_file:
517 Handler.access_token = Path(args.access_token_file).read_text().strip()
518 server_class = ThreadingHTTPServer
519 if ":" in host:
520 # "::" is dual-stack on Linux (bindv6only=0).
521 server_class = type("Server6", (ThreadingHTTPServer,), {"address_family": socket.AF_INET6})
522 server = server_class((host, int(port)), Handler)
523 server.daemon_threads = True
524 log.info("serving on http://%s", args.listen)
525 server.serve_forever()
526
527
528if __name__ == "__main__":
529 main()