"""LiveKit speech adapter to the durable Hope coordinator, never a reasoning loop.""" from __future__ import annotations import asyncio import json import uuid from dataclasses import replace import httpx from livekit.agents import llm from livekit.agents.types import DEFAULT_API_CONNECT_OPTIONS try: from agent.hermes_llm import HermesLLMError, _sse except ImportError: from hermes_llm import HermesLLMError, _sse class ConversationLLM(llm.LLM): def __init__(self, *, control_url, secret, room_id, on_progress, on_wait_notice=None, on_answer_start=None): super().__init__() self.root = control_url.rstrip('/') + '/internal/chat/rooms/' + room_id self.headers = {'X-Control-Secret': secret} self.client = httpx.AsyncClient(follow_redirects=False, timeout=httpx.Timeout(660, connect=10)) self.on_progress, self.on_wait_notice, self.on_answer_start = on_progress, on_wait_notice, on_answer_start @property def model(self): return 'hermes-agent' @property def provider(self): return 'hermes' def chat(self, *, chat_ctx, tools=None, conn_options=DEFAULT_API_CONNECT_OPTIONS, **kwargs): return ConversationStream(self, chat_ctx, replace(conn_options, max_retry=0)) async def aclose(self): await super().aclose() await self.client.aclose() class ConversationStream(llm.LLMStream): def __init__(self, bridge, chat_ctx, conn_options): self.bridge = bridge self.operation_id = next((m.id for m in reversed(chat_ctx.items) if isinstance(m, llm.ChatMessage) and m.role == 'user' and getattr(m, 'id', None)), uuid.uuid4().hex) self.run_id = None self.answer_started = False self._close_task = None super().__init__(bridge, chat_ctx=chat_ctx, tools=[], conn_options=conn_options) async def aclose(self): # LiveKit can close a stream from interruption and teardown concurrently. # One underlying closer prevents a second cancellation from aborting # the first closer's uncertain-admission reconciliation. if self._close_task is None: self._close_task = asyncio.create_task(super().aclose()) await asyncio.shield(self._close_task) async def callback(self, callback, *args): if callback: try: await asyncio.wait_for(callback(*args), 5) except Exception: pass # feedback failure never repeats a tool action async def progress(self, status, message): await self.callback(self.bridge.on_progress, {'type': 'assistant_progress', 'request_id': self.run_id or self.operation_id, 'status': status, 'message': message}) async def _run(self): user = next((m for m in reversed(self.chat_ctx.items) if isinstance(m, llm.ChatMessage) and m.role == 'user'), None) text = '\n'.join(p for p in user.content if isinstance(p, str)) if user else '' if not text.strip(): raise HermesLLMError('empty_voice_turn') timer = None admission = None admission_rejected = False try: # Do not abandon uncertain admission on barge-in. A successfully # admitted remote action must be identified and cancelled, not left # running just because the speech request lost its response. admission = asyncio.create_task(self.bridge.client.post(self.bridge.root + '/messages', headers=self.bridge.headers, json={'content': text, 'idempotency_key': self.operation_id, 'modality': 'voice'}, timeout=httpx.Timeout(20, connect=5))) response = await asyncio.shield(admission) if response.status_code != 202: admission_rejected = 400 <= response.status_code < 500 raise HermesLLMError('voice_admission_failed') self.run_id = response.json()['run']['id'] await self.progress('working', 'Working on your request.') timer = asyncio.create_task(self.notice()) # Voice streams only completed public content. This avoids reading # a draft that the backend subsequently replaces or rejects. final = None async with self.bridge.client.stream('GET', self.bridge.root + '/runs/' + self.run_id + '/events', headers=self.bridge.headers) as stream: if stream.status_code != 200: raise HermesLLMError('voice_stream_failed') async for event, raw in _sse(stream): item = json.loads(raw) if event == 'progress': await self.progress('working', 'Working on your request.') if event == 'message': message = item.get('message', item) if message.get('role') == 'assistant' and message.get('status') == 'completed': final = message.get('content', '') if event in ('run', 'done'): run = item.get('run', item) if run.get('status') in ('failed', 'cancelled'): raise HermesLLMError('voice_run_' + run['status']) if run.get('status') == 'awaiting_approval': await self.progress('working', 'Approval needed. Open this conversation to review it.') if event == 'done': break if final: self.answer_started = True await self.callback(self.bridge.on_answer_start) self._event_ch.send_nowait(llm.ChatChunk(id=self.operation_id, delta=llm.ChoiceDelta(role='assistant', content=final))) else: raise HermesLLMError('empty_voice_answer') await self.progress('completed', 'Completed.') except asyncio.CancelledError: # Persist cancellation by operation identity even when admission's # response was lost. The coordinator tombstones the key, so a later # identical admission cannot start an uncancelled remote action. try: await asyncio.shield(self.bridge.client.post(self.bridge.root + '/operations/' + self.operation_id + '/cancel', headers=self.bridge.headers, timeout=httpx.Timeout(10, connect=5))) except Exception: pass if not self.run_id and admission: try: admitted = await asyncio.shield(admission) if admitted.status_code == 202: self.run_id = admitted.json()['run']['id'] except Exception: # A transport failure can be an uncertain acceptance. Retry # only the identical stable key; the coordinator deduplicates # admission, then return the run identity for cancellation. try: admitted = await self.bridge.client.post(self.bridge.root + '/messages', headers=self.bridge.headers, json={'content': text, 'idempotency_key': self.operation_id, 'modality': 'voice'}, timeout=httpx.Timeout(20, connect=5)) if admitted.status_code == 202: self.run_id = admitted.json()['run']['id'] except Exception: pass if self.run_id: try: await asyncio.shield(self.bridge.client.post(self.bridge.root + '/runs/' + self.run_id + '/cancel', headers=self.bridge.headers, timeout=httpx.Timeout(10, connect=5))) except Exception: pass await self.progress('failed', 'Request cancelled.') raise except Exception: # A failed HTTP receipt is not proof admission failed. Fail safely # by tombstoning the stable operation key, including when JSON/SSE # decoding failed after the coordinator already accepted the turn. if not admission_rejected: try: await self.bridge.client.post(self.bridge.root + '/operations/' + self.operation_id + '/cancel', headers=self.bridge.headers, timeout=httpx.Timeout(10, connect=5)) except Exception: pass await self.progress('failed', 'Hope could not complete this response.') raise HermesLLMError('voice_coordinator_failed') from None finally: if timer: timer.cancel() await asyncio.gather(timer, return_exceptions=True) async def notice(self): await asyncio.sleep(12) if not self.answer_started: await self.callback(self.bridge.on_wait_notice, "I'm still working on that. I'll let you know when it's ready.")