hypernodes 0.4.8

Hierarchical, Modular Data Pipelines for AI/ML

Core Pattern • Caching • Daft • Docs • Install

Why HyperNodes?

Data pipelines are run more than written — you execute the same code on the same data hundreds of times during development. Every run costs time and money.

HyperNodes solves this with two core ideas:

✨ The Three-Step Pattern

1️⃣ Think Singular

Write functions that process one item. No batch loops, no complexity — just clean, testable code.

from hypernodes import Pipeline, node

@node(output_name="cleaned")
def clean(text: str) -> str:
    return text.strip().lower()

@node(output_name="word_count")
def count(cleaned: str) -> int:
    return len(cleaned.split())

# Create a named pipeline
text_processor = Pipeline(nodes=[clean, count], name="text_processor")

result = text_processor.run(inputs={"text": "  Hello World  "})
# {'cleaned': 'hello world', 'word_count': 2}

The text_processor pipeline

2️⃣ Compose

Pipelines are nodes. Use text_processor inside a larger workflow — it appears as a single unit.

@node(output_name="report")
def summarize(word_count: int, cleaned: str) -> str:
    return f"Processed: {word_count} words"

# text_processor becomes a node in the analysis pipeline
analysis = Pipeline(nodes=[text_processor.as_node(), summarize], name="analysis")

Collapsed view — text_processor as a single node

Expand to see the internals:

Expanded view — see inside text_processor

3️⃣ Scale

Process many items with .map(). The same single-item logic runs over a collection.

run() vs map() at a glance

Single-item execution versus per-item fan-out.

# Process multiple texts
results = text_processor.map(
    inputs={"text": ["Hello", "World", "Foo Bar"]},
    map_over="text"
)
# [{'cleaned': 'hello', 'word_count': 1}, 
#  {'cleaned': 'world', 'word_count': 1},
#  {'cleaned': 'foo bar', 'word_count': 2}]

Or wrap it as a batch processor in a larger pipeline:

batch_processor = text_processor.as_node(
    name="batch_text_processor",
    map_over="texts",
    input_mapping={"texts": "text"},
    output_mapping={"word_count": "word_counts"}
)

Batch processing with map_over

📦 Smart Caching

HyperNodes implements a content-addressable cache that keeps runs fast and correct.

How it works

It hashes:

• Function signature (source code, closures, defaults)
• Input values (per item in .map())
• Dependencies (upstream node signatures)

Only nodes whose hashes changed will execute.

Enable caching in 3 steps

1. Pick a backend – DiskCache ships by default; swap in Redis/S3 later.
2. Attach it to an engine – SeqEngine(cache=DiskCache(path=".cache")).
3. Run pipelines as usual – pipeline.run(...) or pipeline.map(...) automatically uses the cache.

from hypernodes import Pipeline, DiskCache, SeqEngine

engine = SeqEngine(cache=DiskCache(path=".cache"))
pipeline = Pipeline(nodes=[clean, count], engine=engine)
pipeline.run(inputs={"text": "hello"})  # warm cache

Incremental map runs (per-item signatures)

Caching shines when datasets change gradually. Each item inside .map() gets its own signature, so adding a single record does not blow away the cache for the entire batch.

from hypernodes import Pipeline, SeqEngine, DiskCache, node

@node(output_name="features")
def featurize(text: str) -> dict[str, int]:
    print(f"computing -> {text}")
    return {"len": len(text)}

engine = SeqEngine(cache=DiskCache(path=".cache"))
featurizer = Pipeline(nodes=[featurize], engine=engine)

# 1) Initial batch – every item is a miss
featurizer.map(inputs={"text": ["alpha", "beta"]}, map_over="text")

# 2) Append one more record – only "delta" recomputes
featurizer.map(inputs={"text": ["alpha", "beta", "delta"]}, map_over="text")

Tip: Visualize before long runs. Cached (grey) vs running (colored) nodes stand out in the HTML/SVG artifact.

featurizer.visualize(
    filename="outputs/cache_walkthrough.svg",
    engine="graphviz",
    separate_outputs=True,
    depth=None,
)

Node-level reuse across complex DAGs

Because caching happens per node, you can pause/resume multi-stage DAGs without re-running upstream work. Here’s a more realistic extraction → embedding pipeline:

@node(output_name="cleaned")
def clean(text: str) -> str:
    return text.strip().lower()

@node(output_name="embedding")
def embed(cleaned: str, model: str) -> list[float]:
    return encoder[model](cleaned)  # Expensive!

@node(output_name="doc")
def package(cleaned: str, embedding: list[float]) -> dict:
    return {"text": cleaned, "vector": embedding}

pipe = Pipeline(nodes=[clean, embed, package], engine=SeqEngine(cache=DiskCache(".cache")))
pipe.run(inputs={"text": "Hello", "model": "v1"})

# Change only the final step; upstream nodes stay cached
@node(output_name="doc")
def package(cleaned: str, embedding: list[float]) -> dict:
    return {"text": cleaned, "vector": embedding, "v": 2}

pipe.run(inputs={"text": "Hello", "model": "v1"})
# clean() and embed() hit cache, only package() reruns

Use pipeline.engine.cache.inspect(signature) (or custom cache utilities) if you ever want to programmatically verify whether a node is cached before triggering a heavy run.

Code-aware invalidations (no stale runs)

The cache is just as smart about code changes. HyperNodes hashes the function body and anything it touches (imports, closures, default arguments). If a helper changes, the dependent node automatically invalidates.

# helpers.py
def normalize(text):
    return text.lower()

@node(output_name="clean")
def clean(text: str) -> str:
    return normalize(text)

pipe = Pipeline(nodes=[clean], engine=SeqEngine(cache=DiskCache(".cache")))
pipe.run(inputs={"text": "Hello"})

# Change helpers.py -> normalize()
def normalize(text):
    return text.lower().strip()

pipe.run(inputs={"text": "Hello"})  # cache miss – dependency changed!

Default arguments, environment variables, and closure state are also part of the hash.

Need to know why a node reran? Enable LOG_HYPERNODES_CACHE=1 (env var) to log cache hits/misses, or plug in a callback that records on_node_cached events for your observability system.

🚀 Scale with Daft

When you're ready for distributed execution, plug in Daft — same pipeline code, now runs on a cluster.

from hypernodes.engines import DaftEngine

engine = DaftEngine(use_batch_udf=True)
pipeline = Pipeline(nodes=[clean, count], engine=engine)

# Auto-batched, distributed, with per-item caching
results = pipeline.map(inputs={"text": texts * 10000}, map_over="text")

Daft provides:

• Distributed execution — laptop to cluster, no code changes
• Auto-batching — optimal batch sizes computed automatically
• Per-item caching — even in distributed mode, each item is cached independently

💡 Inspiration

HyperNodes stands on the shoulders of giants. It was inspired by and grew from working with:

• Pipefunc: For its elegant approach to function composition.
• Apache Hamilton: For its paradigm of defining dataflows using standard Python functions.

HyperNodes aims to bring native support for hierarchical pipelines and advanced caching to this ecosystem.

📚 Documentation

The full documentation is available in the docs/ directory:

• Introduction & Quick Start
• Core Concepts
  • Nodes
  • Pipelines
  • Execution
• Data Processing
  • Mapping (Parallel Processing)
  • Nesting Pipelines
  • Caching
• Scaling & Optimization
  • Engines Overview
  • Distributed Daft Engine
  • Stateful Parameters
• Observability
  • Visualization
  • Telemetry & Progress

🚀 Installation

pip install hypernodes
# or with uv
uv add hypernodes

📄 License

MIT License - see LICENSE file for details.