Feature: Dynamic Workflow Mode Implementation for Conditional Edges

[WenPulse]

Overview

Building upon the existing LangGraph framework, I developed and implemented a complete workflow in my project. I discovered that using add_conditional_edges in the workflow had certain flaws in executing subsequent nodes. To address this issue, I temporarily applied a monkey patch in my project to fix the problems related to conditional node execution within the workflow. Additionally, I designed an Analyzer to maintain the directed graph of execution paths. This improvement allows users to enable the new execution mode by setting workflow_mode=True during compilation. If not set or set to False, the original execution mode is used by default. I have submitted a PR and hope that this approach can be adopted into LangGraph.

---

Problem Description

Prerequisite: In the workflow, each node should execute only once unless special handling logic is set.

When handling conditional branches with LangGraph, the following execution issues were discovered:

1. Set up a selector node (A) with two executable branches, Type 1 and Type 2:

   • Type 1: Execute text spell correction (Node B).
   • Type 2: Execute both text spell correction (Node B) and text syntax correction (Node C).

2. When the results of both types point to Node E, LangGraph's parallel processing mechanism causes Node E to execute only once.

import operator
from typing import Annotated, Any

from typing_extensions import TypedDict
from langgraph.graph import END, START, StateGraph

class State(TypedDict):
    aggregate: Annotated[list, operator.add]
    type: str


class ReturnNodeValue:
    def __init__(self, node_secret: str):
        self._value = node_secret

    def __call__(self, state: State) -> Any:
        print(f"Adding {self._value} to {state['aggregate']}")
        return {"aggregate": [self._value]}


builder = StateGraph(State)
builder.add_node("A", ReturnNodeValue("I'm A"))
builder.add_node("B", ReturnNodeValue("I'm B"))
builder.add_node("C", ReturnNodeValue("I'm C"))
builder.add_node("E", ReturnNodeValue("I'm E"))
intermediates = ["B", "C"]
builder.add_conditional_edges(
    "A",
    lambda state: ["B"] if state["type"] == "one" else ["B", "C"],
    path_map=intermediates,
)
builder.add_edge(START, "A")
builder.add_edge("B", "E")
builder.add_edge("C", "E")
builder.add_edge("E", END)
graph = builder.compile()
graph.invoke({"aggregate": [], "type": "two", "fanout_values": []})

Adding I'm A to []
Adding I'm B to ["I'm A"]
Adding I'm C to ["I'm A"]
Adding I'm E to ["I'm A", "I'm B", "I'm C"]

3. However, if Node D is added after Node B, making Node D execute Node E, then for Type 2, Node E is executed twice (paths A → C → E and A → B → D → E), which does not align with the expected workflow execution.

builder.add_node("A", ReturnNodeValue("I'm A"))
builder.add_node("B", ReturnNodeValue("I'm B"))
builder.add_node("C", ReturnNodeValue("I'm C"))
builder.add_node("D", ReturnNodeValue("I'm D"))
builder.add_node("E", ReturnNodeValue("I'm E"))
intermediates = ["B", "C"]
builder.add_conditional_edges(
    "A",
    lambda state: ["B"] if state["type"] == "one" else ["B", "C"],
    path_map=intermediates,
)
builder.add_edge(START, "A")
builder.add_edge("B", "D")
builder.add_edge("D", "E")
builder.add_edge("C", "E")
builder.add_edge("E", END)
graph = builder.compile()
graph.invoke({"aggregate": [], "type": "two", "fanout_values": []})

Adding I'm A to []
Adding I'm B to ["I'm A"]
Adding I'm C to ["I'm A"]
Adding I'm D to ["I'm A", "I'm B", "I'm C"]
Adding I'm E to ["I'm A", "I'm B", "I'm C"]
Adding I'm E to ["I'm A", "I'm B", "I'm C", "I'm D", "I'm E"]

Attempts in Existing LangGraph:

• If Nodes C and D are set to both succeed before executing Node E, Node E can never be triggered in Type 1.

