"""LiveKit text streaming bridge to a trusted, per-user Hermes API. Hermes owns tool execution. Local tool schemas, tool choices, reasoning, and wait notices are never sent as chat history. Callers must use the notice callback to speak out-of-band. LiveKit session.say(..., add_to_chat_ctx=False) keeps history clean but queues behind the current response; prompt notices require independently managed audio output (never interrupt the Hermes turn). """ from __future__ import annotations import asyncio import json import math import uuid from collections.abc import AsyncIterator, Awaitable, Callable from dataclasses import replace from typing import Any import httpx from livekit.agents import APIError, llm as lk_llm from livekit.agents.types import ( APIConnectOptions, DEFAULT_API_CONNECT_OPTIONS, NOT_GIVEN, NotGivenOr, ) ProgressCallback = Callable[[dict[str, Any]], Awaitable[None]] WaitNoticeCallback = Callable[[str], Awaitable[None]] WAIT_NOTICE = "I'm working on that. I'll let you know when it's ready." class HermesLLMError(APIError): """Stable, redacted, non-retryable failure; ``code`` identifies the cause.""" def __init__(self, code: str) -> None: self.code = code super().__init__(f"Hermes request failed ({code})", retryable=False) def _messages(chat_ctx: lk_llm.ChatContext) -> list[dict[str, str]]: messages = [] for item in chat_ctx.items: if not isinstance(item, lk_llm.ChatMessage): continue if item.role not in ("system", "user", "assistant"): continue # Preserve text exactly, including the system prompt; drop media and # function-call/result items rather than stringifying sensitive objects. parts = [part for part in item.content if isinstance(part, str)] if parts: messages.append({"role": item.role, "content": "\n".join(parts)}) return messages async def _sse(response: httpx.Response) -> AsyncIterator[tuple[str, str]]: """HTTPX handles split UTF-8/CRLF; this layer handles SSE event framing.""" event = "message" data: list[str] = [] size = 0 async for line in response.aiter_lines(): if not line: if data: yield event, "\n".join(data) event, data, size = "message", [], 0 continue if line.startswith(":"): continue field, sep, value = line.partition(":") if sep and value.startswith(" "): value = value[1:] size += len(line) if size > 1024 * 1024: raise HermesLLMError("event_too_large") if field == "event": event = value elif field == "data": data.append(value) if data: yield event, "\n".join(data) class HermesLLM(lk_llm.LLM): """One per-user bridge, with stable room/session headers and no retries. ``base_url`` is trusted server configuration (origin, optionally ending in /v1), never browser input. ``on_progress`` receives assistant_progress dicts with request_id, status (working/completed/failed), and message. Callback failures do not replay or abort Hermes actions. An injected ``http_client`` is caller-owned; otherwise this instance owns and closes its HTTP client. ``on_answer_start`` is an optional async zero-argument hook invoked once, before the first content chunk, to quickly cancel/clear independent notice audio. Callbacks should schedule notice playback, not await full playout. ``timeout_seconds`` bounds the whole request, not just idle reads. """ def __init__( self, *, base_url: str, api_key: str, user_id: str, room_id: str, on_progress: ProgressCallback, on_wait_notice: WaitNoticeCallback | None = None, on_answer_start: Callable[[], Awaitable[None]] | None = None, timeout_seconds: float = 600.0, http_client: httpx.AsyncClient | None = None, ) -> None: super().__init__() url = httpx.URL(base_url) if url.scheme not in ("http", "https") or not url.host or url.userinfo or url.query or url.fragment: raise ValueError("base_url must be a trusted HTTP(S) API URL") if not api_key or not user_id or not room_id: raise ValueError("api_key, user_id and room_id are required") if any("\r" in value or "\n" in value for value in (api_key, user_id, room_id)): raise ValueError("credentials and session identifiers cannot contain newlines") if not math.isfinite(timeout_seconds) or timeout_seconds <= 0: raise ValueError("timeout_seconds must be finite and positive") root = base_url.rstrip("/") self._url = root + ("/chat/completions" if root.endswith("/v1") else "/v1/chat/completions") self._headers = { "Authorization": f"Bearer {api_key}", "X-Hermes-Session-Id": room_id, "X-Hermes-Session-Key": f"hope:{user_id}", "Accept": "text/event-stream", } self._on_progress = on_progress self._on_wait_notice = on_wait_notice self._on_answer_start = on_answer_start self._timeout_seconds = timeout_seconds self._owns_client = http_client is None self._client = http_client if http_client is not None else httpx.AsyncClient(follow_redirects=False) @property def model(self) -> str: return "hermes-agent" @property def provider(self) -> str: return "hermes" def chat( self, *, chat_ctx: lk_llm.ChatContext, tools: list[lk_llm.Tool] | None = None, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: NotGivenOr[lk_llm.ToolChoice] = NOT_GIVEN, extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN, ) -> HermesLLMStream: # Intentionally ignore all local tools/options; none can override the # trusted model, messages, reasoning policy, or session scope. return HermesLLMStream(self, chat_ctx=chat_ctx, conn_options=replace(conn_options, max_retry=0)) async def aclose(self) -> None: await super().aclose() if self._owns_client: await self._client.aclose() class HermesLLMStream(lk_llm.LLMStream): def __init__(self, llm: HermesLLM, *, chat_ctx: lk_llm.ChatContext, conn_options: APIConnectOptions) -> None: self._hermes = llm self.request_id = uuid.uuid4().hex self._has_answer = False self._noticed = False super().__init__(llm, chat_ctx=chat_ctx, tools=[], conn_options=conn_options) async def _callback(self, callback: Callable[..., Awaitable[None]], *args: Any) -> None: try: await asyncio.wait_for(callback(*args), timeout=5.0) except Exception: # UI/TTS failure is not a reason to retry an action. Never log # callback exceptions: they may contain credentials or payloads. pass async def _progress(self, status: str, message: str) -> None: await self._callback(self._hermes._on_progress, { "type": "assistant_progress", "request_id": self.request_id, "status": status, "message": message, }) async def _notice(self) -> None: if self._noticed or self._has_answer or self._hermes._on_wait_notice is None: return self._noticed = True # latch before awaiting, including competing timer await self._callback(self._hermes._on_wait_notice, WAIT_NOTICE) async def _latency_notice(self) -> None: await asyncio.sleep(3.0) if not self._has_answer: await self._progress("working", "Working on your request.") await self._notice() async def _run(self) -> None: timer = asyncio.create_task(self._latency_notice()) try: await asyncio.wait_for(self._consume(), timeout=self._hermes._timeout_seconds) await self._progress("completed", "Completed.") except asyncio.CancelledError: await self._progress("failed", "Request cancelled.") raise except Exception as exc: code = exc.code if isinstance(exc, HermesLLMError) else ( "timeout" if isinstance(exc, (asyncio.TimeoutError, httpx.TimeoutException)) else "transport_error" ) await self._progress("failed", f"Hermes request failed ({code}).") # Suppress upstream exceptions/bodies/URLs in the logged chain. raise HermesLLMError(code) from None finally: timer.cancel() await asyncio.gather(timer, return_exceptions=True) async def _consume(self) -> None: payload = { "model": "hermes-agent", "stream": True, "messages": _messages(self.chat_ctx), "model_options": {"reasoning": {"enabled": False}}, } finished = False async with self._hermes._client.stream( "POST", self._hermes._url, headers=self._hermes._headers, json=payload, timeout=httpx.Timeout(self._hermes._timeout_seconds, connect=10.0), follow_redirects=False, ) as response: if response.status_code != 200: raise HermesLLMError(f"http_{response.status_code}") if "text/event-stream" not in response.headers.get("content-type", "").lower(): raise HermesLLMError("invalid_content_type") async for event, raw in _sse(response): if raw.strip() == "[DONE]": finished = True break if event in ("approval.request", "hermes.approval.request"): # Voice has no approval-response transport. Fail immediately # instead of leaving an action waiting until the long timeout. raise HermesLLMError("approval_unsupported") if event not in ("message", "hermes.tool.progress", "hermes.status", "hermes.error", "error"): continue try: item = json.loads(raw) except (ValueError, TypeError): raise HermesLLMError("invalid_sse") from None if not isinstance(item, dict): raise HermesLLMError("invalid_sse") hermes = item.get("hermes") if event in ("error", "hermes.error") or item.get("error") or ( isinstance(hermes, dict) and hermes.get("error") ): raise HermesLLMError("upstream_error") if event == "hermes.status": if item.get("status") in ("failed", "error"): raise HermesLLMError("upstream_error") await self._progress("working", "Working on your request.") await self._notice() continue if event == "hermes.tool.progress": status = item.get("status", "working") if status in ("failed", "error"): raise HermesLLMError("tool_failed") message = item.get("message") if not isinstance(message, str): name = item.get("tool_name", item.get("tool", item.get("name"))) message = f"Working on {name}." if isinstance(name, str) else "Working on your request." await self._progress("working", message) if status not in ("completed", "complete", "success", "finished"): await self._notice() continue choices = item.get("choices", []) if not isinstance(choices, list): raise HermesLLMError("invalid_sse") for choice in choices: if not isinstance(choice, dict) or not isinstance(choice.get("delta", {}), dict): raise HermesLLMError("invalid_sse") if choice.get("index", 0) != 0: continue finish_reason = choice.get("finish_reason") if finish_reason in ("error", "failed", "content_filter"): raise HermesLLMError("upstream_error") if finish_reason in ("approval", "approval_required"): raise HermesLLMError("approval_unsupported") if finish_reason is not None: finished = True content = choice.get("delta", {}).get("content") # No reasoning fallback or local tool dispatch, ever. if isinstance(content, str) and content: if not self._has_answer: self._has_answer = True # latch before cancelling notice if self._hermes._on_answer_start is not None: await self._callback(self._hermes._on_answer_start) self._event_ch.send_nowait(lk_llm.ChatChunk( id=self.request_id, delta=lk_llm.ChoiceDelta(role="assistant", content=content), )) if not finished: raise HermesLLMError("incomplete_stream") if not self._has_answer: raise HermesLLMError("empty_response")