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

Deprecate expect_stripped_words option from LLMAssistantAggregatorParams, when used with the newer LLMAssistantAggregator, which now handles word spacing automatically.

Browse Source 
This commit does not change how it works in the older `LLMAssistantContextAggregator`.
This commit is contained in:
Paul Kompfner
2025-10-30 15:48:15 -04:00
parent 5e00133e64
commit ac5734d0ed
18 changed files with 130 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -89,7 +89,9 @@ class LLMAssistantAggregatorParams:
Parameters:
expect_stripped_words: Whether to expect and handle stripped words
in text frames by adding spaces between tokens.
in text frames by adding spaces between tokens. This parameter is
ignored when used with the newer LLMAssistantAggregator, which
handles word spacing automatically.
"""
expect_stripped_words: bool = True

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

@@ -13,6 +13,7 @@ LLM processing, and text-to-speech components in conversational AI pipelines.
import asyncio
import json
import warnings
from abc import abstractmethod
from typing import Any, Dict, List, Literal, Optional, Set
@@ -65,6 +66,7 @@ from pipecat.processors.aggregators.llm_response import (
LLMUserAggregatorParams,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.utils.string import concatenate_aggregated_text
from pipecat.utils.time import time_now_iso8601
@@ -88,7 +90,7 @@ class LLMContextAggregator(FrameProcessor):
self._context = context
self._role = role
self._aggregation: str = ""
self._aggregation: List[str] = []
@property
def messages(self) -> List[LLMContextMessage]:
@@ -168,13 +170,21 @@ class LLMContextAggregator(FrameProcessor):
async def reset(self):
"""Reset the aggregation state."""
self._aggregation = ""
self._aggregation = []
@abstractmethod
async def push_aggregation(self):
"""Push the current aggregation downstream."""
pass
def aggregation_string(self) -> str:
"""Get the current aggregation as a string.
Returns:
The concatenated aggregation string.
"""
return concatenate_aggregated_text(self._aggregation)
class LLMUserAggregator(LLMContextAggregator):
"""User LLM aggregator that processes speech-to-text transcriptions.
@@ -212,8 +222,6 @@ class LLMUserAggregator(LLMContextAggregator):
self._turn_params: Optional[SmartTurnParams] = None
if "aggregation_timeout" in kwargs:
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
@@ -307,7 +315,7 @@ class LLMUserAggregator(LLMContextAggregator):
async def _process_aggregation(self):
"""Process the current aggregation and push it downstream."""
aggregation = self._aggregation
aggregation = self.aggregation_string()
await self.reset()
self._context.add_message({"role": self.role, "content": aggregation})
frame = LLMContextFrame(self._context)
@@ -355,7 +363,7 @@ class LLMUserAggregator(LLMContextAggregator):
"""
async def should_interrupt(strategy: BaseInterruptionStrategy):
await strategy.append_text(self._aggregation)
await strategy.append_text(self.aggregation_string())
return await strategy.should_interrupt()
return any([await should_interrupt(s) for s in self._interruption_strategies])
@@ -425,7 +433,7 @@ class LLMUserAggregator(LLMContextAggregator):
if not text.strip():
return
self._aggregation += f" {text}" if self._aggregation else text
self._aggregation.append(text)
# We just got a final result, so let's reset interim results.
self._seen_interim_results = False
# Reset aggregation timer.
@@ -550,23 +558,31 @@ class LLMAssistantAggregator(LLMContextAggregator):
Args:
context: The OpenAI LLM context for conversation storage.
params: Configuration parameters for aggregation behavior.
**kwargs: Additional arguments. Supports deprecated 'expect_stripped_words'.
**kwargs: Additional arguments.
"""
super().__init__(context=context, role="assistant", **kwargs)
self._params = params or LLMAssistantAggregatorParams()
if "expect_stripped_words" in kwargs:
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"Parameter 'expect_stripped_words' is deprecated, use 'params' instead.",
"Parameter 'expect_stripped_words' is deprecated. "
"LLMAssistantAggregator now handles word spacing automatically.",
DeprecationWarning,
)
self._params.expect_stripped_words = kwargs["expect_stripped_words"]
if params and not params.expect_stripped_words:
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"params.expect_stripped_words is deprecated. "
"LLMAssistantAggregator now handles word spacing automatically.",
DeprecationWarning,
)
self._started = 0
self._function_calls_in_progress: Dict[str, Optional[FunctionCallInProgressFrame]] = {}
self._context_updated_tasks: Set[asyncio.Task] = set()
@@ -629,7 +645,7 @@ class LLMAssistantAggregator(LLMContextAggregator):
if not self._aggregation:
return
aggregation = self._aggregation.strip()
aggregation = self.aggregation_string()
await self.reset()
if aggregation:
@@ -793,10 +809,11 @@ class LLMAssistantAggregator(LLMContextAggregator):
if not self._started:
return
if self._params.expect_stripped_words:
self._aggregation += f" {frame.text}" if self._aggregation else frame.text
else:
self._aggregation += frame.text
# Make sure we really have text (spaces count, too!)
if len(frame.text) == 0:
return
self._aggregation.append(frame.text)
def _context_updated_task_finished(self, task: asyncio.Task):
self._context_updated_tasks.discard(task)

