wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #4217 from pipecat-ai/filipi/async_tools

Browse Source 
Supporting async function calls.
This commit is contained in:
Filipi da Silva Fuchter
2026-04-07 09:35:03 -03:00
committed by GitHub
parent a451c42dc7 d8dc6bc7d0
commit 6eccd16543
17 changed files with 1066 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
changelog/4217.added.2.md Normal file
View File

@@ -0,0 +1 @@
- Added `group_parallel_tools` parameter to `LLMService` (default `True`). When `True`, all function calls from the same LLM response batch share a group ID and the LLM is triggered exactly once after the last call completes. Set to `False` to trigger inference independently for each function call result as it arrives.

1
changelog/4217.added.md Normal file
View File

@@ -0,0 +1 @@
- Added async function call support to `register_function()` and `register_direct_function()` via `cancel_on_interruption=False`. When set to `False`, the LLM continues the conversation immediately without waiting for the function result. The result is injected back into the context as a `developer` message once available, triggering a new LLM inference at that point.

1
changelog/4217.changed.md Normal file
View File

@@ -0,0 +1 @@
- When multiple function calls are returned in a single LLM response, by default (when `group_parallel_tools=True`) the LLM is now triggered exactly once after the last call in the batch completes, rather than waiting for all function calls.

1
changelog/4217.fixed.2.md Normal file
View File

@@ -0,0 +1 @@
- Fixed `BaseOutputTransport` discarding pending `UninterruptibleFrame` items (e.g. function-call context updates) when an interruption arrived. The audio task is now kept alive and only interruptible frames are drained when uninterruptible frames are present in the queue.

1
changelog/4217.fixed.md Normal file
View File

@@ -0,0 +1 @@
- Fixed spurious LLM inference being triggered when a function call result arrived while the user was actively speaking. The context frame is now suppressed until the user stops speaking.

174
examples/function-calling/function-calling-anthropic-async.py Normal file
View File

@@ -0,0 +1,174 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
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 LLMRunFrame
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 fetch_weather_from_api(params: FunctionCallParams):
# Simulate a long-running API call, so we can test async function calls (cancel_on_interruption=False).
await asyncio.sleep(20)
await params.result_callback({"conditions": "nice", "temperature": "75"})
async def fetch_restaurant_recommendation(params: FunctionCallParams):
await params.result_callback({"name": "The Golden Dragon"})
# 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. Respond to what the user said in a creative, helpful, and brief way.",
),
)
# 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,
cancel_on_interruption=False,
timeout_secs=30,
)
llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation)
weather_function = FunctionSchema(
name="get_current_weather",
description="Get the current weather",
properties={
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
},
required=["location"],
)
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])
context = LLMContext(tools=tools)
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
context,
user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()),
)
pipeline = Pipeline(
[
transport.input(), # Transport user input
stt,
user_aggregator, # User spoken responses
llm, # LLM
tts, # TTS
transport.output(), # Transport bot output
assistant_aggregator, # Assistant spoken responses and tool context
]
)
task = PipelineTask(
pipeline,
params=PipelineParams(
enable_metrics=True,
enable_usage_metrics=True,
),
idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
)
@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()

250
examples/function-calling/function-calling-google-async.py Normal file
View File

