Dynamic Edge Conditions and Conditional Branching in ChatDev Workflow Graphs

ChatDev implements dynamic edge conditions and conditional branching through a declarative YAML configuration system combined with runtime condition managers, payload processors, and Map/Tree execution modes that determine workflow paths based on message content.

ChatDev, maintained by OpenBMB, structures AI agent workflows as directed graphs where nodes represent execution units and edges control message flow. Unlike static workflow engines, ChatDev supports dynamic edge conditions that evaluate at runtime whether an edge should be traversed, enabling sophisticated conditional branching logic. This article examines the implementation details, configuration schemas, and practical applications of these mechanisms using actual source code from the repository.

Understanding ChatDev's Graph Execution Model

In ChatDev, a workflow graph consists of nodes (literal, human, agent types) connected by edges that carry messages. The system provides three orthogonal mechanisms to control this flow dynamically:

• Edge Conditions determine whether an edge is active for a specific payload, implemented via EdgeConditionConfig schemas and runtime managers.
• Edge Processors transform the message payload before it reaches the downstream node using EdgeProcessorConfig and concrete processors.
• Dynamic Edge Modes (Map and Tree) control whether downstream nodes are replicated for parallel processing or split-and-reduction operations, configured through DynamicEdgeConfig.

These components work together in workflow/graph_manager.py, where GraphManager._initiate_edges assembles the runtime graph by attaching condition and processor objects to each edge payload.

Configuring Edge Conditions

Edge conditions declare the criteria that must be satisfied for a message to traverse a specific edge. You define these conditions declaratively in YAML, which the engine converts into executable logic at runtime.

Declarative Schema Definition

In your graph YAML, conditions appear under the condition key of an edge definition:

edges:
  - from: A
    to: B
    condition:
      type: function
      config:
        name: is_positive

The type field selects the condition implementation, while config provides type-specific parameters. The EdgeConditionConfig.from_dict method in entity/configs/edge/edge_condition.py (lines 250-260) normalizes this input and resolves the concrete schema via get_edge_condition_schema, delegating to specialized classes like FunctionEdgeConditionConfig.

Runtime Evaluation

During graph construction, GraphManager._initiate_edges stores the condition configuration in the edge payload. At execution time, the engine instantiates the appropriate EdgeConditionManager based on the condition type:

• FunctionEdgeConditionManager (runtime/edge/conditions/function_manager.py, lines 21-44) builds a callable evaluator that invokes user-defined Python functions from the functions/edge_condition/ directory.
• KeywordEdgeConditionManager (runtime/edge/conditions/keyword_manager.py) evaluates keyword matching rules against the message content.

When traversing an edge, the manager's process method calls _process_with_condition, returning True to allow passage or False to skip the edge. This mechanism enables conditional branching where only edges satisfied by the current payload are active.

Transforming Payloads with Edge Processors

After a condition passes, the payload can undergo transformation before reaching the downstream node. This occurs through edge processors defined under the process key in YAML:

process:
  type: regex_extract
  config:
    pattern: '```(?P<code>.*?)```'
    group: code
    dotall: true
    on_no_match: pass

EdgeProcessorConfig.from_dict in entity/configs/edge/edge_processor.py (lines 84-98) parses this configuration into concrete processor configs such as RegexEdgeProcessorConfig. At execution time, RegexEdgePayloadProcessor in runtime/edge/processors/regex_processor.py (lines 31-55) extracts the specified pattern, applies optional templating, and returns a new Message instance.

Function-based processors follow a similar pattern, delegating to FunctionEdgePayloadProcessor in runtime/edge/processors/function_processor.py to execute custom transformation logic.

Advanced Branching: Map and Tree Dynamic Modes

For complex workflows requiring parallel execution or hierarchical reduction, ChatDev offers dynamic edge modes configured through DynamicEdgeConfig in entity/configs/edge/dynamic_edge_config.py.

Map Mode

Map mode splits the incoming message into independent pieces and creates virtual sub-executions for each piece:

edges:
  - from: split_input
    to: process_each
    dynamic:
      type: map
      split:
        type: regex
        pattern: ',\s*'

The engine uses the specified edge processor (here, a regex splitter) to divide the payload, then replicates the downstream node for each segment, executing them in parallel. Results are collected into a list that becomes the aggregated payload for subsequent nodes.

