Logging and Debugging ChatDev Workflow Execution Traces: A Complete Guide

ChatDev provides a structured logging subsystem through WorkflowLogger, StructuredLogger, and LogManager that captures hierarchical JSON execution traces for every node, model call, and tool invocation in AI workflows.

ChatDev is an open-source framework for orchestrating multi-agent AI workflows available at OpenBMB/ChatDev. Understanding how to log and debug execution traces is essential for monitoring complex graph-based pipelines and diagnosing failures in production environments. This guide examines the three-tier logging architecture that enables real-time observability and post-mortem analysis of workflow runs through fine-grained, JSON-structured event capture.

The Three-Tier Logging Architecture

ChatDev embeds a structured logging subsystem consisting of three tightly-coupled components that work together to record every significant event, duration, and hierarchical context during workflow execution.

WorkflowLogger: Core Event Capture

The WorkflowLogger class in utils/logger.py serves as the primary interface for creating LogEntry objects and maintaining an in-memory log list. When instantiated, it receives a unique workflow_id derived from the graph name and timestamp, plus a configured LogLevel (defaulting to DEBUG).

Key responsibilities include:

• Filtering events based on the current log_level setting
• Capturing execution path hierarchies via self.current_path
• Escaping complex objects using _json_safe for JSON serialization
• Forwarding structured messages to the StructuredLogger component

StructuredLogger: JSON Output Stream

The StructuredLogger in utils/structured_logger.py provides a thin wrapper around Python’s standard logging module, emitting JSON-formatted logs suitable for downstream ingestion by ELK stacks, Splunk, or custom analytics pipelines. When use_structured_logging=True (the default), the WorkflowLogger creates a StructuredLogger instance via get_workflow_logger(self.workflow_id) at line 78.

This component serializes LogEntry objects and writes them to logs/{workflow_id}.log or stdout, respecting environment variables such as WORKFLOW_LOG_FILE and LOG_LEVEL.

LogManager: Backward Compatibility Shim

The LogManager in utils/log_manager.py acts as a compatibility layer throughout the codebase, delegating timing-context managers and event-recording helpers to the underlying WorkflowLogger. Legacy code continues using log_manager.node_timer(...) while the actual implementation resides in the core logger.

How Workflow Execution Traces Are Generated

Logger Instantiation in workflow/graph.py

When a Workflow object initializes in workflow/graph.py (around line 121), it creates a dedicated logger instance:

def _create_logger(self) -> WorkflowLogger:
    """Create and return a logger instance."""
    return WorkflowLogger(self.graph.name, self.graph.log_level)

The workflow engine automatically configures the logger with the graph's name and log level, establishing the foundation for all subsequent trace collection.

LogEntry Lifecycle and Schema

Each event generates a LogEntry object (defined in utils/logger.py, lines 41-62) containing:

• timestamp: Unix timestamp of the event
• level: Severity level (DEBUG, INFO, WARNING, ERROR, CRITICAL)
• node_id: Identifier for the executing node
• event_type: Enum value from entity.enums categorizing the event
• message: Human-readable description
• details: JSON-safe dictionary with contextual data
• execution_path: Current node stack showing hierarchical position
• duration: Execution time in seconds (when applicable)

The add_log method (lines 81-136) processes these entries by filtering against the current log_level, capturing the execution path, converting objects to JSON-safe formats, appending to the internal log list, optionally printing to console, and propagating to the StructuredLogger for persistent storage.

Timing Context Managers for Performance Tracking

WorkflowLogger provides built-in timing helpers including node_timer, model_timer, and tool_timer (lines 106-164). These context managers calculate durations automatically:

@contextmanager
def node_timer(self, node_id: str):
    self.__init_timers__()
    start_time = time.time()
    try: 
        yield
    finally:
        self._timers[node_id] = time.time() - start_time

LogManager forwards these context managers (lines 30-70) so developers can instrument code blocks without directly accessing the core logger.

Event Recording API Reference

During execution, the graph calls high-level record methods at specific lifecycle points. These methods populate the details field, include durations from associated timers, and update the execution_path hierarchy:

Method	Trigger	Location in workflow/graph.py
record_node_start / record_node_end	Node entry and exit	Lines 559-571 and 583-594
record_edge_process	Edge traversal	Line ~620
record_model_call	Model API invocation	Line ~660
record_tool_call	External tool usage	Line ~690
record_human_interaction	Human-in-the-loop steps	Line ~720
record_workflow_start / record_workflow_end	Whole-workflow lifecycle	Lines ~540 and 735
record_thinking_process	Specialized AI reasoning stages	Line ~730
record_memory_operation	Memory read/write operations	Line ~750

Each call automatically captures the current execution context, enabling precise reconstruction of the workflow state at any point in the trace.

Retrieving and Analyzing Execution Traces

Exporting Logs to JSON

After workflow completion, extract the complete execution trace using:

# Get JSON string for network transmission or file persistence

json_logs = logger.to_json()

# Get Python dictionary for programmatic analysis

log_dict = logger.to_dict()