@@ -0,0 +1,250 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
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 LLMRunFrame, TTSSpeakFrame, UserImageRequestFrame
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.processors.frame_processor import FrameDirection
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import (
create_transport,
get_transport_client_id,
maybe_capture_participant_camera,
)
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
load_dotenv(override=True)
async def get_weather(params: FunctionCallParams):
# Simulate a long-running API call, so we can test async function calls (cancel_on_interruption=False).
await asyncio.sleep(20)
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):
"""Fetch the user image and push it to the LLM.
When called, this function pushes a UserImageRequestFrame upstream to the
transport. As a result, the transport will request the user image and push a
UserImageRawFrame downstream which will be added to the context by the LLM
assistant aggregator. The result_callback will be invoked once the image is
retrieved and processed.
"""
user_id = params.arguments["user_id"]
question = params.arguments["question"]
logger.debug(f"Requesting image with user_id={user_id}, question={question}")
# Request a user image frame and indicate that it should be added to the
# context. Also associate it to the function call. Pass the result_callback
# so it can be invoked when the image is actually retrieved.
await params.llm.push_frame(
UserImageRequestFrame(
user_id=user_id,
text=question,
append_to_context=True,
function_name=params.function_name,
tool_call_id=params.tool_call_id,
result_callback=params.result_callback,
),
FrameDirection.UPSTREAM,
)
# 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,
video_in_enabled=True,
),
"webrtc": lambda: TransportParams(
audio_in_enabled=True,
audio_out_enabled=True,
video_in_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
),
)
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 three tools: get_weather, get_restaurant_recommendation, and get_image.
You can respond to questions about the weather using the get_weather tool.
You can answer questions about the user's video stream using the get_image tool. Some examples of phrases that \
indicate you should use the get_image tool are:
- What do you see?
- What's in the video?
- Can you describe the video?
- Tell me about what you see.
- Tell me something interesting about what you see.
- What's happening in the video?
"""
llm = GoogleLLMService(
api_key=os.getenv("GOOGLE_API_KEY"),
settings=GoogleLLMService.Settings(
system_instruction=system_prompt,
),
)
llm.register_function("get_weather", get_weather, cancel_on_interruption=False, timeout_secs=30)
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",
description="Get the current weather",
properties={
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
"format": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "The temperature unit to use. Infer this from the user's location.",
},
},
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="Called when the user requests a description of their camera feed",
properties={
"user_id": {
"type": "string",
"description": "The ID of the user to grab the image from",
},
"question": {
"type": "string",
"description": "The question that the user is asking about the image",
},
},
required=["user_id", "question"],
)
tools = ToolsSchema(standard_tools=[weather_function, get_image_function, restaurant_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,
),
idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
)
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
logger.info(f"Client connected: {client}")
await maybe_capture_participant_camera(transport, client)
client_id = get_transport_client_id(transport, client)
# Kick off the conversation.
context.add_message(
{
"role": "developer",
"content": f"Please introduce yourself to the user. Use '{client_id}' as the user ID during function calls.",
}
)
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()

189
examples/function-calling/function-calling-openai-async.py Normal file
View File

@@ -0,0 +1,189 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
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 (
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.llm_service import FunctionCallParams
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.services.openai.stt import OpenAISTTService
from pipecat.services.openai.tts import OpenAITTSService
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 fetch_weather_from_api(params: FunctionCallParams):
# Simulate a long-running API call, so we can test async function calls.
await asyncio.sleep(20)
await params.result_callback({"conditions": "nice", "temperature": "75"})
async def fetch_restaurant_recommendation(params: FunctionCallParams):
await params.result_callback({"name": "The Golden Dragon"})
# 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 = OpenAISTTService(
api_key=os.getenv("OPENAI_API_KEY"),
settings=OpenAISTTService.Settings(
model="gpt-4o-transcribe",
prompt="Expect words related weather, such as temperature and conditions. And restaurant names.",
),
)
tts = OpenAITTSService(
api_key=os.getenv("OPENAI_API_KEY"),
settings=OpenAITTSService.Settings(
voice="ballad",
),
instructions="Please speak clearly and at a moderate pace.",
)
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. Respond to what the user said in a creative, helpful, and brief way.",
),
)
# 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,
cancel_on_interruption=False,
timeout_secs=30,
)
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",
description="Get the current weather",
properties={
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
"format": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "The temperature unit to use. Infer this from the user's location.",
},
},
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"],
)
tools = ToolsSchema(standard_tools=[weather_function, restaurant_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()

191
examples/function-calling/function-calling-openai-responses-async.py Normal file
View File

@@ -0,0 +1,191 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
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 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 fetch_weather_from_api(params: FunctionCallParams):
# Simulate a long-running API call, so we can test async function calls.
await asyncio.sleep(20)
await params.result_callback({"conditions": "nice", "temperature": "75"})
async def fetch_restaurant_recommendation(params: FunctionCallParams):
await params.result_callback({"name": "The Golden Dragon"})
# 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. Respond to what the user said in a creative, helpful, and brief way.",
),
)
# 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,
cancel_on_interruption=False,
timeout_secs=30,
)
llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation)
@llm.event_handler("on_connection_error")
async def on_connection_error(service, error):
logger.error(f"LLM connection error: {error}")
@llm.event_handler("on_function_calls_started")
async def on_function_calls_started(service, function_calls):
# Avoid appending this filler message to the LLM context — it would
# alter the conversation history and prevent
# OpenAIResponsesLLMService's previous_response_id optimization from
# matching, forcing a full context resend.
await tts.queue_frame(TTSSpeakFrame("Let me check on that.", append_to_context=False))
weather_function = FunctionSchema(
name="get_current_weather",
description="Get the current weather",
properties={
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
"format": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "The temperature unit to use. Infer this from the user's location.",
},
},
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"],
)
tools = ToolsSchema(standard_tools=[weather_function, restaurant_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,
),
idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
)
@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()

7
src/pipecat/frames/frames.py
View File

@@ -1643,12 +1643,19 @@ class FunctionCallInProgressFrame(ControlFrame, UninterruptibleFrame):
tool_call_id: Unique identifier for this function call.
arguments: Arguments passed to the function.
cancel_on_interruption: Whether to cancel this call if interrupted.
When ``False`` the call is treated as asynchronous: the LLM
continues the conversation immediately without waiting for the
result, and the result is injected later via a developer message.
group_id: Identifier shared by all function calls originating from the
same LLM response batch. Used to determine when the last call in a
group completes so the LLM can be triggered exactly once.
"""
function_name: str
tool_call_id: str
arguments: Any
cancel_on_interruption: bool = False
group_id: Optional[str] = None
@dataclass

