# Dynamic Edge Conditions and Conditional Branching in ChatDev Workflow Graphs

> Explore dynamic edge conditions and conditional branching in ChatDev workflow graphs. Learn how YAML config, condition managers, and execution modes control workflow paths based on message content for efficient development.

- Repository: [OpenBMB/ChatDev](https://github.com/OpenBMB/ChatDev)
- Tags: how-to-guide
- Published: 2026-04-01

---

**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`](https://github.com/OpenBMB/ChatDev/blob/main/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:

```yaml
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`](https://github.com/OpenBMB/ChatDev/blob/main/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`](https://github.com/OpenBMB/ChatDev/blob/main/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`](https://github.com/OpenBMB/ChatDev/blob/main/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:

```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`](https://github.com/OpenBMB/ChatDev/blob/main/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`](https://github.com/OpenBMB/ChatDev/blob/main/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`](https://github.com/OpenBMB/ChatDev/blob/main/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`](https://github.com/OpenBMB/ChatDev/blob/main/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:

```yaml
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:

```yaml
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`](https://github.com/OpenBMB/ChatDev/blob/main/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:

```yaml
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/`:

```python

# 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`](https://github.com/OpenBMB/ChatDev/blob/main/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`](https://github.com/OpenBMB/ChatDev/blob/main/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`](https://github.com/OpenBMB/ChatDev/blob/main/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.