From 7519725321629ba2df7ae443f5f82ddbbdcd2702 Mon Sep 17 00:00:00 2001 From: mkbk Date: Thu, 18 Jun 2026 12:08:41 +0800 Subject: [PATCH] =?UTF-8?q?[=E8=BF=9E=E7=BB=AD=E5=AF=B9=E8=AF=9D=E5=88=A4?= =?UTF-8?q?=E6=96=AD]=EF=BC=9A=E5=AE=8C=E6=88=90=E8=87=AA=E5=8A=A8?= =?UTF-8?q?=E6=8C=81=E7=BB=AD=E5=AF=B9=E8=AF=9D=E5=92=8C=E6=92=AD=E6=8A=A5?= =?UTF-8?q?=E6=89=93=E6=96=AD=EF=BC=8C=E5=8C=85=E5=90=AB=E5=9B=9E=E5=A4=8D?= =?UTF-8?q?=E6=84=8F=E5=9B=BE=E5=88=A4=E6=96=AD=E3=80=81=E5=85=8D=E5=94=A4?= =?UTF-8?q?=E9=86=92=E8=BF=BD=E9=97=AE=E5=92=8C=E6=89=93=E6=96=AD=E5=9B=9E?= =?UTF-8?q?=E5=BD=92=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 7 + README.md | 19 +- .../design.md | 31 ++ .../proposal.md | 3 + .../specs/voice-pet-pipeline/spec.md | 46 ++ .../tasks.md | 8 + src/owner_voice_pet/assistant_pipeline.py | 417 +++++++++++++++--- src/owner_voice_pet/cli.py | 55 +-- src/owner_voice_pet/config.py | 54 +++ src/owner_voice_pet/continuation.py | 205 +++++++++ src/owner_voice_pet/events.py | 7 + src/owner_voice_pet/real_live_check.py | 2 + src/owner_voice_pet/tts.py | 27 +- tests/test_continuation.py | 81 ++++ tests/test_live_runtime.py | 193 +++++++- tests/test_models_config.py | 23 + 16 files changed, 1059 insertions(+), 119 deletions(-) create mode 100644 src/owner_voice_pet/continuation.py create mode 100644 tests/test_continuation.py diff --git a/.env.example b/.env.example index 1e03f3f..d55e123 100644 --- a/.env.example +++ b/.env.example @@ -38,6 +38,13 @@ OWNER_SPEECH_MODELS_DIR=models OWNER_CONTEXT_MODE=session_memory OWNER_CONTEXT_MAX_MESSAGES=12 OWNER_CONTEXT_MAX_CHARS=12000 +OWNER_CONTINUOUS_DIALOG_ENABLED=1 +OWNER_CONTINUATION_DECISION_PROVIDER=hybrid +OWNER_CONTINUATION_CONFIDENCE_THRESHOLD=0.65 +OWNER_FOLLOWUP_LISTEN_TIMEOUT_MS=3000 +OWNER_BARGE_IN_ENABLED=1 +OWNER_BARGE_IN_MIN_SPEECH_MS=250 +OWNER_BARGE_IN_ECHO_GUARD_MS=500 OWNER_WAKE_WORD=小杰小杰 OWNER_SAMPLE_RATE=16000 OWNER_CHANNELS=1 diff --git a/README.md b/README.md index b8925b3..34c00c5 100644 --- a/README.md +++ b/README.md @@ -16,6 +16,8 @@ - 本地模型：`models/` 存放 `sherpa-onnx` wake/VAD/STT/denoiser 模型，目录不提交 Git。 - Pipeline：`run-live` 使用 stage 化 `VoiceAssistantPipeline`，通过事件总线输出终端状态。 - 主说话人端点：默认 `OWNER_ENDPOINT_MODE=primary_speaker`，本轮音色消失后结束录音，避免背景噪声拖慢 STT。 +- 自动持续对话：助手回复后自动判断是否需要继续听用户回答，默认规则优先、LLM 小分类器兜底，不确定就恢复待机。 +- 播报打断：助手播报超过回声保护期后，如果检测到有效用户语音和实时字幕，会停止剩余播报并处理新输入。 - 临时上下文：同一次 `run-live` 进程内携带最近 user/assistant 历史，退出即清空。 ## 首次准备 @@ -63,6 +65,13 @@ OWNER_ASR_MODEL=mimo-v2.5-asr OWNER_TTS_MODEL=mimo-v2.5-tts OWNER_TTS_VOICE=mimo_default OWNER_CONTEXT_MODE=session_memory +OWNER_CONTINUOUS_DIALOG_ENABLED=1 +OWNER_CONTINUATION_DECISION_PROVIDER=hybrid +OWNER_CONTINUATION_CONFIDENCE_THRESHOLD=0.65 +OWNER_FOLLOWUP_LISTEN_TIMEOUT_MS=3000 +OWNER_BARGE_IN_ENABLED=1 +OWNER_BARGE_IN_MIN_SPEECH_MS=250 +OWNER_BARGE_IN_ECHO_GUARD_MS=500 ``` `OWNER_WAKE_PROVIDER=local_kws` 表示唤醒词“小杰小杰”由本地模型检测。唤醒命中后会先本地播报 `OWNER_WAKE_ACK_TEXT=我在`，再开始听取问题。`OWNER_SPEECH_PROVIDER=local` 表示正式问题 STT、实时字幕和 TTS 都走本地模型或 macOS 本地能力；云端只接收 final 文本和本次会话历史用于 LLM 回复。 @@ -71,6 +80,10 @@ OWNER_CONTEXT_MODE=session_memory `OWNER_REALTIME_TRANSCRIPT_ENABLED=1` 表示录音期间会使用本地 streaming STT 实时显示中间转写，终端会输出 `实时转写：...`。为了过滤噪音，实时字幕默认不会显示 `家`、`家确` 这类很短的瞬时误识别；最终发送给 LLM 的内容仍只以 `转写结果：...` 为准。`OWNER_REALTIME_TRANSCRIPT_IDLE_TIMEOUT_MS=1500` 表示已有实时字幕后，如果 1.5 秒内没有新的文字输出，就直接结束本轮录音进入 final STT；设置为 `0` 可以关闭这个停滞端点。`OWNER_REALTIME_TRANSCRIPT_ENABLED=0` 可以临时关闭实时显示。 +`OWNER_CONTINUOUS_DIALOG_ENABLED=1` 表示每轮回复播放后会自动判断是否继续对话。若助手回复里明显在问用户、要求补充信息或让用户选择，终端会输出 `继续对话：3秒内可直接回答`，这 3 秒内可以不用再说“小杰小杰”。若助手只是完成回答、报错、拒绝或判断不确定，就直接恢复待机。`OWNER_CONTINUATION_DECISION_PROVIDER=hybrid` 表示先用本地规则判断，规则不确定时才调用云端 LLM 做小分类；低于 `OWNER_CONTINUATION_CONFIDENCE_THRESHOLD=0.65` 的结果按待机处理。 + +`OWNER_BARGE_IN_ENABLED=1` 表示播报中允许打断。播放开始后的 `OWNER_BARGE_IN_ECHO_GUARD_MS=500` 毫秒内忽略麦克风输入，之后如果检测到至少 `OWNER_BARGE_IN_MIN_SPEECH_MS=250` 毫秒有效用户语音，并且 realtime STT 给出有效 partial，就停止剩余播报。上下文只记录已经完整播出的 assistant 句子，未播出的内容不会写入临时历史。 + ## 本地模型 首次运行前必须准备本地语音模型；同一脚本会下载 wake、VAD、2025 中文 CTC STT 和 GTCRN denoiser 模型： @@ -137,15 +150,15 @@ python3.11 scripts/download_speech_models.py --dir models .venv/bin/python -m owner_voice_pet run-live ``` -运行后终端状态来自 pipeline event bus，会显示待机、唤醒命中、应答中、请说出问题、录音中、检测到用户语音、实时转写、用户语音结束、转写中、转写结果、思考中、播放中、恢复待机等状态。说“小杰小杰”，听到“我在”且看到“请说出问题”后再提问；听到回复后可以再次说“小杰小杰”继续下一轮。本次进程内会携带临时历史，程序退出后不保存。背景噪声下如果实时字幕仍偶发短错字，先看最终 `转写结果`；最终文本才会进入 LLM。 +运行后终端状态来自 pipeline event bus，会显示待机、唤醒命中、应答中、请说出问题、录音中、检测到用户语音、实时转写、用户语音结束、转写中、转写结果、思考中、播放中、继续对话或恢复待机等状态。说“小杰小杰”，听到“我在”且看到“请说出问题”后再提问；如果助手回复后判断需要你继续回答，可以在 3 秒内直接说下一句，不需要再次唤醒。若助手已经完成回答，会自动恢复待机。本次进程内会携带临时历史，程序退出后不保存。背景噪声下如果实时字幕仍偶发短错字，先看最终 `转写结果`；最终文本才会进入 LLM。 ## 验证 ```bash .venv/bin/python -m compileall src tests scripts .venv/bin/python -m unittest discover -s tests -.venv/bin/python -m owner_voice_pet simulate-live --turns 2 -.venv/bin/python -m owner_voice_pet real-live-check --turns 2 +.venv/bin/python -m owner_voice_pet simulate-live --turns 3 +.venv/bin/python -m owner_voice_pet real-live-check --turns 2 --no-playback .venv/bin/python -m owner_voice_pet acceptance .venv/bin/python -m owner_voice_pet validate-assets .venv/bin/python -m owner_voice_pet security-check diff --git a/openspec/changes/separate-wake-and-realtime-transcript/design.md b/openspec/changes/separate-wake-and-realtime-transcript/design.md index a34ab54..e625ea5 100644 --- a/openspec/changes/separate-wake-and-realtime-transcript/design.md +++ b/openspec/changes/separate-wake-and-realtime-transcript/design.md @@ -295,6 +295,37 @@ The real LLM provider is wrapped by `RecordingLlmProvider`. For every cloud LLM Timing output is diagnostic metadata only. It SHALL NOT include `OWNER_LLM_API_KEY`, raw audio, denoised audio, full provider request headers, or any other secret-bearing configuration. +## Automatic Continuous Dialog Decision + +持续对话不再等同于“每次回复后固定打开 3 秒追问窗口”。新的回复后状态流为： + +```text +llm_started + -> tts_started/playback_finished + -> continuation_decision_started + -> continuation_decision_made(continue|standby) + -> standby_resumed + OR followup_listening -> capture_started -> transcript_final -> llm_started ... +``` + +`ContinuationDecisionStage` 使用 hybrid provider： + +1. `RuleContinuationDecisionProvider` 先用可解释规则判断。明确问题、补充信息请求、选择题、等待用户下一步的表达判定为 `continue`；完成性陈述、播报结果、拒绝/报错、寒暄结束和不确定表达判定为 `standby`。 +2. `LlmContinuationDecisionProvider` 只在规则返回 uncertain 时调用云端 LLM 小分类器。分类器 prompt 要求只输出 `continue` 或 `standby`，可附带 `confidence`；解析失败或低于 `OWNER_CONTINUATION_CONFIDENCE_THRESHOLD` 时按 `standby`。 +3. `HybridContinuationDecisionProvider` 是默认 provider，由 `OWNER_CONTINUATION_DECISION_PROVIDER=hybrid` 启用；`rule` 可用于完全离线决策。 + +Follow-up 捕获复用现有 capture、实时字幕、final STT、LLM、TTS 流程，但跳过 wake 和 ACK。为了让“3 秒内无回答”真正快退，follow-up 捕获临时把 recorder no-speech timeout 限制为 `OWNER_FOLLOWUP_LISTEN_TIMEOUT_MS`；若返回 `VAD_TIMEOUT_NO_SPEECH`，事件层输出 `followup_timeout` 和 `continuous_session_ended`，然后恢复 standby。 + +## Barge-in Playback Interruption + +第一版打断只解决 turn-based 播放中的用户抢话，不做电话式 full-duplex。实现边界如下： + +1. `MacSayTtsProvider` 产物优先转成 16 kHz/16-bit/mono PCM，使播放阶段可以切成约 100 ms 小段，避免整段 `afplay` 阻塞。 +2. `SpeakStage` 播放每个小段后检查 microphone queue。`OWNER_BARGE_IN_ECHO_GUARD_MS` 内只 flush/忽略，降低播报回声误触发。 +3. echo guard 之后，若 VAD 累计有效语音达到 `OWNER_BARGE_IN_MIN_SPEECH_MS` 且 realtime STT partial 有效，发出 `barge_in_detected` 并停止剩余句子/剩余回复播放。 +4. 已完整播放的句子文本才写入 process-local assistant context；未播出的句子和后续未播放 token 不写入上下文。 +5. 打断后的用户输入进入免唤醒 follow-up 捕获路径，保留实时字幕和 final STT，最终仍只有 final transcript 进入 LLM。 + ## Migration No database migration. Users should run: diff --git a/openspec/changes/separate-wake-and-realtime-transcript/proposal.md b/openspec/changes/separate-wake-and-realtime-transcript/proposal.md index 02c3805..1982455 100644 --- a/openspec/changes/separate-wake-and-realtime-transcript/proposal.md +++ b/openspec/changes/separate-wake-and-realtime-transcript/proposal.md @@ -485,6 +485,8 @@ standby 14. `Real live check CLI`：新增 `real-live-check` 可重复命令，把完整真实 Provider 链路验收从一次性脚本升级为仓库内正式工具。 15. `Real live check timing`：新增真实流程命令计时输出，覆盖命令发起时间、LLM 请求发送时间、各阶段响应耗时和 pipeline 事件时间线。 16. `Realtime transcript idle endpoint`：新增已有实时字幕后 1.5 秒无新文字即结束录音的端点要求，避免用户已停说但 capture 继续等待 VAD/音色端点。 +17. `Automatic continuous dialog decision`：新增回复后自动判断继续/待机能力，使用规则优先、LLM 小分类兜底，不确定时默认回到待机。 +18. `Barge-in playback interruption`：新增播报中用户有效说话打断能力，使用 echo guard、最短语音时长和 realtime partial 共同确认，避免播放回声误触发。 ### 删除项 @@ -510,6 +512,7 @@ standby 12. M12：`real-live-check` 正式 CLI、文档、测试和真实命令验收完成后提交；仍保留物理麦克风 `run-live` 真人体验验收。 13. M13：`real-live-check` 计时输出完成后提交，使后续排查可以直接看到每段耗时。 14. M14：实时字幕停滞端点完成后提交，用户真实 `run-live` 若已有 partial 后 1.5 秒无新文字，应快速进入 final STT。 +15. M15：自动持续对话判断与播报打断完成后提交，用户真实 `run-live` 应能在助手反问时免唤醒继续回答，在助手已完成回答时自动恢复待机。 估时： diff --git a/openspec/changes/separate-wake-and-realtime-transcript/specs/voice-pet-pipeline/spec.md b/openspec/changes/separate-wake-and-realtime-transcript/specs/voice-pet-pipeline/spec.md index 9543857..dfe4989 100644 --- a/openspec/changes/separate-wake-and-realtime-transcript/specs/voice-pet-pipeline/spec.md +++ b/openspec/changes/separate-wake-and-realtime-transcript/specs/voice-pet-pipeline/spec.md @@ -135,6 +135,52 @@ The system SHALL provide a deterministic simulated microphone acceptance command - **WHEN** the simulated microphone transport runs out of frames before the expected turn completes - **THEN** the command SHALL fail with a structured error instead of waiting indefinitely +### Requirement: Automatic continuous dialog decision +The live runtime SHALL automatically decide after each assistant reply whether to continue listening for a direct follow-up answer or to end the continuous session and return to wake standby. + +#### Scenario: Assistant asks the user a direct follow-up question +- **WHEN** an assistant reply clearly asks the user a question, requests missing information, asks the user to choose among options, or otherwise waits for the user's next answer +- **THEN** the continuation decision SHALL be `continue` with confidence at or above the configured threshold +- **AND** the runtime SHALL enter follow-up listening after playback without requiring the wake word again + +#### Scenario: Assistant completes the answer +- **WHEN** an assistant reply only provides an answer, reports a result, refuses, reports an error, ends a greeting, or is ambiguous about needing more user input +- **THEN** the continuation decision SHALL be `standby` +- **AND** the runtime SHALL return to wake standby after playback + +#### Scenario: Hybrid decision falls back to an LLM classifier +- **WHEN** deterministic rules cannot decide continue or standby +- **THEN** the hybrid provider SHALL call the configured cloud LLM with a classifier prompt that can only resolve to `continue` or `standby` +- **AND** a low-confidence, malformed, timed-out, or failed classifier result SHALL be treated as `standby` + +#### Scenario: Follow-up answer arrives in the listening window +- **WHEN** the decision is `continue` and the user starts speaking within `OWNER_FOLLOWUP_LISTEN_TIMEOUT_MS` +- **THEN** the runtime SHALL capture, realtime-transcribe, final-transcribe, send to LLM, synthesize, and play the answer without local wake detection and without the wake acknowledgement phrase + +#### Scenario: Follow-up listening times out +- **WHEN** the decision is `continue` but no user speech is detected before `OWNER_FOLLOWUP_LISTEN_TIMEOUT_MS` +- **THEN** the runtime SHALL emit `followup_timeout` +- **AND** it SHALL emit `continuous_session_ended` +- **AND** it SHALL return to wake standby + +### Requirement: Barge-in playback interruption +The live runtime SHALL support a first-version barge-in path that can stop assistant playback when the user starts a valid new utterance during TTS playback. + +#### Scenario: Echo guard suppresses early playback echo +- **WHEN** TTS playback starts and microphone input appears before `OWNER_BARGE_IN_ECHO_GUARD_MS` +- **THEN** the runtime SHALL NOT treat that input as a user barge-in + +#### Scenario: Valid user speech interrupts playback +- **WHEN** TTS playback has passed the echo guard, VAD observes at least `OWNER_BARGE_IN_MIN_SPEECH_MS` of user speech, and realtime STT produces a valid partial transcript +- **THEN** the runtime SHALL emit `barge_in_detected` +- **AND** it SHALL stop remaining playback chunks and emit `playback_interrupted` +- **AND** it SHALL process the user's new utterance as the next input without requiring wake + +#### Scenario: Interrupted assistant context is partial +- **WHEN** playback is interrupted before the full assistant reply is spoken +- **THEN** the process-local conversation context SHALL append only the assistant text that had been fully spoken before interruption +- **AND** unspoken assistant text SHALL NOT be appended to the session history + ### Requirement: Real provider live fixture acceptance The system SHALL support scripted full-chain acceptance that drives the live pipeline with generated microphone PCM while using real local speech providers, the configured cloud LLM, and local TTS/playback capability. diff --git a/openspec/changes/separate-wake-and-realtime-transcript/tasks.md b/openspec/changes/separate-wake-and-realtime-transcript/tasks.md index 536d0d1..4065e86 100644 --- a/openspec/changes/separate-wake-and-realtime-transcript/tasks.md +++ b/openspec/changes/separate-wake-and-realtime-transcript/tasks.md @@ -118,3 +118,11 @@ - [x] 15.2 实现配置与 capture 端点；前置条件：15.1 完成；验收标准：`AppConfig`、`.env.example`、`--show-config` 支持新配置，`VoiceAssistantPipeline` 在已有 partial 后 1500 ms 无新文本时调用 recorder finish；测试要点：配置解析和 show-config 单测；优先级：P0；预计：45 分钟。 - [x] 15.3 补充 recorder finish 和回归测试；前置条件：15.2 完成；验收标准：持续有语音但 partial 不再推进时，segment end_reason 为 `partial_transcript_idle`，final STT 仍执行且 partial 不进上下文；测试要点：live runtime 单测；优先级：P0；预计：45 分钟。 - [x] 15.4 执行门禁并提交“实时字幕端点”模块；前置条件：15.1 至 15.3 完成；验收标准：compileall、unittest、simulate-live、security-check、OpenSpec strict、git diff check 通过后中文提交；优先级：P0；预计：45 分钟。 + +## 16. 自动持续对话判断与播报打断 + +- [x] 16.1 更新 OpenSpec 和 README 描述自动持续对话、免唤醒追问和播报打断；前置条件：用户确认不再固定 3 秒追问窗口；验收标准：proposal/design/spec/tasks/README 明确 `ContinuationDecisionStage`、hybrid 规则、follow-up 3 秒窗口、barge-in echo guard 和默认 standby 策略；测试要点：OpenSpec strict；优先级：P0；预计：45 分钟。 +- [x] 16.2 实现配置、事件和续聊决策 Provider；前置条件：16.1 完成；验收标准：`AppConfig`、`.env.example`、`--show-config` 支持持续对话配置，新增事件可被 reporter/test 消费，规则优先和 LLM 兜底 provider 可单测；测试要点：规则 continue/standby、不确定低置信 standby、高置信 continue；优先级：P0；预计：60 分钟。 +- [x] 16.3 重构 `VoiceAssistantPipeline/TurnController` 支持自动 follow-up；前置条件：16.2 完成；验收标准：回复后先判定 continue/standby，continue 时播放完进入 `followup_listening`，3 秒内用户直接回答无需 wake/ACK，超时恢复 standby；测试要点：连续对话、follow-up timeout、上下文只写 final STT；优先级：P0；预计：60 分钟。 +- [x] 16.4 实现可打断播放路径；前置条件：16.3 完成；验收标准：TTS 播放前 500 ms 忽略回声，之后检测到至少 250 ms 有效用户语音且 realtime partial 有效时触发 `barge_in_detected -> playback_interrupted`，上下文只写已播出的 assistant 文本；测试要点：guard window 不打断、有效输入打断、未播文本不进上下文；优先级：P0；预计：60 分钟。 +- [x] 16.5 执行门禁并提交“连续对话判断”模块；前置条件：16.1 至 16.4 完成；验收标准：compileall、unittest、simulate-live --turns 3、real-live-check --turns 2 --no-playback、security-check、model-check、device-check、OpenSpec strict、git diff check 通过后提交；优先级：P0；预计：60 分钟。 diff --git a/src/owner_voice_pet/assistant_pipeline.py b/src/owner_voice_pet/assistant_pipeline.py index a9b30aa..4036674 100644 --- a/src/owner_voice_pet/assistant_pipeline.py +++ b/src/owner_voice_pet/assistant_pipeline.py @@ -5,12 +5,20 @@ from typing import Protocol from .audio_preprocess import NoopAudioPreprocessor from .config import AppConfig +from .continuation import ContinuationDecision, ContinuationDecisionProvider, build_continuation_decider from .conversation import ConversationContext from .events import ( ACK_STARTED, + BARGE_IN_DETECTED, CAPTURE_STARTED, + CONTINUATION_DECISION_MADE, + CONTINUATION_DECISION_STARTED, + CONTINUOUS_SESSION_ENDED, + FOLLOWUP_LISTENING, + FOLLOWUP_TIMEOUT, LLM_STARTED, PLAYBACK_FINISHED, + PLAYBACK_INTERRUPTED, QUESTION_PROMPT, RECOVERING, SPEECH_ENDED, @@ -61,6 +69,8 @@ class TurnResult: assistant_text: str = "" error: ProviderError | None = None states: list[PipelineState] = field(default_factory=list) + completed_turns: int = 0 + failed_turns: int = 0 @dataclass(slots=True) @@ -71,6 +81,12 @@ class RuntimeSummary: last_error: ProviderError | None = None +@dataclass(slots=True) +class SpeakResult: + spoken_text: str + interrupted: bool = False + + class TurnController: def __init__( self, @@ -87,6 +103,7 @@ class TurnController: ack_tts: TtsProvider, context: ConversationContext, event_bus: PipelineEventBus, + continuation_decider: ContinuationDecisionProvider, sentence_buffer: SentenceBuffer | None = None, ) -> None: self.config = config @@ -101,8 +118,10 @@ class TurnController: self.ack_tts = ack_tts self.context = context self.event_bus = event_bus + self.continuation_decider = continuation_decider self.sentence_buffer = sentence_buffer or SentenceBuffer() self._states: list[PipelineState] = [] + self._pending_capture_frames: list[AudioFrame] = [] def run_turn(self, turn_id: int) -> TurnResult: self._states = [] @@ -129,7 +148,20 @@ class TurnController: if ack_error is not None: return ack_error self._event(QUESTION_PROMPT, PipelineState.SPEECH_DETECTING, "请说出问题", turn_id=turn_id) - user_segment = self._capture_segment(turn_id, state_message="录音中：正在听取问题") + return self._capture_and_transcribe(turn_id, state_message="录音中：正在听取问题") + + def _capture_and_transcribe( + self, + turn_id: int, + *, + state_message: str, + no_speech_timeout_ms: int | None = None, + ) -> str | ProviderError: + user_segment = self._capture_segment( + turn_id, + state_message=state_message, + no_speech_timeout_ms=no_speech_timeout_ms, + ) if isinstance(user_segment, ProviderError): return user_segment self._event(STT_STARTED, PipelineState.TRANSCRIBING, "转写中：正在识别问题", turn_id=turn_id) @@ -158,54 +190,73 @@ class TurnController: self.wakeword.reset() return None - def _capture_segment(self, turn_id: int, *, state_message: str) -> AudioSegment | ProviderError: - self.vad_recorder.reset() - self.vad_recorder.provider.reset() - self.audio_preprocessor.reset() - realtime_session = self._start_realtime_transcript() - last_partial_ms: int | None = None - self._event(CAPTURE_STARTED, PipelineState.RECORDING, state_message, turn_id=turn_id) - while True: - frames = self.transport.read_frames(timeout_ms=100) - if not frames: - continue - for frame in frames: - try: - frame = self.audio_preprocessor.process_frame(frame) - except ProviderError as exc: - return exc - was_started = self.vad_recorder.started - result = self.vad_recorder.feed(frame) - if not was_started and self.vad_recorder.started: - self._event(SPEECH_STARTED, PipelineState.RECORDING, "检测到用户语音", turn_id=turn_id) - if isinstance(result, ProviderError): - return result - if self.vad_recorder.started and realtime_session is not None: - if self._emit_realtime_transcript(realtime_session, frame, turn_id): - last_partial_ms = frame.timestamp_ms - if isinstance(result, AudioSegment): - if realtime_session is not None: - self._finish_realtime_transcript(realtime_session, turn_id) - self._event( - SPEECH_ENDED, - PipelineState.RECORDING, - "用户语音结束", - turn_id=turn_id, - payload={"end_reason": result.metadata.get("end_reason", "")}, - ) - return result - if self._should_end_after_realtime_idle(last_partial_ms, frame.timestamp_ms): - if realtime_session is not None: - self._finish_realtime_transcript(realtime_session, turn_id) - result = self.vad_recorder.finish("partial_transcript_idle") - self._event( - SPEECH_ENDED, - PipelineState.RECORDING, - "用户语音结束", - turn_id=turn_id, - payload={"end_reason": result.metadata.get("end_reason", "")}, - ) - return result + def _capture_segment( + self, + turn_id: int, + *, + state_message: str, + no_speech_timeout_ms: int | None = None, + ) -> AudioSegment | ProviderError: + original_no_speech_timeout_ms = self.vad_recorder.no_speech_timeout_ms + if no_speech_timeout_ms is not None: + self.vad_recorder.no_speech_timeout_ms = no_speech_timeout_ms + try: + self.vad_recorder.reset() + self.vad_recorder.provider.reset() + self.audio_preprocessor.reset() + realtime_session = self._start_realtime_transcript() + last_partial_ms: int | None = None + self._event(CAPTURE_STARTED, PipelineState.RECORDING, state_message, turn_id=turn_id) + while True: + frames = self._read_capture_frames(timeout_ms=100) + if not frames: + continue + for frame in frames: + try: + frame = self.audio_preprocessor.process_frame(frame) + except ProviderError as exc: + return exc + was_started = self.vad_recorder.started + result = self.vad_recorder.feed(frame) + if not was_started and self.vad_recorder.started: + self._event(SPEECH_STARTED, PipelineState.RECORDING, "检测到用户语音", turn_id=turn_id) + if isinstance(result, ProviderError): + return result + if self.vad_recorder.started and realtime_session is not None: + if self._emit_realtime_transcript(realtime_session, frame, turn_id): + last_partial_ms = frame.timestamp_ms + if isinstance(result, AudioSegment): + if realtime_session is not None: + self._finish_realtime_transcript(realtime_session, turn_id) + self._event( + SPEECH_ENDED, + PipelineState.RECORDING, + "用户语音结束", + turn_id=turn_id, + payload={"end_reason": result.metadata.get("end_reason", "")}, + ) + return result + if self._should_end_after_realtime_idle(last_partial_ms, frame.timestamp_ms): + if realtime_session is not None: + self._finish_realtime_transcript(realtime_session, turn_id) + result = self.vad_recorder.finish("partial_transcript_idle") + self._event( + SPEECH_ENDED, + PipelineState.RECORDING, + "用户语音结束", + turn_id=turn_id, + payload={"end_reason": result.metadata.get("end_reason", "")}, + ) + return result + finally: + self.vad_recorder.no_speech_timeout_ms = original_no_speech_timeout_ms + + def _read_capture_frames(self, *, timeout_ms: int) -> list[AudioFrame]: + if self._pending_capture_frames: + frames = list(self._pending_capture_frames) + self._pending_capture_frames.clear() + return frames + return self.transport.read_frames(timeout_ms=timeout_ms) def _start_realtime_transcript(self) -> RealtimeTranscriptSession | None: if not self.config.realtime_transcript_enabled or self.realtime_stt is None: @@ -259,19 +310,86 @@ class TurnController: return exc def _reply_to_user(self, user_text: str, turn_id: int) -> TurnResult: + completed_turns = 0 + current_user_text = user_text + current_turn_id = turn_id + last_assistant_text = "" + while True: + reply_result = self._reply_once(current_user_text, current_turn_id, completed_turns=completed_turns) + completed_turns += reply_result.completed_turns + if not reply_result.success: + return reply_result + last_assistant_text = reply_result.assistant_text + if reply_result.error is not None: + return reply_result + if reply_result.states and reply_result.states[-1] == PipelineState.INTERRUPTED: + followup = self._listen_for_followup(current_turn_id + 1, interrupted=True) + else: + decision = self._decide_continuation(current_user_text, last_assistant_text, current_turn_id) + if not decision.should_continue: + self._event( + CONTINUOUS_SESSION_ENDED, + PipelineState.WAKE_LISTENING, + "", + turn_id=current_turn_id, + payload={"decision": decision.action, "reason": decision.reason}, + ) + self._event( + STANDBY_RESUMED, + PipelineState.WAKE_LISTENING, + "恢复待机：可继续唤醒", + turn_id=current_turn_id, + ) + return TurnResult( + True, + current_user_text, + last_assistant_text, + states=list(self._states), + completed_turns=completed_turns, + ) + followup = self._listen_for_followup(current_turn_id + 1, interrupted=False) + if followup is None: + return TurnResult( + True, + current_user_text, + last_assistant_text, + states=list(self._states), + completed_turns=completed_turns, + ) + if isinstance(followup, ProviderError): + return self._recover(followup, current_turn_id + 1, completed_turns=completed_turns) + current_turn_id += 1 + current_user_text = followup + + def _reply_once(self, user_text: str, turn_id: int, *, completed_turns: int) -> TurnResult: self.context.append_user(user_text) self._event(LLM_STARTED, PipelineState.THINKING, "思考中：正在生成回复", turn_id=turn_id) assistant_text = "" + spoken_parts: list[str] = [] + interrupted = False try: for delta in self.llm.stream_reply(self.context.build_llm_messages()): assistant_text += delta.text_delta for sentence in self.sentence_buffer.feed(delta.text_delta, bool(delta.finish_reason)): - self._speak(sentence, turn_id) - for sentence in self.sentence_buffer.flush(): - self._speak(sentence, turn_id) + speak_result = self._speak(sentence, turn_id) + if speak_result.interrupted: + interrupted = True + break + spoken_parts.append(speak_result.spoken_text) + if interrupted: + break + if interrupted: + self.sentence_buffer.flush() + else: + for sentence in self.sentence_buffer.flush(): + speak_result = self._speak(sentence, turn_id) + if speak_result.interrupted: + interrupted = True + break + spoken_parts.append(speak_result.spoken_text) except ProviderError as exc: - return self._recover(exc, turn_id) - if not assistant_text.strip(): + return self._recover(exc, turn_id, completed_turns=completed_turns) + if not assistant_text.strip() and not "".join(spoken_parts).strip(): return self._recover( ProviderError( ErrorCode.LLM_EMPTY_REPLY, @@ -281,19 +399,153 @@ class TurnController: "llm", ), turn_id, + completed_turns=completed_turns, ) - self.context.append_assistant(assistant_text) - self._event(STANDBY_RESUMED, PipelineState.WAKE_LISTENING, "恢复待机：可继续唤醒", turn_id=turn_id) - return TurnResult(True, user_text, assistant_text, states=list(self._states)) + spoken_text = "".join(spoken_parts) + if spoken_text.strip(): + self.context.append_assistant(spoken_text) + states = list(self._states) + if interrupted: + states.append(PipelineState.INTERRUPTED) + return TurnResult( + True, + user_text, + spoken_text or assistant_text, + states=states, + completed_turns=1, + ) - def _speak(self, sentence: str, turn_id: int) -> None: + def _decide_continuation(self, user_text: str, assistant_text: str, turn_id: int) -> ContinuationDecision: + if not self.config.continuous_dialog_enabled: + return ContinuationDecision("standby", 1.0, "continuous dialog disabled", "config") + self._event(CONTINUATION_DECISION_STARTED, PipelineState.THINKING, "", turn_id=turn_id) + decision = self.continuation_decider.decide( + user_text=user_text, + assistant_text=assistant_text, + history=self.context.messages(), + ) + self._event( + CONTINUATION_DECISION_MADE, + PipelineState.THINKING, + "", + turn_id=turn_id, + payload={ + "action": decision.action, + "confidence": decision.confidence, + "reason": decision.reason, + "provider": decision.provider, + }, + ) + return decision + + def _listen_for_followup(self, turn_id: int, *, interrupted: bool) -> str | ProviderError | None: + if not interrupted: + seconds = max(1, round(self.config.followup_listen_timeout_ms / 1000)) + self._event( + FOLLOWUP_LISTENING, + PipelineState.RECORDING, + f"继续对话：{seconds}秒内可直接回答", + turn_id=turn_id, + ) + user_text = self._capture_and_transcribe( + turn_id, + state_message="录音中：正在听取追问" if not interrupted else "录音中：正在听取打断内容", + no_speech_timeout_ms=self.config.followup_listen_timeout_ms, + ) + if isinstance(user_text, ProviderError) and user_text.code == ErrorCode.VAD_TIMEOUT_NO_SPEECH: + self._event( + FOLLOWUP_TIMEOUT, + PipelineState.WAKE_LISTENING, + "追问超时：未检测到用户回答", + turn_id=turn_id, + ) + self._event(CONTINUOUS_SESSION_ENDED, PipelineState.WAKE_LISTENING, "", turn_id=turn_id) + self._event(STANDBY_RESUMED, PipelineState.WAKE_LISTENING, "恢复待机：可继续唤醒", turn_id=turn_id) + return None + return user_text + + def _speak(self, sentence: str, turn_id: int) -> SpeakResult: self._event(TTS_STARTED, PipelineState.SPEAKING, "播放中：正在播报回复", turn_id=turn_id) segment = self.tts.synthesize(sentence) - playback = self.transport.play_pcm(segment) - if playback.error: - raise playback.error + if not self._can_interrupt_playback(segment): + playback = self.transport.play_pcm(segment) + if playback.error: + raise playback.error + self._event(PLAYBACK_FINISHED, PipelineState.SPEAKING, "", turn_id=turn_id) + self._drain_input_after_playback() + return SpeakResult(sentence) + if self._play_interruptible(segment, turn_id=turn_id): + return SpeakResult("", interrupted=True) self._event(PLAYBACK_FINISHED, PipelineState.SPEAKING, "", turn_id=turn_id) self._drain_input_after_playback() + return SpeakResult(sentence) + + def _can_interrupt_playback(self, segment: AudioSegment) -> bool: + return ( + self.config.barge_in_enabled + and self.realtime_stt is not None + and segment.duration_ms > self.config.barge_in_echo_guard_ms + and not segment.metadata.get("format") + ) + + def _play_interruptible(self, segment: AudioSegment, *, turn_id: int) -> bool: + elapsed_ms = 0 + guard_cleared = False + self.vad_recorder.provider.reset() + realtime_session = self._start_realtime_transcript() + speech_ms = 0 + partial_seen = False + pending_frames: list[AudioFrame] = [] + for chunk in _audio_chunks(segment, chunk_ms=100): + playback = self.transport.play_pcm(chunk) + if playback.error: + raise playback.error + elapsed_ms += chunk.duration_ms + if elapsed_ms < self.config.barge_in_echo_guard_ms: + continue + if not guard_cleared: + self.transport.flush_input() + guard_cleared = True + continue + detected, speech_ms, partial_seen, new_frames = self._detect_barge_in( + realtime_session, + turn_id=turn_id, + speech_ms=speech_ms, + partial_seen=partial_seen, + ) + pending_frames.extend(new_frames) + if detected: + self._pending_capture_frames.extend(pending_frames) + self._event(BARGE_IN_DETECTED, PipelineState.INTERRUPTED, "检测到用户打断", turn_id=turn_id) + self._event(PLAYBACK_INTERRUPTED, PipelineState.INTERRUPTED, "播报已打断", turn_id=turn_id) + if realtime_session is not None: + realtime_session.finish() + return True + if realtime_session is not None: + realtime_session.finish() + return False + + def _detect_barge_in( + self, + realtime_session: RealtimeTranscriptSession | None, + *, + turn_id: int, + speech_ms: int, + partial_seen: bool, + ) -> tuple[bool, int, bool, list[AudioFrame]]: + frames = self.transport.read_frames(timeout_ms=0) + if not frames: + return False, speech_ms, partial_seen, [] + for frame in frames: + result = self.vad_recorder.provider.analyze(frame) + if result.is_speech: + speech_ms += int(frame.metadata.get("duration_ms", 20)) + if realtime_session is not None and self._emit_realtime_transcript(realtime_session, frame, turn_id): + partial_seen = True + else: + speech_ms = 0 + detected = speech_ms >= self.config.barge_in_min_speech_ms and partial_seen + return detected, speech_ms, partial_seen, frames def _drain_input_after_playback(self) -> None: self.transport.flush_input() @@ -306,11 +558,11 @@ class TurnController: remaining_ms -= timeout_ms self.transport.flush_input() - def _recover(self, error: ProviderError, turn_id: int) -> TurnResult: + def _recover(self, error: ProviderError, turn_id: int, *, completed_turns: int = 0) -> TurnResult: self._event(STAGE_ERROR, PipelineState.ERROR_RECOVERING, error.message, turn_id=turn_id, payload={"error": error}) self._event(RECOVERING, PipelineState.ERROR_RECOVERING, "恢复待机：本轮已结束", turn_id=turn_id) self._event(STANDBY_RESUMED, PipelineState.WAKE_LISTENING, "恢复待机：可继续唤醒", turn_id=turn_id) - return TurnResult(False, error=error, states=list(self._states)) + return TurnResult(False, error=error, states=list(self._states), completed_turns=completed_turns, failed_turns=1) def _event( self, @@ -342,6 +594,7 @@ class VoiceAssistantPipeline: ack_tts: TtsProvider | None = None, reporter: RuntimeReporter | None = None, event_bus: PipelineEventBus | None = None, + continuation_decider: ContinuationDecisionProvider | None = None, sentence_buffer: SentenceBuffer | None = None, ) -> None: self.config = config @@ -357,6 +610,11 @@ class VoiceAssistantPipeline: self.context = context self.reporter = reporter self.event_bus = event_bus or PipelineEventBus() + self.continuation_decider = continuation_decider or build_continuation_decider( + config.continuation_decision_provider, + llm, + threshold=config.continuation_confidence_threshold, + ) if reporter is not None: self.event_bus.subscribe(self._report_event) self.sentence_buffer = sentence_buffer or SentenceBuffer() @@ -373,6 +631,7 @@ class VoiceAssistantPipeline: ack_tts=self.ack_tts, context=context, event_bus=self.event_bus, + continuation_decider=self.continuation_decider, sentence_buffer=self.sentence_buffer, ) @@ -401,10 +660,11 @@ class VoiceAssistantPipeline: while True: turn_id = completed + failed + 1 result = self.run_turn(turn_id) + completed += result.completed_turns if result.success: - completed += 1 + completed += 0 if result.completed_turns else 1 else: - failed += 1 + failed += result.failed_turns or 1 last_error = result.error if once: break @@ -432,3 +692,30 @@ class VoiceAssistantPipeline: handler(event) return dispatch_pipeline_event(self.reporter, event) + + +def _audio_chunks(segment: AudioSegment, *, chunk_ms: int) -> list[AudioSegment]: + if chunk_ms <= 0 or segment.duration_ms <= chunk_ms: + return [segment] + bytes_per_ms = max(1, int(segment.sample_rate * segment.channels * 2 / 1000)) + chunk_bytes = max(2 * segment.channels, bytes_per_ms * chunk_ms) + chunk_bytes -= chunk_bytes % (2 * segment.channels) + chunks: list[AudioSegment] = [] + offset = 0 + start_ms = segment.start_time_ms + while offset < len(segment.pcm): + data = segment.pcm[offset : offset + chunk_bytes] + duration_ms = max(1, int(len(data) / bytes_per_ms)) + chunks.append( + AudioSegment( + data, + segment.sample_rate, + segment.channels, + start_ms, + start_ms + duration_ms, + dict(segment.metadata), + ) + ) + offset += len(data) + start_ms += duration_ms + return chunks diff --git a/src/owner_voice_pet/cli.py b/src/owner_voice_pet/cli.py index 77733a4..8fddf7f 100644 --- a/src/owner_voice_pet/cli.py +++ b/src/owner_voice_pet/cli.py @@ -4,6 +4,7 @@ import argparse import json import re import subprocess +from dataclasses import replace from pathlib import Path from .audio_preprocess import SherpaOnnxDenoiserPreprocessor @@ -96,6 +97,13 @@ def main(argv: list[str] | None = None) -> int: "tts_voice": config.tts_voice, "speech_models_dir": str(config.speech_models_dir), "context_mode": config.context_mode, + "continuous_dialog_enabled": config.continuous_dialog_enabled, + "continuation_decision_provider": config.continuation_decision_provider, + "continuation_confidence_threshold": config.continuation_confidence_threshold, + "followup_listen_timeout_ms": config.followup_listen_timeout_ms, + "barge_in_enabled": config.barge_in_enabled, + "barge_in_min_speech_ms": config.barge_in_min_speech_ms, + "barge_in_echo_guard_ms": config.barge_in_echo_guard_ms, }, ensure_ascii=False, sort_keys=True, @@ -192,52 +200,7 @@ def main(argv: list[str] | None = None) -> int: if args.command == "llm-smoke": config = AppConfig.from_dotenv(args.env_file) if args.no_stream: - config = AppConfig( - wake_word=config.wake_word, - sample_rate=config.sample_rate, - channels=config.channels, - llm_base_url=config.llm_base_url, - llm_api_key=config.llm_api_key, - llm_model=config.llm_model, - llm_api_style=config.llm_api_style, - llm_stream=False, - realtime_transcript_enabled=config.realtime_transcript_enabled, - realtime_transcript_idle_timeout_ms=config.realtime_transcript_idle_timeout_ms, - audio_input_device=config.audio_input_device, - audio_output_device=config.audio_output_device, - asset_dir=config.asset_dir, - log_dir=config.log_dir, - wake_provider=config.wake_provider, - wake_keywords_file=config.wake_keywords_file, - wake_kws_threshold=config.wake_kws_threshold, - wake_kws_score=config.wake_kws_score, - wake_ack_text=config.wake_ack_text, - post_playback_drain_ms=config.post_playback_drain_ms, - pipeline_mode=config.pipeline_mode, - endpoint_mode=config.endpoint_mode, - noise_filter_enabled=config.noise_filter_enabled, - noise_filter_provider=config.noise_filter_provider, - wake_denoise_enabled=config.wake_denoise_enabled, - speaker_profile_ms=config.speaker_profile_ms, - speaker_profile_min_ms=config.speaker_profile_min_ms, - speaker_absent_ms=config.speaker_absent_ms, - speaker_similarity_threshold=config.speaker_similarity_threshold, - speaker_min_rms=config.speaker_min_rms, - vad_provider=config.vad_provider, - vad_threshold=config.vad_threshold, - vad_min_duration_ms=config.vad_min_duration_ms, - vad_end_silence_ms=config.vad_end_silence_ms, - vad_no_speech_timeout_ms=config.vad_no_speech_timeout_ms, - vad_max_recording_ms=config.vad_max_recording_ms, - speech_provider=config.speech_provider, - asr_model=config.asr_model, - tts_model=config.tts_model, - tts_voice=config.tts_voice, - speech_models_dir=config.speech_models_dir, - context_mode=config.context_mode, - context_max_messages=config.context_max_messages, - context_max_chars=config.context_max_chars, - ) + config = replace(config, llm_stream=False) try: provider = OpenAICompatibleLlmProvider(config, timeout_s=30) messages = [ConversationContext().build_llm_messages()[0]] diff --git a/src/owner_voice_pet/config.py b/src/owner_voice_pet/config.py index 340f31c..7e02a6d 100644 --- a/src/owner_voice_pet/config.py +++ b/src/owner_voice_pet/config.py @@ -52,6 +52,13 @@ class AppConfig: context_mode: str = "session_memory" context_max_messages: int = 12 context_max_chars: int = 12000 + continuous_dialog_enabled: bool = True + continuation_decision_provider: str = "hybrid" + continuation_confidence_threshold: float = 0.65 + followup_listen_timeout_ms: int = 3000 + barge_in_enabled: bool = True + barge_in_min_speech_ms: int = 250 + barge_in_echo_guard_ms: int = 500 @classmethod def from_dotenv(cls, path: str | Path = ".env", prefix: str = "OWNER_") -> "AppConfig": @@ -111,6 +118,18 @@ class AppConfig: context_mode=(get("CONTEXT_MODE", "session_memory") or "session_memory").lower(), context_max_messages=int(get("CONTEXT_MAX_MESSAGES", "12") or "12"), context_max_chars=int(get("CONTEXT_MAX_CHARS", "12000") or "12000"), + continuous_dialog_enabled=(get("CONTINUOUS_DIALOG_ENABLED", "1") or "1").lower() + not in {"0", "false", "no"}, + continuation_decision_provider=( + get("CONTINUATION_DECISION_PROVIDER", "hybrid") or "hybrid" + ).lower(), + continuation_confidence_threshold=float( + get("CONTINUATION_CONFIDENCE_THRESHOLD", "0.65") or "0.65" + ), + followup_listen_timeout_ms=int(get("FOLLOWUP_LISTEN_TIMEOUT_MS", "3000") or "3000"), + barge_in_enabled=(get("BARGE_IN_ENABLED", "1") or "1").lower() not in {"0", "false", "no"}, + 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"), ) @classmethod @@ -340,6 +359,41 @@ class AppConfig: "startup", ) ) + if self.continuation_decision_provider not in {"hybrid", "rule", "llm"}: + errors.append( + ProviderError( + ErrorCode.CONFIG_MISSING_VALUE, + "OWNER_CONTINUATION_DECISION_PROVIDER must be hybrid, rule, or llm", + False, + "config", + "startup", + ) + ) + if not 0 < self.continuation_confidence_threshold <= 1: + errors.append( + ProviderError( + ErrorCode.CONFIG_MISSING_VALUE, + "OWNER_CONTINUATION_CONFIDENCE_THRESHOLD must be in (0, 1]", + False, + "config", + "startup", + ) + ) + for name, value in { + "OWNER_FOLLOWUP_LISTEN_TIMEOUT_MS": self.followup_listen_timeout_ms, + "OWNER_BARGE_IN_MIN_SPEECH_MS": self.barge_in_min_speech_ms, + "OWNER_BARGE_IN_ECHO_GUARD_MS": self.barge_in_echo_guard_ms, + }.items(): + if value < 0: + errors.append( + ProviderError( + ErrorCode.CONFIG_MISSING_VALUE, + f"{name} must be non-negative", + False, + "config", + "startup", + ) + ) return errors def api_url(self, path: str) -> str: diff --git a/src/owner_voice_pet/continuation.py b/src/owner_voice_pet/continuation.py new file mode 100644 index 0000000..3fa65f1 --- /dev/null +++ b/src/owner_voice_pet/continuation.py @@ -0,0 +1,205 @@ +from __future__ import annotations + +import json +import re +from collections.abc import Sequence +from dataclasses import dataclass +from typing import Protocol + +from .models import Message +from .protocols import LlmProvider + + +@dataclass(frozen=True, slots=True) +class ContinuationDecision: + action: str + confidence: float + reason: str + provider: str + + @property + def should_continue(self) -> bool: + return self.action == "continue" + + +class ContinuationDecisionProvider(Protocol): + def decide( + self, + *, + user_text: str, + assistant_text: str, + history: Sequence[Message], + ) -> ContinuationDecision: + ... + + +class RuleContinuationDecisionProvider: + _continue_patterns = ( + "你想", + "你要", + "你需要", + "需要我", + "要不要", + "是否需要", + "可以告诉我", + "告诉我", + "请告诉我", + "你希望", + "你更想", + "哪一个", + "哪一部分", + "哪种", + "哪个", + "请选择", + "选一个", + "请补充", + "需要补充", + "补充一下", + "继续吗", + ) + _standby_patterns = ( + "这是", + "已经", + "完成", + "好了", + "好的", + "没问题", + "不客气", + "无法", + "不能", + "抱歉", + "出错", + "失败", + ) + + def decide( + self, + *, + user_text: str, + assistant_text: str, + history: Sequence[Message], + ) -> ContinuationDecision: + text = _normalize(assistant_text) + if not text: + return ContinuationDecision("standby", 1.0, "empty assistant reply", "rule") + if _has_direct_question(text) or any(pattern in text for pattern in self._continue_patterns): + return ContinuationDecision("continue", 0.86, "assistant asks for user input", "rule") + if any(pattern in text for pattern in self._standby_patterns): + return ContinuationDecision("standby", 0.82, "assistant appears to finish the answer", "rule") + if len(text) <= 18 and not text.endswith(("?", "？", "吗", "呢")): + return ContinuationDecision("standby", 0.72, "short non-question reply", "rule") + return ContinuationDecision("unknown", 0.0, "rule uncertain", "rule") + + +class LlmContinuationDecisionProvider: + def __init__(self, llm: LlmProvider, *, threshold: float = 0.65) -> None: + self.llm = llm + self.threshold = threshold + + def decide( + self, + *, + user_text: str, + assistant_text: str, + history: Sequence[Message], + ) -> ContinuationDecision: + prompt = ( + "你是语音助手的持续对话分类器。判断助手刚才的回复是否需要用户继续直接回答。\n" + "只允许输出 JSON: {\"action\":\"continue|standby\",\"confidence\":0到1,\"reason\":\"短原因\"}。\n" + "如果不确定，action 必须是 standby。\n" + f"用户上一句: {user_text}\n" + f"助手回复: {assistant_text}\n" + ) + messages = [ + Message("system", "你只做二分类，不生成正常对话回复。", 0.0), + Message("user", prompt, 0.0), + ] + try: + raw = "".join(delta.text_delta for delta in self.llm.stream_reply(messages)).strip() + except Exception as exc: + return ContinuationDecision("standby", 0.0, f"classifier failed: {exc}", "llm") + parsed = _parse_classifier_output(raw) + if parsed is None: + return ContinuationDecision("standby", 0.0, "classifier output malformed", "llm") + action, confidence, reason = parsed + if action not in {"continue", "standby"}: + return ContinuationDecision("standby", 0.0, "classifier action invalid", "llm") + if confidence < self.threshold: + return ContinuationDecision("standby", confidence, "classifier confidence below threshold", "llm") + return ContinuationDecision(action, confidence, reason or "classifier decision", "llm") + + +class HybridContinuationDecisionProvider: + def __init__( + self, + rule_provider: RuleContinuationDecisionProvider, + llm_provider: LlmContinuationDecisionProvider | None, + *, + threshold: float = 0.65, + ) -> None: + self.rule_provider = rule_provider + self.llm_provider = llm_provider + self.threshold = threshold + + def decide( + self, + *, + user_text: str, + assistant_text: str, + history: Sequence[Message], + ) -> ContinuationDecision: + rule = self.rule_provider.decide(user_text=user_text, assistant_text=assistant_text, history=history) + if rule.action in {"continue", "standby"} and rule.confidence >= self.threshold: + return rule + if self.llm_provider is None: + return ContinuationDecision("standby", rule.confidence, "rule uncertain and no llm fallback", "hybrid") + llm = self.llm_provider.decide(user_text=user_text, assistant_text=assistant_text, history=history) + if llm.action == "continue" and llm.confidence >= self.threshold: + return ContinuationDecision("continue", llm.confidence, llm.reason, "hybrid") + return ContinuationDecision("standby", llm.confidence, llm.reason, "hybrid") + + +def build_continuation_decider( + provider_name: str, + llm: LlmProvider, + *, + threshold: float, +) -> ContinuationDecisionProvider: + rule = RuleContinuationDecisionProvider() + if provider_name == "rule": + return rule + llm_provider = LlmContinuationDecisionProvider(llm, threshold=threshold) + if provider_name == "llm": + return llm_provider + return HybridContinuationDecisionProvider(rule, llm_provider, threshold=threshold) + + +def _normalize(text: str) -> str: + return re.sub(r"\s+", "", text.strip()) + + +def _has_direct_question(text: str) -> bool: + if text.endswith(("?", "？")): + return True + return bool(re.search(r"(你|您).{0,12}(吗|呢|么|什么|哪|是否|要不要|需要不需要)", text)) + + +def _parse_classifier_output(raw: str) -> tuple[str, float, str] | None: + clean = raw.strip() + if not clean: + return None + try: + data = json.loads(clean) + action = str(data.get("action", "")).strip().lower() + confidence = float(data.get("confidence", 1.0 if action in {"continue", "standby"} else 0.0)) + reason = str(data.get("reason", "")).strip() + return action, max(0.0, min(1.0, confidence)), reason + except (ValueError, TypeError, json.JSONDecodeError): + lowered = clean.lower() + if lowered in {"continue", "standby"}: + return lowered, 1.0, "plain classifier output" + if "continue" in lowered and "standby" not in lowered: + return "continue", 1.0, "plain classifier output" + if "standby" in lowered and "continue" not in lowered: + return "standby", 1.0, "plain classifier output" + return None diff --git a/src/owner_voice_pet/events.py b/src/owner_voice_pet/events.py index 490c1e9..fe0b574 100644 --- a/src/owner_voice_pet/events.py +++ b/src/owner_voice_pet/events.py @@ -21,6 +21,13 @@ LLM_STARTED = "llm_started" TTS_STARTED = "tts_started" PLAYBACK_FINISHED = "playback_finished" STANDBY_RESUMED = "standby_resumed" +CONTINUATION_DECISION_STARTED = "continuation_decision_started" +CONTINUATION_DECISION_MADE = "continuation_decision_made" +FOLLOWUP_LISTENING = "followup_listening" +FOLLOWUP_TIMEOUT = "followup_timeout" +BARGE_IN_DETECTED = "barge_in_detected" +PLAYBACK_INTERRUPTED = "playback_interrupted" +CONTINUOUS_SESSION_ENDED = "continuous_session_ended" STAGE_ERROR = "stage_error" RECOVERING = "recovering" diff --git a/src/owner_voice_pet/real_live_check.py b/src/owner_voice_pet/real_live_check.py index 868dc4e..99da425 100644 --- a/src/owner_voice_pet/real_live_check.py +++ b/src/owner_voice_pet/real_live_check.py @@ -229,6 +229,8 @@ def run_real_live_check( speech_provider="local", noise_filter_enabled=True, realtime_transcript_enabled=True, + continuous_dialog_enabled=False, + barge_in_enabled=False, llm_stream=False, post_playback_drain_ms=0, ) diff --git a/src/owner_voice_pet/tts.py b/src/owner_voice_pet/tts.py index f97065b..432ff4d 100644 --- a/src/owner_voice_pet/tts.py +++ b/src/owner_voice_pet/tts.py @@ -9,6 +9,7 @@ import subprocess import tempfile import urllib.error import urllib.request +import wave from collections.abc import Callable from pathlib import Path from typing import Any @@ -104,8 +105,9 @@ class MacSayTtsProvider: "tts", ) with tempfile.TemporaryDirectory() as tmp: - output = Path(tmp) / "speech.aiff" - command = ["say", "-o", str(output)] + aiff_output = Path(tmp) / "speech.aiff" + wav_output = Path(tmp) / "speech.wav" + command = ["say", "-o", str(aiff_output)] if self.voice: command.extend(["-v", self.voice]) command.append(clean) @@ -119,7 +121,23 @@ class MacSayTtsProvider: "macos-say", "tts", ) from exc - data = output.read_bytes() + try: + subprocess.run( + ["afconvert", "-f", "WAVE", "-d", "LEI16@16000", "-c", "1", str(aiff_output), str(wav_output)], + check=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + with wave.open(str(wav_output), "rb") as handle: + sample_rate = handle.getframerate() + channels = handle.getnchannels() + data = handle.readframes(handle.getnframes()) + metadata = {"text": clean} + except (FileNotFoundError, subprocess.CalledProcessError, wave.Error): + data = aiff_output.read_bytes() + sample_rate = 16000 + channels = 1 + metadata = {"text": clean, "format": "aiff"} if not data: raise ProviderError( ErrorCode.TTS_EMPTY_AUDIO, @@ -128,7 +146,8 @@ class MacSayTtsProvider: "macos-say", "tts", ) - return AudioSegment(data, 16000, 1, 0, max(120, len(clean) * 45), {"text": clean, "format": "aiff"}) + duration_ms = int(len(data) / max(1, sample_rate * channels * 2) * 1000) + return AudioSegment(data, sample_rate, channels, 0, max(120, duration_ms, len(clean) * 45), metadata) class CloudTtsProvider: diff --git a/tests/test_continuation.py b/tests/test_continuation.py new file mode 100644 index 0000000..fb0c88e --- /dev/null +++ b/tests/test_continuation.py @@ -0,0 +1,81 @@ +from __future__ import annotations + +import unittest + +from owner_voice_pet.continuation import ( + HybridContinuationDecisionProvider, + LlmContinuationDecisionProvider, + RuleContinuationDecisionProvider, +) +from owner_voice_pet.models import Message, ReplyDelta + + +class FakeClassifierLlm: + def __init__(self, text: str) -> None: + self.text = text + self.calls: list[list[Message]] = [] + + def stream_reply(self, messages): + self.calls.append(list(messages)) + yield ReplyDelta(self.text, finish_reason="stop") + + +class ContinuationDecisionTests(unittest.TestCase): + def test_rule_continue_for_assistant_question(self) -> None: + decision = RuleContinuationDecisionProvider().decide( + user_text="讲讲天气", + assistant_text="你想继续听哪一部分？", + history=[], + ) + + self.assertEqual(decision.action, "continue") + self.assertGreaterEqual(decision.confidence, 0.65) + + def test_rule_standby_for_completed_answer(self) -> None: + decision = RuleContinuationDecisionProvider().decide( + user_text="今天天气怎么样", + assistant_text="这是今天的天气。", + history=[], + ) + + self.assertEqual(decision.action, "standby") + self.assertGreaterEqual(decision.confidence, 0.65) + + def test_hybrid_low_confidence_llm_defaults_to_standby(self) -> None: + llm = FakeClassifierLlm('{"action":"continue","confidence":0.3,"reason":"弱"}') + provider = HybridContinuationDecisionProvider( + RuleContinuationDecisionProvider(), + LlmContinuationDecisionProvider(llm, threshold=0.65), + threshold=0.65, + ) + + decision = provider.decide( + user_text="继续", + assistant_text="我还可以从背景原因和下一步影响两个方向继续展开", + history=[], + ) + + self.assertEqual(decision.action, "standby") + self.assertEqual(len(llm.calls), 1) + + def test_hybrid_high_confidence_llm_can_continue(self) -> None: + llm = FakeClassifierLlm('{"action":"continue","confidence":0.91,"reason":"等待用户选择"}') + provider = HybridContinuationDecisionProvider( + RuleContinuationDecisionProvider(), + LlmContinuationDecisionProvider(llm, threshold=0.65), + threshold=0.65, + ) + + decision = provider.decide( + user_text="继续", + assistant_text="我还可以从背景原因和下一步影响两个方向继续展开", + history=[], + ) + + self.assertEqual(decision.action, "continue") + self.assertEqual(decision.provider, "hybrid") + self.assertEqual(len(llm.calls), 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_live_runtime.py b/tests/test_live_runtime.py index 9cc2d96..0adda7f 100644 --- a/tests/test_live_runtime.py +++ b/tests/test_live_runtime.py @@ -6,9 +6,14 @@ from owner_voice_pet.config import AppConfig from owner_voice_pet.conversation import ConversationContext from owner_voice_pet.events import ( ACK_STARTED, + BARGE_IN_DETECTED, CAPTURE_STARTED, + CONTINUATION_DECISION_MADE, + FOLLOWUP_LISTENING, + FOLLOWUP_TIMEOUT, LLM_STARTED, PLAYBACK_FINISHED, + PLAYBACK_INTERRUPTED, SPEECH_ENDED, SPEECH_STARTED, STANDBY_RESUMED, @@ -21,7 +26,7 @@ from owner_voice_pet.events import ( PipelineEventBus, ) from owner_voice_pet.llm import MockLlmProvider -from owner_voice_pet.models import AudioFrame, AudioSegment, Transcript +from owner_voice_pet.models import AudioFrame, AudioSegment, PlaybackResult, ReplyDelta, Transcript, TransportHealth from owner_voice_pet.assistant_pipeline import VoiceAssistantPipeline from owner_voice_pet.runtime import build_live_runtime from owner_voice_pet.stt import MetadataSttProvider, SherpaOnnxSttProvider @@ -95,6 +100,19 @@ class QueueSttProvider: return Transcript(text, "zh", 1.0, segment.duration_ms, "queue-stt") +class QueueLlmProvider: + def __init__(self, replies: list[list[str]]) -> None: + self.replies = [list(item) for item in replies] + self.calls = [] + + def stream_reply(self, messages): + self.calls.append(list(messages)) + chunks = self.replies.pop(0) + for chunk in chunks: + yield ReplyDelta(chunk) + yield ReplyDelta("", finish_reason="stop") + + class RecordingReporter: def __init__(self) -> None: self.statuses: list[str] = [] @@ -188,6 +206,45 @@ def make_runtime( return runtime, stt, llm, transport, reporter +class PlaybackInjectedTransport(MemoryAudioTransport): + def __init__( + self, + frames: list[AudioFrame], + *, + inject_after_play_count: int, + injected_frames: list[AudioFrame], + ) -> None: + super().__init__(frames, flush_clears_input=False) + self.inject_after_play_count = inject_after_play_count + self.injected_frames = list(injected_frames) + self.play_count = 0 + + def play_pcm(self, segment: AudioSegment, interrupt: bool = False) -> PlaybackResult: + result = super().play_pcm(segment, interrupt=interrupt) + self.play_count += 1 + if self.play_count == self.inject_after_play_count: + for frame in self.injected_frames: + self.inject(frame) + return result + + def health(self) -> TransportHealth: + return TransportHealth(True, True, "playback injected") + + +def silence_frames(start_id: int, start_ms: int, count: int) -> list[AudioFrame]: + return [ + AudioFrame( + b"\x00\x00", + 16000, + 1, + start_ms + index * 20, + start_id + index, + {"duration_ms": 20, "speech": False}, + ) + for index in range(count) + ] + + class LiveRuntimeTests(unittest.TestCase): def test_repeated_runtime_runs_two_turns_and_returns_to_standby(self) -> None: runtime, stt, llm, transport, reporter = make_runtime(["第一问", "第二问"]) @@ -359,6 +416,140 @@ class LiveRuntimeTests(unittest.TestCase): self.assertIsInstance(runtime.stt, SherpaOnnxSttProvider) self.assertIsNotNone(runtime.realtime_stt) + def test_assistant_followup_question_enters_listening_without_second_wake(self) -> None: + frames = [wake_frame(0, 0)] + frames.extend(segment_frames(1, 20, partials=["第一问", "第一问"])) + frames.extend(segment_frames(10, 400, partials=["继续内容", "继续内容"])) + transport = MemoryAudioTransport(frames, flush_clears_input=False) + stt = QueueSttProvider(["第一问", "继续内容"]) + llm = QueueLlmProvider([["你想继续听哪一部分？"], ["这是补充回答。"]]) + reporter = RecordingReporter() + runtime = VoiceAssistantPipeline( + config=AppConfig( + llm_api_key="secret", + speech_provider="cloud", + wake_ack_text="", + followup_listen_timeout_ms=3000, + ), + transport=transport, + wakeword=KeywordWakeWordProvider(), + vad_recorder=VadRecorder(EnergyVadProvider(), min_duration_ms=40, end_silence_ms=40), + stt=stt, + realtime_stt=MetadataSttProvider(), + llm=llm, + tts=SineTtsProvider(), + context=ConversationContext(), + reporter=reporter, + event_bus=PipelineEventBus(), + ) + + summary = runtime.run(max_turns=1) + event_types = [event.type for event in runtime.event_bus.events] + + self.assertEqual(summary.completed_turns, 2) + self.assertEqual(event_types.count(WAKE_DETECTED), 1) + self.assertEqual(event_types.count(FOLLOWUP_LISTENING), 1) + self.assertEqual(event_types.count(STT_STARTED), 2) + self.assertEqual(len(llm.calls), 2) + self.assertEqual(reporter.transcripts, ["第一问", "继续内容"]) + second_call_text = [message.content for message in llm.calls[1]] + self.assertIn("第一问", second_call_text) + self.assertIn("你想继续听哪一部分？", second_call_text) + + def test_followup_timeout_returns_to_standby(self) -> None: + frames = [wake_frame(0, 0)] + frames.extend(segment_frames(1, 20, partials=["第一问", "第一问"])) + frames.extend(silence_frames(20, 400, 170)) + transport = MemoryAudioTransport(frames, flush_clears_input=False) + stt = QueueSttProvider(["第一问"]) + llm = QueueLlmProvider([["你想继续听哪一部分？"]]) + reporter = RecordingReporter() + runtime = VoiceAssistantPipeline( + config=AppConfig( + llm_api_key="secret", + speech_provider="cloud", + wake_ack_text="", + followup_listen_timeout_ms=3000, + ), + transport=transport, + wakeword=KeywordWakeWordProvider(), + vad_recorder=VadRecorder(EnergyVadProvider(), min_duration_ms=40, end_silence_ms=40), + stt=stt, + realtime_stt=MetadataSttProvider(), + llm=llm, + tts=SineTtsProvider(), + context=ConversationContext(), + reporter=reporter, + event_bus=PipelineEventBus(), + ) + + summary = runtime.run(max_turns=1) + event_types = [event.type for event in runtime.event_bus.events] + + self.assertEqual(summary.completed_turns, 1) + self.assertEqual(summary.failed_turns, 0) + self.assertIn(FOLLOWUP_TIMEOUT, event_types) + self.assertEqual(event_types[-1], STANDBY_RESUMED) + self.assertEqual(len(llm.calls), 1) + + def test_barge_in_interrupts_playback_and_keeps_only_spoken_context(self) -> None: + first_question_frames = [wake_frame(0, 0)] + first_question_frames.extend(segment_frames(1, 20, partials=["第一问", "第一问"])) + barge_frames = [ + AudioFrame(b"\xff\x7f", 16000, 1, 600, 20, {"duration_ms": 20, "speech": True, "partial_transcript": "等一下"}), + AudioFrame(b"\xff\x7f", 16000, 1, 620, 21, {"duration_ms": 20, "speech": True, "partial_transcript": "等一下"}), + *silence_frames(22, 640, 3), + ] + transport = PlaybackInjectedTransport( + first_question_frames, + inject_after_play_count=8, + injected_frames=barge_frames, + ) + stt = QueueSttProvider(["第一问", "打断问题"]) + llm = QueueLlmProvider( + [ + [ + "已播出一句。", + "这是一段需要被打断的很长很长很长很长很长很长很长很长的回复内容，没有播放完。", + ], + ["这是新回答。"], + ] + ) + reporter = RecordingReporter() + runtime = VoiceAssistantPipeline( + config=AppConfig( + llm_api_key="secret", + speech_provider="cloud", + wake_ack_text="", + barge_in_enabled=True, + barge_in_echo_guard_ms=500, + barge_in_min_speech_ms=40, + followup_listen_timeout_ms=3000, + ), + transport=transport, + wakeword=KeywordWakeWordProvider(), + vad_recorder=VadRecorder(EnergyVadProvider(), min_duration_ms=40, end_silence_ms=40), + stt=stt, + realtime_stt=MetadataSttProvider(), + llm=llm, + tts=SineTtsProvider(), + context=ConversationContext(), + reporter=reporter, + event_bus=PipelineEventBus(), + ) + + summary = runtime.run(max_turns=2) + event_types = [event.type for event in runtime.event_bus.events] + context_texts = [message.content for message in runtime.context.messages()] + + self.assertEqual(summary.completed_turns, 2) + self.assertIn(BARGE_IN_DETECTED, event_types) + self.assertIn(PLAYBACK_INTERRUPTED, event_types) + self.assertIn("已播出一句。", context_texts) + self.assertNotIn("这是一段需要被打断的很长很长很长很长很长很长很长很长的回复内容，没有播放完。", context_texts) + self.assertEqual(reporter.transcripts[-1], "打断问题") + self.assertEqual(len(llm.calls), 2) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_models_config.py b/tests/test_models_config.py index 000b3b8..a964714 100644 --- a/tests/test_models_config.py +++ b/tests/test_models_config.py @@ -106,6 +106,13 @@ class ModelsConfigTests(unittest.TestCase): self.assertEqual(config.tts_voice, "mimo_default") self.assertEqual(str(config.speech_models_dir), "models") self.assertEqual(config.context_mode, "session_memory") + self.assertTrue(config.continuous_dialog_enabled) + self.assertEqual(config.continuation_decision_provider, "hybrid") + self.assertEqual(config.continuation_confidence_threshold, 0.65) + self.assertEqual(config.followup_listen_timeout_ms, 3000) + self.assertTrue(config.barge_in_enabled) + self.assertEqual(config.barge_in_min_speech_ms, 250) + self.assertEqual(config.barge_in_echo_guard_ms, 500) self.assertTrue(config.llm_stream) self.assertEqual(config.validate_basic(), []) @@ -139,6 +146,22 @@ class ModelsConfigTests(unittest.TestCase): errors = config.validate_basic() self.assertTrue(any("OWNER_NOISE_FILTER_PROVIDER" in error.message for error in errors)) + def test_continuation_config_is_validated(self) -> None: + config = AppConfig( + continuation_decision_provider="invalid", + continuation_confidence_threshold=2.0, + followup_listen_timeout_ms=-1, + barge_in_min_speech_ms=-1, + barge_in_echo_guard_ms=-1, + ) + errors = config.validate_basic() + + self.assertTrue(any("OWNER_CONTINUATION_DECISION_PROVIDER" in error.message for error in errors)) + self.assertTrue(any("OWNER_CONTINUATION_CONFIDENCE_THRESHOLD" in error.message for error in errors)) + self.assertTrue(any("OWNER_FOLLOWUP_LISTEN_TIMEOUT_MS" in error.message for error in errors)) + self.assertTrue(any("OWNER_BARGE_IN_MIN_SPEECH_MS" in error.message for error in errors)) + self.assertTrue(any("OWNER_BARGE_IN_ECHO_GUARD_MS" in error.message for error in errors)) + def test_speaker_similarity_threshold_range_is_validated(self) -> None: config = AppConfig(speaker_similarity_threshold=1.5) errors = config.validate_basic()