diff --git a/examples/foundational/14-function-calling.py b/examples/foundational/14-function-calling.py index 35a02743b..e1432b6ca 100644 --- a/examples/foundational/14-function-calling.py +++ b/examples/foundational/14-function-calling.py @@ -11,7 +11,7 @@ import sys from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner -from pipecat.pipeline.task import PipelineTask +from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.services.cartesia import CartesiaTTSService from pipecat.services.openai import OpenAILLMContext, OpenAILLMService from pipecat.transports.services.daily import DailyParams, DailyTransport @@ -115,13 +115,21 @@ async def main(): ] ) - task = PipelineTask(pipeline) + task = PipelineTask( + pipeline, + PipelineParams( + allow_interruptions=True, + enable_metrics=True, + enable_usage_metrics=True, + report_only_initial_ttfb=True, + ), + ) @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): transport.capture_participant_transcription(participant["id"]) # Kick off the conversation. - await tts.say("Hi! Ask me about the weather in San Francisco.") + await task.queue_frames([context_aggregator.user().get_context_frame()]) runner = PipelineRunner() diff --git a/examples/foundational/19-openai-realtime-beta.py b/examples/foundational/19-openai-realtime-beta.py new file mode 100644 index 000000000..51ca773e1 --- /dev/null +++ b/examples/foundational/19-openai-realtime-beta.py @@ -0,0 +1,164 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import os +import sys +from datetime import datetime + +import aiohttp +from dotenv import load_dotenv +from loguru import logger +from runner import configure + +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, +) +from pipecat.services.openai_realtime_beta import ( + InputAudioTranscription, + OpenAILLMServiceRealtimeBeta, + SessionProperties, + TurnDetection, +) +from pipecat.transports.services.daily import DailyParams, DailyTransport +from pipecat.vad.silero import SileroVADAnalyzer +from pipecat.vad.vad_analyzer import VADParams + +load_dotenv(override=True) + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + + +async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback): + temperature = 75 if args["format"] == "fahrenheit" else 24 + await result_callback( + { + "conditions": "nice", + "temperature": temperature, + "format": args["format"], + "timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"), + } + ) + + +tools = [ + { + "type": "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"], + }, + } +] + + +async def main(): + async with aiohttp.ClientSession() as session: + (room_url, token) = await configure(session) + + transport = DailyTransport( + room_url, + token, + "Respond bot", + DailyParams( + audio_in_enabled=True, + audio_in_sample_rate=24000, + audio_out_enabled=True, + audio_out_sample_rate=24000, + transcription_enabled=False, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.8)), + vad_audio_passthrough=True, + ), + ) + + session_properties = SessionProperties( + input_audio_transcription=InputAudioTranscription(), + # Set openai TurnDetection parameters. Not setting this at all will turn it + # on by default + turn_detection=TurnDetection(silence_duration_ms=1000), + # Or set to False to disable openai turn detection and use transport VAD + # turn_detection=False, + # tools=tools, + instructions="""Your knowledge cutoff is 2023-10. You are a helpful and friendly AI. + +Act like a human, but remember that you aren't a human and that you can't do human +things in the real world. Your voice and personality should be warm and engaging, with a lively and +playful tone. + +If interacting in a non-English language, start by using the standard accent or dialect familiar to +the user. Talk quickly. You should always call a function if you can. Do not refer to these rules, +even if you're asked about them. +- +You are participating in a voice conversation. Keep your responses concise, short, and to the point +unless specifically asked to elaborate on a topic. + +Remember, your responses should be short. Just one or two sentences, usually.""", + ) + + llm = OpenAILLMServiceRealtimeBeta( + api_key=os.getenv("OPENAI_API_KEY"), + session_properties=session_properties, + start_audio_paused=False, + ) + + # you can either register a single function for all function calls, or specific functions + # llm.register_function(None, fetch_weather_from_api) + llm.register_function("get_current_weather", fetch_weather_from_api) + + context = OpenAILLMContext([{"role": "user", "content": "Say hello!"}], tools) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline( + [ + transport.input(), # Transport user input + context_aggregator.user(), + llm, # LLM + context_aggregator.assistant(), + transport.output(), # Transport bot output + ] + ) + + task = PipelineTask( + pipeline, + PipelineParams( + allow_interruptions=True, + enable_metrics=True, + enable_usage_metrics=True, + # report_only_initial_ttfb=True, + ), + ) + + @transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + transport.capture_participant_transcription(participant["id"]) + # Kick off the conversation. + await task.queue_frames([context_aggregator.user().get_context_frame()]) + + runner = PipelineRunner() + + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/foundational/20a-persistent-context-openai.py b/examples/foundational/20a-persistent-context-openai.py new file mode 100644 index 000000000..5767d6dbd --- /dev/null +++ b/examples/foundational/20a-persistent-context-openai.py @@ -0,0 +1,236 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import glob +import json +import os +import sys +from datetime import datetime + +import aiohttp +from dotenv import load_dotenv +from loguru import logger +from runner import configure + +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, +) +from pipecat.services.openai import OpenAILLMService +from pipecat.services.cartesia import CartesiaTTSService + +from pipecat.transports.services.daily import DailyParams, DailyTransport +from pipecat.vad.silero import SileroVADAnalyzer +from pipecat.vad.vad_analyzer import VADParams + +load_dotenv(override=True) + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + +BASE_FILENAME = "/tmp/pipecat_conversation_" +tts = None + + +async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback): + temperature = 75 if args["format"] == "fahrenheit" else 24 + await result_callback( + { + "conditions": "nice", + "temperature": temperature, + "format": args["format"], + "timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"), + } + ) + + +async def get_saved_conversation_filenames( + function_name, tool_call_id, args, llm, context, result_callback +): + # Construct the full pattern including the BASE_FILENAME + full_pattern = f"{BASE_FILENAME}*.json" + + # Use glob to find all matching files + matching_files = glob.glob(full_pattern) + logger.debug(f"matching files: {matching_files}") + + await result_callback({"filenames": matching_files}) + + +async def save_conversation(function_name, tool_call_id, args, llm, context, result_callback): + timestamp = datetime.now().strftime("%Y-%m-%d_%H:%M:%S") + filename = f"{BASE_FILENAME}{timestamp}.json" + logger.debug(f"writing conversation to {filename}\n{json.dumps(context.messages, indent=4)}") + try: + with open(filename, "w") as file: + messages = context.get_messages_for_persistent_storage() + # remove the last message, which is the instruction we just gave to save the conversation + messages.pop() + json.dump(messages, file, indent=2) + await result_callback({"success": True}) + except Exception as e: + await result_callback({"success": False, "error": str(e)}) + + +async def load_conversation(function_name, tool_call_id, args, llm, context, result_callback): + global tts + filename = args["filename"] + logger.debug(f"loading conversation from {filename}") + try: + with open(filename, "r") as file: + context.set_messages(json.load(file)) + logger.debug( + f"loaded conversation from {filename}\n{json.dumps(context.messages, indent=4)}" + ) + await tts.say("Ok, I've loaded that conversation.") + except Exception as e: + await result_callback({"success": False, "error": str(e)}) + + +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.", + }, +] +tools = [ + { + "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"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "save_conversation", + "description": "Save the current conversatione. Use this function to persist the current conversation to external storage.", + "parameters": { + "type": "object", + "properties": {}, + "required": [], + }, + }, + }, + { + "type": "function", + "function": { + "name": "get_saved_conversation_filenames", + "description": "Get a list of saved conversation histories. Returns a list of filenames. Each filename includes a date and timestamp. Each file is conversation history that can be loaded into this session.", + "parameters": { + "type": "object", + "properties": {}, + "required": [], + }, + }, + }, + { + "type": "function", + "function": { + "name": "load_conversation", + "description": "Load a conversation history. Use this function to load a conversation history into the current session.", + "parameters": { + "type": "object", + "properties": { + "filename": { + "type": "string", + "description": "The filename of the conversation history to load.", + } + }, + "required": ["filename"], + }, + }, + }, +] + + +async def main(): + global tts + async with aiohttp.ClientSession() as session: + (room_url, token) = await configure(session) + + transport = DailyTransport( + room_url, + token, + "Respond bot", + DailyParams( + audio_out_enabled=True, + transcription_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.8)), + ), + ) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady + ) + + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + + # you can either register a single function for all function calls, or specific functions + # llm.register_function(None, fetch_weather_from_api) + llm.register_function("get_current_weather", fetch_weather_from_api) + llm.register_function("save_conversation", save_conversation) + llm.register_function("get_saved_conversation_filenames", get_saved_conversation_filenames) + llm.register_function("load_conversation", load_conversation) + + context = OpenAILLMContext(messages, tools) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline( + [ + transport.input(), # Transport user input + context_aggregator.user(), + llm, # LLM + tts, + context_aggregator.assistant(), + transport.output(), # Transport bot output + ] + ) + + task = PipelineTask( + pipeline, + PipelineParams( + allow_interruptions=True, + enable_metrics=True, + enable_usage_metrics=True, + # report_only_initial_ttfb=True, + ), + ) + + @transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + transport.capture_participant_transcription(participant["id"]) + # Kick off the conversation. + await task.queue_frames([context_aggregator.user().get_context_frame()]) + + runner = PipelineRunner() + + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/foundational/20b-persistent-context-openai-realtime.py b/examples/foundational/20b-persistent-context-openai-realtime.py new file mode 100644 index 000000000..27a2b7de6 --- /dev/null +++ b/examples/foundational/20b-persistent-context-openai-realtime.py @@ -0,0 +1,262 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import glob +import json +import os +import sys +from datetime import datetime + +import aiohttp +from dotenv import load_dotenv +from loguru import logger +from runner import configure + +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, +) +from pipecat.services.openai_realtime_beta import ( + InputAudioTranscription, + OpenAILLMServiceRealtimeBeta, + SessionProperties, + TurnDetection, +) +from pipecat.transports.services.daily import DailyParams, DailyTransport +from pipecat.vad.silero import SileroVADAnalyzer +from pipecat.vad.vad_analyzer import VADParams + +load_dotenv(override=True) + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + +BASE_FILENAME = "/tmp/pipecat_conversation_" + + +async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback): + temperature = 75 if args["format"] == "fahrenheit" else 24 + await result_callback( + { + "conditions": "nice", + "temperature": temperature, + "format": args["format"], + "timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"), + } + ) + + +async def get_saved_conversation_filenames( + function_name, tool_call_id, args, llm, context, result_callback +): + # Construct the full pattern including the BASE_FILENAME + full_pattern = f"{BASE_FILENAME}*.json" + + # Use glob to find all matching files + matching_files = glob.glob(full_pattern) + logger.debug(f"matching files: {matching_files}") + + await result_callback({"filenames": matching_files}) + + +# async def get_saved_conversation_filenames( +# function_name, tool_call_id, args, llm, context, result_callback +# ): +# pattern = re.compile(re.escape(BASE_FILENAME) + "\\d{8}_\\d{6}\\.json$") +# matching_files = [] + +# for filename in os.listdir("."): +# if pattern.match(filename): +# matching_files.append(filename) + +# await result_callback({"filenames": matching_files}) + + +async def save_conversation(function_name, tool_call_id, args, llm, context, result_callback): + timestamp = datetime.now().strftime("%Y-%m-%d_%H:%M:%S") + filename = f"{BASE_FILENAME}{timestamp}.json" + logger.debug(f"writing conversation to {filename}\n{json.dumps(context.messages, indent=4)}") + try: + with open(filename, "w") as file: + messages = context.get_messages_for_persistent_storage() + # remove the last message, which is the instruction we just gave to save the conversation + messages.pop() + json.dump(messages, file, indent=2) + await result_callback({"success": True}) + except Exception as e: + await result_callback({"success": False, "error": str(e)}) + + +async def load_conversation(function_name, tool_call_id, args, llm, context, result_callback): + async def _reset(): + filename = args["filename"] + logger.debug(f"loading conversation from {filename}") + try: + with open(filename, "r") as file: + context.set_messages(json.load(file)) + await llm.reset_conversation() + await llm._create_response() + except Exception as e: + await result_callback({"success": False, "error": str(e)}) + + asyncio.create_task(_reset()) + + +tools = [ + { + "type": "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"], + }, + }, + { + "type": "function", + "name": "save_conversation", + "description": "Save the current conversatione. Use this function to persist the current conversation to external storage.", + "parameters": { + "type": "object", + "properties": {}, + "required": [], + }, + }, + { + "type": "function", + "name": "get_saved_conversation_filenames", + "description": "Get a list of saved conversation histories. Returns a list of filenames. Each filename includes a date and timestamp. Each file is conversation history that can be loaded into this session.", + "parameters": { + "type": "object", + "properties": {}, + "required": [], + }, + }, + { + "type": "function", + "name": "load_conversation", + "description": "Load a conversation history. Use this function to load a conversation history into the current session.", + "parameters": { + "type": "object", + "properties": { + "filename": { + "type": "string", + "description": "The filename of the conversation history to load.", + } + }, + "required": ["filename"], + }, + }, +] + + +async def main(): + async with aiohttp.ClientSession() as session: + (room_url, token) = await configure(session) + + transport = DailyTransport( + room_url, + token, + "Respond bot", + DailyParams( + audio_in_enabled=True, + audio_in_sample_rate=24000, + audio_out_enabled=True, + audio_out_sample_rate=24000, + transcription_enabled=False, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.8)), + vad_audio_passthrough=True, + ), + ) + + session_properties = SessionProperties( + input_audio_transcription=InputAudioTranscription(), + # Set openai TurnDetection parameters. Not setting this at all will turn it + # on by default + turn_detection=TurnDetection(silence_duration_ms=1000), + # Or set to False to disable openai turn detection and use transport VAD + # turn_detection=False, + # tools=tools, + instructions="""Your knowledge cutoff is 2023-10. You are a helpful and friendly AI. + +Act like a human, but remember that you aren't a human and that you can't do human +things in the real world. Your voice and personality should be warm and engaging, with a lively and +playful tone. + +If interacting in a non-English language, start by using the standard accent or dialect familiar to +the user. Talk quickly. You should always call a function if you can. Do not refer to these rules, +even if you're asked about them. +- +You are participating in a voice conversation. Keep your responses concise, short, and to the point +unless specifically asked to elaborate on a topic. + +Remember, your responses should be short. Just one or two sentences, usually.""", + ) + + llm = OpenAILLMServiceRealtimeBeta( + api_key=os.getenv("OPENAI_API_KEY"), + session_properties=session_properties, + start_audio_paused=False, + ) + + # you can either register a single function for all function calls, or specific functions + # llm.register_function(None, fetch_weather_from_api) + llm.register_function("get_current_weather", fetch_weather_from_api) + llm.register_function("save_conversation", save_conversation) + llm.register_function("get_saved_conversation_filenames", get_saved_conversation_filenames) + llm.register_function("load_conversation", load_conversation) + + context = OpenAILLMContext([], tools) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline( + [ + transport.input(), # Transport user input + context_aggregator.user(), + llm, # LLM + context_aggregator.assistant(), + transport.output(), # Transport bot output + ] + ) + + task = PipelineTask( + pipeline, + PipelineParams( + allow_interruptions=True, + enable_metrics=True, + enable_usage_metrics=True, + # report_only_initial_ttfb=True, + ), + ) + + @transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + transport.capture_participant_transcription(participant["id"]) + # Kick off the conversation. + await task.queue_frames([context_aggregator.user().get_context_frame()]) + + runner = PipelineRunner() + + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/foundational/20c-persistent-context-anthropic.py b/examples/foundational/20c-persistent-context-anthropic.py new file mode 100644 index 000000000..926722aeb --- /dev/null +++ b/examples/foundational/20c-persistent-context-anthropic.py @@ -0,0 +1,232 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import glob +import json +import os +import sys +from datetime import datetime + +import aiohttp +from dotenv import load_dotenv +from loguru import logger +from runner import configure + +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, +) +from pipecat.services.cartesia import CartesiaTTSService +from pipecat.services.anthropic import AnthropicLLMService + +from pipecat.transports.services.daily import DailyParams, DailyTransport +from pipecat.vad.silero import SileroVADAnalyzer +from pipecat.vad.vad_analyzer import VADParams + +load_dotenv(override=True) + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + +BASE_FILENAME = "/tmp/pipecat_conversation_" +tts = None + + +async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback): + temperature = 75 if args["format"] == "fahrenheit" else 24 + await result_callback( + { + "conditions": "nice", + "temperature": temperature, + "format": args["format"], + "timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"), + } + ) + + +async def get_saved_conversation_filenames( + function_name, tool_call_id, args, llm, context, result_callback +): + # Construct the full pattern including the BASE_FILENAME + full_pattern = f"{BASE_FILENAME}*.json" + + # Use glob to find all matching files + matching_files = glob.glob(full_pattern) + logger.debug(f"matching files: {matching_files}") + + await result_callback({"filenames": matching_files}) + + +async def save_conversation(function_name, tool_call_id, args, llm, context, result_callback): + timestamp = datetime.now().strftime("%Y-%m-%d_%H:%M:%S") + filename = f"{BASE_FILENAME}{timestamp}.json" + logger.debug(f"writing conversation to {filename}\n{json.dumps(context.messages, indent=4)}") + try: + with open(filename, "w") as file: + # todo: extract 'system' into the first message in the list + messages = context.get_messages_for_persistent_storage() + # remove the last message, which is the instruction we just gave to save the conversation + messages.pop() + json.dump(messages, file, indent=2) + await result_callback({"success": True}) + except Exception as e: + await result_callback({"success": False, "error": str(e)}) + + +async def load_conversation(function_name, tool_call_id, args, llm, context, result_callback): + global tts + filename = args["filename"] + logger.debug(f"loading conversation from {filename}") + try: + with open(filename, "r") as file: + context.set_messages(json.load(file)) + logger.debug( + f"loaded conversation from {filename}\n{json.dumps(context.messages, indent=4)}" + ) + await tts.say("Ok, I've loaded that conversation.") + except Exception as e: + await result_callback({"success": False, "error": str(e)}) + + +# Test message munging ... +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.", + }, + {"role": "user", "content": ""}, + {"role": "assistant", "content": []}, + {"role": "user", "content": "Tell me"}, + {"role": "user", "content": "a joke"}, +] +tools = [ + { + "name": "get_current_weather", + "description": "Get the current weather", + "input_schema": { + "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"], + }, + }, + { + "name": "save_conversation", + "description": "Save the current conversation. Use this function to persist the current conversation to external storage.", + "input_schema": { + "type": "object", + "properties": {}, + "required": [], + }, + }, + { + "name": "get_saved_conversation_filenames", + "description": "Get a list of saved conversation histories. Returns a list of filenames. Each filename includes a date and timestamp. Each file is conversation history that can be loaded into this session.", + "input_schema": { + "type": "object", + "properties": {}, + "required": [], + }, + }, + { + "name": "load_conversation", + "description": "Load a conversation history. Use this function to load a conversation history into the current session.", + "input_schema": { + "type": "object", + "properties": { + "filename": { + "type": "string", + "description": "The filename of the conversation history to load.", + } + }, + "required": ["filename"], + }, + }, +] + + +async def main(): + global tts + async with aiohttp.ClientSession() as session: + (room_url, token) = await configure(session) + + transport = DailyTransport( + room_url, + token, + "Respond bot", + DailyParams( + audio_out_enabled=True, + transcription_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.8)), + ), + ) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady + ) + + llm = AnthropicLLMService( + api_key=os.getenv("ANTHROPIC_API_KEY"), model="claude-3-5-sonnet-20240620" + ) + + # you can either register a single function for all function calls, or specific functions + # llm.register_function(None, fetch_weather_from_api) + llm.register_function("get_current_weather", fetch_weather_from_api) + llm.register_function("save_conversation", save_conversation) + llm.register_function("get_saved_conversation_filenames", get_saved_conversation_filenames) + llm.register_function("load_conversation", load_conversation) + + context = OpenAILLMContext(messages, tools) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline( + [ + transport.input(), # Transport user input + context_aggregator.user(), + llm, # LLM + tts, + context_aggregator.assistant(), + transport.output(), # Transport bot output + ] + ) + + task = PipelineTask( + pipeline, + PipelineParams( + allow_interruptions=True, + enable_metrics=True, + enable_usage_metrics=True, + # report_only_initial_ttfb=True, + ), + ) + + @transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + transport.capture_participant_transcription(participant["id"]) + # Kick off the conversation. + await task.queue_frames([context_aggregator.user().get_context_frame()]) + + runner = PipelineRunner() + + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/pyproject.toml b/pyproject.toml index 1aaf0a5e2..45ccc2002 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -54,11 +54,11 @@ livekit = [ "livekit~=0.13.1", "tenacity~=9.0.0" ] lmnt = [ "lmnt~=1.1.4" ] local = [ "pyaudio~=0.2.14" ] moondream = [ "einops~=0.8.0", "timm~=1.0.8", "transformers~=4.44.0" ] -openai = [ "openai~=1.37.2" ] +openai = [ "openai~=1.50.2", "websockets~=12.0", "python-deepcompare~=1.0.1" ] openpipe = [ "openpipe~=4.24.0" ] playht = [ "pyht~=0.0.28" ] silero = [ "onnxruntime>=1.16.1" ] -together = [ "together~=1.2.7" ] +together = [ "openai~=1.50.2" ] websocket = [ "websockets~=12.0", "fastapi~=0.115.0" ] whisper = [ "faster-whisper~=1.0.3" ] xtts = [ "resampy~=0.4.3" ] diff --git a/src/pipecat/processors/aggregators/openai_llm_context.py b/src/pipecat/processors/aggregators/openai_llm_context.py index a8db23d07..9ea5c3e63 100644 --- a/src/pipecat/processors/aggregators/openai_llm_context.py +++ b/src/pipecat/processors/aggregators/openai_llm_context.py @@ -132,6 +132,23 @@ class OpenAILLMContext: msgs.append(msg) return json.dumps(msgs) + def from_standard_message(self, message): + return message + + # convert a message in this LLM's format to one or more messages in OpenAI format + def to_standard_messages(self, obj) -> list: + return [obj] + + def get_messages_for_initializing_history(self): + return self._messages + + def get_messages_for_persistent_storage(self): + messages = [] + for m in self._messages: + standard_messages = self.to_standard_messages(m) + messages.extend(standard_messages) + return messages + def set_tool_choice(self, tool_choice: ChatCompletionToolChoiceOptionParam | NotGiven): self._tool_choice = tool_choice @@ -168,6 +185,7 @@ class OpenAILLMContext: llm: FrameProcessor, run_llm: bool = True, ) -> None: + logger.debug(f"Calling function {function_name} with arguments {arguments}") # Push a SystemFrame downstream. This frame will let our assistant context aggregator # know that we are in the middle of a function call. Some contexts/aggregators may # not need this. But some definitely do (Anthropic, for example). diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index a5be18c12..f1e7ea023 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -486,7 +486,7 @@ class RTVIBotLLMTextProcessor(RTVIFrameProcessor): await self.push_frame(frame, direction) - if isinstance(frame, TextFrame): + if type(frame) is TextFrame: await self._handle_text(frame) async def _handle_text(self, frame: TextFrame): @@ -503,7 +503,7 @@ class RTVIBotTTSTextProcessor(RTVIFrameProcessor): await self.push_frame(frame, direction) - if isinstance(frame, TextFrame): + if type(frame) is TextFrame: await self._handle_text(frame) async def _handle_text(self, frame: TextFrame): diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index b5ef17b48..7ca043e97 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -47,6 +47,7 @@ class AIService(FrameProcessor): super().__init__(**kwargs) self._model_name: str = "" self._settings: Dict[str, Any] = {} + self._session_properties: Dict[str, Any] = {} @property def model_name(self) -> str: @@ -66,11 +67,44 @@ class AIService(FrameProcessor): pass async def _update_settings(self, settings: Dict[str, Any]): + from pipecat.services.openai_realtime_beta.events import ( + SessionProperties, + ) + for key, value in settings.items(): + print("Update request for:", key, value) + if key in self._settings: - logger.debug(f"Updating setting {key} to: [{value}] for {self.name}") + logger.debug(f"Updating LLM setting {key} to: [{value}]") self._settings[key] = value + elif key in SessionProperties.model_fields: + print("Attempting to update", key, value) + + try: + from pipecat.services.openai_realtime_beta.events import ( + TurnDetection, + ) + + if isinstance(self._session_properties, SessionProperties): + current_properties = self._session_properties + else: + current_properties = SessionProperties(**self._session_properties) + + if key == "turn_detection" and isinstance(value, dict): + turn_detection = TurnDetection(**value) + setattr(current_properties, key, turn_detection) + else: + setattr(current_properties, key, value) + + validated_properties = SessionProperties.model_validate( + current_properties.model_dump() + ) + logger.debug(f"Updating LLM setting {key} to: [{value}]") + self._session_properties = validated_properties.model_dump() + except Exception as e: + logger.warning(f"Unexpected error updating session property {key}: {e}") elif key == "model": + logger.debug(f"Updating LLM setting {key} to: [{value}]") self.set_model_name(value) else: logger.warning(f"Unknown setting for {self.name} service: {key}") diff --git a/src/pipecat/services/anthropic.py b/src/pipecat/services/anthropic.py index 1b7064209..a87ed9423 100644 --- a/src/pipecat/services/anthropic.py +++ b/src/pipecat/services/anthropic.py @@ -267,7 +267,7 @@ class AnthropicLLMService(LLMService): context = None if isinstance(frame, OpenAILLMContextFrame): - context = frame.context + context: "AnthropicLLMContext" = AnthropicLLMContext.upgrade_to_anthropic(frame.context) elif isinstance(frame, LLMMessagesFrame): context = AnthropicLLMContext.from_messages(frame.messages) elif isinstance(frame, VisionImageRawFrame): @@ -332,6 +332,14 @@ class AnthropicLLMContext(OpenAILLMContext): self.system = system + @staticmethod + def upgrade_to_anthropic(obj: OpenAILLMContext) -> "AnthropicLLMContext": + logger.debug(f"Upgrading to Anthropic: {obj}") + if isinstance(obj, OpenAILLMContext) and not isinstance(obj, AnthropicLLMContext): + obj.__class__ = AnthropicLLMContext + obj._restructure_from_openai_messages() + return obj + @classmethod def from_openai_context(cls, openai_context: OpenAILLMContext): self = cls( @@ -361,6 +369,100 @@ class AnthropicLLMContext(OpenAILLMContext): self._messages[:] = messages self._restructure_from_openai_messages() + # convert a message in Anthropic format into one or more messages in OpenAI format + def to_standard_messages(self, obj): + # todo: image format (?) + # tool_use + role = obj.get("role") + content = obj.get("content") + if role == "assistant": + if isinstance(content, str): + return [{"role": role, "content": [{"type": "text", "text": content}]}] + elif isinstance(content, list): + text_items = [] + tool_items = [] + for item in content: + if item["type"] == "text": + text_items.append({"type": "text", "text": item["text"]}) + elif item["type"] == "tool_use": + tool_items.append( + { + "type": "function", + "id": item["id"], + "function": { + "name": item["name"], + "arguments": json.dumps(item["input"]), + }, + } + ) + messages = [] + if text_items: + messages.append({"role": role, "content": text_items}) + if tool_items: + messages.append({"role": role, "tool_calls": tool_items}) + return messages + elif role == "user": + if isinstance(content, str): + return [{"role": role, "content": [{"type": "text", "text": content}]}] + elif isinstance(content, list): + text_items = [] + tool_items = [] + for item in content: + if item["type"] == "text": + text_items.append({"type": "text", "text": item["text"]}) + elif item["type"] == "tool_result": + tool_items.append( + { + "role": "tool", + "tool_call_id": item["tool_use_id"], + "content": item["content"], + } + ) + messages = [] + if text_items: + messages.append({"role": role, "content": text_items}) + messages.extend(tool_items) + return messages + + def from_standard_message(self, message): + # todo: image messages (?) + if message["role"] == "tool": + return { + "role": "user", + "content": [ + { + "type": "tool_result", + "tool_use_id": message["tool_call_id"], + "content": message["content"], + }, + ], + } + if message.get("tool_calls"): + tc = message["tool_calls"] + ret = {"role": "assistant", "content": []} + for tool_call in tc: + function = tool_call["function"] + arguments = json.loads(function["arguments"]) + new_tool_use = { + "type": "tool_use", + "id": tool_call["id"], + "name": function["name"], + "input": arguments, + } + ret["content"].append(new_tool_use) + return ret + # check for empty text strings + content = message.get("content") + if isinstance(content, str): + if content == "": + content = "(empty)" + elif isinstance(content, list): + for item in content: + if item["type"] == "text" and item["text"] == "": + item["text"] = "(empty)" + + return message + def add_image_frame_message( self, *, format: str, size: tuple[int, int], image: bytes, text: str = None ): @@ -429,6 +531,12 @@ class AnthropicLLMContext(OpenAILLMContext): return self.messages def _restructure_from_openai_messages(self): + # first, map across self._messages calling self.from_standard_message(m) to modify messages in place + try: + self._messages[:] = [self.from_standard_message(m) for m in self._messages] + except Exception as e: + logger.error(f"Error mapping messages: {e}") + # See if we should pull the system message out of our context.messages list. (For # compatibility with Open AI messages format.) if self.messages and self.messages[0]["role"] == "system": @@ -442,6 +550,39 @@ class AnthropicLLMContext(OpenAILLMContext): self.system = self.messages[0]["content"] self.messages.pop(0) + # Merge consecutive messages with the same role. + i = 0 + while i < len(self.messages) - 1: + current_message = self.messages[i] + next_message = self.messages[i + 1] + if current_message["role"] == next_message["role"]: + # Convert content to list of dictionaries if it's a string + if isinstance(current_message["content"], str): + current_message["content"] = [ + {"type": "text", "text": current_message["content"]} + ] + if isinstance(next_message["content"], str): + next_message["content"] = [{"type": "text", "text": next_message["content"]}] + # Concatenate the content + current_message["content"].extend(next_message["content"]) + # Remove the next message from the list + self.messages.pop(i + 1) + else: + i += 1 + + # Avoid empty content in messages + for message in self.messages: + if isinstance(message["content"], str) and message["content"] == "": + message["content"] = "(empty)" + elif isinstance(message["content"], list) and len(message["content"]) == 0: + message["content"] = [{"type": "text", "text": "(empty)"}] + + def get_messages_for_persistent_storage(self): + messages = super().get_messages_for_persistent_storage() + if self.system: + messages.insert(0, {"role": "system", "content": self.system}) + return messages + def get_messages_for_logging(self) -> str: msgs = [] for message in self.messages: diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index a93d40210..44d707cf2 100644 --- a/src/pipecat/services/openai.py +++ b/src/pipecat/services/openai.py @@ -63,6 +63,7 @@ except ModuleNotFoundError as e: ) raise Exception(f"Missing module: {e}") + ValidVoice = Literal["alloy", "echo", "fable", "onyx", "nova", "shimmer"] VALID_VOICES: Dict[str, ValidVoice] = { @@ -468,7 +469,7 @@ class OpenAIUserContextAggregator(LLMUserContextAggregator): if frame.user_id in self._context._user_image_request_context: del self._context._user_image_request_context[frame.user_id] elif isinstance(frame, UserImageRawFrame): - # Push a new AnthropicImageMessageFrame with the text context we cached + # Push a new OpenAIImageMessageFrame with the text context we cached # downstream to be handled by our assistant context aggregator. This is # necessary so that we add the message to the context in the right order. text = self._context._user_image_request_context.get(frame.user_id) or "" @@ -495,8 +496,10 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator): self._function_calls_in_progress.clear() self._function_call_finished = None elif isinstance(frame, FunctionCallInProgressFrame): + logger.debug(f"FunctionCallInProgressFrame: {frame}") self._function_calls_in_progress[frame.tool_call_id] = frame elif isinstance(frame, FunctionCallResultFrame): + logger.debug(f"FunctionCallResultFrame: {frame}") if frame.tool_call_id in self._function_calls_in_progress: del self._function_calls_in_progress[frame.tool_call_id] self._function_call_result = frame diff --git a/src/pipecat/services/openai_realtime_beta/__init__.py b/src/pipecat/services/openai_realtime_beta/__init__.py new file mode 100644 index 000000000..ebd37d148 --- /dev/null +++ b/src/pipecat/services/openai_realtime_beta/__init__.py @@ -0,0 +1,2 @@ +from .events import InputAudioTranscription, SessionProperties, TurnDetection +from .llm_and_context import OpenAILLMServiceRealtimeBeta diff --git a/src/pipecat/services/openai_realtime_beta/events.py b/src/pipecat/services/openai_realtime_beta/events.py new file mode 100644 index 000000000..0515012e3 --- /dev/null +++ b/src/pipecat/services/openai_realtime_beta/events.py @@ -0,0 +1,433 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# +# + +import json +import uuid +from typing import Any, Dict, List, Literal, Optional, Union + +from pydantic import BaseModel, Field + +# +# session properties +# + + +class InputAudioTranscription(BaseModel): + model: Optional[str] = "whisper-1" + + +class TurnDetection(BaseModel): + type: Optional[Literal["server_vad"]] = "server_vad" + threshold: Optional[float] = 0.5 + prefix_padding_ms: Optional[int] = 300 + silence_duration_ms: Optional[int] = 800 + + +class SessionProperties(BaseModel): + modalities: Optional[List[Literal["text", "audio"]]] = None + instructions: Optional[str] = None + voice: Optional[str] = None + input_audio_format: Optional[Literal["pcm16", "g711_ulaw", "g711_alaw"]] = None + output_audio_format: Optional[Literal["pcm16", "g711_ulaw", "g711_alaw"]] = None + input_audio_transcription: Optional[InputAudioTranscription] = None + # set turn_detection to False to disable turn detection + turn_detection: Optional[Union[TurnDetection, bool]] = Field(default=None) + tools: Optional[List[Dict]] = None + tool_choice: Optional[Literal["auto", "none", "required"]] = None + temperature: Optional[float] = None + max_response_output_tokens: Optional[Union[int, Literal["inf"]]] = None + + +# +# context +# + + +class ItemContent(BaseModel): + type: Literal["text", "audio", "input_text", "input_audio"] + text: Optional[str] = None + audio: Optional[str] = None # base64-encoded audio + transcript: Optional[str] = None + + +class ConversationItem(BaseModel): + id: str = Field(default_factory=lambda: str(uuid.uuid4().hex)) + object: Optional[Literal["realtime.item"]] = None + type: Literal["message", "function_call", "function_call_output"] + status: Optional[Literal["completed", "in_progress", "incomplete"]] = None + # role and content are present for message items + role: Optional[Literal["user", "assistant", "system"]] = None + content: Optional[List[ItemContent]] = None + # these four fields are present for function_call items + call_id: Optional[str] = None + name: Optional[str] = None + arguments: Optional[str] = None + output: Optional[str] = None + + +class RealtimeConversation(BaseModel): + id: str + object: Literal["realtime.conversation"] + + +class ResponseProperties(BaseModel): + modalities: Optional[List[Literal["text", "audio"]]] = ["audio", "text"] + instructions: Optional[str] = None + voice: Optional[str] = None + output_audio_format: Optional[Literal["pcm16", "g711_ulaw", "g711_alaw"]] = None + tools: Optional[List[Dict]] = [] + tool_choice: Optional[Literal["auto", "none", "required"]] = None + temperature: Optional[float] = None + max_response_output_tokens: Optional[Union[int, Literal["inf"]]] = None + + +# +# error class +# +class RealtimeError(BaseModel): + type: str + code: Optional[str] = "" + message: str + param: Optional[str] = None + + +# +# client events +# + + +class ClientEvent(BaseModel): + event_id: str = Field(default_factory=lambda: str(uuid.uuid4())) + + +class SessionUpdateEvent(ClientEvent): + type: Literal["session.update"] = "session.update" + session: SessionProperties + + def model_dump(self, *args, **kwargs) -> Dict[str, Any]: + dump = super().model_dump(*args, **kwargs) + + # Handle turn_detection so that False is serialized as null + if "turn_detection" in dump["session"]: + if dump["session"]["turn_detection"] is False: + dump["session"]["turn_detection"] = None + + return dump + + +class InputAudioBufferAppendEvent(ClientEvent): + type: Literal["input_audio_buffer.append"] = "input_audio_buffer.append" + audio: str # base64-encoded audio + + +class InputAudioBufferCommitEvent(ClientEvent): + type: Literal["input_audio_buffer.commit"] = "input_audio_buffer.commit" + + +class InputAudioBufferClearEvent(ClientEvent): + type: Literal["input_audio_buffer.clear"] = "input_audio_buffer.clear" + + +class ConversationItemCreateEvent(ClientEvent): + type: Literal["conversation.item.create"] = "conversation.item.create" + previous_item_id: Optional[str] = None + item: ConversationItem + + +class ConversationItemTruncateEvent(ClientEvent): + type: Literal["conversation.item.truncate"] = "conversation.item.truncate" + item_id: str + content_index: int + audio_end_ms: int + + +class ConversationItemDeleteEvent(ClientEvent): + type: Literal["conversation.item.delete"] = "conversation.item.delete" + item_id: str + + +class ResponseCreateEvent(ClientEvent): + type: Literal["response.create"] = "response.create" + response: Optional[ResponseProperties] = None + + +class ResponseCancelEvent(ClientEvent): + type: Literal["response.cancel"] = "response.cancel" + + +# +# server events +# + + +class ServerEvent(BaseModel): + event_id: str + type: str + + class Config: + arbitrary_types_allowed = True + + +class SessionCreatedEvent(ServerEvent): + type: Literal["session.created"] + session: SessionProperties + + +class SessionUpdatedEvent(ServerEvent): + type: Literal["session.updated"] + session: SessionProperties + + +class ConversationCreated(ServerEvent): + type: Literal["conversation.created"] + conversation: RealtimeConversation + + +class ConversationItemCreated(ServerEvent): + type: Literal["conversation.item.created"] + previous_item_id: Optional[str] = None + item: ConversationItem + + +class ConversationItemInputAudioTranscriptionCompleted(ServerEvent): + type: Literal["conversation.item.input_audio_transcription.completed"] + item_id: str + content_index: int + transcript: str + + +class ConversationItemInputAudioTranscriptionFailed(ServerEvent): + type: Literal["conversation.item.input_audio_transcription.failed"] + item_id: str + content_index: int + error: RealtimeError + + +class ConversationItemTruncated(ServerEvent): + type: Literal["conversation.item.truncated"] + item_id: str + content_index: int + audio_end_ms: int + + +class ConversationItemDeleted(ServerEvent): + type: Literal["conversation.item.deleted"] + item_id: str + + +class ResponseCreated(ServerEvent): + type: Literal["response.created"] + response: "Response" + + +class ResponseDone(ServerEvent): + type: Literal["response.done"] + response: "Response" + + +class ResponseOutputItemAdded(ServerEvent): + type: Literal["response.output_item.added"] + response_id: str + output_index: int + item: ConversationItem + + +class ResponseOutputItemDone(ServerEvent): + type: Literal["response.output_item.done"] + response_id: str + output_index: int + item: ConversationItem + + +class ResponseContentPartAdded(ServerEvent): + type: Literal["response.content_part.added"] + response_id: str + item_id: str + output_index: int + content_index: int + part: ItemContent + + +class ResponseContentPartDone(ServerEvent): + type: Literal["response.content_part.done"] + response_id: str + item_id: str + output_index: int + content_index: int + part: ItemContent + + +class ResponseTextDelta(ServerEvent): + type: Literal["response.text.delta"] + response_id: str + item_id: str + output_index: int + content_index: int + delta: str + + +class ResponseTextDone(ServerEvent): + type: Literal["response.text.done"] + response_id: str + item_id: str + output_index: int + content_index: int + text: str + + +class ResponseAudioTranscriptDelta(ServerEvent): + type: Literal["response.audio_transcript.delta"] + response_id: str + item_id: str + output_index: int + content_index: int + delta: str + + +class ResponseAudioTranscriptDone(ServerEvent): + type: Literal["response.audio_transcript.done"] + response_id: str + item_id: str + output_index: int + content_index: int + transcript: str + + +class ResponseAudioDelta(ServerEvent): + type: Literal["response.audio.delta"] + response_id: str + item_id: str + output_index: int + content_index: int + delta: str # base64-encoded audio + + +class ResponseAudioDone(ServerEvent): + type: Literal["response.audio.done"] + response_id: str + item_id: str + output_index: int + content_index: int + + +class ResponseFunctionCallArgumentsDelta(ServerEvent): + type: Literal["response.function_call_arguments.delta"] + response_id: str + item_id: str + output_index: int + call_id: str + delta: str + + +class ResponseFunctionCallArgumentsDone(ServerEvent): + type: Literal["response.function_call_arguments.done"] + response_id: str + item_id: str + output_index: int + call_id: str + arguments: str + + +class InputAudioBufferSpeechStarted(ServerEvent): + type: Literal["input_audio_buffer.speech_started"] + audio_start_ms: int + item_id: str + + +class InputAudioBufferSpeechStopped(ServerEvent): + type: Literal["input_audio_buffer.speech_stopped"] + audio_end_ms: int + item_id: str + + +class InputAudioBufferCommitted(ServerEvent): + type: Literal["input_audio_buffer.committed"] + previous_item_id: Optional[str] = None + item_id: str + + +class InputAudioBufferCleared(ServerEvent): + type: Literal["input_audio_buffer.cleared"] + + +class ErrorEvent(ServerEvent): + type: Literal["error"] + error: RealtimeError + + +class RateLimitsUpdated(ServerEvent): + type: Literal["rate_limits.updated"] + rate_limits: List[Dict[str, Any]] + + +class TokenDetails(BaseModel): + cached_tokens: Optional[int] = 0 + text_tokens: Optional[int] = 0 + audio_tokens: Optional[int] = 0 + + class Config: + extra = "allow" + + +class Usage(BaseModel): + total_tokens: int + input_tokens: int + output_tokens: int + input_token_details: TokenDetails + output_token_details: TokenDetails + + +class Response(BaseModel): + id: str + object: Literal["realtime.response"] + status: Literal["completed", "in_progress", "incomplete", "cancelled", "failed"] + status_details: Any + output: List[ConversationItem] + usage: Optional[Usage] = None + + +_server_event_types = { + "error": ErrorEvent, + "session.created": SessionCreatedEvent, + "session.updated": SessionUpdatedEvent, + "conversation.created": ConversationCreated, + "input_audio_buffer.committed": InputAudioBufferCommitted, + "input_audio_buffer.cleared": InputAudioBufferCleared, + "input_audio_buffer.speech_started": InputAudioBufferSpeechStarted, + "input_audio_buffer.speech_stopped": InputAudioBufferSpeechStopped, + "conversation.item.created": ConversationItemCreated, + "conversation.item.input_audio_transcription.completed": ConversationItemInputAudioTranscriptionCompleted, + "conversation.item.input_audio_transcription.failed": ConversationItemInputAudioTranscriptionFailed, + "conversation.item.truncated": ConversationItemTruncated, + "conversation.item.deleted": ConversationItemDeleted, + "response.created": ResponseCreated, + "response.done": ResponseDone, + "response.output_item.added": ResponseOutputItemAdded, + "response.output_item.done": ResponseOutputItemDone, + "response.content_part.added": ResponseContentPartAdded, + "response.content_part.done": ResponseContentPartDone, + "response.text.delta": ResponseTextDelta, + "response.text.done": ResponseTextDone, + "response.audio_transcript.delta": ResponseAudioTranscriptDelta, + "response.audio_transcript.done": ResponseAudioTranscriptDone, + "response.audio.delta": ResponseAudioDelta, + "response.audio.done": ResponseAudioDone, + "response.function_call_arguments.delta": ResponseFunctionCallArgumentsDelta, + "response.function_call_arguments.done": ResponseFunctionCallArgumentsDone, + "rate_limits.updated": RateLimitsUpdated, +} + + +def parse_server_event(str): + try: + event = json.loads(str) + event_type = event["type"] + if event_type not in _server_event_types: + raise Exception(f"Unimplemented server event type: {event_type}") + return _server_event_types[event_type].model_validate(event) + except Exception as e: + raise Exception(f"{e} \n\n{str}") diff --git a/src/pipecat/services/openai_realtime_beta/llm_and_context.py b/src/pipecat/services/openai_realtime_beta/llm_and_context.py new file mode 100644 index 000000000..173ee5103 --- /dev/null +++ b/src/pipecat/services/openai_realtime_beta/llm_and_context.py @@ -0,0 +1,755 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import base64 +import copy +import json +import time +from dataclasses import dataclass + +import websockets +from loguru import logger + +from pipecat.frames.frames import ( + BotStoppedSpeakingFrame, + CancelFrame, + DataFrame, + EndFrame, + ErrorFrame, + Frame, + FunctionCallResultFrame, + InputAudioRawFrame, + LLMFullResponseEndFrame, + LLMFullResponseStartFrame, + LLMMessagesAppendFrame, + LLMMessagesUpdateFrame, + LLMSetToolsFrame, + LLMUpdateSettingsFrame, + StartFrame, + StartInterruptionFrame, + StopInterruptionFrame, + TextFrame, + TranscriptionFrame, + TTSAudioRawFrame, + TTSStartedFrame, + TTSStoppedFrame, + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame, +) +from pipecat.metrics.metrics import LLMTokenUsage +from pipecat.processors.aggregators.openai_llm_context import ( + OpenAILLMContext, + OpenAILLMContextFrame, +) +from pipecat.processors.frame_processor import FrameDirection +from pipecat.services.ai_services import LLMService +from pipecat.services.openai import ( + OpenAIAssistantContextAggregator, + OpenAIContextAggregatorPair, + OpenAIUserContextAggregator, +) +from pipecat.utils.time import time_now_iso8601 + +from . import events +from .events import SessionProperties + +# websocket logger -- in case needed for debugging send/recv +# import logging +# logging.basicConfig( +# format="%(message)s", +# level=logging.DEBUG, +# ) + + +@dataclass +class _InternalMessagesUpdateFrame(DataFrame): + context: "OpenAIRealtimeLLMContext" + + +@dataclass +class _InternalFunctionCallResultFrame(DataFrame): + result_frame: FunctionCallResultFrame + + +@dataclass +class _CurrentAudioResponse: + item_id: str + content_index: int + start_time_ms: int + total_size: int = 0 + + +class OpenAIUnhandledFunctionException(Exception): + pass + + +class OpenAIRealtimeLLMContext(OpenAILLMContext): + def __init__(self, messages=None, tools=None, **kwargs): + super().__init__(messages=messages, tools=tools, **kwargs) + self.__setup_local() + + def __setup_local(self): + self.llm_needs_settings_update = True + self.llm_needs_initial_messages = True + self._session_instructions = "" + + return + + @staticmethod + def upgrade_to_realtime(obj: OpenAILLMContext) -> "OpenAIRealtimeLLMContext": + if isinstance(obj, OpenAILLMContext) and not isinstance(obj, OpenAIRealtimeLLMContext): + obj.__class__ = OpenAIRealtimeLLMContext + obj.__setup_local() + return obj + + # todo + # - finish implementing all frames + # - add message conversion functions to OpenAILLMContext base class + + def from_standard_message(self, message): + if message.get("role") == "assistant" and message.get("tool_calls"): + tc = message.get("tool_calls")[0] + return events.ConversationItem( + type="function_call", + call_id=tc["id"], + name=tc["function"]["name"], + arguments=tc["function"]["arguments"], + ) + logger.error(f"Unhandled message type in from_standard_message: {message}") + + def get_messages_for_initializing_history(self): + # We can't load a long conversation history into the openai realtime api yet. (The API/model + # forgets that it can do audio, if you do a series of `conversation.item.create` calls.) So + # our general strategy until this is fixed is just to put everything into a first "user" + # message as a single input. + if not self.messages: + return [] + + messages = copy.deepcopy(self.messages) + + # If we have a "system" message as our first message, let's pull that out into session + # "instructions" + if messages[0].get("role") == "system": + self.llm_needs_settings_update = True + system = messages.pop(0) + content = system.get("content") + if isinstance(content, str): + self._session_instructions = content + elif isinstance(content, list): + self._session_instructions = content[0].get("text") + if not messages: + return [] + + # If we have just a single "user" item, we can just send it normally + if len(messages) == 1 and messages[0].get("role") == "user": + return messages + + # Otherwise, let's pack everything into a single "user" message with a bit of + # explanation for the LLM + intro_text = """ + This is a previously saved conversation. Please treat this conversation history as a + starting point for the current conversation.""" + + trailing_text = """ + This is the end of the previously saved conversation. Please continue the conversation + from here. If the last message is a user instruction or question, act on that instruction + or answer the question. If the last message is an assistant response, simple say that you + are ready to continue the conversation.""" + + return [ + { + "role": "user", + "type": "message", + "content": [ + { + "type": "input_text", + "text": "\n\n".join( + [intro_text, json.dumps(messages, indent=2), trailing_text] + ), + } + ], + } + ] + + def add_user_content_item_as_message(self, item): + message = { + "role": "user", + "content": [{"type": "text", "text": item.content[0].transcript}], + } + self.add_message(message) + + def add_assistant_content_item_as_message(self, item): + message = {"role": "assistant", "content": []} + for content in item.content: + if content.type == "audio": + message["content"].append({"type": "text", "text": content.transcript}) + else: + logger.error(f"Unhandled content type in assistant item: {content.type} - {item}") + self.add_message(message) + + +class OpenAIRealtimeUserContextAggregator(OpenAIUserContextAggregator): + async def process_frame( + self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM + ): + await super().process_frame(frame, direction) + # Parent does not push LLMMessagesUpdateFrame. This ensures that in a typical pipeline, + # messages are only processed by the user context aggregator, which is generally what we want. But + # we also need to send new messages over the websocket, so the openai realtime API has them + # in its context. + if isinstance(frame, LLMMessagesUpdateFrame): + await self.push_frame(_InternalMessagesUpdateFrame(context=self._context)) + + # Parent also doesn't push the LLMSetToolsFrame. + if isinstance(frame, LLMSetToolsFrame): + await self.push_frame(frame, direction) + + async def _push_aggregation(self): + # for the moment, ignore all user input coming into the pipeline. + # todo: think about whether/how to fix this to allow for text input from + # upstream (transport/transcription, or other sources) + pass + + +class OpenAIRealtimeAssistantContextAggregator(OpenAIAssistantContextAggregator): + async def _push_aggregation(self): + # the only thing we implement here is function calling. in all other cases, messages + # are added to the context when we receive openai realtime api events + if not self._function_call_result: + return + + self._reset() + try: + frame = self._function_call_result + self._function_call_result = None + if frame.result: + # The "tool_call" message from the LLM that triggered the function call + self._context.add_message( + { + "role": "assistant", + "tool_calls": [ + { + "id": frame.tool_call_id, + "function": { + "name": frame.function_name, + "arguments": json.dumps(frame.arguments), + }, + "type": "function", + } + ], + } + ) + # The result of the function call. Need to add this both to our context here and to + # the openai realtime api context. + result_message = { + "role": "tool", + "content": json.dumps(frame.result), + "tool_call_id": frame.tool_call_id, + } + + self._context.add_message(result_message) + # The standard function callback code path pushes the FunctionCallResultFrame from the llm itself, + # so we didn't have a chance to add the result to the openai realtime api context. Let's push a + # special frame to do that. + await self._user_context_aggregator.push_frame( + _InternalFunctionCallResultFrame(result_frame=frame) + ) + run_llm = frame.run_llm + + 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}") + + +class OpenAILLMServiceRealtimeBeta(LLMService): + def __init__( + self, + *, + api_key: str, + base_url="wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview-2024-10-01", + session_properties: events.SessionProperties = events.SessionProperties(), + start_audio_paused: bool = False, + send_transcription_frames: bool = True, + **kwargs, + ): + super().__init__(base_url=base_url, **kwargs) + self.api_key = api_key + self.base_url = base_url + + self._session_properties: events.SessionProperties = session_properties + self._audio_input_paused = start_audio_paused + self._send_transcription_frames = send_transcription_frames + self._websocket = None + self._receive_task = None + self._context = None + + self._disconnecting = False + self._api_session_ready = False + self._run_llm_when_api_session_ready = False + + self._current_assistant_response = None + self._current_audio_response = None + + self._messages_added_manually = {} + self._user_and_response_message_tuple = None + + def can_generate_metrics(self) -> bool: + return True + + def set_audio_input_paused(self, paused: bool): + self._audio_input_paused = paused + + # + # standard AIService frame handling + # + + async def start(self, frame: StartFrame): + await super().start(frame) + await self._connect() + + async def stop(self, frame: EndFrame): + await super().stop(frame) + await self._disconnect() + + async def cancel(self, frame: CancelFrame): + await super().cancel(frame) + await self._disconnect() + + # + # speech and interruption handling + # + + async def _handle_interruption(self): + if self._session_properties.turn_detection is None: + await self.send_client_event(events.InputAudioBufferClearEvent()) + await self.send_client_event(events.ResponseCancelEvent()) + await self._truncate_current_audio_response() + await self.stop_all_metrics() + if self._current_assistant_response: + await self.push_frame(LLMFullResponseEndFrame()) + await self.push_frame(TTSStoppedFrame()) + + async def _handle_user_started_speaking(self, frame): + if self._session_properties.turn_detection is None: + await self._handle_interruption() + + async def _handle_user_stopped_speaking(self, frame): + if self._session_properties.turn_detection is None: + await self.send_client_event(events.InputAudioBufferCommitEvent()) + await self.send_client_event(events.ResponseCreateEvent()) + + async def _handle_bot_stopped_speaking(self): + self._current_audio_response = None + + async def _truncate_current_audio_response(self): + # if the bot is still speaking, truncate the last message + if self._current_audio_response: + current = self._current_audio_response + self._current_audio_response = None + elapsed_ms = int(time.time() * 1000 - current.start_time_ms) + await self.send_client_event( + events.ConversationItemTruncateEvent( + item_id=current.item_id, + content_index=current.content_index, + audio_end_ms=elapsed_ms, + ) + ) + + + # + # frame processing + # + # StartFrame, StopFrame, CancelFrame implemented in base class + # + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if isinstance(frame, TranscriptionFrame): + pass + elif isinstance(frame, OpenAILLMContextFrame): + context: OpenAIRealtimeLLMContext = OpenAIRealtimeLLMContext.upgrade_to_realtime( + frame.context + ) + if not self._context: + self._context = context + elif frame.context is not self._context: + # If the context has changed, reset the conversation + self._context = context + await self.reset_conversation() + # Run the LLM at next opportunity + await self._create_response() + elif isinstance(frame, InputAudioRawFrame): + if not self._audio_input_paused: + await self._send_user_audio(frame) + elif isinstance(frame, StartInterruptionFrame): + await self._handle_interruption() + elif isinstance(frame, UserStartedSpeakingFrame): + await self._handle_user_started_speaking(frame) + elif isinstance(frame, UserStoppedSpeakingFrame): + await self._handle_user_stopped_speaking(frame) + elif isinstance(frame, BotStoppedSpeakingFrame): + await self._handle_bot_stopped_speaking() + elif isinstance(frame, LLMMessagesAppendFrame): + await self._handle_messages_append(frame) + elif isinstance(frame, _InternalMessagesUpdateFrame): + self._context = frame.context + elif isinstance(frame, LLMUpdateSettingsFrame): + self._session_properties = SessionProperties(**frame.settings) + await self._update_settings() + elif isinstance(frame, LLMSetToolsFrame): + await self._update_settings() + elif isinstance(frame, _InternalFunctionCallResultFrame): + await self._handle_function_call_result(frame.result_frame) + + await self.push_frame(frame, direction) + + async def _handle_messages_append(self, frame): + logger.error("!!! NEED TO IMPLEMENT MESSAGES APPEND") + + async def _handle_function_call_result(self, frame): + item = events.ConversationItem( + type="function_call_output", + call_id=frame.tool_call_id, + output=json.dumps(frame.result), + ) + await self.send_client_event(events.ConversationItemCreateEvent(item=item)) + + # + # websocket communication + # + + async def send_client_event(self, event: events.ClientEvent): + await self._ws_send(event.model_dump(exclude_none=True)) + + async def _connect(self): + try: + if self._websocket: + # Here we assume that if we have a websocket, we are connected. We + # handle disconnections in the send/recv code paths. + return + self._websocket = await websockets.connect( + uri=self.base_url, + extra_headers={ + "Authorization": f"Bearer {self.api_key}", + "OpenAI-Beta": "realtime=v1", + }, + ) + self._receive_task = self.get_event_loop().create_task(self._receive_task_handler()) + except Exception as e: + logger.error(f"{self} initialization error: {e}") + self._websocket = None + + async def _disconnect(self): + try: + self._disconnecting = True + self._api_session_ready = False + await self.stop_all_metrics() + if self._websocket: + await self._websocket.close() + self._websocket = None + if self._receive_task: + self._receive_task.cancel() + try: + await asyncio.wait_for(self._receive_task, timeout=1.0) + except asyncio.TimeoutError: + logger.warning("Timed out waiting for receive task to finish") + self._receive_task = None + self._disconnecting = False + except Exception as e: + logger.error(f"{self} error disconnecting: {e}") + + async def _ws_send(self, realtime_message): + try: + if self._websocket: + await self._websocket.send(json.dumps(realtime_message)) + except Exception as e: + if self._disconnecting: + return + logger.error(f"Error sending message to websocket: {e}") + # In server-to-server contexts, a WebSocket error should be quite rare. Given how hard + # it is to recover from a send-side error with proper state management, and that exponential + # backoff for retries can have cost/stability implications for a service cluster, let's just + # treat a send-side error as fatal. + await self.push_error(ErrorFrame(error=f"Error sending client event: {e}", fatal=True)) + + async def _update_settings(self): + settings = self._session_properties + # tools given in the context override the tools in the session properties + if self._context and self._context.tools: + settings.tools = self._context.tools + # instructions in the context come from an initial "system" message in the + # messages list, and override instructions in the session properties + if self._context and self._context._session_instructions: + settings.instructions = self._context._session_instructions + await self.send_client_event(events.SessionUpdateEvent(session=settings)) + + # + # inbound server event handling + # https://platform.openai.com/docs/api-reference/realtime-server-events + # + + async def _receive_task_handler(self): + try: + async for message in self._websocket: + evt = events.parse_server_event(message) + if evt.type == "session.created": + await self._handle_evt_session_created(evt) + elif evt.type == "session.updated": + await self._handle_evt_session_updated(evt) + elif evt.type == "response.audio.delta": + await self._handle_evt_audio_delta(evt) + elif evt.type == "response.audio.done": + await self._handle_evt_audio_done(evt) + elif evt.type == "conversation.item.created": + await self._handle_evt_conversation_item_created(evt) + elif evt.type == "conversation.item.input_audio_transcription.completed": + await self.handle_evt_input_audio_transcription_completed(evt) + elif evt.type == "response.done": + await self._handle_evt_response_done(evt) + elif evt.type == "input_audio_buffer.speech_started": + await self._handle_evt_speech_started(evt) + elif evt.type == "input_audio_buffer.speech_stopped": + await self._handle_evt_speech_stopped(evt) + elif evt.type == "response.audio_transcript.delta": + await self._handle_evt_audio_transcript_delta(evt) + elif evt.type == "error": + await self._handle_evt_error(evt) + # errors are fatal, so exit the receive loop + return + + else: + # logger.debug(f"!!! Unhandled event: {evt}") + pass + except asyncio.CancelledError: + logger.debug("websocket receive task cancelled") + except Exception as e: + logger.error(f"{self} exception: {e}") + + async def _handle_evt_session_created(self, evt): + # session.created is received right after connecting. Send a message + # to configure the session properties. + await self._update_settings() + + async def _handle_evt_session_updated(self, evt): + # If this is our first context frame, run the LLM + self._api_session_ready = True + # Now that we've configured the session, we can run the LLM if we need to. + if self._run_llm_when_api_session_ready: + self._run_llm_when_api_session_ready = False + await self._create_response() + + async def _handle_evt_audio_delta(self, evt): + # note: ttfb is faster by 1/2 RTT than ttfb as measured for other services, since we're getting + # this event from the server + await self.stop_ttfb_metrics() + if not self._current_audio_response: + self._current_audio_response = _CurrentAudioResponse( + item_id=evt.item_id, + content_index=evt.content_index, + start_time_ms=int(time.time() * 1000), + ) + await self.push_frame(TTSStartedFrame()) + audio = base64.b64decode(evt.delta) + self._current_audio_response.total_size += len(audio) + frame = TTSAudioRawFrame( + audio=audio, + sample_rate=24000, + num_channels=1, + ) + await self.push_frame(frame) + + async def _handle_evt_audio_done(self, evt): + if self._current_audio_response: + await self.push_frame(TTSStoppedFrame()) + # Don't clear the self._current_audio_response here. We need to wait until we + # receive a BotStoppedSpeakingFrame from the output transport. + + async def _handle_evt_conversation_item_created(self, evt): + # This will get sent from the server every time a new "message" is added + # to the server's conversation state, whether we create it via the API + # or the server creates it from LLM output. + if self._messages_added_manually.get(evt.item.id): + del self._messages_added_manually[evt.item.id] + return + + if evt.item.role == "user": + # We need to wait for completion of both user message and response message. Then we'll + # add both to the context. User message is complete when we have a "transcript" field + # that is not None. Response message is complete when we get a "response.done" event. + self._user_and_response_message_tuple = (evt.item, {"done": False, "output": []}) + elif evt.item.role == "assistant": + self._current_assistant_response = evt.item + await self.push_frame(LLMFullResponseStartFrame()) + + async def handle_evt_input_audio_transcription_completed(self, evt): + if self._send_transcription_frames: + await self.push_frame( + # no way to get a language code? + TranscriptionFrame(evt.transcript, "", time_now_iso8601()) + ) + pair = self._user_and_response_message_tuple + if pair: + user, assistant = pair + user.content[0].transcript = evt.transcript + if assistant["done"]: + self._user_and_response_message_tuple = None + self._context.add_user_content_item_as_message(user) + await self._handle_assistant_output(assistant["output"]) + else: + # User message without preceding conversation.item.created. Bug? + logger.warning(f"Transcript for unknown user message: {evt}") + + async def _handle_evt_response_done(self, evt): + # todo: figure out whether there's anything we need to do for "cancelled" events + # usage metrics + tokens = LLMTokenUsage( + prompt_tokens=evt.response.usage.input_tokens, + completion_tokens=evt.response.usage.output_tokens, + total_tokens=evt.response.usage.total_tokens, + ) + await self.start_llm_usage_metrics(tokens) + await self.stop_processing_metrics() + await self.push_frame(LLMFullResponseEndFrame()) + self._current_assistant_response = None + # response content + pair = self._user_and_response_message_tuple + if pair: + user, assistant = pair + assistant["done"] = True + assistant["output"] = evt.response.output + if user.content[0].transcript is not None: + self._user_and_response_message_tuple = None + self._context.add_user_content_item_as_message(user) + await self._handle_assistant_output(assistant["output"]) + else: + # Response message without preceding user message. Add it to the context. + await self._handle_assistant_output(evt.response.output) + + async def _handle_evt_audio_transcript_delta(self, evt): + if evt.delta: + await self.push_frame(TextFrame(evt.delta)) + + async def _handle_evt_speech_started(self, evt): + await self._truncate_current_audio_response() + # todo: might need to guard sending these when we fully support using either openai + # turn detection of Pipecat turn detection + await self._start_interruption() # cancels this processor task + await self.push_frame(StartInterruptionFrame()) # cancels downstream tasks + await self.push_frame(UserStartedSpeakingFrame()) + + async def _handle_evt_speech_stopped(self, evt): + await self.start_ttfb_metrics() + await self.start_processing_metrics() + await self._stop_interruption() + await self.push_frame(StopInterruptionFrame()) + await self.push_frame(UserStoppedSpeakingFrame()) + + async def _handle_evt_error(self, evt): + # Errors are fatal to this connection. Send an ErrorFrame. + await self.push_error(ErrorFrame(error=f"Error: {evt}", fatal=True)) + + async def _handle_assistant_output(self, output): + # logger.debug(f"!!! HANDLE Assistant output: {output}") + # We haven't seen intermixed audio and function_call items in the same response. But let's + # try to write logic that handles that, if it does happen. + messages = [item for item in output if item.type == "message"] + function_calls = [item for item in output if item.type == "function_call"] + for item in messages: + self._context.add_assistant_content_item_as_message(item) + await self._handle_function_call_items(function_calls) + + async def _handle_function_call_items(self, items): + total_items = len(items) + for index, item in enumerate(items): + function_name = item.name + tool_id = item.call_id + arguments = json.loads(item.arguments) + if self.has_function(function_name): + run_llm = index == total_items - 1 + if function_name in self._callbacks.keys(): + await self.call_function( + context=self._context, + tool_call_id=tool_id, + function_name=function_name, + arguments=arguments, + run_llm=run_llm, + ) + elif None in self._callbacks.keys(): + await self.call_function( + context=self._context, + tool_call_id=tool_id, + function_name=function_name, + arguments=arguments, + run_llm=run_llm, + ) + else: + raise OpenAIUnhandledFunctionException( + f"The LLM tried to call a function named '{function_name}', but there isn't a callback registered for that function." + ) + + # + # state and client events for the current conversation + # https://platform.openai.com/docs/api-reference/realtime-client-events + # + + async def reset_conversation(self): + # Disconnect/reconnect is the safest way to start a new conversation. + # Note that this will fail if called from the receive task. + logger.debug("Resetting conversation") + await self._disconnect() + if self._context: + self._context.llm_needs_settings_update = True + self._context.llm_needs_initial_messages = True + await self._connect() + + async def _create_response(self): + if not self._api_session_ready: + self._run_llm_when_api_session_ready = True + return + + if self._context.llm_needs_initial_messages: + messages = self._context.get_messages_for_initializing_history() + for item in messages: + evt = events.ConversationItemCreateEvent(item=item) + self._messages_added_manually[evt.item.id] = True + await self.send_client_event(evt) + self._context.llm_needs_initial_messages = False + + if self._context.llm_needs_settings_update: + await self._update_settings() + self._context.llm_needs_settings_update = False + + logger.debug(f"Creating response: {self._context.get_messages_for_logging()}") + + await self.push_frame(LLMFullResponseStartFrame()) + await self.start_processing_metrics() + await self.start_ttfb_metrics() + await self.send_client_event( + events.ResponseCreateEvent( + response=events.ResponseProperties(modalities=["audio", "text"]) + ) + ) + + async def _send_user_audio(self, frame): + payload = base64.b64encode(frame.audio).decode("utf-8") + await self.send_client_event(events.InputAudioBufferAppendEvent(audio=payload)) + + def create_context_aggregator( + self, context: OpenAILLMContext, *, assistant_expect_stripped_words: bool = False + ) -> OpenAIContextAggregatorPair: + OpenAIRealtimeLLMContext.upgrade_to_realtime(context) + user = OpenAIRealtimeUserContextAggregator(context) + assistant = OpenAIRealtimeAssistantContextAggregator( + user, expect_stripped_words=assistant_expect_stripped_words + ) + return OpenAIContextAggregatorPair(_user=user, _assistant=assistant)