Lightweight research project pipeline framework with DAG tracing
Project description
Research Pipelines
A lightweight Python framework for tracing the DAG (directed acyclic graph) of research experiments. Automatically track datasets, models, and evaluations arguments and function-dependencies, then persist everything to wandb or local storage. This is especially useful for plotting of further evaluation of a trained model, as we can recreate the a function call or just the arguments of a traced function. By design, it is a pickle-free solution that relies on recording primitve arguments.
Just decorate function during training like:
@evaluation()
def evaluate(model_obj, metric: str):
return {"score": 0.95}
It turns a huge, messy notebook into something simple like:
# (select a traced run and load its saved configurations)
# rebuild the arguments such that we can call evaluate ourselves
model_obj, metric = query.build_arguments(
evaluate
)
# load saved weights
model_obj.load_state_dict(state_dict)
# call evaluate
evaluate(model_obj, metric)
# do some plotting
Features
- Automatic DAG Tracing: Decorators automatically detect when traced objects are used as dependencies
- Configuration Persistence: Basic types (str, int, float, bool, None) are automatically captured and stored
- Flexible Rebuilding: The query backend allows for calling the traced functions again, even if they depend on other traced functions
- Pluggable Backends: Use PickleBackend for testing or WandBBackend for production wandb integration
- Zero Boilerplate: Apply decorators and your functions/classes are automatically traced
- Recursive Dependency Resolution: Full transitive closure of all dependencies
Installation (Dev)
# Clone or create the project
cd research_pipelines
# Create conda environment
conda create -n research_pipelines python=3.11
# Activate environment
conda activate research_pipelines
# Install package in editable mode
pip install -e .
# Optional: Install the Torch example extra
pip install ".[example]"
# Optional: Install wandb backend
pip install ".[wandb]"
Quick Start
from research_pipelines.decorators import dataset, model, evaluation
from research_pipelines.dag import build_dag
# Decorate your functions
@dataset()
def load_data(path: str, split: str):
# Load your data...
return {"data": [...], "metadata": {...}}
@model()
def train_model(train_data, architecture: str, lr: float):
# Non-basic args (train_data) become dependencies
# Basic args (architecture, lr) become config
return trained_model
@evaluation()
def evaluate(model_obj, metric: str):
return {"score": 0.95}
# Execute your pipeline
data = load_data(path="/data/train.csv", split="train")
model = train_model(train_data=data, architecture="bert", lr=0.001)
results = evaluate(model_obj=model, metric="accuracy")
# Print the DAG
dag = build_dag()
for obj_id, obj in dag.items():
print(f"{obj['type']}: {obj['config']}, depends on: {obj['dependencies']}")
Rebuild the traced object
The traced objects are not pickled, instead the arguments the functions are called with are saved.
import research_pipelines.query as query
# we can now easily call the functions with the recorded arguments via build()
dataset = query.build(
load_data
)
# or just get the arguments such that we can call it ourselves
model_obj, metric = query.build_arguments(
evaluate
)
model_obj.load_state_dict(state_dict)
evaluate(model_obj, metric)
How It Works
1. Decoration
Apply @dataset(), @model(), @evaluation(), or generic @traced(traced_type="...") to your functions or class constructors:
@dataset()
def load_data(path: str, split: str):
return load_from_disk(path)
@model()
class MyModel:
def __init__(self, layers: int, dataset_input):
self.layers = layers
self.data = dataset_input
2. Automatic Tracing
When you call a decorated function/constructor:
- Arguments are classified:
- Basic types (str, int, float, bool, None): stored as configuration
- Traced objects (returned from other @traced functions): become dependencies
- Other types: ignored (can be supplied manually later)
- Unique ID is generated for this object
- Configuration (basic args + type) is persisted to backend
- Dependencies (other traced object IDs) are recorded
3. DAG Structure
The framework automatically builds a DAG:
dataset_1 (config: path="/data/train.csv", split="train")
↓
model_1 (config: architecture="bert", lr=0.001, depends_on: [dataset_1])
↓
eval_1 (config: metric="accuracy", depends_on: [model_1])
4. Backend Persistence
Choose a backend to persist configurations:
PickleBackend (default for testing):
from research_pipelines.backends.pickle_backend import PickleBackend
from research_pipelines.backends.manager import set_backend
backend = PickleBackend(directory=".traced_configs")
set_backend(backend)
WandBBackend (for wandb integration):
import wandb
from research_pipelines.backends.wandb_backend import WandBBackend
from research_pipelines.backends.manager import set_backend
wandb.init(project="my_project")
backend = WandBBackend()
set_backend(backend)
# Configs are automatically logged to wandb.run.config
API Reference
Decorators
from research_pipelines.decorators import dataset, model, evaluation, traced
@dataset()
def load_data(...):
"""Traces a dataset creation function/class."""
pass
@model()
def train(...):
"""Traces a model creation function/class."""
pass
@evaluation()
def eval(...):
"""Traces an evaluation function/class."""
pass
@traced(traced_type="custom")
def my_function(...):
"""Generic tracer with custom type."""
pass
DAG Operations
from research_pipelines.dag import (
build_dag,
get_dependencies_recursive,
detect_circular_dependencies,
export_dag,
get_root_objects,
get_leaf_objects,
get_objects_by_type,
get_dependents,
)
# Build full DAG
dag = build_dag()
# Get all transitive dependencies
deps = get_dependencies_recursive(object_id)
# Check for cycles
has_cycles = detect_circular_dependencies()
# Export for serialization
dag_export = export_dag()
# Find roots (datasets with no dependencies)
roots = get_root_objects()
# Find leaves (objects nothing depends on)
leaves = get_leaf_objects()
# Filter by type
datasets = get_objects_by_type("dataset")
models = get_objects_by_type("model")
# Find what depends on an object
dependents = get_dependents(object_id)
Backends
from research_pipelines.backends.manager import get_backend, set_backend
# Get active backend
backend = get_backend()
# Set custom backend
set_backend(my_backend)
# Backend interface
class Backend(ABC):
def log_config(object_id, config_dict, dependencies):
"""Persist config for an object."""
pass
def get_config(object_id):
"""Retrieve config for an object."""
pass
def load_all():
"""Load all configs."""
pass
def clear():
"""Clear all configs."""
pass
Configuration Format
Configurations are stored as dictionaries with the following structure:
{
"object_id_1": {
"callable": "examples.simple_pipeline:load_dataset"
"config": {
"path": "/data/train.csv",
"split": "train",
"batch_size": 32,
},
"dependencies": [],
},
"object_id_2": {
"callable": "examples.simple_pipeline:create_model"
"config": {
"architecture": "bert",
"learning_rate": 0.001,
},
"dependencies": ["object_id_1"],
},
}
When using WandBBackend, this is stored directly in wandb.run.config.
Examples
See examples/simple_pipeline.py for a complete end-to-end example.
Run it:
conda activate research_pipelines
python examples/simple_pipeline.py
Testing
All tests use PickleBackend and are fully isolated:
conda activate research_pipelines
pytest tests/ -v
Development
The framework is organized into modules:
src/research_pipelines/core.py- Core tracing logicsrc/research_pipelines/decorators.py- @dataset, @model, @evaluation decoratorssrc/research_pipelines/backends/- Backend implementationsbase.py- Abstract Backend interfacepickle_backend.py- PickleBackend (testing)wandb_backend.py- WandBBackend (wandb integration)manager.py- Global backend management
src/research_pipelines/dag.py- DAG utilitiestests/- Test suite (61 tests, all passing)
Key Design Decisions
- Lazy Imports: wandb is only imported when WandBBackend is used
- Automatic Dependency Detection: Uses Python's
id()to track object identity - In-Memory Registry: Separate from backend storage, enables DAG operations
- UUID v4 IDs: Unique, collision-free object identifiers
- Type-Based Filtering: Basic types automatically detected and persisted
- Pluggable Backends: Easy to add custom storage implementations
Limitations & Future Work
- No support for custom object serialization (by design)
- No execution timing/profiling (configuration-only tracking)
- No automatic versioning/hashing of objects
License
MIT
Project details
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file research_pipelines-0.1.0.tar.gz.
File metadata
- Download URL: research_pipelines-0.1.0.tar.gz
- Upload date:
- Size: 29.6 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.2.0 CPython/3.13.9
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
8c10657a85aaff3df2bd4fb7ec8e7d1e72fb4fd2a0729ff05880ab30a025f176
|
|
| MD5 |
1a168f520f685af0e04af9ff4a6e469b
|
|
| BLAKE2b-256 |
8f2810bcb5d56fa1976ea333ea9f7a15ca5e27a0c5dc4577e2bb5be5174056a7
|
File details
Details for the file research_pipelines-0.1.0-py3-none-any.whl.
File metadata
- Download URL: research_pipelines-0.1.0-py3-none-any.whl
- Upload date:
- Size: 23.3 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.2.0 CPython/3.13.9
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
cb4a06dd2870c90d6c09ab776a83e2a70f63e409026d58f7479699f7123f6f3e
|
|
| MD5 |
9d94f3353381d7d40c31bffe41c3162a
|
|
| BLAKE2b-256 |
85bd7f7076bd4517ea71879dd8a14ec13b9271cbf4a66b4112dbff4f905db898
|