82
src/pipecat/processors/aggregators/llm_response_universal.py
View File

@@ -831,6 +831,8 @@ class LLMAssistantAggregator(LLMContextAggregator):
self._function_calls_image_results: Dict[str, UserImageRawFrame] = {}
self._context_updated_tasks: Set[asyncio.Task] = set()
self._user_speaking: bool = False
self._assistant_turn_start_timestamp = ""
self._thought_append_to_context = False
@@ -935,6 +937,12 @@ class LLMAssistantAggregator(LLMContextAggregator):
await self._handle_user_image_frame(frame)
elif isinstance(frame, AssistantImageRawFrame):
await self._handle_assistant_image_frame(frame)
elif isinstance(frame, UserStartedSpeakingFrame):
self._user_speaking = True
await self.push_frame(frame, direction)
elif isinstance(frame, UserStoppedSpeakingFrame):
self._user_speaking = False
await self.push_frame(frame, direction)
else:
await self.push_frame(frame, direction)
@@ -1019,13 +1027,31 @@ class LLMAssistantAggregator(LLMContextAggregator):
],
}
)
self._context.add_message(
{
"role": "tool",
"content": "IN_PROGRESS",
"tool_call_id": frame.tool_call_id,
}
)
is_async = not frame.cancel_on_interruption
if is_async:
self._context.add_message(
{
"role": "tool",
"content": json.dumps(
{
"type": "async_tool",
"status": "started",
"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.",
}
),
"tool_call_id": frame.tool_call_id,
}
)
else:
self._context.add_message(
{
"role": "tool",
"content": "IN_PROGRESS",
"tool_call_id": frame.tool_call_id,
}
)
self._function_calls_in_progress[frame.tool_call_id] = frame
@@ -1039,16 +1065,34 @@ 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
# Update context with the function call result
if frame.result:
result = json.dumps(frame.result, ensure_ascii=False)
self._update_function_call_result(frame.function_name, frame.tool_call_id, result)
result = json.dumps(frame.result, ensure_ascii=False) if frame.result else "COMPLETED"
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, "COMPLETED")
self._update_function_call_result(frame.function_name, frame.tool_call_id, result)
run_llm = False
@@ -1070,10 +1114,18 @@ class LLMAssistantAggregator(LLMContextAggregator):
# If the frame is indicating we should run the LLM, do it.
run_llm = frame.run_llm
else:
# If this is the last function call in progress, run the LLM.
run_llm = not bool(self._function_calls_in_progress)
# Run the LLM when this is the last function call in the group
# to complete. If group_id is set, only consider sibling calls;
# 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
for f in self._function_calls_in_progress.values()
)
else:
run_llm = True
if run_llm:
if run_llm and not self._user_speaking:
await self.push_context_frame(FrameDirection.UPSTREAM)
# Call the `on_context_updated` callback once the function call result

