Batch Execution of Multiple ChatDev Workflows Programmatically: A Complete Guide

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 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 wraps the definition with runtime metadata including the output root, log level, and variable overrides.
3. Execution Context: GraphContext in 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 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 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.

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.

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.

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 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: Contains the public run_workflow entry point that prepares the graph context and returns a WorkflowRunResult.
• workflow/graph.py: Implements GraphExecutor, which handles memory initialization, strategy selection (DAG/Cycle/MajorityVote), and node execution.
• workflow/graph_context.py: Defines GraphContext, the mutable runtime container that manages nodes, edges, and cycle detection for each session.
• entity/graph_config.py: Stores GraphConfig, the immutable wrapper for parsed YAML metadata and runtime variables.
• check/check.py: Provides load_config for YAML validation and GraphDefinition construction.
• server/routes/execute_sync.py: Exposes the HTTP API with optional SSE streaming via _run_workflow_with_logger.
• utils/task_input.py: Transforms prompts and file lists into Message objects for graph input.
• 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) is best for local automation and integrates directly with your existing Python codebase. The HTTP API (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.

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