diff --git a/CHANGELOG.md b/CHANGELOG.md index 359b2a2f4..5ab7c7121 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,89 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Expanded support for univeral `LLMContext` to `OpenAIRealtimeLLMService`. + As a reminder, the context-setup pattern when using `LLMContext` is: + + ```python + context = LLMContext(messages, tools) + context_aggregator = LLMContextAggregatorPair( + context, + # This part is `OpenAIRealtimeLLMService`-specific. + # `expect_stripped_words=False` needed when OpenAI Realtime used with + # "audio" modality (the default). + assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False), + ) + ``` + + (Note that even though `OpenAIRealtimeLLMService` now supports the universal + `LLMContext`, it is not meant to be swapped out for another LLM service at + runtime with `LLMSwitcher`.) + + Note: `TranscriptionFrame`s and `InterimTranscriptionFrame`s now go upstream + from `OpenAIRealtimeLLMService`, so if you're using `TranscriptProcessor`, + say, you'll want to adjust accordingly: + + ```python + pipeline = Pipeline( + [ + transport.input(), + context_aggregator.user(), + + # BEFORE + llm, + transcript.user(), + + # AFTER + transcript.user(), + llm, + + transport.output(), + transcript.assistant(), + context_aggregator.assistant(), + ] + ) + ``` + + Also worth noting: whether or not you use the new context-setup pattern with + `OpenAIRealtimeLLMService`, some types have changed under the hood: + + ```python + ## BEFORE: + + # Context aggregator type + context_aggregator: OpenAIContextAggregatorPair + + # Context frame type + frame: OpenAILLMContextFrame + + # Context type + context: OpenAIRealtimeLLMContext + # or + context: OpenAILLMContext + + ## AFTER: + + # Context aggregator type + context_aggregator: LLMContextAggregatorPair + + # Context frame type + frame: LLMContextFrame + + # Context type + context: LLMContext + ``` + + Also note that `RealtimeMessagesUpdateFrame` and + `RealtimeFunctionCallResultFrame` have been deprecated, since they're no + longer used by `OpenAIRealtimeLLMService`. OpenAI Realtime now works more + like other LLM services in Pipecat, relying on updates to its context, pushed + by context aggregators, to update its internal state. Listen for + `LLMContextFrame`s for context updates. + + Finally, `LLMTextFrame`s are no longer pushed from `OpenAIRealtimeLLMService` + when it's configured with `output_modalities=['audio']`. If you need + to process its output, listen for `TTSTextFrame`s instead. + - Expanded support for universal `LLMContext` to `GeminiLiveLLMService`. As a reminder, the context-setup pattern when using `LLMContext` is: @@ -25,7 +108,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 (Note that even though `GeminiLiveLLMService` now supports the universal `LLMContext`, it is not meant to be swapped out for another LLM service at - runtime.) + runtime with `LLMSwitcher`.) Worth noting: whether or not you use the new context-setup pattern with `GeminiLiveLLMService`, some types have changed under the hood: @@ -79,6 +162,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Deprecated +- The `send_transcription_frames` argument to `OpenAIRealtimeLLMService` is + deprecated. Transcription frames are now always sent. They go upstream, to be + handled by the user context aggregator. See "Added" section for details. + +- Types in `pipecat.services.openai.realtime.context` and + `pipecat.services.openai.realtime.frames` are deprecated, as they're no + longer used by `OpenAIRealtimeLLMService`. See "Added" section for details. + - `SimliVideoService` `simli_config` parameter is deprecated. Use `api_key` and `face_id` parameters instead. @@ -121,7 +212,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 (Note that even though `AWSNovaSonicLLMService` now supports the universal `LLMContext`, it is not meant to be swapped out for another LLM service at - runtime.) + runtime with `LLMSwitcher`.) Worth noting: whether or not you use the new context-setup pattern with `AWSNovaSonicLLMService`, some types have changed under the hood: @@ -200,8 +291,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 deprecated. Transcription frames are now always sent. They go upstream, to be handled by the user context aggregator. See "Added" section for details. -- Types in `pipecat.services.aws.nova_sonic.context` have been deprecated due - to changes to support `LLMContext`. See "Changed" section for details. +- Types in `pipecat.services.aws.nova_sonic.context` are deprecated, as they're + no longer used by `AWSNovaSonicLLMService`. See "Added" section for + details. ### Fixed @@ -5375,4 +5467,4 @@ a bit. ## [0.0.2] - 2024-03-12 -Initial public release. \ No newline at end of file +Initial public release. diff --git a/examples/foundational/19-openai-realtime.py b/examples/foundational/19-openai-realtime.py index f182d7c8c..6907ec196 100644 --- a/examples/foundational/19-openai-realtime.py +++ b/examples/foundational/19-openai-realtime.py @@ -5,6 +5,7 @@ # +import asyncio import os from datetime import datetime @@ -14,12 +15,14 @@ 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, TranscriptionMessage +from pipecat.frames.frames import LLMRunFrame, LLMSetToolsFrame, TranscriptionMessage from pipecat.observers.loggers.transcription_log_observer import TranscriptionLogObserver 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.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams +from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair from pipecat.processors.transcript_processor import TranscriptProcessor from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport @@ -52,6 +55,18 @@ async def fetch_weather_from_api(params: FunctionCallParams): ) +async def get_news(params: FunctionCallParams): + await params.result_callback( + { + "news": [ + "Massive UFO currently hovering above New York City", + "Stock markets reach all-time highs", + "Living dinosaur species discovered in the Amazon rainforest", + ], + } + ) + + async def fetch_restaurant_recommendation(params: FunctionCallParams): await params.result_callback({"name": "The Golden Dragon"}) @@ -73,6 +88,13 @@ weather_function = FunctionSchema( required=["location", "format"], ) +get_news_function = FunctionSchema( + name="get_news", + description="Get the current news.", + properties={}, + required=[], +) + restaurant_function = FunctionSchema( name="get_restaurant_recommendation", description="Get a restaurant recommendation", @@ -140,10 +162,6 @@ 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. -You have access to the following tools: -- get_current_weather: Get the current weather for a given location. -- get_restaurant_recommendation: Get a restaurant recommendation for a given location. - Remember, your responses should be short. Just one or two sentences, usually. Respond in English.""", ) @@ -157,25 +175,31 @@ Remember, your responses should be short. Just one or two sentences, usually. Re # llm.register_function(None, fetch_weather_from_api) llm.register_function("get_current_weather", fetch_weather_from_api) llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) + llm.register_function("get_news", get_news) transcript = TranscriptProcessor() # Create a standard OpenAI LLM context object using the normal messages format. The # OpenAIRealtimeLLMService will convert this internally to messages that the # openai WebSocket API can understand. - context = OpenAILLMContext( + context = LLMContext( [{"role": "user", "content": "Say hello!"}], tools, ) - context_aggregator = llm.create_context_aggregator(context) + context_aggregator = LLMContextAggregatorPair( + context, + # `expect_stripped_words=False` needed when OpenAI Realtime used with + # "audio" modality (the default) + assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False), + ) pipeline = Pipeline( [ transport.input(), # Transport user input context_aggregator.user(), + transcript.user(), # LLM pushes TranscriptionFrames upstream llm, # LLM - transcript.user(), # Placed after the LLM, as LLM pushes TranscriptionFrames downstream transport.output(), # Transport bot output transcript.assistant(), # After the transcript output, to time with the audio output context_aggregator.assistant(), @@ -198,6 +222,13 @@ Remember, your responses should be short. Just one or two sentences, usually. Re # Kick off the conversation. await task.queue_frames([LLMRunFrame()]) + # Add a new tool at runtime after a delay. + await asyncio.sleep(15) + new_tools = ToolsSchema( + standard_tools=[weather_function, restaurant_function, get_news_function] + ) + await task.queue_frames([LLMSetToolsFrame(tools=new_tools)]) + @transport.event_handler("on_client_disconnected") async def on_client_disconnected(transport, client): logger.info(f"Client disconnected") diff --git a/examples/foundational/19a-azure-realtime.py b/examples/foundational/19a-azure-realtime.py index c4b0fc02a..7d9cf1b4b 100644 --- a/examples/foundational/19a-azure-realtime.py +++ b/examples/foundational/19a-azure-realtime.py @@ -18,7 +18,9 @@ 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.openai_llm_context import OpenAILLMContext +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams +from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport from pipecat.services.azure.realtime.llm import AzureRealtimeLLMService @@ -155,10 +157,10 @@ Remember, your responses should be short. Just one or two sentences, usually. Re llm.register_function("get_current_weather", fetch_weather_from_api) llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) - # Create a standard OpenAI LLM context object using the normal messages format. The + # Create a standard LLM context object using the normal messages format. The # OpenAIRealtimeBetaLLMService will convert this internally to messages that the # openai WebSocket API can understand. - context = OpenAILLMContext( + context = LLMContext( [{"role": "user", "content": "Say hello!"}], # [{"role": "user", "content": [{"type": "text", "text": "Say hello!"}]}], # [ @@ -173,7 +175,12 @@ Remember, your responses should be short. Just one or two sentences, usually. Re tools, ) - context_aggregator = llm.create_context_aggregator(context) + context_aggregator = LLMContextAggregatorPair( + context, + # `expect_stripped_words=False` needed when OpenAI Realtime used with + # "audio" modality (the default) + assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False), + ) pipeline = Pipeline( [ diff --git a/examples/foundational/19b-openai-realtime-text.py b/examples/foundational/19b-openai-realtime-text.py index bb63a4814..c1f33b7bf 100644 --- a/examples/foundational/19b-openai-realtime-text.py +++ b/examples/foundational/19b-openai-realtime-text.py @@ -18,7 +18,8 @@ from pipecat.frames.frames import LLMRunFrame, TranscriptionMessage 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.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair from pipecat.processors.transcript_processor import TranscriptProcessor from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport @@ -169,20 +170,20 @@ Remember, your responses should be short. Just one or two sentences, usually. Re # Create a standard OpenAI LLM context object using the normal messages format. The # OpenAIRealtimeLLMService will convert this internally to messages that the # openai WebSocket API can understand. - context = OpenAILLMContext( + context = LLMContext( [{"role": "user", "content": "Say hello!"}], tools, ) - context_aggregator = llm.create_context_aggregator(context) + context_aggregator = LLMContextAggregatorPair(context) pipeline = Pipeline( [ transport.input(), # Transport user input context_aggregator.user(), + transcript.user(), # LLM pushes TranscriptionFrames upstream llm, # LLM tts, # TTS - transcript.user(), # Placed after the LLM, as LLM pushes TranscriptionFrames downstream transport.output(), # Transport bot output transcript.assistant(), # After the transcript output, to time with the audio output context_aggregator.assistant(), diff --git a/examples/foundational/20b-persistent-context-openai-realtime.py b/examples/foundational/20b-persistent-context-openai-realtime.py index 629a17c67..e3f018c16 100644 --- a/examples/foundational/20b-persistent-context-openai-realtime.py +++ b/examples/foundational/20b-persistent-context-openai-realtime.py @@ -13,14 +13,15 @@ from datetime import datetime 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 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.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport from pipecat.services.deepgram.stt import DeepgramSTTService @@ -69,11 +70,11 @@ async def save_conversation(params: FunctionCallParams): 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(params.context.messages, indent=4)}" + f"writing conversation to {filename}\n{json.dumps(params.context.get_messages(), indent=4)}" ) try: with open(filename, "w") as file: - messages = params.context.get_messages_for_persistent_storage() + messages = params.context.get_messages() # remove the last message, which is the instruction we just gave to save the conversation messages.pop() json.dump(messages, file, indent=2) @@ -90,6 +91,10 @@ async def load_conversation(params: FunctionCallParams): with open(filename, "r") as file: params.context.set_messages(json.load(file)) await params.llm.reset_conversation() + # NOTE: we manually create a response here rather than relying + # on the function callback to trigger one since we've reset the + # conversation so the remote service doesn't know about the + # in-progress tool call. await params.llm._create_response() except Exception as e: await params.result_callback({"success": False, "error": str(e)}) @@ -97,14 +102,12 @@ async def load_conversation(params: FunctionCallParams): asyncio.create_task(_reset()) -tools = [ - { - "type": "function", - "name": "get_current_weather", - "description": "Get the current weather", - "parameters": { - "type": "object", - "properties": { +tools = ToolsSchema( + standard_tools=[ + FunctionSchema( + name="get_current_weather", + description="Get the current weather", + properties={ "location": { "type": "string", "description": "The city and state, e.g. San Francisco, CA", @@ -115,45 +118,33 @@ tools = [ "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": { + required=["location", "format"], + ), + FunctionSchema( + name="save_conversation", + description="Save the current conversatione. Use this function to persist the current conversation to external storage.", + properties={}, + required=[], + ), + FunctionSchema( + 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.", + properties={}, + required=[], + ), + FunctionSchema( + name="load_conversation", + description="Load a conversation history. Use this function to load a conversation history into the current session.", + properties={ "filename": { "type": "string", "description": "The filename of the conversation history to load.", } }, - "required": ["filename"], - }, - }, -] + required=["filename"], + ), + ] +) # We store functions so objects (e.g. SileroVADAnalyzer) don't get @@ -224,8 +215,8 @@ Remember, your responses should be short. Just one or two sentences, usually.""" 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) + context = LLMContext([{"role": "user", "content": "Say hello!"}], tools) + context_aggregator = LLMContextAggregatorPair(context) pipeline = Pipeline( [ diff --git a/src/pipecat/adapters/schemas/tools_schema.py b/src/pipecat/adapters/schemas/tools_schema.py index 05710616d..d0d798569 100644 --- a/src/pipecat/adapters/schemas/tools_schema.py +++ b/src/pipecat/adapters/schemas/tools_schema.py @@ -22,9 +22,12 @@ class AdapterType(Enum): Parameters: GEMINI: Google Gemini adapter - currently the only service supporting custom tools. + SHIM: Backward compatibility shim for creating ToolsSchemas from lists of tools in + any format, used by LLMContext.from_openai_context. """ GEMINI = "gemini" # that is the only service where we are able to add custom tools for now + SHIM = "shim" # for use as backward compatibility shim for creating ToolsSchemas from list of tools in any format class ToolsSchema: diff --git a/src/pipecat/adapters/services/aws_nova_sonic_adapter.py b/src/pipecat/adapters/services/aws_nova_sonic_adapter.py index 60f12798b..dcc42ba68 100644 --- a/src/pipecat/adapters/services/aws_nova_sonic_adapter.py +++ b/src/pipecat/adapters/services/aws_nova_sonic_adapter.py @@ -16,7 +16,7 @@ from loguru import logger from pipecat.adapters.base_llm_adapter import BaseLLMAdapter from pipecat.adapters.schemas.function_schema import FunctionSchema -from pipecat.adapters.schemas.tools_schema import ToolsSchema +from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema from pipecat.processors.aggregators.llm_context import LLMContext, LLMContextMessage @@ -210,4 +210,18 @@ class AWSNovaSonicLLMAdapter(BaseLLMAdapter[AWSNovaSonicLLMInvocationParams]): List of dictionaries in AWS Nova Sonic function format. """ functions_schema = tools_schema.standard_tools - return [self._to_aws_nova_sonic_function_format(func) for func in functions_schema] + standard_tools = [ + self._to_aws_nova_sonic_function_format(func) for func in functions_schema + ] + + # For backward compatibility, AWS Nova Sonic can still be used with + # tools in dict format, even though it always uses `LLMContext` under + # the hood (via `LLMContext.from_openai_context()`). + # To support this behavior, we use "shimmed" custom tools here. + # (We maintain this backward compatibility because users aren't + # *knowingly* opting into the new `LLMContext`.) + shimmed_tools = [] + if tools_schema.custom_tools: + shimmed_tools = tools_schema.custom_tools.get(AdapterType.SHIM, []) + + return standard_tools + shimmed_tools diff --git a/src/pipecat/adapters/services/open_ai_realtime_adapter.py b/src/pipecat/adapters/services/open_ai_realtime_adapter.py index 2ff629e2e..3d3650633 100644 --- a/src/pipecat/adapters/services/open_ai_realtime_adapter.py +++ b/src/pipecat/adapters/services/open_ai_realtime_adapter.py @@ -6,12 +6,18 @@ """OpenAI Realtime LLM adapter for Pipecat.""" -from typing import Any, Dict, List, TypedDict +import copy +import json +from dataclasses import dataclass +from typing import Any, Dict, List, Optional, TypedDict + +from loguru import logger from pipecat.adapters.base_llm_adapter import BaseLLMAdapter from pipecat.adapters.schemas.function_schema import FunctionSchema -from pipecat.adapters.schemas.tools_schema import ToolsSchema -from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema +from pipecat.processors.aggregators.llm_context import LLMContext, LLMContextMessage +from pipecat.services.openai.realtime import events class OpenAIRealtimeLLMInvocationParams(TypedDict): @@ -20,7 +26,9 @@ class OpenAIRealtimeLLMInvocationParams(TypedDict): This is a placeholder until support for universal LLMContext machinery is added for OpenAI Realtime. """ - pass + system_instruction: Optional[str] + messages: List[events.ConversationItem] + tools: List[Dict[str, Any]] class OpenAIRealtimeLLMAdapter(BaseLLMAdapter): @@ -33,7 +41,7 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter): @property def id_for_llm_specific_messages(self) -> str: """Get the identifier used in LLMSpecificMessage instances for OpenAI Realtime.""" - raise NotImplementedError("Universal LLMContext is not yet supported for OpenAI Realtime.") + return "openai-realtime" def get_llm_invocation_params(self, context: LLMContext) -> OpenAIRealtimeLLMInvocationParams: """Get OpenAI Realtime-specific LLM invocation parameters from a universal LLM context. @@ -46,7 +54,13 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter): Returns: Dictionary of parameters for invoking OpenAI Realtime's API. """ - raise NotImplementedError("Universal LLMContext is not yet supported for OpenAI Realtime.") + messages = self._from_universal_context_messages(self.get_messages(context)) + return { + "system_instruction": messages.system_instruction, + "messages": messages.messages, + # NOTE: LLMContext's tools are guaranteed to be a ToolsSchema (or NOT_GIVEN) + "tools": self.from_standard_tools(context.tools) or [], + } def get_messages_for_logging(self, context) -> List[Dict[str, Any]]: """Get messages from a universal LLM context in a format ready for logging about OpenAI Realtime. @@ -61,7 +75,124 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter): Returns: List of messages in a format ready for logging about OpenAI Realtime. """ - raise NotImplementedError("Universal LLMContext is not yet supported for OpenAI Realtime.") + # NOTE: this is the same as in OpenAIAdapter, as that's what it was + # prior to a refactor. Worth noting that for OpenAI Realtime + # specifically, not everything handled here is necessarily supported + # (or supported yet). + msgs = [] + for message in self.get_messages(context): + msg = copy.deepcopy(message) + if "content" in msg: + if isinstance(msg["content"], list): + for item in msg["content"]: + if item["type"] == "image_url": + if item["image_url"]["url"].startswith("data:image/"): + item["image_url"]["url"] = "data:image/..." + if item["type"] == "input_audio": + item["input_audio"]["data"] = "..." + if "mime_type" in msg and msg["mime_type"].startswith("image/"): + msg["data"] = "..." + msgs.append(msg) + return msgs + + @dataclass + class ConvertedMessages: + """Container for OpenAI-formatted messages converted from universal context.""" + + messages: List[events.ConversationItem] + system_instruction: Optional[str] = None + + def _from_universal_context_messages( + self, universal_context_messages: List[LLMContextMessage] + ) -> ConvertedMessages: + # 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 universal_context_messages: + return self.ConvertedMessages(messages=[]) + + messages = copy.deepcopy(universal_context_messages) + system_instruction = None + + # If we have a "system" message as our first message, let's pull that out into session + # "instructions" + if messages[0].get("role") == "system": + system = messages.pop(0) + content = system.get("content") + if isinstance(content, str): + system_instruction = content + elif isinstance(content, list): + system_instruction = content[0].get("text") + if not messages: + return self.ConvertedMessages(messages=[], system_instruction=system_instruction) + + # 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 self.ConvertedMessages( + messages=[self._from_universal_context_message(messages[0])], + system_instruction=system_instruction, + ) + + # 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 self.ConvertedMessages( + messages=[ + { + "role": "user", + "type": "message", + "content": [ + { + "type": "input_text", + "text": "\n\n".join( + [intro_text, json.dumps(messages, indent=2), trailing_text] + ), + } + ], + } + ], + system_instruction=system_instruction, + ) + + def _from_universal_context_message( + self, message: LLMContextMessage + ) -> events.ConversationItem: + if message.get("role") == "user": + content = message.get("content") + if isinstance(message.get("content"), list): + content = "" + for c in message.get("content"): + if c.get("type") == "text": + content += " " + c.get("text") + else: + logger.error( + f"Unhandled content type in context message: {c.get('type')} - {message}" + ) + return events.ConversationItem( + role="user", + type="message", + content=[events.ItemContent(type="input_text", text=content)], + ) + 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_universal_context_message: {message}") @staticmethod def _to_openai_realtime_function_format(function: FunctionSchema) -> Dict[str, Any]: @@ -94,4 +225,18 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter): List of function definitions in OpenAI Realtime format. """ functions_schema = tools_schema.standard_tools - return [self._to_openai_realtime_function_format(func) for func in functions_schema] + standard_tools = [ + self._to_openai_realtime_function_format(func) for func in functions_schema + ] + + # For backward compatibility, OpenAI Realtime can still be used with + # tools in dict format, even though it always uses `LLMContext` under + # the hood (via `LLMContext.from_openai_context()`). + # To support this behavior, we use "shimmed" custom tools here. + # (We maintain this backward compatibility because users aren't + # *knowingly* opting into the new `LLMContext`.) + shimmed_tools = [] + if tools_schema.custom_tools: + shimmed_tools = tools_schema.custom_tools.get(AdapterType.SHIM, []) + + return standard_tools + shimmed_tools diff --git a/src/pipecat/processors/aggregators/llm_context.py b/src/pipecat/processors/aggregators/llm_context.py index 913566909..8dc79fb50 100644 --- a/src/pipecat/processors/aggregators/llm_context.py +++ b/src/pipecat/processors/aggregators/llm_context.py @@ -15,7 +15,6 @@ service-specific adapter. """ import base64 -import copy import io from dataclasses import dataclass from typing import TYPE_CHECKING, Any, List, Optional, TypeAlias, Union @@ -29,7 +28,7 @@ from openai.types.chat import ( ) from PIL import Image -from pipecat.adapters.schemas.tools_schema import ToolsSchema +from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema from pipecat.frames.frames import AudioRawFrame if TYPE_CHECKING: @@ -83,9 +82,17 @@ class LLMContext: Returns: New LLMContext instance with converted messages and settings. """ + # Convert tools to ToolsSchema if needed. + # If the tools are already a ToolsSchema, this is a no-op. + # Otherwise, we wrap them in a shim ToolsSchema. + converted_tools = openai_context.tools + if isinstance(converted_tools, list): + converted_tools = ToolsSchema( + standard_tools=[], custom_tools={AdapterType.SHIM: converted_tools} + ) return LLMContext( messages=openai_context.get_messages(), - tools=openai_context.tools, + tools=converted_tools, tool_choice=openai_context.tool_choice, ) @@ -119,6 +126,33 @@ class LLMContext: """ return self.get_messages() + def get_messages_for_persistent_storage(self) -> List[LLMContextMessage]: + """Get messages suitable for persistent storage. + + NOTE: the only reason this method exists is because we're "silently" + switching from OpenAILLMContext to LLMContext under the hood in some + services and don't want to trip up users who may have been relying on + this method, which is part of the public API of OpenAILLMContext but + doesn't need to be for LLMContext. + + .. deprecated:: + Use `get_messages()` instead. + + Returns: + List of conversation messages. + """ + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "get_messages_for_persistent_storage() is deprecated, use get_messages() instead.", + DeprecationWarning, + stacklevel=2, + ) + + return self.get_messages() + def get_messages(self, llm_specific_filter: Optional[str] = None) -> List[LLMContextMessage]: """Get the current messages list. diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index 69a8dd280..9f1e04fe0 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -290,6 +290,12 @@ class LLMUserAggregator(LLMContextAggregator): await self._handle_llm_messages_update(frame) elif isinstance(frame, LLMSetToolsFrame): self.set_tools(frame.tools) + # Push the LLMSetToolsFrame as well, since speech-to-speech LLM + # services (like OpenAI Realtime) may need to know about tool + # changes; unlike text-based LLM services they won't just "pick up + # the change" on the next LLM run, as the LLM is continuously + # running. + await self.push_frame(frame, direction) elif isinstance(frame, LLMSetToolChoiceFrame): self.set_tool_choice(frame.tool_choice) elif isinstance(frame, SpeechControlParamsFrame): diff --git a/src/pipecat/services/azure/realtime/llm.py b/src/pipecat/services/azure/realtime/llm.py index 1193b82d4..66ba95eea 100644 --- a/src/pipecat/services/azure/realtime/llm.py +++ b/src/pipecat/services/azure/realtime/llm.py @@ -38,7 +38,7 @@ class AzureRealtimeLLMService(OpenAIRealtimeLLMService): Args: api_key: The API key for the Azure OpenAI service. base_url: The full Azure WebSocket endpoint URL including api-version and deployment. - Example: "wss://my-project.openai.azure.com/openai/realtime?api-version=2024-10-01-preview&deployment=my-realtime-deployment" + Example: "wss://my-project.openai.azure.com/openai/realtime?api-version=2025-04-01-preview&deployment=my-realtime-deployment" **kwargs: Additional arguments passed to parent OpenAIRealtimeLLMService. """ super().__init__(base_url=base_url, api_key=api_key, **kwargs) diff --git a/src/pipecat/services/openai/realtime/context.py b/src/pipecat/services/openai/realtime/context.py index cb1c0a9f5..91c6e74d5 100644 --- a/src/pipecat/services/openai/realtime/context.py +++ b/src/pipecat/services/openai/realtime/context.py @@ -4,7 +4,85 @@ # SPDX-License-Identifier: BSD 2-Clause License # -"""OpenAI Realtime LLM context and aggregator implementations.""" +"""OpenAI Realtime LLM context and aggregator implementations. + +.. deprecated:: 0.0.92 + OpenAI Realtime no longer uses types from this module under the hood. + It now uses `LLMContext` and `LLMContextAggregatorPair`. + Using the new patterns should allow you to not need types from this module. + + BEFORE: + ``` + # Setup + context = OpenAILLMContext(messages, tools) + context_aggregator = llm.create_context_aggregator(context) + + # Context aggregator type + context_aggregator: OpenAIContextAggregatorPair + + # Context frame type + frame: OpenAILLMContextFrame + + # Context type + context: OpenAIRealtimeLLMContext + # or + context: OpenAILLMContext + ``` + + AFTER: + ``` + # Setup + context = LLMContext(messages, tools) + context_aggregator = LLMContextAggregatorPair(context) + + # Context aggregator type + context_aggregator: LLMContextAggregatorPair + + # Context frame type + frame: LLMContextFrame + + # Context type + context: LLMContext + ``` +""" + +import warnings + +with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "Types in pipecat.services.openai.realtime.llm (or " + "pipecat.services.openai_realtime.llm) are deprecated. \n" + "OpenAI Realtime no longer uses types from this module under the hood. \n" + "It now uses `LLMContext` and `LLMContextAggregatorPair`. \n" + "Using the new patterns should allow you to not need types from this module.\n\n" + "BEFORE:\n" + "```\n" + "# Setup\n" + "context = OpenAILLMContext(messages, tools)\n" + "context_aggregator = llm.create_context_aggregator(context)\n\n" + "# Context aggregator type\n" + "context_aggregator: OpenAIContextAggregatorPair\n\n" + "# Context frame type\n" + "frame: OpenAILLMContextFrame\n\n" + "# Context type\n" + "context: OpenAIRealtimeLLMContext\n" + "# or\n" + "context: OpenAILLMContext\n\n" + "```\n\n" + "AFTER:\n" + "```\n" + "# Setup\n" + "context = LLMContext(messages, tools)\n" + "context_aggregator = LLMContextAggregatorPair(context)\n\n" + "# Context aggregator type\n" + "context_aggregator: LLMContextAggregatorPair\n\n" + "# Context frame type\n" + "frame: LLMContextFrame\n\n" + "# Context type\n" + "context: LLMContext\n\n" + "```\n", + ) import copy import json diff --git a/src/pipecat/services/openai/realtime/frames.py b/src/pipecat/services/openai/realtime/frames.py index 8617c6efd..39cfd9757 100644 --- a/src/pipecat/services/openai/realtime/frames.py +++ b/src/pipecat/services/openai/realtime/frames.py @@ -4,7 +4,28 @@ # SPDX-License-Identifier: BSD 2-Clause License # -"""Custom frame types for OpenAI Realtime API integration.""" +"""Custom frame types for OpenAI Realtime API integration. + +.. deprecated:: 0.0.92 + OpenAI Realtime no longer uses types from this module under the hood. + + It now works more like most LLM services in Pipecat, relying on updates to + its context, pushed by context aggregators, to update its internal state. + + Listen for `LLMContextFrame`s for context updates. +""" + +import warnings + +with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "Types in pipecat.services.openai.realtime.frames are deprecated. \n" + "OpenAI Realtime no longer uses types from this module under the hood. \n\n" + "It now works more like other LLM services in Pipecat, relying on updates to \n" + "its context, pushed by context aggregators, to update its internal state.\n\n" + "Listen for `LLMContextFrame`s for context updates.\n" + ) from dataclasses import dataclass from typing import TYPE_CHECKING diff --git a/src/pipecat/services/openai/realtime/llm.py b/src/pipecat/services/openai/realtime/llm.py index 8b3d500eb..012604eb8 100644 --- a/src/pipecat/services/openai/realtime/llm.py +++ b/src/pipecat/services/openai/realtime/llm.py @@ -14,7 +14,9 @@ from typing import Optional from loguru import logger -from pipecat.adapters.services.open_ai_realtime_adapter import OpenAIRealtimeLLMAdapter +from pipecat.adapters.services.open_ai_realtime_adapter import ( + OpenAIRealtimeLLMAdapter, +) from pipecat.frames.frames import ( BotStoppedSpeakingFrame, CancelFrame, @@ -41,10 +43,12 @@ from pipecat.frames.frames import ( UserStoppedSpeakingFrame, ) from pipecat.metrics.metrics import LLMTokenUsage +from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.aggregators.llm_response import ( LLMAssistantAggregatorParams, LLMUserAggregatorParams, ) +from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContext, OpenAILLMContextFrame, @@ -57,12 +61,6 @@ from pipecat.utils.time import time_now_iso8601 from pipecat.utils.tracing.service_decorators import traced_openai_realtime, traced_stt from . import events -from .context import ( - OpenAIRealtimeAssistantContextAggregator, - OpenAIRealtimeLLMContext, - OpenAIRealtimeUserContextAggregator, -) -from .frames import RealtimeFunctionCallResultFrame, RealtimeMessagesUpdateFrame try: from websockets.asyncio.client import connect as websocket_connect @@ -108,22 +106,39 @@ class OpenAIRealtimeLLMService(LLMService): base_url: str = "wss://api.openai.com/v1/realtime", session_properties: Optional[events.SessionProperties] = None, start_audio_paused: bool = False, - send_transcription_frames: bool = True, + send_transcription_frames: Optional[bool] = None, **kwargs, ): """Initialize the OpenAI Realtime LLM service. Args: api_key: OpenAI API key for authentication. - model: OpenAI model name. Defaults to "gpt-4o-realtime-preview-2025-06-03". + model: OpenAI model name. Defaults to "gpt-realtime". base_url: WebSocket base URL for the realtime API. Defaults to "wss://api.openai.com/v1/realtime". session_properties: Configuration properties for the realtime session. If None, uses default SessionProperties. start_audio_paused: Whether to start with audio input paused. Defaults to False. - send_transcription_frames: Whether to emit transcription frames. Defaults to True. + send_transcription_frames: Whether to emit transcription frames. + + .. deprecated:: 0.0.92 + This parameter is deprecated and will be removed in a future version. + Transcription frames are always sent. + **kwargs: Additional arguments passed to parent LLMService. """ + if send_transcription_frames is not None: + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "`send_transcription_frames` is deprecated and will be removed in a future version. " + "Transcription frames are always sent.", + DeprecationWarning, + stacklevel=2, + ) + full_url = f"{base_url}?model={model}" super().__init__(base_url=full_url, **kwargs) @@ -135,10 +150,11 @@ class OpenAIRealtimeLLMService(LLMService): session_properties or events.SessionProperties() ) 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._context: LLMContext = None + + self._llm_needs_conversation_setup = True self._disconnecting = False self._api_session_ready = False @@ -148,8 +164,8 @@ class OpenAIRealtimeLLMService(LLMService): self._current_audio_response = None self._messages_added_manually = {} - self._user_and_response_message_tuple = None self._pending_function_calls = {} # Track function calls by call_id + self._completed_tool_calls = set() self._register_event_handler("on_conversation_item_created") self._register_event_handler("on_conversation_item_updated") @@ -347,22 +363,13 @@ class OpenAIRealtimeLLMService(LLMService): if isinstance(frame, TranscriptionFrame): pass - elif isinstance(frame, OpenAILLMContextFrame): - context: OpenAIRealtimeLLMContext = OpenAIRealtimeLLMContext.upgrade_to_realtime( + elif isinstance(frame, (LLMContextFrame, OpenAILLMContextFrame)): + context = ( frame.context + if isinstance(frame, LLMContextFrame) + else LLMContext.from_openai_context(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, LLMContextFrame): - raise NotImplementedError( - "Universal LLMContext is not yet supported for OpenAI Realtime." - ) + await self._handle_context(context) elif isinstance(frame, InputAudioRawFrame): if not self._audio_input_paused: await self._send_user_audio(frame) @@ -376,29 +383,33 @@ class OpenAIRealtimeLLMService(LLMService): await self._handle_bot_stopped_speaking() elif isinstance(frame, LLMMessagesAppendFrame): await self._handle_messages_append(frame) - elif isinstance(frame, RealtimeMessagesUpdateFrame): - self._context = frame.context elif isinstance(frame, LLMUpdateSettingsFrame): self._session_properties = events.SessionProperties(**frame.settings) await self._update_settings() elif isinstance(frame, LLMSetToolsFrame): await self._update_settings() - elif isinstance(frame, RealtimeFunctionCallResultFrame): - await self._handle_function_call_result(frame.result_frame) await self.push_frame(frame, direction) + async def _handle_context(self, context: LLMContext): + if not self._context: + # We got our initial context + self._context = context + # Initialize our bookkeeping of already-completed tool calls in + # the context + await self._process_completed_function_calls(send_new_results=False) + # Run the LLM at next opportunity + await self._create_response() + else: + # We got an updated context. + # This may contain a new user message or tool call result. + self._context = context + # Send results for newly-completed function calls, if any. + await self._process_completed_function_calls(send_new_results=True) + 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 # @@ -439,16 +450,21 @@ class OpenAIRealtimeLLMService(LLMService): if self._receive_task: await self.cancel_task(self._receive_task, timeout=1.0) self._receive_task = None + self._completed_tool_calls = set() 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: + if not self._disconnecting and self._websocket: await self._websocket.send(json.dumps(realtime_message)) except Exception as e: - if self._disconnecting: + if self._disconnecting or not self._websocket: + # We're in the process of disconnecting. + # (If not self._websocket, that could indicate that we + # somehow *started* the websocket send attempt while we still + # had a connection) 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 @@ -459,13 +475,20 @@ class OpenAIRealtimeLLMService(LLMService): 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 + + if self._context: + adapter: OpenAIRealtimeLLMAdapter = self.get_llm_adapter() + llm_invocation_params = adapter.get_llm_invocation_params(self._context) + + # tools given in the context override the tools in the session properties + if llm_invocation_params["tools"]: + settings.tools = llm_invocation_params["tools"] + + # instructions in the context come from an initial "system" message in the + # messages list, and override instructions in the session properties + if llm_invocation_params["system_instruction"]: + settings.instructions = llm_invocation_params["system_instruction"] + await self.send_client_event(events.SessionUpdateEvent(session=settings)) # @@ -571,12 +594,7 @@ class OpenAIRealtimeLLMService(LLMService): 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": + if evt.item.role == "assistant": self._current_assistant_response = evt.item await self.push_frame(LLMFullResponseStartFrame()) @@ -587,11 +605,11 @@ class OpenAIRealtimeLLMService(LLMService): # For now, no additional logic needed beyond the event handler call async def _handle_evt_input_audio_transcription_delta(self, evt): - if self._send_transcription_frames: - await self.push_frame( - # no way to get a language code? - InterimTranscriptionFrame(evt.delta, "", time_now_iso8601(), result=evt) - ) + await self.push_frame( + # no way to get a language code? + InterimTranscriptionFrame(evt.delta, "", time_now_iso8601(), result=evt), + direction=FrameDirection.UPSTREAM, + ) @traced_stt async def _handle_user_transcription( @@ -608,22 +626,12 @@ class OpenAIRealtimeLLMService(LLMService): """ await self._call_event_handler("on_conversation_item_updated", evt.item_id, None) - if self._send_transcription_frames: - await self.push_frame( - # no way to get a language code? - TranscriptionFrame(evt.transcript, "", time_now_iso8601(), result=evt) - ) - await self._handle_user_transcription(evt.transcript, True, Language.EN) - 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) - else: - # User message without preceding conversation.item.created. Bug? - logger.warning(f"Transcript for unknown user message: {evt}") + await self.push_frame( + # no way to get a language code? + TranscriptionFrame(evt.transcript, "", time_now_iso8601(), result=evt), + FrameDirection.UPSTREAM, + ) + await self._handle_user_transcription(evt.transcript, True, Language.EN) async def _handle_conversation_item_retrieved(self, evt: events.ConversationItemRetrieved): futures = self._retrieve_conversation_item_futures.pop(evt.item.id, None) @@ -653,26 +661,17 @@ class OpenAIRealtimeLLMService(LLMService): # response content for item in evt.response.output: await self._call_event_handler("on_conversation_item_updated", item.id, item) - 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) - else: - # Response message without preceding user message (standalone response) - # Function calls in this response were already processed immediately when arguments were complete - logger.debug(f"Handling standalone response: {evt.response.id}") async def _handle_evt_text_delta(self, evt): + # We receive text deltas (as opposed to audio transcript deltas) when + # the output modality is "text" if evt.delta: await self.push_frame(LLMTextFrame(evt.delta)) async def _handle_evt_audio_transcript_delta(self, evt): + # We receive audio transcript deltas (as opposed to text deltas) when + # the output modality is "audio" (the default) if evt.delta: - await self.push_frame(LLMTextFrame(evt.delta)) await self.push_frame(TTSTextFrame(evt.delta)) async def _handle_evt_function_call_arguments_done(self, evt): @@ -760,9 +759,11 @@ class OpenAIRealtimeLLMService(LLMService): """ logger.debug("Resetting conversation") await self._disconnect() - if self._context: - self._context.llm_needs_settings_update = True - self._context.llm_needs_initial_messages = True + + # Prepare to setup server-side conversation from local context again + self._llm_needs_conversation_setup = True + await self._process_completed_function_calls(send_new_results=False) + await self._connect() @traced_openai_realtime(operation="llm_request") @@ -771,19 +772,29 @@ class OpenAIRealtimeLLMService(LLMService): self._run_llm_when_api_session_ready = True return - if self._context.llm_needs_initial_messages: - messages = self._context.get_messages_for_initializing_history() + adapter: OpenAIRealtimeLLMAdapter = self.get_llm_adapter() + + # Configure the LLM for this session if needed + if self._llm_needs_conversation_setup: + logger.debug( + f"Setting up conversation on OpenAI Realtime LLM service with initial messages: {adapter.get_messages_for_logging(self._context)}" + ) + + # Send initial messages + llm_invocation_params = adapter.get_llm_invocation_params(self._context) + messages = llm_invocation_params["messages"] 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: + # Send new settings if needed await self._update_settings() - self._context.llm_needs_settings_update = False - logger.debug(f"Creating response: {self._context.get_messages_for_logging()}") + # We're done configuring the LLM for this session + self._llm_needs_conversation_setup = False + + logger.debug(f"Creating response") await self.push_frame(LLMFullResponseStartFrame()) await self.start_processing_metrics() @@ -794,19 +805,50 @@ class OpenAIRealtimeLLMService(LLMService): ) ) + async def _process_completed_function_calls(self, send_new_results: bool): + # Check for set of completed function calls in the context + sent_new_result = False + for message in self._context.get_messages(): + if message.get("role") and message.get("content") != "IN_PROGRESS": + tool_call_id = message.get("tool_call_id") + if tool_call_id and tool_call_id not in self._completed_tool_calls: + # Found a newly-completed function call - send the result to the service + if send_new_results: + sent_new_result = True + await self._send_tool_result(tool_call_id, message.get("content")) + self._completed_tool_calls.add(tool_call_id) + + # If we reported any new tool call results to the service, trigger + # another response + if sent_new_result: + await self._create_response() + async def _send_user_audio(self, frame): payload = base64.b64encode(frame.audio).decode("utf-8") await self.send_client_event(events.InputAudioBufferAppendEvent(audio=payload)) + async def _send_tool_result(self, tool_call_id: str, result: str): + item = events.ConversationItem( + type="function_call_output", + call_id=tool_call_id, + output=json.dumps(result), + ) + await self.send_client_event(events.ConversationItemCreateEvent(item=item)) + def create_context_aggregator( self, context: OpenAILLMContext, *, user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(), assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(), - ) -> OpenAIContextAggregatorPair: + ) -> LLMContextAggregatorPair: """Create an instance of OpenAIContextAggregatorPair from an OpenAILLMContext. + NOTE: this method exists only for backward compatibility. New code + should instead do: + context = LLMContext(...) + context_aggregator = LLMContextAggregatorPair(context) + Constructor keyword arguments for both the user and assistant aggregators can be provided. Args: @@ -819,11 +861,41 @@ class OpenAIRealtimeLLMService(LLMService): the user and one for the assistant, encapsulated in an OpenAIContextAggregatorPair. """ - context.set_llm_adapter(self.get_llm_adapter()) - - OpenAIRealtimeLLMContext.upgrade_to_realtime(context) - user = OpenAIRealtimeUserContextAggregator(context, params=user_params) + # Log warning about transcription frame direction change in 0.0.92. + # We're putting this warning here rather than in the constructor so + # that it shows up for folks who haven't updated their code at all + # since 0.0.92, gives them a way to acknowledge and dismiss the + # warning, and encourages adoption of a new preferred pattern. + logger.warning( + "As of version 0.0.92, TranscriptionFrames and InterimTranscriptionFrames " + "now go upstream from OpenAIRealtimeLLMService, so if you're using " + "TranscriptProcessor, say, you'll want to adjust accordingly:\n\n" + "pipeline = Pipeline(\n" + " [\n" + " transport.input(),\n" + " context_aggregator.user(),\n\n" + " # BEFORE\n" + " llm,\n" + " transcript.user(),\n\n" + " # AFTER\n" + " transcript.user(),\n" + " llm,\n\n" + " transport.output(),\n" + " transcript.assistant(),\n" + " context_aggregator.assistant(),\n" + " ]\n" + ")\n\n" + "Also, LLMTextFrames are no longer pushed from " + "OpenAIRealtimeLLMService when it's configured with " + "output_modalities=['audio']. Listen for TTSTextFrames instead.\n\n" + "Once you've made the appropriate changes (if needed), you can " + "dismiss this warning by updating to the new context-setup pattern:\n\n" + " context = LLMContext(messages, tools)\n" + " context_aggregator = LLMContextAggregatorPair(context)\n" + ) + context = LLMContext.from_openai_context(context) assistant_params.expect_stripped_words = False - assistant = OpenAIRealtimeAssistantContextAggregator(context, params=assistant_params) - return OpenAIContextAggregatorPair(_user=user, _assistant=assistant) + return LLMContextAggregatorPair( + context, user_params=user_params, assistant_params=assistant_params + ) diff --git a/src/pipecat/services/openai_realtime/context.py b/src/pipecat/services/openai_realtime/context.py index 58f1cfe75..79a01b980 100644 --- a/src/pipecat/services/openai_realtime/context.py +++ b/src/pipecat/services/openai_realtime/context.py @@ -4,18 +4,15 @@ # SPDX-License-Identifier: BSD 2-Clause License # -"""OpenAI Realtime LLM context and aggregator implementations.""" +"""OpenAI Realtime LLM context and aggregator implementations. -import warnings +.. deprecated:: 0.0.91 + OpenAI Realtime no longer uses types from this module under the hood. + It now uses `LLMContext` and `LLMContextAggregatorPair`. + Using the new patterns should allow you to not need types from this module. + + See deprecation warning in pipecat.services.openai.realtime.context for + more details. +""" from pipecat.services.openai.realtime.context import * - -with warnings.catch_warnings(): - warnings.simplefilter("always") - warnings.warn( - "Types in pipecat.services.openai_realtime.context are deprecated. " - "Please use the equivalent types from " - "pipecat.services.openai.realtime.context instead.", - DeprecationWarning, - stacklevel=2, - ) diff --git a/src/pipecat/utils/tracing/service_decorators.py b/src/pipecat/utils/tracing/service_decorators.py index cf1ba912c..3935a4afc 100644 --- a/src/pipecat/utils/tracing/service_decorators.py +++ b/src/pipecat/utils/tracing/service_decorators.py @@ -905,7 +905,9 @@ def traced_openai_realtime(operation: str) -> Callable: # Capture context messages being sent if hasattr(self, "_context") and self._context: try: - messages = self._context.get_messages_for_logging() + messages = self.get_llm_adapter().get_messages_for_logging( + self._context + ) if messages: operation_attrs["context_messages"] = json.dumps(messages) except Exception as e: