Merge pull request #4230 from pipecat-ai/filipi/async_tools_stream

Support for streaming multiple responses via function calls

changelog/4230.added.md

@@ -0,0 +1 @@
+- Added support for streaming intermediate results from async function calls. Call `result_callback` multiple times with `properties=FunctionCallResultProperties(is_final=False)` to push incremental updates, then call it once more (with `is_final=True`, the default) to deliver the final result. Only valid for functions registered with `cancel_on_interruption=False`.

changelog/4230.fixed.md

@@ -0,0 +1 @@
+- Fixed duplicate LLM replies that could occur when multiple async function call results arrived while an LLM request was already queued.

examples/function-calling/function-calling-anthropic-async-stream.py

@@ -0,0 +1,208 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Example: async function call with intermediate updates.
+
+The ``track_current_location`` tool simulates a GPS tracker reporting the
+device's position during a road trip from San Francisco to San Diego.  It
+sends two intermediate updates (via ``params.result_callback`` with
+``is_final=False``) as the vehicle passes through cities along the way, then
+delivers the final destination (via ``params.result_callback``).  Each update
+returns the same structure with a different city:
+
+  Update 1 – {gps, city: "San Francisco"}   ← trip start
+  Update 2 – {gps, city: "Los Angeles"}     ← passing through
+  Final     – {gps, city: "San Diego"}      ← destination reached
+
+Because the function is registered with ``cancel_on_interruption=False``, the
+LLM can keep talking while the trip is in progress; each position update
+arrives as a developer message so the LLM can narrate the journey to the user.
+"""
+
+import asyncio
+import os
+
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.adapters.schemas.function_schema import FunctionSchema
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import (
+    FunctionCallResultProperties,
+    LLMRunFrame,
+    TTSSpeakFrame,
+)
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response_universal import (
+    LLMContextAggregatorPair,
+    LLMUserAggregatorParams,
+)
+from pipecat.runner.types import RunnerArguments
+from pipecat.runner.utils import create_transport
+from pipecat.services.anthropic.llm import AnthropicLLMService
+from pipecat.services.cartesia.tts import CartesiaTTSService
+from pipecat.services.deepgram.stt import DeepgramSTTService
+from pipecat.services.llm_service import FunctionCallParams
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.daily.transport import DailyParams
+from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
+
+load_dotenv(override=True)
+
+
+async def track_current_location(params: FunctionCallParams):
+    """Simulate a GPS tracker reporting position during a road trip.
+
+    Step 1 – San Francisco (trip start)     (update)
+    Step 2 – Los Angeles   (passing through) (update)
+    Step 3 – San Diego     (destination)     (final result)
+    """
+
+    # First update: initial city estimate.
+    gps = {"lat": 37.7310, "lng": -122.4527}
+    await params.result_callback(
+        {"gps": gps, "city": "San Francisco"},
+        properties=FunctionCallResultProperties(is_final=False),
+    )
+
+    # Second update: revised city estimate.
+    await asyncio.sleep(10)
+    gps = {"lat": 33.96003, "lng": -118.40639}
+    await params.result_callback(
+        {"gps": gps, "city": "Los Angeles"},
+        properties=FunctionCallResultProperties(is_final=False),
+    )
+
+    # Final result: confirmed city.
+    await asyncio.sleep(10)
+    gps = {"lat": 32.743569, "lng": -117.20466}
+    await params.result_callback({"gps": gps, "city": "San Diego"})
+
+
+# We use lambdas to defer transport parameter creation until the transport
+# type is selected at runtime.
+transport_params = {
+    "daily": lambda: DailyParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+    "twilio": lambda: FastAPIWebsocketParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+}
+
+
+async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
+    logger.info(f"Starting bot")
+
+    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+    tts = CartesiaTTSService(
+        api_key=os.getenv("CARTESIA_API_KEY"),
+        settings=CartesiaTTSService.Settings(
+            voice="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
+        ),
+    )
+
+    llm = AnthropicLLMService(
+        api_key=os.getenv("ANTHROPIC_API_KEY"),
+        settings=AnthropicLLMService.Settings(
+            system_instruction=(
+                "You are a helpful assistant in a voice conversation. "
+                "Your responses will be spoken aloud, so avoid emojis, bullet points, or other "
+                "formatting that can't be spoken. "
+                "You have access to a function that starts tracking the user's location and "
+                "provides regular updates on it. When you receive the final location, tell the user "
+                "the destination has been reached."
+            ),
+        ),
+    )
+
+    # cancel_on_interruption=False makes this an async function call: the LLM
+    # continues the conversation immediately and receives updates/result later.
+    llm.register_function(
+        "track_current_location",
+        track_current_location,
+        cancel_on_interruption=False,
+        timeout_secs=30,
+    )
+
+    @llm.event_handler("on_function_calls_started")
+    async def on_function_calls_started(service, function_calls):
+        await tts.queue_frame(TTSSpeakFrame("Sure, tracking your location now."))
+
+    location_function = FunctionSchema(
+        name="track_current_location",
+        description="Start tracking the user's current GPS location, reporting position updates until the user reaches their destination.",
+        properties={},
+        required=[],
+    )
+    tools = ToolsSchema(standard_tools=[location_function])
+
+    context = LLMContext(tools=tools)
+    user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
+        context,
+        user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()),
+    )
+
+    pipeline = Pipeline(
+        [
+            transport.input(),
+            stt,
+            user_aggregator,
+            llm,
+            tts,
+            transport.output(),
+            assistant_aggregator,
+        ]
+    )
+
+    task = PipelineTask(
+        pipeline,
+        params=PipelineParams(
+            enable_metrics=True,
+            enable_usage_metrics=True,
+        ),
+    )
+
+    @transport.event_handler("on_client_connected")
+    async def on_client_connected(transport, client):
+        logger.info(f"Client connected")
+        # Kick off the conversation.
+        context.add_message(
+            {"role": "developer", "content": "Please introduce yourself to the user."}
+        )
+        await task.queue_frames([LLMRunFrame()])
+
+    @transport.event_handler("on_client_disconnected")
+    async def on_client_disconnected(transport, client):
+        logger.info(f"Client disconnected")
+        await task.cancel()
+
+    runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
+
+    await runner.run(task)
+
+
+async def bot(runner_args: RunnerArguments):
+    """Main bot entry point compatible with Pipecat Cloud."""
+    transport = await create_transport(runner_args, transport_params)
+    await run_bot(transport, runner_args)
+
+
+if __name__ == "__main__":
+    from pipecat.runner.run import main
+
+    main()

examples/function-calling/function-calling-google-async-stream.py

@@ -0,0 +1,208 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Example: async function call with intermediate updates.
+
+The ``track_current_location`` tool simulates a GPS tracker reporting the
+device's position during a road trip from San Francisco to San Diego.  It
+sends two intermediate updates (via ``params.result_callback`` with
+``is_final=False``) as the vehicle passes through cities along the way, then
+delivers the final destination (via ``params.result_callback``).  Each update
+returns the same structure with a different city:
+
+  Update 1 – {gps, city: "San Francisco"}   ← trip start
+  Update 2 – {gps, city: "Los Angeles"}     ← passing through
+  Final     – {gps, city: "San Diego"}      ← destination reached
+
+Because the function is registered with ``cancel_on_interruption=False``, the
+LLM can keep talking while the trip is in progress; each position update
+arrives as a developer message so the LLM can narrate the journey to the user.
+"""
+
+import asyncio
+import os
+
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.adapters.schemas.function_schema import FunctionSchema
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import (
+    FunctionCallResultProperties,
+    LLMRunFrame,
+    TTSSpeakFrame,
+)
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response_universal import (
+    LLMContextAggregatorPair,
+    LLMUserAggregatorParams,
+)
+from pipecat.runner.types import RunnerArguments
+from pipecat.runner.utils import create_transport
+from pipecat.services.cartesia.tts import CartesiaTTSService
+from pipecat.services.deepgram.stt import DeepgramSTTService
+from pipecat.services.google.llm import GoogleLLMService
+from pipecat.services.llm_service import FunctionCallParams
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.daily.transport import DailyParams
+from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
+
+load_dotenv(override=True)
+
+
+async def track_current_location(params: FunctionCallParams):
+    """Simulate a GPS tracker reporting position during a road trip.
+
+    Step 1 – San Francisco (trip start)     (update)
+    Step 2 – Los Angeles   (passing through) (update)
+    Step 3 – San Diego     (destination)     (final result)
+    """
+
+    # First update: initial city estimate.
+    gps = {"lat": 37.7310, "lng": -122.4527}
+    await params.result_callback(
+        {"gps": gps, "city": "San Francisco"},
+        properties=FunctionCallResultProperties(is_final=False),
+    )
+
+    # Second update: revised city estimate.
+    await asyncio.sleep(10)
+    gps = {"lat": 33.96003, "lng": -118.40639}
+    await params.result_callback(
+        {"gps": gps, "city": "Los Angeles"},
+        properties=FunctionCallResultProperties(is_final=False),
+    )
+
+    # Final result: confirmed city.
+    await asyncio.sleep(10)
+    gps = {"lat": 32.743569, "lng": -117.20466}
+    await params.result_callback({"gps": gps, "city": "San Diego"})
+
+
+# We use lambdas to defer transport parameter creation until the transport
+# type is selected at runtime.
+transport_params = {
+    "daily": lambda: DailyParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+    "twilio": lambda: FastAPIWebsocketParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+}
+
+
+async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
+    logger.info(f"Starting bot")
+
+    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+    tts = CartesiaTTSService(
+        api_key=os.getenv("CARTESIA_API_KEY"),
+        settings=CartesiaTTSService.Settings(
+            voice="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
+        ),
+    )
+
+    llm = GoogleLLMService(
+        api_key=os.getenv("GOOGLE_API_KEY"),
+        settings=GoogleLLMService.Settings(
+            system_instruction=(
+                "You are a helpful assistant in a voice conversation. "
+                "Your responses will be spoken aloud, so avoid emojis, bullet points, or other "
+                "formatting that can't be spoken. "
+                "You have access to a function that starts tracking the user's location and "
+                "provides regular updates on it. When you receive the final location, tell the user "
+                "the destination has been reached."
+            ),
+        ),
+    )
+
+    # cancel_on_interruption=False makes this an async function call: the LLM
+    # continues the conversation immediately and receives updates/result later.
+    llm.register_function(
+        "track_current_location",
+        track_current_location,
+        cancel_on_interruption=False,
+        timeout_secs=30,
+    )
+
+    @llm.event_handler("on_function_calls_started")
+    async def on_function_calls_started(service, function_calls):
+        await tts.queue_frame(TTSSpeakFrame("Sure, tracking your location now."))
+
+    location_function = FunctionSchema(
+        name="track_current_location",
+        description="Start tracking the user's current GPS location, reporting position updates until the user reaches their destination.",
+        properties={},
+        required=[],
+    )
+    tools = ToolsSchema(standard_tools=[location_function])
+
+    context = LLMContext(tools=tools)
+    user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
+        context,
+        user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()),
+    )
+
+    pipeline = Pipeline(
+        [
+            transport.input(),
+            stt,
+            user_aggregator,
+            llm,
+            tts,
+            transport.output(),
+            assistant_aggregator,
+        ]
+    )
+
+    task = PipelineTask(
+        pipeline,
+        params=PipelineParams(
+            enable_metrics=True,
+            enable_usage_metrics=True,
+        ),
+    )
+
+    @transport.event_handler("on_client_connected")
+    async def on_client_connected(transport, client):
+        logger.info(f"Client connected")
+        # Kick off the conversation.
+        context.add_message(
+            {"role": "developer", "content": "Please introduce yourself to the user."}
+        )
+        await task.queue_frames([LLMRunFrame()])
+
+    @transport.event_handler("on_client_disconnected")
+    async def on_client_disconnected(transport, client):
+        logger.info(f"Client disconnected")
+        await task.cancel()
+
+    runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
+
+    await runner.run(task)
+
+
+async def bot(runner_args: RunnerArguments):
+    """Main bot entry point compatible with Pipecat Cloud."""
+    transport = await create_transport(runner_args, transport_params)
+    await run_bot(transport, runner_args)
+
+
+if __name__ == "__main__":
+    from pipecat.runner.run import main
+
+    main()