# builder.add_edge("C", "E")
builder.add_edge(["C", "D"], "E")
builder.add_edge("E", END)
graph = builder.compile()
graph.invoke({"aggregate": [], "type": "one", "fanout_values": []})

Adding I'm A to []
Adding I'm B to ["I'm A"]
Adding I'm D to ["I'm A", "I'm B"]

• If nodes C and D are set to execute simultaneously or if node E is triggered after node C executes alone, then node E will be executed repeatedly.

builder.add_edge("C", "E")
builder.add_edge(["C", "D"], "E")
builder.add_edge("E", END)
graph = builder.compile()
graph.invoke({"aggregate": [], "type": "two", "fanout_values": []})

Adding I'm A to []
Adding I'm B to ["I'm A"]
Adding I'm C to ["I'm A"]
Adding I'm D to ["I'm A", "I'm B", "I'm C"]
Adding I'm E to ["I'm A", "I'm B", "I'm C"]
Adding I'm E to ["I'm A", "I'm B", "I'm C", "I'm D", "I'm E"]

Since each node is dynamically generated, specific branch paths cannot be determined before workflow execution. Therefore, it is necessary to dynamically choose the appropriate path during execution.

---

Solution

Added an Analyzer to LangGraph's source code to achieve the aforementioned requirements. Its main functionalities include:

1. Path Connectivity Maintenance: In the presence of conditional nodes, use a greedy algorithm to find the maximum connectivity method for executing subsequent nodes linked to the conditional node.
2. Trigger Condition Management: For example, Node E’s default trigger condition requires both Node C and Node D to execute successfully.
3. Dynamic Path Adjustment: During execution, if conditional nodes are involved, the Analyzer retrieves the actual execution path of the conditional nodes and breaks the paths that cannot be executed in the directed graph, thereby dynamically adjusting Node E’s execution trigger conditions.

Execution in this Case:

1. Type 1:
   • The selector activates the B branch, making the path A → C invalid.
   • The only valid path from the start node to E is A → B → D → E.
   • Node E’s trigger is set to D.

builder.add_conditional_edges(
    "A",
    lambda state: ["B"] if state["type"] == "one" else ["B", "C"],
    path_map=["B","C"],   
)
builder.add_edge(START, "A")
builder.add_edge("B", "D")
builder.add_edge("C", "E")
builder.add_edge("D", "E")
builder.add_edge("E", END)
graph = builder.compile(workflow_mode=True)
graph.invoke({"aggregate": [], "type": "one", "fanout_values": []})

Adding I'm A to []
Adding I'm B to ["I'm A"]
Adding I'm D to ["I'm A", "I'm B"]
Adding I'm E to ["I'm A", "I'm B", "I'm D"]

2. Type 2:
   • The selector activates both B and C branches, making all paths valid.
   • The valid paths from the start node to E include two paths: A → C → E and A → B → D → E.
   • Node E’s trigger is set to require both C and D to execute.

def route_BC_or_B(state: State) -> Literal["B", "C"]:
    if state["type"] == "one":
        return ["B"]
    return ["B", "C"]
builder.add_conditional_edges(
    "A",
    route_BC_or_B,
)
builder.add_edge(START, "A")
builder.add_edge("B", "D")
builder.add_edge("C", "E")
builder.add_edge("D", "E")
builder.add_edge("E", END)
graph = builder.compile(workflow_mode=True)
graph.invoke({"aggregate": [], "type": "two", "fanout_values": []})

Adding I'm A to []
Adding I'm B to ["I'm A"]
Adding I'm C to ["I'm A"]
Adding I'm D to ["I'm A", "I'm B", "I'm C"]
Adding I'm E to ["I'm A", "I'm B", "I'm C", "I'm D"]

Through these improvements, each node executes only once within the workflow. The trigger conditions for subsequent nodes are dynamically adjusted based on the actual execution paths, catering to different types of text processing requirements.

---

Usage

To use the improved execution mode in LangGraph, set workflow_mode=True in compile. The specific steps are as follows:

