diff --git a/examples/foundational/12-describe-image-openai-responses.py b/examples/foundational/12-describe-image-openai-responses.py new file mode 100644 index 000000000..a3c113cb2 --- /dev/null +++ b/examples/foundational/12-describe-image-openai-responses.py @@ -0,0 +1,139 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + + +import os + +from dotenv import load_dotenv +from loguru import logger +from PIL import Image + +from pipecat.audio.vad.silero import SileroVADAnalyzer +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.llm_response_universal import ( + LLMContextAggregatorPair, + LLMUserAggregatorParams, +) +from pipecat.runner.types import RunnerArguments +from pipecat.runner.utils import create_transport +from pipecat.services.cartesia.tts import CartesiaTTSService +from pipecat.services.deepgram.stt import DeepgramSTTService +from pipecat.services.openai.responses.llm import OpenAIResponsesLLMService +from pipecat.transports.base_transport import BaseTransport, TransportParams +from pipecat.transports.daily.transport import DailyParams + +load_dotenv(override=True) + + +# We use lambdas to defer transport parameter creation until the transport +# type is selected at runtime. +transport_params = { + "daily": lambda: DailyParams( + audio_in_enabled=True, + audio_out_enabled=True, + ), + "webrtc": lambda: TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + ), +} + + +async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): + logger.info(f"Starting bot") + + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + settings=CartesiaTTSService.Settings( + voice="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady + ), + ) + + llm = OpenAIResponsesLLMService( + api_key=os.getenv("OPENAI_API_KEY"), + settings=OpenAIResponsesLLMService.Settings( + system_instruction="You are a helpful assistant in a voice conversation. Your responses will be spoken aloud, so avoid emojis, bullet points, or other formatting that can't be spoken. Respond to what the user said in a creative, helpful, and brief way. You are also able to describe images.", + ), + ) + + context = LLMContext() + user_aggregator, assistant_aggregator = LLMContextAggregatorPair( + context, + user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()), + ) + + pipeline = Pipeline( + [ + transport.input(), # Transport user input + stt, # STT + user_aggregator, # User responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + assistant_aggregator, # Assistant spoken responses + ] + ) + + 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") + + if not runner_args.body: + script_dir = os.path.dirname(__file__) + runner_args.body = { + "image_path": os.path.join(script_dir, "assets", "cat.jpg"), + "question": "Describe this image", + } + + image_path = runner_args.body["image_path"] + question = runner_args.body["question"] + + # Kick off the conversation. + image = Image.open(image_path) + message = await LLMContext.create_image_message( + image=image.tobytes(), + format="RGB", + size=image.size, + text=question, + ) + context.add_message(message) + await task.queue_frames([LLMRunFrame()]) + + @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/14d-function-calling-openai-responses-video.py b/examples/foundational/14d-function-calling-openai-responses-video.py new file mode 100644 index 000000000..440c51cc1 --- /dev/null +++ b/examples/foundational/14d-function-calling-openai-responses-video.py @@ -0,0 +1,195 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + + +import os + +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.vad.silero import SileroVADAnalyzer +from pipecat.frames.frames import LLMRunFrame, TTSSpeakFrame, 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.llm_response_universal import ( + LLMContextAggregatorPair, + LLMUserAggregatorParams, +) +from pipecat.processors.frame_processor import FrameDirection +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.llm_service import FunctionCallParams +from pipecat.services.openai.responses.llm import OpenAIResponsesLLMService +from pipecat.transports.base_transport import BaseTransport, TransportParams +from pipecat.transports.daily.transport import DailyParams + +load_dotenv(override=True) + + +async def fetch_user_image(params: FunctionCallParams): + """Fetch the user image and push it to the LLM. + + When called, this function pushes a UserImageRequestFrame upstream to the + transport. As a result, the transport will request the user image and push a + UserImageRawFrame downstream which will be added to the context by the LLM + assistant aggregator. The result_callback will be invoked once the image is + retrieved and processed. + """ + user_id = params.arguments["user_id"] + question = params.arguments["question"] + logger.debug(f"Requesting image with user_id={user_id}, question={question}") + + # Request a user image frame and indicate that it should be added to the + # context. Also associate it to the function call. Pass the result_callback + # so it can be invoked when the image is actually retrieved. + await params.llm.push_frame( + UserImageRequestFrame( + user_id=user_id, + text=question, + append_to_context=True, + function_name=params.function_name, + tool_call_id=params.tool_call_id, + result_callback=params.result_callback, + ), + FrameDirection.UPSTREAM, + ) + + +# We use lambdas to defer transport parameter creation until the transport +# type is selected at runtime. +transport_params = { + "daily": lambda: DailyParams( + audio_in_enabled=True, + audio_out_enabled=True, + video_in_enabled=True, + ), + "webrtc": lambda: TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + video_in_enabled=True, + ), +} + + +async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): + logger.info(f"Starting bot") + + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + settings=CartesiaTTSService.Settings( + voice="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady + ), + ) + + llm = OpenAIResponsesLLMService( + api_key=os.getenv("OPENAI_API_KEY"), + settings=OpenAIResponsesLLMService.Settings( + system_instruction="You are a helpful assistant in a voice conversation. Your responses will be spoken aloud, so avoid emojis, bullet points, or other formatting that can't be spoken. Respond to what the user said in a creative, helpful, and brief way. You are able to describe images from the user camera.", + ), + ) + llm.register_function("fetch_user_image", fetch_user_image) + + @llm.event_handler("on_function_calls_started") + async def on_function_calls_started(service, function_calls): + await tts.queue_frame(TTSSpeakFrame("Let me check on that.", append_to_context=False)) + + 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]) + + context = LLMContext(tools=tools) + user_aggregator, assistant_aggregator = LLMContextAggregatorPair( + context, + user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()), + ) + + pipeline = Pipeline( + [ + transport.input(), # Transport user input + stt, # STT + user_aggregator, # User responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + assistant_aggregator, # Assistant spoken responses + ] + ) + + 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") + + await maybe_capture_participant_camera(transport, client) + + client_id = get_transport_client_id(transport, client) + + # Kick off the conversation. + context.add_message( + { + "role": "user", + "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): + logger.info(f"Client disconnected") + await task.cancel() + + @tts.event_handler("on_tts_request") + async def on_tts_request(tts, context_id: str, text: str): + logger.debug(f"On TTS request: {context_id}: {text}") + + 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/scripts/evals/run-release-evals.py b/scripts/evals/run-release-evals.py index 9671703ba..19492ba62 100644 --- a/scripts/evals/run-release-evals.py +++ b/scripts/evals/run-release-evals.py @@ -154,6 +154,7 @@ TESTS_07 = [ TESTS_12 = [ ("12-describe-image-openai.py", EVAL_VISION_IMAGE(eval_speaks_first=True)), + ("12-describe-image-openai-responses.py", EVAL_VISION_IMAGE(eval_speaks_first=True)), ("12a-describe-image-anthropic.py", EVAL_VISION_IMAGE(eval_speaks_first=True)), ("12b-describe-image-aws.py", EVAL_VISION_IMAGE(eval_speaks_first=True)), ("12c-describe-image-gemini-flash.py", EVAL_VISION_IMAGE(eval_speaks_first=True)), @@ -193,6 +194,7 @@ TESTS_14 = [ ("14d-function-calling-gemini-flash-video.py", EVAL_VISION_CAMERA), ("14d-function-calling-moondream-video.py", EVAL_VISION_CAMERA), ("14d-function-calling-openai-video.py", EVAL_VISION_CAMERA), + ("14d-function-calling-openai-responses-video.py", EVAL_VISION_CAMERA), # Currently not working. # ("14c-function-calling-together.py", EVAL_WEATHER), # ("14l-function-calling-deepseek.py", EVAL_WEATHER),