# Batch Execution of Multiple ChatDev Workflows Programmatically: A Complete Guide

> Learn to programmatically run multiple ChatDev workflows in batch. Explore the stateless run_workflow SDK function for isolated sessions and efficient execution. Improve your development process.

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

---

**You can execute multiple ChatDev workflows in batch by calling the stateless `run_workflow` SDK function in a loop or thread pool, as each invocation creates an isolated session with its own output directory and memory context.**

ChatDev 2.0 (also known as **DevAll**) represents workflows as declarative YAML graphs that define agents, memory stores, and execution policies. When you need to process multiple tasks programmatically, the OpenBMB/ChatDev repository provides a thread-safe, stateless SDK that handles batch execution of multiple ChatDev workflows without manual session management. The architecture ensures that every call to `run_workflow` instantiates fresh resources, making parallel processing safe and predictable.

## Understanding the ChatDev Workflow Execution Pipeline

Before implementing batch execution, it helps to understand how a single workflow moves from a YAML file to a completed result. The pipeline consists of five distinct layers, each handled by specific modules in the codebase:

1. **YAML Parsing and Validation**: The `check.check.load_config` function in [`check/check.py`](https://github.com/OpenBMB/ChatDev/blob/main/check/check.py) parses the workflow file, validates it against a JSON-Schema, and builds an immutable `GraphDefinition` object.
2. **Runtime Configuration**: `GraphConfig.from_definition` in [`entity/graph_config.py`](https://github.com/OpenBMB/ChatDev/blob/main/entity/graph_config.py) wraps the definition with runtime metadata including the output root, log level, and variable overrides.
3. **Execution Context**: `GraphContext` in [`workflow/graph_context.py`](https://github.com/OpenBMB/ChatDev/blob/main/workflow/graph_context.py) transforms the static configuration into a mutable, execution-ready container that holds node and edge objects, manages the output directory, and performs cycle detection.
4. **Graph Execution**: `GraphExecutor` in [`workflow/graph.py`](https://github.com/OpenBMB/ChatDev/blob/main/workflow/graph.py) builds the complete execution environment—including the tool manager, function manager, token tracker, and memory stores—and runs the graph using strategies like DAG, cycle, or majority-vote.
5. **SDK Facade**: `runtime.sdk.run_workflow` in [`runtime/sdk.py`](https://github.com/OpenBMB/ChatDev/blob/main/runtime/sdk.py) orchestrates the above layers into a single Python function that returns a `WorkflowRunResult` containing the final message and metadata.

This architecture ensures that `run_workflow` is entirely self-contained, creating a fresh `GraphContext` and output session for every invocation.

## Why Batch Execution Works Out-of-the-Box

The ChatDev SDK requires no special configuration for batch processing because of three core design decisions:

- **Stateless Execution**: Each call to `run_workflow` creates a unique `session_name` and dedicated `OUTPUT_ROOT` subdirectory, ensuring that concurrent runs never share mutable state or overwrite each other's artifacts.
- **Thread-Safe Resource Isolation**: `AttachmentStore`, `MemoryManager`, and the token tracker are instantiated per-workflow rather than globally, allowing parallel executions to safely use the same underlying LLM client without race conditions.
- **Dynamic Variable Injection**: The `variables` parameter lets you inject environment-style values (such as API keys or model names) on a per-run basis, which is essential when batching workflows that target different backends or require different credentials.

## Method 1: Synchronous Batch Execution Using the Python SDK

For small batches or sequential processing, iterate over a list of job dictionaries and call `run_workflow` for each entry. Each job can specify its own YAML file, prompt, attachments, and variable overrides.

```python
from pathlib import Path
from runtime.sdk import run_workflow
import json

# Define batch jobs with distinct configurations

batch = [
    {
        "yaml": "yaml_instance/data_visualization_basic.yaml",
        "prompt": "Create a bar chart for the sales data in sales.csv.",
        "attachments": ["datasets/sales.csv"],
        "variables": {"API_KEY": "sk-xxx-data-vis", "MODEL": "gpt-4o"},
    },
    {
        "yaml": "yaml_instance/blender_3d_builder_simple.yaml",
        "prompt": "Generate a 3-D model of a festive Christmas tree.",
        "attachments": [],
        "variables": {"API_KEY": "sk-xxx-blender"},
    },
]

results = []
for job in batch:
    result = run_workflow(
        yaml_file=job["yaml"],
        task_prompt=job["prompt"],
        attachments=job.get("attachments"),
        variables=job.get("variables"),
        session_name=f"batch_{Path(job['yaml']).stem}",  # Deterministic naming

    )
    
    results.append({
        "session": result.meta_info.session_name,
        "output": result.final_message.text_content() if result.final_message else "(no output)",
        "token_usage": result.meta_info.token_usage,
        "output_dir": str(result.meta_info.output_dir),
    })

# Persist batch summary

Path("batch_summary.json").write_text(json.dumps(results, ensure_ascii=False, indent=2))

```

The `session_name` parameter ensures each run writes to a distinct subdirectory under `WareHouse/`, while `meta_info.output_dir` provides the absolute path to retrieve generated artifacts like code files or images.

## Method 2: Parallel Batch Execution with Concurrent Futures

When processing many independent workflows, use a `ThreadPoolExecutor` to run jobs concurrently. The SDK is I/O-bound due to LLM API calls and file operations, so Python's Global Interpreter Lock (GIL) does not become a bottleneck.

```python
import concurrent.futures
from runtime.sdk import run_workflow

def execute_job(job):
    """Wrapper to unpack job dict and run workflow."""
    return run_workflow(
        yaml_file=job["yaml"],
        task_prompt=job["prompt"],
        attachments=job.get("attachments", []),
        variables=job.get("variables", {}),
        session_name=job.get("session_name"),
    )

batch_jobs = [
    # Same structure as synchronous example

    {"yaml": "yaml_instance/task1.yaml", "prompt": "Refactor this code.", "variables": {}},
    {"yaml": "yaml_instance/task2.yaml", "prompt": "Write documentation.", "variables": {}},
]

with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
    futures = {executor.submit(execute_job, job): job for job in batch_jobs}
    
    for future in concurrent.futures.as_completed(futures):
        job = futures[future]
        try:
            res = future.result()
            print(f"[{res.meta_info.session_name}] Completed successfully")
        except Exception as exc:
            print(f"[{job['yaml']}] Failed: {exc}")

```

Each thread maintains its own `GraphContext` and `MemoryManager` instances, ensuring that token tracking and memory persistence remain isolated across parallel executions.

## Method 3: Asynchronous Batch Execution via the HTTP API

If you are running the ChatDev FastAPI server (default port 6400), you can drive batches through the `/api/workflow/run` endpoint. This approach supports Server-Sent Events (SSE) for real-time log streaming, which is valuable for monitoring long-running 3D rendering or research workflows.

```python
import aiohttp
import asyncio

async def run_one(session, yaml, prompt, attachments=None, variables=None):
    payload = {
        "yaml_file": yaml,
        "task_prompt": prompt,
        "attachments": attachments or [],
        "variables": variables or {},
    }
    
    async with session.post(
        "http://localhost:6400/api/workflow/run",
        json=payload,
        headers={"Accept": "text/event-stream"}  # Request SSE streaming

    ) as resp:
        if resp.headers["content-type"] == "text/event-stream":
            async for line in resp.content:
                if line.startswith(b"event:"):
                    # Parse or forward SSE events to UI

                    print(line.decode().strip())
        else:
            data = await resp.json()
            return data

async def main():
    async with aiohttp.ClientSession() as session:
        jobs = [
            ("yaml_instance/data_visualization_basic.yaml", "Chart Q1 data.", ["data.csv"]),
            ("yaml_instance/deep_research_v1.yaml", "Summarize RL papers.", []),
        ]
        await asyncio.gather(*(run_one(session, y, p, a) for y, p, a in jobs))

asyncio.run(main())

```

The server endpoint in [`server/routes/execute_sync.py`](https://github.com/OpenBMB/ChatDev/blob/main/server/routes/execute_sync.py) uses the same `run_workflow` logic as the SDK, ensuring behavioral parity between local Python execution and remote HTTP calls.

## Core Source Files for Programmatic Workflow Automation

Understanding these modules helps when debugging batch failures or extending functionality:

- **[`runtime/sdk.py`](https://github.com/OpenBMB/ChatDev/blob/main/runtime/sdk.py)**: Contains the public `run_workflow` entry point that prepares the graph context and returns a `WorkflowRunResult`.
- **[`workflow/graph.py`](https://github.com/OpenBMB/ChatDev/blob/main/workflow/graph.py)**: Implements `GraphExecutor`, which handles memory initialization, strategy selection (DAG/Cycle/MajorityVote), and node execution.
- **[`workflow/graph_context.py`](https://github.com/OpenBMB/ChatDev/blob/main/workflow/graph_context.py)**: Defines `GraphContext`, the mutable runtime container that manages nodes, edges, and cycle detection for each session.
- **[`entity/graph_config.py`](https://github.com/OpenBMB/ChatDev/blob/main/entity/graph_config.py)**: Stores `GraphConfig`, the immutable wrapper for parsed YAML metadata and runtime variables.
- **[`check/check.py`](https://github.com/OpenBMB/ChatDev/blob/main/check/check.py)**: Provides `load_config` for YAML validation and `GraphDefinition` construction.
- **[`server/routes/execute_sync.py`](https://github.com/OpenBMB/ChatDev/blob/main/server/routes/execute_sync.py)**: Exposes the HTTP API with optional SSE streaming via `_run_workflow_with_logger`.
- **[`utils/task_input.py`](https://github.com/OpenBMB/ChatDev/blob/main/utils/task_input.py)**: Transforms prompts and file lists into `Message` objects for graph input.
- **[`utils/attachments.py`](https://github.com/OpenBMB/ChatDev/blob/main/utils/attachments.py)**: Manages the `AttachmentStore` for persisting uploaded files per workflow session.

## Summary

- **Stateless Design**: ChatDev's `run_workflow` function creates isolated sessions automatically, making batch execution of multiple ChatDev workflows safe without manual sandboxing.
- **Flexible Entry Points**: Use the Python SDK for direct integration, `concurrent.futures` for parallel processing, or the FastAPI HTTP endpoint for distributed architectures.
- **Per-Run Configuration**: Override variables, prompts, and attachments for each job individually via the `variables` and `attachments` parameters.
- **Isolated Outputs**: Each workflow writes to a unique subdirectory under `WareHouse/`, accessible via `result.meta_info.output_dir`.
- **Thread Safety**: Resource managers (`MemoryManager`, `AttachmentStore`) are instantiated per-workflow, supporting safe multi-threaded execution.

## Frequently Asked Questions

### Can I run hundreds of ChatDev workflows simultaneously?

Yes, but you should limit concurrency based on your LLM provider's rate limits and your local file system capacity. While the SDK is thread-safe, the underlying LLM API may throttle requests. Use a `ThreadPoolExecutor` with controlled `max_workers` (e.g., 4-10) and implement exponential backoff for API rate limit errors.

### How do I isolate outputs when running workflows in parallel?

Isolation is automatic. Each call to `run_workflow` generates a unique `session_name` (or uses the one you provide) and creates a dedicated output directory under `WareHouse/<session_name>/`. The `result.meta_info.output_dir` field provides the absolute path to retrieve artifacts, ensuring parallel runs never overwrite files.

### Does batch execution support different LLM backends for each job?

Yes. Pass backend-specific credentials and model names through the `variables` dictionary in each job definition. The `GraphConfig` class injects these variables into the YAML configuration at runtime, allowing one batch to mix workflows targeting OpenAI, Azure, or local models.

### What is the difference between using the SDK and the HTTP API for batches?

The Python SDK ([`runtime/sdk.py`](https://github.com/OpenBMB/ChatDev/blob/main/runtime/sdk.py)) is best for local automation and integrates directly with your existing Python codebase. The HTTP API ([`server/routes/execute_sync.py`](https://github.com/OpenBMB/ChatDev/blob/main/server/routes/execute_sync.py)) is ideal for distributed systems or when you need to stream logs via SSE to a web interface. Both use identical core logic in `GraphExecutor`, so results are consistent across methods.