1. Set the Option:

   Add workflow_mode=True in the compilation configuration to enable the dynamic workflow mode.

   from langgraph.graph import StateGraph
   
   builder = StateGraph(State)
   graph = builder.compile(workflow_mode=True)

   Note: When workflow_mode=True, either the path_map parameter or a Literal return is required. Only one of them is needed. The path_map or Literal must cover all possible execution paths.

2. Default Behavior:

   If not set or if workflow_mode is set to False, the original execution mode of LangGraph is used, maintaining backward compatibility.

---

Here is my PR link

[WenPulse]

I have also attempted to use Command and goto to solve this issue, but it has been nearly impractical. This is because, in the workflow projects I create, all nodes are manually added and connected by the client. When connecting nodes, I provide the add_edges method for regular nodes and the add_conditional_edges method for conditional nodes.

I am unable to set Command for different nodes based on the final connections and operational state of the graph generated by the client. The only solution is to progressively disconnect unreachable nodes during execution to ensure the correct execution of subsequent nodes following conditional edges.

I believe the root cause of this problem is that, in a real workflow, when dealing with the subsequent nodes of conditional nodes, it is not feasible to execute based solely on simple "AND" or "OR" conditions such as "all preceding nodes are satisfied" or "any one of the preceding conditions is met."

[WenPulse]

If you think the previous example is somewhat extreme, consider this example: when a node receives inputs from both conditional branches and regular nodes, according to the original LangGraph, this generally causes the node to execute twice.

import operator
from typing import Annotated, Literal, Any

from typing_extensions import TypedDict
from langgraph.graph import END, START, StateGraph

def reduce_fanouts(left, right):
    if left is None:
        left = []
    if not right:
        # Overwrite
        return []
    return left + right


class State(TypedDict):
    # The operator.add reducer fn makes this append-only
    aggregate: Annotated[list, operator.add]
    which: str


class ReturnNodeValue:
    def __init__(self, node_secret: str):
        self._value = node_secret

    def __call__(self, state: State) -> Any:
        print(f"Adding {self._value} to {state['aggregate']}")
        return {"aggregate": [self._value]}


builder = StateGraph(State)
builder.add_node("a", ReturnNodeValue("I'm A"))
builder.add_edge(START, "a")
builder.add_node("b", ReturnNodeValue("I'm B"))
builder.add_node("c", ReturnNodeValue("I'm C"))
builder.add_node("e", ReturnNodeValue("I'm E"))


def route_b_or_e(state: State) -> Literal["b", "e"]:
    if state["which"] == "b":
        return ["b"]
    return ["e"]


intermediates = ["b", "e"]
builder.add_conditional_edges(
    "a",
    route_b_or_e,
)
builder.add_edge(START, "a")
builder.add_edge(START, "c")
builder.add_edge("c", "e")
builder.add_edge("b","e")
builder.add_edge("e", END)
graph = builder.compile()
graph.invoke({"aggregate": [], "which": "b", "fanout_values": []})

originl LangGraph

Adding I'm A to []
Adding I'm C to []
Adding I'm B to ["I'm A", "I'm C"]
Adding I'm E to ["I'm A", "I'm C"]
Adding I'm E to ["I'm A", "I'm C", "I'm B", "I'm E"]

use workflow mode

Adding I'm A to []
Adding I'm C to []
Adding I'm B to ["I'm A", "I'm C"]
Adding I'm E to ["I'm A", "I'm C", "I'm B"]

[WenPulse]

@nfcampos I sincerely apologize for disturbing you amidst your busy schedule. Could you please take some time to review my idea and the associated code to determine if they are reasonable? I will actively respond based on your feedback ！

[nfcampos]

Hi, I commented on the PR too, this would change langgraph into a DAG framework, we started langgraph precisely to support graphs either cycles that are needed for LLM applications

[WenPulse]

Thank you for your comments. I have provided additional reasons for this change in the PR. If you still disagree with this modification approach, you may choose to close the PR or further elaborate your concerns. I will explore alternative methods to address this issue. Regardless of whether the PR is merged, I greatly appreciate your response and support.