examples/function-calling/function-calling-openai-async-stream.py

@@ -0,0 +1,205 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Example: async function call with intermediate updates.
+
+The ``track_current_location`` tool simulates a GPS tracker reporting the
+device's position during a road trip from San Francisco to San Diego.  It
+sends two intermediate updates (via ``params.result_callback`` with
+``is_final=False``) as the vehicle passes through cities along the way, then
+delivers the final destination (via ``params.result_callback``).  Each update
+returns the same structure with a different city:
+
+  Update 1 – {gps, city: "San Francisco"}   ← trip start
+  Update 2 – {gps, city: "Los Angeles"}     ← passing through
+  Final     – {gps, city: "San Diego"}      ← destination reached
+
+Because the function is registered with ``cancel_on_interruption=False``, the
+LLM can keep talking while the trip is in progress; each position update
+arrives as a developer message so the LLM can narrate the journey to the user.
+"""
+
+import asyncio
+import os
+
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.adapters.schemas.function_schema import FunctionSchema
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import (
+    FunctionCallResultProperties,
+    LLMRunFrame,
+    TTSSpeakFrame,
+)
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response_universal import (
+    LLMContextAggregatorPair,
+    LLMUserAggregatorParams,
+)
+from pipecat.runner.types import RunnerArguments
+from pipecat.runner.utils import create_transport
+from pipecat.services.cartesia.tts import CartesiaTTSService
+from pipecat.services.deepgram.stt import DeepgramSTTService
+from pipecat.services.llm_service import FunctionCallParams
+from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.daily.transport import DailyParams
+from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
+
+load_dotenv(override=True)
+
+
+async def track_current_location(params: FunctionCallParams):
+    """Simulate a GPS tracker reporting position during a road trip.
+
+    Step 1 – San Francisco (trip start)     (update)
+    Step 2 – Los Angeles   (passing through) (update)
+    Step 3 – San Diego     (destination)     (final result)
+    """
+
+    # First update: initial city estimate.
+    gps = {"lat": 37.7310, "lng": -122.4527}
+    await params.result_callback(
+        {"gps": gps, "city": "San Francisco"},
+        properties=FunctionCallResultProperties(is_final=False),
+    )
+
+    # Second update: revised city estimate.
+    await asyncio.sleep(10)
+    gps = {"lat": 33.96003, "lng": -118.40639}
+    await params.result_callback(
+        {"gps": gps, "city": "Los Angeles"},
+        properties=FunctionCallResultProperties(is_final=False),
+    )
+
+    # Final result: confirmed city.
+    await asyncio.sleep(10)
+    gps = {"lat": 32.743569, "lng": -117.20466}
+    await params.result_callback({"gps": gps, "city": "San Diego"})
+
+
+# We use lambdas to defer transport parameter creation until the transport
+# type is selected at runtime.
+transport_params = {
+    "daily": lambda: DailyParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+    "twilio": lambda: FastAPIWebsocketParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+}
+
+
+async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
+    logger.info(f"Starting bot")
+
+    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+    tts = CartesiaTTSService(
+        api_key=os.getenv("CARTESIA_API_KEY"),
+        settings=CartesiaTTSService.Settings(
+            voice="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
+        ),
+    )
+
+    llm = OpenAILLMService(
+        api_key=os.getenv("OPENAI_API_KEY"),
+        settings=OpenAILLMService.Settings(
+            system_instruction=(
+                "You are a helpful assistant in a voice conversation. "
+                "Your responses will be spoken aloud, so avoid emojis, bullet points, or other "
+                "formatting that can't be spoken. "
+                "You have access to a function that starts tracking the user's location and "
+                "provides regular updates on it. When you receive the final location, tell the user "
+                "the destination has been reached."
+            ),
+        ),
+    )
+
+    # cancel_on_interruption=False makes this an async function call: the LLM
+    # continues the conversation immediately and receives updates/result later.
+    llm.register_function(
+        "track_current_location",
+        track_current_location,
+        cancel_on_interruption=False,
+        timeout_secs=30,
+    )
+
+    @llm.event_handler("on_function_calls_started")
+    async def on_function_calls_started(service, function_calls):
+        await tts.queue_frame(TTSSpeakFrame("Sure, tracking your location now."))
+
+    location_function = FunctionSchema(
+        name="track_current_location",
+        description="Start tracking the user's current GPS location, reporting position updates until the user reaches their destination.",
+        properties={},
+        required=[],
+    )
+    tools = ToolsSchema(standard_tools=[location_function])
+
+    context = LLMContext(tools=tools)
+    user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
+        context,
+        user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()),
+    )
+
+    pipeline = Pipeline(
+        [
+            transport.input(),
+            stt,
+            user_aggregator,
+            llm,
+            tts,
+            transport.output(),
+            assistant_aggregator,
+        ]
+    )
+
+    task = PipelineTask(
+        pipeline,
+        params=PipelineParams(
+            enable_metrics=True,
+            enable_usage_metrics=True,
+        ),
+    )
+
+    @transport.event_handler("on_client_connected")
+    async def on_client_connected(transport, client):
+        logger.info(f"Client connected")
+        # Kick off the conversation.
+        await task.queue_frames([LLMRunFrame()])
+
+    @transport.event_handler("on_client_disconnected")
+    async def on_client_disconnected(transport, client):
+        logger.info(f"Client disconnected")
+        await task.cancel()
+
+    runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
+
+    await runner.run(task)
+
+
+async def bot(runner_args: RunnerArguments):
+    """Main bot entry point compatible with Pipecat Cloud."""
+    transport = await create_transport(runner_args, transport_params)
+    await run_bot(transport, runner_args)
+
+
+if __name__ == "__main__":
+    from pipecat.runner.run import main
+
+    main()

examples/function-calling/function-calling-openai-responses-async-stream.py

@@ -0,0 +1,205 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Example: async function call with intermediate updates.
+
+The ``track_current_location`` tool simulates a GPS tracker reporting the
+device's position during a road trip from San Francisco to San Diego.  It
+sends two intermediate updates (via ``params.result_callback`` with
+``is_final=False``) as the vehicle passes through cities along the way, then
+delivers the final destination (via ``params.result_callback``).  Each update returns the same structure with a
+different city:
+
+  Update 1 – {gps, city: "San Francisco"}   ← trip start
+  Update 2 – {gps, city: "Los Angeles"}     ← passing through
+  Final     – {gps, city: "San Diego"}      ← destination reached
+
+Because the function is registered with ``cancel_on_interruption=False``, the
+LLM can keep talking while the trip is in progress; each position update
+arrives as a developer message so the LLM can narrate the journey to the user.
+"""
+
+import asyncio
+import os
+
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.adapters.schemas.function_schema import FunctionSchema
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import (
+    FunctionCallResultProperties,
+    LLMRunFrame,
+    TTSSpeakFrame,
+)
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response_universal import (
+    LLMContextAggregatorPair,
+    LLMUserAggregatorParams,
+)
+from pipecat.runner.types import RunnerArguments
+from pipecat.runner.utils import create_transport
+from pipecat.services.cartesia.tts import CartesiaTTSService
+from pipecat.services.deepgram.stt import DeepgramSTTService
+from pipecat.services.llm_service import FunctionCallParams
+from pipecat.services.openai.responses.llm import OpenAIResponsesLLMService
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.daily.transport import DailyParams
+from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
+
+load_dotenv(override=True)
+
+
+async def track_current_location(params: FunctionCallParams):
+    """Simulate a GPS tracker reporting position during a road trip.
+
+    Step 1 – San Francisco (trip start)     (update)
+    Step 2 – Los Angeles   (passing through) (update)
+    Step 3 – San Diego     (destination)     (final result)
+    """
+
+    # First update: initial city estimate.
+    gps = {"lat": 37.7310, "lng": -122.4527}
+    await params.result_callback(
+        {"gps": gps, "city": "San Francisco"},
+        properties=FunctionCallResultProperties(is_final=False),
+    )
+
+    # Second update: revised city estimate.
+    await asyncio.sleep(10)
+    gps = {"lat": 33.96003, "lng": -118.40639}
+    await params.result_callback(
+        {"gps": gps, "city": "Los Angeles"},
+        properties=FunctionCallResultProperties(is_final=False),
+    )
+
+    # Final result: confirmed city.
+    await asyncio.sleep(10)
+    gps = {"lat": 32.743569, "lng": -117.20466}
+    await params.result_callback({"gps": gps, "city": "San Diego"})
+
+
+# We use lambdas to defer transport parameter creation until the transport
+# type is selected at runtime.
+transport_params = {
+    "daily": lambda: DailyParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+    "twilio": lambda: FastAPIWebsocketParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+}
+
+
+async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
+    logger.info(f"Starting bot")
+
+    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+    tts = CartesiaTTSService(
+        api_key=os.getenv("CARTESIA_API_KEY"),
+        settings=CartesiaTTSService.Settings(
+            voice="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
+        ),
+    )
+
+    llm = OpenAIResponsesLLMService(
+        api_key=os.getenv("OPENAI_API_KEY"),
+        settings=OpenAIResponsesLLMService.Settings(
+            system_instruction=(
+                "You are a helpful assistant in a voice conversation. "
+                "Your responses will be spoken aloud, so avoid emojis, bullet points, or other "
+                "formatting that can't be spoken. "
+                "You have access to a function that starts tracking a moving device's location and "
+                "provides regular updates on it. When you receive the final location, tell the user "
+                "the destination has been reached and announce the final city."
+            ),
+        ),
+    )
+
+    # cancel_on_interruption=False makes this an async function call: the LLM
+    # continues the conversation immediately and receives updates/result later.
+    llm.register_function(
+        "track_current_location",
+        track_current_location,
+        cancel_on_interruption=False,
+        timeout_secs=30,
+    )
+
+    @llm.event_handler("on_function_calls_started")
+    async def on_function_calls_started(service, function_calls):
+        await tts.queue_frame(TTSSpeakFrame("Sure, tracking your location now."))
+
+    location_function = FunctionSchema(
+        name="track_current_location",
+        description="Track the device's current GPS location during a road trip, reporting position updates as the vehicle moves through cities until it reaches the final destination.",
+        properties={},
+        required=[],
+    )
+    tools = ToolsSchema(standard_tools=[location_function])
+
+    context = LLMContext(tools=tools)
+    user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
+        context,
+        user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()),
+    )
+
+    pipeline = Pipeline(
+        [
+            transport.input(),
+            stt,
+            user_aggregator,
+            llm,
+            tts,
+            transport.output(),
+            assistant_aggregator,
+        ]
+    )
+
+    task = PipelineTask(
+        pipeline,
+        params=PipelineParams(
+            enable_metrics=True,
+            enable_usage_metrics=True,
+        ),
+    )
+
+    @transport.event_handler("on_client_connected")
+    async def on_client_connected(transport, client):
+        logger.info(f"Client connected")
+        # Kick off the conversation.
+        await task.queue_frames([LLMRunFrame()])
+
+    @transport.event_handler("on_client_disconnected")
+    async def on_client_disconnected(transport, client):
+        logger.info(f"Client disconnected")
+        await task.cancel()
+
+    runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
+
+    await runner.run(task)
+
+
+async def bot(runner_args: RunnerArguments):
+    """Main bot entry point compatible with Pipecat Cloud."""
+    transport = await create_transport(runner_args, transport_params)
+    await run_bot(transport, runner_args)
+
+
+if __name__ == "__main__":
+    from pipecat.runner.run import main
+
+    main()

