153 lines
8.4 KiB
Python
153 lines
8.4 KiB
Python
"""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.")
|