diff --git a/examples/foundational/14-function-calling.py b/examples/foundational/14-function-calling.py index b5aba449c..9141029ca 100644 --- a/examples/foundational/14-function-calling.py +++ b/examples/foundational/14-function-calling.py @@ -34,7 +34,12 @@ logger.add(sys.stderr, level="DEBUG") async def start_fetch_weather(function_name, llm, context): - await llm.push_frame(TextFrame("Let me check on that.")) + # note: we can't push a frame to the LLM here. the bot + # can interrupt itself and/or cause audio overlapping glitches. + # possible question for Aleix and Chad about what the right way + # to trigger speech is, now, with the new queues/async/sync refactors. + # await llm.push_frame(TextFrame("Let me check on that.")) + logger.debug(f"Starting fetch_weather_from_api with function_name: {function_name}") async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback): @@ -106,11 +111,11 @@ async def main(): pipeline = Pipeline( [ - fl_in, + # fl_in, transport.input(), context_aggregator.user(), llm, - fl_out, + # fl_out, tts, transport.output(), context_aggregator.assistant(), diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 8059b904b..f7faa8ef0 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -585,6 +585,7 @@ class FunctionCallResultFrame(DataFrame): tool_call_id: str arguments: str result: Any + run_llm: bool = True @dataclass diff --git a/src/pipecat/processors/aggregators/openai_llm_context.py b/src/pipecat/processors/aggregators/openai_llm_context.py index 83ec3e57f..4bf3f042c 100644 --- a/src/pipecat/processors/aggregators/openai_llm_context.py +++ b/src/pipecat/processors/aggregators/openai_llm_context.py @@ -133,6 +133,7 @@ class OpenAILLMContext: tool_call_id: str, arguments: str, llm: FrameProcessor, + run_llm: bool = True, ) -> None: # Push a SystemFrame downstream. This frame will let our assistant context aggregator # know that we are in the middle of a function call. Some contexts/aggregators may @@ -153,6 +154,7 @@ class OpenAILLMContext: tool_call_id=tool_call_id, arguments=arguments, result=result, + run_llm=run_llm, ) ) diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index 5eadb475b..a46ad3fab 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -110,7 +110,13 @@ class LLMService(AIService): return function_name in self._callbacks.keys() async def call_function( - self, *, context: OpenAILLMContext, tool_call_id: str, function_name: str, arguments: str + self, + *, + context: OpenAILLMContext, + tool_call_id: str, + function_name: str, + arguments: str, + run_llm: bool, ) -> None: f = None if function_name in self._callbacks.keys(): @@ -120,7 +126,12 @@ class LLMService(AIService): else: return None await context.call_function( - f, function_name=function_name, tool_call_id=tool_call_id, arguments=arguments, llm=self + f, + function_name=function_name, + tool_call_id=tool_call_id, + arguments=arguments, + llm=self, + run_llm=run_llm, ) # QUESTION FOR CB: maybe this isn't needed anymore? diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index b17dd7397..73dae4644 100644 --- a/src/pipecat/services/openai.py +++ b/src/pipecat/services/openai.py @@ -273,26 +273,21 @@ class BaseOpenAILLMService(LLMService): functions_list.append(function_name) arguments_list.append(arguments) tool_id_list.append(tool_call_id) - for function_name, arguments, tool_id in zip( - functions_list, arguments_list, tool_id_list + + total_items = len(functions_list) + for index, (function_name, arguments, tool_id) in enumerate( + zip(functions_list, arguments_list, tool_id_list), start=1 ): if self.has_function(function_name): - await self._handle_function_call(context, tool_id, function_name, arguments) - else: - raise OpenAIUnhandledFunctionException( - f"The LLM tried to call a function named '{function_name}', but there isn't a callback registered for that function." + run_llm = index == total_items + arguments = json.loads(arguments) + await self.call_function( + context=context, + function_name=function_name, + arguments=arguments, + tool_call_id=tool_id, + run_llm=run_llm, ) - # re-prompt to get a human answer after all the functions are called - await self._process_context(context) - - async def _handle_function_call(self, context, tool_call_id, function_name, arguments): - arguments = json.loads(arguments) - await self.call_function( - context=context, - tool_call_id=tool_call_id, - function_name=function_name, - arguments=arguments, - ) async def _update_settings(self, frame: LLMUpdateSettingsFrame): if frame.model is not None: @@ -486,31 +481,34 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator): def __init__(self, user_context_aggregator: OpenAIUserContextAggregator, **kwargs): super().__init__(context=user_context_aggregator._context, **kwargs) self._user_context_aggregator = user_context_aggregator - self._function_call_in_progress = None + self._function_calls_in_progress = {} self._function_call_result = None async def process_frame(self, frame, direction): await super().process_frame(frame, direction) # See note above about not calling push_frame() here. if isinstance(frame, StartInterruptionFrame): - self._function_call_in_progress = None + self._function_calls_in_progress.clear() self._function_call_finished = None + logger.debug("clearing function calls in progress") elif isinstance(frame, FunctionCallInProgressFrame): - self._function_call_in_progress = frame + self._function_calls_in_progress[frame.tool_call_id] = frame + logger.debug( + f"FunctionCallInProgressFrame: {frame.tool_call_id} {self._function_calls_in_progress}" + ) elif isinstance(frame, FunctionCallResultFrame): - if ( - self._function_call_in_progress - and self._function_call_in_progress.tool_call_id == frame.tool_call_id - ): - self._function_call_in_progress = None + logger.debug( + f"FunctionCallResultFrame: {frame.tool_call_id} {self._function_calls_in_progress}" + ) + if frame.tool_call_id in self._function_calls_in_progress: + del self._function_calls_in_progress[frame.tool_call_id] self._function_call_result = frame # TODO-CB: Kwin wants us to refactor this out of here but I REFUSE await self._push_aggregation() else: logger.warning( - "FunctionCallResultFrame tool_call_id does not match FunctionCallInProgressFrame tool_call_id" + "FunctionCallResultFrame tool_call_id does not match any function call in progress" ) - self._function_call_in_progress = None self._function_call_result = None async def _push_aggregation(self): @@ -549,7 +547,7 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator): "tool_call_id": frame.tool_call_id, } ) - run_llm = True + run_llm = frame.run_llm else: self._context.add_message({"role": "assistant", "content": aggregation})