flowstepper 0.1.0

DAG-based pipeline visualizer — log from any language, explore in the browser

Flowstepper

Visualize pipelines as interactive DAGs — from log files.

Works like TensorBoard: instrument your code with the SDK, run flowstepper, and explore your pipeline in a browser.

---

What is Flowstepper?

Flowstepper is an observability tool for pipelines. It lets you:

• Instrument your pipeline stages with a lightweight SDK (zero-blocking, production-safe)
• Visualize the DAG structure, node timings, and runtime metadata in a browser
• Debug slow stages, data drops, and execution errors across runs
• Compare different runs side-by-side to track regressions

The SDK writes structured .flowlog files. The viewer reads them — no database, no external service, no setup beyond pip install.

---

Quick Start

Install

pip install flowstepper

Instrument your pipeline

from flowstepper.sdk import BaseNode, Pipeline

# Optional: pick a colour per kind shown in the UI
COLORS = {
    "source": "#3b82f6",
    "ranker": "#22c55e",
    "filter": "#f59e0b",
}

with Pipeline("recommendation", log_dir="./logs", node_kind_colors=COLORS) as p:
    src  = p.add_node(BaseNode("catalog", kind="source"))
    pre  = p.add_node(BaseNode("scorer", kind="ranker"))
    rank = p.add_node(BaseNode("ltr_model", kind="ranker",
                               tags={"model_name": "xgb", "model_version": "v3"}))
    filt = p.add_node(BaseNode("dedup", kind="filter"))

    p.connect(src, pre)
    p.connect(pre, rank)
    p.connect(rank, filt)

    with p.node_scope(src):
        items = fetch_catalog()
        p.record(src, item_count=len(items))

    with p.node_scope(pre):
        scored = score(items)
        p.record(pre, scored_count=len(scored))

    with p.node_scope(rank):
        ranked = ltr_model.rank(scored)

    with p.node_scope(filt):
        results = dedup(ranked)
        p.record(filt, output_count=len(results))

Launch the viewer

flowstepper --logdir ./logs
# Open http://localhost:6006

---

Features

Feature	Description
Interactive DAG	Auto-laid-out graph using dagre; click nodes to inspect metadata
Timeline trace	Gantt-style execution timeline, similar to Chrome DevTools
Live reload	--reload flag tails new runs as they land
Zero-overhead sampling	sample_rate=0.0 makes all SDK calls pure no-ops
Production-safe	Logging failures never propagate; configurable drop policies
No external deps	Just files — no database, no broker, no cloud account
Language-agnostic format	Any language can write .flowlog files

---

SDK

Nodes

There is a single BaseNode class. Every node has a name, a free-form kind string (any label you choose — e.g. "retrieval", "ranker", "tool_call"), and optional tags:

from flowstepper.sdk import BaseNode

BaseNode("catalog", kind="source")
BaseNode("ltr", kind="ranker", tags={"model_name": "xgb", "model_version": "v3"})

Colours for each kind are supplied once at pipeline init via node_kind_colors={"source": "#3b82f6", ...}.

Pipeline API

# Add nodes
node = p.add_node(BaseNode("ranker", kind="ranker",
                           tags={"model_name": "xgb"}))

# Declare edges
p.connect(source_node, target_node)                        # stream (default)
p.connect(source_node, target_node, kind="batch")          # batch transfer
p.connect(signal_node, target_node, kind="signal")         # control signal

# Track execution
with p.node_scope(node) as scope:
    result = run_stage()
    p.record(node, latency_ms=12.4, output_size=len(result))
    if not result:
        scope.is_error = True  # soft-error flag

# Inspect
print(p.run_id)          # unique run identifier
print(p.is_sampled)      # whether this run is being logged
print(p.dropped_events)  # events dropped due to queue pressure

Logging Configuration

from flowstepper.sdk import Pipeline, LoggingConfig, DropPolicy

# Development — log everything
with Pipeline("my_pipeline", log_dir="./logs") as p:
    ...

# Production — 1% sample rate, bounded queue
cfg = LoggingConfig(
    sample_rate=0.01,
    queue_maxsize=10_000,
    drop_policy=DropPolicy.DROP_NEWEST,
    flush_interval_ms=200,
    flush_batch_size=256,
    on_drop=lambda n: metrics.increment("flowstepper.drops", n),
    on_error=lambda e: logger.warning("flowstepper error", exc_info=e),
)

with Pipeline("my_pipeline", log_dir="./logs", config=cfg) as p:
    ...

Built-in presets:

LoggingConfig.for_development()   # ALWAYS sample, sync flush
LoggingConfig.for_production()    # RATE sample, async flush, DROP_NEWEST
LoggingConfig.disabled()          # NEVER sample — all calls are no-ops

---

Architecture

User Code ──► SDK ──► async queue ──► Logger (background thread) ──► .flowlog files
                                                                             │
                                                                    LogParser (reader/)
                                                                             │
                                                                    FastAPI Server
                                                                             │
                                                                    React SPA (browser)

Key design principles:

• Hot-path zero-blocking — sampling decision made once at pipeline entry; all logging calls enqueue asynchronously
• Bounded memory — configurable queue size with drop policies (DROP_NEWEST, DROP_OLDEST, BLOCK)
• Error isolation — logging failures are never propagated to caller code
• Stateless reader — LogParser is a pure function: Path → PipelineRun
• Language-agnostic — log format is newline-delimited JSON; any language can produce logs

Log Format

Logs are stored as newline-delimited JSON at <log_dir>/<pipeline_name>/<run_id>.flowlog.

Eight event types: pipeline_start, pipeline_end, node_declared, edge_declared, node_start, node_end, node_error, metadata.

See Log Format Specification for the full schema.

---

Installation from Source

git clone https://github.com/pavelthei/flowstepper
cd flowstepper
pip install -e ".[dev]"

The editable install automatically builds the frontend bundle. To rebuild manually after frontend changes:

./flowstepper/scripts/build_frontend.sh

Running Tests

pytest tests/                              # all tests
pytest tests/sdk/test_logger.py -v         # single file
ruff check flowstepper tests               # lint
mypy flowstepper                           # type check

---

SDK Language Support

Language	Status
Python 3	Supported
Java	Planned
Go	Planned

---

Documentation

• Architecture — design rationale and component overview
• Log Format — event schema specification
• Python SDK — full SDK reference

---

License

MIT