The to_json() method (line 86 in utils/logger.py) serializes the entire log history, while to_dict() (line 76) provides a structured Python dictionary.

Execution Summary Statistics

Generate aggregated metrics using:

summary = logger.get_execution_summary()
print(f"Total duration: {summary['total_duration']}")
print(f"Error count: {summary['error_count']}")
print(f"Warning count: {summary['warning_count']}")

The get_execution_summary() method (lines 51-75) aggregates total duration across all nodes, counts errors and warnings by severity level, and reports the final execution path.

Practical Implementation Examples

Creating a Logger for a New Workflow

from utils.logger import WorkflowLogger
from entity.enums import LogLevel

# Create a logger with structured JSON output

logger = WorkflowLogger(
    workflow_id="my_workflow_001",
    log_level=LogLevel.DEBUG,
    use_structured_logging=True,
    log_to_console=True
)

Recording Node Execution with Timing

def run_my_node(node, inputs):
    logger.enter_node(node.id, inputs, node_type=node.type)
    
    with logger.node_timer(node.id):
        # Execute node logic

        output = process_inputs(inputs)
    
    logger.exit_node(
        node.id, 
        output, 
        duration=logger.get_timer(node.id)
    )

Using LogManager for Legacy Integration

from utils.log_manager import LogManager
from utils.logger import WorkflowLogger

runtime_logger = WorkflowLogger("example")
log_manager = LogManager(runtime_logger)

# Record tool execution with metadata

log_manager.record_tool_call(
    node_id="search_tool_1",
    tool_name="web_search",
    tool_result="{'results': [...]}",
    success=True
)

Configuring Structured Log Destinations

Set environment variables before launching the workflow:

export WORKFLOW_LOG_FILE=logs/production_run.log
export LOG_LEVEL=INFO

StructuredLogger respects these values via get_workflow_logger() in utils/structured_logger.py (lines 71-76).

Key Source Files

File	Purpose
utils/logger.py	Core WorkflowLogger, LogEntry dataclass, JSON-safe conversion, console and structured output routing
utils/structured_logger.py	Minimal JSON logger built on Python’s logging; provides info/debug/warning/error/critical and specialized helpers like log_request and log_response
utils/log_manager.py	Compatibility layer delegating timing context managers and event-recording calls to WorkflowLogger
workflow/graph.py	Orchestrates graph execution, creates the logger instance, and emits all trace events via LogManager
workflow/runtime/runtime_builder.py	Instantiates RuntimeContext with a LogManager; integrates logging into the runtime lifecycle
entity/enums.py	Enumerations for LogLevel, EventType, and CallStage used throughout the logging API
workflow/executor/dag_executor.py	Concrete executor wrapping node processing in timers and calling LogManager methods
workflow/executor/cycle_executor.py	Cycle-aware executor with specialized logging for iterative workflows
server/services/websocket_logger.py	Streams structured logs over WebSocket to UI clients for real-time debugging

Summary

ChatDev’s logging architecture provides comprehensive, hierarchical, and JSON-structured traces for every workflow run. By combining WorkflowLogger, StructuredLogger, and the LogManager shim, developers can:

• Capture fine-grained events including node lifecycle transitions, model invocations, tool executions, and human interactions
• Track precise execution timings via built-in context managers like node_timer and model_timer
• Export logs instantly to console, files, or WebSocket streams using to_json() and get_execution_summary()
• Analyze performance bottlenecks through per-node duration aggregation
• Integrate with external monitoring solutions via structured JSON output

These capabilities enable teams to debug complex AI orchestration pipelines, profile performance across different model versions, and maintain audit trails for compliance requirements.

Frequently Asked Questions

How do I change the log level for a specific workflow run?

Pass the desired LogLevel enum value when instantiating WorkflowLogger. Valid options defined in entity/enums.py include DEBUG, INFO, WARNING, ERROR, and CRITICAL. The add_log method filters events based on this level before recording them. Alternatively, set the global LOG_LEVEL environment variable to affect all StructuredLogger instances.

What is the performance impact of structured logging on workflow execution?

The overhead is minimal because WorkflowLogger uses in-memory list operations for self.logs and delegates JSON serialization to the StructuredLogger asynchronously. Timing context managers use simple time.time() calculations stored in the _timers dictionary. For high-throughput scenarios, you can disable console output by setting log_to_console=False or reduce verbosity by setting log_level=LogLevel.WARNING.

How can I correlate logs across distributed workflow executions?

The workflow_id parameter serves as the correlation identifier. When creating the logger via WorkflowLogger(workflow_id="unique_run_123"), this ID propagates to all LogEntry objects and the structured log file at logs/unique_run_123.log. Include this ID in your centralized logging platform to group related events and reconstruct end-to-end execution flows across multiple services.

Can I extend the logging system to capture custom event types?

Yes. While the standard API covers EventType enums defined in entity/enums.py, you can use the low-level add_log method directly to inject custom events. Instantiate a LogEntry with your desired event_type string (or extend the enum), populate the details dictionary with relevant metadata, and call logger.add_log(entry). Ensure your details dictionary contains only JSON-serializable types or use _json_safe() for conversion.

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