diff --git a/examples/foundational/22b-natural-conversation-proposal.py b/examples/foundational/22b-natural-conversation-proposal.py index 417aeca76..9b3ff1d7a 100644 --- a/examples/foundational/22b-natural-conversation-proposal.py +++ b/examples/foundational/22b-natural-conversation-proposal.py @@ -9,8 +9,9 @@ import os from dotenv import load_dotenv from loguru import logger -from openai.types.chat import ChatCompletionToolParam +from pipecat.adapters.schemas.function_schema import FunctionSchema +from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.frames.frames import ( CancelFrame, @@ -19,6 +20,7 @@ from pipecat.frames.frames import ( FunctionCallInProgressFrame, FunctionCallResultFrame, InterruptionFrame, + LLMContextFrame, LLMRunFrame, StartFrame, SystemFrame, @@ -32,10 +34,8 @@ from pipecat.pipeline.parallel_pipeline import ParallelPipeline from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask -from pipecat.processors.aggregators.openai_llm_context import ( - OpenAILLMContext, - OpenAILLMContextFrame, -) +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair from pipecat.processors.filters.function_filter import FunctionFilter from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.processors.user_idle_processor import UserIdleProcessor @@ -66,13 +66,13 @@ class StatementJudgeContextFilter(FrameProcessor): await self.push_frame(frame, direction) return - # We only want to handle OpenAILLMContextFrames, and only want to push through a simplified + # We only want to handle LLMContextFrames, and only want to push through a simplified # context frame that contains a system prompt and the most recent user messages, # concatenated. - if isinstance(frame, OpenAILLMContextFrame): + if isinstance(frame, LLMContextFrame): logger.debug(f"Context Frame: {frame}") # Take text content from the most recent user messages. - messages = frame.context.messages + messages = frame.context.get_messages() user_text_messages = [] last_assistant_message = None for message in reversed(messages): @@ -100,7 +100,7 @@ class StatementJudgeContextFilter(FrameProcessor): if last_assistant_message: messages.append(last_assistant_message) messages.append({"role": "user", "content": user_message}) - await self.push_frame(OpenAILLMContextFrame(OpenAILLMContext(messages))) + await self.push_frame(LLMContextFrame(LLMContext(messages))) class CompletenessCheck(FrameProcessor): @@ -231,22 +231,26 @@ class TurnDetectionLLM(Pipeline): async def pass_only_llm_trigger_frames(frame): return ( - isinstance(frame, OpenAILLMContextFrame) + isinstance(frame, LLMContextFrame) or isinstance(frame, InterruptionFrame) or isinstance(frame, FunctionCallInProgressFrame) or isinstance(frame, FunctionCallResultFrame) ) + async def filter_all(frame): + return False + super().__init__( [ ParallelPipeline( [ - # Ignore everything except an OpenAILLMContextFrame. Pass a specially constructed + # Ignore everything except an LLMContextFrame. Pass a specially constructed # simplified context frame to the statement classifier LLM. The only frame this # sub-pipeline will output is a UserStoppedSpeakingFrame. statement_judge_context_filter, statement_llm, completeness_check, + FunctionFilter(filter=filter_all, direction=FrameDirection.UPSTREAM), ], [ # Block everything except frames that trigger LLM inference. @@ -302,30 +306,23 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): async def on_function_calls_started(service, function_calls): await tts.queue_frame(TTSSpeakFrame("Let me check on that.")) - tools = [ - ChatCompletionToolParam( - type="function", - function={ - "name": "get_current_weather", - "description": "Get the current weather", - "parameters": { - "type": "object", - "properties": { - "location": { - "type": "string", - "description": "The city and state, e.g. San Francisco, CA", - }, - "format": { - "type": "string", - "enum": ["celsius", "fahrenheit"], - "description": "The temperature unit to use. Infer this from the users location.", - }, - }, - "required": ["location", "format"], - }, + weather_function = FunctionSchema( + name="get_current_weather", + description="Get the current weather", + properties={ + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", }, - ) - ] + "format": { + "type": "string", + "enum": ["celsius", "fahrenheit"], + "description": "The temperature unit to use. Infer this from the users location.", + }, + }, + required=["location", "format"], + ) + tools = ToolsSchema(standard_tools=[weather_function]) messages = [ { @@ -334,8 +331,8 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): }, ] - context = OpenAILLMContext(messages, tools) - context_aggregator = llm_main.create_context_aggregator(context) + context = LLMContext(messages, tools) + context_aggregator = LLMContextAggregatorPair(context) # LLM + turn detection (with an extra LLM as a judge) llm = TurnDetectionLLM(llm_main) @@ -369,7 +366,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): await task.queue_frames([LLMRunFrame()]) @transport.event_handler("on_app_message") - async def on_app_message(transport, message): + async def on_app_message(transport, message, sender): logger.debug(f"Received app message: {message}") if "message" not in message: return diff --git a/examples/foundational/22c-natural-conversation-mixed-llms.py b/examples/foundational/22c-natural-conversation-mixed-llms.py index e4c554b26..461fab08d 100644 --- a/examples/foundational/22c-natural-conversation-mixed-llms.py +++ b/examples/foundational/22c-natural-conversation-mixed-llms.py @@ -9,8 +9,9 @@ import os from dotenv import load_dotenv from loguru import logger -from openai.types.chat import ChatCompletionToolParam +from pipecat.adapters.schemas.function_schema import FunctionSchema +from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.frames.frames import ( CancelFrame, @@ -19,6 +20,7 @@ from pipecat.frames.frames import ( FunctionCallInProgressFrame, FunctionCallResultFrame, InterruptionFrame, + LLMContextFrame, LLMRunFrame, StartFrame, SystemFrame, @@ -32,10 +34,8 @@ from pipecat.pipeline.parallel_pipeline import ParallelPipeline from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask -from pipecat.processors.aggregators.openai_llm_context import ( - OpenAILLMContext, - OpenAILLMContextFrame, -) +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair from pipecat.processors.filters.function_filter import FunctionFilter from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.processors.user_idle_processor import UserIdleProcessor @@ -272,11 +272,11 @@ class StatementJudgeContextFilter(FrameProcessor): await self.push_frame(frame, direction) return - # We only want to handle OpenAILLMContextFrames, and only want to push through a simplified + # We only want to handle LLMContextFrames, and only want to push through a simplified # context frame that contains a system prompt and the most recent user messages, - if isinstance(frame, OpenAILLMContextFrame): + if isinstance(frame, LLMContextFrame): # Take text content from the most recent user messages. - messages = frame.context.messages + messages = frame.context.get_messages() user_text_messages = [] last_assistant_message = None for message in reversed(messages): @@ -303,7 +303,7 @@ class StatementJudgeContextFilter(FrameProcessor): if last_assistant_message: messages.append(last_assistant_message) messages.append({"role": "user", "content": user_message}) - await self.push_frame(OpenAILLMContextFrame(OpenAILLMContext(messages))) + await self.push_frame(LLMContextFrame(LLMContext(messages))) class CompletenessCheck(FrameProcessor): @@ -425,12 +425,15 @@ class TurnDetectionLLM(Pipeline): async def pass_only_llm_trigger_frames(frame): return ( - isinstance(frame, OpenAILLMContextFrame) + isinstance(frame, LLMContextFrame) or isinstance(frame, InterruptionFrame) or isinstance(frame, FunctionCallInProgressFrame) or isinstance(frame, FunctionCallResultFrame) ) + async def filter_all(frame): + return False + super().__init__( [ ParallelPipeline( @@ -440,12 +443,13 @@ class TurnDetectionLLM(Pipeline): FunctionFilter(filter=block_user_stopped_speaking), ], [ - # Ignore everything except an OpenAILLMContextFrame. Pass a specially constructed + # Ignore everything except an LLMContextFrame. Pass a specially constructed # simplified context frame to the statement classifier LLM. The only frame this # sub-pipeline will output is a UserStoppedSpeakingFrame. statement_judge_context_filter, statement_llm, completeness_check, + FunctionFilter(filter=filter_all, direction=FrameDirection.UPSTREAM), ], [ # Block everything except frames that trigger LLM inference. @@ -505,30 +509,23 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): async def on_function_calls_started(service, function_calls): await tts.queue_frame(TTSSpeakFrame("Let me check on that.")) - tools = [ - ChatCompletionToolParam( - type="function", - function={ - "name": "get_current_weather", - "description": "Get the current weather", - "parameters": { - "type": "object", - "properties": { - "location": { - "type": "string", - "description": "The city and state, e.g. San Francisco, CA", - }, - "format": { - "type": "string", - "enum": ["celsius", "fahrenheit"], - "description": "The temperature unit to use. Infer this from the users location.", - }, - }, - "required": ["location", "format"], - }, + weather_function = FunctionSchema( + name="get_current_weather", + description="Get the current weather", + properties={ + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", }, - ) - ] + "format": { + "type": "string", + "enum": ["celsius", "fahrenheit"], + "description": "The temperature unit to use. Infer this from the users location.", + }, + }, + required=["location", "format"], + ) + tools = ToolsSchema(standard_tools=[weather_function]) messages = [ { @@ -537,8 +534,8 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): }, ] - context = OpenAILLMContext(messages, tools) - context_aggregator = llm_main.create_context_aggregator(context) + context = LLMContext(messages, tools) + context_aggregator = LLMContextAggregatorPair(context) # LLM + turn detection (with an extra LLM as a judge) llm = TurnDetectionLLM(llm_main) @@ -577,7 +574,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): await task.queue_frames([LLMRunFrame()]) @transport.event_handler("on_app_message") - async def on_app_message(transport, message): + async def on_app_message(transport, message, sender): logger.debug(f"Received app message: {message}") if "message" not in message: return diff --git a/examples/foundational/22d-natural-conversation-gemini-audio.py b/examples/foundational/22d-natural-conversation-gemini-audio.py index d7ecf2ba7..5819c9cb9 100644 --- a/examples/foundational/22d-natural-conversation-gemini-audio.py +++ b/examples/foundational/22d-natural-conversation-gemini-audio.py @@ -9,7 +9,6 @@ import os import time from dotenv import load_dotenv -from google.genai.types import Content, Part from loguru import logger from pipecat.audio.vad.silero import SileroVADAnalyzer @@ -21,6 +20,7 @@ from pipecat.frames.frames import ( FunctionCallResultFrame, InputAudioRawFrame, InterruptionFrame, + LLMContextFrame, LLMFullResponseStartFrame, LLMRunFrame, StartFrame, @@ -34,20 +34,18 @@ from pipecat.pipeline.parallel_pipeline import ParallelPipeline from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.aggregators.llm_response import ( LLMAssistantAggregatorParams, LLMAssistantResponseAggregator, ) -from pipecat.processors.aggregators.openai_llm_context import ( - OpenAILLMContext, - OpenAILLMContextFrame, -) +from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair from pipecat.processors.filters.function_filter import FunctionFilter from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport from pipecat.services.cartesia.tts import CartesiaTTSService -from pipecat.services.google.llm import GoogleLLMContext, GoogleLLMService +from pipecat.services.google.llm import GoogleLLMService from pipecat.services.llm_service import LLMService from pipecat.sync.base_notifier import BaseNotifier from pipecat.sync.event_notifier import EventNotifier @@ -375,7 +373,7 @@ class AudioAccumulator(FrameProcessor): await super().process_frame(frame, direction) # ignore context frame - if isinstance(frame, OpenAILLMContextFrame): + if isinstance(frame, LLMContextFrame): return if isinstance(frame, TranscriptionFrame): @@ -392,9 +390,9 @@ class AudioAccumulator(FrameProcessor): f"Processing audio buffer seconds: ({len(self._audio_frames)}) ({len(data)}) {len(data) / 2 / 16000}" ) self._user_speaking = False - context = GoogleLLMContext() + context = LLMContext() context.add_audio_frames_message(audio_frames=self._audio_frames) - await self.push_frame(OpenAILLMContextFrame(context=context)) + await self.push_frame(LLMContextFrame(context=context)) elif isinstance(frame, InputAudioRawFrame): # Append the audio frame to our buffer. Treat the buffer as a ring buffer, dropping the oldest # frames as necessary. @@ -513,7 +511,7 @@ class LLMAggregatorBuffer(LLMAssistantResponseAggregator): class ConversationAudioContextAssembler(FrameProcessor): """Takes the single-message context generated by the AudioAccumulator and adds it to the conversation LLM's context.""" - def __init__(self, context: OpenAILLMContext, **kwargs): + def __init__(self, context: LLMContext, **kwargs): super().__init__(**kwargs) self._context = context @@ -525,11 +523,10 @@ class ConversationAudioContextAssembler(FrameProcessor): await self.push_frame(frame, direction) return - if isinstance(frame, OpenAILLMContextFrame): - GoogleLLMContext.upgrade_to_google(self._context) - last_message = frame.context.messages[-1] + if isinstance(frame, LLMContextFrame): + last_message = frame.context.get_messages()[-1] self._context._messages.append(last_message) - await self.push_frame(OpenAILLMContextFrame(context=self._context)) + await self.push_frame(LLMContextFrame(context=self._context)) class OutputGate(FrameProcessor): @@ -543,7 +540,7 @@ class OutputGate(FrameProcessor): def __init__( self, notifier: BaseNotifier, - context: OpenAILLMContext, + context: LLMContext, llm_transcription_buffer: LLMAggregatorBuffer, **kwargs, ): @@ -610,19 +607,23 @@ class OutputGate(FrameProcessor): self._gate_task = None async def _gate_task_handler(self): - await self._notifier.wait() + while True: + try: + await self._notifier.wait() - transcription = await self._transcription_buffer.wait_for_transcription() or "-" - self._context.add_message(Content(role="user", parts=[Part(text=transcription)])) + transcription = await self._transcription_buffer.wait_for_transcription() or "-" + self._context.add_message({"role": "user", "content": transcription}) - self.open_gate() - for frame, direction in self._frames_buffer: - await self.push_frame(frame, direction) - self._frames_buffer = [] + self.open_gate() + for frame, direction in self._frames_buffer: + await self.push_frame(frame, direction) + self._frames_buffer = [] + except asyncio.CancelledError: + break class TurnDetectionLLM(Pipeline): - def __init__(self, llm: LLMService, context: OpenAILLMContext): + def __init__(self, llm: LLMService, context: LLMContext): # This is the LLM that will transcribe user speech. tx_llm = GoogleLLMService( name="Transcriber", @@ -648,10 +649,10 @@ class TurnDetectionLLM(Pipeline): # as complete or incomplete. # statement_judge_context_filter = StatementJudgeAudioContextAccumulator(notifier=notifier) - audio_accumulater = AudioAccumulator() + audio_accumulator = AudioAccumulator() # This sends a UserStoppedSpeakingFrame and triggers the notifier event completeness_check = CompletenessCheck( - notifier=notifier, audio_accumulator=audio_accumulater + notifier=notifier, audio_accumulator=audio_accumulator ) async def block_user_stopped_speaking(frame): @@ -667,7 +668,7 @@ class TurnDetectionLLM(Pipeline): super().__init__( [ - audio_accumulater, + audio_accumulator, ParallelPipeline( [ # Pass everything except UserStoppedSpeaking to the elements after @@ -734,8 +735,8 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): system_instruction=conversation_system_instruction, ) - context = OpenAILLMContext() - context_aggregator = conversation_llm.create_context_aggregator(context) + context = LLMContext() + context_aggregator = LLMContextAggregatorPair(context) llm = TurnDetectionLLM(conversation_llm, context) @@ -761,12 +762,10 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): @transport.event_handler("on_client_connected") async def on_client_connected(transport, client): logger.info(f"Client connected") - # Kick off the conversation. - await task.queue_frames([LLMRunFrame()]) @transport.event_handler("on_app_message") - async def on_app_message(transport, message): - logger.debug(f"Received app message: {message}") + async def on_app_message(transport, message, sender): + logger.debug(f"Received app message: {message}, sender: {sender}") # TODO: revert if "message" not in message: return diff --git a/src/pipecat/transports/smallwebrtc/transport.py b/src/pipecat/transports/smallwebrtc/transport.py index 4a9ba9341..4b2437be1 100644 --- a/src/pipecat/transports/smallwebrtc/transport.py +++ b/src/pipecat/transports/smallwebrtc/transport.py @@ -66,7 +66,7 @@ class SmallWebRTCCallbacks(BaseModel): on_client_disconnected: Called when a client disconnects. """ - on_app_message: Callable[[Any], Awaitable[None]] + on_app_message: Callable[[Any, str], Awaitable[None]] on_client_connected: Callable[[SmallWebRTCConnection], Awaitable[None]] on_client_disconnected: Callable[[SmallWebRTCConnection], Awaitable[None]] @@ -254,7 +254,7 @@ class SmallWebRTCClient: @self._webrtc_connection.event_handler("app-message") async def on_app_message(connection: SmallWebRTCConnection, message: Any): - await self._handle_app_message(message) + await self._handle_app_message(message, connection.pc_id) def _convert_frame(self, frame_array: np.ndarray, format_name: str) -> np.ndarray: """Convert a video frame to RGB format based on the input format. @@ -512,9 +512,9 @@ class SmallWebRTCClient: if not self._closing: await self._callbacks.on_client_disconnected(self._webrtc_connection) - async def _handle_app_message(self, message: Any): + async def _handle_app_message(self, message: Any, sender: str): """Handle incoming application messages.""" - await self._callbacks.on_app_message(message) + await self._callbacks.on_app_message(message, sender) def _can_send(self): """Check if the connection is ready for sending data.""" @@ -935,11 +935,11 @@ class SmallWebRTCTransport(BaseTransport): if self._output: await self._output.queue_frame(frame, FrameDirection.DOWNSTREAM) - async def _on_app_message(self, message: Any): + async def _on_app_message(self, message: Any, sender: str): """Handle incoming application messages.""" if self._input: await self._input.push_app_message(message) - await self._call_event_handler("on_app_message", message) + await self._call_event_handler("on_app_message", message, sender) async def _on_client_connected(self, webrtc_connection): """Handle client connection events."""