76 lines
2.4 KiB
Python
76 lines
2.4 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from pipecat.frames.frames import (
|
||
|
|
Frame,
|
||
|
|
InterruptionFrame,
|
||
|
|
LLMFullResponseEndFrame,
|
||
|
|
LLMFullResponseStartFrame,
|
||
|
|
LLMTextFrame,
|
||
|
|
OutputTransportMessageUrgentFrame,
|
||
|
|
TTSSpeakFrame,
|
||
|
|
)
|
||
|
|
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
|
||
|
|
|
||
|
|
|
||
|
|
class ProductTextStreamProcessor(FrameProcessor):
|
||
|
|
"""Mirrors LLM text frames as streaming protocol events."""
|
||
|
|
|
||
|
|
def __init__(self) -> None:
|
||
|
|
super().__init__()
|
||
|
|
self._aggregation: list[str] = []
|
||
|
|
self._turn_active = False
|
||
|
|
|
||
|
|
async def process_frame(self, frame: Frame, direction: FrameDirection) -> None:
|
||
|
|
await super().process_frame(frame, direction)
|
||
|
|
|
||
|
|
if isinstance(frame, LLMFullResponseStartFrame):
|
||
|
|
await self._start_turn()
|
||
|
|
elif isinstance(frame, LLMTextFrame):
|
||
|
|
if frame.text:
|
||
|
|
await self._delta(frame.text)
|
||
|
|
elif isinstance(frame, LLMFullResponseEndFrame):
|
||
|
|
await self._end_turn(interrupted=False)
|
||
|
|
elif isinstance(frame, InterruptionFrame):
|
||
|
|
await self._end_turn(interrupted=True)
|
||
|
|
elif isinstance(frame, TTSSpeakFrame):
|
||
|
|
text = frame.text or ""
|
||
|
|
await self._start_turn()
|
||
|
|
if text:
|
||
|
|
await self._delta(text)
|
||
|
|
await self._end_turn(interrupted=False)
|
||
|
|
|
||
|
|
await self.push_frame(frame, direction)
|
||
|
|
|
||
|
|
async def _start_turn(self) -> None:
|
||
|
|
if self._turn_active:
|
||
|
|
return
|
||
|
|
self._turn_active = True
|
||
|
|
self._aggregation = []
|
||
|
|
await self._emit("response.text.started")
|
||
|
|
|
||
|
|
async def _delta(self, text: str) -> None:
|
||
|
|
if not self._turn_active:
|
||
|
|
await self._start_turn()
|
||
|
|
self._aggregation.append(text)
|
||
|
|
await self._emit("response.text.delta", text=text)
|
||
|
|
|
||
|
|
async def _end_turn(self, *, interrupted: bool) -> None:
|
||
|
|
if not self._turn_active:
|
||
|
|
return
|
||
|
|
full_text = "".join(self._aggregation)
|
||
|
|
self._turn_active = False
|
||
|
|
self._aggregation = []
|
||
|
|
await self._emit(
|
||
|
|
"response.text.final",
|
||
|
|
text=full_text,
|
||
|
|
interrupted=interrupted,
|
||
|
|
)
|
||
|
|
|
||
|
|
async def _emit(self, event_type: str, **payload: object) -> None:
|
||
|
|
await self.push_frame(
|
||
|
|
OutputTransportMessageUrgentFrame(
|
||
|
|
message={"type": event_type, **payload},
|
||
|
|
),
|
||
|
|
FrameDirection.DOWNSTREAM,
|
||
|
|
)
|