Implementing Human-in-the-Loop Feedback in ChatDev Agent Workflows: A Complete Technical Guide

ChatDev implements human-in-the-loop feedback through a three-layer architecture consisting of the PromptChannel transport protocol, the HumanPromptService orchestration layer, and the HumanNodeExecutor runtime node, enabling pluggable human interaction across CLI, WebSocket, and custom enterprise integrations.

ChatDev, the multi-agent collaborative development framework by OpenBMB, provides native support for pausing agent workflows to collect human feedback through its modular human-in-the-loop architecture. Implementing human-in-the-loop feedback in ChatDev agent workflows requires understanding how the transport layer, service orchestration, and graph execution interact to create seamless breakpoints for operator input. This guide examines the source code implementation across the utils/human_prompt.py service layer and runtime/node/executor/human_executor.py execution engine to demonstrate how to integrate interactive human approval steps into production agent graphs.

The Three-Layer Human-in-the-Loop Architecture

According to the OpenBMB/ChatDev source code, the framework decouples human interaction into three distinct responsibilities: transport, orchestration, and execution. This separation allows developers to swap input mechanisms (CLI, WebSocket, Slack) without modifying workflow graph logic.

PromptChannel Transport Layer

The PromptChannel protocol in utils/human_prompt.py defines the contract for all human interaction back-ends. Any transport implementation must provide a request method matching this signature:

class PromptChannel(Protocol):
    def request(
        *,
        node_id: str,
        task: str,
        inputs: Optional[str] = None,
        metadata: Optional[Dict[str, Any]] = None,
    ) -> PromptResult: ...

ChatDev ships with two concrete implementations:

• CliPromptChannel – Reads from standard input (stdin) for local development and debugging scenarios.
• WebPromptChannel – Located in server/services/prompt_channel.py, this class posts JSON payloads via WebSocket and blocks until the browser client responds.

Developers can inject custom channels through the resolve_prompt_channel hook in utils/human_prompt.py. The GraphExecutor lazily initializes the service via _ensure_human_prompt_service in workflow/graph.py, falling back to CliPromptChannel if no custom hook is registered.

HumanPromptService Orchestration

The HumanPromptService class in utils/human_prompt.py acts as a singleton coordinator that manages the lifecycle of human input requests. When a human node executes, the service performs six critical operations:

1. Thread Safety – Acquires an async lock (self._lock) to guarantee only one active prompt per workflow thread.
2. Timing – Invokes LogManager.human_timer to measure elapsed wait time for observability.
3. Delegation – Calls self._channel.request to delegate to the concrete transport.
4. Normalization – Ensures the result contains string-only text and a list of MessageBlock objects.
5. Sanitization – Strips unsafe Unicode bytes and merges metadata.
6. Recording – Persists the interaction via LogManager.record_human_interaction for audit trails.

The service is instantiated once per workflow execution and shared across all human nodes via the runtime context.

HumanNodeExecutor Runtime Integration

The HumanNodeExecutor in runtime/node/executor/human_executor.py bridges the service layer and the graph engine. Its execute method performs four steps:

1. Validates that the node type is "human" and that a HumanConfig exists.
2. Renders node inputs as plain text using self._inputs_to_text(inputs).
3. Retrieves the shared HumanPromptService from NodeExecutor.context and invokes request.
4. Wraps the returned PromptResult into a Message with MessageRole.USER and returns it to the graph.

The executor is automatically registered for the "human" node type in runtime/node/builtin_nodes.py, ensuring seamless instantiation without manual wiring.

Configuring Human Nodes in Workflow Graphs

Human breakpoints are declared declaratively in YAML workflow definitions. The HumanConfig schema in entity/configs/node/human.py parses the optional description field used to prompt operators.

nodes:
  - id: ask_user
    type: human
    config:
      description: |
        Please review the generated design and confirm if it meets the requirements.

If the description field is omitted, the node renders only the formatted inputs passed from upstream nodes. The type: human key triggers the HumanNodeExecutor instantiation at runtime.

Implementing Pluggable Feedback Channels

CLI Fallback for Local Development

When no custom workspace hook is provided, ChatDev automatically selects CliPromptChannel. This implementation prints a formatted banner to stdout and blocks on Python’s built-in input() function:

===== HUMAN INPUT REQUIRED =====
=== Task for human (ask_user) ===
Please review the generated design and confirm if it meets the requirements.
=== Your response: ===

This mode requires no additional infrastructure and is ideal for debugging graphs locally.