31
src/pipecat/processors/frame_processor.py
View File

@@ -48,6 +48,7 @@ from pipecat.observers.base_observer import BaseObserver, FrameProcessed, FrameP
from pipecat.processors.metrics.frame_processor_metrics import FrameProcessorMetrics
from pipecat.utils.asyncio.task_manager import BaseTaskManager
from pipecat.utils.base_object import BaseObject
from pipecat.utils.frame_queue import FrameQueue
class FrameDirection(Enum):
@@ -228,7 +229,7 @@ class FrameProcessor(BaseObject):
# called. To resume processing frames we need to call
# `resume_processing_frames()` which will wake up the event.
self.__should_block_frames = False
self.__process_queue = asyncio.Queue()
self.__process_queue = FrameQueue(frame_getter=lambda item: item[0])
self.__process_event: Optional[asyncio.Event] = None
self.__process_frame_task: Optional[asyncio.Task] = None
self.__process_current_frame: Optional[Frame] = None
@@ -818,9 +819,14 @@ class FrameProcessor(BaseObject):
async def _start_interruption(self):
"""Start handling an interruption by cancelling current tasks."""
try:
if isinstance(self.__process_current_frame, UninterruptibleFrame):
# We don't want to cancel UninterruptibleFrame, so we simply
# cleanup the queue.
current_is_uninterruptible = isinstance(
self.__process_current_frame, UninterruptibleFrame
)
if current_is_uninterruptible or self.__process_queue.has_uninterruptible:
# We don't want to cancel an UninterruptibleFrame (either the
# one currently being processed or one waiting in the queue),
# so we simply cleanup the queue keeping only
# UninterruptibleFrames.
self.__reset_process_queue()
else:
# Cancel and re-create the process task.
@@ -920,22 +926,7 @@ class FrameProcessor(BaseObject):
def __reset_process_queue(self):
"""Reset non-system frame processing queue."""
# Create a new queue to insert UninterruptibleFrame frames.
new_queue = asyncio.Queue()
# Process current queue and keep UninterruptibleFrame frames.
while not self.__process_queue.empty():
item = self.__process_queue.get_nowait()
frame = item[0]
if isinstance(frame, UninterruptibleFrame):
new_queue.put_nowait(item)
self.__process_queue.task_done()
# Put back UninterruptibleFrame frames into our process queue.
while not new_queue.empty():
item = new_queue.get_nowait()
self.__process_queue.put_nowait(item)
new_queue.task_done()
self.__process_queue.reset()
async def __cancel_process_task(self):
"""Cancel the non-system frame processing task."""

31
src/pipecat/services/llm_service.py
View File

