wx44wx/engine-v5-pipecat-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update xfyun asr streaming boundary

Browse Source
This commit is contained in:
Xin Wang
2026-05-22 00:06:15 +08:00
parent 2ecb3ef506
commit c0fb0c51a2
2 changed files with 61 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
README.md
View File

@@ -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": {

105
engine/xfyun_asr.py
View File

@@ -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: