Visualization

This section covers the real-time monitoring, dashboard, and performance visualization capabilities in the Kailash SDK.

Overview

The visualization system provides comprehensive real-time monitoring and performance analysis for workflow execution:

  • Real-time Dashboards: Live monitoring with streaming metrics

  • Performance Reports: Multi-format comprehensive reports

  • Interactive Charts: Chart.js integration for web dashboards

  • API Access: REST and WebSocket endpoints for custom integrations

  • Resource Monitoring: CPU, memory, and I/O tracking

  • Bottleneck Analysis: Automatic performance issue detection

Real-time Dashboard

The core component for live workflow monitoring with background metrics collection.

class kailash.visualization.dashboard.RealTimeDashboard(task_manager: TaskManager, config: DashboardConfig | None = None)[source]

Bases: object

Real-time dashboard for workflow monitoring.

This class provides comprehensive real-time monitoring capabilities including live metrics collection, interactive visualizations, and status reporting for workflow execution.

Usage:

dashboard = RealTimeDashboard(task_manager) dashboard.start_monitoring() # Dashboard runs in background dashboard.generate_live_report(“output.html”) dashboard.stop_monitoring()

Parameters:
__init__(task_manager: TaskManager, config: DashboardConfig | None = None)[source]

Initialize real-time dashboard.

Parameters:

  • task_manager (TaskManager) – TaskManager instance for data access

  • config (DashboardConfig | None) – Dashboard configuration options

start_monitoring(run_id: str | None = None)[source]

Start real-time monitoring for a workflow run.

Parameters:

run_id (str | None) – Specific run to monitor, or None for latest

stop_monitoring()[source]

Stop real-time monitoring.

add_metrics_callback(callback: Any)[source]

Add callback for metrics updates.

Parameters:

callback (Any) – Function that takes LiveMetrics as argument

add_status_callback(callback: Any)[source]

Add callback for status changes.

Parameters:

callback (Any) – Function that takes (event_type, count) as arguments

get_current_metrics() → LiveMetrics | None[source]

Get the most recent metrics.

Return type:

LiveMetrics | None

get_metrics_history(minutes: int | None = None) → list[LiveMetrics][source]

Get metrics history for specified time period.

Parameters:

minutes (int | None) – Number of minutes of history to return

Returns:

List of metrics within time period

Return type:

list[LiveMetrics]

generate_live_report(output_path: str | Path, include_charts: bool = True) → Path[source]

Generate comprehensive live dashboard report.

Parameters:

  • output_path (str | Path) – Path to save HTML dashboard

  • include_charts (bool) – Whether to include performance charts

Returns:

Path to generated dashboard file

Return type:

Path

Dashboard Configuration

Configuration options for customizing dashboard behavior and appearance.

class kailash.visualization.dashboard.DashboardConfig(update_interval: float = 1.0, max_history_points: int = 100, auto_refresh: bool = True, show_completed: bool = True, show_failed: bool = True, theme: str = 'light')[source]

Bases: object

Configuration for dashboard components.

Variables:

  • update_interval (float) – Seconds between dashboard updates

  • max_history_points (int) – Maximum data points to keep in memory

  • auto_refresh (bool) – Whether to automatically refresh data

  • show_completed (bool) – Whether to show completed tasks

  • show_failed (bool) – Whether to show failed tasks

  • theme (str) – Dashboard color theme (‘light’ or ‘dark’)

Parameters:

  • update_interval (float)

  • max_history_points (int)

  • auto_refresh (bool)

  • show_completed (bool)

  • show_failed (bool)

  • theme (str)

update_interval: float = 1.0
max_history_points: int = 100
auto_refresh: bool = True
show_completed: bool = True
show_failed: bool = True
theme: str = 'light'
__init__(update_interval: float = 1.0, max_history_points: int = 100, auto_refresh: bool = True, show_completed: bool = True, show_failed: bool = True, theme: str = 'light') → None
Parameters:

  • update_interval (float)

  • max_history_points (int)

  • auto_refresh (bool)

  • show_completed (bool)

  • show_failed (bool)

  • theme (str)

