Fix: Handle run_in_parallel=False, simplify pending function call tra… #147
Conversation
|
The latest updates on your projects. Learn more about Vercel for Git ↗︎
|
markbackman
force-pushed
the
mb/function-simplified-handle-parallel-calls
branch
from
f0490bf to
462b7d6
Compare
markbackman
force-pushed
the
mb/function-simplified-handle-parallel-calls
branch
from
462b7d6 to
3ac577b
Compare
|
@kompfner this is ready for review. |
src/pipecat_flows/manager.py
Outdated
| logger.debug( | ||
| f"{'Transition-only function called for' if is_transition_only_function else 'Function handler completed for'} {name}" | ||
| ) | ||
|
|
||
| # For edge functions, prevent LLM completion until transition (run_llm=False) | ||
| # For node functions, allow immediate completion (run_llm=True) | ||
| # Determine if this is an edge function | ||
| has_explicit_transition = bool(transition_to) or bool(transition_callback) |
There was a problem hiding this comment.
Super nit: the only reason we were storing has_explicit_transition before was to help determine how to invoke on_context_updated_edge() (though, looking back on it, I don't think it was necessary). If we don't need this var anymore, we can just get rid of it and just do:
is_edge_function = bool(next_node) or bool(transition_to) or bool(transition_callback)
Otherwise the reader might think it's significant.
There was a problem hiding this comment.
Thanks for catching this 🙇
| else: | ||
| # New style: (args, result, flow_manager) | ||
| await transition_callback(arguments, result, self) | ||
| except Exception as e: |
There was a problem hiding this comment.
Ah, looks like here you removed some duplicate logic that had been invoking result_callback with an error, right? It's logic that we already have in. transiition_func(), right?
There was a problem hiding this comment.
Right, I think this is the right call. transition_func() is responsible for the actual function call. That leaves _execute_transition to transition nodes. It's possible to have a successful function calls and a failed transition. Given that, I think this makes sense.
| @@ -614,6 +587,8 @@ async def _set_node(self, node_id: str, node_config: NodeConfig) -> None: | |||
| raise FlowTransitionError(f"{self.__class__.__name__} must be initialized first") | |||
|
|
|||
| try: | |||
| self._pending_transition = None | |||
There was a problem hiding this comment.
Why do we clear _pending_transition here?
There was a problem hiding this comment.
Asking because it seems like we already clear it in _execute_transition().
There was a problem hiding this comment.
You can still call set_node directly, which is probably less common as something that would happen outside of the context of a transition callback (or call set_node_from_config). It seems safer (maybe for now only) to reset in both places. I don't feel strongly about it, but that's why I added it. WDYT?
There was a problem hiding this comment.
Ah I see. You want it so that if somehow the developer directly invoked set_node() from outside the context of an ongoing tool-call-based transition, then it would effectively cancel it. Is my understanding right?
If so, might be worth a comment above this line to provide rationale.
There was a problem hiding this comment.
That's right. I think that makes sense, as an explicitly called function should take precedent. I'll add a comment.
tests/test_manager.py
Outdated
| ], | ||
| } | ||
|
|
||
| await flow_manager._set_node("test", node_config) |
There was a problem hiding this comment.
I didn't update all the _set_node calls in my last PR (replacing them with set_node_from_config(), but I probably ought to have...I think we should prefer to not add more private method calls.
There was a problem hiding this comment.
I updated this.
There was a problem hiding this comment.
Oh nice, you not only updated this line, but a bunch of others as well. Going the extra mile—thanks!
|
|
||
| # Check if all function calls are complete using Pipecat's state | ||
| assistant_aggregator = self._context_aggregator.assistant() | ||
| if not assistant_aggregator.has_function_calls_in_progress: |
There was a problem hiding this comment.
We need to bump the minimum pipecat-ai version requirement in pyproject.toml to pick up this functionality.
There was a problem hiding this comment.
Indeed! We should hold off on merging this PR until the pipecat-ai release is cut.
…cking
#145 but on top of the
pk/function-simplification-explorationbranch. This now relies on Pipecat to do the pending function call tracking and removes the counter and tracking logic from Flows, which is a nice simplification.DO NOT MERGE YET.
Requires an updated pipecat-ai version due to: pipecat-ai/pipecat#2006.