25
src/pipecat/processors/transcript_processor.py
View File

@@ -26,6 +26,7 @@ from pipecat.frames.frames import (
TTSTextFrame,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.utils.string import concatenate_aggregated_text
from pipecat.utils.time import time_now_iso8601
@@ -140,29 +141,7 @@ class AssistantTranscriptProcessor(BaseTranscriptProcessor):
Result: "Hello there how are you"
"""
if self._current_text_parts and self._aggregation_start_time:
# Check specifically for space characters, previously isspace() was used
# but that includes all whitespace characters (e.g. \n), not just spaces.
has_leading_spaces = any(
part and part[0] == " " for part in self._current_text_parts[1:]
)
has_trailing_spaces = any(
part and part[-1] == " " for part in self._current_text_parts[:-1]
)
# If there are embedded spaces in the fragments, use direct concatenation
contains_spacing_between_fragments = has_leading_spaces or has_trailing_spaces
# Apply corresponding joining method
if contains_spacing_between_fragments:
# Fragments already have spacing - just concatenate
content = "".join(self._current_text_parts)
else:
# Word-by-word fragments - join with spaces
content = " ".join(self._current_text_parts)
# Clean up any excessive whitespace
content = content.strip()
content = concatenate_aggregated_text(self._current_text_parts)
if content:
logger.trace(f"Emitting aggregated assistant message: {content}")
message = TranscriptionMessage(

39
src/pipecat/utils/string.py
View File

@@ -18,7 +18,7 @@ Dependencies:
"""
import re
from typing import FrozenSet, Optional, Sequence, Tuple
from typing import FrozenSet, List, Optional, Sequence, Tuple
import nltk
from loguru import logger
@@ -196,3 +196,40 @@ def parse_start_end_tags(
return (None, len(text))
return (None, current_tag_index)
def concatenate_aggregated_text(text_parts: List[str]) -> str:
"""Concatenate a list of text parts into a single string.
This function joins the provided list of text parts into a single string,
taking into account whether or not the parts already contain spacing.
This function is useful for aggregating text segments received from LLMs or
transcription services.
Args:
text_parts: A list of strings representing parts of text to concatenate.
Returns:
A single concatenated string.
"""
# Check specifically for space characters, previously isspace() was used
# but that includes all whitespace characters (e.g. \n), not just spaces.
has_leading_spaces = any(part and part[0] == " " for part in text_parts[1:])
has_trailing_spaces = any(part and part[-1] == " " for part in text_parts[:-1])
# If there are embedded spaces in the fragments, use direct concatenation
contains_spacing_between_fragments = has_leading_spaces or has_trailing_spaces
# Apply corresponding joining method
if contains_spacing_between_fragments:
# Fragments already have spacing - just concatenate
result = "".join(text_parts)
else:
# Word-by-word fragments - join with spaces
result = " ".join(text_parts)
# Clean up any excessive whitespace
result = result.strip()
return result