Return type:

None

Live Metrics

Data models for real-time performance metrics.

class kailash.visualization.dashboard.LiveMetrics(timestamp: datetime = <factory>, active_tasks: int = 0, completed_tasks: int = 0, failed_tasks: int = 0, total_cpu_usage: float = 0.0, total_memory_usage: float = 0.0, throughput: float = 0.0, avg_task_duration: float = 0.0)[source]

Bases: object

Container for live performance metrics.

Variables:

  • timestamp (datetime.datetime) – When metrics were collected

  • active_tasks (int) – Number of currently running tasks

  • completed_tasks (int) – Number of completed tasks

  • failed_tasks (int) – Number of failed tasks

  • total_cpu_usage (float) – System-wide CPU usage percentage

  • total_memory_usage (float) – System-wide memory usage in MB

  • throughput (float) – Tasks completed per minute

  • avg_task_duration (float) – Average task execution time

Parameters:
timestamp: datetime
active_tasks: int = 0
completed_tasks: int = 0
failed_tasks: int = 0
total_cpu_usage: float = 0.0
total_memory_usage: float = 0.0
throughput: float = 0.0
avg_task_duration: float = 0.0
__init__(timestamp: datetime = <factory>, active_tasks: int = 0, completed_tasks: int = 0, failed_tasks: int = 0, total_cpu_usage: float = 0.0, total_memory_usage: float = 0.0, throughput: float = 0.0, avg_task_duration: float = 0.0) → None
Parameters:
Return type:

None

Performance Reporter

Generate comprehensive performance reports in multiple formats.

class kailash.visualization.reports.WorkflowPerformanceReporter(task_manager: TaskManager, config: ReportConfig | None = None)[source]

Bases: object

Comprehensive workflow performance report generator.

This class provides detailed performance analysis and reporting capabilities for workflow executions, including insights, recommendations, and comparative analysis across multiple runs.

Usage:

reporter = WorkflowPerformanceReporter(task_manager) report = reporter.generate_report(run_id, output_path=”report.html”)

Parameters:

  • task_manager (TaskManager)

  • config (ReportConfig | None)

__init__(task_manager: TaskManager, config: ReportConfig | None = None)[source]

Initialize performance reporter.

Parameters:

  • task_manager (TaskManager) – TaskManager instance for data access

  • config (ReportConfig | None) – Report configuration options

generate_report(run_id: str, output_path: str | Path | None = None, format: ReportFormat = ReportFormat.HTML, compare_runs: list[str] | None = None) → Path[source]

Generate comprehensive performance report.

Parameters:

  • run_id (str) – Workflow run to analyze

  • output_path (str | Path | None) – Path to save report file

  • format (ReportFormat) – Output format for the report

  • compare_runs (list[str] | None) – List of run IDs to compare against

Returns:

Path to generated report file

Return type:

Path

Report Formats

Supported output formats for performance reports.

class kailash.visualization.reports.ReportFormat(value)[source]

Bases: Enum

Supported report output formats.

HTML = 'html'
MARKDOWN = 'markdown'
JSON = 'json'
PDF = 'pdf'
classmethod __contains__(member)

Return True if member is a member of this enum raises TypeError if member is not an enum member

note: in 3.12 TypeError will no longer be raised, and True will also be returned if member is the value of a member in this enum

classmethod __getitem__(name)

Return the member matching name.

classmethod __iter__()

Return members in definition order.

classmethod __len__()

Return the number of members (no aliases)

Performance Insights

Structured performance analysis and recommendations.

class kailash.visualization.reports.PerformanceInsight(category: str, severity: str, title: str, description: str, recommendation: str, metrics: dict[str, ~typing.Any]=<factory>)[source]

Bases: object

Container for performance insights and recommendations.

Variables:

  • category (str) – Type of insight (‘bottleneck’, ‘optimization’, ‘warning’)

  • severity (str) – Severity level (‘low’, ‘medium’, ‘high’, ‘critical’)

  • title (str) – Brief insight title

  • description (str) – Detailed description

  • recommendation (str) – Actionable recommendation

  • metrics (dict[str, Any]) – Supporting metrics data

