Merge pull request #1683 from pipecat-ai/aleix/run-function-calls-sequentially

run function calls sequentially or in parallel

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.
 

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 = [
         {

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
 

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",

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.
 

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",

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",

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",

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",

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",

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",

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",

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",

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",

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",

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 = [
         {

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

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

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",

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",

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
+You have three tools available to you:
-for the weather, call this function.
+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."}],

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",

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",

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
 
 

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):

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:
+        if run_llm:
-                await self.push_context_frame(FrameDirection.UPSTREAM)
+            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

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(
+                        args = json.loads(json_accumulator) if json_accumulator else {}
-                            context=context,
+                        function_calls.append(
-                            tool_call_id=tool_use_block.id,
+                            FunctionCallFromLLM(
-                            function_name=tool_use_block.name,
+                                context=context,
-                            arguments=json.loads(json_accumulator) if json_accumulator else dict(),
+                                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

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(
+                                function_calls.append(
-                                    context=context,
+                                    FunctionCallFromLLM(
-                                    tool_call_id=tool_use_block["id"],
+                                        context=context,
-                                    function_name=tool_use_block["name"],
+                                        tool_call_id=tool_use_block["id"],
-                                    arguments=arguments,
+                                        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

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,
+                tool_call_id=f.id,
-                function_name=call.name,
+                function_name=f.name,
-                arguments=call.args,
+                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):

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(
+                                function_calls.append(
-                                    context=context,
+                                    FunctionCallFromLLM(
-                                    tool_call_id=id,
+                                        context=context,
-                                    function_name=function_call.name,
+                                        tool_call_id=id,
-                                    arguments=function_call.args or {},
+                                        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:

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)
-                    arguments = json.loads(arguments)
+
-                    await self.call_function(
+                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)

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(
+    async def run_function_calls(self, function_calls: Sequence[FunctionCallFromLLM]):
-        self,
+        if len(function_calls) == 0:
-        *,
-        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():
             return
 
-        task = self.create_task(
+        await self._call_event_handler("on_function_calls_started", function_calls)
-            self._run_function_call(context, tool_call_id, function_name, arguments, run_llm)
-        )
 
-        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(
+    async def _create_sequential_runner_task(self):
-        self,
+        if not self._sequential_runner_task:
-        context: OpenAILLMContext,
+            self._sequential_runner_queue = asyncio.Queue()
-        tool_call_id: str,
+            self._sequential_runner_task = self.create_task(self._sequential_runner_handler())
-        function_name: str,
+
-        arguments: Mapping[str, Any],
+    async def _cancel_sequential_runner_task(self):
-        run_llm: bool = True,
+        if self._sequential_runner_task:
-    ):
+            await self.cancel_task(self._sequential_runner_task)
-        if function_name in self._functions.keys():
+            self._sequential_runner_task = None
-            entry = self._functions[function_name]
+
+    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
+        # Push a function call in-progress downstream. This frame will let our
-        # know that we are in the middle of a function call. Some contexts/aggregators may
+        # assistant context aggregator know that we are in the middle of a
-        # not need this. But some definitely do (Anthropic, for example).
+        # function call. Some contexts/aggregators may not need this. But some
-        # Also push a SystemFrame upstream for use by other processors, like STTMuteFilter.
+        # definitely do (Anthropic, for example).  Also push it upstream for use
+        # by other processors, like STTMuteFilter.
         progress_frame_downstream = FunctionCallInProgressFrame(
-            function_name=function_name,
+            function_name=runner_item.function_name,
-            tool_call_id=tool_call_id,
+            tool_call_id=runner_item.tool_call_id,
-            arguments=arguments,
+            arguments=runner_item.arguments,
-            cancel_on_interruption=entry.cancel_on_interruption,
+            cancel_on_interruption=item.cancel_on_interruption,
         )
         progress_frame_upstream = FunctionCallInProgressFrame(
-            function_name=function_name,
+            function_name=runner_item.function_name,
-            tool_call_id=tool_call_id,
+            tool_call_id=runner_item.tool_call_id,
-            arguments=arguments,
+            arguments=runner_item.arguments,
-            cancel_on_interruption=entry.cancel_on_interruption,
+            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,
+                function_name=runner_item.function_name,
-                tool_call_id=tool_call_id,
+                tool_call_id=runner_item.tool_call_id,
-                arguments=arguments,
+                arguments=runner_item.arguments,
                 result=result,
+                run_llm=runner_item.run_llm,
                 properties=properties,
             )
             result_frame_upstream = FunctionCallResultFrame(
-                function_name=function_name,
+                function_name=runner_item.function_name,
-                tool_call_id=tool_call_id,
+                tool_call_id=runner_item.tool_call_id,
-                arguments=arguments,
+                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(
+            await item.handler(
-                function_name, tool_call_id, arguments, self, context, function_call_result_callback
+                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,
+                function_name=runner_item.function_name,
-                tool_call_id=tool_call_id,
+                tool_call_id=runner_item.tool_call_id,
-                arguments=arguments,
+                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:
+        for task, runner_item in self._function_call_tasks.items():
-            if name == function_name:
+            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(
+                frame = FunctionCallCancelFrame(function_name=name, tool_call_id=tool_call_id)
-                    function_name=function_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 task in self._function_call_tasks:
-        if tuple_to_remove:
+            del self._function_call_tasks[task]
-            self._function_call_tasks.discard(tuple_to_remove)
             # 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.

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(
+            function_calls = []
-                zip(functions_list, arguments_list, tool_id_list), start=1
+
+            for function_name, arguments, tool_id in zip(
+                functions_list, arguments_list, tool_id_list
             ):
-                if self.has_function(function_name):
+                arguments = json.loads(arguments)
-                    run_llm = False
+                function_calls.append(
-                    arguments = json.loads(arguments)
+                    FunctionCallFromLLM(
-                    await self.call_function(
                         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)

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)
+        function_calls = []
-        for index, item in enumerate(items):
+        for item in items:
-            function_name = item.name
+            args = json.loads(item.arguments)
-            tool_id = item.call_id
+            function_calls.append(
-            arguments = json.loads(item.arguments)
+                FunctionCallFromLLM(
-            if self.has_function(function_name):
+                    context=self._context,
-                run_llm = index == total_items - 1
+                    tool_call_id=item.call_id,
-                if function_name in self._functions.keys() or None in self._functions.keys():
+                    function_name=item.name,
-                    await self.call_function(
+                    arguments=args,
-                        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."
                 )
+            )
+        await self.run_function_calls(function_calls)
 
     #
     # state and client events for the current conversation