src/pipecat/frames/frames.py

@@ -663,10 +663,14 @@ class FunctionCallResultProperties:
     Parameters:
         run_llm: Whether to run the LLM after receiving this result.
         on_context_updated: Callback to execute when context is updated.
+        is_final: Whether this is the final result for the function call. When
+            ``False`` the result is treated as an intermediate update. Defaults to ``True``.
+            Only meaningful for async function calls (``cancel_on_interruption=False``).
     """
 
     run_llm: Optional[bool] = None
     on_context_updated: Optional[Callable[[], Awaitable[None]]] = None
+    is_final: bool = True
 
 
 @dataclass

src/pipecat/processors/aggregators/llm_response_universal.py

@@ -25,6 +25,8 @@ from pipecat.audio.vad.vad_analyzer import VADAnalyzer
 from pipecat.audio.vad.vad_controller import VADController
 from pipecat.frames.frames import (
     AssistantImageRawFrame,
+    BotStartedSpeakingFrame,
+    BotStoppedSpeakingFrame,
     CancelFrame,
     EndFrame,
     Frame,
@@ -832,6 +834,13 @@ class LLMAssistantAggregator(LLMContextAggregator):
         self._context_updated_tasks: Set[asyncio.Task] = set()
 
         self._user_speaking: bool = False
+        self._bot_speaking: bool = False
+
+        # When a function call result arrives while the bot is speaking, we defer the LLM
+        # re-invocation until the bot stops speaking. This flag is set to True in that case
+        # so that `BotStoppedSpeakingFrame` knows to push a context frame. Multiple results
+        # arriving in the same speaking window are bundled into a single deferred push.
+        self._push_context_on_bot_stopped_speaking: bool = False
 
         self._assistant_turn_start_timestamp = ""
 
@@ -872,6 +881,7 @@ class LLMAssistantAggregator(LLMContextAggregator):
         """Reset the aggregation state."""
         await super().reset()
         await self._reset_thought_aggregation()  # Just to be safe
+        self._push_context_on_bot_stopped_speaking = False
 
     async def _reset_thought_aggregation(self):
         """Reset the thought aggregation state."""
@@ -943,6 +953,15 @@ class LLMAssistantAggregator(LLMContextAggregator):
         elif isinstance(frame, UserStoppedSpeakingFrame):
             self._user_speaking = False
             await self.push_frame(frame, direction)
+        elif isinstance(frame, BotStartedSpeakingFrame):
+            self._bot_speaking = True
+            await self.push_frame(frame, direction)
+        elif isinstance(frame, BotStoppedSpeakingFrame):
+            self._bot_speaking = False
+            await self.push_frame(frame, direction)
+            if self._push_context_on_bot_stopped_speaking and not self._user_speaking:
+                logger.debug(f"{self}: Bot stopped speaking — pushing deferred context frame!")
+                await self.push_context_frame(FrameDirection.UPSTREAM)
         else:
             await self.push_frame(frame, direction)
 
@@ -973,6 +992,15 @@ class LLMAssistantAggregator(LLMContextAggregator):
 
         return aggregation
 
+    async def push_context_frame(self, direction: FrameDirection = FrameDirection.DOWNSTREAM):
+        """Push a context frame in the specified direction.
+
+        Args:
+            direction: The direction to push the frame (upstream or downstream).
+        """
+        await super().push_context_frame(direction)
+        self._push_context_on_bot_stopped_speaking = False
+
     async def _handle_llm_run(self, frame: LLMRunFrame):
         await self.push_context_frame(FrameDirection.UPSTREAM)
 
@@ -1036,9 +1064,12 @@ class LLMAssistantAggregator(LLMContextAggregator):
                     "content": json.dumps(
                         {
                             "type": "async_tool",
-                            "status": "started",
+                            "status": "running",
                             "tool_call_id": frame.tool_call_id,
-                            "description": "The tool associated with this tool_call_id is still in progress, and the result is not yet available. It will be provided in a subsequent message with the same tool_call_id.",
+                            "description": "An asynchronous task associated with this tool_call_id has started running. "
+                            + "Expect results to arrive later as developer messages that look roughly like this one (with 'type=async_tool' and a matching tool_call_id) but with a 'result' field. "
+                            + "Note that there *may* be more than one result (i.e., a stream of results), but there doesn't have to be (there may be only one). "
+                            + "The last result will come in a message with 'status=finished'.",
                         }
                     ),
                     "tool_call_id": frame.tool_call_id,
@@ -1066,33 +1097,14 @@ class LLMAssistantAggregator(LLMContextAggregator):
             return
 
         in_progress_frame = self._function_calls_in_progress[frame.tool_call_id]
-        is_async = not in_progress_frame.cancel_on_interruption if in_progress_frame else False
         group_id = in_progress_frame.group_id if in_progress_frame else None
-
-        del self._function_calls_in_progress[frame.tool_call_id]
-
         properties = frame.properties
+        is_final = frame.properties.is_final if frame.properties else True
 
-        result = json.dumps(frame.result, ensure_ascii=False) if frame.result else "COMPLETED"
+        if is_final:
-
+            await self._handle_function_call_finished(frame, in_progress_frame)
-        if is_async:
-            # For async function calls instead of updating the existing IN_PROGRESS tool message we inject
-            # a new developer message so the LLM is notified of the completed result.
-            self._context.add_message(
-                {
-                    "role": "developer",
-                    "content": json.dumps(
-                        {
-                            "type": "async_tool",
-                            "tool_call_id": frame.tool_call_id,
-                            "status": "finished",
-                            "result": result,
-                        }
-                    ),
-                }
-            )
         else:
-            self._update_function_call_result(frame.function_name, frame.tool_call_id, result)
+            await self._handle_function_call_intermediate_result(frame, in_progress_frame)
 
         run_llm = False
 
@@ -1119,14 +1131,38 @@ class LLMAssistantAggregator(LLMContextAggregator):
                 # otherwise always execute as soon as we receive the result.
                 if group_id:
                     run_llm = not any(
-                        f is not None and f.group_id == group_id
+                        f is not None
+                        and f.group_id == group_id
+                        # We are now able to receive "updates", so the current
+                        # frame can still be in the in progress list, and we need to
+                        # ignore it.
+                        and f.tool_call_id != frame.tool_call_id
                         for f in self._function_calls_in_progress.values()
                     )
                 else:
                     run_llm = True
 
         if run_llm and not self._user_speaking:
-            await self.push_context_frame(FrameDirection.UPSTREAM)
+            if self.has_queued_frame(FunctionCallResultFrame):
+                # Another FunctionCallResultFrame is already queued. Defer the context push
+                # to bundle all results into a single LLM call instead of triggering one
+                # inference pass per result. The context will be pushed once the last
+                # function call in the queue is processed.
+                logger.debug(
+                    f"{self}: More FunctionCallResultFrames queued — deferring context frame push."
+                )
+            elif self._bot_speaking:
+                # Defer the context frame push until the bot finishes speaking. If multiple
+                # function call results arrive while the bot is speaking, they all accumulate
+                # in the context and a single push is performed once speaking stops, preventing
+                # the LLM from running multiple times and producing duplicated responses.
+                # This should be an edge case, since it would require a FunctionCallResultFrame
+                # being queued between an LLM response start and end frame.
+                logger.debug(f"{self}: Bot is speaking — deferring context frame push.")
+                self._push_context_on_bot_stopped_speaking = True
+            else:
+                logger.debug(f"{self}: Pushing context frame!")
+                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
@@ -1137,6 +1173,70 @@ class LLMAssistantAggregator(LLMContextAggregator):
             self._context_updated_tasks.add(task)
             task.add_done_callback(self._context_updated_task_finished)
 
+    async def _handle_function_call_intermediate_result(
+        self, frame: FunctionCallResultFrame, in_progress_frame: FunctionCallInProgressFrame
+    ):
+        """Handle an intermediate result for an async function call.
+
+        Injects an intermediate developer message into the context without
+        removing the call from the in-progress map.
+        """
+        if not frame.result:
+            logger.warning(f"{self} result_callback called with is_final=False but no result!")
+            return
+
+        result = json.dumps(frame.result, ensure_ascii=False)
+        self._context.add_message(
+            {
+                "role": "developer",
+                "content": json.dumps(
+                    {
+                        "type": "async_tool",
+                        "tool_call_id": frame.tool_call_id,
+                        "status": "running",
+                        "description": "This is an intermediate result for the asynchronous task associated with this tool_call_id. "
+                        + "The task is still running. More intermediate results may follow, or the next result may be the final one with 'status=finished'.",
+                        "result": result,
+                    }
+                ),
+            }
+        )
+
+    async def _handle_function_call_finished(
+        self, frame: FunctionCallResultFrame, in_progress_frame: FunctionCallInProgressFrame
+    ):
+        """Handle the final result of a function call.
+
+        Removes the call from the in-progress map, updates the context, and
+        triggers LLM inference when appropriate.
+        """
+        is_async = not in_progress_frame.cancel_on_interruption
+        del self._function_calls_in_progress[frame.tool_call_id]
+
+        result = json.dumps(frame.result, ensure_ascii=False) if frame.result else "COMPLETED"
+
+        if is_async:
+            # For async function calls inject a developer message so the LLM is
+            # notified of the completed result instead of updating the IN_PROGRESS
+            # tool message.
+            self._context.add_message(
+                {
+                    "role": "developer",
+                    "content": json.dumps(
+                        {
+                            "type": "async_tool",
+                            "tool_call_id": frame.tool_call_id,
+                            "status": "finished",
+                            "description": "This is the final result for the asynchronous task associated with this tool_call_id. "
+                            + "The task has completed. No further results will arrive for this tool_call_id.",
+                            "result": result,
+                        }
+                    ),
+                }
+            )
+        else:
+            self._update_function_call_result(frame.function_name, frame.tool_call_id, result)
+
     async def _handle_function_call_cancel(self, frame: FunctionCallCancelFrame):
         logger.debug(
             f"{self} FunctionCallCancelFrame: [{frame.function_name}:{frame.tool_call_id}]"

src/pipecat/processors/frame_processor.py

@@ -25,6 +25,7 @@ from typing import (
     Optional,
     Tuple,
     Type,
+    Union,
 )
 
 from loguru import logger
@@ -928,6 +929,21 @@ class FrameProcessor(BaseObject):
         """Reset non-system frame processing queue."""
         self.__process_queue.reset()
 
+    def has_queued_frame(self, frame_type: Union[Type[Frame], Type[UninterruptibleFrame]]) -> bool:
+        """Return True if a frame of the given type is waiting in the processing queue.
+
+        Delegates to :meth:`FrameQueue.has_frame` so the check is O(distinct
+        enqueued types) with no queue scanning.  ``frame_type`` may be any
+        ``Frame`` subclass or ``UninterruptibleFrame`` (a mixin).
+
+        Args:
+            frame_type: The frame class (or mixin) to look for.
+
+        Returns:
+            True if at least one matching frame is queued.
+        """
+        return self.__process_queue.has_frame(frame_type)
+
     async def __cancel_process_task(self):
         """Cancel the non-system frame processing task."""
         if self.__process_frame_task:

src/pipecat/services/llm_service.py

@@ -73,7 +73,10 @@ FunctionCallHandler = Callable[["FunctionCallParams"], Awaitable[None]]
 class FunctionCallResultCallback(Protocol):
     """Protocol for function call result callbacks.
 
