dagpipe

DagPipe

让 AI 工作流可靠上线的保障层：故障恢复、模式验证和成本路由

NeurIPS 2025 的研究分析了 1,642 条真实世界的多 Agent 执行轨迹， 发现 7 个最先进的开源系统中，故障率高达 41–86.7%。 根本原因：错误级联传播，一个节点发生故障就会破坏所有下游节点。

DagPipe 从结构上杜绝级联故障。

每个节点的输出都会在下一个节点执行前独立验证并保存检查点。节点 4 发生故障，不会破坏节点 1、2 或 3。无需删除任何内容，只需重新运行。DagPipe 会自动从中断处准确恢复。

Pipeline: research → outline → draft → edit → publish
                                  ↑
                            crashed here

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

零基础设施。零订阅费用。完全通过免费层级的 API 运行。

安装

pip install dagpipe-core

要求： Python 3.12+ · pydantic >= 2.0 · pyyaml · 一个免费的 Groq API 密钥（无需信用卡）

三种使用 DagPipe 的方式

面向开发者：安装库，在 Python 中构建具备故障恢复能力的 LLM 流水线：

pip install dagpipe-core

面向非编程人员：用通俗语言描述工作流，即可获得可下载的 zip 文件，其中包含可用于生产环境、具备故障恢复能力的流水线代码。无需编写代码： 👉 Apify 上的流水线生成器（$0.05/次运行）

面向 AI Agent 和 IDE 用户：通过 MCP 直接连接。无需编写代码，即可在 Claude Desktop、Cursor、Windsurf 或任何兼容 MCP 的客户端中使用 DagPipe： 👉 Smithery 上的 DagPipe Generator MCP

生成器输出的是 DagPipe 流水线，因为每个生成的 zip 文件默认都内置了故障恢复、模式验证和成本路由。其他 LLM 流水线框架都不具备这些功能。

为什么选择 DagPipe？

v0.2.3 新增内容

v0.2.3 为软件包仓库添加了官方 MCP Registry 元数据和标识符，支持在官方 MCP Registry 中一键发现。

v0.2.2 新增内容

v0.2.2 通过优化元数据、使项目描述更清晰，以及改进 AI Agent 分类，提升了在 PyPI 上的可发现性。

v0.2.1 新增内容

v0.2.1 带来了关键的生成器可靠性修复，以及一项广受期待的 DX 功能：

• verbose=True 输出：向 PipelineOrchestrator 传入 verbose=True，即可在命令行中实时查看各节点的进度更新，包括执行时间、节点描述和运行成本。
• 生成器核心修复：MCP 和 Apify 生成器提示词现在会严格要求位置参数 model=None，确保 YAML 拓扑依赖键查找准确无误（DEPENDENCY_NODE_ID），并开箱即支持原生 .csv、.xml、.html 和 .md 保存节点。

v0.2.0 新增内容

v0.2.0 是一次重大的架构升级。所有功能均为可选，并且完全向后兼容，这意味着现有的 v0.1.x 流水线无需修改即可运行。

智能模型路由器

路由器会根据任务复杂度自动选择要使用的 AI 模型。简单任务交给快速的免费模型，困难任务则升级到能力更强的模型。如果模型失败或触发速率限制，它会自动尝试下一个模型。

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,
)

实时模型注册表

用于记录免费层模型可用性和定价的自维护数据库。每 24 小时自动刷新。如果某个模型离线，注册表会将其标记为不可用，路由器也会停止向其路由，无需人工干预。

语义输出断言

除了模式验证外，节点还可以声明语义规则。如果 LLM 返回的数据结构有效但逻辑错误，断言会捕获问题并强制模型自行纠正：

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",
)

上下文隔离

节点只会接收其声明的依赖项的输出。处理不可信网络数据的节点无法访问凭据，也无法访问无关流水线分支的输出。

死信队列