Parameters:
category: str
severity: str
title: str
description: str
recommendation: str
metrics: dict[str, Any]
__init__(category: str, severity: str, title: str, description: str, recommendation: str, metrics: dict[str, ~typing.Any]=<factory>) → None
Parameters:
Return type:

None

Dashboard API

REST API interface for accessing metrics programmatically.

class kailash.visualization.api.SimpleDashboardAPI(task_manager: TaskManager, dashboard_config: DashboardConfig | None = None)[source]

Bases: object

Simplified API interface for dashboard functionality without FastAPI.

This class provides dashboard API functionality using standard Python libraries for environments where FastAPI is not available or desired.

Parameters:
__init__(task_manager: TaskManager, dashboard_config: DashboardConfig | None = None)[source]

Initialize simple API interface.

Parameters:

  • task_manager (TaskManager) – TaskManager instance for data access

  • dashboard_config (DashboardConfig | None) – Configuration for dashboard components

get_runs(limit: int = 10, offset: int = 0) → list[dict[str, Any]][source]

Get list of workflow runs.

Parameters:
Return type:

list[dict[str, Any]]

get_run_details(run_id: str) → dict[str, Any] | None[source]

Get details for a specific run.

Parameters:

run_id (str)

Return type:

dict[str, Any] | None

start_monitoring(run_id: str | None = None) → dict[str, Any][source]

Start real-time monitoring.

Parameters:

run_id (str | None)

Return type:

dict[str, Any]

stop_monitoring() → dict[str, Any][source]

Stop real-time monitoring.

Return type:

dict[str, Any]

get_current_metrics() → dict[str, Any] | None[source]

Get current live metrics.

Return type:

dict[str, Any] | None

get_metrics_history(minutes: int = 30) → list[dict[str, Any]][source]

Get metrics history.

Parameters:

minutes (int)

Return type:

list[dict[str, Any]]

generate_report(run_id: str, format: str = 'html', output_path: str | Path | None = None, compare_runs: list[str] | None = None) → Path[source]

Generate performance report.

Parameters:
Return type:

Path

generate_dashboard(output_path: str | Path | None = None) → Path[source]

Generate live dashboard HTML.

Parameters:

output_path (str | Path | None)

Return type:

Path

export_metrics_json(output_path: str | Path | None = None) → Path[source]

Export current metrics as JSON.

Parameters:

output_path (str | Path | None)

Return type:

Path

WebSocket Server

FastAPI-based server for real-time metrics streaming.

Note

This component requires FastAPI to be installed. Install with: pip install fastapi uvicorn

class kailash.visualization.api.DashboardAPIServer(task_manager: TaskManager, dashboard_config: DashboardConfig | None = None, cors_origins: list[str] | None = None, require_auth: bool = True, auth_config: Any = None, external_auth_reason: str | None = None, auth_exempt_paths: list[str] | None = None)[source]

Bases: object

FastAPI server for dashboard API endpoints.

This class provides a complete REST API server for accessing real-time workflow performance data and dashboard components.

