feat: integrate OCI agent streaming and web voice client
This commit is contained in:
@@ -18,10 +18,17 @@ Non-responsibilities:
|
||||
|
||||
import pytest
|
||||
import asyncio
|
||||
from contextlib import aclosing
|
||||
|
||||
from voice_assistant.voice_pipeline import (
|
||||
PipelineAudioChunk,
|
||||
PipelineAgentResponse,
|
||||
PipelineTextDelta,
|
||||
VoicePipeline,
|
||||
)
|
||||
|
||||
from voice_assistant.audio_gateway.app.session import SessionState
|
||||
from voice_assistant.turn_manager import TurnManager, TurnState
|
||||
from voice_assistant.voice_pipeline import VoicePipeline
|
||||
from voice_assistant.agent.mock import MockAgentAdapter
|
||||
from voice_assistant.tts.mock import MockTTSAdapter
|
||||
|
||||
@@ -818,3 +825,268 @@ async def test_pipeline_rejects_new_turn_while_previous_task_runs():
|
||||
assert second_turn.turn_id == 2
|
||||
assert session.active_turn_id == 2
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_stream_turn_response_produces_audio_before_agent_finishes():
|
||||
"""
|
||||
Verify that TTS audio can be delivered while the Agent
|
||||
is still generating its response.
|
||||
"""
|
||||
|
||||
class StreamingAgent:
|
||||
def __init__(self):
|
||||
self.first_delta_sent = asyncio.Event()
|
||||
self.release_generation = asyncio.Event()
|
||||
self.finished = False
|
||||
|
||||
async def stream(self, session_id, turn_id, text):
|
||||
yield "Olá, como posso ajudar?"
|
||||
|
||||
self.first_delta_sent.set()
|
||||
|
||||
await self.release_generation.wait()
|
||||
|
||||
yield " Estou à disposição."
|
||||
self.finished = True
|
||||
|
||||
class StreamingTTS:
|
||||
async def synthesize(self, text):
|
||||
yield text.encode("utf-8")
|
||||
|
||||
agent = StreamingAgent()
|
||||
|
||||
pipeline = VoicePipeline(
|
||||
session=SessionState(session_id="stream-session"),
|
||||
turn_manager=TurnManager(),
|
||||
agent=agent,
|
||||
tts=StreamingTTS(),
|
||||
)
|
||||
|
||||
turn = pipeline.start_turn()
|
||||
pipeline.on_final(turn.turn_id, "Olá")
|
||||
|
||||
first_audio_received = asyncio.Event()
|
||||
received_events = []
|
||||
|
||||
async def consume():
|
||||
async with aclosing(
|
||||
pipeline.stream_turn_response(
|
||||
session_id="stream-session",
|
||||
turn_id=turn.turn_id,
|
||||
text="Olá",
|
||||
)
|
||||
) as stream:
|
||||
async for event in stream:
|
||||
received_events.append(event)
|
||||
|
||||
if isinstance(event, PipelineAudioChunk):
|
||||
first_audio_received.set()
|
||||
|
||||
task = asyncio.create_task(consume())
|
||||
|
||||
try:
|
||||
await asyncio.wait_for(
|
||||
first_audio_received.wait(),
|
||||
timeout=2.0,
|
||||
)
|
||||
|
||||
assert agent.finished is False
|
||||
|
||||
assert any(
|
||||
isinstance(event, PipelineTextDelta)
|
||||
for event in received_events
|
||||
)
|
||||
|
||||
assert any(
|
||||
isinstance(event, PipelineAudioChunk)
|
||||
for event in received_events
|
||||
)
|
||||
|
||||
agent.release_generation.set()
|
||||
|
||||
await asyncio.wait_for(task, timeout=2.0)
|
||||
|
||||
assert agent.finished is True
|
||||
|
||||
assert any(
|
||||
isinstance(event, PipelineAgentResponse)
|
||||
for event in received_events
|
||||
)
|
||||
|
||||
assert (
|
||||
pipeline.turn_manager.active_turn.state
|
||||
== TurnState.COMPLETED
|
||||
)
|
||||
|
||||
finally:
|
||||
agent.release_generation.set()
|
||||
|
||||
if not task.done():
|
||||
task.cancel()
|
||||
|
||||
await asyncio.gather(
|
||||
task,
|
||||
return_exceptions=True,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_stream_turn_response_propagates_agent_failure():
|
||||
class FailingStreamingAgent:
|
||||
async def stream(self, session_id, turn_id, text):
|
||||
yield "Olá, como posso ajudar?"
|
||||
raise RuntimeError("Simulated OCI streaming failure.")
|
||||
|
||||
class StreamingTTS:
|
||||
async def synthesize(self, text):
|
||||
yield text.encode("utf-8")
|
||||
|
||||
pipeline = VoicePipeline(
|
||||
session=SessionState(session_id="agent-failure-session"),
|
||||
turn_manager=TurnManager(),
|
||||
agent=FailingStreamingAgent(),
|
||||
tts=StreamingTTS(),
|
||||
)
|
||||
|
||||
turn = pipeline.start_turn()
|
||||
pipeline.on_final(turn.turn_id, "Olá")
|
||||
|
||||
async def consume():
|
||||
async with aclosing(
|
||||
pipeline.stream_turn_response(
|
||||
session_id="agent-failure-session",
|
||||
turn_id=turn.turn_id,
|
||||
text="Olá",
|
||||
)
|
||||
) as stream:
|
||||
return [event async for event in stream]
|
||||
|
||||
with pytest.raises(ExceptionGroup) as exc_info:
|
||||
await asyncio.wait_for(consume(), timeout=2.0)
|
||||
|
||||
assert any(
|
||||
isinstance(error, RuntimeError)
|
||||
and "Simulated OCI streaming failure" in str(error)
|
||||
for error in exc_info.value.exceptions
|
||||
)
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_stream_turn_response_cancels_agent_on_tts_failure():
|
||||
class BlockingStreamingAgent:
|
||||
def __init__(self):
|
||||
self.cancelled = asyncio.Event()
|
||||
|
||||
async def stream(self, session_id, turn_id, text):
|
||||
try:
|
||||
yield "Olá, como posso ajudar?"
|
||||
await asyncio.Event().wait()
|
||||
finally:
|
||||
self.cancelled.set()
|
||||
|
||||
class FailingTTS:
|
||||
async def synthesize(self, text):
|
||||
raise RuntimeError("Simulated TTS failure.")
|
||||
yield b"" # Keep this method an async generator.
|
||||
|
||||
agent = BlockingStreamingAgent()
|
||||
|
||||
pipeline = VoicePipeline(
|
||||
session=SessionState(session_id="tts-failure-session"),
|
||||
turn_manager=TurnManager(),
|
||||
agent=agent,
|
||||
tts=FailingTTS(),
|
||||
)
|
||||
|
||||
turn = pipeline.start_turn()
|
||||
pipeline.on_final(turn.turn_id, "Olá")
|
||||
|
||||
async def consume():
|
||||
async with aclosing(
|
||||
pipeline.stream_turn_response(
|
||||
session_id="tts-failure-session",
|
||||
turn_id=turn.turn_id,
|
||||
text="Olá",
|
||||
)
|
||||
) as stream:
|
||||
async for _ in stream:
|
||||
pass
|
||||
|
||||
with pytest.raises(ExceptionGroup) as exc_info:
|
||||
await asyncio.wait_for(consume(), timeout=2.0)
|
||||
|
||||
assert any(
|
||||
isinstance(error, RuntimeError)
|
||||
and "Simulated TTS failure" in str(error)
|
||||
for error in exc_info.value.exceptions
|
||||
)
|
||||
|
||||
assert agent.cancelled.is_set()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_stream_turn_response_external_cancellation():
|
||||
class BlockingAgent:
|
||||
def __init__(self):
|
||||
self.closed = asyncio.Event()
|
||||
|
||||
async def stream(self, session_id, turn_id, text):
|
||||
try:
|
||||
yield "Olá, como posso ajudar?"
|
||||
await asyncio.Event().wait()
|
||||
finally:
|
||||
self.closed.set()
|
||||
|
||||
class BlockingTTS:
|
||||
def __init__(self):
|
||||
self.started = asyncio.Event()
|
||||
self.closed = asyncio.Event()
|
||||
|
||||
async def synthesize(self, text):
|
||||
try:
|
||||
self.started.set()
|
||||
await asyncio.Event().wait()
|
||||
yield b"audio"
|
||||
finally:
|
||||
self.closed.set()
|
||||
|
||||
agent = BlockingAgent()
|
||||
tts = BlockingTTS()
|
||||
|
||||
pipeline = VoicePipeline(
|
||||
session=SessionState(session_id="cancel-session"),
|
||||
turn_manager=TurnManager(),
|
||||
agent=agent,
|
||||
tts=tts,
|
||||
)
|
||||
|
||||
turn = pipeline.start_turn()
|
||||
pipeline.on_final(turn.turn_id, "Olá")
|
||||
|
||||
async def consume():
|
||||
async with aclosing(
|
||||
pipeline.stream_turn_response(
|
||||
session_id="cancel-session",
|
||||
turn_id=turn.turn_id,
|
||||
text="Olá",
|
||||
)
|
||||
) as stream:
|
||||
async for _ in stream:
|
||||
pass
|
||||
|
||||
task = asyncio.create_task(consume())
|
||||
|
||||
try:
|
||||
await asyncio.wait_for(tts.started.wait(), timeout=2.0)
|
||||
|
||||
task.cancel()
|
||||
|
||||
with pytest.raises(asyncio.CancelledError):
|
||||
await asyncio.wait_for(task, timeout=2.0)
|
||||
|
||||
assert agent.closed.is_set()
|
||||
assert tts.closed.is_set()
|
||||
|
||||
finally:
|
||||
if not task.done():
|
||||
task.cancel()
|
||||
|
||||
await asyncio.gather(task, return_exceptions=True)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user