当节点耗尽所有重试次数后，DagPipe 会将完整的失败上下文保存到 .dagpipe/checkpoints/{node_id}.failed.json，其中包括传入的内容、发生的错误以及手动修正说明。不会丢失任何失败上下文。

可插拔检查点后端

通过实现 CheckpointStorage 协议，可将文件系统存储替换为 Redis、S3 或任何自定义后端。

快速开始

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"]["summary"]
    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']['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=messages[0]["content"]
    ).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"})

运行中途崩溃？ 不必删除任何内容，重新运行即可。DagPipe 会读取检查点并自动跳过已完成的节点。

工作原理

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

📦 核心模块

dagpipe.dag：编排器

核心引擎。可从 Python 列表或 YAML 文件加载 DAG，按依赖关系对节点排序，并通过检查点和重试机制执行节点。

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：崩溃恢复

每次成功执行后，都会将节点输出保存到磁盘。恢复运行时，已完成的节点会被完全跳过。

v0.2.0 新增：CheckpointStorage Protocol，让你可以将默认文件系统后端替换为 Redis、S3 或任何自定义存储：

from dagpipe.dag import CheckpointStorage

class RedisCheckpoint(CheckpointStorage):
    def save(self, id: str, data: dict): redis_client.set(id, json.dumps(data))
    def load(self, id: str): return json.loads(redis_client.get(id) or "null")
    def exists(self, id: str): return redis_client.exists(id) > 0
    def clear(self): redis_client.flushdb()

（注意：在 v0.2.0 中，直接向 Orchestrator 传入 checkpoint_dir 已弃用。请改用 checkpoint_backend=FilesystemCheckpoint(path)。）

dagpipe.router：智能模型选择

将任务路由到能够处理该任务的最低成本模型。跟踪速率限制额度，并在重试时升级模型。

from dagpipe.router import ModelRouter, classify_complexity

score = classify_complexity("implement OAuth authentication", token_count=1200)
# → 0.8 (high complexity, triggers high_complexity_fn)

router = ModelRouter(
    low_complexity_fn=cheap_model,   low_label="7b",
    high_complexity_fn=smart_model,  high_label="70b",
    fallback_fn=backup_model,        fallback_label="backup",
)
fn, label = router.route(complexity=0.8)

dagpipe.constrained：保证结构化输出

使用 Pydantic schema 验证包装任何 LLM 调用。验证失败时，会将错误注入提示词并自动重试。

from pydantic import BaseModel
from dagpipe.constrained import constrained_generate

class ArticleOutput(BaseModel):
    title: str
    body: str
    word_count: int

result = constrained_generate(
    messages=[{"role": "user", "content": "Write a short article about AI."}],
    schema=ArticleOutput,
    llm_call_fn=my_llm,
    max_retries=3,
)
# result is a validated ArticleOutput instance, every time

dagpipe.registry：实时模型注册表 （v0.2.0 新增）

自动维护模型可用性和定价的数据库。每 24 小时刷新一次。如果模型离线，注册表会自动将其标记为不可用。

from dagpipe.registry import ModelRegistry

registry = ModelRegistry()
available = registry.list_available()   # live model list
cost = registry.get_cost("llama-3.3-70b-versatile")  # per-token pricing

📝 YAML 流水线定义

# my_pipeline.yaml
nodes:
  - id: research
    fn: research_fn
    complexity: 0.4
  - id: summarize
    fn: summarize_fn
    depends_on: [research]
    complexity: 0.5
  - id: publish
    fn: publish_fn
    depends_on: [summarize]
    is_deterministic: true

零成本技术栈

DagPipe 专为完全使用免费套餐运行而设计：

⚠️ Gemini 2.0 Flash 已于 2026 年 3 月 3 日停用。请将所有旧模型引用更新为 gemini-2.5-flash。

将其中任意一个配置为 low_complexity_fn、high_complexity_fn 或 fallback_fn。DagPipe 不绑定任何提供商，任何 Python 可调用对象都可以使用。

🔌 MCP 服务

