[结束提示音优化]:完成Codex通知音资产内置,包含音频提取、配置默认值和回归测试

This commit is contained in:
mkbk
2026-06-18 14:10:34 +08:00
parent 80549fb312
commit d83f5a19cb
14 changed files with 102 additions and 8 deletions
+1
View File
@@ -46,6 +46,7 @@ OWNER_BARGE_IN_ENABLED=1
OWNER_BARGE_IN_MIN_SPEECH_MS=250
OWNER_BARGE_IN_ECHO_GUARD_MS=500
OWNER_END_CHIME_ENABLED=1
OWNER_END_CHIME_FILE=assets/sounds/codex-notification.wav
OWNER_END_CHIME_FREQUENCY_HZ=880
OWNER_END_CHIME_DURATION_MS=140
OWNER_WAKE_WORD=小杰小杰
+1
View File
@@ -8,6 +8,7 @@ __pycache__/
*.py[cod]
*.aiff
*.wav
!assets/sounds/codex-notification.wav
*.flac
.pytest_cache/
.mypy_cache/
+2 -1
View File
@@ -73,6 +73,7 @@ OWNER_BARGE_IN_ENABLED=1
OWNER_BARGE_IN_MIN_SPEECH_MS=250
OWNER_BARGE_IN_ECHO_GUARD_MS=500
OWNER_END_CHIME_ENABLED=1
OWNER_END_CHIME_FILE=assets/sounds/codex-notification.wav
OWNER_END_CHIME_FREQUENCY_HZ=880
OWNER_END_CHIME_DURATION_MS=140
```
@@ -87,7 +88,7 @@ OWNER_END_CHIME_DURATION_MS=140
`OWNER_BARGE_IN_ENABLED=1` 表示播报中允许打断。播放开始后的 `OWNER_BARGE_IN_ECHO_GUARD_MS=500` 毫秒内忽略麦克风输入,之后如果检测到至少 `OWNER_BARGE_IN_MIN_SPEECH_MS=250` 毫秒有效用户语音,并且 realtime STT 给出有效 partial,就停止剩余播报。上下文只记录已经完整播出的 assistant 句子,未播出的内容不会写入临时历史。
`OWNER_END_CHIME_ENABLED=1` 表示对话自然结束或追问超时恢复待机前会播放一声本地短提示音。提示音不走 TTS,也不会写入上下文;`OWNER_END_CHIME_FREQUENCY_HZ``OWNER_END_CHIME_DURATION_MS` 可以调整音高和时长,设置 `OWNER_END_CHIME_ENABLED=0` 可以关闭。
`OWNER_END_CHIME_ENABLED=1` 表示对话自然结束或追问超时恢复待机前会播放一声项目内置提示音,默认文件是 `assets/sounds/codex-notification.wav`。提示音不走 TTS,也不会写入上下文;如果 `OWNER_END_CHIME_FILE` 指向的文件缺失,会回退到本地合成短音,`OWNER_END_CHIME_FREQUENCY_HZ``OWNER_END_CHIME_DURATION_MS` 只影响这个回退音。设置 `OWNER_END_CHIME_ENABLED=0` 可以关闭。
## 本地模型
Binary file not shown.
+2 -1
View File
@@ -27,7 +27,7 @@ from .llm import MockLlmProvider, OpenAICompatibleLlmProvider
from .pipeline import PipelineResult, VoicePipeline
from .runtime import LiveVoiceRuntime, RuntimeSummary, TerminalRuntimeReporter, TurnResult, build_live_runtime
from .simulation import run_simulated_live
from .tts import CloudTtsProvider, MacSayTtsProvider, SentenceBuffer, SineTtsProvider, make_prompt_chime, sanitize_tts_text
from .tts import CloudTtsProvider, MacSayTtsProvider, SentenceBuffer, SineTtsProvider, make_end_chime, make_prompt_chime, sanitize_tts_text
from .assets import validate_pet_assets
from .ui import ConsolePetWindow, PetStateController, PetVisualState
@@ -70,6 +70,7 @@ __all__ = [
"MacSayTtsProvider",
"SentenceBuffer",
"SineTtsProvider",
"make_end_chime",
"make_prompt_chime",
"sanitize_tts_text",
"validate_pet_assets",
+3 -2
View File
@@ -47,7 +47,7 @@ from .protocols import (
WakeWordProvider,
)
from .stt import is_valid_transcript_text
from .tts import SentenceBuffer, make_prompt_chime, sanitize_tts_text
from .tts import SentenceBuffer, make_end_chime, sanitize_tts_text
from .vad import VadRecorder
@@ -491,7 +491,8 @@ class TurnController:
def _play_end_chime(self) -> None:
if not self.config.end_chime_enabled:
return
segment = make_prompt_chime(
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,
+1
View File
@@ -105,6 +105,7 @@ def main(argv: list[str] | None = None) -> int:
"barge_in_min_speech_ms": config.barge_in_min_speech_ms,
"barge_in_echo_guard_ms": config.barge_in_echo_guard_ms,
"end_chime_enabled": config.end_chime_enabled,
"end_chime_file": str(config.end_chime_file),
"end_chime_frequency_hz": config.end_chime_frequency_hz,
"end_chime_duration_ms": config.end_chime_duration_ms,
},
+5
View File
@@ -60,6 +60,7 @@ class AppConfig:
barge_in_min_speech_ms: int = 250
barge_in_echo_guard_ms: int = 500
end_chime_enabled: bool = True
end_chime_file: Path = Path("assets/sounds/codex-notification.wav")
end_chime_frequency_hz: int = 880
end_chime_duration_ms: int = 140
@@ -134,6 +135,10 @@ class AppConfig:
barge_in_min_speech_ms=int(get("BARGE_IN_MIN_SPEECH_MS", "250") or "250"),
barge_in_echo_guard_ms=int(get("BARGE_IN_ECHO_GUARD_MS", "500") or "500"),
end_chime_enabled=(get("END_CHIME_ENABLED", "1") or "1").lower() not in {"0", "false", "no"},
end_chime_file=Path(
get("END_CHIME_FILE", "assets/sounds/codex-notification.wav")
or "assets/sounds/codex-notification.wav"
),
end_chime_frequency_hz=int(get("END_CHIME_FREQUENCY_HZ", "880") or "880"),
end_chime_duration_ms=int(get("END_CHIME_DURATION_MS", "140") or "140"),
)
+3 -2
View File
@@ -33,7 +33,7 @@ from .models import AudioSegment, ErrorCode, PipelineState, ProviderError
from .protocols import AudioTransport, LlmProvider, RealtimeSttProvider, SttProvider, TtsProvider, WakeWordProvider
from .stt import CloudAsrSttProvider, SherpaOnnxSttProvider, is_valid_transcript_text
from .transport import SoundDeviceAudioTransport
from .tts import CloudTtsProvider, MacSayTtsProvider, SentenceBuffer, make_prompt_chime, sanitize_tts_text
from .tts import CloudTtsProvider, MacSayTtsProvider, SentenceBuffer, make_end_chime, sanitize_tts_text
from .vad import EnergyVadProvider, HybridVadProvider, PrimarySpeakerVadRecorder, SherpaOnnxVadProvider, VadRecorder
from .wakeword import SherpaOnnxKeywordWakeWordProvider
@@ -343,7 +343,8 @@ class LiveVoiceRuntime:
def _play_end_chime(self) -> None:
if not self.config.end_chime_enabled:
return
segment = make_prompt_chime(
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,
+49 -1
View File
@@ -170,7 +170,55 @@ def make_prompt_chime(
channels,
0,
duration_ms,
{"chime": "end", "text": "end_chime"},
{"chime": "end", "text": "end_chime", "source": "synthetic"},
)
def make_end_chime(
*,
file_path: str | Path | None = None,
frequency_hz: int = 880,
duration_ms: int = 140,
sample_rate: int = 16000,
channels: int = 1,
) -> AudioSegment:
if file_path is not None:
segment = _load_chime_file(Path(file_path))
if segment is not None:
return segment
return make_prompt_chime(
frequency_hz=frequency_hz,
duration_ms=duration_ms,
sample_rate=sample_rate,
channels=channels,
)
def _load_chime_file(path: Path) -> AudioSegment | None:
if not path.is_file():
return None
data = path.read_bytes()
if not data:
return None
suffix = path.suffix.lower().lstrip(".") or "wav"
sample_rate = 16000
channels = 1
duration_ms = 120
if suffix == "wav":
try:
with wave.open(str(path), "rb") as handle:
sample_rate = handle.getframerate()
channels = handle.getnchannels()
duration_ms = int(handle.getnframes() / max(1, sample_rate) * 1000)
except wave.Error:
pass
return AudioSegment(
data,
sample_rate,
channels,
0,
max(20, duration_ms),
{"chime": "end", "text": "end_chime", "format": suffix, "path": str(path), "source": "file"},
)
+1
View File
@@ -47,6 +47,7 @@ class CliAcceptanceTests(unittest.TestCase):
self.assertEqual(data["realtime_transcript_idle_timeout_ms"], 1500)
self.assertEqual(data["context_mode"], "session_memory")
self.assertTrue(data["end_chime_enabled"])
self.assertEqual(data["end_chime_file"], "assets/sounds/codex-notification.wav")
self.assertEqual(data["end_chime_duration_ms"], 140)
self.assertNotIn("secret-value", str(data))
+16
View File
@@ -272,7 +272,9 @@ class LiveRuntimeTests(unittest.TestCase):
self.assertEqual(len(transport.played_segments), 6)
self.assertEqual(transport.flush_count, 6)
self.assertEqual(transport.played_segments[2].metadata["chime"], "end")
self.assertEqual(transport.played_segments[2].metadata["source"], "file")
self.assertEqual(transport.played_segments[5].metadata["chime"], "end")
self.assertEqual(transport.played_segments[5].metadata["source"], "file")
self.assertEqual(reporter.transcripts, ["第一问", "第二问"])
self.assertIn("应答中:我在", reporter.statuses)
self.assertLess(reporter.statuses.index("唤醒命中"), reporter.statuses.index("应答中:我在"))
@@ -339,6 +341,7 @@ class LiveRuntimeTests(unittest.TestCase):
self.assertEqual(len(stt.calls), 1)
self.assertEqual(transport.flush_count, 3)
self.assertEqual(transport.played_segments[-1].metadata["chime"], "end")
self.assertEqual(transport.played_segments[-1].metadata["source"], "file")
def test_no_ack_text_does_not_drain_before_capture(self) -> None:
runtime, stt, _, transport, reporter = make_runtime(["第一问"], wake_ack_text="")
@@ -351,6 +354,18 @@ class LiveRuntimeTests(unittest.TestCase):
self.assertEqual(len(transport.played_segments), 2)
self.assertEqual(transport.flush_count, 2)
self.assertEqual(transport.played_segments[-1].metadata["chime"], "end")
self.assertEqual(transport.played_segments[-1].metadata["source"], "file")
def test_end_chime_can_be_disabled(self) -> None:
runtime, _, _, transport, _ = make_runtime(["第一问"], wake_ack_text="")
runtime.config = AppConfig(llm_api_key="secret", speech_provider="cloud", wake_ack_text="", end_chime_enabled=False)
runtime.controller.config = runtime.config
summary = runtime.run(max_turns=1)
self.assertEqual(summary.completed_turns, 1)
self.assertEqual(len(transport.played_segments), 1)
self.assertNotIn("chime", transport.played_segments[-1].metadata)
def test_temporary_context_is_sent_to_second_llm_call(self) -> None:
runtime, _, llm, _, _ = make_runtime(["第一问", "第二问"])
@@ -598,6 +613,7 @@ class LiveRuntimeTests(unittest.TestCase):
self.assertEqual(event_types[-1], STANDBY_RESUMED)
self.assertEqual(len(llm.calls), 1)
self.assertEqual(transport.played_segments[-1].metadata["chime"], "end")
self.assertEqual(transport.played_segments[-1].metadata["source"], "file")
def test_completed_reply_returns_to_standby_without_cloud_classifier_delay(self) -> None:
frames = [wake_frame(0, 0)]
+1
View File
@@ -114,6 +114,7 @@ class ModelsConfigTests(unittest.TestCase):
self.assertEqual(config.barge_in_min_speech_ms, 250)
self.assertEqual(config.barge_in_echo_guard_ms, 500)
self.assertTrue(config.end_chime_enabled)
self.assertEqual(str(config.end_chime_file), "assets/sounds/codex-notification.wav")
self.assertEqual(config.end_chime_frequency_hz, 880)
self.assertEqual(config.end_chime_duration_ms, 140)
self.assertTrue(config.llm_stream)
+17 -1
View File
@@ -12,7 +12,7 @@ from owner_voice_pet.models import AudioFrame, ErrorCode, Message, PipelineState
from owner_voice_pet.pipeline import VoicePipeline
from owner_voice_pet.stt import MetadataSttProvider
from owner_voice_pet.transport import MemoryAudioTransport
from owner_voice_pet.tts import CloudTtsProvider, SentenceBuffer, SineTtsProvider, make_prompt_chime, sanitize_tts_text
from owner_voice_pet.tts import CloudTtsProvider, SentenceBuffer, SineTtsProvider, make_end_chime, make_prompt_chime, sanitize_tts_text
from owner_voice_pet.vad import EnergyVadProvider, VadRecorder
from owner_voice_pet.wakeword import KeywordWakeWordProvider
@@ -78,8 +78,24 @@ class PipelineLlmTtsTests(unittest.TestCase):
self.assertEqual(segment.duration_ms, 140)
self.assertEqual(segment.metadata["chime"], "end")
self.assertEqual(segment.metadata["source"], "synthetic")
self.assertGreater(len(segment.pcm), 0)
def test_end_chime_uses_file_when_present(self) -> None:
segment = make_end_chime(file_path="assets/sounds/codex-notification.wav")
self.assertEqual(segment.metadata["source"], "file")
self.assertEqual(segment.metadata["path"], "assets/sounds/codex-notification.wav")
self.assertEqual(segment.metadata["format"], "wav")
self.assertGreater(segment.duration_ms, 0)
def test_end_chime_falls_back_to_synthetic_when_file_missing(self) -> None:
segment = make_end_chime(file_path="/tmp/owner-voice-pet-missing-chime.wav", duration_ms=140)
self.assertEqual(segment.metadata["source"], "synthetic")
self.assertEqual(segment.metadata["chime"], "end")
self.assertEqual(segment.duration_ms, 140)
def test_cloud_tts_posts_speech_request(self) -> None:
class FakeResponse:
def __enter__(self):