@@ -8,6 +8,7 @@
import asyncio
import json
import uuid
import warnings
from dataclasses import dataclass
from typing import (
@@ -118,6 +119,9 @@ class FunctionCallRegistryItem:
function_name: The name of the function (None for catch-all handler).
handler: The handler for processing function call parameters.
cancel_on_interruption: Whether to cancel the call on interruption.
When ``False`` the call is treated as asynchronous: the LLM
continues the conversation immediately without waiting for the
result, and the result is injected later via a developer message.
timeout_secs: Optional per-tool timeout in seconds. Overrides the global
``function_call_timeout_secs`` for this specific function.
"""
@@ -141,6 +145,9 @@ class FunctionCallRunnerItem:
arguments: The arguments for the function.
context: The LLM context.
run_llm: Optional flag to control LLM execution after function call.
group_id: Shared identifier for all function calls from the same LLM
response batch. Used to trigger the LLM exactly once when the last
call in the group completes.
"""
registry_item: FunctionCallRegistryItem
@@ -149,6 +156,7 @@ class FunctionCallRunnerItem:
arguments: Mapping[str, Any]
context: LLMContext
run_llm: Optional[bool] = None
group_id: Optional[str] = None
class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
@@ -184,6 +192,7 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
def __init__(
self,
run_in_parallel: bool = True,
group_parallel_tools: bool = True,
function_call_timeout_secs: Optional[float] = None,
settings: Optional[LLMSettings] = None,
**kwargs,
@@ -193,6 +202,10 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
Args:
run_in_parallel: Whether to run function calls in parallel or sequentially.
Defaults to True.
group_parallel_tools: Whether to group parallel function calls so the LLM
is triggered exactly once after all calls in the batch complete. When
False, each function call result triggers the LLM independently as it
arrives. Defaults to True.
function_call_timeout_secs: Optional timeout in seconds for deferred function
calls.
settings: The runtime-updatable settings for the LLM service.
@@ -207,6 +220,7 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
**kwargs,
)
self._run_in_parallel = run_in_parallel
self._group_parallel_tools = group_parallel_tools
self._function_call_timeout_secs = function_call_timeout_secs
self._filter_incomplete_user_turns: bool = False
self._base_system_instruction: Optional[str] = None
@@ -547,7 +561,10 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
handler: The function handler. Should accept a single FunctionCallParams
parameter.
cancel_on_interruption: Whether to cancel this function call when an
interruption occurs. Defaults to True.
interruption occurs. When ``False`` the call is treated as
asynchronous: the LLM continues the conversation immediately
without waiting for the result, and the result is injected later
via a developer message. Defaults to True.
timeout_secs: Optional per-tool timeout in seconds. Overrides the global
``function_call_timeout_secs`` for this specific function. Defaults to
None, which uses the global timeout.
@@ -577,7 +594,10 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
Args:
handler: The direct function to register. Must follow DirectFunction protocol.
cancel_on_interruption: Whether to cancel this function call when an
interruption occurs. Defaults to True.
interruption occurs. When ``False`` the call is treated as
asynchronous: the LLM continues the conversation immediately
without waiting for the result, and the result is injected later
via a developer message. Defaults to True.
timeout_secs: Optional per-tool timeout in seconds. Overrides the global
``function_call_timeout_secs`` for this specific function. Defaults to
None, which uses the global timeout.
@@ -638,6 +658,11 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
await self.broadcast_frame(FunctionCallsStartedFrame, function_calls=function_calls)
# When group_parallel_tools is True all calls share a group_id so the
# aggregator triggers the LLM exactly once after the last one completes.
# When False, group_id is None and each result triggers inference independently.
group_id = str(uuid.uuid4()) if self._group_parallel_tools else None
runner_items = []
for function_call in function_calls:
if function_call.function_name in self._functions.keys():
@@ -657,6 +682,7 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
tool_call_id=function_call.tool_call_id,
arguments=function_call.arguments,
context=function_call.context,
group_id=group_id,
)
)
@@ -725,6 +751,7 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
tool_call_id=runner_item.tool_call_id,
arguments=runner_item.arguments,
cancel_on_interruption=item.cancel_on_interruption,
group_id=runner_item.group_id,
)
timeout_task: Optional[asyncio.Task] = None

