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

Support customization over the way the assistant aggregator aggregates LLMTextFrames when tts_skip is on

Browse Source
This commit is contained in:
mattie ruth backman
2025-10-30 16:58:47 -04:00
parent 29417ba44d
commit 0d2c5286fa
2 changed files with 20 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/pipecat/processors/aggregators/llm_response.py
View File

@@ -59,6 +59,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
OpenAILLMContextFrame, OpenAILLMContextFrame,
) )
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.utils.text.base_text_aggregator import BaseTextAggregator
from pipecat.utils.time import time_now_iso8601 from pipecat.utils.time import time_now_iso8601
@@ -95,6 +96,7 @@ class LLMAssistantAggregatorParams:
""" """
expect_stripped_words: bool = True expect_stripped_words: bool = True
llm_text_aggregator: Optional[BaseTextAggregator] = None
class LLMFullResponseAggregator(FrameProcessor): class LLMFullResponseAggregator(FrameProcessor):

36
src/pipecat/processors/aggregators/llm_response_universal.py
View File

@@ -68,7 +68,9 @@ from pipecat.processors.aggregators.llm_response import (
LLMUserAggregatorParams, LLMUserAggregatorParams,
) )
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.utils.string import concatenate_aggregated_text, match_endofsentence from pipecat.utils.string import concatenate_aggregated_text
from pipecat.utils.text.base_text_aggregator import BaseTextAggregator
from pipecat.utils.text.simple_text_aggregator import SimpleTextAggregator
from pipecat.utils.time import time_now_iso8601 from pipecat.utils.time import time_now_iso8601
@@ -597,7 +599,9 @@ class LLMAssistantAggregator(LLMContextAggregator):
self._function_calls_in_progress: Dict[str, Optional[FunctionCallInProgressFrame]] = {} self._function_calls_in_progress: Dict[str, Optional[FunctionCallInProgressFrame]] = {}
self._context_updated_tasks: Set[asyncio.Task] = set() self._context_updated_tasks: Set[asyncio.Task] = set()
self._llm_aggregation: str = "" self._llm_text_aggregator: BaseTextAggregator = (
self._params.llm_text_aggregator or SimpleTextAggregator()
)
self._skip_tts: Optional[bool] = None self._skip_tts: Optional[bool] = None
@property @property
@@ -821,8 +825,6 @@ class LLMAssistantAggregator(LLMContextAggregator):
async def _handle_llm_text(self, frame: LLMTextFrame): async def _handle_llm_text(self, frame: LLMTextFrame):
await self._handle_text(frame) await self._handle_text(frame)
if self._skip_tts or frame.skip_tts:
self._llm_aggregation += frame.text
await self._maybe_push_llm_aggregation(frame) await self._maybe_push_llm_aggregation(frame)
async def _handle_llm_end(self, frame: LLMFullResponseEndFrame): async def _handle_llm_end(self, frame: LLMFullResponseEndFrame):
@@ -833,30 +835,28 @@ class LLMAssistantAggregator(LLMContextAggregator):
async def _maybe_push_llm_aggregation( async def _maybe_push_llm_aggregation(
self, frame: LLMFullResponseStartFrame | LLMTextFrame | LLMFullResponseEndFrame self, frame: LLMFullResponseStartFrame | LLMTextFrame | LLMFullResponseEndFrame
): ):
should_push = False aggregate = None
should_reset_aggregator = False
if self._skip_tts and not frame.skip_tts: if self._skip_tts and not frame.skip_tts:
# if the skip_tts flag switches, to false, push the current aggregation # if the skip_tts flag switches, to false, push the current aggregation
should_push = True aggregate = self._llm_text_aggregator.text
should_reset_aggregator = True
self._skip_tts = frame.skip_tts self._skip_tts = frame.skip_tts
if self._skip_tts: if self._skip_tts:
if self._skip_tts and isinstance(frame, LLMFullResponseEndFrame): if self._skip_tts and isinstance(frame, LLMFullResponseEndFrame):
# on end frame, always push the aggregation # on end frame, always push the aggregation
should_push = True aggregate = self._llm_text_aggregator.text
elif len(self._llm_aggregation) > 0 and match_endofsentence(self._llm_aggregation): should_reset_aggregator = True
# push aggregation on end of sentence elif isinstance(frame, LLMTextFrame):
should_push = True aggregate = await self._llm_text_aggregator.aggregate(frame.text)
if not should_push: if not aggregate:
return return
text = self._llm_aggregation.lstrip("\n") llm_frame = AggregatedLLMTextFrame(text=aggregate.text, aggregated_by=aggregate.type)
if not text.strip():
# don't push empty text
return
llm_frame = AggregatedLLMTextFrame(text=text, aggregated_by="sentence")
await self.push_frame(llm_frame) await self.push_frame(llm_frame)
self._llm_aggregation = "" if should_reset_aggregator:
await self._llm_text_aggregator.reset()
async def _handle_text(self, frame: TextFrame): async def _handle_text(self, frame: TextFrame):
if not self._started or not frame.append_to_context: if not self._started or not frame.append_to_context: