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