Dagpipe

DagPipe

The reliability layer that makes AI workflows safe to ship: crash recovery, schema validation, and cost routing

---

NeurIPS 2025 research analyzing 1,642 real-world multi-agent execution traces

found a 41–86.7% failure rate across 7 state-of-the-art open-source systems.

The root cause: cascading error propagation, where one failed node corrupts all downstream nodes.

DagPipe makes cascade failure structurally impossible.

Every node's output is independently validated and checkpointed before the next node executes. A failure at node 4 cannot corrupt nodes 1, 2, or 3. Delete nothing. Just re-run. DagPipe resumes exactly where it stopped, automatically.

Pipeline: research → outline → draft → edit → publish

↑ crashed here

Re-run → research ✓ (restored) → outline ✓ (restored) → draft (re-runs) → ...

Zero infrastructure. Zero subscription. Runs entirely on free-tier APIs.

---

Install

pip install dagpipe-core

Requirements: Python 3.12+ · pydantic >= 2.0 · pyyaml · A free Groq API key (no credit card)

---

Three Ways to Use DagPipe

For developers: install the library and build crash-proof LLM pipelines in Python:

pip install dagpipe-core

For non-coders: describe your workflow in plain English, receive production-ready crash-proof pipeline code as a downloadable zip. No coding required: 👉 Pipeline Generator on Apify ($0.05/run)

For AI agents and IDE users: connect directly via MCP. Use DagPipe from Claude Desktop, Cursor, Windsurf, or any MCP-compatible client without writing any code: 👉 DagPipe Generator MCP on Smithery

The generator outputs DagPipe pipelines, as every generated zip already has crash recovery, schema validation, and cost routing built in by default. No other LLM pipeline framework ships this.

---

Why DagPipe?

---

What's New in v0.2.3

v0.2.3 adds the official MCP Registry metadata and identifier to the package repository, enabling one-click discovery on the official MCP Registry.

---

What's New in v0.2.2

v0.2.2 improves PyPI discoverability with optimized metadata, a clearer project description, and enhanced AI agent categorization.

---

What's New in v0.2.1

v0.2.1 brings crucial generator reliability fixes and a highly requested DX feature:

• `verbose=True` Output: Pass verbose=True to the PipelineOrchestrator to get real-time, per-node CLI progress updates with execution times, node descriptions, and running costs.
• Generator Core Fixes: The MCP and Apify generator prompts now strictly enforce the model=None positional argument, guarantee exact YAML topological dependency key lookups (DEPENDENCY_NODE_ID), and support native .csv, .xml, .html, and .md save nodes out of the box.

---

What's New in v0.2.0

v0.2.0 was a significant architectural upgrade. All features are opt-in and fully backwards compatible, meaning existing v0.1.x pipelines run without changes.

Smart Model Router

The router automatically selects which AI model to use based on task complexity. Simple tasks go to fast, free models. Hard tasks escalate to more capable models. If a model fails or rate-limits, it tries the next one, automatically.

router = ModelRouter(

low_complexity_fn=gemini_flash, low_label="gemini_flash", high_complexity_fn=llama33_70b, high_label="llama_70b", fallback_fn=gemini_flash, fallback_label="fallback", complexity_threshold=0.6, )

Live Model Registry

A self-maintaining database of free-tier model availability and pricing. Refreshes every 24 hours automatically. If a model goes offline, the registry marks it unavailable and the router stops routing to it, without any manual intervention.

Semantic Output Assertions

Beyond schema validation, nodes can declare semantic rules. If the LLM returns structurally valid but logically wrong data, the assertion catches it and forces self-correction:

DAGNode(

id="revenue", fn_name="calculate_revenue", assert_fn=lambda out: 0 < out.get("revenue", 0) < 1e12, assert_message="Revenue must be a realistic positive number", )

Context Isolation

Nodes only receive outputs from their declared dependencies. A node processing untrusted web data cannot access credentials or outputs from unrelated pipeline branches.

Dead Letter Queue

When a node exhausts all retries, DagPipe saves the complete failure context to .dagpipe/checkpoints/{node_id}.failed.json, including what was passed in, what error occurred, and instructions for manual correction. No failure context is ever lost.