-    Handles the result of an LLM function call execution.
+    Used for both final results and intermediate updates. Pass
+    ``properties=FunctionCallResultProperties(is_final=False)`` to send an
+    intermediate update (only valid for async function calls registered with
+    ``cancel_on_interruption=False``).
     """
 
     async def __call__(
@@ -82,8 +85,9 @@ class FunctionCallResultCallback(Protocol):
         """Call the result callback.
 
         Args:
-            result: The result of the function call.
+            result: The result of the function call, or an intermediate update.
-            properties: Optional properties for the result.
+            properties: Optional properties. Set ``is_final=False`` to send an
+                intermediate update instead of the final result.
         """
         ...
 
@@ -98,7 +102,10 @@ class FunctionCallParams:
         arguments: The arguments for the function.
         llm: The LLMService instance being used.
         context: The LLM context.
-        result_callback: Callback to handle the result of the function call.
+        result_callback: Callback to deliver the result of the function call.
+            For async function calls (``cancel_on_interruption=False``), call
+            it with ``properties=FunctionCallResultProperties(is_final=False)``
+            to push intermediate updates before the final result.
     """
 
     function_name: str
@@ -756,10 +763,21 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
 
         timeout_task: Optional[asyncio.Task] = None
 
-        # Define a callback function that pushes a FunctionCallResultFrame upstream & downstream.
+        # Single callback for both intermediate updates and final results.
+        # Pass properties=FunctionCallResultProperties(is_final=False) for updates.
         async def function_call_result_callback(
             result: Any, *, properties: Optional[FunctionCallResultProperties] = None
         ):
