Python: Workflow Edge Groups

[TaoChenOSU]

Motivation and Context

Addresses: #357
Related comment: #271 (comment)

Description

1. Introducing Edge groups
2. Adding the following edge groups:
   • SingleEdgeGroup
   • FanOutEdgeGroup
   • FanInEdgeGroup
   • SwitchCaseEdgeGroup
3. Adding tests and samples

Contribution Checklist

• The code builds clean without any errors or warnings
• The PR follows the Contribution Guidelines
• All unit tests pass, and I have added new tests where possible
• I didn't break anyone 😄

[Copilot]

Pull Request Overview

This pull request introduces the concept of "Edge Groups" to the workflow framework, providing a more organized and flexible way to handle connections between workflow executors. The PR implements several types of edge groups to support different connection patterns, moving away from individual edge management to group-based edge management.

• Introduces the abstract EdgeGroup base class and five concrete implementations: SingleEdgeGroup, SourceEdgeGroup, TargetEdgeGroup, ConditionalEdgeGroup, and PartitioningEdgeGroup
• Updates the workflow builder to use edge groups instead of individual edges, providing new methods for conditional and partitioning fan-out patterns
• Refactors the validation system and runner to work with edge groups while maintaining backward compatibility

Reviewed Changes

Copilot reviewed 12 out of 12 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
step_06_map_reduce.py	Cleanup of unused imports and code formatting improvements
step_00_foundation_patterns.py	New comprehensive sample demonstrating all edge group patterns with simple arithmetic operations
test_workflow_builder.py	Updated test assertion to check edge groups count instead of individual edges
test_validation.py	Updated validation tests to use SingleEdgeGroup instead of raw Edge objects
test_runner.py	Updated runner tests to use SingleEdgeGroup for consistency
test_edge.py	Added extensive test coverage for all new edge group implementations
_workflow.py	Major refactor to use edge groups, added new builder methods for conditional and partitioning patterns
_validation.py	Updated validation logic to work with edge groups while maintaining type compatibility checks
_typing_utils.py	Added handling for Any type in type checking utility
_runner.py	Refactored to work with edge groups instead of individual edges
_executor.py	Enhanced executor ID generation to include class name for better debugging
_edge.py	Major expansion adding the EdgeGroup hierarchy and all concrete implementations

[lokitoth]

Something to consider here is that keeping the low-level set of edge types the minimal needed to implement all of the conceptual edges we want to support makes the toolkit more portable acros platforms. To that end, SourceEdgeGroup and ConditionalEdgeGroup become specializations of PartitionerEdgeGroup (if you allow the partitioner to be None, and interpret that as SourceEdgeGroup, and a partition function which always returns only a single index is equivalent to a conditional edge group).

At the same time, having a richer set of types could make tooling easier, as it would allow tools to better understand the intent of an edge-group (conditional vs. partitioner). (Maybe we should support some kind of metadata through which higher-level toolkits could include information for tooling/visualization?)

[TaoChenOSU]

Something to consider here is that keeping the low-level set of edge types the minimal needed to implement all of the conceptual edges we want to support makes the toolkit more portable acros platforms. To that end, SourceEdgeGroup and ConditionalEdgeGroup become specializations of PartitionerEdgeGroup (if you allow the partitioner to be None, and interpret that as SourceEdgeGroup, and a partition function which always returns only a single index is equivalent to a conditional edge group).

At the same time, having a richer set of types could make tooling easier, as it would allow tools to better understand the intent of an edge-group (conditional vs. partitioner). (Maybe we should support some kind of metadata through which higher-level toolkits could include information for tooling/visualization?)

Good point!

1. I think the SourceEdgeGroup is indeed a specialized version of the PartitioningEdgeGroup with a None partitioner. Currently, it's the opposite. The PartitioningEdgeGroup is a specialized version of the SourceEdgeGroup. But I don't think the ConditionalEdgeGroup can be a special case of partitioning unless we have a way to restrict the partitioner function to always return a single index, which would require a different signture.
2. Are you proposing some metadata for the type of the group?
   For example,

@property
def type(self) -> str:
    return self.__class__.__name__

[lokitoth]

The reason I think ConditionalEdgeGroup is a "special case" is that you can implement it on top of PartitionerEdgeGroup. So when someone calls .add_conditional_fan_out_edges, rather than create a new type, create a partitioner function that returns the index of the first condition that passes (or the index of the "else" edge, if any).

Are you proposing some metadata for the type of the group?

No, I do not really have a concrete proposal. Just if we do what I suggest above, we want some way for tooling to be able to go back to the higher-level concept. Like the [DesignerXYZAttribute] in .NET. Probably need to be talking to the tooling teams to really understand the requirements here, if any.

[eavanvalkenburg]

I have some big disagreements here, mostly with the terminology, and I think the code can be much simpler!

[eavanvalkenburg]

Python Test Coverage Report •

File	Stmts	Miss	Cover	Missing
packages/workflow/agent_framework_workflow
_edge.py	168	9	94%	74, 86, 99, 118, 123, 128, 133, 138, 417
_executor.py	109	20	81%	62, 80, 236–240, 245, 247–250, 254–256, 258, 262, 264–265, 330
_runner.py	170	28	83%	84–86, 88, 166–168, 189–193, 197, 201–202, 223–224, 236–237, 248–250, 254, 261–263, 268–269
_typing_utils.py	19	9	52%	18, 30, 37–40, 47–48, 54
_validation.py	211	21	90%	197, 315, 318, 376, 380, 467, 469, 472–473, 475–477, 479–481, 484–488, 493
_workflow.py	181	33	81%	52, 118, 140, 190, 198–200, 212, 226, 344, 364–365, 367–369, 384–386, 394–395, 404–410, 415, 417, 489, 491, 513, 515
TOTAL	4441	1002	77%

Python Unit Test Overview

Tests	Skipped	Failures	Errors	Time
511	31 💤	0 ❌	0 🔥	8.689s ⏱️

[eavanvalkenburg]

small comment on the naming, but looking much better!