File size: 3,423 Bytes
f440f03 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 | """Tests realtime voice assistant arhitektūrai."""
from __future__ import annotations
import pytest
from maris_core.voice.livekit_agent import (
build_livekit_agent_metadata,
load_livekit_voice_agent_settings,
require_livekit_agents_runtime,
)
from maris_core.voice.realtime import RealtimeVoiceAssistant, VoiceSessionState
class _FakeSttProvider:
def stream_transcript(self, audio_chunks, *, language: str, session_id: str):
del audio_chunks, language, session_id
yield {"text": "Sveiki", "is_final": False}
yield {"text": "Sveiki, palīdzi man", "is_final": True}
class _FakeLlmProvider:
def stream_response(self, transcript: str, *, session_id: str, persona_id: str | None):
del transcript, session_id, persona_id
yield "Labdien! "
yield "Es palīdzēšu."
class _FakeTtsProvider:
def stream_audio(self, text_chunks, *, language: str, voice: str, session_id: str):
del language, voice, session_id
for index, chunk in enumerate(text_chunks):
yield {"audio": chunk.encode("utf-8"), "mime_type": "audio/wav", "is_final": index == 1}
def test_load_livekit_voice_agent_settings_and_metadata() -> None:
settings = load_livekit_voice_agent_settings(
{
"LIVEKIT_URL": "wss://livekit.example",
"LIVEKIT_API_KEY": "key",
"LIVEKIT_API_SECRET": "secret",
"MARIS_VOICE_STT_PROVIDER": "speechmatics",
"MARIS_VOICE_TTS_PROVIDER": "azure",
}
)
metadata = build_livekit_agent_metadata(settings)
assert settings.livekit_url == "wss://livekit.example"
assert metadata["runtime"] == "livekit-agents"
assert metadata["features"]["barge_in"] is True
def test_realtime_voice_assistant_streams_full_turn() -> None:
assistant = RealtimeVoiceAssistant(
stt=_FakeSttProvider(),
llm=_FakeLlmProvider(),
tts=_FakeTtsProvider(),
)
events = list(
assistant.handle_turn([b"audio"], session=VoiceSessionState(session_id="voice-1"))
)
assert events[0]["type"] == "vad"
assert any(event["type"] == "transcript" and event["is_final"] for event in events)
assert any(event["type"] == "llm_delta" for event in events)
assert any(event["type"] == "tts_audio" for event in events)
assert events[-1]["type"] == "complete"
assert events[-1]["text"] == "Labdien! Es palīdzēšu."
def test_realtime_voice_assistant_supports_barge_in() -> None:
assistant = RealtimeVoiceAssistant(
stt=_FakeSttProvider(),
llm=_FakeLlmProvider(),
tts=_FakeTtsProvider(),
)
session = VoiceSessionState(session_id="voice-2")
iterator = assistant.handle_turn([b"audio"], session=session)
next(iterator) # vad
next(iterator) # partial transcript
next(iterator) # final transcript
first_llm_delta = next(iterator)
interrupt_event = assistant.interrupt(session)
barge_in_event = next(iterator)
assert first_llm_delta["type"] == "llm_delta"
assert interrupt_event["type"] == "barge_in"
assert barge_in_event["type"] == "barge_in"
assert barge_in_event["interrupted"] is True
def test_require_livekit_agents_runtime_raises_clear_error() -> None:
with pytest.raises(RuntimeError) as exc_info:
require_livekit_agents_runtime()
assert "LiveKit Agents runtime nav instalēts" in str(exc_info.value)
|