diff --git a/openspec/changes/separate-wake-and-realtime-transcript/tasks.md b/openspec/changes/separate-wake-and-realtime-transcript/tasks.md index 197840c..794da48 100644 --- a/openspec/changes/separate-wake-and-realtime-transcript/tasks.md +++ b/openspec/changes/separate-wake-and-realtime-transcript/tasks.md @@ -49,7 +49,7 @@ ## 7. 开源语音助手式 Pipeline 重构 - [x] 7.1 更新 OpenSpec 以描述 stage 化 pipeline、事件总线、TurnController 和主说话人端点；前置条件：公开参考已确认；验收标准：proposal/design/spec/tasks 覆盖新架构和任务；测试要点：OpenSpec strict；优先级：P0；预计：45 分钟。 -- [ ] 7.2 实现 pipeline event bus 和终端事件映射；前置条件：7.1 完成；验收标准：所有 live 用户可见状态由事件产生；测试要点：事件顺序和终端文案测试；优先级：P0；预计：60 分钟。 +- [x] 7.2 实现 pipeline event bus 和终端事件映射；前置条件：7.1 完成；验收标准：所有 live 用户可见状态由事件产生；测试要点：事件顺序和终端文案测试；优先级：P0；预计：60 分钟。 - [ ] 7.3 实现 `TurnController` 和 `VoiceAssistantPipeline`；前置条件：7.2 完成；验收标准：`run-live` 使用统一 pipeline，成功/失败 turn 均恢复待机；测试要点：两轮 fake runtime、错误恢复、上下文回归；优先级：P0；预计：60 分钟。 - [ ] 7.4 实现本轮主说话人端点；前置条件：7.3 完成；验收标准：主说话人音色消失约 300 ms 后结束采集；测试要点：一次提问后背景噪声不拖尾、短暂停顿不断句、画像不足回退；优先级：P0；预计：60 分钟。 - [ ] 7.5 更新 README、`.env.example`、本地 `.env` 非密钥配置；前置条件：7.2 至 7.4 完成；验收标准：运行说明匹配新 pipeline；测试要点：`--show-config` 不泄露 key；优先级：P0；预计：30 分钟。 diff --git a/src/owner_voice_pet/__init__.py b/src/owner_voice_pet/__init__.py index 75ae765..b313647 100644 --- a/src/owner_voice_pet/__init__.py +++ b/src/owner_voice_pet/__init__.py @@ -1,6 +1,7 @@ """Owner voice pet pipeline package.""" from .config import AppConfig +from .events import PipelineEvent, PipelineEventBus from .models import ( AudioFrame, AudioSegment, @@ -29,6 +30,8 @@ from .ui import ConsolePetWindow, PetStateController, PetVisualState __all__ = [ "AppConfig", + "PipelineEvent", + "PipelineEventBus", "AudioFrame", "AudioSegment", "AudioRingBuffer", diff --git a/src/owner_voice_pet/events.py b/src/owner_voice_pet/events.py new file mode 100644 index 0000000..d20cd42 --- /dev/null +++ b/src/owner_voice_pet/events.py @@ -0,0 +1,77 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Callable + +from .models import PipelineState, ProviderError + + +PIPELINE_STARTED = "pipeline_started" +WAKE_LISTENING = "wake_listening" +WAKE_DETECTED = "wake_detected" +ACK_STARTED = "ack_started" +QUESTION_PROMPT = "question_prompt" +CAPTURE_STARTED = "capture_started" +SPEECH_STARTED = "speech_started" +SPEECH_ENDED = "speech_ended" +STT_STARTED = "stt_started" +TRANSCRIPT_FINAL = "transcript_final" +LLM_STARTED = "llm_started" +TTS_STARTED = "tts_started" +PLAYBACK_FINISHED = "playback_finished" +STANDBY_RESUMED = "standby_resumed" +STAGE_ERROR = "stage_error" +RECOVERING = "recovering" + + +@dataclass(frozen=True, slots=True) +class PipelineEvent: + type: str + turn_id: int | None = None + state: PipelineState | None = None + message: str = "" + payload: dict[str, Any] = field(default_factory=dict) + + +class PipelineEventBus: + def __init__(self) -> None: + self.events: list[PipelineEvent] = [] + self._listeners: list[Callable[[PipelineEvent], None]] = [] + + def subscribe(self, listener: Callable[[PipelineEvent], None]) -> None: + self._listeners.append(listener) + + def emit( + self, + event_type: str, + *, + turn_id: int | None = None, + state: PipelineState | None = None, + message: str = "", + payload: dict[str, Any] | None = None, + ) -> PipelineEvent: + event = PipelineEvent(event_type, turn_id=turn_id, state=state, message=message, payload=payload or {}) + self.events.append(event) + for listener in list(self._listeners): + listener(event) + return event + + +def dispatch_pipeline_event(reporter: Any, event: PipelineEvent) -> None: + if event.type == TRANSCRIPT_FINAL: + reporter.transcript(str(event.payload.get("text", event.message)), final=True, turn_id=event.turn_id) + return + if event.type == STAGE_ERROR: + error = event.payload.get("error") + if isinstance(error, ProviderError): + reporter.error(error.stage, error.code.value, error.message, turn_id=event.turn_id) + return + reporter.error( + str(event.payload.get("stage", "pipeline")), + str(event.payload.get("code", "STAGE_ERROR")), + event.message, + turn_id=event.turn_id, + ) + return + if event.message: + reporter.status((event.state.value if event.state else event.type), event.message, turn_id=event.turn_id) diff --git a/src/owner_voice_pet/runtime.py b/src/owner_voice_pet/runtime.py index 009e431..6054b80 100644 --- a/src/owner_voice_pet/runtime.py +++ b/src/owner_voice_pet/runtime.py @@ -6,6 +6,26 @@ from typing import Protocol from .config import AppConfig from .conversation import ConversationContext +from .events import ( + ACK_STARTED, + CAPTURE_STARTED, + LLM_STARTED, + PipelineEvent, + PipelineEventBus, + PLAYBACK_FINISHED, + QUESTION_PROMPT, + RECOVERING, + SPEECH_ENDED, + SPEECH_STARTED, + STAGE_ERROR, + STANDBY_RESUMED, + STT_STARTED, + TRANSCRIPT_FINAL, + TTS_STARTED, + WAKE_DETECTED, + WAKE_LISTENING, + dispatch_pipeline_event, +) from .llm import OpenAICompatibleLlmProvider from .models import AudioSegment, ErrorCode, PipelineState, ProviderError from .protocols import AudioTransport, LlmProvider, SttProvider, TtsProvider, WakeWordProvider @@ -28,6 +48,9 @@ class RuntimeReporter(Protocol): class TerminalRuntimeReporter: + def handle_event(self, event: PipelineEvent) -> None: + dispatch_pipeline_event(self, event) + def status(self, state: str, message: str, *, turn_id: int | None = None) -> None: prefix = f"[第{turn_id}轮] " if turn_id is not None else "" print(f"{prefix}{message}", flush=True) @@ -73,6 +96,7 @@ class LiveVoiceRuntime: context: ConversationContext, ack_tts: TtsProvider | None = None, reporter: RuntimeReporter | None = None, + event_bus: PipelineEventBus | None = None, sentence_buffer: SentenceBuffer | None = None, ) -> None: self.config = config @@ -85,6 +109,8 @@ class LiveVoiceRuntime: self.ack_tts = ack_tts or tts self.context = context self.reporter = reporter or TerminalRuntimeReporter() + self.event_bus = event_bus or PipelineEventBus() + self.event_bus.subscribe(self._report_event) self.sentence_buffer = sentence_buffer or SentenceBuffer() self._states: list[PipelineState] = [] @@ -130,7 +156,12 @@ class LiveVoiceRuntime: def run_turn(self, turn_id: int) -> TurnResult: self._states = [] try: - self._state(PipelineState.WAKE_LISTENING, "待机：等待唤醒词“小杰小杰”", turn_id=turn_id) + self._event( + WAKE_LISTENING, + PipelineState.WAKE_LISTENING, + "待机：等待唤醒词“小杰小杰”", + turn_id=turn_id, + ) user_text = self._wait_for_wake_and_user_text(turn_id) if isinstance(user_text, ProviderError): return self._recover(user_text, turn_id) @@ -145,15 +176,15 @@ class LiveVoiceRuntime: wake_error = self._wait_for_local_wake(turn_id) if wake_error is not None: return wake_error - self._state(PipelineState.SPEECH_DETECTING, "唤醒命中", turn_id=turn_id) + self._event(WAKE_DETECTED, PipelineState.SPEECH_DETECTING, "唤醒命中", turn_id=turn_id) ack_error = self._acknowledge_wake(turn_id) if ack_error is not None: return ack_error - self._state(PipelineState.SPEECH_DETECTING, "请说出问题", turn_id=turn_id) + self._event(QUESTION_PROMPT, PipelineState.SPEECH_DETECTING, "请说出问题", turn_id=turn_id) user_segment = self._capture_segment(turn_id, state_message="录音中：正在听取问题") if isinstance(user_segment, ProviderError): return user_segment - self._state(PipelineState.TRANSCRIBING, "转写中：正在识别问题", turn_id=turn_id) + self._event(STT_STARTED, PipelineState.TRANSCRIBING, "转写中：正在识别问题", turn_id=turn_id) transcript = self.stt.transcribe(user_segment) user_text = transcript.normalized_text if not is_valid_transcript_text(user_text): @@ -164,7 +195,7 @@ class LiveVoiceRuntime: "live-runtime", "stt", ) - self.reporter.transcript(user_text, final=True, turn_id=turn_id) + self._event(TRANSCRIPT_FINAL, PipelineState.TRANSCRIBING, "", turn_id=turn_id, payload={"text": user_text}) return user_text def _wait_for_local_wake(self, turn_id: int) -> ProviderError | None: @@ -182,16 +213,26 @@ class LiveVoiceRuntime: def _capture_segment(self, turn_id: int, *, state_message: str) -> AudioSegment | ProviderError: self.vad_recorder.reset() self.vad_recorder.provider.reset() - self._state(PipelineState.RECORDING, state_message, turn_id=turn_id) + 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: + 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 isinstance(result, AudioSegment): + self._event( + SPEECH_ENDED, + PipelineState.RECORDING, + "用户语音结束", + turn_id=turn_id, + payload={"end_reason": result.metadata.get("end_reason", "")}, + ) return result def _acknowledge_wake(self, turn_id: int) -> ProviderError | None: @@ -200,7 +241,7 @@ class LiveVoiceRuntime: self._drain_input_after_playback() return None try: - self._state(PipelineState.SPEAKING, f"应答中：{text}", turn_id=turn_id) + self._event(ACK_STARTED, PipelineState.SPEAKING, f"应答中：{text}", turn_id=turn_id) segment = self.ack_tts.synthesize(text) playback = self.transport.play_pcm(segment) if playback.error: @@ -223,7 +264,7 @@ class LiveVoiceRuntime: def _reply_to_user(self, user_text: str, turn_id: int) -> TurnResult: self.context.append_user(user_text) - self._state(PipelineState.THINKING, "思考中：正在生成回复", turn_id=turn_id) + self._event(LLM_STARTED, PipelineState.THINKING, "思考中：正在生成回复", turn_id=turn_id) assistant_text = "" try: for delta in self.llm.stream_reply(self.context.build_llm_messages()): @@ -246,26 +287,42 @@ class LiveVoiceRuntime: turn_id, ) self.context.append_assistant(assistant_text) - self._state(PipelineState.WAKE_LISTENING, "恢复待机：可继续唤醒", turn_id=turn_id) + self._event(STANDBY_RESUMED, PipelineState.WAKE_LISTENING, "恢复待机：可继续唤醒", turn_id=turn_id) return TurnResult(True, user_text, assistant_text, states=list(self._states)) def _speak(self, sentence: str, turn_id: int) -> None: - self._state(PipelineState.SPEAKING, "播放中：正在播报回复", turn_id=turn_id) + 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 + self._event(PLAYBACK_FINISHED, PipelineState.SPEAKING, "播放完成", turn_id=turn_id) self._drain_input_after_playback() def _recover(self, error: ProviderError, turn_id: int) -> TurnResult: - self.reporter.error(error.stage, error.code.value, error.message, turn_id=turn_id) - self._state(PipelineState.ERROR_RECOVERING, "恢复待机：本轮已结束", turn_id=turn_id) - self._state(PipelineState.WAKE_LISTENING, "恢复待机：可继续唤醒", turn_id=turn_id) + 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)) - def _state(self, state: PipelineState, message: str, *, turn_id: int) -> None: + def _event( + self, + event_type: str, + state: PipelineState, + message: str, + *, + turn_id: int, + payload: dict[str, object] | None = None, + ) -> None: self._states.append(state) - self.reporter.status(state.value, message, turn_id=turn_id) + self.event_bus.emit(event_type, turn_id=turn_id, state=state, message=message, payload=payload) + + def _report_event(self, event: PipelineEvent) -> None: + handler = getattr(self.reporter, "handle_event", None) + if callable(handler): + handler(event) + return + dispatch_pipeline_event(self.reporter, event) def build_live_runtime(config: AppConfig, reporter: RuntimeReporter | None = None) -> LiveVoiceRuntime: diff --git a/tests/test_live_runtime.py b/tests/test_live_runtime.py index c7fced3..7894856 100644 --- a/tests/test_live_runtime.py +++ b/tests/test_live_runtime.py @@ -4,6 +4,21 @@ import unittest from owner_voice_pet.config import AppConfig from owner_voice_pet.conversation import ConversationContext +from owner_voice_pet.events import ( + ACK_STARTED, + CAPTURE_STARTED, + LLM_STARTED, + PLAYBACK_FINISHED, + SPEECH_ENDED, + SPEECH_STARTED, + STANDBY_RESUMED, + STT_STARTED, + TRANSCRIPT_FINAL, + TTS_STARTED, + WAKE_DETECTED, + WAKE_LISTENING, + PipelineEventBus, +) from owner_voice_pet.llm import MockLlmProvider from owner_voice_pet.models import AudioFrame, AudioSegment, Transcript from owner_voice_pet.runtime import LiveVoiceRuntime @@ -79,6 +94,7 @@ def make_runtime(texts: list[str], context: ConversationContext | None = None) - llm = MockLlmProvider(["这是答复。"]) tts = SineTtsProvider() reporter = RecordingReporter() + event_bus = PipelineEventBus() runtime = LiveVoiceRuntime( config=AppConfig(llm_api_key="secret", speech_provider="cloud", post_playback_drain_ms=0), transport=transport, @@ -89,6 +105,7 @@ def make_runtime(texts: list[str], context: ConversationContext | None = None) - tts=tts, context=context or ConversationContext(), reporter=reporter, + event_bus=event_bus, ) return runtime, stt, llm, transport, reporter @@ -108,6 +125,23 @@ class LiveRuntimeTests(unittest.TestCase): self.assertLess(reporter.statuses.index("应答中：我在"), reporter.statuses.index("请说出问题")) self.assertLess(reporter.statuses.index("请说出问题"), reporter.statuses.index("录音中：正在听取问题")) self.assertIn("恢复待机：可继续唤醒", reporter.statuses[-1]) + event_types = [event.type for event in runtime.event_bus.events] + expected_order = [ + WAKE_LISTENING, + WAKE_DETECTED, + ACK_STARTED, + CAPTURE_STARTED, + SPEECH_STARTED, + SPEECH_ENDED, + STT_STARTED, + TRANSCRIPT_FINAL, + LLM_STARTED, + TTS_STARTED, + PLAYBACK_FINISHED, + STANDBY_RESUMED, + ] + positions = [event_types.index(item) for item in expected_order] + self.assertEqual(positions, sorted(positions)) def test_temporary_context_is_sent_to_second_llm_call(self) -> None: runtime, _, llm, _, _ = make_runtime(["第一问", "第二问"])