Authentication fails CLOSED (#2112): require_auth defaults to True and construction raises ServerAuthNotConfiguredError unless a credential source is configured.

Usage:

# export KAILASH_JWT_SECRET=<at least 32 bytes> api_server = DashboardAPIServer(task_manager) api_server.start_server(host=”0.0.0.0”, port=8000)

# Or, to run without authentication – an explicit opt-out that logs a # loud WARN naming the exposure: api_server = DashboardAPIServer(task_manager, require_auth=False)

Parameters:
__init__(task_manager: TaskManager, dashboard_config: DashboardConfig | None = None, cors_origins: list[str] | None = None, require_auth: bool = True, auth_config: Any = None, external_auth_reason: str | None = None, auth_exempt_paths: list[str] | None = None)[source]

Initialize API server.

Parameters:

  • task_manager (TaskManager) – TaskManager instance for data access

  • dashboard_config (DashboardConfig | None) – Configuration for dashboard components

  • cors_origins (list[str] | None) – Allowed CORS origins. Defaults to [] (no cross-origin browser access), which is what this server has always used; supplying a list is what makes the CORS layer actually usable from a dashboard front-end.

  • require_auth (bool) –

    Whether every request must be authenticated. Defaults to ``True`` (fail-closed) and this is a BREAKING change – see kailash.utils.server_auth. This class builds its own FastAPI application and ships its own start_server() under uvicorn, so it is a server in its own right: reachable without WorkflowServer, without create_gateway(), and untouched by the six surfaces PR #2100 closed. Un-gated it served GET /api/v1/runs, GET /api/v1/runs/{id}/tasks, report downloads and two websocket streams to anonymous callers – run history and per-run task breakdowns describing what the system runs and when. There is no execute route here, so this is anonymous DISCLOSURE rather than anonymous code execution.

    Construction RAISES ServerAuthNotConfiguredError when no credential source is configured, rather than serving that surface openly. Set KAILASH_JWT_SECRET (or KAILASH_API_KEY_<NAME>, or pass auth_config=) to configure one. Pass require_auth=False to run without authentication – an explicit opt-out that logs a loud WARN.

  • auth_config (Any) – Explicit JWTConfig (or a dict of its fields) to authenticate with, bypassing environment lookup.

  • external_auth_reason (str | None) – Non-empty string declaring that an ASGI middleware OUTSIDE this server authenticates every request, so this server installs none. A blank string is rejected – a reason that names nothing is an undocumented hole.

  • auth_exempt_paths (list[str] | None) – Extra paths exempt from authentication, on top of the health-probe defaults. /docs, /openapi.json and the /api/v1/* routes are NOT exempt by default; each describes or exposes the protected surface.

Raises:

  • ImportError – FastAPI is not installed.

  • ServerAuthNotConfiguredError – require_auth=True and no credential source is configured.

start_server(host: str = '127.0.0.1', port: int = 8000, **kwargs)[source]

Start the API server.

Parameters:

  • host (str) – Host to bind to

  • port (int) – Port to bind to

  • **kwargs – Additional uvicorn server options

Performance Visualizer

Static performance analysis and chart generation.

class kailash.visualization.performance.PerformanceVisualizer(task_manager: TaskManager)[source]

Bases: object

Creates performance reports from task execution metrics.

Generates Markdown with tables and Mermaid bar charts — renders natively in GitHub, VS Code, JetBrains, and any Markdown viewer.

Parameters:

task_manager (TaskManager)

__init__(task_manager: TaskManager)[source]
Parameters:

task_manager (TaskManager)

create_run_performance_summary(run_id: str, output_dir: Path | None = None) → dict[str, Path][source]
Parameters:

  • run_id (str)

  • output_dir (Path | None)

Return type:

dict[str, Path]

compare_runs(run_ids: list[str], output_path: Path | None = None) → Path[source]
Parameters:
Return type:

Path

Usage Examples

Basic Real-time Monitoring

from kailash.visualization.dashboard import RealTimeDashboard, DashboardConfig
from kailash.tracking import TaskManager
from kailash.runtime.local import LocalRuntime

# Setup components
task_manager = TaskManager()
config = DashboardConfig(
    update_interval=1.0,
    max_history_points=100,
    auto_refresh=True,
    theme="light"
)

# Create dashboard
dashboard = RealTimeDashboard(task_manager, config)

# Start monitoring
dashboard.start_monitoring()

# Execute workflow with monitoring
with LocalRuntime() as runtime:
    results, run_id = runtime.execute(workflow, task_manager)

# Generate live dashboard
dashboard.generate_live_report("dashboard.html", include_charts=True)
dashboard.stop_monitoring()

Performance Report Generation

from kailash.visualization.reports import WorkflowPerformanceReporter, ReportFormat

# Create reporter
reporter = WorkflowPerformanceReporter(task_manager)

# Generate comprehensive HTML report
report_path = reporter.generate_report(
    run_id,
    output_path="performance_report.html",
    format=ReportFormat.HTML,
    compare_runs=[previous_run_id]
)

# Generate Markdown report
md_report = reporter.generate_report(
    run_id,
    format=ReportFormat.MARKDOWN
)

API-based Monitoring

from kailash.visualization.api import SimpleDashboardAPI

# Create API interface
api = SimpleDashboardAPI(task_manager)
api.start_monitoring()

# Get current metrics
metrics = api.get_current_metrics()
print(f"Active tasks: {metrics['active_tasks']}")

# Get historical data
history = api.get_metrics_history(minutes=30)

# Stop monitoring
api.stop_monitoring()

WebSocket Streaming Server

from kailash.visualization.api import DashboardAPIServer
import asyncio

# Create server
server = DashboardAPIServer(task_manager, port=8000)

# Start server (runs async)
async def run_server():
    await server.start()

# In your client (JavaScript):
# const ws = new WebSocket('ws://localhost:8000/api/v1/metrics/stream');
# ws.onmessage = (event) => {
#     const metrics = JSON.parse(event.data);
#     // Update dashboard with real-time metrics
# };

Real-time Callbacks

# Add custom callbacks for real-time events
def on_metrics_update(metrics):
    print(f"CPU: {metrics.total_cpu_usage:.1f}%, Memory: {metrics.total_memory_usage:.1f}MB")

def on_status_change(event_type, count):
    if event_type == "task_completed":
        print(f"✅ {count} task(s) completed")
    elif event_type == "task_failed":
        print(f"❌ {count} task(s) failed")

dashboard.add_metrics_callback(on_metrics_update)
dashboard.add_status_callback(on_status_change)

Dashboard Features

The real-time dashboard provides:

Live Metrics

  • Active, completed, and failed task counts

  • Real-time CPU and memory usage

  • Throughput metrics (tasks per minute)

  • I/O statistics and data transfer rates

Interactive Charts

  • Timeline charts with Chart.js integration

  • Resource usage graphs over time

  • Task status progression visualization

  • Performance comparison charts

Responsive Design

  • Mobile-friendly layout

  • Auto-refresh capabilities

  • Dark/light theme support

  • Customizable update intervals

Export Options

  • HTML dashboards with embedded JavaScript

  • JSON metrics logs for external analysis

  • Markdown reports for documentation

  • PNG/SVG chart exports (with matplotlib)

Architecture

The visualization system follows a modular architecture:

        graph TB
    subgraph "Real-time Layer"
        A[RealTimeDashboard]
        B[LiveMetrics]
        C[DashboardConfig]
    end

    subgraph "Reporting Layer"
        D[WorkflowPerformanceReporter]
        E[PerformanceInsight]
        F[ReportFormat]
    end

    subgraph "API Layer"
        G[SimpleDashboardAPI]
        H[DashboardAPIServer]
        I[WebSocket Streaming]
    end

    subgraph "Static Analysis"
        J[PerformanceVisualizer]
        K[Chart Generation]
        L[Metrics Analysis]
    end

    A --> B
    A --> C
    A --> G

    D --> E
    D --> F

    G --> H
    H --> I

    J --> K
    J --> L

    A --> D
    G --> J

Best Practices

Real-time Monitoring

  • Use appropriate update intervals (1-5 seconds for active monitoring)

  • Limit history points to prevent memory issues (100-500 points)

  • Stop monitoring when done to free resources

  • Use callbacks for custom event handling

Performance Reports

  • Generate reports after workflow completion

  • Compare multiple runs to identify trends

  • Include relevant run metadata and context

  • Use appropriate output formats for your use case

API Integration

  • Use WebSocket streaming for real-time dashboards

  • Implement proper error handling and reconnection

  • Rate limit API calls to prevent performance impact

  • Cache metrics data for better performance

Resource Management

  • Monitor system resources during metrics collection

  • Use background threads to avoid blocking workflow execution

  • Implement proper cleanup and resource disposal

  • Consider storage requirements for long-running monitoring