319 lines
12 KiB
Python
319 lines
12 KiB
Python
import asyncio
|
|
import time
|
|
import uuid
|
|
|
|
from services.ai_voice_runtime_service.audiosocket import EnergyVAD
|
|
from services.ai_voice_runtime_service.media_runtime import AudioSocketMediaRuntime, MediaActor, MediaRegistration
|
|
from services.ai_voice_runtime_service.providers.asr import (
|
|
ASRProvider,
|
|
ASRTranscription,
|
|
StreamingASRPartial,
|
|
StreamingASRProvider,
|
|
StreamingASRUnavailable,
|
|
)
|
|
from services.ai_voice_runtime_service.providers.tts import TTSSynthesis, TTSProvider
|
|
from services.shared.models import VoiceAITurnDecisionOut
|
|
|
|
|
|
class _StubTTSProvider(TTSProvider):
|
|
name = "stub-tts"
|
|
|
|
def synthesize(
|
|
self,
|
|
text: str,
|
|
*,
|
|
language: str | None = None,
|
|
style_hints: dict[str, object] | None = None,
|
|
) -> TTSSynthesis:
|
|
del language, style_hints
|
|
assert text
|
|
return TTSSynthesis(
|
|
text=text,
|
|
audio_bytes=(b"\x10\x00" * 960),
|
|
sample_rate_hz=24000,
|
|
)
|
|
|
|
|
|
def test_media_runtime_voice_v2_uses_streaming_sidecar_for_partial_and_final_asr_stable():
|
|
reply_starts: list[tuple[str, str | None, float]] = []
|
|
turns: list[tuple[str, str]] = []
|
|
|
|
class _BatchASRProvider(ASRProvider):
|
|
name = "batch-asr"
|
|
|
|
def transcribe(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription:
|
|
raise AssertionError("batch ASR should not be used when streaming sidecar is active")
|
|
|
|
class _StreamingProvider(StreamingASRProvider):
|
|
name = "streaming-sidecar"
|
|
supports_streaming = True
|
|
|
|
def __init__(self) -> None:
|
|
self.events: list[tuple[str, float]] = []
|
|
self.chunk_count = 0
|
|
|
|
def open_stream(self, session_id: str, *, language_hint: str | None = None) -> str:
|
|
assert session_id
|
|
assert language_hint == "ru"
|
|
self.events.append(("open", time.monotonic()))
|
|
return "stream-1"
|
|
|
|
def push_pcm(self, stream_id: str, pcm_bytes: bytes) -> None:
|
|
assert stream_id == "stream-1"
|
|
assert pcm_bytes
|
|
self.chunk_count += 1
|
|
self.events.append(("push", time.monotonic()))
|
|
|
|
def poll_partial(self, stream_id: str) -> StreamingASRPartial | None:
|
|
assert stream_id == "stream-1"
|
|
if self.chunk_count < 2:
|
|
return None
|
|
self.events.append(("partial", time.monotonic()))
|
|
return StreamingASRPartial(
|
|
text="need work schedule",
|
|
language="ru",
|
|
confidence=0.84,
|
|
is_final=False,
|
|
is_stable=True,
|
|
)
|
|
|
|
def finalize(self, stream_id: str) -> ASRTranscription:
|
|
assert stream_id == "stream-1"
|
|
self.events.append(("finalize", time.monotonic()))
|
|
return ASRTranscription(text="need work schedule", language="ru", confidence=0.91)
|
|
|
|
def close_stream(self, stream_id: str) -> None:
|
|
assert stream_id == "stream-1"
|
|
self.events.append(("close", time.monotonic()))
|
|
|
|
streaming_provider = _StreamingProvider()
|
|
registration = MediaRegistration(
|
|
voice_session_id="avs_voice_v2_streaming",
|
|
call_id="call_voice_v2_streaming",
|
|
interaction_id="int_voice_v2_streaming",
|
|
ai_session_id="ais_voice_v2_streaming",
|
|
language="ru",
|
|
media_uuid=str(uuid.uuid4()),
|
|
queue_code="voice_lab_ai",
|
|
queue_id="que_voice_lab_ai",
|
|
agent_profile="voice_support",
|
|
voice_v2_enabled=True,
|
|
voice_v2_ack_mode="immediate_short",
|
|
voice_v2_streaming_tts=True,
|
|
voice_v2_partial_asr=True,
|
|
voice_v2_duplex=True,
|
|
voice_v2_streaming_asr_backend="local_sidecar",
|
|
voice_v2_prebaked_ack=False,
|
|
)
|
|
|
|
runtime = AudioSocketMediaRuntime(
|
|
enabled=True,
|
|
host="127.0.0.1",
|
|
port=0,
|
|
frame_ms=20,
|
|
idle_timeout_seconds=2.0,
|
|
registration_wait_timeout_seconds=0.5,
|
|
min_speech_ms=40,
|
|
trailing_silence_ms=40,
|
|
max_turn_ms=2000,
|
|
asr_provider=_BatchASRProvider(),
|
|
streaming_asr_provider=streaming_provider,
|
|
tts_provider=_StubTTSProvider(),
|
|
load_registration_by_media_uuid=lambda value: None,
|
|
mark_media_connected=lambda session_id, value: None,
|
|
mark_media_ended=lambda session_id, reason: None,
|
|
touch_media_frame=lambda session_id: None,
|
|
set_state=lambda session_id, state, handoff_reason, metadata: None,
|
|
get_pending_greeting=lambda session_id: None,
|
|
mark_reply_started=lambda session_id, text, is_greeting, reply_phase=None: None,
|
|
mark_reply_delivered=lambda session_id, text, is_greeting, reply_phase=None: None,
|
|
mark_reply_discarded=lambda session_id, text, reply_phase, reason: None,
|
|
plan_reply=lambda session_id, text, metadata, kind: None,
|
|
record_latency=lambda session_id, metric, latency_ms: None,
|
|
process_turn=lambda session_id, transcript_text, language, barge_in, metadata: (
|
|
turns.append((str((metadata or {}).get("reply_phase") or "final"), transcript_text))
|
|
or VoiceAITurnDecisionOut(
|
|
language=language or "ru",
|
|
intent="schedule",
|
|
reply_text="Please name the city.",
|
|
confidence=0.92,
|
|
needs_handoff=False,
|
|
handoff_reason=None,
|
|
case_action="keep_open",
|
|
kb_refs=[],
|
|
summary_text="reply ready",
|
|
model="stub-voice",
|
|
latency_ms=1,
|
|
status="active",
|
|
)
|
|
),
|
|
request_handoff=lambda session_id, customer_request_text, decision: None,
|
|
handle_media_error=lambda session_id, message, metadata: None,
|
|
)
|
|
|
|
async def _fake_speak_reply(
|
|
actor: MediaActor,
|
|
text: str,
|
|
*,
|
|
is_greeting: bool,
|
|
style_hints: dict[str, object] | None = None,
|
|
reply_phase: str | None = "main",
|
|
) -> None:
|
|
del actor, is_greeting, style_hints
|
|
reply_starts.append((text, reply_phase, time.monotonic()))
|
|
|
|
runtime._speak_reply = _fake_speak_reply # type: ignore[method-assign]
|
|
|
|
async def _scenario() -> None:
|
|
actor = MediaActor(
|
|
registration=registration,
|
|
reader=asyncio.StreamReader(),
|
|
writer=None, # type: ignore[arg-type]
|
|
vad=EnergyVAD(frame_ms=20, min_speech_ms=40, trailing_silence_ms=40, max_turn_ms=2000),
|
|
frame_ms=20,
|
|
frame_bytes=320,
|
|
)
|
|
runtime._reset_live_turn_state(actor)
|
|
await runtime._ensure_streaming_asr(actor)
|
|
assert actor.asr_stream_id == "stream-1"
|
|
|
|
pcm_chunk = (1000).to_bytes(2, "little", signed=True) * 160
|
|
await asyncio.to_thread(streaming_provider.push_pcm, actor.asr_stream_id, pcm_chunk)
|
|
await asyncio.to_thread(streaming_provider.push_pcm, actor.asr_stream_id, pcm_chunk)
|
|
actor.asr_poll_due_monotonic = 0.0
|
|
await runtime._poll_streaming_partial(actor)
|
|
|
|
actor.speech_started_monotonic = time.monotonic()
|
|
actor.speech_ended_monotonic = actor.speech_started_monotonic
|
|
await runtime._process_utterance(actor, pcm_chunk * 14, False)
|
|
|
|
asyncio.run(_scenario())
|
|
|
|
phases = [phase for _, phase, _ in reply_starts]
|
|
finalize_started_at = next(ts for name, ts in streaming_provider.events if name == "finalize")
|
|
assert [text for phase, text in turns if phase == "final"] == ["need work schedule"]
|
|
assert "open" in [name for name, _ in streaming_provider.events]
|
|
assert "close" in [name for name, _ in streaming_provider.events]
|
|
assert phases[:2] == ["ack", "main"]
|
|
assert reply_starts[0][2] <= finalize_started_at
|
|
assert any(text == "Please name the city." and phase == "main" for text, phase, _ in reply_starts)
|
|
|
|
|
|
def test_media_runtime_voice_v2_falls_back_when_streaming_sidecar_is_unavailable_stable():
|
|
turns: list[str] = []
|
|
|
|
class _BatchASRProvider(ASRProvider):
|
|
name = "batch-asr"
|
|
|
|
def transcribe(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription:
|
|
assert audio_bytes
|
|
return ASRTranscription(text="need operator", language=language_hint or "ru", confidence=0.88)
|
|
|
|
class _UnavailableStreamingProvider(StreamingASRProvider):
|
|
name = "missing-sidecar"
|
|
supports_streaming = True
|
|
|
|
def open_stream(self, session_id: str, *, language_hint: str | None = None) -> str:
|
|
del session_id, language_hint
|
|
raise StreamingASRUnavailable("sidecar down")
|
|
|
|
registration = MediaRegistration(
|
|
voice_session_id="avs_voice_v2_fallback",
|
|
call_id="call_voice_v2_fallback",
|
|
interaction_id="int_voice_v2_fallback",
|
|
ai_session_id="ais_voice_v2_fallback",
|
|
language="ru",
|
|
media_uuid=str(uuid.uuid4()),
|
|
queue_code="voice_lab_ai",
|
|
queue_id="que_voice_lab_ai",
|
|
agent_profile="voice_support",
|
|
voice_v2_enabled=True,
|
|
voice_v2_ack_mode="immediate_short",
|
|
voice_v2_streaming_tts=True,
|
|
voice_v2_partial_asr=True,
|
|
voice_v2_duplex=True,
|
|
voice_v2_streaming_asr_backend="local_sidecar",
|
|
voice_v2_prebaked_ack=False,
|
|
)
|
|
|
|
runtime = AudioSocketMediaRuntime(
|
|
enabled=True,
|
|
host="127.0.0.1",
|
|
port=0,
|
|
frame_ms=20,
|
|
idle_timeout_seconds=2.0,
|
|
registration_wait_timeout_seconds=0.5,
|
|
min_speech_ms=40,
|
|
trailing_silence_ms=40,
|
|
max_turn_ms=2000,
|
|
asr_provider=_BatchASRProvider(),
|
|
streaming_asr_provider=_UnavailableStreamingProvider(),
|
|
tts_provider=_StubTTSProvider(),
|
|
load_registration_by_media_uuid=lambda value: None,
|
|
mark_media_connected=lambda session_id, value: None,
|
|
mark_media_ended=lambda session_id, reason: None,
|
|
touch_media_frame=lambda session_id: None,
|
|
set_state=lambda session_id, state, handoff_reason, metadata: None,
|
|
get_pending_greeting=lambda session_id: None,
|
|
mark_reply_started=lambda session_id, text, is_greeting, reply_phase=None: None,
|
|
mark_reply_delivered=lambda session_id, text, is_greeting, reply_phase=None: None,
|
|
mark_reply_discarded=lambda session_id, text, reply_phase, reason: None,
|
|
plan_reply=lambda session_id, text, metadata, kind: None,
|
|
record_latency=lambda session_id, metric, latency_ms: None,
|
|
process_turn=lambda session_id, transcript_text, language, barge_in, metadata: (
|
|
turns.append(transcript_text)
|
|
or VoiceAITurnDecisionOut(
|
|
language=language or "ru",
|
|
intent="handoff",
|
|
reply_text="Connecting you to an operator.",
|
|
confidence=0.9,
|
|
needs_handoff=False,
|
|
handoff_reason=None,
|
|
case_action="keep_open",
|
|
kb_refs=[],
|
|
summary_text="reply ready",
|
|
model="stub-voice",
|
|
latency_ms=1,
|
|
status="active",
|
|
)
|
|
),
|
|
request_handoff=lambda session_id, customer_request_text, decision: None,
|
|
handle_media_error=lambda session_id, message, metadata: None,
|
|
)
|
|
|
|
async def _fake_speak_reply(
|
|
actor: MediaActor,
|
|
text: str,
|
|
*,
|
|
is_greeting: bool,
|
|
style_hints: dict[str, object] | None = None,
|
|
reply_phase: str | None = "main",
|
|
) -> None:
|
|
del actor, text, is_greeting, style_hints, reply_phase
|
|
|
|
runtime._speak_reply = _fake_speak_reply # type: ignore[method-assign]
|
|
|
|
async def _scenario() -> None:
|
|
actor = MediaActor(
|
|
registration=registration,
|
|
reader=asyncio.StreamReader(),
|
|
writer=None, # type: ignore[arg-type]
|
|
vad=EnergyVAD(frame_ms=20, min_speech_ms=40, trailing_silence_ms=40, max_turn_ms=2000),
|
|
frame_ms=20,
|
|
frame_bytes=320,
|
|
)
|
|
runtime._reset_live_turn_state(actor)
|
|
await runtime._ensure_streaming_asr(actor)
|
|
assert actor.asr_streaming_enabled is False
|
|
|
|
pcm_chunk = (1000).to_bytes(2, "little", signed=True) * 160
|
|
actor.speech_started_monotonic = time.monotonic()
|
|
actor.speech_ended_monotonic = actor.speech_started_monotonic
|
|
await runtime._process_utterance(actor, pcm_chunk * 4, False)
|
|
|
|
asyncio.run(_scenario())
|
|
|
|
assert turns == ["need operator"]
|
|
assert registration.voice_v2_duplex is True
|
|
assert registration.voice_v2_partial_asr is True
|