From c9f922c4798c6f1ba4857e0cca9c6dc8ae7d6923 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Wed, 28 Jan 2026 15:38:40 -0300 Subject: [PATCH 1/4] Removed an overridden method that was identical to the parent implementation. --- src/pipecat/services/google/llm.py | 17 ----------------- 1 file changed, 17 deletions(-) diff --git a/src/pipecat/services/google/llm.py b/src/pipecat/services/google/llm.py index adb2664bb..fd6c2c54e 100644 --- a/src/pipecat/services/google/llm.py +++ b/src/pipecat/services/google/llm.py @@ -40,7 +40,6 @@ from pipecat.frames.frames import ( LLMThoughtStartFrame, LLMThoughtTextFrame, LLMUpdateSettingsFrame, - UserImageRawFrame, ) from pipecat.metrics.metrics import LLMTokenUsage from pipecat.processors.aggregators.llm_context import LLMContext @@ -199,22 +198,6 @@ class GoogleAssistantContextAggregator(OpenAIAssistantContextAggregator): if part.function_response and part.function_response.id == tool_call_id: part.function_response.response = {"value": json.dumps(result)} - async def handle_user_image_frame(self, frame: UserImageRawFrame): - """Handle user image frame. - - Args: - frame: Frame containing user image data and request context. - """ - await self._update_function_call_result( - frame.request.function_name, frame.request.tool_call_id, "COMPLETED" - ) - self._context.add_image_frame_message( - format=frame.format, - size=frame.size, - image=frame.image, - text=frame.request.context, - ) - @dataclass class GoogleContextAggregatorPair: From d3f4cbb6201bf980b93e12df32ae69f9a0984ed7 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Wed, 28 Jan 2026 15:39:06 -0300 Subject: [PATCH 2/4] Providing a way to defer the function call results. --- src/pipecat/frames/frames.py | 2 ++ .../processors/aggregators/llm_response.py | 5 ++++ .../aggregators/llm_response_universal.py | 10 ++++--- src/pipecat/services/llm_service.py | 28 ++++++++++++++----- 4 files changed, 34 insertions(+), 11 deletions(-) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 26f9af9d2..8df97d313 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -1466,6 +1466,7 @@ class UserImageRequestFrame(SystemFrame): video_source: Specific video source to capture from. function_name: Name of function that generated this request (if any). tool_call_id: Tool call ID if generated by function call (if any). + result_callback: Optional callback to invoke when the image is retrieved. context: [DEPRECATED] Optional context for the image request. """ @@ -1475,6 +1476,7 @@ class UserImageRequestFrame(SystemFrame): video_source: Optional[str] = None function_name: Optional[str] = None tool_call_id: Optional[str] = None + result_callback: Optional[Any] = None context: Optional[Any] = None def __post_init__(self): diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index 1b53e9934..e11ededbf 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -1042,6 +1042,11 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): del self._function_calls_in_progress[frame.request.tool_call_id] + # Call the result_callback if provided. This signals that the image + # has been retrieved and the function call can now complete. + if frame.request and frame.request.result_callback: + await frame.request.result_callback(None) + await self.handle_user_image_frame(frame) await self.push_aggregation() await self.push_context_frame(FrameDirection.UPSTREAM) diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index 3a8bbb542..200010591 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -941,16 +941,18 @@ class LLMAssistantAggregator(LLMContextAggregator): async def _handle_user_image_frame(self, frame: UserImageRawFrame): image_appended = False - # Check if this image is a result of a function call if so, let's cache. - # TODO(aleix): The function call might have already been executed - # because FunctionCallResultFrame was just faster, in that case we just - # push the context frame now. + # Check if this image is a result of a function call. if ( frame.request and frame.request.tool_call_id and frame.request.tool_call_id in self._function_calls_in_progress ): self._function_calls_image_results[frame.request.tool_call_id] = frame + + # Call the result_callback if provided. This signals that the image + # has been retrieved and the function call can now complete. + if frame.request.result_callback: + await frame.request.result_callback(None) else: image_appended = await self._maybe_append_image_to_context(frame) diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index 0368c8feb..035f1b2f2 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -170,17 +170,22 @@ class LLMService(AIService): # However, subclasses should override this with a more specific adapter when necessary. adapter_class: Type[BaseLLMAdapter] = OpenAILLMAdapter - def __init__(self, run_in_parallel: bool = True, **kwargs): + def __init__( + self, run_in_parallel: bool = True, function_call_timeout_secs: float = 10.0, **kwargs + ): """Initialize the LLM service. Args: run_in_parallel: Whether to run function calls in parallel or sequentially. Defaults to True. + function_call_timeout_secs: Timeout in seconds for deferred function calls. + Defaults to 10.0 seconds. **kwargs: Additional arguments passed to the parent AIService. """ super().__init__(**kwargs) self._run_in_parallel = run_in_parallel + self._function_call_timeout_secs = function_call_timeout_secs self._start_callbacks = {} self._adapter = self.adapter_class() self._functions: Dict[Optional[str], FunctionCallRegistryItem] = {} @@ -596,14 +601,26 @@ class LLMService(AIService): cancel_on_interruption=item.cancel_on_interruption, ) - callback_executed = False + # Start a timeout task for deferred function calls + async def timeout_handler(): + await asyncio.sleep(self._function_call_timeout_secs) + logger.warning( + f"{self} Function call [{runner_item.function_name}:{runner_item.tool_call_id}] timed out after {self._function_call_timeout_secs} seconds" + ) + await function_call_result_callback(None) + + timeout_task = self.create_task(timeout_handler()) # Define a callback function that pushes a FunctionCallResultFrame upstream & downstream. async def function_call_result_callback( result: Any, *, properties: Optional[FunctionCallResultProperties] = None ): - nonlocal callback_executed - callback_executed = True + nonlocal timeout_task + + # Cancel timeout task if it exists + if timeout_task and not timeout_task.done(): + await self.cancel_task(timeout_task) + await self.broadcast_frame( FunctionCallResultFrame, function_name=runner_item.function_name, @@ -653,9 +670,6 @@ class LLMService(AIService): error_message = f"Error executing function call [{runner_item.function_name}]: {e}" logger.error(f"{self} {error_message}") await self.push_error(error_msg=error_message, exception=e, fatal=False) - finally: - if not callback_executed: - await function_call_result_callback(None) async def _cancel_function_call(self, function_name: Optional[str]): cancelled_tasks = set() From c8496dfb8e6a3b4c96624a36db9cb7df6f46ad49 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Wed, 28 Jan 2026 15:39:21 -0300 Subject: [PATCH 3/4] Updated the examples which use UserImageRequestFrame to defer the function call result. --- .../14d-function-calling-anthropic-video.py | 13 +++++-------- .../foundational/14d-function-calling-aws-video.py | 13 +++++-------- .../14d-function-calling-gemini-flash-video.py | 13 +++++-------- .../14d-function-calling-moondream-video.py | 13 +++++-------- .../14d-function-calling-openai-video.py | 13 +++++-------- .../foundational/14e-function-calling-google.py | 13 +++++-------- .../foundational/20d-persistent-context-gemini.py | 10 +++------- 7 files changed, 33 insertions(+), 55 deletions(-) diff --git a/examples/foundational/14d-function-calling-anthropic-video.py b/examples/foundational/14d-function-calling-anthropic-video.py index 8097ee988..c9dacaa75 100644 --- a/examples/foundational/14d-function-calling-anthropic-video.py +++ b/examples/foundational/14d-function-calling-anthropic-video.py @@ -48,14 +48,16 @@ async def fetch_user_image(params: FunctionCallParams): 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. + 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. + # 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, @@ -63,16 +65,11 @@ async def fetch_user_image(params: FunctionCallParams): append_to_context=True, function_name=params.function_name, tool_call_id=params.tool_call_id, + result_callback=params.result_callback, ), FrameDirection.UPSTREAM, ) - await params.result_callback(None) - - # Instead of None, it's possible to also provide a tool call answer to - # tell the LLM that we are grabbing the image to analyze. - # await params.result_callback({"result": "Image is being captured."}) - # We store functions so objects (e.g. SileroVADAnalyzer) don't get # instantiated. The function will be called when the desired transport gets diff --git a/examples/foundational/14d-function-calling-aws-video.py b/examples/foundational/14d-function-calling-aws-video.py index b972f4097..9faa925aa 100644 --- a/examples/foundational/14d-function-calling-aws-video.py +++ b/examples/foundational/14d-function-calling-aws-video.py @@ -48,14 +48,16 @@ async def fetch_user_image(params: FunctionCallParams): 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. + 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. + # 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, @@ -63,16 +65,11 @@ async def fetch_user_image(params: FunctionCallParams): append_to_context=True, function_name=params.function_name, tool_call_id=params.tool_call_id, + result_callback=params.result_callback, ), FrameDirection.UPSTREAM, ) - await params.result_callback(None) - - # Instead of None, it's possible to also provide a tool call answer to - # tell the LLM that we are grabbing the image to analyze. - # await params.result_callback({"result": "Image is being captured."}) - # We store functions so objects (e.g. SileroVADAnalyzer) don't get # instantiated. The function will be called when the desired transport gets diff --git a/examples/foundational/14d-function-calling-gemini-flash-video.py b/examples/foundational/14d-function-calling-gemini-flash-video.py index ba9dcd266..857eb92b6 100644 --- a/examples/foundational/14d-function-calling-gemini-flash-video.py +++ b/examples/foundational/14d-function-calling-gemini-flash-video.py @@ -48,14 +48,16 @@ async def fetch_user_image(params: FunctionCallParams): 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. + 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. + # 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, @@ -63,16 +65,11 @@ async def fetch_user_image(params: FunctionCallParams): append_to_context=True, function_name=params.function_name, tool_call_id=params.tool_call_id, + result_callback=params.result_callback, ), FrameDirection.UPSTREAM, ) - await params.result_callback(None) - - # Instead of None, it's possible to also provide a tool call answer to - # tell the LLM that we are grabbing the image to analyze. - # await params.result_callback({"result": "Image is being captured."}) - # We store functions so objects (e.g. SileroVADAnalyzer) don't get # instantiated. The function will be called when the desired transport gets diff --git a/examples/foundational/14d-function-calling-moondream-video.py b/examples/foundational/14d-function-calling-moondream-video.py index 0b10854ea..3e64d5bd8 100644 --- a/examples/foundational/14d-function-calling-moondream-video.py +++ b/examples/foundational/14d-function-calling-moondream-video.py @@ -57,7 +57,8 @@ async def fetch_user_image(params: FunctionCallParams): 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. + UserImageRawFrame downstream. The result_callback will be invoked once the + image is retrieved and processed. """ user_id = params.arguments["user_id"] question = params.arguments["question"] @@ -65,7 +66,8 @@ async def fetch_user_image(params: FunctionCallParams): # Request a user image frame. In this case, we don't want the requested # image to be added to the context because we will process it with - # Moondream. Also associate it to the function call. + # Moondream. 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, @@ -73,16 +75,11 @@ async def fetch_user_image(params: FunctionCallParams): append_to_context=False, function_name=params.function_name, tool_call_id=params.tool_call_id, + result_callback=params.result_callback, ), FrameDirection.UPSTREAM, ) - await params.result_callback(None) - - # Instead of None, it's possible to also provide a tool call answer to - # tell the LLM that we are grabbing the image to analyze. - # await params.result_callback({"result": "Image is being captured."}) - class MoondreamTextFrameWrapper(FrameProcessor): """Wraps Moondream-provided TextFrames with LLM response start/end frames. diff --git a/examples/foundational/14d-function-calling-openai-video.py b/examples/foundational/14d-function-calling-openai-video.py index 3afebe7bb..ec201cc3a 100644 --- a/examples/foundational/14d-function-calling-openai-video.py +++ b/examples/foundational/14d-function-calling-openai-video.py @@ -49,14 +49,16 @@ async def fetch_user_image(params: FunctionCallParams): 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. + 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. + # 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, @@ -64,16 +66,11 @@ async def fetch_user_image(params: FunctionCallParams): append_to_context=True, function_name=params.function_name, tool_call_id=params.tool_call_id, + result_callback=params.result_callback, ), FrameDirection.UPSTREAM, ) - await params.result_callback(None) - - # Instead of None, it's possible to also provide a tool call answer to - # tell the LLM that we are grabbing the image to analyze. - # await params.result_callback({"result": "Image is being captured."}) - # We store functions so objects (e.g. SileroVADAnalyzer) don't get # instantiated. The function will be called when the desired transport gets diff --git a/examples/foundational/14e-function-calling-google.py b/examples/foundational/14e-function-calling-google.py index 4e7abf997..56cdd53c1 100644 --- a/examples/foundational/14e-function-calling-google.py +++ b/examples/foundational/14e-function-calling-google.py @@ -58,14 +58,16 @@ async def get_image(params: FunctionCallParams): 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. + 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. + # 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, @@ -73,16 +75,11 @@ async def get_image(params: FunctionCallParams): append_to_context=True, function_name=params.function_name, tool_call_id=params.tool_call_id, + result_callback=params.result_callback, ), FrameDirection.UPSTREAM, ) - await params.result_callback(None) - - # Instead of None, it's possible to also provide a tool call answer to - # tell the LLM that we are grabbing the image to analyze. - # await params.result_callback({"result": "Image is being captured."}) - # We store functions so objects (e.g. SileroVADAnalyzer) don't get # instantiated. The function will be called when the desired transport gets diff --git a/examples/foundational/20d-persistent-context-gemini.py b/examples/foundational/20d-persistent-context-gemini.py index e0e2949a3..4b434c4c1 100644 --- a/examples/foundational/20d-persistent-context-gemini.py +++ b/examples/foundational/20d-persistent-context-gemini.py @@ -66,7 +66,8 @@ async def get_image(params: FunctionCallParams): 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. + # 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, @@ -74,16 +75,11 @@ async def get_image(params: FunctionCallParams): append_to_context=True, function_name=params.function_name, tool_call_id=params.tool_call_id, + result_callback=params.result_callback, ), FrameDirection.UPSTREAM, ) - await params.result_callback(None) - - # Instead of None, it's possible to also provide a tool call answer to - # tell the LLM that we are grabbing the image to analyze. - # await params.result_callback({"result": "Image is being captured."}) - async def get_saved_conversation_filenames(params: FunctionCallParams): # Construct the full pattern including the BASE_FILENAME From 5543bc56f378b85df463418339ddf96b041de70c Mon Sep 17 00:00:00 2001 From: filipi87 Date: Wed, 28 Jan 2026 15:43:59 -0300 Subject: [PATCH 4/4] Add changelog files for PR #3571 Co-Authored-By: Claude Sonnet 4.5 --- changelog/3571.added.2.md | 1 + changelog/3571.added.md | 1 + changelog/3571.changed.md | 4 ++++ 3 files changed, 6 insertions(+) create mode 100644 changelog/3571.added.2.md create mode 100644 changelog/3571.added.md create mode 100644 changelog/3571.changed.md diff --git a/changelog/3571.added.2.md b/changelog/3571.added.2.md new file mode 100644 index 000000000..f9af8dde0 --- /dev/null +++ b/changelog/3571.added.2.md @@ -0,0 +1 @@ +- Added `function_call_timeout_secs` parameter to `LLMService` to configure timeout for deferred function calls (defaults to 10.0 seconds). diff --git a/changelog/3571.added.md b/changelog/3571.added.md new file mode 100644 index 000000000..927bbcbc4 --- /dev/null +++ b/changelog/3571.added.md @@ -0,0 +1 @@ +- Added `result_callback` parameter to `UserImageRequestFrame` to support deferred function call results. diff --git a/changelog/3571.changed.md b/changelog/3571.changed.md new file mode 100644 index 000000000..eb44cae0a --- /dev/null +++ b/changelog/3571.changed.md @@ -0,0 +1,4 @@ +- ⚠️ Changed function call handling to use timeout-based completion instead of immediate callback execution. + - Function calls that defer their results (e.g., `UserImageRequestFrame`) now use a timeout mechanism + - The `result_callback` is invoked automatically when the deferred operation completes or after timeout + - This change affects examples using `UserImageRequestFrame` - the `result_callback` should now be passed to the frame instead of being called immediately