13
src/pipecat/transports/base_output.py
View File

@@ -48,6 +48,7 @@ from pipecat.frames.frames import (
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.transports.base_transport import TransportParams
from pipecat.utils.frame_queue import FrameQueue
from pipecat.utils.time import nanoseconds_to_seconds
BOT_VAD_STOP_SECS = 0.35
@@ -518,14 +519,20 @@ class BaseOutputTransport(FrameProcessor):
_: The start interruption frame (unused).
"""
# Cancel tasks.
await self._cancel_audio_task()
await self._cancel_clock_task()
await self._cancel_video_task()
if self._audio_queue.has_uninterruptible:
# Keep the audio task running but drain all interruptible frames
# so the pending UninterruptibleFrames are still delivered.
self._audio_queue.reset()
else:
await self._cancel_audio_task()
self._create_audio_task()
# Create tasks.
self._create_video_task()
self._create_clock_task()
self._create_audio_task()
# Let's send a bot stopped speaking if we have to.
await self._bot_stopped_speaking()
@@ -609,7 +616,7 @@ class BaseOutputTransport(FrameProcessor):
def _create_audio_task(self):
"""Create the audio processing task."""
if not self._audio_task:
self._audio_queue = asyncio.Queue()
self._audio_queue = FrameQueue()
self._audio_task = self._transport.create_task(self._audio_task_handler())
async def _cancel_audio_task(self):

64
src/pipecat/utils/context/llm_context_summarization.py
View File

@@ -10,6 +10,7 @@ This module provides reusable functionality for automatically compressing conver
context when token limits are reached, enabling efficient long-running conversations.
"""
import json
import warnings
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, List, Optional
@@ -383,6 +384,35 @@ class LLMContextSummarizationUtil:
return total
@staticmethod
def _is_tool_message_pending(content: str) -> bool:
"""Return True if a tool message content represents an unresolved call.
A tool message is considered pending (unresolved) when its content is
the synchronous ``"IN_PROGRESS"`` sentinel or the async
``{"type": "async_tool", "status": "started"}`` marker — both indicate
that the actual result has not yet been written back to the context.
Args:
content: The ``content`` field of a tool-role context message.
Returns:
True if the tool call should be treated as still in progress.
"""
if content == "IN_PROGRESS":
return True
try:
parsed = json.loads(content)
if (
isinstance(parsed, dict)
and parsed.get("type") == "async_tool"
and parsed.get("status") == "started"
):
return True
except (json.JSONDecodeError, ValueError):
pass
return False
@staticmethod
def _get_earliest_function_call_not_resolved_in_range(
messages: List[dict], start_idx: int, summary_end: int
@@ -391,9 +421,13 @@ class LLMContextSummarizationUtil:
Scans messages from ``start_idx`` up to (but not including)
``summary_end`` to identify tool calls whose responses either don't
exist yet or fall in the kept portion of the context (>= summary_end).
exist yet, fall in the kept portion of the context (>= summary_end),
or are still marked as ``IN_PROGRESS`` (async calls whose results have
not yet arrived).
This prevents summarizing tool call requests when their responses would
remain in the kept context as orphans, which the OpenAI API rejects.
remain in the kept context as orphans, which the OpenAI API rejects,
and avoids summarizing async function calls before their results arrive.
Args:
messages: List of messages to check.
@@ -430,11 +464,33 @@ class LLMContextSummarizationUtil:
if tool_call_id:
pending_tool_calls[tool_call_id] = i
# Check for tool results
# Check for tool results — treat IN_PROGRESS and async "started"
# messages as still pending so they are not summarized away before
# their results arrive.
if role == "tool":
tool_call_id = msg.get("tool_call_id")
if tool_call_id and tool_call_id in pending_tool_calls:
pending_tool_calls.pop(tool_call_id)
if not LLMContextSummarizationUtil._is_tool_message_pending(
msg.get("content", "")
):
pending_tool_calls.pop(tool_call_id)
# Check for async tool completion — a developer message with
# {"type": "async_tool", "status": "finished"} signals that the
# async result has arrived and the call is now resolved.
if role == "developer":
try:
parsed = json.loads(msg.get("content", ""))
if (
isinstance(parsed, dict)
and parsed.get("type") == "async_tool"
and parsed.get("status") == "finished"
):
tool_call_id = parsed.get("tool_call_id")
if tool_call_id and tool_call_id in pending_tool_calls:
pending_tool_calls.pop(tool_call_id)
except (json.JSONDecodeError, ValueError):
pass
# If we have pending tool calls, return the earliest index
if pending_tool_calls:

71
src/pipecat/utils/frame_queue.py Normal file
View File

@@ -0,0 +1,71 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Frame queue utilities for Pipecat pipeline processors."""
import asyncio
from typing import Any, Callable
from pipecat.frames.frames import Frame, UninterruptibleFrame
class FrameQueue(asyncio.Queue):
"""An asyncio.Queue that tracks whether any UninterruptibleFrame is enqueued.
Extends ``asyncio.Queue`` and maintains an O(1) ``has_uninterruptible``
flag so interrupt-handling code can decide whether to cancel a task or
merely drain non-uninterruptible items without scanning the queue.
Items may be raw ``Frame`` objects or tuples whose first element is a
``Frame`` (e.g. ``(frame, direction, callback)``). Pass a ``frame_getter``
callable to extract the frame from each item; the default treats the item
itself as the frame.
Also exposes a ``reset()`` helper that drains all non-``UninterruptibleFrame``
items while keeping uninterruptible ones in place.
"""
def __init__(self, frame_getter: Callable[[Any], Frame] = lambda item: item):
"""Initialize the FrameQueue.
Args:
frame_getter: Callable that extracts a ``Frame`` from a queue item.
Defaults to the identity function (item is a raw ``Frame``).
Pass ``lambda item: item[0]`` when items are
``(frame, direction, callback)`` tuples.
"""
super().__init__()
self._frame_getter = frame_getter
self._uninterruptible_count: int = 0
@property
def has_uninterruptible(self) -> bool:
"""Return True if any UninterruptibleFrame is currently in the queue."""
return self._uninterruptible_count > 0
def _put(self, item: Any) -> None:
if isinstance(self._frame_getter(item), UninterruptibleFrame):
self._uninterruptible_count += 1
super()._put(item)
def _get(self) -> Any:
item = super()._get()
if isinstance(self._frame_getter(item), UninterruptibleFrame):
self._uninterruptible_count -= 1
return item
def reset(self) -> None:
"""Remove all non-UninterruptibleFrame items, keeping uninterruptible ones."""
kept: asyncio.Queue = asyncio.Queue()
while not self.empty():
item = self.get_nowait()
if isinstance(self._frame_getter(item), UninterruptibleFrame):
kept.put_nowait(item)
self.task_done()
while not kept.empty():
item = kept.get_nowait()
self.put_nowait(item)
kept.task_done()

4
tests/test_context_aggregators_universal.py
View File

@@ -869,7 +869,7 @@ class TestLLMAssistantAggregator(unittest.IsolatedAsyncioTestCase):
function_name="get_weather",
tool_call_id="1",
arguments={"location": "Los Angeles"},
cancel_on_interruption=False,
cancel_on_interruption=True,
),
SleepFrame(),
FunctionCallResultFrame(
@@ -901,7 +901,7 @@ class TestLLMAssistantAggregator(unittest.IsolatedAsyncioTestCase):
function_name="get_weather",
tool_call_id="1",
arguments={"location": "Los Angeles"},
cancel_on_interruption=False,
cancel_on_interruption=True,
),
SleepFrame(),
FunctionCallResultFrame(