WebSocket Integration for Web UI

For production deployments using the ChatDev Web UI, WebPromptChannel collaborates with two server-side services:

• SessionExecutionController (server/services/session_execution.py) – Stores a Future object awaiting human input and resolves it when the client posts a response.
• WebsocketManager – Sends the human_input_required JSON message to the browser via the _notify_human_prompt method.

The channel blocks on session_controller.wait_for_human_input until the front-end submits a payload:

{
  "type": "human_input_required",
  "data": {
    "node_id": "ask_user",
    "input": "...rendered inputs...",
    "task_description": "Please review the generated design..."
  }
}

Upon submission, the service attaches any files via AttachmentService and unblocks the workflow.

Building Custom Channels (Enterprise Integration Example)

You can implement PromptChannel to integrate with enterprise tools like Slack, Microsoft Teams, or internal ticketing systems. Below is a pattern for a Slack webhook integration:

from utils.human_prompt import PromptChannel, PromptResult
import requests

class SlackPromptChannel(PromptChannel):
    def __init__(self, webhook_url: str):
        self.webhook_url = webhook_url

    def request(self, *, node_id: str, task: str, inputs: str | None = None, metadata=None) -> PromptResult:
        payload = {
            "text": f"*Human task* ({node_id}):\n{task}\n\n*Inputs*:\n{inputs or ''}"
        }
        resp = requests.post(self.webhook_url, json=payload)
        resp.raise_for_status()
        # In production, block here waiting for Slack callback or timeout

        return PromptResult(text="✅ Approved via Slack", blocks=None)

# Register via workspace hook

def workspace_hook_factory(runtime):
    runtime.workspace_hook = type("Hook", (), {
        "get_prompt_channel": lambda self: SlackPromptChannel("https://hooks.slack.com/services/...")
    })()
    return runtime

Pass workspace_hook_factory to the runtime initialization to override the default CLI behavior.

Session Management and Blocking Mechanics

The WebSocket implementation uses an async Future pattern to pause graph execution without consuming worker threads. When WebPromptChannel.request is invoked:

1. It calls session_controller.set_waiting_for_input to update session state.
2. It awaits session_controller.wait_for_human_input, which blocks until the external client POSTs a response to the server endpoint.
3. The controller resolves the Future with the user’s text and attachments, which the channel converts into a PromptResult.

This design allows the workflow engine to handle thousands of concurrent sessions while consuming minimal resources during human wait periods.

Summary

Implementing human-in-the-loop feedback in ChatDev agent workflows leverages a clean separation of concerns across three architectural layers:

• PromptChannel defines the transport contract for gathering input, with built-in CLI and WebSocket implementations located in utils/human_prompt.py and server/services/prompt_channel.py.
• HumanPromptService orchestrates request locking, timing, sanitization, and logging, ensuring consistent behavior across all human nodes.
• HumanNodeExecutor integrates human breakpoints into the graph execution engine, automatically handling nodes with type: human as defined in YAML configurations.

Developers can extend the system by implementing the PromptChannel protocol and registering custom channels through workspace hooks, enabling integration with Slack, REST APIs, or proprietary internal tools without modifying core framework code.

Frequently Asked Questions

How do I add a human approval step to a ChatDev workflow?

Add a node with type: human to your workflow YAML file and provide a description field in the config section. When the graph executor reaches this node, it will pause and invoke the configured PromptChannel to gather operator input before proceeding to downstream tasks.

What is the difference between PromptChannel and HumanPromptService?

PromptChannel is the transport abstraction responsible for the actual I/O (reading from CLI, WebSocket, or external APIs), while HumanPromptService is the orchestration singleton that manages request locking, timing instrumentation, result normalization, and observability logging. The service delegates to the channel but handles cross-cutting concerns like thread safety and audit trails.

Can I implement human-in-the-loop feedback without the Web UI?

Yes. If no custom PromptChannel is provided via a workspace hook, ChatDev automatically defaults to CliPromptChannel, which reads from standard input. This allows full human-in-the-loop functionality in headless environments, CI/CD pipelines, or remote SSH sessions without requiring a browser interface.

How does ChatDev prevent race conditions in human input requests?

The HumanPromptService in utils/human_prompt.py uses an async lock (self._lock) to ensure that only one human input request is active per workflow thread at any time. This prevents multiple concurrent nodes from simultaneously prompting the operator and guarantees sequential, deterministic feedback collection.

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