Pluggable Checkpoint Backends

Swap out filesystem storage for Redis, S3, or any custom backend by implementing the CheckpointStorage protocol.

---

Quickstart

from pathlib import Path

from dagpipe.dag import PipelineOrchestrator, DAGNode, FilesystemCheckpoint from dagpipe.router import ModelRouter from dagpipe.constrained import constrained_generate

1. Define your node functions

def research(context, model): prompt = [{"role": "user", "content": f"Research: {context['topic']}"}] return {"summary": model(prompt)}

def write_draft(context, model): summary = context"research" prompt = [{"role": "user", "content": f"Write an article based on: {summary}"}] return {"draft": model(prompt)}

def publish(context, model): print(f"Publishing: {context'write_draft'[:100]}...") return {"status": "published"}

2. Wire your LLM providers (any Python callable works)

import groq from google.genai import Client

groq_client = groq.Groq() def llama33_70b(messages): return groq_client.chat.completions.create( model="llama-3.3-70b-versatile", messages=messages ).choices[0].message.content

gemini_client = Client() def gemini_flash(messages): return gemini_client.models.generate_content( model="gemini-2.5-flash", contents=messages0 ).text

3. Build the router

router = ModelRouter( low_complexity_fn=gemini_flash, low_label="gemini_flash", high_complexity_fn=llama33_70b, high_label="llama_70b", fallback_fn=gemini_flash, fallback_label="fallback", complexity_threshold=0.6, )

4. Define the DAG

nodes = [ DAGNode(id="research", fn_name="research", complexity=0.4), DAGNode(id="write_draft", fn_name="write_draft", complexity=0.7, depends_on=["research"]), DAGNode(id="publish", fn_name="publish", depends_on=["write_draft"], is_deterministic=True), ]

5. Run it

orchestrator = PipelineOrchestrator( nodes=nodes, node_registry={ "research": research, "write_draft": write_draft, "publish": publish, }, router=router, checkpoint_backend=FilesystemCheckpoint(Path(".dagpipe/checkpoints")), max_retries=3, on_node_complete=lambda node_id, result, duration: print(f" ✓ {node_id} ({duration:.1f}s)"), )

result = orchestrator.run(initial_state={"topic": "AI in African fintech"})

Crash mid-run? Delete nothing. Just re-run. DagPipe reads the checkpoints and skips completed nodes automatically.

---

How It Works

Your Tasks (YAML or Python list of DAGNodes)

│ ▼ ┌──────────────────┐ │ Topological │ resolves execution order, │ Sort │ detects cycles before running └────────┬─────────┘ │ ┌─────────▼──────────┐ │ Checkpoint │ restores any completed nodes │ Restore │ from previous runs └─────────┬──────────┘ │ ┌───────▼────────┐ │ For each node │◄─────────────────────────┐ └───────┬────────┘ │ │ │ ┌─────────▼──────────┐ ┌────────────────┐ │ │ ModelRouter │───▶│ low / high / │ │ │ (complexity score)│ │ fallback fn │ │ └─────────┬──────────┘ └────────────────┘ │ │ │ ┌─────────▼──────────┐ │ │ Constrained │ forces valid output │ │ Generator │ retries on failure │ └─────────┬──────────┘ │ │ │ ┌─────────▼──────────┐ │ │ Checkpoint Save │ writes result to disk │ └─────────┬──────────┘ │ │ │ crash here = resume from ✓ next node

---

📦 Core Modules

dagpipe.dag: The Orchestrator

The central engine. Loads a DAG from a Python list or YAML file, sorts nodes by dependency, and executes them with checkpointing and retry.

from dagpipe.dag import PipelineOrchestrator, DAGNode, load_dag

Load from YAML

nodes = load_dag(Path("my_pipeline.yaml"))

Or define in Python

nodes = [DAGNode(id="step_a", fn_name="fn_a", complexity=0.3)]

dagpipe.checkpoints: Crash Recovery

Saves node output to disk after every successful execution. On resume, completed nodes are skipped entirely.

New in v0.2.0: the CheckpointStorage Protocol, allowing you to swap the default filesystem backend for Redis, S3, or any custom store:

from dagpipe.dag import C