diff --git a/CHANGELOG.md b/CHANGELOG.md index 9c4940edf..61f3b6e76 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added a new frame `FunctionCallsStartedFrame`. This frame is pushed both + upstream and downstream from the LLM service to indicate that one or more + function calls are going to be executed. + +- Added LLM services `on_function_calls_started` event. This event will be + triggered when the LLM service receives function calls from the model and is + going to start executing them. + +- Function calls can now be executed sequentially (in the order received in the + completion) by passing `run_in_parallel=False` when creating your LLM + service. By default, if the LLM completion returns 2 or more function calls + they run concurrently. In both cases, concurrently and sequentially, a new LLM + completion will run when the last function call finishes. + - Added OpenTelemetry tracing for `GeminiMultimodalLiveLLMService` and `OpenAIRealtimeBetaLLMService`. @@ -53,6 +67,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- Function calls are now cancelled by default if there's an interruption. To + disable this behavior you can set `cancel_on_interruption=False` when + registering the function call. Since function calls are executed as tasks you + can tell if a function call has been cancelled by catching the + `asyncio.CancelledError` exception (and don't forget to raise it again!). + - Updated OpenTelemetry tracing attribute `metrics.ttfb_ms` to `metrics.ttfb`. The attribute reports TTFB in seconds. diff --git a/examples/foundational/14-function-calling.py b/examples/foundational/14-function-calling.py index 7c80416b1..b4e09c99a 100644 --- a/examples/foundational/14-function-calling.py +++ b/examples/foundational/14-function-calling.py @@ -30,10 +30,13 @@ load_dotenv(override=True) async def fetch_weather_from_api(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) await params.result_callback({"conditions": "nice", "temperature": "75"}) +async def fetch_restaurant_recommendation(params: FunctionCallParams): + await params.result_callback({"name": "The Golden Dragon"}) + + # We store functions so objects (e.g. SileroVADAnalyzer) don't get # instantiated. The function will be called when the desired transport gets # selected. @@ -71,6 +74,11 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # You can also register a function_name of None to get all functions # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) + + @llm.event_handler("on_function_calls_started") + async def on_function_calls_started(service, function_calls): + await tts.queue_frame(TTSSpeakFrame("Let me check on that.")) weather_function = FunctionSchema( name="get_current_weather", @@ -88,7 +96,18 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si }, required=["location", "format"], ) - tools = ToolsSchema(standard_tools=[weather_function]) + restaurant_function = FunctionSchema( + name="get_restaurant_recommendation", + description="Get a restaurant recommendation", + properties={ + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + }, + required=["location"], + ) + tools = ToolsSchema(standard_tools=[weather_function, restaurant_function]) messages = [ { diff --git a/examples/foundational/14a-function-calling-anthropic.py b/examples/foundational/14a-function-calling-anthropic.py index f46a42db1..bf34e211c 100644 --- a/examples/foundational/14a-function-calling-anthropic.py +++ b/examples/foundational/14a-function-calling-anthropic.py @@ -33,6 +33,10 @@ async def get_weather(params: FunctionCallParams): await params.result_callback(f"The weather in {location} is currently 72 degrees and sunny.") +async def fetch_restaurant_recommendation(params: FunctionCallParams): + await params.result_callback({"name": "The Golden Dragon"}) + + # We store functions so objects (e.g. SileroVADAnalyzer) don't get # instantiated. The function will be called when the desired transport gets # selected. @@ -66,9 +70,11 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si ) llm = AnthropicLLMService( - api_key=os.getenv("ANTHROPIC_API_KEY"), model="claude-3-7-sonnet-latest" + api_key=os.getenv("ANTHROPIC_API_KEY"), + model="claude-3-7-sonnet-latest", ) llm.register_function("get_weather", get_weather) + llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) weather_function = FunctionSchema( name="get_weather", @@ -81,7 +87,18 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si }, required=["location"], ) - tools = ToolsSchema(standard_tools=[weather_function]) + restaurant_function = FunctionSchema( + name="get_restaurant_recommendation", + description="Get a restaurant recommendation", + properties={ + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + }, + required=["location"], + ) + tools = ToolsSchema(standard_tools=[weather_function, restaurant_function]) # todo: test with very short initial user message diff --git a/examples/foundational/14c-function-calling-together.py b/examples/foundational/14c-function-calling-together.py index 54545b0b2..1188f32e2 100644 --- a/examples/foundational/14c-function-calling-together.py +++ b/examples/foundational/14c-function-calling-together.py @@ -30,7 +30,6 @@ load_dotenv(override=True) async def fetch_weather_from_api(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -74,6 +73,10 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + @llm.event_handler("on_function_calls_started") + async def on_function_calls_started(service, function_calls): + await tts.queue_frame(TTSSpeakFrame("Let me check on that.")) + weather_function = FunctionSchema( name="get_current_weather", description="Get the current weather", diff --git a/examples/foundational/14e-function-calling-gemini.py b/examples/foundational/14e-function-calling-gemini.py index 4e6a582f3..ccbe952fb 100644 --- a/examples/foundational/14e-function-calling-gemini.py +++ b/examples/foundational/14e-function-calling-gemini.py @@ -35,11 +35,14 @@ client_id = "" async def get_weather(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) location = params.arguments["location"] await params.result_callback(f"The weather in {location} is currently 72 degrees and sunny.") +async def fetch_restaurant_recommendation(params: FunctionCallParams): + await params.result_callback({"name": "The Golden Dragon"}) + + async def get_image(params: FunctionCallParams): question = params.arguments["question"] logger.debug(f"Requesting image with user_id={client_id}, question={question}") @@ -93,6 +96,11 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si llm = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY"), model="gemini-2.0-flash-001") llm.register_function("get_weather", get_weather) llm.register_function("get_image", get_image) + llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) + + @llm.event_handler("on_function_calls_started") + async def on_function_calls_started(service, function_calls): + await tts.queue_frame(TTSSpeakFrame("Let me check on that.")) weather_function = FunctionSchema( name="get_weather", @@ -110,6 +118,17 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si }, required=["location", "format"], ) + restaurant_function = FunctionSchema( + name="get_restaurant_recommendation", + description="Get a restaurant recommendation", + properties={ + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + }, + required=["location"], + ) get_image_function = FunctionSchema( name="get_image", description="Get an image from the video stream.", @@ -121,14 +140,14 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si }, required=["question"], ) - tools = ToolsSchema(standard_tools=[weather_function, get_image_function]) + tools = ToolsSchema(standard_tools=[weather_function, get_image_function, restaurant_function]) system_prompt = """\ You are a helpful assistant who converses with a user and answers questions. Respond concisely to general questions. Your response will be turned into speech so use only simple words and punctuation. -You have access to two tools: get_weather and get_image. +You have access to three tools: get_weather, get_restaurant_recommendation, and get_image. You can respond to questions about the weather using the get_weather tool. diff --git a/examples/foundational/14f-function-calling-groq.py b/examples/foundational/14f-function-calling-groq.py index 698845028..3ff56a915 100644 --- a/examples/foundational/14f-function-calling-groq.py +++ b/examples/foundational/14f-function-calling-groq.py @@ -31,7 +31,6 @@ load_dotenv(override=True) async def fetch_weather_from_api(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -74,6 +73,10 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + @llm.event_handler("on_function_calls_started") + async def on_function_calls_started(service, function_calls): + await tts.queue_frame(TTSSpeakFrame("Let me check on that.")) + weather_function = FunctionSchema( name="get_current_weather", description="Get the current weather", diff --git a/examples/foundational/14h-function-calling-azure.py b/examples/foundational/14h-function-calling-azure.py index a54fb1429..25cf7defa 100644 --- a/examples/foundational/14h-function-calling-azure.py +++ b/examples/foundational/14h-function-calling-azure.py @@ -30,7 +30,6 @@ load_dotenv(override=True) async def fetch_weather_from_api(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -75,6 +74,10 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + @llm.event_handler("on_function_calls_started") + async def on_function_calls_started(service, function_calls): + await tts.queue_frame(TTSSpeakFrame("Let me check on that.")) + weather_function = FunctionSchema( name="get_current_weather", description="Get the current weather", diff --git a/examples/foundational/14i-function-calling-fireworks.py b/examples/foundational/14i-function-calling-fireworks.py index 5467e685a..028d9fa64 100644 --- a/examples/foundational/14i-function-calling-fireworks.py +++ b/examples/foundational/14i-function-calling-fireworks.py @@ -30,7 +30,6 @@ load_dotenv(override=True) async def fetch_weather_from_api(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -74,6 +73,10 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + @llm.event_handler("on_function_calls_started") + async def on_function_calls_started(service, function_calls): + await tts.queue_frame(TTSSpeakFrame("Let me check on that.")) + weather_function = FunctionSchema( name="get_current_weather", description="Get the current weather", diff --git a/examples/foundational/14j-function-calling-nim.py b/examples/foundational/14j-function-calling-nim.py index f9b437f83..d254e0d4f 100644 --- a/examples/foundational/14j-function-calling-nim.py +++ b/examples/foundational/14j-function-calling-nim.py @@ -30,7 +30,6 @@ load_dotenv(override=True) async def fetch_weather_from_api(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -72,6 +71,10 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + @llm.event_handler("on_function_calls_started") + async def on_function_calls_started(service, function_calls): + await tts.queue_frame(TTSSpeakFrame("Let me check on that.")) + weather_function = FunctionSchema( name="get_current_weather", description="Get the current weather", diff --git a/examples/foundational/14k-function-calling-cerebras.py b/examples/foundational/14k-function-calling-cerebras.py index 7aa5a2af4..70ffceffd 100644 --- a/examples/foundational/14k-function-calling-cerebras.py +++ b/examples/foundational/14k-function-calling-cerebras.py @@ -30,7 +30,6 @@ load_dotenv(override=True) async def fetch_weather_from_api(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -71,6 +70,10 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + @llm.event_handler("on_function_calls_started") + async def on_function_calls_started(service, function_calls): + await tts.queue_frame(TTSSpeakFrame("Let me check on that.")) + weather_function = FunctionSchema( name="get_current_weather", description="Get the current weather", diff --git a/examples/foundational/14l-function-calling-deepseek.py b/examples/foundational/14l-function-calling-deepseek.py index 0daa814e6..7e992942d 100644 --- a/examples/foundational/14l-function-calling-deepseek.py +++ b/examples/foundational/14l-function-calling-deepseek.py @@ -30,7 +30,6 @@ load_dotenv(override=True) async def fetch_weather_from_api(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -71,6 +70,10 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + @llm.event_handler("on_function_calls_started") + async def on_function_calls_started(service, function_calls): + await tts.queue_frame(TTSSpeakFrame("Let me check on that.")) + weather_function = FunctionSchema( name="get_current_weather", description="Get the current weather", diff --git a/examples/foundational/14m-function-calling-openrouter.py b/examples/foundational/14m-function-calling-openrouter.py index 13487b3b6..023f725f6 100644 --- a/examples/foundational/14m-function-calling-openrouter.py +++ b/examples/foundational/14m-function-calling-openrouter.py @@ -30,7 +30,6 @@ load_dotenv(override=True) async def fetch_weather_from_api(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -75,6 +74,10 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + @llm.event_handler("on_function_calls_started") + async def on_function_calls_started(service, function_calls): + await tts.queue_frame(TTSSpeakFrame("Let me check on that.")) + weather_function = FunctionSchema( name="get_current_weather", description="Get the current weather", diff --git a/examples/foundational/14o-function-calling-gemini-openai-format.py b/examples/foundational/14o-function-calling-gemini-openai-format.py index 8e1e0fdf1..8f0036e07 100644 --- a/examples/foundational/14o-function-calling-gemini-openai-format.py +++ b/examples/foundational/14o-function-calling-gemini-openai-format.py @@ -30,7 +30,6 @@ load_dotenv(override=True) async def fetch_weather_from_api(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -71,6 +70,10 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + @llm.event_handler("on_function_calls_started") + async def on_function_calls_started(service, function_calls): + await tts.queue_frame(TTSSpeakFrame("Let me check on that.")) + weather_function = FunctionSchema( name="get_current_weather", description="Get the current weather", diff --git a/examples/foundational/14p-function-calling-gemini-vertex-ai.py b/examples/foundational/14p-function-calling-gemini-vertex-ai.py index 42567787a..4348b663a 100644 --- a/examples/foundational/14p-function-calling-gemini-vertex-ai.py +++ b/examples/foundational/14p-function-calling-gemini-vertex-ai.py @@ -30,7 +30,6 @@ load_dotenv(override=True) async def fetch_weather_from_api(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -76,6 +75,10 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + @llm.event_handler("on_function_calls_started") + async def on_function_calls_started(service, function_calls): + await tts.queue_frame(TTSSpeakFrame("Let me check on that.")) + weather_function = FunctionSchema( name="get_current_weather", description="Get the current weather", diff --git a/examples/foundational/14q-function-calling-qwen.py b/examples/foundational/14q-function-calling-qwen.py index d3740d25f..6e07a97e1 100644 --- a/examples/foundational/14q-function-calling-qwen.py +++ b/examples/foundational/14q-function-calling-qwen.py @@ -30,7 +30,6 @@ load_dotenv(override=True) async def fetch_weather_from_api(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -72,6 +71,10 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + @llm.event_handler("on_function_calls_started") + async def on_function_calls_started(service, function_calls): + await tts.queue_frame(TTSSpeakFrame("Let me check on that.")) + weather_function = FunctionSchema( name="get_current_weather", description="Get the current weather", diff --git a/examples/foundational/14r-function-calling-aws.py b/examples/foundational/14r-function-calling-aws.py index f7796157a..4443bec27 100644 --- a/examples/foundational/14r-function-calling-aws.py +++ b/examples/foundational/14r-function-calling-aws.py @@ -32,6 +32,10 @@ async def fetch_weather_from_api(params: FunctionCallParams): await params.result_callback({"conditions": "nice", "temperature": "75"}) +async def fetch_restaurant_recommendation(params: FunctionCallParams): + await params.result_callback({"name": "The Golden Dragon"}) + + # We store functions so objects (e.g. SileroVADAnalyzer) don't get # instantiated. The function will be called when the desired transport gets # selected. @@ -74,6 +78,7 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # You can also register a function_name of None to get all functions # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) weather_function = FunctionSchema( name="get_current_weather", @@ -91,7 +96,18 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si }, required=["location", "format"], ) - tools = ToolsSchema(standard_tools=[weather_function]) + restaurant_function = FunctionSchema( + name="get_restaurant_recommendation", + description="Get a restaurant recommendation", + properties={ + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + }, + required=["location"], + ) + tools = ToolsSchema(standard_tools=[weather_function, restaurant_function]) messages = [ { diff --git a/examples/foundational/19-openai-realtime-beta.py b/examples/foundational/19-openai-realtime-beta.py index f1dd7269d..81e8dc6c3 100644 --- a/examples/foundational/19-openai-realtime-beta.py +++ b/examples/foundational/19-openai-realtime-beta.py @@ -45,6 +45,10 @@ async def fetch_weather_from_api(params: FunctionCallParams): ) +async def fetch_restaurant_recommendation(params: FunctionCallParams): + await params.result_callback({"name": "The Golden Dragon"}) + + weather_function = FunctionSchema( name="get_current_weather", description="Get the current weather", @@ -62,8 +66,20 @@ weather_function = FunctionSchema( required=["location", "format"], ) +restaurant_function = FunctionSchema( + name="get_restaurant_recommendation", + description="Get a restaurant recommendation", + properties={ + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + }, + required=["location"], +) + # Create tools schema -tools = ToolsSchema(standard_tools=[weather_function]) +tools = ToolsSchema(standard_tools=[weather_function, restaurant_function]) # We store functions so objects (e.g. SileroVADAnalyzer) don't get @@ -100,7 +116,7 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # turn_detection=False, input_audio_noise_reduction=InputAudioNoiseReduction(type="near_field"), # tools=tools, - instructions="""Your knowledge cutoff is 2023-10. You are a helpful and friendly AI. + instructions="""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 @@ -113,6 +129,10 @@ 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.""", ) @@ -125,6 +145,7 @@ Remember, your responses should be short. Just one or two sentences, usually.""" # 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("get_restaurant_recommendation", fetch_restaurant_recommendation) # Create a standard OpenAI LLM context object using the normal messages format. The # OpenAIRealtimeBetaLLMService will convert this internally to messages that the diff --git a/examples/foundational/19a-azure-realtime-beta.py b/examples/foundational/19a-azure-realtime-beta.py index 76180e397..54f0302e5 100644 --- a/examples/foundational/19a-azure-realtime-beta.py +++ b/examples/foundational/19a-azure-realtime-beta.py @@ -43,6 +43,10 @@ async def fetch_weather_from_api(params: FunctionCallParams): ) +async def fetch_restaurant_recommendation(params: FunctionCallParams): + await params.result_callback({"name": "The Golden Dragon"}) + + # Define weather function using standardized schema weather_function = FunctionSchema( name="get_current_weather", @@ -61,8 +65,20 @@ weather_function = FunctionSchema( required=["location", "format"], ) +restaurant_function = FunctionSchema( + name="get_restaurant_recommendation", + description="Get a restaurant recommendation", + properties={ + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + }, + required=["location"], +) + # Create tools schema -tools = ToolsSchema(standard_tools=[weather_function]) +tools = ToolsSchema(standard_tools=[weather_function, restaurant_function]) # We store functions so objects (e.g. SileroVADAnalyzer) don't get @@ -98,7 +114,7 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # 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. + instructions="""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 @@ -111,6 +127,10 @@ 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.""", ) @@ -124,6 +144,7 @@ Remember, your responses should be short. Just one or two sentences, usually.""" # 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("get_restaurant_recommendation", fetch_restaurant_recommendation) # Create a standard OpenAI LLM context object using the normal messages format. The # OpenAIRealtimeBetaLLMService will convert this internally to messages that the diff --git a/examples/foundational/22b-natural-conversation-proposal.py b/examples/foundational/22b-natural-conversation-proposal.py index 58259b266..1b03f341e 100644 --- a/examples/foundational/22b-natural-conversation-proposal.py +++ b/examples/foundational/22b-natural-conversation-proposal.py @@ -198,7 +198,6 @@ class OutputGate(FrameProcessor): async def fetch_weather_from_api(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -245,6 +244,10 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + @llm.event_handler("on_function_calls_started") + async def on_function_calls_started(service, function_calls): + await tts.queue_frame(TTSSpeakFrame("Let me check on that.")) + tools = [ ChatCompletionToolParam( type="function", diff --git a/examples/foundational/22c-natural-conversation-mixed-llms.py b/examples/foundational/22c-natural-conversation-mixed-llms.py index 7d6a1ce18..0bcbf75bd 100644 --- a/examples/foundational/22c-natural-conversation-mixed-llms.py +++ b/examples/foundational/22c-natural-conversation-mixed-llms.py @@ -402,7 +402,6 @@ class OutputGate(FrameProcessor): async def fetch_weather_from_api(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -449,6 +448,10 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + @llm.event_handler("on_function_calls_started") + async def on_function_calls_started(service, function_calls): + await tts.queue_frame(TTSSpeakFrame("Let me check on that.")) + tools = [ ChatCompletionToolParam( type="function", diff --git a/examples/foundational/26b-gemini-multimodal-live-function-calling.py b/examples/foundational/26b-gemini-multimodal-live-function-calling.py index 84c7d5045..15db2faa0 100644 --- a/examples/foundational/26b-gemini-multimodal-live-function-calling.py +++ b/examples/foundational/26b-gemini-multimodal-live-function-calling.py @@ -40,11 +40,17 @@ async def fetch_weather_from_api(params: FunctionCallParams): ) +async def fetch_restaurant_recommendation(params: FunctionCallParams): + await params.result_callback({"name": "The Golden Dragon"}) + + system_instruction = """ You are a helpful assistant who can answer questions and use tools. -You have a tool called "get_current_weather" that can be used to get the current weather. If the user asks -for the weather, call this function. +You have three tools available to you: +1. get_current_weather: Use this tool to get the current weather in a specific location. +2. get_restaurant_recommendation: Use this tool to get a restaurant recommendation in a specific location. +3. google_search: Use this tool to search the web for information. """ @@ -101,9 +107,21 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si }, required=["location", "format"], ) + restaurant_function = FunctionSchema( + name="get_restaurant_recommendation", + description="Get a restaurant recommendation", + properties={ + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + }, + required=["location"], + ) search_tool = {"google_search": {}} tools = ToolsSchema( - standard_tools=[weather_function], custom_tools={AdapterType.GEMINI: [search_tool]} + standard_tools=[weather_function, restaurant_function], + custom_tools={AdapterType.GEMINI: [search_tool]}, ) llm = GeminiMultimodalLiveLLMService( @@ -113,6 +131,7 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si ) llm.register_function("get_current_weather", fetch_weather_from_api) + llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) context = OpenAILLMContext( [{"role": "user", "content": "Say hello."}], diff --git a/examples/open-telemetry/jaeger/bot.py b/examples/open-telemetry/jaeger/bot.py index 74e587410..0f3084fd3 100644 --- a/examples/open-telemetry/jaeger/bot.py +++ b/examples/open-telemetry/jaeger/bot.py @@ -49,7 +49,6 @@ if IS_TRACING_ENABLED: async def fetch_weather_from_api(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -93,6 +92,10 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + @llm.event_handler("on_function_calls_started") + async def on_function_calls_started(service, function_calls): + await tts.queue_frame(TTSSpeakFrame("Let me check on that.")) + weather_function = FunctionSchema( name="get_current_weather", description="Get the current weather", diff --git a/examples/open-telemetry/langfuse/bot.py b/examples/open-telemetry/langfuse/bot.py index 22ff399a0..9c4f34695 100644 --- a/examples/open-telemetry/langfuse/bot.py +++ b/examples/open-telemetry/langfuse/bot.py @@ -46,7 +46,6 @@ if IS_TRACING_ENABLED: async def fetch_weather_from_api(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -90,6 +89,10 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + @llm.event_handler("on_function_calls_started") + async def on_function_calls_started(service, function_calls): + await tts.queue_frame(TTSSpeakFrame("Let me check on that.")) + weather_function = FunctionSchema( name="get_current_weather", description="Get the current weather", diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 45c1fc6c2..6ad0f089f 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -667,6 +667,32 @@ class MetricsFrame(SystemFrame): data: List[MetricsData] +@dataclass +class FunctionCallFromLLM: + """Represents a function call returned by the LLM to be registered for execution. + + Attributes: + function_name (str): The name of the function. + tool_call_id (str): A unique identifier for the function call. + arguments (Mapping[str, Any]): The arguments for the function. + context (OpenAILLMContext): The LLM context. + + """ + + function_name: str + tool_call_id: str + arguments: Mapping[str, Any] + context: Any + + +@dataclass +class FunctionCallsStartedFrame(SystemFrame): + """A frame signaling that one or more function call execution is going to + start.""" + + function_calls: Sequence[FunctionCallFromLLM] + + @dataclass class FunctionCallInProgressFrame(SystemFrame): """A frame signaling that a function call is in progress.""" @@ -701,6 +727,7 @@ class FunctionCallResultFrame(SystemFrame): tool_call_id: str arguments: Any result: Any + run_llm: Optional[bool] = None properties: Optional[FunctionCallResultProperties] = None diff --git a/src/pipecat/pipeline/runner.py b/src/pipecat/pipeline/runner.py index 7ac07064f..23c43c06e 100644 --- a/src/pipecat/pipeline/runner.py +++ b/src/pipecat/pipeline/runner.py @@ -59,7 +59,7 @@ class PipelineRunner(BaseObject): await asyncio.gather(*[t.stop_when_done() for t in self._tasks.values()]) async def cancel(self): - logger.debug(f"Canceling runner {self}") + logger.debug(f"Cancelling runner {self}") await asyncio.gather(*[t.cancel() for t in self._tasks.values()]) def _setup_sigint(self): @@ -72,7 +72,7 @@ class PipelineRunner(BaseObject): self._sig_task = asyncio.create_task(self._sig_cancel()) async def _sig_cancel(self): - logger.warning(f"Interruption detected. Canceling runner {self}") + logger.warning(f"Interruption detected. Cancelling runner {self}") await self.cancel() def _gc_collect(self): diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index be9c6f77f..ae2382cd6 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -23,6 +23,7 @@ from pipecat.frames.frames import ( FunctionCallCancelFrame, FunctionCallInProgressFrame, FunctionCallResultFrame, + FunctionCallsStartedFrame, InterimTranscriptionFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, @@ -500,7 +501,7 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): self._params.expect_stripped_words = kwargs["expect_stripped_words"] self._started = 0 - self._function_calls_in_progress: Dict[str, FunctionCallInProgressFrame] = {} + self._function_calls_in_progress: Dict[str, Optional[FunctionCallInProgressFrame]] = {} self._context_updated_tasks: Set[asyncio.Task] = set() async def handle_aggregation(self, aggregation: str): @@ -538,6 +539,8 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): self.set_tools(frame.tools) elif isinstance(frame, LLMSetToolChoiceFrame): self.set_tool_choice(frame.tool_choice) + elif isinstance(frame, FunctionCallsStartedFrame): + await self._handle_function_calls_started(frame) elif isinstance(frame, FunctionCallInProgressFrame): await self._handle_function_call_in_progress(frame) elif isinstance(frame, FunctionCallResultFrame): @@ -574,6 +577,12 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): self._started = 0 self.reset() + async def _handle_function_calls_started(self, frame: FunctionCallsStartedFrame): + function_names = [f"{f.function_name}:{f.tool_call_id}" for f in frame.function_calls] + logger.debug(f"{self} FunctionCallsStartedFrame: {function_names}") + for function_call in frame.function_calls: + self._function_calls_in_progress[function_call.tool_call_id] = None + async def _handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame): logger.debug( f"{self} FunctionCallInProgressFrame: [{frame.function_name}:{frame.tool_call_id}]" @@ -597,19 +606,22 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): await self.handle_function_call_result(frame) + run_llm = False + # Run inference if the function call result requires it. if frame.result: - run_llm = False - if properties and properties.run_llm is not None: - # If the tool call result has a run_llm property, use it + # If the tool call result has a run_llm property, use it. run_llm = properties.run_llm + elif frame.run_llm is not None: + # If the frame is indicating we should run the LLM, do it. + run_llm = frame.run_llm else: - # Default behavior is to run the LLM if there are no function calls in progress + # If this is the last function call in progress, run the LLM. run_llm = not bool(self._function_calls_in_progress) - if run_llm: - await self.push_context_frame(FrameDirection.UPSTREAM) + if run_llm: + await self.push_context_frame(FrameDirection.UPSTREAM) # Call the `on_context_updated` callback once the function call result # is added to the context. Also, run this in a separate task to make diff --git a/src/pipecat/services/anthropic/llm.py b/src/pipecat/services/anthropic/llm.py index 2c4e9078d..236f269fa 100644 --- a/src/pipecat/services/anthropic/llm.py +++ b/src/pipecat/services/anthropic/llm.py @@ -45,7 +45,7 @@ from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContextFrame, ) from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.llm_service import LLMService +from pipecat.services.llm_service import FunctionCallFromLLM, LLMService from pipecat.utils.tracing.service_decorators import traced_llm try: @@ -202,9 +202,8 @@ class AnthropicLLMService(LLMService): tool_use_block = None json_accumulator = "" + function_calls = [] async for event in response: - # logger.debug(f"Anthropic LLM event: {event}") - # Aggregate streaming content, create frames, trigger events if event.type == "content_block_delta": @@ -226,11 +225,14 @@ class AnthropicLLMService(LLMService): and event.delta.stop_reason == "tool_use" ): if tool_use_block: - await self.call_function( - context=context, - tool_call_id=tool_use_block.id, - function_name=tool_use_block.name, - arguments=json.loads(json_accumulator) if json_accumulator else dict(), + args = json.loads(json_accumulator) if json_accumulator else {} + function_calls.append( + FunctionCallFromLLM( + context=context, + tool_call_id=tool_use_block.id, + function_name=tool_use_block.name, + arguments=args, + ) ) # Calculate usage. Do this here in its own if statement, because there may be usage @@ -277,6 +279,8 @@ class AnthropicLLMService(LLMService): if total_input_tokens >= 1024: context.turns_above_cache_threshold += 1 + await self.run_function_calls(function_calls) + except asyncio.CancelledError: # If we're interrupted, we won't get a complete usage report. So set our flag to use the # token estimate. The reraise the exception so all the processors running in this task diff --git a/src/pipecat/services/aws/llm.py b/src/pipecat/services/aws/llm.py index a90620b20..dcec91463 100644 --- a/src/pipecat/services/aws/llm.py +++ b/src/pipecat/services/aws/llm.py @@ -21,6 +21,7 @@ from pipecat.adapters.services.bedrock_adapter import AWSBedrockLLMAdapter from pipecat.frames.frames import ( Frame, FunctionCallCancelFrame, + FunctionCallFromLLM, FunctionCallInProgressFrame, FunctionCallResultFrame, LLMFullResponseEndFrame, @@ -708,6 +709,7 @@ class AWSBedrockLLMService(LLMService): tool_use_block = None json_accumulator = "" + function_calls = [] for event in response["stream"]: # Handle text content if "contentBlockDelta" in event: @@ -740,11 +742,13 @@ class AWSBedrockLLMService(LLMService): # Only call function if it's not the no_operation tool if not using_noop_tool: - await self.call_function( - context=context, - tool_call_id=tool_use_block["id"], - function_name=tool_use_block["name"], - arguments=arguments, + function_calls.append( + FunctionCallFromLLM( + context=context, + tool_call_id=tool_use_block["id"], + function_name=tool_use_block["name"], + arguments=arguments, + ) ) else: logger.debug("Ignoring no_operation tool call") @@ -758,7 +762,7 @@ class AWSBedrockLLMService(LLMService): completion_tokens += usage.get("outputTokens", 0) cache_read_input_tokens += usage.get("cacheReadInputTokens", 0) cache_creation_input_tokens += usage.get("cacheWriteInputTokens", 0) - + await self.run_function_calls(function_calls) except asyncio.CancelledError: # If we're interrupted, we won't get a complete usage report. So set our flag to use the # token estimate. The reraise the exception so all the processors running in this task diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index 6dd93720b..894e53285 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -52,7 +52,7 @@ from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContextFrame, ) from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.llm_service import LLMService +from pipecat.services.llm_service import FunctionCallFromLLM, LLMService from pipecat.services.openai.llm import ( OpenAIAssistantContextAggregator, OpenAIUserContextAggregator, @@ -891,13 +891,18 @@ class GeminiMultimodalLiveLLMService(LLMService): return if not self._context: logger.error("Function calls are not supported without a context object.") - for call in function_calls: - await self.call_function( + + function_calls_llm = [ + FunctionCallFromLLM( context=self._context, - tool_call_id=call.id, - function_name=call.name, - arguments=call.args, + tool_call_id=f.id, + function_name=f.name, + arguments=f.args, ) + for f in function_calls + ] + + await self.run_function_calls(function_calls_llm) @traced_gemini_live(operation="llm_response") async def _handle_evt_turn_complete(self, evt): diff --git a/src/pipecat/services/google/llm.py b/src/pipecat/services/google/llm.py index 5efbbe8c0..b63b3786c 100644 --- a/src/pipecat/services/google/llm.py +++ b/src/pipecat/services/google/llm.py @@ -42,7 +42,7 @@ from pipecat.processors.aggregators.openai_llm_context import ( ) from pipecat.processors.frame_processor import FrameDirection from pipecat.services.google.frames import LLMSearchResponseFrame -from pipecat.services.llm_service import LLMService +from pipecat.services.llm_service import FunctionCallFromLLM, LLMService from pipecat.services.openai.llm import ( OpenAIAssistantContextAggregator, OpenAIUserContextAggregator, @@ -557,6 +557,7 @@ class GoogleLLMService(LLMService): ) await self.stop_ttfb_metrics() + function_calls = [] async for chunk in response: if chunk.usage_metadata: prompt_tokens += chunk.usage_metadata.prompt_token_count or 0 @@ -576,11 +577,13 @@ class GoogleLLMService(LLMService): function_call = part.function_call id = function_call.id or str(uuid.uuid4()) logger.debug(f"Function call: {function_call.name}:{id}") - await self.call_function( - context=context, - tool_call_id=id, - function_name=function_call.name, - arguments=function_call.args or {}, + function_calls.append( + FunctionCallFromLLM( + context=context, + tool_call_id=id, + function_name=function_call.name, + arguments=function_call.args or {}, + ) ) if ( @@ -621,6 +624,8 @@ class GoogleLLMService(LLMService): "rendered_content": rendered_content, "origins": origins, } + + await self.run_function_calls(function_calls) except DeadlineExceeded: await self._call_event_handler("on_completion_timeout") except Exception as e: diff --git a/src/pipecat/services/google/llm_openai.py b/src/pipecat/services/google/llm_openai.py index 94b99072a..a497cb229 100644 --- a/src/pipecat/services/google/llm_openai.py +++ b/src/pipecat/services/google/llm_openai.py @@ -10,6 +10,8 @@ import os from openai import AsyncStream from openai.types.chat import ChatCompletionChunk +from pipecat.services.llm_service import FunctionCallFromLLM + # Suppress gRPC fork warnings os.environ["GRPC_ENABLE_FORK_SUPPORT"] = "false" @@ -18,7 +20,6 @@ from loguru import logger from pipecat.frames.frames import LLMTextFrame from pipecat.metrics.metrics import LLMTokenUsage from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext -from pipecat.services.openai.base_llm import OpenAIUnhandledFunctionException from pipecat.services.openai.llm import OpenAILLMService @@ -112,25 +113,26 @@ class GoogleLLMOpenAIBetaService(OpenAILLMService): logger.debug( f"Function list: {functions_list}, Arguments list: {arguments_list}, Tool ID list: {tool_id_list}" ) - for index, (function_name, arguments, tool_id) in enumerate( - zip(functions_list, arguments_list, tool_id_list), start=1 + + function_calls = [] + for function_name, arguments, tool_id in zip( + functions_list, arguments_list, tool_id_list ): if function_name == "": # TODO: Remove the _process_context method once Google resolves the bug # where the index is incorrectly set to None instead of returning the actual index, # which currently results in an empty function name(''). continue - if self.has_function(function_name): - run_llm = False - arguments = json.loads(arguments) - await self.call_function( + + arguments = json.loads(arguments) + + function_calls.append( + FunctionCallFromLLM( context=context, + tool_call_id=tool_id, function_name=function_name, arguments=arguments, - tool_call_id=tool_id, - 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." ) + ) + + await self.run_function_calls(function_calls) diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index 21b62325d..5619cd35e 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -7,18 +7,23 @@ import asyncio import inspect from dataclasses import dataclass -from typing import Any, Awaitable, Callable, Mapping, Optional, Protocol, Set, Tuple, Type +from typing import Any, Awaitable, Callable, Dict, Mapping, Optional, Protocol, Sequence, Type from loguru import logger from pipecat.adapters.base_llm_adapter import BaseLLMAdapter from pipecat.adapters.services.open_ai_adapter import OpenAILLMAdapter from pipecat.frames.frames import ( + CancelFrame, + EndFrame, Frame, FunctionCallCancelFrame, + FunctionCallFromLLM, FunctionCallInProgressFrame, FunctionCallResultFrame, FunctionCallResultProperties, + FunctionCallsStartedFrame, + StartFrame, StartInterruptionFrame, UserImageRequestFrame, ) @@ -41,22 +46,6 @@ class FunctionCallResultCallback(Protocol): ) -> None: ... -@dataclass -class FunctionCallEntry: - """Represents an internal entry for a function call. - - Attributes: - function_name (Optional[str]): The name of the function. - handler (FunctionCallHandler): The handler for processing function call parameters. - cancel_on_interruption (bool): Flag indicating whether to cancel the call on interruption. - - """ - - function_name: Optional[str] - handler: FunctionCallHandler - cancel_on_interruption: bool - - @dataclass class FunctionCallParams: """Parameters for a function call. @@ -79,20 +68,78 @@ class FunctionCallParams: result_callback: FunctionCallResultCallback +@dataclass +class FunctionCallRegistryItem: + """Represents an entry in our function call registry. This is what the user + registers. + + Attributes: + function_name (Optional[str]): The name of the function. + handler (FunctionCallHandler): The handler for processing function call parameters. + cancel_on_interruption (bool): Flag indicating whether to cancel the call on interruption. + + """ + + function_name: Optional[str] + handler: FunctionCallHandler + cancel_on_interruption: bool + + +@dataclass +class FunctionCallRunnerItem: + """Represents an internal function call entry to our function call + runner. The runner executes function calls in order. + + Attributes: + registry_name (Optional[str]): The function call name registration (could be None). + function_name (str): The name of the function. + tool_call_id (str): A unique identifier for the function call. + arguments (Mapping[str, Any]): The arguments for the function. + context (OpenAILLMContext): The LLM context. + + """ + + registry_item: FunctionCallRegistryItem + function_name: str + tool_call_id: str + arguments: Mapping[str, Any] + context: OpenAILLMContext + run_llm: Optional[bool] = None + + class LLMService(AIService): - """This class is a no-op but serves as a base class for LLM services.""" + """This is the base class for all LLM services. It handles function calling + registration and execution. The class also provides event handlers. + + An event to know when an LLM service completion timeout occurs: + + @task.event_handler("on_completion_timeout") + async def on_completion_timeout(service): + ... + + And an event to know that function calls have been received from the LLM + service and that we are going to start executing them: + + @task.event_handler("on_function_calls_started") + async def on_function_calls_started(service, function_calls: Sequence[FunctionCallFromLLM]): + ... + + """ # OpenAILLMAdapter is used as the default adapter since it aligns with most LLM implementations. # However, subclasses should override this with a more specific adapter when necessary. adapter_class: Type[BaseLLMAdapter] = OpenAILLMAdapter - def __init__(self, **kwargs): + def __init__(self, run_in_parallel: bool = True, **kwargs): super().__init__(**kwargs) - self._functions = {} + self._run_in_parallel = run_in_parallel self._start_callbacks = {} self._adapter = self.adapter_class() - self._function_call_tasks: Set[Tuple[asyncio.Task, str, str]] = set() + self._functions: Dict[Optional[str], FunctionCallRegistryItem] = {} + self._function_call_tasks: Dict[asyncio.Task, FunctionCallRunnerItem] = {} + self._sequential_runner_task: Optional[asyncio.Task] = None + self._register_event_handler("on_function_calls_started") self._register_event_handler("on_completion_timeout") def get_llm_adapter(self) -> BaseLLMAdapter: @@ -107,13 +154,28 @@ class LLMService(AIService): ) -> Any: pass + async def start(self, frame: StartFrame): + await super().start(frame) + if not self._run_in_parallel: + await self._create_sequential_runner_task() + + async def stop(self, frame: EndFrame): + await super().stop(frame) + if not self._run_in_parallel: + await self._cancel_sequential_runner_task() + + async def cancel(self, frame: CancelFrame): + await super().cancel(frame) + if not self._run_in_parallel: + await self._cancel_sequential_runner_task() + async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) if isinstance(frame, StartInterruptionFrame): await self._handle_interruptions(frame) - async def _handle_interruptions(self, frame: StartInterruptionFrame): + async def _handle_interruptions(self, _: StartInterruptionFrame): for function_name, entry in self._functions.items(): if entry.cancel_on_interruption: await self._cancel_function_call(function_name) @@ -124,11 +186,11 @@ class LLMService(AIService): handler: Any, start_callback=None, *, - cancel_on_interruption: bool = False, + cancel_on_interruption: bool = True, ): # Registering a function with the function_name set to None will run # that handler for all functions - self._functions[function_name] = FunctionCallEntry( + self._functions[function_name] = FunctionCallRegistryItem( function_name=function_name, handler=handler, cancel_on_interruption=cancel_on_interruption, @@ -157,25 +219,43 @@ class LLMService(AIService): return True return function_name in self._functions.keys() - async def call_function( - self, - *, - context: OpenAILLMContext, - tool_call_id: str, - function_name: str, - arguments: Mapping[str, Any], - run_llm: bool = True, - ): - if not function_name in self._functions.keys() and not None in self._functions.keys(): + async def run_function_calls(self, function_calls: Sequence[FunctionCallFromLLM]): + if len(function_calls) == 0: return - task = self.create_task( - self._run_function_call(context, tool_call_id, function_name, arguments, run_llm) - ) + await self._call_event_handler("on_function_calls_started", function_calls) - self._function_call_tasks.add((task, tool_call_id, function_name)) + # Push frame both downstream and upstream + started_frame_downstream = FunctionCallsStartedFrame(function_calls=function_calls) + started_frame_upstream = FunctionCallsStartedFrame(function_calls=function_calls) + await self.push_frame(started_frame_downstream, FrameDirection.DOWNSTREAM) + await self.push_frame(started_frame_upstream, FrameDirection.UPSTREAM) - task.add_done_callback(self._function_call_task_finished) + for function_call in function_calls: + if function_call.function_name in self._functions.keys(): + item = self._functions[function_call.function_name] + elif None in self._functions.keys(): + item = self._functions[None] + else: + logger.warning( + f"{self} is calling '{function_call.function_name}', but it's not registered." + ) + continue + + runner_item = FunctionCallRunnerItem( + registry_item=item, + function_name=function_call.function_name, + tool_call_id=function_call.tool_call_id, + arguments=function_call.arguments, + context=function_call.context, + ) + + if self._run_in_parallel: + task = self.create_task(self._run_function_call(runner_item)) + self._function_call_tasks[task] = runner_item + task.add_done_callback(self._function_call_task_finished) + else: + await self._sequential_runner_queue.put(runner_item) async def call_start_function(self, context: OpenAILLMContext, function_name: str): if function_name in self._start_callbacks.keys(): @@ -203,43 +283,57 @@ class LLMService(AIService): FrameDirection.UPSTREAM, ) - async def _run_function_call( - self, - context: OpenAILLMContext, - tool_call_id: str, - function_name: str, - arguments: Mapping[str, Any], - run_llm: bool = True, - ): - if function_name in self._functions.keys(): - entry = self._functions[function_name] + async def _create_sequential_runner_task(self): + if not self._sequential_runner_task: + self._sequential_runner_queue = asyncio.Queue() + self._sequential_runner_task = self.create_task(self._sequential_runner_handler()) + + async def _cancel_sequential_runner_task(self): + if self._sequential_runner_task: + await self.cancel_task(self._sequential_runner_task) + self._sequential_runner_task = None + + async def _sequential_runner_handler(self): + while True: + runner_item = await self._sequential_runner_queue.get() + task = self.create_task(self._run_function_call(runner_item)) + self._function_call_tasks[task] = runner_item + # Since we run tasks sequentially we don't need to call + # task.add_done_callback(self._function_call_task_finished). + await self.wait_for_task(task) + del self._function_call_tasks[task] + + async def _run_function_call(self, runner_item: FunctionCallRunnerItem): + if runner_item.function_name in self._functions.keys(): + item = self._functions[runner_item.function_name] elif None in self._functions.keys(): - entry = self._functions[None] + item = self._functions[None] else: return logger.debug( - f"{self} Calling function [{function_name}:{tool_call_id}] with arguments {arguments}" + f"{self} Calling function [{runner_item.function_name}:{runner_item.tool_call_id}] with arguments {runner_item.arguments}" ) # NOTE(aleix): This needs to be removed after we remove the deprecation. - await self.call_start_function(context, function_name) + await self.call_start_function(runner_item.context, runner_item.function_name) - # 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). - # Also push a SystemFrame upstream for use by other processors, like STTMuteFilter. + # Push a function call in-progress 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). Also push it upstream for use + # by other processors, like STTMuteFilter. progress_frame_downstream = FunctionCallInProgressFrame( - function_name=function_name, - tool_call_id=tool_call_id, - arguments=arguments, - cancel_on_interruption=entry.cancel_on_interruption, + function_name=runner_item.function_name, + tool_call_id=runner_item.tool_call_id, + arguments=runner_item.arguments, + cancel_on_interruption=item.cancel_on_interruption, ) progress_frame_upstream = FunctionCallInProgressFrame( - function_name=function_name, - tool_call_id=tool_call_id, - arguments=arguments, - cancel_on_interruption=entry.cancel_on_interruption, + function_name=runner_item.function_name, + tool_call_id=runner_item.tool_call_id, + arguments=runner_item.arguments, + cancel_on_interruption=item.cancel_on_interruption, ) # Push frame both downstream and upstream @@ -251,24 +345,26 @@ class LLMService(AIService): result: Any, *, properties: Optional[FunctionCallResultProperties] = None ): result_frame_downstream = FunctionCallResultFrame( - function_name=function_name, - tool_call_id=tool_call_id, - arguments=arguments, + function_name=runner_item.function_name, + tool_call_id=runner_item.tool_call_id, + arguments=runner_item.arguments, result=result, + run_llm=runner_item.run_llm, properties=properties, ) result_frame_upstream = FunctionCallResultFrame( - function_name=function_name, - tool_call_id=tool_call_id, - arguments=arguments, + function_name=runner_item.function_name, + tool_call_id=runner_item.tool_call_id, + arguments=runner_item.arguments, result=result, + run_llm=runner_item.run_llm, properties=properties, ) await self.push_frame(result_frame_downstream, FrameDirection.DOWNSTREAM) await self.push_frame(result_frame_upstream, FrameDirection.UPSTREAM) - signature = inspect.signature(entry.handler) + signature = inspect.signature(item.handler) if len(signature.parameters) > 1: import warnings @@ -279,24 +375,32 @@ class LLMService(AIService): DeprecationWarning, ) - await entry.handler( - function_name, tool_call_id, arguments, self, context, function_call_result_callback + await item.handler( + runner_item.function_name, + runner_item.tool_call_id, + runner_item.arguments, + self, + runner_item.context, + function_call_result_callback, ) else: params = FunctionCallParams( - function_name=function_name, - tool_call_id=tool_call_id, - arguments=arguments, + function_name=runner_item.function_name, + tool_call_id=runner_item.tool_call_id, + arguments=runner_item.arguments, llm=self, - context=context, + context=runner_item.context, result_callback=function_call_result_callback, ) - await entry.handler(params) + await item.handler(params) - async def _cancel_function_call(self, function_name: str): + async def _cancel_function_call(self, function_name: Optional[str]): cancelled_tasks = set() - for task, tool_call_id, name in self._function_call_tasks: - if name == function_name: + for task, runner_item in self._function_call_tasks.items(): + if runner_item.registry_item.function_name == function_name: + name = runner_item.function_name + tool_call_id = runner_item.tool_call_id + # We remove the callback because we are going to cancel the task # now, otherwise we will be removing it from the set while we # are iterating. @@ -306,23 +410,20 @@ class LLMService(AIService): await self.cancel_task(task) - frame = FunctionCallCancelFrame( - function_name=function_name, tool_call_id=tool_call_id - ) + frame = FunctionCallCancelFrame(function_name=name, tool_call_id=tool_call_id) await self.push_frame(frame) - logger.debug(f"{self} Function call [{name}:{tool_call_id}] has been cancelled") - cancelled_tasks.add(task) + logger.debug(f"{self} Function call [{name}:{tool_call_id}] has been cancelled") + # Remove all cancelled tasks from our set. for task in cancelled_tasks: self._function_call_task_finished(task) def _function_call_task_finished(self, task: asyncio.Task): - tuple_to_remove = next((t for t in self._function_call_tasks if t[0] == task), None) - if tuple_to_remove: - self._function_call_tasks.discard(tuple_to_remove) + if task in self._function_call_tasks: + del self._function_call_tasks[task] # The task is finished so this should exit immediately. We need to # do this because otherwise the task manager would report a dangling # task if we don't remove it. diff --git a/src/pipecat/services/openai/base_llm.py b/src/pipecat/services/openai/base_llm.py index e90f87f3c..2badfed96 100644 --- a/src/pipecat/services/openai/base_llm.py +++ b/src/pipecat/services/openai/base_llm.py @@ -34,14 +34,10 @@ from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContextFrame, ) from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.llm_service import LLMService +from pipecat.services.llm_service import FunctionCallFromLLM, LLMService from pipecat.utils.tracing.service_decorators import traced_llm -class OpenAIUnhandledFunctionException(Exception): - pass - - class BaseOpenAILLMService(LLMService): """This is the base for all services that use the AsyncOpenAI client. @@ -260,23 +256,22 @@ class BaseOpenAILLMService(LLMService): arguments_list.append(arguments) tool_id_list.append(tool_call_id) - for index, (function_name, arguments, tool_id) in enumerate( - zip(functions_list, arguments_list, tool_id_list), start=1 + function_calls = [] + + for function_name, arguments, tool_id in zip( + functions_list, arguments_list, tool_id_list ): - if self.has_function(function_name): - run_llm = False - arguments = json.loads(arguments) - await self.call_function( + arguments = json.loads(arguments) + function_calls.append( + FunctionCallFromLLM( context=context, + tool_call_id=tool_id, function_name=function_name, arguments=arguments, - tool_call_id=tool_id, - 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." ) + ) + + await self.run_function_calls(function_calls) async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) diff --git a/src/pipecat/services/openai_realtime_beta/openai.py b/src/pipecat/services/openai_realtime_beta/openai.py index 9957cb134..4ea5843bf 100644 --- a/src/pipecat/services/openai_realtime_beta/openai.py +++ b/src/pipecat/services/openai_realtime_beta/openai.py @@ -48,7 +48,7 @@ from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContextFrame, ) from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.llm_service import LLMService +from pipecat.services.llm_service import FunctionCallFromLLM, LLMService from pipecat.services.openai.llm import OpenAIContextAggregatorPair from pipecat.transcriptions.language import Language from pipecat.utils.time import time_now_iso8601 @@ -78,10 +78,6 @@ class CurrentAudioResponse: total_size: int = 0 -class OpenAIUnhandledFunctionException(Exception): - pass - - class OpenAIRealtimeBetaLLMService(LLMService): # Overriding the default adapter to use the OpenAIRealtimeLLMAdapter one. adapter_class = OpenAIRealtimeLLMAdapter @@ -587,25 +583,18 @@ class OpenAIRealtimeBetaLLMService(LLMService): 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._functions.keys() or None in self._functions.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." + function_calls = [] + for item in items: + args = json.loads(item.arguments) + function_calls.append( + FunctionCallFromLLM( + context=self._context, + tool_call_id=item.call_id, + function_name=item.name, + arguments=args, ) + ) + await self.run_function_calls(function_calls) # # state and client events for the current conversation