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

Fix text stream frame forwarding order and transport message type.

Browse Source 
Push LLM/TTS frames downstream before emitting product events so the pipeline keeps correct ordering, and use OutputTransportMessageFrame instead of the urgent variant.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Xin Wang
2026-05-24 21:56:04 +08:00
parent f301af396d
commit b918eec5c1
1 changed files with 9 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
engine/text_stream.py
View File

@@ -6,7 +6,7 @@ from pipecat.frames.frames import (
LLMFullResponseEndFrame, LLMFullResponseEndFrame,
LLMFullResponseStartFrame, LLMFullResponseStartFrame,
LLMTextFrame, LLMTextFrame,
OutputTransportMessageUrgentFrame, OutputTransportMessageFrame,
TTSSpeakFrame, TTSSpeakFrame,
) )
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
@@ -38,24 +38,29 @@ class ProductTextStreamProcessor(FrameProcessor):
await super().process_frame(frame, direction) await super().process_frame(frame, direction)
if isinstance(frame, LLMFullResponseStartFrame): if isinstance(frame, LLMFullResponseStartFrame):
await self.push_frame(frame, direction)
await self._start_turn() await self._start_turn()
elif isinstance(frame, LLMTextFrame): elif isinstance(frame, LLMTextFrame):
await self.push_frame(frame, direction)
if frame.text: if frame.text:
await self._delta(frame.text) await self._delta(frame.text)
elif isinstance(frame, LLMFullResponseEndFrame): elif isinstance(frame, LLMFullResponseEndFrame):
await self.push_frame(frame, direction)
await self._end_turn(interrupted=False) await self._end_turn(interrupted=False)
elif isinstance(frame, InterruptionFrame): elif isinstance(frame, InterruptionFrame):
await self.push_frame(frame, direction)
await self._end_turn(interrupted=True) await self._end_turn(interrupted=True)
elif isinstance(frame, TTSSpeakFrame): elif isinstance(frame, TTSSpeakFrame):
# Fixed-text / direct-speech path: there's no LLM cycle, so # Fixed-text / direct-speech path: there's no LLM cycle, so
# synthesize one started/delta/final sequence for the spoken text. # synthesize one started/delta/final sequence for the spoken text.
text = frame.text or "" text = frame.text or ""
await self.push_frame(frame, direction)
await self._start_turn() await self._start_turn()
if text: if text:
await self._delta(text) await self._delta(text)
await self._end_turn(interrupted=False) await self._end_turn(interrupted=False)
else:
await self.push_frame(frame, direction) await self.push_frame(frame, direction)
async def _start_turn(self) -> None: async def _start_turn(self) -> None:
if self._turn_active: if self._turn_active:
@@ -86,7 +91,7 @@ class ProductTextStreamProcessor(FrameProcessor):
async def _emit(self, event_type: str, **payload: object) -> None: async def _emit(self, event_type: str, **payload: object) -> None:
await self.push_frame( await self.push_frame(
OutputTransportMessageUrgentFrame( OutputTransportMessageFrame(
message={"type": event_type, **payload}, message={"type": event_type, **payload},
), ),
FrameDirection.DOWNSTREAM, FrameDirection.DOWNSTREAM,