0.50.0 - GraphFactory Required Pattern

🚨 BREAKING CHANGE: graph Parameter Removed

This release introduces a breaking change to improve concurrency safety and ensure workflow isolation. The graph parameter in the Workflow and Subflow constructors has been removed and replaced with a mandatory graph_factory parameter.

Why this change was made

Previously, passing a Graph instance directly to a workflow could lead to stateful, shared graph instances across multiple workflow executions. This could cause unpredictable behavior in concurrent environments.

To ensure that every workflow execution is isolated and runs with a fresh, stateless graph, we now require a graph_factory. This is a callable (like a function) that returns a new Graph instance each time it's called. This change guarantees predictable behavior and prevents state conflicts.

How to Migrate

Updating your code is straightforward. Instead of creating your Graph instance and passing it in, you will now wrap your graph creation logic in a function and pass that function to the graph_factory parameter.

Before

# Instantiate the nodes
first_node = FirstNode()
# ... other nodes

# Create the workflow graph instance directly
workflow_graph = Graph(
    source=first_node,
    sink=final_node,
    edges=[
        # ... edges
    ]
)

# Create the workflow, passing the graph instance
sample_workflow = Workflow[SampleWorkflowState, SampleWorkflowStore](
    name="Getting Started Example Workflow",
    graph=workflow_graph, # <-- This parameter is now removed
    store_factory=lambda: SampleWorkflowStore(
        initial_state=SampleWorkflowState(items=["one", "two"])
    )
)

After

# Create a factory function that returns a new Graph instance
def create_graph() -> Graph:
    """
    Factory function to create a new instance of the sample workflow graph.
    This ensures that each workflow execution gets a fresh, isolated graph,
    preventing state conflicts in concurrent environments.
    """
    # Instantiate the nodes inside the factory
    first_node = FirstNode()
    # ... other nodes

    # Return a new Graph instance
    return Graph(
        source=first_node,
        sink=final_node,
        edges=[
            # ... edges
        ]
    )

# Create the workflow, passing the factory function
sample_workflow = Workflow[SampleWorkflowState, SampleWorkflowStore](
    name="Getting Started Example Workflow",
    graph_factory=create_graph, # <-- Use the new graph_factory parameter
    store_factory=lambda: SampleWorkflowStore(
        initial_state=SampleWorkflowState(items=["one", "two"])
    )
)

✨ New & Improved API Documentation

To support this change, we've introduced two new protocols, GraphFactory and StoreFactory, which are now fully documented in our API reference. This will provide better type hinting and clarity for developers.