From c0fb0c51a28df51c7b16b307976d2165d3f3969b Mon Sep 17 00:00:00 2001 From: Xin Wang Date: Fri, 22 May 2026 00:06:15 +0800 Subject: [PATCH] Update xfyun asr streaming boundary --- README.md | 9 ++-- engine/xfyun_asr.py | 105 +++++++++++++++++++++++--------------------- 2 files changed, 61 insertions(+), 53 deletions(-) diff --git a/README.md b/README.md index f7644b9..aa40419 100644 --- a/README.md +++ b/README.md @@ -170,9 +170,12 @@ default Smart Turn v3 analyzer, so the engine no longer loads the ### Xfyun ASR The STT provider can be switched to iFlytek/Xfyun's streaming voice dictation -WebSocket API. The engine sends PCM chunks as `encoding: "raw"` and emits -`input.transcript.interim` events with the current full interim transcript as -Xfyun results arrive, followed by the existing `input.transcript.final` event. +WebSocket API. The engine opens the xfyun websocket when Pipecat VAD detects +the user has started speaking, keeps it open across brief pauses, and closes it +only when Pipecat's user-turn strategy declares the logical turn complete. It +sends PCM chunks as `encoding: "raw"` and emits `input.transcript.interim` +events with the current full interim transcript as Xfyun results arrive, +followed by the existing `input.transcript.final` event. ```json "stt": { diff --git a/engine/xfyun_asr.py b/engine/xfyun_asr.py index 07d9260..81c8b5a 100644 --- a/engine/xfyun_asr.py +++ b/engine/xfyun_asr.py @@ -23,7 +23,6 @@ from pipecat.frames.frames import ( TranscriptionFrame, UserStoppedSpeakingFrame, VADUserStartedSpeakingFrame, - VADUserStoppedSpeakingFrame, ) from pipecat.processors.frame_processor import FrameDirection from pipecat.services.settings import STTSettings @@ -77,20 +76,14 @@ class XfyunASRService(STTService): self._audio_buffer = bytearray() self._sent_first_frame = False self._sent_final_frame = False + self._finalizing_turn = False self._partials: list[str] = [] self._last_text = "" - # Text already finalized by xfyun within the current aggregator turn. - # xfyun may emit several status=2 segments within one turn (brief - # user pauses, or the engine's VAD stopping/restarting before the - # turn timeout fires); each segment resets `_partials`/`_last_text`, - # but interim frames pushed to clients should still grow - # monotonically across segments. Reset on the aggregator-level - # `UserStoppedSpeakingFrame` (broadcast by the user aggregator once - # per turn at turn end) — NOT on `VADUserStartedSpeakingFrame`, - # which fires on every brief resume within the same turn and would - # clobber the accumulator mid-utterance, making the bubble appear - # to "un-stream" backwards. - self._turn_committed_text = "" + # Text already emitted as TranscriptionFrame deltas for Pipecat's + # user-turn strategy and context aggregator. The xfyun websocket now + # spans a full logical user turn, so UI interim frames can use + # `_last_text` directly while this cursor prevents duplicate context. + self._turn_transcription_text = "" async def cleanup(self) -> None: await self._close_utterance() @@ -108,20 +101,12 @@ class XfyunASRService(STTService): await super().process_frame(frame, direction) if isinstance(frame, UserStoppedSpeakingFrame): - # Aggregator-level turn end (broadcast by the user aggregator - # once per turn). At this point the xfyun websocket has already - # been closed via `_finish_utterance` on VAD stop, so it's safe - # to clear the cross-segment text accumulator here. Doing it on - # turn end rather than turn start avoids a frame-ordering race - # where the first interim of a new turn could be prefixed with - # the previous turn's committed text (the aggregator broadcasts - # `UserStartedSpeakingFrame` only after `VADUserStartedSpeakingFrame` - # has already propagated through this processor). - self._turn_committed_text = "" + # Aggregator-level turn end (broadcast once per logical user turn). + # This is the only boundary that finalizes/closes the xfyun + # websocket, so brief VAD pauses do not restart the ASR session. + await self._finish_utterance() elif isinstance(frame, VADUserStartedSpeakingFrame): await self._start_utterance() - elif isinstance(frame, VADUserStoppedSpeakingFrame): - await self._finish_utterance() async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame | None, None]: if not audio: @@ -150,6 +135,7 @@ class XfyunASRService(STTService): self._audio_buffer.clear() self._partials = [] self._last_text = "" + self._turn_transcription_text = "" self._sent_first_frame = False self._sent_final_frame = False @@ -180,6 +166,7 @@ class XfyunASRService(STTService): return if not self._sent_final_frame: + self._finalizing_turn = True await self._send_payload({"data": {"status": 2}}) self.request_finalize() self._sent_final_frame = True @@ -201,6 +188,7 @@ class XfyunASRService(STTService): self._audio_buffer.clear() self._sent_first_frame = False self._sent_final_frame = False + self._finalizing_turn = False async def _flush_audio_buffer(self, *, final: bool) -> None: while len(self._audio_buffer) >= self._frame_size: @@ -283,38 +271,55 @@ class XfyunASRService(STTService): text = self._apply_recognition_result(recognition) if text and text != self._last_text: self._last_text = text - await self.push_frame( - InterimTranscriptionFrame( - self._turn_committed_text + text, - self._user_id, - time_now_iso8601(), - _language_or_none(self._language), - result=payload, + if not self._finalizing_turn: + await self.push_frame( + InterimTranscriptionFrame( + text, + self._user_id, + time_now_iso8601(), + _language_or_none(self._language), + result=payload, + ) ) - ) + await self._push_transcription_delta(text, result=payload, finalized=False) if data.get("status") == 2: final_text = self._last_text - if final_text: + if final_text and not self._finalizing_turn: self.confirm_finalize() - # Emit just this segment's text. The pipecat user aggregator - # concatenates TranscriptionFrames within a VAD turn, so we - # must NOT prepend `_turn_committed_text` here or the - # aggregated turn text would double-count earlier segments. - await self.push_frame( - TranscriptionFrame( - final_text, - self._user_id, - time_now_iso8601(), - _language_or_none(self._language), - result=payload, - ) - ) - # Accumulate so the next sub-session's interim frames carry - # the full turn so far (used for client UI display only). - self._turn_committed_text += final_text + await self._push_transcription_delta(final_text, result=payload, finalized=True) await self._close_utterance() + async def _push_transcription_delta( + self, + text: str, + *, + result: dict[str, Any], + finalized: bool, + ) -> None: + if text.startswith(self._turn_transcription_text): + delta = text[len(self._turn_transcription_text) :] + else: + logger.debug( + "Xfyun transcript replacement is not append-only; " + "continuing with the new suffix for turn aggregation" + ) + delta = text + + if not delta and not finalized: + return + + self._turn_transcription_text = text + await self.push_frame( + TranscriptionFrame( + delta, + self._user_id, + time_now_iso8601(), + _language_or_none(self._language), + result=result, + ) + ) + def _apply_recognition_result(self, recognition: dict[str, Any]) -> str: partial = _extract_text_from_result(recognition) if not partial: