From d7d409df606ee4637a470e7d6055abec0f95b893 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 28 Oct 2025 15:09:46 -0700 Subject: [PATCH] examples(foundational): move 12-* to 14-*-video --- CHANGELOG.md | 11 +- .../foundational/12b-describe-video-gpt-4o.py | 184 ------------------ .../12c-describe-video-anthropic.py | 184 ------------------ ...> 14d-function-calling-anthropic-video.py} | 105 +++------- ...s.py => 14d-function-calling-aws-video.py} | 138 +++++++------ ...4d-function-calling-gemini-flash-video.py} | 135 ++++++------- ...> 14d-function-calling-moondream-video.py} | 120 +++++++----- ...y => 14d-function-calling-openai-video.py} | 105 ++++------ 8 files changed, 275 insertions(+), 707 deletions(-) delete mode 100644 examples/foundational/12b-describe-video-gpt-4o.py delete mode 100644 examples/foundational/12c-describe-video-anthropic.py rename examples/foundational/{14b-function-calling-anthropic-video.py => 14d-function-calling-anthropic-video.py} (63%) rename examples/foundational/{12d-describe-video-aws.py => 14d-function-calling-aws-video.py} (58%) rename examples/foundational/{12a-describe-video-gemini-flash.py => 14d-function-calling-gemini-flash-video.py} (55%) rename examples/foundational/{12-describe-video.py => 14d-function-calling-moondream-video.py} (58%) rename examples/foundational/{14d-function-calling-video.py => 14d-function-calling-openai-video.py} (63%) diff --git a/CHANGELOG.md b/CHANGELOG.md index 07596b3b9..b4de8325c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -216,6 +216,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Fixed an issue in `HumeTTSService` that was only using Octave 2, which does + not support the `description` field. Now, if a description is provided, it + switches to Octave 1. + - Fixed an issue where `DailyTransport` would timeout prematurely on join and on leave. @@ -225,7 +229,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Fixed an issue in `ServiceSwitcher` where the `STTService`s would result in all STT services producing `TranscriptionFrame`s. -- Fixed an issue in `HumeTTSService` that was only using Octave 2, which does not support the `description` field. Now, if a description is provided, it switches to Octave 1. +### Other + +- Updated all vision 12-series foundational examples to use function calling to + request for a camera image and also to push `LLMMessagesAppendFrame` with the + retrieved image. For the specific `Moondream` example (`12-describe-video.py`) + we now use a regular LLM and a parallel pipeline with the `MoondreamService`. ## [0.0.91] - 2025-10-21 diff --git a/examples/foundational/12b-describe-video-gpt-4o.py b/examples/foundational/12b-describe-video-gpt-4o.py deleted file mode 100644 index 894d70d7b..000000000 --- a/examples/foundational/12b-describe-video-gpt-4o.py +++ /dev/null @@ -1,184 +0,0 @@ -# -# Copyright (c) 2024–2025, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -import os -from typing import Optional - -from dotenv import load_dotenv -from loguru import logger - -from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams -from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3 -from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.audio.vad.vad_analyzer import VADParams -from pipecat.frames.frames import ( - Frame, - LLMContextFrame, - TextFrame, - TTSSpeakFrame, - UserImageRawFrame, - UserImageRequestFrame, -) -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.user_response import UserResponseAggregator -from pipecat.processors.frame_processor import FrameDirection, FrameProcessor -from pipecat.runner.types import RunnerArguments -from pipecat.runner.utils import ( - create_transport, - get_transport_client_id, - maybe_capture_participant_camera, -) -from pipecat.services.cartesia.tts import CartesiaTTSService -from pipecat.services.deepgram.stt import DeepgramSTTService -from pipecat.services.openai.llm import OpenAILLMService -from pipecat.transports.base_transport import BaseTransport, TransportParams -from pipecat.transports.daily.transport import DailyParams - -load_dotenv(override=True) - - -class UserImageRequester(FrameProcessor): - """Converts incoming text into requests for user images.""" - - def __init__(self, participant_id: Optional[str] = None): - super().__init__() - self._participant_id = participant_id - - def set_participant_id(self, participant_id: str): - self._participant_id = participant_id - - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - if self._participant_id and isinstance(frame, TextFrame): - await self.push_frame( - UserImageRequestFrame(self._participant_id, context=frame.text), - FrameDirection.UPSTREAM, - ) - else: - await self.push_frame(frame, direction) - - -class UserImageProcessor(FrameProcessor): - """Converts incoming user images into context frames.""" - - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - if isinstance(frame, UserImageRawFrame): - if frame.request and frame.request.context: - context = LLMContext() - context.add_image_frame_message( - image=frame.image, - text=frame.request.context, - size=frame.size, - format=frame.format, - ) - frame = LLMContextFrame(context) - await self.push_frame(frame) - else: - await self.push_frame(frame, direction) - - -# We store functions so objects (e.g. SileroVADAnalyzer) don't get -# instantiated. The function will be called when the desired transport gets -# selected. -transport_params = { - "daily": lambda: DailyParams( - audio_in_enabled=True, - audio_out_enabled=True, - video_in_enabled=True, - vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)), - turn_analyzer=LocalSmartTurnAnalyzerV3(params=SmartTurnParams()), - ), - "webrtc": lambda: TransportParams( - audio_in_enabled=True, - audio_out_enabled=True, - video_in_enabled=True, - vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)), - turn_analyzer=LocalSmartTurnAnalyzerV3(params=SmartTurnParams()), - ), -} - - -async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): - logger.info(f"Starting bot") - - user_response = UserResponseAggregator() - - # Initialize the image requester without setting the participant ID yet - image_requester = UserImageRequester() - - image_processor = UserImageProcessor() - - stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) - - # OpenAI GPT-4o for vision analysis - openai = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) - - tts = CartesiaTTSService( - api_key=os.getenv("CARTESIA_API_KEY"), - voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady - ) - - pipeline = Pipeline( - [ - transport.input(), - stt, - user_response, - image_requester, - image_processor, - openai, - tts, - transport.output(), - ] - ) - - task = PipelineTask( - pipeline, - params=PipelineParams( - enable_metrics=True, - enable_usage_metrics=True, - ), - idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, - ) - - @transport.event_handler("on_client_connected") - async def on_client_connected(transport, client): - logger.info(f"Client connected: {client}") - - await maybe_capture_participant_camera(transport, client) - - # Set the participant ID in the image requester - client_id = get_transport_client_id(transport, client) - image_requester.set_participant_id(client_id) - - # Welcome message - await task.queue_frame(TTSSpeakFrame("Hi there! Feel free to ask me about what I see.")) - - @transport.event_handler("on_client_disconnected") - async def on_client_disconnected(transport, client): - logger.info(f"Client disconnected") - await task.cancel() - - runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) - - await runner.run(task) - - -async def bot(runner_args: RunnerArguments): - """Main bot entry point compatible with Pipecat Cloud.""" - transport = await create_transport(runner_args, transport_params) - await run_bot(transport, runner_args) - - -if __name__ == "__main__": - from pipecat.runner.run import main - - main() diff --git a/examples/foundational/12c-describe-video-anthropic.py b/examples/foundational/12c-describe-video-anthropic.py deleted file mode 100644 index a8134d535..000000000 --- a/examples/foundational/12c-describe-video-anthropic.py +++ /dev/null @@ -1,184 +0,0 @@ -# -# Copyright (c) 2024–2025, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -import os -from typing import Optional - -from dotenv import load_dotenv -from loguru import logger - -from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams -from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3 -from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.audio.vad.vad_analyzer import VADParams -from pipecat.frames.frames import ( - Frame, - LLMContextFrame, - TextFrame, - TTSSpeakFrame, - UserImageRawFrame, - UserImageRequestFrame, -) -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.user_response import UserResponseAggregator -from pipecat.processors.frame_processor import FrameDirection, FrameProcessor -from pipecat.runner.types import RunnerArguments -from pipecat.runner.utils import ( - create_transport, - get_transport_client_id, - maybe_capture_participant_camera, -) -from pipecat.services.anthropic.llm import AnthropicLLMService -from pipecat.services.cartesia.tts import CartesiaTTSService -from pipecat.services.deepgram.stt import DeepgramSTTService -from pipecat.transports.base_transport import BaseTransport, TransportParams -from pipecat.transports.daily.transport import DailyParams - -load_dotenv(override=True) - - -class UserImageRequester(FrameProcessor): - """Converts incoming text into requests for user images.""" - - def __init__(self, participant_id: Optional[str] = None): - super().__init__() - self._participant_id = participant_id - - def set_participant_id(self, participant_id: str): - self._participant_id = participant_id - - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - if self._participant_id and isinstance(frame, TextFrame): - await self.push_frame( - UserImageRequestFrame(self._participant_id, context=frame.text), - FrameDirection.UPSTREAM, - ) - else: - await self.push_frame(frame, direction) - - -class UserImageProcessor(FrameProcessor): - """Converts incoming user images into context frames.""" - - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - if isinstance(frame, UserImageRawFrame): - if frame.request and frame.request.context: - context = LLMContext() - context.add_image_frame_message( - image=frame.image, - text=frame.request.context, - size=frame.size, - format=frame.format, - ) - frame = LLMContextFrame(context) - await self.push_frame(frame) - else: - await self.push_frame(frame, direction) - - -# We store functions so objects (e.g. SileroVADAnalyzer) don't get -# instantiated. The function will be called when the desired transport gets -# selected. -transport_params = { - "daily": lambda: DailyParams( - audio_in_enabled=True, - audio_out_enabled=True, - video_in_enabled=True, - vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)), - turn_analyzer=LocalSmartTurnAnalyzerV3(params=SmartTurnParams()), - ), - "webrtc": lambda: TransportParams( - audio_in_enabled=True, - audio_out_enabled=True, - video_in_enabled=True, - vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)), - turn_analyzer=LocalSmartTurnAnalyzerV3(params=SmartTurnParams()), - ), -} - - -async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): - logger.info(f"Starting bot") - - user_response = UserResponseAggregator() - - # Initialize the image requester without setting the participant ID yet - image_requester = UserImageRequester() - - image_processor = UserImageProcessor() - - stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) - - # Anthropic for vision analysis - anthropic = AnthropicLLMService(api_key=os.getenv("ANTHROPIC_API_KEY")) - - tts = CartesiaTTSService( - api_key=os.getenv("CARTESIA_API_KEY"), - voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady - ) - - pipeline = Pipeline( - [ - transport.input(), - stt, - user_response, - image_requester, - image_processor, - anthropic, - tts, - transport.output(), - ] - ) - - task = PipelineTask( - pipeline, - params=PipelineParams( - enable_metrics=True, - enable_usage_metrics=True, - ), - idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, - ) - - @transport.event_handler("on_client_connected") - async def on_client_connected(transport, client): - logger.info(f"Client connected: {client}") - - await maybe_capture_participant_camera(transport, client) - - # Set the participant ID in the image requester - client_id = get_transport_client_id(transport, client) - image_requester.set_participant_id(client_id) - - # Welcome message - await task.queue_frame(TTSSpeakFrame("Hi there! Feel free to ask me about what I see.")) - - @transport.event_handler("on_client_disconnected") - async def on_client_disconnected(transport, client): - logger.info(f"Client disconnected") - await task.cancel() - - runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) - - await runner.run(task) - - -async def bot(runner_args: RunnerArguments): - """Main bot entry point compatible with Pipecat Cloud.""" - transport = await create_transport(runner_args, transport_params) - await run_bot(transport, runner_args) - - -if __name__ == "__main__": - from pipecat.runner.run import main - - main() diff --git a/examples/foundational/14b-function-calling-anthropic-video.py b/examples/foundational/14d-function-calling-anthropic-video.py similarity index 63% rename from examples/foundational/14b-function-calling-anthropic-video.py rename to examples/foundational/14d-function-calling-anthropic-video.py index 009f59500..a4daed481 100644 --- a/examples/foundational/14b-function-calling-anthropic-video.py +++ b/examples/foundational/14d-function-calling-anthropic-video.py @@ -4,8 +4,6 @@ # SPDX-License-Identifier: BSD 2-Clause License # - -import asyncio import os from dotenv import load_dotenv @@ -39,34 +37,21 @@ from pipecat.transports.daily.transport import DailyParams load_dotenv(override=True) -# Global variable to store the client ID -client_id = "" - - -async def get_weather(params: FunctionCallParams): - location = params.arguments["location"] - await params.result_callback(f"The weather in {location} is currently 72 degrees and sunny.") - - -async def get_image(params: FunctionCallParams): +async def fetch_user_image(params: FunctionCallParams): + user_id = params.arguments["user_id"] question = params.arguments["question"] - logger.debug(f"Requesting image with user_id={client_id}, question={question}") + logger.debug(f"Requesting image with user_id={user_id}, question={question}") - # Request the image frame + # Request the user image frame. Note that this image is associated to a + # function call and will be handled by the LLM assistant aggregators. await params.llm.request_image_frame( - user_id=client_id, + user_id=user_id, function_name=params.function_name, tool_call_id=params.tool_call_id, text_content=question, ) - # Wait a short time for the frame to be processed - await asyncio.sleep(0.5) - - # Return a result to complete the function call - await params.result_callback( - f"I've captured an image from your camera and I'm analyzing what you asked about: {question}" - ) + await params.result_callback(None) # We store functions so objects (e.g. SileroVADAnalyzer) don't get @@ -100,70 +85,32 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady ) - llm = AnthropicLLMService( - api_key=os.getenv("ANTHROPIC_API_KEY"), - model="claude-3-7-sonnet-latest", - params=AnthropicLLMService.InputParams(enable_prompt_caching=True), - ) - llm.register_function("get_weather", get_weather) - llm.register_function("get_image", get_image) + # Anthropic for vision analysis + llm = AnthropicLLMService(api_key=os.getenv("ANTHROPIC_API_KEY")) + llm.register_function("fetch_user_image", fetch_user_image) - weather_function = FunctionSchema( - name="get_weather", - description="Get the current weather", + fetch_image_function = FunctionSchema( + name="fetch_user_image", + description="Called when the user requests a description of their camera feed", properties={ - "location": { + "user_id": { "type": "string", - "description": "The city and state, e.g. San Francisco, CA", + "description": "The ID of the user to grab the image from", }, - }, - required=["location"], - ) - get_image_function = FunctionSchema( - name="get_image", - description="Get an image from the video stream.", - properties={ "question": { "type": "string", - "description": "The question that the user is asking about the image.", - } + "description": "The question that the user is asking about the image", + }, }, - required=["question"], + required=["user_id", "question"], ) - tools = ToolsSchema(standard_tools=[weather_function, get_image_function]) - - system_prompt = """\ -You are a helpful assistant who converses with a user and answers questions. Respond concisely to general questions. - -Your response will be turned into speech so use only simple words and punctuation. - -You have access to two tools: get_weather and get_image. - -You can respond to questions about the weather using the get_weather tool. - -You can answer questions about the user's video stream using the get_image tool. Some examples of phrases that \ -indicate you should use the get_image tool are: -- What do you see? -- What's in the video? -- Can you describe the video? -- Tell me about what you see. -- Tell me something interesting about what you see. -- What's happening in the video? - -If you need to use a tool, simply use the tool. Do not tell the user the tool you are using. Be brief and concise. - """ + tools = ToolsSchema(standard_tools=[fetch_image_function]) messages = [ { "role": "system", - "content": [ - { - "type": "text", - "text": system_prompt, - } - ], + "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way. You are able to describe images from the user camera.", }, - {"role": "user", "content": "Start the conversation by introducing yourself."}, ] context = LLMContext(messages, tools) @@ -173,11 +120,11 @@ If you need to use a tool, simply use the tool. Do not tell the user the tool yo [ transport.input(), # Transport user input stt, # STT - context_aggregator.user(), # User speech to text + context_aggregator.user(), # User responses llm, # LLM tts, # TTS transport.output(), # Transport bot output - context_aggregator.assistant(), # Assistant spoken responses and tool context + context_aggregator.assistant(), # Assistant spoken responses ] ) @@ -196,10 +143,16 @@ If you need to use a tool, simply use the tool. Do not tell the user the tool yo await maybe_capture_participant_camera(transport, client) - global client_id + # Set the participant ID in the image requester client_id = get_transport_client_id(transport, client) # Kick off the conversation. + messages.append( + { + "role": "system", + "content": f"Please introduce yourself to the user. Use '{client_id}' as the user ID during function calls.", + } + ) await task.queue_frames([LLMRunFrame()]) @transport.event_handler("on_client_disconnected") diff --git a/examples/foundational/12d-describe-video-aws.py b/examples/foundational/14d-function-calling-aws-video.py similarity index 58% rename from examples/foundational/12d-describe-video-aws.py rename to examples/foundational/14d-function-calling-aws-video.py index 5436b81ba..78bacbedf 100644 --- a/examples/foundational/12d-describe-video-aws.py +++ b/examples/foundational/14d-function-calling-aws-video.py @@ -5,29 +5,22 @@ # import os -from typing import Optional from dotenv import load_dotenv from loguru import logger +from pipecat.adapters.schemas.function_schema import FunctionSchema +from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3 from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.audio.vad.vad_analyzer import VADParams -from pipecat.frames.frames import ( - Frame, - LLMContextFrame, - TextFrame, - TTSSpeakFrame, - UserImageRawFrame, - UserImageRequestFrame, -) +from pipecat.frames.frames import LLMRunFrame 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.user_response import UserResponseAggregator -from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import ( create_transport, @@ -37,54 +30,28 @@ from pipecat.runner.utils import ( from pipecat.services.aws.llm import AWSBedrockLLMService from pipecat.services.cartesia.tts import CartesiaTTSService from pipecat.services.deepgram.stt import DeepgramSTTService +from pipecat.services.llm_service import FunctionCallParams from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.daily.transport import DailyParams load_dotenv(override=True) -class UserImageRequester(FrameProcessor): - """Converts incoming text into requests for user images.""" +async def fetch_user_image(params: FunctionCallParams): + user_id = params.arguments["user_id"] + question = params.arguments["question"] + logger.debug(f"Requesting image with user_id={user_id}, question={question}") - def __init__(self, participant_id: Optional[str] = None): - super().__init__() - self._participant_id = participant_id + # Request the user image frame. Note that this image is associated to a + # function call and will be handled by the LLM assistant aggregators. + await params.llm.request_image_frame( + user_id=user_id, + function_name=params.function_name, + tool_call_id=params.tool_call_id, + text_content=question, + ) - def set_participant_id(self, participant_id: str): - self._participant_id = participant_id - - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - if self._participant_id and isinstance(frame, TextFrame): - await self.push_frame( - UserImageRequestFrame(self._participant_id, context=frame.text), - FrameDirection.UPSTREAM, - ) - else: - await self.push_frame(frame, direction) - - -class UserImageProcessor(FrameProcessor): - """Converts incoming user images into context frames.""" - - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - if isinstance(frame, UserImageRawFrame): - if frame.request and frame.request.context: - # Note: AWS Bedrock does not yet support the universal LLMContext - context = LLMContext() - context.add_image_frame_message( - image=frame.image, - text=frame.request.context, - size=frame.size, - format=frame.format, - ) - frame = LLMContextFrame(context) - await self.push_frame(frame) - else: - await self.push_frame(frame, direction) + await params.result_callback(None) # We store functions so objects (e.g. SileroVADAnalyzer) don't get @@ -111,17 +78,15 @@ transport_params = { async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): logger.info(f"Starting bot") - user_response = UserResponseAggregator() - - # Initialize the image requester without setting the participant ID yet - image_requester = UserImageRequester() - - image_processor = UserImageProcessor() - stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady + ) + # AWS for vision analysis - aws = AWSBedrockLLMService( + llm = AWSBedrockLLMService( aws_region="us-west-2", model="us.anthropic.claude-3-7-sonnet-20250219-v1:0", # Note: usually, prefer providing latency="optimized" param. @@ -129,22 +94,44 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): # which we need for image input. params=AWSBedrockLLMService.InputParams(temperature=0.8), ) + llm.register_function("fetch_user_image", fetch_user_image) - tts = CartesiaTTSService( - api_key=os.getenv("CARTESIA_API_KEY"), - voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady + fetch_image_function = FunctionSchema( + name="fetch_user_image", + description="Called when the user requests a description of their camera feed", + properties={ + "user_id": { + "type": "string", + "description": "The ID of the user to grab the image from", + }, + "question": { + "type": "string", + "description": "The question that the user is asking about the image", + }, + }, + required=["user_id", "question"], ) + tools = ToolsSchema(standard_tools=[fetch_image_function]) + + messages = [ + { + "role": "system", + "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way. You are able to describe images from the user camera.", + }, + ] + + context = LLMContext(messages, tools) + context_aggregator = LLMContextAggregatorPair(context) pipeline = Pipeline( [ - transport.input(), - stt, - user_response, - image_requester, - image_processor, - aws, - tts, - transport.output(), + transport.input(), # Transport user input + stt, # STT + context_aggregator.user(), # User responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + context_aggregator.assistant(), # Assistant spoken responses ] ) @@ -165,10 +152,15 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): # Set the participant ID in the image requester client_id = get_transport_client_id(transport, client) - image_requester.set_participant_id(client_id) - # Welcome message - await task.queue_frame(TTSSpeakFrame("Hi there! Feel free to ask me about what I see.")) + # Kick off the conversation. + messages.append( + { + "role": "system", + "content": f"Please introduce yourself to the user. Use '{client_id}' as the user ID during function calls.", + } + ) + await task.queue_frames([LLMRunFrame()]) @transport.event_handler("on_client_disconnected") async def on_client_disconnected(transport, client): diff --git a/examples/foundational/12a-describe-video-gemini-flash.py b/examples/foundational/14d-function-calling-gemini-flash-video.py similarity index 55% rename from examples/foundational/12a-describe-video-gemini-flash.py rename to examples/foundational/14d-function-calling-gemini-flash-video.py index 63c1fd677..a669e1e46 100644 --- a/examples/foundational/12a-describe-video-gemini-flash.py +++ b/examples/foundational/14d-function-calling-gemini-flash-video.py @@ -5,29 +5,22 @@ # import os -from typing import Optional from dotenv import load_dotenv from loguru import logger +from pipecat.adapters.schemas.function_schema import FunctionSchema +from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3 from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.audio.vad.vad_analyzer import VADParams -from pipecat.frames.frames import ( - Frame, - LLMContextFrame, - TextFrame, - TTSSpeakFrame, - UserImageRawFrame, - UserImageRequestFrame, -) +from pipecat.frames.frames import LLMRunFrame 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.user_response import UserResponseAggregator -from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import ( create_transport, @@ -37,53 +30,28 @@ from pipecat.runner.utils import ( from pipecat.services.cartesia.tts import CartesiaTTSService from pipecat.services.deepgram.stt import DeepgramSTTService from pipecat.services.google.llm import GoogleLLMService +from pipecat.services.llm_service import FunctionCallParams from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.daily.transport import DailyParams load_dotenv(override=True) -class UserImageRequester(FrameProcessor): - """Converts incoming text into requests for user images.""" +async def fetch_user_image(params: FunctionCallParams): + user_id = params.arguments["user_id"] + question = params.arguments["question"] + logger.debug(f"Requesting image with user_id={user_id}, question={question}") - def __init__(self, participant_id: Optional[str] = None): - super().__init__() - self._participant_id = participant_id + # Request the user image frame. Note that this image is associated to a + # function call and will be handled by the LLM assistant aggregators. + await params.llm.request_image_frame( + user_id=user_id, + function_name=params.function_name, + tool_call_id=params.tool_call_id, + text_content=question, + ) - def set_participant_id(self, participant_id: str): - self._participant_id = participant_id - - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - if self._participant_id and isinstance(frame, TextFrame): - await self.push_frame( - UserImageRequestFrame(self._participant_id, context=frame.text), - FrameDirection.UPSTREAM, - ) - else: - await self.push_frame(frame, direction) - - -class UserImageProcessor(FrameProcessor): - """Converts incoming user images into context frames.""" - - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - if isinstance(frame, UserImageRawFrame): - if frame.request and frame.request.context: - context = LLMContext() - context.add_image_frame_message( - image=frame.image, - text=frame.request.context, - size=frame.size, - format=frame.format, - ) - frame = LLMContextFrame(context) - await self.push_frame(frame) - else: - await self.push_frame(frame, direction) + await params.result_callback(None) # We store functions so objects (e.g. SileroVADAnalyzer) don't get @@ -110,33 +78,53 @@ transport_params = { async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): logger.info(f"Starting bot") - user_response = UserResponseAggregator() - - # Initialize the image requester without setting the participant ID yet - image_requester = UserImageRequester() - - image_processor = UserImageProcessor() - stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) - # Google Gemini model for vision analysis - google = GoogleLLMService(model="gemini-2.0-flash-001", api_key=os.getenv("GOOGLE_API_KEY")) - tts = CartesiaTTSService( api_key=os.getenv("CARTESIA_API_KEY"), voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady ) + # Google Gemini model for vision analysis + llm = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY")) + llm.register_function("fetch_user_image", fetch_user_image) + + fetch_image_function = FunctionSchema( + name="fetch_user_image", + description="Called when the user requests a description of their camera feed", + properties={ + "user_id": { + "type": "string", + "description": "The ID of the user to grab the image from", + }, + "question": { + "type": "string", + "description": "The question that the user is asking about the image", + }, + }, + required=["user_id", "question"], + ) + tools = ToolsSchema(standard_tools=[fetch_image_function]) + + messages = [ + { + "role": "system", + "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way. You are able to describe images from the user camera.", + }, + ] + + context = LLMContext(messages, tools) + context_aggregator = LLMContextAggregatorPair(context) + pipeline = Pipeline( [ - transport.input(), - stt, - user_response, - image_requester, - image_processor, - google, - tts, - transport.output(), + transport.input(), # Transport user input + stt, # STT + context_aggregator.user(), # User responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + context_aggregator.assistant(), # Assistant spoken responses ] ) @@ -157,10 +145,15 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): # Set the participant ID in the image requester client_id = get_transport_client_id(transport, client) - image_requester.set_participant_id(client_id) - # Welcome message - await task.queue_frame(TTSSpeakFrame("Hi there! Feel free to ask me about what I see.")) + # Kick off the conversation. + messages.append( + { + "role": "system", + "content": f"Please introduce yourself to the user. Use '{client_id}' as the user ID during function calls.", + } + ) + await task.queue_frames([LLMRunFrame()]) @transport.event_handler("on_client_disconnected") async def on_client_disconnected(transport, client): diff --git a/examples/foundational/12-describe-video.py b/examples/foundational/14d-function-calling-moondream-video.py similarity index 58% rename from examples/foundational/12-describe-video.py rename to examples/foundational/14d-function-calling-moondream-video.py index eb783ad75..0bad950ed 100644 --- a/examples/foundational/12-describe-video.py +++ b/examples/foundational/14d-function-calling-moondream-video.py @@ -5,28 +5,29 @@ # import os -from typing import Optional from dotenv import load_dotenv from loguru import logger +from pipecat.adapters.schemas.function_schema import FunctionSchema +from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3 from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.audio.vad.vad_analyzer import VADParams from pipecat.frames.frames import ( Frame, - LLMContextFrame, - TextFrame, - TTSSpeakFrame, + LLMRunFrame, UserImageRawFrame, UserImageRequestFrame, + VisionImageRawFrame, ) +from pipecat.pipeline.parallel_pipeline import ParallelPipeline from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineTask from pipecat.processors.aggregators.llm_context import LLMContext -from pipecat.processors.aggregators.user_response import UserResponseAggregator +from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import ( @@ -36,33 +37,27 @@ from pipecat.runner.utils import ( ) from pipecat.services.cartesia.tts import CartesiaTTSService from pipecat.services.deepgram.stt import DeepgramSTTService +from pipecat.services.llm_service import FunctionCallParams from pipecat.services.moondream.vision import MoondreamService +from pipecat.services.openai.llm import OpenAILLMService from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.daily.transport import DailyParams load_dotenv(override=True) -class UserImageRequester(FrameProcessor): - """Converts incoming text into requests for user images.""" +async def fetch_user_image(params: FunctionCallParams): + user_id = params.arguments["user_id"] + question = params.arguments["question"] + logger.debug(f"Requesting image with user_id={user_id}, question={question}") - def __init__(self, participant_id: Optional[str] = None): - super().__init__() - self._participant_id = participant_id + # Request the user image frame frame. In this case we don't use + # `llm.request_image_frame()` because we don't want the LLM to analyze it. + await params.llm.push_frame( + UserImageRequestFrame(user_id=user_id, context=question), FrameDirection.UPSTREAM + ) - def set_participant_id(self, participant_id: str): - self._participant_id = participant_id - - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - if self._participant_id and isinstance(frame, TextFrame): - await self.push_frame( - UserImageRequestFrame(self._participant_id, context=frame.text), - FrameDirection.UPSTREAM, - ) - else: - await self.push_frame(frame, direction) + await params.result_callback(None) class UserImageProcessor(FrameProcessor): @@ -73,14 +68,12 @@ class UserImageProcessor(FrameProcessor): if isinstance(frame, UserImageRawFrame): if frame.request and frame.request.context: - context = LLMContext() - context.add_image_frame_message( + frame = VisionImageRawFrame( image=frame.image, text=frame.request.context, size=frame.size, format=frame.format, ) - frame = LLMContextFrame(context) await self.push_frame(frame) else: await self.push_frame(frame, direction) @@ -110,16 +103,6 @@ transport_params = { async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): logger.info(f"Starting bot") - user_response = UserResponseAggregator() - - # Initialize the image requester without setting the participant ID yet - image_requester = UserImageRequester() - - image_processor = UserImageProcessor() - - # If you run into weird description, try with use_cpu=True - moondream = MoondreamService() - stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) tts = CartesiaTTSService( @@ -127,16 +110,54 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady ) + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) + llm.register_function("fetch_user_image", fetch_user_image) + + fetch_image_function = FunctionSchema( + name="fetch_user_image", + description="Called when the user requests a description of their camera feed", + properties={ + "user_id": { + "type": "string", + "description": "The ID of the user to grab the image from", + }, + "question": { + "type": "string", + "description": "The question that the user is asking about the image", + }, + }, + required=["user_id", "question"], + ) + tools = ToolsSchema(standard_tools=[fetch_image_function]) + + messages = [ + { + "role": "system", + "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way. You are able to describe images from the user camera.", + }, + ] + + context = LLMContext(messages, tools) + context_aggregator = LLMContextAggregatorPair(context) + + # This will get the get the user image frame and push it to the LLM. + image_processor = UserImageProcessor() + + # If you run into weird description, try with use_cpu=True + moondream = MoondreamService() + pipeline = Pipeline( [ - transport.input(), - stt, - user_response, - image_requester, - image_processor, - moondream, - tts, - transport.output(), + transport.input(), # Transport user input + stt, # STT + context_aggregator.user(), # User responses + ParallelPipeline( + [llm], # LLM + [image_processor, moondream], + ), + tts, # TTS + transport.output(), # Transport bot output + context_aggregator.assistant(), # Assistant spoken responses ] ) @@ -153,10 +174,15 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): # Set the participant ID in the image requester client_id = get_transport_client_id(transport, client) - image_requester.set_participant_id(client_id) - # Welcome message - await task.queue_frame(TTSSpeakFrame("Hi there! Feel free to ask me about what I see.")) + # Kick off the conversation. + messages.append( + { + "role": "system", + "content": f"Please introduce yourself to the user. Use '{client_id}' as the user ID during function calls.", + } + ) + await task.queue_frames([LLMRunFrame()]) @transport.event_handler("on_client_disconnected") async def on_client_disconnected(transport, client): diff --git a/examples/foundational/14d-function-calling-video.py b/examples/foundational/14d-function-calling-openai-video.py similarity index 63% rename from examples/foundational/14d-function-calling-video.py rename to examples/foundational/14d-function-calling-openai-video.py index 48cf95ee9..c6320b1ec 100644 --- a/examples/foundational/14d-function-calling-video.py +++ b/examples/foundational/14d-function-calling-openai-video.py @@ -5,7 +5,6 @@ # -import asyncio import os from dotenv import load_dotenv @@ -39,34 +38,21 @@ from pipecat.transports.daily.transport import DailyParams load_dotenv(override=True) -# Global variable to store the client ID -client_id = "" - - -async def get_weather(params: FunctionCallParams): - location = params.arguments["location"] - await params.result_callback(f"The weather in {location} is currently 72 degrees and sunny.") - - -async def get_image(params: FunctionCallParams): +async def fetch_user_image(params: FunctionCallParams): + user_id = params.arguments["user_id"] question = params.arguments["question"] - logger.debug(f"Requesting image with user_id={client_id}, question={question}") + logger.debug(f"Requesting image with user_id={user_id}, question={question}") - # Request the image frame + # Request the user image frame. Note that this image is associated to a + # function call and will be handled by the LLM assistant aggregators. await params.llm.request_image_frame( - user_id=client_id, + user_id=user_id, function_name=params.function_name, tool_call_id=params.tool_call_id, text_content=question, ) - # Wait a short time for the frame to be processed - await asyncio.sleep(0.5) - - # Return a result to complete the function call - await params.result_callback( - f"I've captured an image from your camera and I'm analyzing what you asked about: {question}" - ) + await params.result_callback(None) # We store functions so objects (e.g. SileroVADAnalyzer) don't get @@ -101,58 +87,30 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): ) llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) - llm.register_function("get_weather", get_weather) - llm.register_function("get_image", get_image) + llm.register_function("fetch_user_image", fetch_user_image) - weather_function = FunctionSchema( - name="get_weather", - description="Get the current weather", + fetch_image_function = FunctionSchema( + name="fetch_user_image", + description="Called when the user requests a description of their camera feed", properties={ - "location": { + "user_id": { "type": "string", - "description": "The city and state, e.g. San Francisco, CA", + "description": "The ID of the user to grab the image from", }, - "format": { - "type": "string", - "enum": ["celsius", "fahrenheit"], - "description": "The temperature unit to use. Infer this from the user's location.", - }, - }, - required=["location"], - ) - get_image_function = FunctionSchema( - name="get_image", - description="Get an image from the video stream.", - properties={ "question": { "type": "string", - "description": "The question that the user is asking about the image.", - } + "description": "The question that the user is asking about the image", + }, }, - required=["question"], + required=["user_id", "question"], ) - tools = ToolsSchema(standard_tools=[weather_function, get_image_function]) + tools = ToolsSchema(standard_tools=[fetch_image_function]) - system_prompt = """\ -You are a helpful assistant who converses with a user and answers questions. Respond concisely to general questions. - -Your response will be turned into speech so use only simple words and punctuation. - -You have access to two tools: get_weather and get_image. - -You can respond to questions about the weather using the get_weather tool. - -You can answer questions about the user's video stream using the get_image tool. Some examples of phrases that \ -indicate you should use the get_image tool are: -- What do you see? -- What's in the video? -- Can you describe the video? -- Tell me about what you see. -- Tell me something interesting about what you see. -- What's happening in the video? -""" messages = [ - {"role": "system", "content": system_prompt}, + { + "role": "system", + "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way. You are able to describe images from the user camera.", + }, ] context = LLMContext(messages, tools) @@ -160,13 +118,13 @@ indicate you should use the get_image tool are: pipeline = Pipeline( [ - transport.input(), - stt, - context_aggregator.user(), - llm, - tts, - transport.output(), - context_aggregator.assistant(), + transport.input(), # Transport user input + stt, # STT + context_aggregator.user(), # User responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + context_aggregator.assistant(), # Assistant spoken responses ] ) @@ -185,10 +143,15 @@ indicate you should use the get_image tool are: await maybe_capture_participant_camera(transport, client) - global client_id client_id = get_transport_client_id(transport, client) # Kick off the conversation. + messages.append( + { + "role": "system", + "content": f"Please introduce yourself to the user. Use '{client_id}' as the user ID during function calls.", + } + ) await task.queue_frames([LLMRunFrame()]) @transport.event_handler("on_client_disconnected")