320 lines
15 KiB
Python
320 lines
15 KiB
Python
"""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 random
|
|
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_SECONDS = 12.0
|
|
WAIT_NOTICES = (
|
|
"I'm still working on that. I'll let you know when it's ready.",
|
|
"This is taking a little longer. I'm on it.",
|
|
"I'm still looking into that. I'll let you know what I find.",
|
|
"I need a little more time on this. I'm still on it.",
|
|
)
|
|
|
|
|
|
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,
|
|
get_model_settings: Callable[[], Awaitable[dict[str, str]]] | 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._last_wait_notice: str | None = None
|
|
self._on_answer_start = on_answer_start
|
|
self._get_model_settings = get_model_settings
|
|
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
|
|
notice = random.choice([
|
|
text for text in WAIT_NOTICES if text != self._hermes._last_wait_notice
|
|
])
|
|
self._hermes._last_wait_notice = notice
|
|
await self._callback(self._hermes._on_wait_notice, notice)
|
|
|
|
async def _latency_notice(self) -> None:
|
|
# Status/tool-start events are not predictions of long latency: Hermes
|
|
# emits them even for ordinary replies. Only elapsed time without an
|
|
# answer warrants spoken feedback; UI progress remains independent.
|
|
await asyncio.sleep(WAIT_NOTICE_SECONDS)
|
|
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:
|
|
settings = {"model": "auto", "thinking_level": "none"}
|
|
if self._hermes._get_model_settings is not None:
|
|
try:
|
|
settings = await self._hermes._get_model_settings()
|
|
except Exception:
|
|
# Never fall back to stale preferences or run after revocation.
|
|
raise HermesLLMError("settings_unavailable") from None
|
|
model, effort = settings.get("model"), settings.get("thinking_level")
|
|
if (not isinstance(model, str) or not 1 <= len(model) <= 200
|
|
or not all(33 <= ord(c) <= 126 for c in model)
|
|
or effort not in ("none", "minimal", "low", "medium", "high", "xhigh")):
|
|
raise HermesLLMError("invalid_settings")
|
|
reasoning = {"enabled": False} if effort == "none" else {"enabled": True, "effort": effort}
|
|
payload = {
|
|
# Explicit provider is required: Hermes otherwise ignores bare model
|
|
# requests by default. Neither provider nor API origin is client input.
|
|
"model": model, "provider": "custom:hope-upstream", "stream": True,
|
|
"messages": _messages(self.chat_ctx),
|
|
"model_options": {"reasoning": reasoning},
|
|
}
|
|
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.")
|
|
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)
|
|
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")
|