+            is_final = properties.is_final if properties else True
+            if not is_final and item.cancel_on_interruption:
+                logger.warning(
+                    f"{self} result_callback called with is_final=False on sync function call"
+                    f" [{runner_item.function_name}:{runner_item.tool_call_id}]."
+                    " Intermediate updates are only valid for async function calls"
+                    " (cancel_on_interruption=False)."
+                )
+                return
+
             nonlocal timeout_task
 
             # Cancel timeout task if it exists

src/pipecat/utils/frame_queue.py

@@ -7,7 +7,7 @@
 """Frame queue utilities for Pipecat pipeline processors."""
 
 import asyncio
-from typing import Any, Callable
+from typing import Any, Callable, Type, Union
 
 from pipecat.frames.frames import Frame, UninterruptibleFrame
 
@@ -41,6 +41,27 @@ class FrameQueue(asyncio.Queue):
         self._frame_getter = frame_getter
         self._uninterruptible_count: int = 0
 
+    def has_frame(self, frame_type: Union[Type[Frame], Type[UninterruptibleFrame]]) -> bool:
+        """Return True if any frame of the given type is in the queue.
+
+        ``frame_type`` may be ``Frame``, ``UninterruptibleFrame`` (a mixin, not a
+        ``Frame`` subclass), or any concrete frame type.
+
+        Note:
+            This inspects the internal `_queue` (deque) of asyncio.Queue.
+            This is not part of the public API but is stable in CPython.
+
+        Args:
+            frame_type: The frame class to check for.
+
+        Returns:
+            True if at least one enqueued frame is an instance of ``frame_type``.
+        """
+        for item in self._queue:
+            if isinstance(self._frame_getter(item), frame_type):
+                return True
+        return False
+
     @property
     def has_uninterruptible(self) -> bool:
         """Return True if any UninterruptibleFrame is currently in the queue."""