Tree Mode

Tree mode implements a split-process-reduce pattern for hierarchical operations like summarization:

edges:
  - from: long_text
    to: summarizer
    dynamic:
      type: tree
      split:
        type: regex
        pattern: '\n\n+'
      config:
        max_parallel: 4
        group_size: 3

The runtime splits the document into paragraphs using SplitConfig, processes each through the downstream node (up to max_parallel concurrently), then uses automatically generated reducer nodes to combine results. This continues recursively—merging every group_size outputs—until a single result remains, as defined in TreeDynamicConfig within entity/configs/dynamic_base.py.

Complete Conditional Branching Implementation

Combining these features enables sophisticated, data-driven workflows. Consider a graph that routes messages based on numeric content while extracting values:

graph:
  id: demo_branch
  nodes:
    - id: start
      type: literal
      config:
        content: "User says: {{input}}"
    - id: check_positive
      type: human
    - id: check_negative
      type: human
  edges:
    - from: start
      to: check_positive
      trigger: true
      condition:
        type: function
        config:
          name: is_positive
      process:
        type: regex_extract
        config:
          pattern: '(\d+)'
          group: 1
    - from: start
      to: check_negative
      trigger: true
      condition:
        type: function
        config:
          name: is_negative

First, create the condition functions in functions/edge_condition/:

# functions/edge_condition/is_positive.py

def is_positive(text: str) -> bool:
    """Return True if text represents a positive number."""
    try:
        return int(text) > 0
    except ValueError:
        return False

FunctionManager automatically discovers functions in this directory at startup. During execution, the engine evaluates both edge conditions in parallel. Only the edge whose condition returns True proceeds, with the payload pre-processed by the regex extractor, effectively implementing conditional branching.

Summary

• Edge conditions use EdgeConditionConfig schemas and runtime managers (FunctionEdgeConditionManager, KeywordEdgeConditionManager) to determine edge traversal based on payload content.
• Edge processors transform messages via EdgeProcessorConfig implementations like RegexEdgePayloadProcessor before they reach downstream nodes.
• Dynamic modes enable parallel processing through Map (replicate and collect) and Tree (split-process-reduce) configurations controlled by DynamicEdgeConfig.
• File locations: Configuration schemas reside in entity/configs/edge/, runtime implementations in runtime/edge/conditions/ and runtime/edge/processors/, and orchestration logic in workflow/graph_manager.py.
• Extensibility: Custom conditions and processors are automatically discovered when placed in functions/edge_condition/ or configured via YAML function references.

Frequently Asked Questions

How do I create a custom condition function in ChatDev?

Create a Python file in the functions/edge_condition/ directory containing a function that accepts a string parameter and returns a boolean. For example, define is_even(text: str) -> bool that parses integers and checks divisibility. The FunctionManager scans this directory at startup and registers your function automatically, making it available for reference in YAML edge condition configurations using the type: function and config.name: is_even parameters.

What is the difference between Map and Tree dynamic edge modes?

Map mode splits the input message into independent chunks and executes the downstream node once per chunk in parallel, collecting all results into a list. Tree mode also splits the input but adds a reduction phase: after processing chunks, results are fed into reducer nodes that combine them recursively until a single output remains, making Tree mode suitable for hierarchical operations like multi-stage summarization where intermediate results must be merged.

How does payload processing interact with edge conditions?

Edge conditions and edge processors operate sequentially during edge traversal. First, the EdgeConditionManager evaluates whether the edge should activate based on the raw payload. If the condition passes (True), the EdgeProcessor transforms the payload (extracting text, applying functions, etc.) before it reaches the downstream node. In dynamic edges, the processor may also handle the splitting logic defined in SplitConfig before the Map or Tree execution mode takes effect.

Where does the runtime decide to spawn parallel executions for dynamic edges?

The decision occurs in GraphManager._initiate_edges within workflow/graph_manager.py (lines 45-60), where dynamic configurations are stored in edge payloads. During actual execution, node executors in runtime/node/executor/ (such as python_executor.py) inspect the dynamic_config attribute. If present, they invoke DynamicEdgeConfig.is_map() or is_tree() to determine whether to spawn virtual sub-executions for parallel processing or hierarchical reduction, delegating split operations to the attached edge processor.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}