diff --git a/changelog/4230.added.md b/changelog/4230.added.md new file mode 100644 index 000000000..0d7d1f124 --- /dev/null +++ b/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`. diff --git a/changelog/4230.fixed.md b/changelog/4230.fixed.md new file mode 100644 index 000000000..3fc505224 --- /dev/null +++ b/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. diff --git a/examples/function-calling/function-calling-anthropic-async-stream.py b/examples/function-calling/function-calling-anthropic-async-stream.py new file mode 100644 index 000000000..953855da0 --- /dev/null +++ b/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() diff --git a/examples/function-calling/function-calling-google-async-stream.py b/examples/function-calling/function-calling-google-async-stream.py new file mode 100644 index 000000000..0a2a1e831 --- /dev/null +++ b/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() diff --git a/examples/function-calling/function-calling-openai-async-stream.py b/examples/function-calling/function-calling-openai-async-stream.py new file mode 100644 index 000000000..5b4412489 --- /dev/null +++ b/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() diff --git a/examples/function-calling/function-calling-openai-responses-async-stream.py b/examples/function-calling/function-calling-openai-responses-async-stream.py new file mode 100644 index 000000000..0862e4813 --- /dev/null +++ b/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() diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index b73f94d26..00f38cab8 100644 --- a/src/pipecat/frames/frames.py +++ b/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 diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index 44e17faef..616636494 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/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_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, - } - ), - } - ) + if is_final: + await self._handle_function_call_finished(frame, in_progress_frame) 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}]" diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index fe9e5b90f..77a35fb29 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/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: diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index 1430aceac..03fcb115f 100644 --- a/src/pipecat/services/llm_service.py +++ b/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. - properties: Optional properties for the result. + result: The result of the function call, or an intermediate update. + 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 diff --git a/src/pipecat/utils/frame_queue.py b/src/pipecat/utils/frame_queue.py index 888bacb8e..64617f770 100644 --- a/src/pipecat/utils/frame_queue.py +++ b/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."""