diff --git a/examples/foundational/07l-interruptible-together.py b/examples/foundational/07l-interruptible-together.py index e2cb55fed..ca3386718 100644 --- a/examples/foundational/07l-interruptible-together.py +++ b/examples/foundational/07l-interruptible-together.py @@ -5,29 +5,24 @@ # import asyncio -import aiohttp import os import sys +import aiohttp +from dotenv import load_dotenv +from loguru import logger +from runner import configure + from pipecat.frames.frames import LLMMessagesFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask -from pipecat.processors.aggregators.llm_response import ( - LLMAssistantResponseAggregator, - LLMUserResponseAggregator, -) +from pipecat.services.ai_services import OpenAILLMContext from pipecat.services.cartesia import CartesiaTTSService from pipecat.services.together import TogetherLLMService from pipecat.transports.services.daily import DailyParams, DailyTransport from pipecat.vad.silero import SileroVADAnalyzer -from runner import configure - -from loguru import logger - -from dotenv import load_dotenv - load_dotenv(override=True) logger.remove(0) @@ -76,17 +71,19 @@ async def main(): }, ] - tma_in = LLMUserResponseAggregator(messages) - tma_out = LLMAssistantResponseAggregator(messages) + context = OpenAILLMContext(messages) + context_aggregator = llm.create_context_aggregator(context) + user_aggregator = context_aggregator.user() + assistant_aggregator = context_aggregator.assistant() pipeline = Pipeline( [ transport.input(), # Transport user input - tma_in, # User responses + user_aggregator, # User responses llm, # LLM tts, # TTS transport.output(), # Transport bot output - tma_out, # Assistant spoken responses + assistant_aggregator, # Assistant spoken responses ] ) diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index 036f5fe47..479746471 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -6,12 +6,6 @@ from typing import List, Type -from pipecat.processors.aggregators.openai_llm_context import ( - OpenAILLMContextFrame, - OpenAILLMContext, -) - -from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.frames.frames import ( Frame, InterimTranscriptionFrame, @@ -22,11 +16,16 @@ from pipecat.frames.frames import ( LLMMessagesUpdateFrame, LLMSetToolsFrame, StartInterruptionFrame, - TranscriptionFrame, TextFrame, + TranscriptionFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame, ) +from pipecat.processors.aggregators.openai_llm_context import ( + OpenAILLMContext, + OpenAILLMContextFrame, +) +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor class LLMResponseAggregator(FrameProcessor): @@ -40,6 +39,7 @@ class LLMResponseAggregator(FrameProcessor): accumulator_frame: Type[TextFrame], interim_accumulator_frame: Type[TextFrame] | None = None, handle_interruptions: bool = False, + expect_stripped_words: bool = True, # if True, need to add spaces between words ): super().__init__() @@ -50,6 +50,7 @@ class LLMResponseAggregator(FrameProcessor): self._accumulator_frame = accumulator_frame self._interim_accumulator_frame = interim_accumulator_frame self._handle_interruptions = handle_interruptions + self._expect_stripped_words = expect_stripped_words # Reset our accumulator state. self._reset() @@ -111,7 +112,10 @@ class LLMResponseAggregator(FrameProcessor): await self.push_frame(frame, direction) elif isinstance(frame, self._accumulator_frame): if self._aggregating: - self._aggregation += f" {frame.text}" if self._aggregation else frame.text + if self._expect_stripped_words: + self._aggregation += f" {frame.text}" if self._aggregation else frame.text + else: + self._aggregation += frame.text # We have recevied a complete sentence, so if we have seen the # end frame and we were still aggregating, it means we should # send the aggregation. @@ -290,7 +294,7 @@ class LLMContextAggregator(LLMResponseAggregator): class LLMAssistantContextAggregator(LLMContextAggregator): - def __init__(self, context: OpenAILLMContext): + def __init__(self, context: OpenAILLMContext, *, expect_stripped_words: bool = True): super().__init__( messages=[], context=context, @@ -299,6 +303,7 @@ class LLMAssistantContextAggregator(LLMContextAggregator): end_frame=LLMFullResponseEndFrame, accumulator_frame=TextFrame, handle_interruptions=True, + expect_stripped_words=expect_stripped_words, ) diff --git a/src/pipecat/services/anthropic.py b/src/pipecat/services/anthropic.py index bc91e4e16..86e1e3726 100644 --- a/src/pipecat/services/anthropic.py +++ b/src/pipecat/services/anthropic.py @@ -110,9 +110,13 @@ class AnthropicLLMService(LLMService): return self._enable_prompt_caching_beta @staticmethod - def create_context_aggregator(context: OpenAILLMContext) -> AnthropicContextAggregatorPair: + def create_context_aggregator( + context: OpenAILLMContext, *, assistant_expect_stripped_words: bool = True + ) -> AnthropicContextAggregatorPair: user = AnthropicUserContextAggregator(context) - assistant = AnthropicAssistantContextAggregator(user) + assistant = AnthropicAssistantContextAggregator( + user, expect_stripped_words=assistant_expect_stripped_words + ) return AnthropicContextAggregatorPair(_user=user, _assistant=assistant) async def set_enable_prompt_caching_beta(self, enable_prompt_caching_beta: bool): @@ -541,8 +545,8 @@ class AnthropicUserContextAggregator(LLMUserContextAggregator): class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator): - def __init__(self, user_context_aggregator: AnthropicUserContextAggregator): - super().__init__(context=user_context_aggregator._context) + def __init__(self, user_context_aggregator: AnthropicUserContextAggregator, **kwargs): + super().__init__(context=user_context_aggregator._context, **kwargs) self._user_context_aggregator = user_context_aggregator self._function_call_in_progress = None self._function_call_result = None @@ -579,7 +583,7 @@ class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator): run_llm = False aggregation = self._aggregation - self._aggregation = "" + self._reset() try: if self._function_call_result: @@ -630,5 +634,8 @@ class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator): if run_llm: await self._user_context_aggregator.push_context_frame() + frame = OpenAILLMContextFrame(self._context) + await self.push_frame(frame) + except Exception as e: logger.error(f"Error processing frame: {e}") diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index f0892b9ca..c17916f2d 100644 --- a/src/pipecat/services/openai.py +++ b/src/pipecat/services/openai.py @@ -336,9 +336,13 @@ class OpenAILLMService(BaseOpenAILLMService): super().__init__(model=model, params=params, **kwargs) @staticmethod - def create_context_aggregator(context: OpenAILLMContext) -> OpenAIContextAggregatorPair: + def create_context_aggregator( + context: OpenAILLMContext, *, assistant_expect_stripped_words: bool = True + ) -> OpenAIContextAggregatorPair: user = OpenAIUserContextAggregator(context) - assistant = OpenAIAssistantContextAggregator(user) + assistant = OpenAIAssistantContextAggregator( + user, expect_stripped_words=assistant_expect_stripped_words + ) return OpenAIContextAggregatorPair(_user=user, _assistant=assistant) @@ -458,8 +462,8 @@ class OpenAIUserContextAggregator(LLMUserContextAggregator): class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator): - def __init__(self, user_context_aggregator: OpenAIUserContextAggregator): - super().__init__(context=user_context_aggregator._context) + def __init__(self, user_context_aggregator: OpenAIUserContextAggregator, **kwargs): + super().__init__(context=user_context_aggregator._context, **kwargs) self._user_context_aggregator = user_context_aggregator self._function_call_in_progress = None self._function_call_result = None @@ -495,7 +499,7 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator): run_llm = False aggregation = self._aggregation - self._aggregation = "" + self._reset() try: if self._function_call_result: @@ -531,5 +535,8 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator): if run_llm: await self._user_context_aggregator.push_context_frame() + frame = OpenAILLMContextFrame(self._context) + await self.push_frame(frame) + except Exception as e: logger.error(f"Error processing frame: {e}") diff --git a/src/pipecat/services/together.py b/src/pipecat/services/together.py index e4068ecfc..3f4d97964 100644 --- a/src/pipecat/services/together.py +++ b/src/pipecat/services/together.py @@ -95,9 +95,13 @@ class TogetherLLMService(LLMService): return True @staticmethod - def create_context_aggregator(context: OpenAILLMContext) -> TogetherContextAggregatorPair: + def create_context_aggregator( + context: OpenAILLMContext, *, assistant_expect_stripped_words: bool = True + ) -> TogetherContextAggregatorPair: user = TogetherUserContextAggregator(context) - assistant = TogetherAssistantContextAggregator(user) + assistant = TogetherAssistantContextAggregator( + user, expect_stripped_words=assistant_expect_stripped_words + ) return TogetherContextAggregatorPair(_user=user, _assistant=assistant) async def set_frequency_penalty(self, frequency_penalty: float): @@ -331,8 +335,8 @@ class TogetherUserContextAggregator(LLMUserContextAggregator): class TogetherAssistantContextAggregator(LLMAssistantContextAggregator): - def __init__(self, user_context_aggregator: TogetherUserContextAggregator): - super().__init__(context=user_context_aggregator._context) + def __init__(self, user_context_aggregator: TogetherUserContextAggregator, **kwargs): + super().__init__(context=user_context_aggregator._context, **kwargs) self._user_context_aggregator = user_context_aggregator self._function_call_in_progress = None self._function_call_result = None @@ -370,7 +374,7 @@ class TogetherAssistantContextAggregator(LLMAssistantContextAggregator): run_llm = False aggregation = self._aggregation - self._aggregation = "" + self._reset() try: if self._function_call_result: @@ -390,5 +394,8 @@ class TogetherAssistantContextAggregator(LLMAssistantContextAggregator): if run_llm: await self._user_context_aggregator.push_messages_frame() + frame = OpenAILLMContextFrame(self._context) + await self.push_frame(frame) + except Exception as e: logger.error(f"Error processing frame: {e}")