From caf35783f89c32905a7e32cfcb10788a2a4d0625 Mon Sep 17 00:00:00 2001 From: mkbk Date: Fri, 19 Jun 2026 13:26:25 +0800 Subject: [PATCH] =?UTF-8?q?[=E5=85=A8=E5=8F=8C=E5=B7=A5=E4=BE=9D=E8=B5=96?= =?UTF-8?q?=E6=8E=A5=E5=85=A5]=EF=BC=9A=E5=AE=8C=E6=88=90=E7=9C=9F?= =?UTF-8?q?=E5=AE=9EAPM=E4=BE=9D=E8=B5=96=E5=AE=89=E8=A3=85=E5=92=8C?= =?UTF-8?q?=E6=8E=A5=E7=BA=BF=EF=BC=8C=E5=8C=85=E5=90=ABaec-audio-processi?= =?UTF-8?q?ng=E3=80=81=E9=9F=B3=E9=A2=91=E8=87=AA=E6=B5=8B=E5=92=8C?= =?UTF-8?q?=E6=9C=80=E7=BB=88=E9=97=A8=E7=A6=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../tasks.md | 4 +- pyproject.toml | 2 +- src/owner_voice_pet/full_duplex_audio.py | 152 +++++++++++++++++- tests/test_cli_acceptance.py | 29 ++-- tests/test_full_duplex_audio.py | 29 +++- 5 files changed, 199 insertions(+), 17 deletions(-) diff --git a/openspec/changes/complete-full-duplex-agent-runtime/tasks.md b/openspec/changes/complete-full-duplex-agent-runtime/tasks.md index 2e6ae1d..2868970 100644 --- a/openspec/changes/complete-full-duplex-agent-runtime/tasks.md +++ b/openspec/changes/complete-full-duplex-agent-runtime/tasks.md @@ -41,5 +41,5 @@ - [x] 6.1 新增 `agent-self-test`；前置条件：Phase 5；优先级：P0；验收标准：三轮覆盖 STT/LLM/TTS/barge-in/memory/tool；测试要点：JSON success。 - [x] 6.2 新增 `audio-self-test`；前置条件：AudioHub/APM；优先级：P0；验收标准：输出 provider/device/echo/latency；测试要点：无 APM 时明确失败。 - [x] 6.3 更新 README 和 `.env.example`；前置条件：6.1-6.2；优先级：P0；验收标准：入口和配置无冲突；测试要点：命令示例准确。 -- [ ] 6.4 执行最终门禁；前置条件：全部实现完成；优先级：P0；验收标准：compileall、unittest、self-test、security、model、device、OpenSpec strict 全通过；测试要点：命令输出。 -- [ ] 6.5 Phase 6 提交；前置条件：6.1-6.4；优先级：P0；验收标准：中文提交 `[全双工自测]...`；测试要点：`git status --short` 为空。 +- [x] 6.4 执行最终门禁；前置条件：全部实现完成；优先级：P0；验收标准：compileall、unittest、self-test、security、model、device、OpenSpec strict 全通过；测试要点：命令输出。 +- [x] 6.5 Phase 6 提交；前置条件：6.1-6.4；优先级：P0；验收标准：中文提交 `[全双工自测]...`；测试要点：`git status --short` 为空。 diff --git a/pyproject.toml b/pyproject.toml index 2963692..df07ffa 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -15,7 +15,7 @@ dependencies = [] [project.optional-dependencies] audio = ["numpy>=1.26", "sounddevice>=0.4.7"] speech = ["numpy>=1.26", "sounddevice>=0.4.7", "sherpa-onnx>=1.13.3"] -full-duplex = ["numpy>=1.26", "sounddevice>=0.4.7"] +full-duplex = ["numpy>=1.26", "sounddevice>=0.4.7", "aec-audio-processing>=1.0.1"] streaming-stt = ["faster-whisper>=1.0"] memory = ["faiss-cpu>=1.8", "numpy>=1.26"] browser = ["playwright>=1.44"] diff --git a/src/owner_voice_pet/full_duplex_audio.py b/src/owner_voice_pet/full_duplex_audio.py index 6b02cbe..d3a951d 100644 --- a/src/owner_voice_pet/full_duplex_audio.py +++ b/src/owner_voice_pet/full_duplex_audio.py @@ -4,6 +4,7 @@ import importlib.util from collections import deque from dataclasses import dataclass from threading import RLock +from typing import Any from typing import Protocol from .config import AppConfig @@ -346,15 +347,142 @@ class FakeWebRtcAudioProcessingProvider: ) +class AecAudioProcessingProvider: + name = "aec_audio_processing" + + def __init__( + self, + *, + sample_rate: int = 16000, + channels: int = 1, + enable_aec: bool = True, + enable_ns: bool = True, + enable_agc: bool = True, + ) -> None: + self.sample_rate = sample_rate + self.channels = channels + self.enable_aec = enable_aec + self.enable_ns = enable_ns + self.enable_agc = enable_agc + self.render_frames: list[AudioFrame] = [] + self._processor: Any = None + self._load_processor() + + def process_capture(self, frame: AudioFrame) -> AudioFrame: + self._validate_format(frame) + try: + pcm = self._process_pcm_chunks(frame.pcm, self._processor.process_stream) + except Exception as exc: + raise ProviderError( + ErrorCode.AUDIO_APM_PROCESS_FAILED, + f"aec_audio_processing capture failed: {exc}", + True, + self.name, + "audio_apm", + ) from exc + metadata = dict(frame.metadata) + metadata["apm_provider"] = self.name + if metadata.get("assistant_echo") and self.render_frames and _rms_int16(pcm) <= _rms_int16(frame.pcm) * 0.85: + metadata["echo_suppressed"] = True + metadata["speech"] = False + return AudioFrame( + pcm, + frame.sample_rate, + frame.channels, + frame.timestamp_ms, + frame.frame_id, + metadata, + ) + + def process_render(self, frame: AudioFrame) -> None: + self._validate_format(frame) + try: + self._process_pcm_chunks(frame.pcm, self._processor.process_reverse_stream) + except Exception as exc: + raise ProviderError( + ErrorCode.AUDIO_APM_PROCESS_FAILED, + f"aec_audio_processing render reference failed: {exc}", + True, + self.name, + "audio_apm", + ) from exc + self.render_frames.append(frame) + + def reset_stream(self) -> None: + self.render_frames.clear() + self._load_processor() + + def health_check(self) -> AudioProcessingHealth: + return AudioProcessingHealth( + provider=self.name, + available=True, + message="aec_audio_processing native WebRTC APM binding loaded", + ) + + def _load_processor(self) -> None: + try: + from aec_audio_processing import AudioProcessor # type: ignore[import-not-found] + + self._processor = AudioProcessor( + enable_aec=self.enable_aec, + enable_ns=self.enable_ns, + enable_agc=self.enable_agc, + enable_vad=False, + ) + self._processor.set_stream_format(self.sample_rate, self.channels, self.sample_rate, self.channels) + self._processor.set_reverse_stream_format(self.sample_rate, self.channels) + except Exception as exc: + raise ProviderError( + ErrorCode.AUDIO_APM_UNAVAILABLE, + f"cannot initialize aec_audio_processing: {exc}", + False, + self.name, + "audio_apm", + ) from exc + + def _process_pcm_chunks(self, pcm: bytes, processor_method) -> bytes: + frame_bytes = max(2 * self.channels, int(self._processor.get_frame_size()) * self.channels * 2) + frame_bytes -= frame_bytes % (2 * self.channels) + if frame_bytes <= 0: + raise ValueError("invalid APM frame size") + processed = bytearray() + offset = 0 + while offset < len(pcm): + chunk = pcm[offset : offset + frame_bytes] + padding = b"" + if len(chunk) < frame_bytes: + padding = b"\x00" * (frame_bytes - len(chunk)) + chunk += padding + output = processor_method(chunk) + if output is None: + output = chunk + if padding: + output = output[: len(output) - len(padding)] + processed.extend(output) + offset += frame_bytes + return bytes(processed) + + def _validate_format(self, frame: AudioFrame) -> None: + if frame.sample_rate != self.sample_rate or frame.channels != self.channels: + raise ProviderError( + ErrorCode.AUDIO_APM_FORMAT_MISMATCH, + "audio frame format does not match aec_audio_processing configuration", + False, + self.name, + "audio_apm", + ) + + def webrtc_apm_probe() -> AudioProcessingHealth: candidates = ( + "aec_audio_processing", "webrtc_audio_processing", "webrtc_audio_processing_module", ) for module_name in candidates: if importlib.util.find_spec(module_name) is not None: return AudioProcessingHealth( - provider="webrtc", + provider=module_name, available=True, message=f"found {module_name}", ) @@ -387,7 +515,15 @@ def build_audio_processing_provider(config: AppConfig) -> AudioProcessingProvide health = webrtc_apm_probe() if health.available: - message = "WebRTC APM binding is detected but native processing is not wired yet" + if health.provider == "aec_audio_processing": + return AecAudioProcessingProvider( + sample_rate=config.sample_rate, + channels=config.channels, + enable_aec=config.audio_aec_enabled, + enable_ns=config.audio_ns_enabled, + enable_agc=config.audio_agc_enabled, + ) + message = f"{health.provider} binding is detected but native processing is not wired yet" if config.audio_apm_required: raise ProviderError( ErrorCode.AUDIO_APM_UNAVAILABLE, @@ -409,3 +545,15 @@ def build_audio_processing_provider(config: AppConfig) -> AudioProcessingProvide fallback_active=True, message=health.message, ) + + +def _rms_int16(pcm: bytes) -> float: + if len(pcm) < 2: + return 0.0 + samples = [ + int.from_bytes(pcm[index : index + 2], "little", signed=True) + for index in range(0, len(pcm) - 1, 2) + ] + if not samples: + return 0.0 + return (sum(sample * sample for sample in samples) / len(samples)) ** 0.5 diff --git a/tests/test_cli_acceptance.py b/tests/test_cli_acceptance.py index 0fa2c0c..c73153c 100644 --- a/tests/test_cli_acceptance.py +++ b/tests/test_cli_acceptance.py @@ -85,9 +85,10 @@ class CliAcceptanceTests(unittest.TestCase): self.assertTrue(data["success"]) self.assertEqual(data["command"], "run-agent-live") self.assertEqual(data["assistant_mode"], "full_duplex_agent") - self.assertFalse(data["full_duplex_runtime_ready"]) + self.assertTrue(data["full_duplex_runtime_ready"]) self.assertEqual(data["audio_apm_provider"], "webrtc") - self.assertEqual(data["audio_apm_error_code"], "AUDIO_APM_UNAVAILABLE") + self.assertEqual(data["audio_apm_error_code"], "") + self.assertTrue(data["audio_apm_available"]) self.assertTrue(data["llm_streaming_enabled"]) self.assertEqual(data["turn_based_entry"], "run-live") @@ -215,16 +216,22 @@ class CliAcceptanceTests(unittest.TestCase): self.assertTrue(data["checks"]["barge_in"]) self.assertTrue(data["checks"]["tool_router"]) - def test_audio_self_test_reports_missing_required_apm(self) -> None: + def test_audio_self_test_reports_missing_required_apm_when_probe_fails(self) -> None: with tempfile.TemporaryDirectory() as tmp: - code, data = self.call( - "--env-file", - str(Path(tmp) / ".env"), - "audio-self-test", - "--duration", - "1", - "--check-echo", - ) + with patch("owner_voice_pet.full_duplex_audio.webrtc_apm_probe") as probe: + probe.return_value = type( + "Health", + (), + {"available": False, "provider": "webrtc", "message": "missing binding"}, + )() + code, data = self.call( + "--env-file", + str(Path(tmp) / ".env"), + "audio-self-test", + "--duration", + "1", + "--check-echo", + ) self.assertEqual(code, 1) self.assertFalse(data["success"]) diff --git a/tests/test_full_duplex_audio.py b/tests/test_full_duplex_audio.py index b64d246..8b90019 100644 --- a/tests/test_full_duplex_audio.py +++ b/tests/test_full_duplex_audio.py @@ -5,6 +5,7 @@ from unittest.mock import patch from owner_voice_pet.config import AppConfig from owner_voice_pet.full_duplex_audio import ( + AecAudioProcessingProvider, AudioHub, CaptureRingBuffer, FakeWebRtcAudioProcessingProvider, @@ -181,12 +182,26 @@ class FullDuplexAudioTests(unittest.TestCase): self.assertIsInstance(provider, NoopAudioProcessingProvider) self.assertTrue(provider.health_check().fallback_active) + def test_webrtc_provider_uses_aec_audio_processing_when_available(self) -> None: + with patch("owner_voice_pet.full_duplex_audio.webrtc_apm_probe") as probe: + probe.return_value = type( + "Health", + (), + {"available": True, "provider": "aec_audio_processing", "message": "found binding"}, + )() + provider = build_audio_processing_provider( + AppConfig(audio_apm_provider="webrtc", audio_apm_required=True) + ) + + self.assertIsInstance(provider, AecAudioProcessingProvider) + self.assertTrue(provider.health_check().available) + def test_detected_but_unwired_webrtc_provider_fails_when_required(self) -> None: with patch("owner_voice_pet.full_duplex_audio.webrtc_apm_probe") as probe: probe.return_value = type( "Health", (), - {"available": True, "message": "found binding"}, + {"available": True, "provider": "webrtc_audio_processing", "message": "found binding"}, )() with self.assertRaises(ProviderError) as raised: build_audio_processing_provider( @@ -195,6 +210,18 @@ class FullDuplexAudioTests(unittest.TestCase): self.assertEqual(raised.exception.code, ErrorCode.AUDIO_APM_UNAVAILABLE) + def test_aec_audio_processing_provider_processes_capture_and_render(self) -> None: + provider = AecAudioProcessingProvider(sample_rate=16000, channels=1) + provider.process_render(frame(1, 0, pcm=b"\x00\x00" * 160)) + + processed = provider.process_capture(frame(2, 20, pcm=b"\x01\x00" * 320, duration_ms=20)) + + self.assertEqual(processed.sample_rate, 16000) + self.assertEqual(processed.channels, 1) + self.assertEqual(len(processed.pcm), 640) + self.assertEqual(processed.metadata["apm_provider"], "aec_audio_processing") + self.assertEqual(len(provider.render_frames), 1) + def test_webrtc_probe_reports_unavailable_without_binding(self) -> None: with patch("importlib.util.find_spec", return_value=None): health = webrtc_apm_probe()