Repository: apache/burr

Issue:
Feat: Build a new Graph as a combination of Graphs
**Is your feature request related to a problem? Please describe.**
I am creating specific Graphs at runtime as the combination of several Graphs depending on a configuration.

**Describe the solution you'd like**
I would need to define a Graph as a merge of 2 or more Graphs with additional transitions, or to be able to get both actions and transitions in the Graph API so that I can extract them from my "sub" graphs and can re-combine them while building a new Graph.

```python
graph_1 = (
    GraphBuilder()
    .with_actions(**actions_1)
    .with_transitions(*transitions_1)
    .build()
)

graph_2 = (
    GraphBuilder()
    .with_actions(**actions_2)
    .with_transitions(*transitions_2)
    .build()
)

final_graph = (
    GraphBuilder()
    .with_graphs(graph_1, graph_2)
    .with_transitions(*additional_transitions)
    .build()
```

**Describe alternatives you've considered**
I could consider passing my 'sub' graphs as a set of actions and transitions instead of Graph, but I find this solution not very elegant and will add complexity to my code as I also might need to use the 'sub' graphs in standalone burr Applications.

Thanks for your feedback,
KR
jbd

Hints from issue comments:
@skrawcz: @jbdebard thanks for the issue. I need to double check the code path whether we handle this in the API style you want, but we have a [RunnableGraph concept](https://burr.dagworks.io/reference/parallelism/#burr.core.parallelism.RunnableGraph) that we created for parallelism that is conceptually what you describe. One wrinkle is that to link graphs we need to specify the start, end, and possibly some state stuff -- did you have any thoughts there?

How much of a blocker is this? Right now the way around it is (which is what I think you're referencing as what you're currently doing?) that you stick those two graphs into two actions and wire it like normal, with the actions then creating the right burr subgraph to invoke - [see this part of the docs](https://burr.dagworks.io/concepts/recursion/). 

E.g. I think this might be a reasonable use case of the [class based action](https://burr.dagworks.io/concepts/actions/#class-based-actions) approach
```python
graph_1 = (
    GraphBuilder()
    .with_actions(**actions_1)
    .with_transitions(*transitions_1)
    .build()
)

graph_2 = (
    GraphBuilder()
    .with_actions(**actions_2)
    .with_transitions(*transitions_2)
    .build()
)


class SubGraphAction(Action):
    
    def __init__(self, sub_graph, reads, writes):
       # constructor
       self.sub_graph = sub_graph
       ...

    @property
    def reads(self) -> list[str]:
        return self.reads

    def run(self, state: State) -> dict:
        # create application given subgraph
        app = ApplicationBuilder().with_graph(self.sub_graph)... # etc
        stuff = app.run(...)
        return {"some_result": "to_save_to_state", ...}

    @property
    def writes(self) -> list[str]:
        return self.writes

    def update(self, result: dict, state: State) -> State:
        return state.update(**result) 


final_graph = (
    GraphBuilder()
    .with_graphs(SubGraphAction(graph1, ...).with_name("graph1"), SubGraphAction(graph_2, ...).with_name("graph2"))
    .with_transitions(*additional_transitions)
    .build()
```
Note: I haven't tested the above code ^ . @elijahbenizzy any thoughts on either approach?
@elijahbenizzy: So yeah, it depends what you want to do. If its just the union of the two graphs, then we should be able to pretty easily allow you to add all the nodes from a graph... But that's not supported. Would be curious to get more of a sense of your use-case if you don't mind sharing?

Otherwise if you want to encapsulate them within an action what @skrawcz recommended is the right approach (and we can always add tooling to make that easier -- it is currently tailored towards parallelism through recursion, but it shouldn't necessarily be coupled to that...).

I think both are feasible, but my guess is you want the union approach from your wording. 

Currently, we have `with_graph` which just adds a prebuilt graph and doesn't work with actions/transitions, but I don't see any reason it couldn't be additive. E.G. we couldn't do what you said, or just have multiple `with_graph` calls. Might have some time to hack it out to ensure that it's feasible -- will report back. Otherwise you can probably implement what you want with a bit of hacking -- have not tested this out (some nuances of how we do actions/assigning names could break it, but in theory it should work...).

Note this assumes that action names are not shared between actions.

Let us know what the preferred approach is and we're happy to scope it out/see if we can hack it in.

```python
graph_1 = (
    GraphBuilder()
    .with_actions(**actions_1)
    .with_transitions(*transitions_1)
    .build()
)

graph_2 = (
    GraphBuilder()
    .with_actions(**actions_2)
    .with_transitions(*transitions_2)
    .build()
)

final_graph = (
    GraphBuilder()
    .with_actions(*[action for action in graph_1.actions + graph_2.actions]) # must be unique
    .with_transitions(*[(transition.from_.name, transition.to.name, transition.condition) for transition in graph_1.transitions + graph_2.transitions])
    .with_transitions(*additional_transitions)
    .build()
```@elijahbenizzy: OK had some fun with this -- mind trying out `pip install burr==0.40.0rc0`? Has the exact API you suggested.

See docs [here](https://burr.dagworks.io/pull/527/reference/application/#burr.core.application.ApplicationBuilder.with_graph)

Additionally I think we'll want to enable @skrawcz's API> this is a limitation of the parallelism design that is coupled to recursion -- shouldn't be too tough to tease apart.

@skrawcz: @jbdebard ^ see the RC version.@jbdebard: Hello @skrawcz and @elijahbenizzy. Thank you very much, I tested `burr==0.40.0rc0` and it works just as expected for me (I used a workaround so far, but will refactor asap).

As for my use case actually, I'm building a pipeline factory that plugs different kinds of models with different kinds of preprocessing at runtime depending on an REST API request. The preprocessing tasks and the models are both `burr.core.Graph` (sometimes just a wrapper around a function, but some other times really needing Graph structure).

The preprocessing part should be pure DAG and I think maybe I should have used Hamilton instead, but I had only 2 month to get it in prod and no time to learn both libraries.

Anyway, a big thank to both of you for being so reactive 👌

Solution patch:
diff --git a/burr/core/application.py b/burr/core/application.py
index 66dc465..a4c15a5 100644
--- a/burr/core/application.py
+++ b/burr/core/application.py
@@ -2065,7 +2065,6 @@ class ApplicationBuilder(Generic[StateType]):
         self.loaded_from_fork: bool = False
         self.tracker = None
         self.graph_builder = None
-        self.prebuilt_graph = None
         self.typing_system = None
         self.parallel_executor_factory = None
         self.state_persister = None
@@ -2143,9 +2142,21 @@ class ApplicationBuilder(Generic[StateType]):
             self.state = State(kwargs)
         return self
 
+    def with_graphs(self, *graphs) -> "ApplicationBuilder[StateType]":
+        """Adds multiple prebuilt graphs -- this just calls :py:meth:`with_graph <burr.core.application.ApplicationBuilder.with_graph>`
+        in a loop! See caveats in :py:meth:`with_graph <burr.core.application.ApplicationBuilder.with_graph>`.
+
+        :param graphs: Graphs to add to the application
+        :return: The application builder for future chaining.
+        """
+        for graph in graphs:
+            self.with_graph(graph)
+        return self
+
     def with_graph(self, graph: Graph) -> "ApplicationBuilder[StateType]":
-        """Adds a prebuilt graph -- this is an alternative to using the with_actions and with_transitions methods.
-        While you will likely use with_actions and with_transitions, you may want this in a few cases:
+        """Adds a prebuilt graph -- this can work in addition to using with_actions and with_transitions methods.
+        This will add all nodes + edges from a prebuilt graph to the current graph. Note that if you add two
+        graphs (or a combination of graphs/nodes/edges), you will need to ensure that there are no node name conflicts.
 
         1. You want to reuse the same graph object for different applications
         2. You want the logic that constructs the graph to be separate from that which constructs the application
@@ -2154,13 +2165,8 @@ class ApplicationBuilder(Generic[StateType]):
         :param graph: Graph object built with the :py:class:`GraphBuilder <burr.core.graph.GraphBuilder>`
         :return: The application builder for future chaining.
         """
-        if self.graph_builder is not None:
-            raise ValueError(
-                BASE_ERROR_MESSAGE
-                + "You have already called `with_actions`, or `with_transitions` -- you currently "
-                "cannot use the with_graph method along with that. Use `with_graph` or the other methods, not both"
-            )
-        self.prebuilt_graph = graph
+        self._initialize_graph_builder()
+        self.graph_builder = self.graph_builder.with_graph(graph)
         return self
 
     def with_parallel_executor(self, executor_factory: lambda: Executor):
@@ -2190,15 +2196,6 @@ class ApplicationBuilder(Generic[StateType]):
         self.parallel_executor_factory = executor_factory
         return self
 
-    def _ensure_no_prebuilt_graph(self):
-        if self.prebuilt_graph is not None:
-            raise ValueError(
-                BASE_ERROR_MESSAGE + "You have already called `with_graph` -- you currently "
-                "cannot use the with_actions, or with_transitions method along with that. "
-                "Use `with_graph` or the other methods, not both."
-            )
-        return self
-
     def _initialize_graph_builder(self):
         if self.graph_builder is None:
             self.graph_builder = GraphBuilder()
@@ -2233,7 +2230,6 @@ class ApplicationBuilder(Generic[StateType]):
         :param action_dict: Actions to add, keyed by name
         :return: The application builder for future chaining.
         """
-        self._ensure_no_prebuilt_graph()
         self._initialize_graph_builder()
         self.graph_builder = self.graph_builder.with_actions(*action_list, **action_dict)
         return self
@@ -2256,7 +2252,6 @@ class ApplicationBuilder(Generic[StateType]):
         :param transitions: Transitions to add
         :return: The application builder for future chaining.
         """
-        self._ensure_no_prebuilt_graph()
         self._initialize_graph_builder()
         self.graph_builder = self.graph_builder.with_transitions(*transitions)
         return self
@@ -2583,15 +2578,13 @@ class ApplicationBuilder(Generic[StateType]):
         self.state = self.state.wipe(delete=[PRIOR_STEP])
 
     def _get_built_graph(self) -> Graph:
-        if self.graph_builder is None and self.prebuilt_graph is None:
+        if self.graph_builder is None:
             raise ValueError(
                 BASE_ERROR_MESSAGE
-                + "You must set the graph using with_graph, or use with_entrypoint, with_actions, and with_transitions"
-                " to build the graph."
+                + "No graph constructs exist. You must call some combination of with_graph, with_entrypoint, "
+                "with_actions, and with_transitions"
             )
-        if self.graph_builder is not None:
-            return self.graph_builder.build()
-        return self.prebuilt_graph
+        return self.graph_builder.build()
 
     def _build_common(self) -> Application:
         graph = self._get_built_graph()
diff --git a/burr/core/graph.py b/burr/core/graph.py
index 6b1ebba..a3a227d 100644
--- a/burr/core/graph.py
+++ b/burr/core/graph.py
@@ -26,6 +26,16 @@ def _validate_actions(actions: Optional[List[Action]]):
     assert_set(actions, "_actions", "with_actions")
     if len(actions) == 0:
         raise ValueError("Must have at least one action in the application!")
+    seen_action_names = set()
+    for action in actions:
+        if action.name in seen_action_names:
+            raise ValueError(
+                f"Action: {action.name} is duplicated in the actions list. "
+                "actions have unique names. This could happen"
+                "if you add two actions with the same name or add a graph that"
+                "has actions with the same name as any that already exist."
+            )
+        seen_action_names.add(action.name)
 
 
 def _validate_transitions(
@@ -321,6 +331,25 @@ class GraphBuilder:
                 self.transitions.append((action, to_, condition))
         return self
 
+    def with_graph(self, graph: Graph) -> "GraphBuilder":
+        """Adds an existing graph to the builder. Note that if you have any name clashes
+        this will error out. This would happen if you add actions with the same name as actions
+        that already exist.
+
+        :param graph: The graph to add
+        :return: The application builder for future chaining.
+        """
+        if self.actions is None:
+            self.actions = []
+        if self.transitions is None:
+            self.transitions = []
+        self.actions.extend(graph.actions)
+        self.transitions.extend(
+            (transition.from_.name, transition.to.name, transition.condition)
+            for transition in graph.transitions
+        )
+        return self
+
     def build(self) -> Graph:
         """Builds/finalizes the graph.
 


Test patch:
diff --git a/tests/core/test_graph.py b/tests/core/test_graph.py
index c2013ee..1dbba42 100644
--- a/tests/core/test_graph.py
+++ b/tests/core/test_graph.py
@@ -87,6 +87,11 @@ def test__validate_actions_empty():
         _validate_actions([])
 
 
+def test__validate_actions_duplicated():
+    with pytest.raises(ValueError, match="duplicated"):
+        _validate_actions([Result("test"), Result("test")])
+
+
 base_counter_action = PassedInAction(
     reads=["count"],
     writes=["count"],
@@ -110,6 +115,33 @@ def test_graph_builder_builds():
     assert len(graph.transitions) == 2
 
 
+def test_graph_builder_with_graph():
+    graph1 = (
+        GraphBuilder()
+        .with_actions(counter=base_counter_action)
+        .with_transitions(("counter", "counter", Condition.expr("count < 10")))
+        .build()
+    )
+    graph2 = (
+        GraphBuilder()
+        .with_actions(counter2=base_counter_action)
+        .with_transitions(("counter2", "counter2", Condition.expr("count < 20")))
+        .build()
+    )
+    graph = (
+        GraphBuilder()
+        .with_graph(graph1)
+        .with_graph(graph2)
+        .with_actions(result=Result("count"))
+        .with_transitions(
+            ("counter", "counter2"),
+            ("counter2", "result"),
+        )
+    )
+    assert len(graph.actions) == 3
+    assert len(graph.transitions) == 4
+
+
 def test_graph_builder_get_next_node():
     graph = (
         GraphBuilder()