feat: workflows v2 #1124
feat: workflows v2 #1124
Conversation
Signed-off-by: MICHAEL DESMOND <[email protected]>
Signed-off-by: MICHAEL DESMOND <[email protected]>
Signed-off-by: MICHAEL DESMOND <[email protected]>
Signed-off-by: MICHAEL DESMOND <[email protected]>
Signed-off-by: MICHAEL DESMOND <[email protected]>
Signed-off-by: MICHAEL DESMOND <[email protected]>
Signed-off-by: MICHAEL DESMOND <[email protected]>
Signed-off-by: MICHAEL DESMOND <[email protected]>
Signed-off-by: MICHAEL DESMOND <[email protected]>
Signed-off-by: MICHAEL DESMOND <[email protected]>
Signed-off-by: MICHAEL DESMOND <[email protected]>
Signed-off-by: MICHAEL DESMOND <[email protected]>
Signed-off-by: MICHAEL DESMOND <[email protected]>
Signed-off-by: MICHAEL DESMOND <[email protected]>
Signed-off-by: MICHAEL DESMOND <[email protected]>
Signed-off-by: MICHAEL DESMOND <[email protected]>
Signed-off-by: MICHAEL DESMOND <[email protected]>
Signed-off-by: MICHAEL DESMOND <[email protected]>
Signed-off-by: MICHAEL DESMOND <[email protected]>
Signed-off-by: MICHAEL DESMOND <[email protected]>
Signed-off-by: MICHAEL DESMOND <[email protected]>
Signed-off-by: MICHAEL DESMOND <[email protected]>
Signed-off-by: MICHAEL DESMOND <[email protected]>
Signed-off-by: MICHAEL DESMOND <[email protected]>
Signed-off-by: MICHAEL DESMOND <[email protected]>
Signed-off-by: MICHAEL DESMOND <[email protected]>
| output: RunnableOutput = ( | ||
| completed_steps[-1].result | ||
| if completed_steps and isinstance(completed_steps[-1].result, RunnableOutput) | ||
| else RunnableOutput(output=[]) | ||
| ) |
There was a problem hiding this comment.
In parallel/branching graphs, this is nondeterministic. Workflows with multiple terminal leaves may not return the intended value. Is this handled?
There was a problem hiding this comment.
Using the .after operator it is possible to build a workflow with multiple terminal steps, in which case you get an empty RunnableOutput from the outer workflow instance. Its up to the programmer of the workflow to properly join to a terminal step that returns RunnableOutput.
| # Enqueue downstream | ||
| for ds_edge in step._downstream_edges: | ||
| ds = ds_edge.target | ||
| enqueue_ds = True | ||
| # Check all upstream edges to determine execution conditions | ||
| for up_edge in ds._upstream_edges: | ||
| up = up_edge.source | ||
|
|
||
| # If upstream has not executed and is its an and edge dont queue | ||
| if not up.has_executed and up_edge.type == "and": | ||
| enqueue_ds = False | ||
| break | ||
|
|
||
| # Check for conditional execution | ||
| if enqueue_ds and ds_edge.condition is not None: | ||
| if ds_edge.condition.fn: | ||
| enqueue_ds = bool(await run_callable(ds_edge.condition.fn, *results) == ds_edge.condition.key) | ||
| else: | ||
| enqueue_ds = bool(step.result == ds_edge.condition.key) | ||
|
|
||
| if enqueue_ds: | ||
| ds.has_executed = False | ||
| await queue.put(ds) | ||
|
|
There was a problem hiding this comment.
If a downstream step has multiple upstream edges marked as or, each upstream completion may enqueue the downstream, causing multiple concurrent executions and potential state races.
There was a problem hiding this comment.
Yes but the Workflow builder is designed to prevent this situation so that the workflow is predictable. I.e. parallel steps have an implicit downstream join step with and edges. And you cant call .then after .branch because you would be joining multiple upstream into a single downstream. This is the tricky part of the entire system.
|
|
||
| if tasks: | ||
| # done, _ = await asyncio.wait(tasks) | ||
| done, _ = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED) |
There was a problem hiding this comment.
I think that other tasks should be awaited or cancelled.
There was a problem hiding this comment.
If you await all of the tasks here then the system will synchronize to the current level in the queue and it wont be able to queue downstream steps that can safely be executed concurrently.
| T = TypeVar("T", bound="Workflow") | ||
|
|
||
|
|
||
| class step: # noqa: N801 |
There was a problem hiding this comment.
This could be a function instead of a class.
| def __get__(self, instance: None, owner: type[T]) -> "step": ... | ||
| @overload | ||
| def __get__(self, instance: T, owner: type[T]) -> WorkflowStep: ... | ||
|
|
||
| def __get__(self, instance: T | None, owner: type[T]) -> Any: |
There was a problem hiding this comment.
These overloads might not be needed as we are no longer using is_start etc.
There was a problem hiding this comment.
These overloads are necessary so that mypy wont get upset about the class being step but the return value of the get being a WorkflowStep.
Which issue(s) does this pull-request address?
Addresses #1025
Description
Lightweight agent orchestration using decorators.
Checklist
General
/ TypeScript
Pythonfor Python changes,TypeScriptfor TypeScript changesCode quality checks
mise check(mise fixto auto-fix)Testing
mise test:unitmise test:e2eDocumentation
mise docs:fix