DagPipe 提供实时 MCP 服务，让任何兼容 MCP 的 AI 客户端都能通过自然语言生成可从崩溃中恢复的流水线。

实时端点：

https://gastronomic-desk--dagpipe-generator-mcp.apify.actor/mcp

已在 Smithery 上架： https://smithery.ai/server/gastronomic-desk/dagpipe-generator

通过 Claude Desktop 连接

将以下内容添加到 claude_desktop_config.json：

{
  "mcpServers": {
    "dagpipe-generator": {
      "url": "https://gastronomic-desk--dagpipe-generator-mcp.apify.actor/mcp?groqApiKey=YOUR_GROQ_KEY",
      "transport": "streamable-http"
    }
  }
}

通过 Cursor / Windsurf 连接

{
  "dagpipe-generator": {
    "url": "https://gastronomic-desk--dagpipe-generator-mcp.apify.actor/mcp?groqApiKey=YOUR_GROQ_KEY",
    "transport": "streamable-http"
  }
}

在 console.groq.com/keys 获取免费的 Groq API 密钥。只需 2 分钟，无需信用卡。

在本地运行（stdio 传输）

git clone https://github.com/devilsfave/dagpipe
cd dagpipe
pip install -r servers/dagpipe-generator/requirements.txt
export GROQ_API_KEY=your_key_here
python servers/dagpipe-generator/server.py

🔒 安全架构

静态执行图

DagPipe 流水线无法自我修改。图会在运行时之前通过 YAML 声明。任何 LLM 调用都无法添加、删除或重新排列节点。因此，DagPipe 可以安全地处理敏感数据；而在 Agent 框架中，LLM 会决定接下来执行什么。

语义输出契约 （v0.2.0+）

除了 schema 验证，节点还可以声明语义断言。结构有效但逻辑错误的数据会被捕获并注入重试提示词，促使节点自我纠正：

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",
)

上下文隔离 （v0.2.0+）

节点只能接收其声明的依赖项的输出。处理不可信网络数据的节点无法访问 API 密钥、凭据或其他无关流水线分支的输出。

死信队列 （v0.2.0+）

节点在所有重试后仍然失败时，DagPipe 会将完整的失败上下文保存到 .dagpipe/checkpoints/{node_id}.failed.json，其中包括传入的上下文、最后一次错误以及手动修正说明。任何失败上下文都不会丢失。

🛒 模板与生态

基于 DagPipe 构建的开箱即用流水线包。下载、填入 API 密钥，然后运行。

🎭 Apify Actors

更多模板即将推出。有使用场景？提交 issue。

📊 项目状态

Phase 1: Core Library + Checkpointing   ████████████████████  COMPLETE (v0.1.0)
Phase 2: PyPI + Templates + Actors      ████████████████████  COMPLETE (v0.1.0)
Phase 3: V2 - Router + Registry + MCP   ████████████████████  COMPLETE (v0.2.3)
Phase 4: Parallel Execution (asyncio)   ░░░░░░░░░░░░░░░░░░░░  ROADMAP  (v0.3.0)

测试覆盖率： 132 项测试 · 5 个模块 · 0 个回归问题 · Python 3.12 + 3.13

❓ 常见问题

🤝 参与贡献

欢迎提交问题和 PR。如果发现错误，请提交问题并附上错误输出和 Python 版本。如果想添加功能，请先提交问题，以便我们在编写代码前就设计达成一致。

pip install dagpipe-core

from dagpipe.dag import PipelineOrchestrator, DAGNode, load_dag
from dagpipe.router import ModelRouter
from dagpipe.constrained import constrained_generate
from dagpipe.registry import ModelRegistry

# Checkpoints saved to .dagpipe/checkpoints/{node_id}.json
# Re-run after crash: completed nodes skipped automatically
orch = PipelineOrchestrator(nodes=nodes, node_registry=registry, router=router, verbose=True)
state, run = orch.run(initial_state={"input": "your data"})

📄 许可证

MIT：为全球开发者社区打造。

由 @devilsfave 构建