From 69c7edd60c8d586bf786f7f442e2217d13d9db42 Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Sat, 28 Sep 2024 21:37:03 -0700 Subject: [PATCH 1/5] pushing context frames from assistant aggregators --- src/pipecat/services/anthropic.py | 5 ++++- src/pipecat/services/openai.py | 5 ++++- src/pipecat/services/together.py | 5 ++++- 3 files changed, 12 insertions(+), 3 deletions(-) diff --git a/src/pipecat/services/anthropic.py b/src/pipecat/services/anthropic.py index bc91e4e16..6a535ef15 100644 --- a/src/pipecat/services/anthropic.py +++ b/src/pipecat/services/anthropic.py @@ -579,7 +579,7 @@ class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator): run_llm = False aggregation = self._aggregation - self._aggregation = "" + self._reset() try: if self._function_call_result: @@ -630,5 +630,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..99d2d7497 100644 --- a/src/pipecat/services/openai.py +++ b/src/pipecat/services/openai.py @@ -495,7 +495,7 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator): run_llm = False aggregation = self._aggregation - self._aggregation = "" + self._reset() try: if self._function_call_result: @@ -531,5 +531,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..935f625ad 100644 --- a/src/pipecat/services/together.py +++ b/src/pipecat/services/together.py @@ -370,7 +370,7 @@ class TogetherAssistantContextAggregator(LLMAssistantContextAggregator): run_llm = False aggregation = self._aggregation - self._aggregation = "" + self._reset() try: if self._function_call_result: @@ -390,5 +390,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}") From 37da7e44cdbd7ec3d580926320b4cc5edc9ae450 Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Sat, 28 Sep 2024 22:01:40 -0700 Subject: [PATCH 2/5] whitespace fix --- .../07l-interruptible-together.py | 27 +++++++++---------- .../processors/aggregators/llm_response.py | 15 +++++------ 2 files changed, 19 insertions(+), 23 deletions(-) diff --git a/examples/foundational/07l-interruptible-together.py b/examples/foundational/07l-interruptible-together.py index e2cb55fed..a7086c941 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, tools) + 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..a3cd63cbd 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): @@ -111,7 +110,7 @@ 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 + self._aggregation += frame.text if self._aggregation else 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. From ed607d5c4b30695c3d7a9a669b3ac707ec62647d Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Sat, 28 Sep 2024 22:02:33 -0700 Subject: [PATCH 3/5] typo fix --- examples/foundational/07l-interruptible-together.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/examples/foundational/07l-interruptible-together.py b/examples/foundational/07l-interruptible-together.py index a7086c941..ca3386718 100644 --- a/examples/foundational/07l-interruptible-together.py +++ b/examples/foundational/07l-interruptible-together.py @@ -71,7 +71,7 @@ async def main(): }, ] - context = OpenAILLMContext(messages, tools) + context = OpenAILLMContext(messages) context_aggregator = llm.create_context_aggregator(context) user_aggregator = context_aggregator.user() assistant_aggregator = context_aggregator.assistant() From fef393dcacf4e1c94d7a819cf3ea35722fbc8a67 Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Sun, 29 Sep 2024 14:11:17 -0700 Subject: [PATCH 4/5] assistant aggregator switch for space padding or not --- src/pipecat/processors/aggregators/llm_response.py | 10 ++++++++-- src/pipecat/services/anthropic.py | 12 ++++++++---- src/pipecat/services/openai.py | 12 ++++++++---- src/pipecat/services/together.py | 12 ++++++++---- 4 files changed, 32 insertions(+), 14 deletions(-) diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index a3cd63cbd..4ea38b89f 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -39,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__() @@ -49,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() @@ -110,7 +112,10 @@ class LLMResponseAggregator(FrameProcessor): await self.push_frame(frame, direction) elif isinstance(frame, self._accumulator_frame): if self._aggregating: - self._aggregation += 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 if self._aggregation else 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. @@ -289,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, @@ -298,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 6a535ef15..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 diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index 99d2d7497..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 diff --git a/src/pipecat/services/together.py b/src/pipecat/services/together.py index 935f625ad..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 From 539e0b66fb5abb446358c0b21ece7162fab63237 Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Mon, 30 Sep 2024 09:26:06 -0700 Subject: [PATCH 5/5] small fix as per aleix --- src/pipecat/processors/aggregators/llm_response.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index 4ea38b89f..479746471 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -115,7 +115,7 @@ class LLMResponseAggregator(FrameProcessor): if self._expect_stripped_words: self._aggregation += f" {frame.text}" if self._aggregation else frame.text else: - self._aggregation += frame.text if self._aggregation else frame.text + 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.