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

Merge pull request #3090 from pipecat-ai/revert_function_calling_pr

Browse Source 
Reverting: Ensure that the function call results respect the previous LLM context
This commit is contained in:
Aleix Conchillo Flaqué
2025-11-18 19:40:58 -08:00
committed by GitHub
parent 62a0f0c0f5 793aca6b8b
commit f93276c64f
2 changed files with 25 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
CHANGELOG.md
View File

@@ -69,11 +69,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Fixed ### Fixed
- Fixed a race condition where, if the LLM received instructions to both produce
text and invoke a function call at the same time, the context would not be
updated before the function call result arrived, causing the bot to repeat
itself.
- Fixed an issue in the `Runner` where, when using `SmallWebRTCTransport`, the - Fixed an issue in the `Runner` where, when using `SmallWebRTCTransport`, the
`request_data` was not being passed to the `SmallWebRTCRunnerArguments` body. `request_data` was not being passed to the `SmallWebRTCRunnerArguments` body.

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

@@ -591,8 +591,6 @@ class LLMAssistantAggregator(LLMContextAggregator):
self._started = 0 self._started = 0
self._function_calls_in_progress: Dict[str, Optional[FunctionCallInProgressFrame]] = {} self._function_calls_in_progress: Dict[str, Optional[FunctionCallInProgressFrame]] = {}
self._context_updated_tasks: Set[asyncio.Task] = set() self._context_updated_tasks: Set[asyncio.Task] = set()
self._function_calls_context_messages = []
self._function_calls_pending_context_updates_callbacks = []
@property @property
def has_function_calls_in_progress(self) -> bool: def has_function_calls_in_progress(self) -> bool:
@@ -649,23 +647,21 @@ class LLMAssistantAggregator(LLMContextAggregator):
async def push_aggregation(self): async def push_aggregation(self):
"""Push the current assistant aggregation with timestamp.""" """Push the current assistant aggregation with timestamp."""
if self._aggregation: if not self._aggregation:
aggregation = self.aggregation_string() return
await self.reset()
if aggregation: aggregation = self.aggregation_string()
self._context.add_message({"role": "assistant", "content": aggregation}) await self.reset()
# Push context frame if aggregation:
await self.push_context_frame() self._context.add_message({"role": "assistant", "content": aggregation})
# Push timestamp frame with current time # Push context frame
timestamp_frame = LLMContextAssistantTimestampFrame(timestamp=time_now_iso8601()) await self.push_context_frame()
await self.push_frame(timestamp_frame)
if self._function_calls_context_messages: # Push timestamp frame with current time
self._flush_function_call_messages_to_context() timestamp_frame = LLMContextAssistantTimestampFrame(timestamp=time_now_iso8601())
await self.push_context_frame(FrameDirection.UPSTREAM) await self.push_frame(timestamp_frame)
async def _handle_llm_run(self, frame: LLMRunFrame): async def _handle_llm_run(self, frame: LLMRunFrame):
await self.push_context_frame(FrameDirection.UPSTREAM) await self.push_context_frame(FrameDirection.UPSTREAM)
@@ -685,23 +681,6 @@ class LLMAssistantAggregator(LLMContextAggregator):
self._started = 0 self._started = 0
await self.reset() await self.reset()
def _flush_function_call_messages_to_context(self):
"""Move all function calls messages into context, then clear the list."""
if self._function_calls_context_messages:
self._context.add_messages(self._function_calls_context_messages)
self._function_calls_context_messages.clear()
# Call the `on_context_updated` callbacks once the function call results
# are added to the context. Run them in separate tasks to make
# sure we don't block the pipeline.
for callback, task_name in self._function_calls_pending_context_updates_callbacks:
task = self.create_task(callback(), task_name)
self._context_updated_tasks.add(task)
task.add_done_callback(self._context_updated_task_finished)
# Clear the pending callbacks list
self._function_calls_pending_context_updates_callbacks.clear()
async def _handle_function_calls_started(self, frame: FunctionCallsStartedFrame): async def _handle_function_calls_started(self, frame: FunctionCallsStartedFrame):
function_names = [f"{f.function_name}:{f.tool_call_id}" for f in frame.function_calls] function_names = [f"{f.function_name}:{f.tool_call_id}" for f in frame.function_calls]
logger.debug(f"{self} FunctionCallsStartedFrame: {function_names}") logger.debug(f"{self} FunctionCallsStartedFrame: {function_names}")
@@ -714,7 +693,7 @@ class LLMAssistantAggregator(LLMContextAggregator):
) )
# Update context with the in-progress function call # Update context with the in-progress function call
self._function_calls_context_messages.append( self._context.add_message(
{ {
"role": "assistant", "role": "assistant",
"tool_calls": [ "tool_calls": [
@@ -729,7 +708,7 @@ class LLMAssistantAggregator(LLMContextAggregator):
], ],
} }
) )
self._function_calls_context_messages.append( self._context.add_message(
{ {
"role": "tool", "role": "tool",
"content": "IN_PROGRESS", "content": "IN_PROGRESS",
@@ -760,13 +739,6 @@ class LLMAssistantAggregator(LLMContextAggregator):
else: 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, "COMPLETED")
# Store the on_context_updated callback along with task name info to be invoked later
if properties and properties.on_context_updated:
task_name = f"{frame.function_name}:{frame.tool_call_id}:on_context_updated"
self._function_calls_pending_context_updates_callbacks.append(
(properties.on_context_updated, task_name)
)
run_llm = False run_llm = False
# Run inference if the function call result requires it. # Run inference if the function call result requires it.
@@ -781,13 +753,17 @@ class LLMAssistantAggregator(LLMContextAggregator):
# If this is the last function call in progress, run the LLM. # If this is the last function call in progress, run the LLM.
run_llm = not bool(self._function_calls_in_progress) run_llm = not bool(self._function_calls_in_progress)
# Only run if the LLM response has completed (not currently generating), if run_llm:
# otherwise defer execution until push_aggregation() is called await self.push_context_frame(FrameDirection.UPSTREAM)
# (triggered by LLMFullResponseEndFrame or interruption).
if not self._started: # Call the `on_context_updated` callback once the function call result
self._flush_function_call_messages_to_context() # is added to the context. Also, run this in a separate task to make
if run_llm: # sure we don't block the pipeline.
await self.push_context_frame(FrameDirection.UPSTREAM) if properties and properties.on_context_updated:
task_name = f"{frame.function_name}:{frame.tool_call_id}:on_context_updated"
task = self.create_task(properties.on_context_updated(), task_name)
self._context_updated_tasks.add(task)
task.add_done_callback(self._context_updated_task_finished)
async def _handle_function_call_cancel(self, frame: FunctionCallCancelFrame): async def _handle_function_call_cancel(self, frame: FunctionCallCancelFrame):
logger.debug( logger.debug(
@@ -802,12 +778,7 @@ class LLMAssistantAggregator(LLMContextAggregator):
del self._function_calls_in_progress[frame.tool_call_id] del self._function_calls_in_progress[frame.tool_call_id]
def _update_function_call_result(self, function_name: str, tool_call_id: str, result: Any): def _update_function_call_result(self, function_name: str, tool_call_id: str, result: Any):
def iter_all(): for message in self._context.get_messages():
yield from self._function_calls_context_messages
# In case on long-running function call, the function may already be added to the context
yield from self._context.get_messages()
for message in iter_all():
if ( if (
not isinstance(message, LLMSpecificMessage) not isinstance(message, LLMSpecificMessage)
and message["role"] == "tool" and message["role"] == "tool"