langchain-ai/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-07-15 12:35:43 -04:00
Code Issues 314 Packages Projects Releases 17 Wiki Activity

[PR #3345] feature:Dynamic Workflow Mode Implementation for Conditional Edges #3229

New Issue
Closed
opened 2026-02-20 17:48:16 -05:00 by yindo · 0 comments
yindo commented 2026-02-20 17:48:16 -05:00
Owner
Copy Link
Copy Source

Original Pull Request: https://github.com/langchain-ai/langgraph/pull/3345

State: closed
Merged: No

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. By applying monkey patch, I temporarily fixed the issues related to conditional node execution within the workflow and 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 hope this approach can be adopted.

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": []})

pic1

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"]
  1. 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": []})

pic2

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"]
  1. 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.

**Original Pull Request:** https://github.com/langchain-ai/langgraph/pull/3345 **State:** closed **Merged:** No --- ## 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. By applying monkey patch, I temporarily fixed the issues related to conditional node execution within the workflow and 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 hope this approach can be adopted. --- ## 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.** ```python 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": []}) ``` ![pic1](https://github.com/user-attachments/assets/a27b16c5-9c82-4908-b7d2-73f5c7293ddd) ```python 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.** ```python 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": []}) ``` ![pic2](https://github.com/user-attachments/assets/98ef41a2-8c53-44e0-aa95-574e38ee99a0) ```python 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.** ```python # 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": []}) ``` ```python 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.** ```python 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": []}) ``` ```python 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. ```python 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": []}) ``` ```python 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. ```python 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": []}) ``` ```python 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. ```python 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.
yindo added the pull-request label 2026-02-20 17:48:16 -05:00
yindo closed this issue 2026-02-20 17:48:16 -05:00
Sign in to join this conversation.
No Branch/Tag Specified
Branches Tags
Labels
Clear labels
agents
awaiting author revision
bug
change
chat models
checkpointer
ci
code quality
code quality
dependencies
documentation
drawing
duplicate
enhancement
good first issue
help wanted
invalid
maintainer
meta
missing information
not maintained by langchain
open-swe-auto
open-swe-max
open-swe-max-auto
pending
performance
pull-request
Mirrored from GitHub Pull Request
 question
requires design
sdk
server
stale
studio
tests
type checking
usage question
v0.5
v1.1
No Label pull-request
Milestone
No items
No Milestone
Projects
Clear projects
No project
Assignees
Clear assignees
yindo
No Assignees
1 Participants
Notifications
Due Date
No due date set.
Dependencies

No dependencies set.

Reference: langchain-ai/langgraph#3229
Delete Branch "%!s()"

Deleting a branch is permanent. Although the deleted branch may continue to exist for a short time before it actually gets removed, it CANNOT be undone in most cases. Continue?