From 8b05e7c8ff09b4c0b8237dc78e27f6b323ec66dd Mon Sep 17 00:00:00 2001 From: mkbk Date: Fri, 19 Jun 2026 13:39:18 +0800 Subject: [PATCH] =?UTF-8?q?[=E5=85=A8=E5=8F=8C=E5=B7=A5=E8=BF=90=E8=A1=8C?= =?UTF-8?q?=E5=85=A5=E5=8F=A3]=EF=BC=9A=E5=AE=8C=E6=88=90run-agent-live?= =?UTF-8?q?=E5=B8=B8=E9=A9=BB=E7=9B=91=E5=90=AC=E9=97=AD=E7=8E=AF=EF=BC=8C?= =?UTF-8?q?=E5=8C=85=E5=90=AB=E9=BA=A6=E5=85=8B=E9=A3=8E=E9=87=87=E9=9B=86?= =?UTF-8?q?=E3=80=81=E8=AF=AD=E9=9F=B3=E8=AF=86=E5=88=AB=E5=92=8C=E7=BB=88?= =?UTF-8?q?=E7=AB=AF=E7=8A=B6=E6=80=81=E8=BE=93=E5=87=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/owner_voice_pet/full_duplex_runtime.py | 530 ++++++++++++++++++++- tests/test_full_duplex_integration.py | 43 +- 2 files changed, 563 insertions(+), 10 deletions(-) diff --git a/src/owner_voice_pet/full_duplex_runtime.py b/src/owner_voice_pet/full_duplex_runtime.py index 8d55723..fe2cf78 100644 --- a/src/owner_voice_pet/full_duplex_runtime.py +++ b/src/owner_voice_pet/full_duplex_runtime.py @@ -1,5 +1,6 @@ from __future__ import annotations +import threading import time from dataclasses import dataclass @@ -11,6 +12,7 @@ from .agent_memory import ( MemoryManager, SQLiteMemoryManager, ) +from .barge_in import BargeInSpeakerGate, TimbreProfile, ensure_interruptible_pcm from .config import AppConfig from .conversation import ConversationContext from .full_duplex_audio import AudioHub, AudioProcessingProvider, build_audio_processing_provider @@ -25,8 +27,11 @@ from .full_duplex_response import ( StreamingTtsProvider, ) from .full_duplex_speech import FakeVadProvider, InterruptController, InterruptionDetector -from .models import AudioFrame, Message, PipelineState +from .llm import OpenAICompatibleLlmProvider +from .models import AudioFrame, AudioSegment, ErrorCode, Message, PipelineState, ProviderError +from .protocols import AudioTransport, LlmProvider, RealtimeSttProvider, SttProvider, TtsProvider from .runtime import RuntimeSummary +from .stt import CloudAsrSttProvider, SherpaOnnxSttProvider, is_valid_transcript_text from .tool_router import ( MemorySaveTool, MemorySearchTool, @@ -36,6 +41,9 @@ from .tool_router import ( ToolResult, ToolRouter, ) +from .transport import SoundDeviceAudioTransport +from .tts import CloudTtsProvider, MacSayTtsProvider, SentenceBuffer, make_end_chime, sanitize_tts_text +from .vad import EnergyVadProvider, HybridVadProvider, SherpaOnnxVadProvider, VadRecorder @dataclass(slots=True) @@ -46,30 +54,195 @@ class FullDuplexRuntimeHealth: message: str +class _AudioHubInputPump: + def __init__( + self, + *, + config: AppConfig, + transport: AudioTransport, + audio_hub: AudioHub, + ) -> None: + self.config = config + self.transport = transport + self.audio_hub = audio_hub + self.error: ProviderError | None = None + self._stop_event = threading.Event() + self._thread: threading.Thread | None = None + self._input_started = False + + def start_input(self) -> None: + if self._input_started: + return + self.transport.start_input( + device_id=self.config.audio_input_device, + sample_rate=self.config.sample_rate, + channels=self.config.channels, + ) + self._input_started = True + + def start_reader(self) -> None: + self.start_input() + if self._thread is not None and self._thread.is_alive(): + return + self._stop_event.clear() + self._thread = threading.Thread(target=self._run, name="owner-full-duplex-audio-hub", daemon=True) + self._thread.start() + + def stop(self) -> None: + self._stop_event.set() + if self._thread is not None: + self._thread.join(timeout=1.0) + self.transport.stop() + self._input_started = False + + def _run(self) -> None: + interval_ms = max(1, self.config.barge_in_listen_interval_ms) + while not self._stop_event.is_set(): + try: + frames = self.transport.read_frames(timeout_ms=interval_ms) + if not frames: + time.sleep(interval_ms / 1000) + continue + for frame in frames: + self.audio_hub.accept_capture(frame) + except ProviderError as exc: + self.error = exc + self._stop_event.set() + except Exception as exc: # pragma: no cover - defensive boundary for device callbacks + self.error = ProviderError( + ErrorCode.AUDIO_APM_PROCESS_FAILED, + f"audio input pump failed: {exc}", + True, + "full-duplex-runtime", + "audio", + ) + self._stop_event.set() + + +class _HubBargeInMonitor: + def __init__( + self, + *, + config: AppConfig, + audio_hub: AudioHub, + speaker_gate: BargeInSpeakerGate, + assistant_reference: AudioSegment | None, + assistant_profile: TimbreProfile, + ) -> None: + self.config = config + self.audio_hub = audio_hub + self.speaker_gate = speaker_gate + self.assistant_reference = assistant_reference + self.assistant_profile = assistant_profile + self.stop_event = threading.Event() + self._shutdown_event = threading.Event() + self._thread: threading.Thread | None = None + self._pending_frames: list[AudioFrame] = [] + self._lock = threading.Lock() + self._vad = EnergyVadProvider() + self._interrupted = False + + @property + def interrupted(self) -> bool: + return self._interrupted + + def pending_frames(self) -> list[AudioFrame]: + with self._lock: + return list(self._pending_frames) + + def start(self) -> None: + self._vad.load() + self._vad.reset() + self._thread = threading.Thread(target=self._run, name="owner-full-duplex-barge-in", daemon=True) + self._thread.start() + + def stop(self) -> None: + self._shutdown_event.set() + if self._thread is not None: + self._thread.join(timeout=1.0) + + def should_stop_playback(self) -> bool: + return self.stop_event.is_set() + + def _run(self) -> None: + subscription = self.audio_hub.subscribe("processed_capture", name="barge-in") + started_at = time.monotonic() + speech_ms = 0 + candidate_frames: list[AudioFrame] = [] + interval_ms = max(1, self.config.barge_in_listen_interval_ms) + echo_guard_ms = max(0, min(self.config.barge_in_echo_guard_ms, 120)) + min_speech_ms = max(40, min(self.config.barge_in_min_speech_ms, self.config.interrupt_target_latency_ms)) + while not self._shutdown_event.is_set() and not self.stop_event.is_set(): + frames = subscription.read_available() + if not frames: + time.sleep(interval_ms / 1000) + continue + elapsed_ms = int((time.monotonic() - started_at) * 1000) + for frame in frames: + if elapsed_ms < echo_guard_ms: + continue + result = self._vad.analyze(frame) + if not result.is_speech or frame.metadata.get("echo_suppressed") or frame.metadata.get("assistant_echo"): + speech_ms = 0 + candidate_frames = [] + continue + accepted = self.speaker_gate.accepts_candidate( + frame, + self.assistant_profile, + self.assistant_reference, + reference_elapsed_ms=elapsed_ms, + ) + if not accepted: + speech_ms = 0 + candidate_frames = [] + continue + speech_ms += frame.duration_ms + candidate_frames.append(frame) + if speech_ms >= min_speech_ms: + with self._lock: + self._pending_frames = list(candidate_frames) + self._interrupted = True + self.stop_event.set() + return + + class FullDuplexAgentRuntime: """Full-duplex Agent runtime boundary. - Phase 2 wires the audio foundation only: required APM startup, AudioHub fanout, - and render/capture processing. Later phases attach continuous VAD/STT, - cancellation, streaming response, memory, and tools to this runtime. + `run()` is the real `run-agent-live` loop: microphone frames are pumped into + AudioHub once, capture/STT/LLM/TTS consume independent subscriptions, and + playback can be stopped by a concurrent barge-in monitor. """ def __init__( self, *, config: AppConfig, + transport: AudioTransport | None = None, processor: AudioProcessingProvider | None = None, audio_hub: AudioHub | None = None, + vad_recorder: VadRecorder | None = None, + stt: SttProvider | None = None, + realtime_stt: RealtimeSttProvider | None = None, + llm: LlmProvider | None = None, + tts: TtsProvider | None = None, llm_provider: StreamingLlmProvider | None = None, tts_provider: StreamingTtsProvider | None = None, context: ConversationContext | None = None, memory_manager: MemoryManager | None = None, memory_manifest: FaissIndexManifest | None = None, tool_router: ToolRouter | None = None, + sentence_buffer: SentenceBuffer | None = None, ) -> None: self.config = config + self.transport = transport or SoundDeviceAudioTransport(output_device=config.audio_output_device) self.processor = processor self.audio_hub = audio_hub + self.vad_recorder = vad_recorder + self.stt = stt + self.realtime_stt = realtime_stt + self.llm = llm + self.tts = tts self.llm_provider = llm_provider self.tts_provider = tts_provider self.context = context or ConversationContext( @@ -92,6 +265,15 @@ class FullDuplexAgentRuntime: self.playback_queue = InterruptiblePlaybackQueue() self.tool_results: list[ToolResult] = [] self.tool_result_messages: list[Message] = [] + self.sentence_buffer = sentence_buffer or SentenceBuffer() + self._input_pump: _AudioHubInputPump | None = None + self._pending_capture_frames: list[AudioFrame] = [] + self._barge_in_gate = BargeInSpeakerGate( + enabled=config.barge_in_speaker_gate_enabled, + user_similarity_threshold=config.barge_in_user_similarity_threshold, + assistant_reject_threshold=config.barge_in_assistant_reject_threshold, + min_rms=config.speaker_min_rms, + ) def load_audio(self) -> FullDuplexRuntimeHealth: if self.audio_hub is None: @@ -113,11 +295,341 @@ class FullDuplexAgentRuntime: def run(self, *, once: bool = False) -> RuntimeSummary: self.load_audio() - if once: - self._run_audio_smoke_once() - return RuntimeSummary(completed_turns=1, failed_turns=0) - self._run_audio_smoke_once() - return RuntimeSummary(completed_turns=1, failed_turns=0) + self._load_runtime_dependencies() + completed = 0 + failed = 0 + last_error: ProviderError | None = None + interrupted = False + self._start_input() + self._status(None, "全双工Agent：启动，直接说话即可") + try: + while True: + turn_id = completed + failed + 1 + try: + result = self._run_live_turn(turn_id) + except ProviderError as exc: + failed += 1 + last_error = exc + self._error(turn_id, exc.stage, exc.code.value, exc.message) + self._recover_to_listening(turn_id) + if once: + break + continue + if result.interrupted: + interrupted = True + if result.completed_turns > 0: + completed += result.completed_turns + else: + failed += result.failed_turns or 1 + last_error = result.last_error + if once: + break + if once and completed >= 1: + break + except KeyboardInterrupt: + interrupted = True + finally: + self.shutdown() + return RuntimeSummary(completed_turns=completed, failed_turns=failed, interrupted=interrupted, last_error=last_error) + + def shutdown(self) -> None: + if self._input_pump is not None: + self._input_pump.stop() + else: + self.transport.stop() + + def _load_runtime_dependencies(self) -> None: + if self.vad_recorder is None: + self.vad_recorder = self._build_vad_recorder() + self.vad_recorder.provider.load() + if self.stt is None: + self.stt = self._build_stt_provider() + self.stt.load() + if self.realtime_stt is None and isinstance(self.stt, SherpaOnnxSttProvider): + self.realtime_stt = self.stt + if self.realtime_stt is not None and self.realtime_stt is not self.stt: + self.realtime_stt.load() + if self.llm is None: + self.llm = OpenAICompatibleLlmProvider(self.config) + if self.tts is None: + self.tts = self._build_tts_provider() + self.tts.load() + + def _start_input(self) -> None: + if self.audio_hub is None: + raise RuntimeError("audio hub is not loaded") + self._input_pump = _AudioHubInputPump( + config=self.config, + transport=self.transport, + audio_hub=self.audio_hub, + ) + self._input_pump.start_input() + + def _ensure_input_reader(self) -> None: + if self._input_pump is None: + self._start_input() + if self._input_pump is not None: + self._input_pump.start_reader() + + def _run_live_turn(self, turn_id: int) -> RuntimeSummary: + self.state_machine = FullDuplexStateMachine() + self.cancellation_graph = CancellationGraph(f"turn-{turn_id}") + self.playback_queue = InterruptiblePlaybackQueue() + self.sentence_buffer = SentenceBuffer() + self.state_machine.transition(PipelineState.LISTENING, event_type="listening_started") + self._status(turn_id, "监听中：请直接说话") + segment = self._capture_user_segment(turn_id) + self._barge_in_gate.remember_user_segment(segment) + user_text = self._transcribe_segment(segment, turn_id) + interrupted = self._reply_to_user(user_text, turn_id) + self._play_end_chime() + if self.state_machine.current_state != PipelineState.LISTENING and self.state_machine.can_transition(PipelineState.LISTENING): + self.state_machine.transition(PipelineState.LISTENING, event_type="turn_finished") + self._status(turn_id, "恢复监听：可直接说话") + return RuntimeSummary(completed_turns=1, failed_turns=0, interrupted=interrupted) + + def _capture_user_segment(self, turn_id: int) -> AudioSegment: + if self.audio_hub is None or self.vad_recorder is None: + raise RuntimeError("runtime dependencies are not loaded") + self.vad_recorder.reset() + self.vad_recorder.provider.reset() + realtime_session = self.realtime_stt.start_stream() if self.realtime_stt is not None else None + subscription = self.audio_hub.subscribe("processed_capture", name=f"turn-{turn_id}-capture") + self._ensure_input_reader() + self._status(turn_id, "录音中：正在听取问题") + speech_started = False + last_partial = "" + while True: + self._raise_input_pump_error() + frames = self._drain_pending_capture_frames() + frames.extend(subscription.read_available()) + if not frames: + time.sleep(max(1, self.config.audio_frame_ms) / 1000) + continue + for frame in frames: + was_started = self.vad_recorder.started + result = self.vad_recorder.feed(frame) + if not was_started and self.vad_recorder.started and not speech_started: + speech_started = True + self._status(turn_id, "检测到用户语音") + if self.vad_recorder.started and realtime_session is not None: + partial = realtime_session.accept_frame(frame) + if partial is not None and partial.normalized_text and partial.normalized_text != last_partial: + last_partial = partial.normalized_text + self._transcript(turn_id, last_partial, final=False) + if isinstance(result, ProviderError): + if result.code == ErrorCode.VAD_TIMEOUT_NO_SPEECH: + self.vad_recorder.reset() + self.vad_recorder.provider.reset() + continue + raise result + if isinstance(result, AudioSegment): + self._status(turn_id, f"用户语音结束：{result.metadata.get('end_reason', 'speech_end')}") + return result + + def _transcribe_segment(self, segment: AudioSegment, turn_id: int) -> str: + if self.stt is None: + raise RuntimeError("STT provider is not loaded") + self._status(turn_id, "转写中：正在识别问题") + transcript = self.stt.transcribe(segment) + user_text = transcript.normalized_text + if not is_valid_transcript_text(user_text): + raise ProviderError( + ErrorCode.STT_EMPTY_TRANSCRIPT, + "STT produced no meaningful user text", + True, + "full-duplex-runtime", + "stt", + ) + self._transcript(turn_id, user_text, final=True) + return user_text + + def _reply_to_user(self, user_text: str, turn_id: int) -> bool: + if self.llm is None: + raise RuntimeError("LLM provider is not loaded") + messages = [*self.conversation_manager.build_messages_for_user(user_text), *self.tool_result_messages] + if self.state_machine.can_transition(PipelineState.THINKING): + self.state_machine.transition(PipelineState.THINKING, event_type="stt_final") + self._status(turn_id, "思考中：正在生成回复") + assistant_text = "" + spoken_parts: list[str] = [] + interrupted = False + for delta in self.llm.stream_reply(messages): + if self.cancellation_graph.root.cancelled: + interrupted = True + break + assistant_text += delta.text_delta + for sentence in self.sentence_buffer.feed(delta.text_delta, bool(delta.finish_reason)): + spoken, was_interrupted = self._speak_sentence(sentence, turn_id) + if spoken: + spoken_parts.append(spoken) + if was_interrupted: + interrupted = True + break + if interrupted: + break + if not interrupted: + for sentence in self.sentence_buffer.flush(): + spoken, was_interrupted = self._speak_sentence(sentence, turn_id) + if spoken: + spoken_parts.append(spoken) + if was_interrupted: + interrupted = True + break + else: + self.sentence_buffer.flush() + spoken_text = "".join(spoken_parts) + if not assistant_text.strip() and not interrupted: + raise ProviderError( + ErrorCode.LLM_EMPTY_REPLY, + "LLM returned no assistant text", + True, + "full-duplex-runtime", + "llm", + ) + self.conversation_manager.commit_user(user_text) + if spoken_text.strip(): + self.conversation_manager.commit_assistant(spoken_text) + if interrupted: + self._status(turn_id, "播报已打断：继续听你说") + return interrupted + + def _speak_sentence(self, sentence: str, turn_id: int) -> tuple[str, bool]: + if self.tts is None or self.audio_hub is None: + raise RuntimeError("TTS or audio hub is not loaded") + spoken_sentence = sanitize_tts_text(sentence) + if not spoken_sentence: + return "", False + if self.state_machine.can_transition(PipelineState.SPEAKING): + self.state_machine.transition(PipelineState.SPEAKING, event_type="tts_started") + self._status(turn_id, "播放中：正在播报回复") + segment = self.tts.synthesize(spoken_sentence) + interruptible = ensure_interruptible_pcm(segment) + if interruptible is None: + playback = self.transport.play_pcm(segment) + if playback.error: + raise playback.error + return spoken_sentence, False + monitor: _HubBargeInMonitor | None = None + if self.config.interrupt_enabled and self.config.barge_in_enabled: + monitor = _HubBargeInMonitor( + config=self.config, + audio_hub=self.audio_hub, + speaker_gate=self._barge_in_gate, + assistant_reference=interruptible, + assistant_profile=self._barge_in_gate.assistant_profile(interruptible), + ) + monitor.start() + + def after_chunk(chunk: AudioSegment, _elapsed_ms: int) -> bool: + self._accept_render_chunk(chunk) + return False + + playback = self.transport.play_pcm_chunks( + interruptible, + chunk_ms=self.config.barge_in_chunk_ms, + after_chunk=after_chunk, + should_stop=monitor.should_stop_playback if monitor is not None else None, + ) + if monitor is not None: + monitor.stop() + if playback.error: + raise playback.error + if monitor is not None and monitor.interrupted: + self._pending_capture_frames.extend(monitor.pending_frames()) + self.cancellation_graph.cancel_all("user interrupted") + if self.state_machine.can_transition(PipelineState.INTERRUPTED): + self.state_machine.transition(PipelineState.INTERRUPTED, event_type="interrupt_detected") + if self.state_machine.can_transition(PipelineState.LISTENING): + self.state_machine.transition(PipelineState.LISTENING, event_type="buffered_user_audio") + return "", True + return spoken_sentence, False + + def _play_end_chime(self) -> None: + if not self.config.end_chime_enabled: + return + segment = make_end_chime( + file_path=self.config.end_chime_file, + frequency_hz=self.config.end_chime_frequency_hz, + duration_ms=self.config.end_chime_duration_ms, + sample_rate=self.config.sample_rate, + channels=self.config.channels, + ) + playback = self.transport.play_pcm(segment) + if playback.error: + return + + def _accept_render_chunk(self, chunk: AudioSegment) -> None: + if self.audio_hub is None: + return + if chunk.sample_rate != self.config.sample_rate or chunk.channels != self.config.channels: + return + frame = AudioFrame( + chunk.pcm, + chunk.sample_rate, + chunk.channels, + chunk.start_time_ms, + max(0, int(time.monotonic() * 1000) % 2_000_000_000), + {"duration_ms": chunk.duration_ms, "assistant_audio": True}, + ) + self.audio_hub.accept_render(frame) + + def _drain_pending_capture_frames(self) -> list[AudioFrame]: + if not self._pending_capture_frames: + return [] + frames = list(self._pending_capture_frames) + self._pending_capture_frames.clear() + return frames + + def _raise_input_pump_error(self) -> None: + if self._input_pump is not None and self._input_pump.error is not None: + raise self._input_pump.error + + def _recover_to_listening(self, turn_id: int) -> None: + if self.state_machine.current_state != PipelineState.RECOVERING and self.state_machine.can_transition(PipelineState.RECOVERING): + self.state_machine.transition(PipelineState.RECOVERING, event_type="stage_error") + if self.state_machine.can_transition(PipelineState.LISTENING): + self.state_machine.transition(PipelineState.LISTENING, event_type="recovered") + self._status(turn_id, "恢复监听：可直接说话") + + def _build_vad_recorder(self) -> VadRecorder: + if self.config.vad_provider == "hybrid": + provider = HybridVadProvider( + SherpaOnnxVadProvider(self.config.speech_models_dir, threshold=self.config.vad_threshold), + EnergyVadProvider(), + ) + elif self.config.vad_provider == "local": + provider = SherpaOnnxVadProvider(self.config.speech_models_dir, threshold=self.config.vad_threshold) + else: + provider = EnergyVadProvider() + return VadRecorder( + provider, + min_duration_ms=self.config.vad_min_duration_ms, + end_silence_ms=self.config.vad_end_silence_ms, + no_speech_timeout_ms=max(self.config.vad_no_speech_timeout_ms, 3_600_000), + max_recording_ms=self.config.vad_max_recording_ms, + ) + + def _build_stt_provider(self) -> SttProvider: + if self.config.speech_provider == "cloud": + return CloudAsrSttProvider(self.config) + return SherpaOnnxSttProvider(str(self.config.speech_models_dir)) + + def _build_tts_provider(self) -> TtsProvider: + if self.config.speech_provider == "cloud": + return CloudTtsProvider(self.config) + return MacSayTtsProvider(voice=self.config.tts_voice if self.config.tts_voice != "mimo_default" else None) + + def _status(self, turn_id: int | None, message: str) -> None: + prefix = f"[第{turn_id}轮] " if turn_id is not None else "" + print(f"{prefix}{message}", flush=True) + + def _transcript(self, turn_id: int, text: str, *, final: bool) -> None: + label = "转写结果" if final else "实时转写" + print(f"[第{turn_id}轮] {label}：{text}", flush=True) + + def _error(self, turn_id: int, stage: str, code: str, message: str) -> None: + print(f"[第{turn_id}轮] {stage}失败：{code} {message}", flush=True) def prepare_interrupt_controller( self, diff --git a/tests/test_full_duplex_integration.py b/tests/test_full_duplex_integration.py index 7602022..a324cd9 100644 --- a/tests/test_full_duplex_integration.py +++ b/tests/test_full_duplex_integration.py @@ -23,11 +23,52 @@ from owner_voice_pet.full_duplex_testing import ( diagnostics_contain_sensitive_data, sanitize_diagnostics, ) -from owner_voice_pet.models import Message, PipelineState +from owner_voice_pet.llm import MockLlmProvider +from owner_voice_pet.models import AudioFrame, Message, PipelineState +from owner_voice_pet.stt import MetadataSttProvider from owner_voice_pet.tool_router import MemorySearchTool, ToolCallRequest, ToolContext, ToolRouter +from owner_voice_pet.transport import MemoryAudioTransport +from owner_voice_pet.tts import SineTtsProvider +from owner_voice_pet.vad import EnergyVadProvider, VadRecorder class FullDuplexIntegrationTests(unittest.TestCase): + def test_run_agent_live_runtime_once_consumes_audio_and_replies(self) -> None: + frames = [ + AudioFrame(b"\xff\x7f", 16000, 1, 0, 1, {"duration_ms": 20, "speech": True, "transcript": "直接提问"}), + AudioFrame(b"\xff\x7f", 16000, 1, 20, 2, {"duration_ms": 20, "speech": True, "transcript": "直接提问"}), + AudioFrame(b"\x00\x00", 16000, 1, 40, 3, {"duration_ms": 20, "speech": False, "transcript": "直接提问"}), + AudioFrame(b"\x00\x00", 16000, 1, 60, 4, {"duration_ms": 20, "speech": False, "transcript": "直接提问"}), + ] + transport = MemoryAudioTransport(frames, flush_clears_input=False) + runtime = FullDuplexAgentRuntime( + config=AppConfig( + audio_apm_provider="fake", + audio_apm_required=False, + memory_enabled=False, + tool_router_enabled=False, + vad_min_duration_ms=40, + vad_end_silence_ms=40, + end_chime_enabled=False, + ), + transport=transport, + vad_recorder=VadRecorder(EnergyVadProvider(), min_duration_ms=40, end_silence_ms=40), + stt=MetadataSttProvider(), + llm=MockLlmProvider(["这是全双工回答。"]), + tts=SineTtsProvider(), + ) + + summary = runtime.run(once=True) + + self.assertEqual(summary.completed_turns, 1) + self.assertEqual(summary.failed_turns, 0) + self.assertEqual([message.role for message in runtime.context.messages()], ["user", "assistant"]) + self.assertEqual(runtime.context.messages()[0].content, "直接提问") + self.assertEqual(runtime.context.messages()[1].content, "这是全双工回答。") + self.assertGreaterEqual(len(transport.played_segments), 1) + self.assertIsNotNone(runtime.audio_hub) + self.assertGreater(runtime.audio_hub.processed_capture.frame_count, 0) + def test_fake_apm_echo_does_not_trigger_interruption(self) -> None: fixture = build_fake_full_duplex_audio_fixture() apm = FakeWebRtcAudioProcessingProvider()