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.
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.
pip install flowstepperfrom 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))flowstepper --logdir ./logs
# Open http://localhost:6006| 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 |
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", ...}.
# 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 pressurefrom 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-opsUser 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 —
LogParseris a pure function:Path → PipelineRun - Language-agnostic — log format is newline-delimited JSON; any language can produce logs
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.
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.shpytest tests/ # all tests
pytest tests/sdk/test_logger.py -v # single file
ruff check flowstepper tests # lint
mypy flowstepper # type check| Language | Status |
|---|---|
| Python 3 | Supported |
| Java | Planned |
| Go | Planned |
- Architecture — design rationale and component overview
- Log Format — event schema specification
- Python SDK — full SDK reference