dagpipe

DagPipe

AI ワークフローを安全に運用するための信頼性レイヤー：クラッシュ復旧、スキーマ検証、コストルーティング

NeurIPS 2025 の研究では、実環境における 1,642 件のマルチエージェント実行トレースを分析し、 最先端のオープンソースシステム 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 の 3 つの使い方

開発者向け：ライブラリをインストールして、クラッシュに強い LLM パイプラインを Python で構築します：

pip install dagpipe-core

コーディングをしない方向け：ワークフローを平易な英語で説明すると、すぐに使えるクラッシュに強いパイプラインのコードをダウンロード可能な zip 形式で受け取れます。コーディングは不要です： 👉 Apify のパイプラインジェネレーター（1 回の実行につき $0.05）

AI エージェントと 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 エージェントのカテゴリ分けの強化により、PyPI で見つけやすくなりました。

v0.2.1 の新機能

v0.2.1 では、重要なジェネレーターの信頼性修正と、要望の多かった DX 機能が追加されました。

• verbose=True の出力：PipelineOrchestrator に verbose=True を渡すと、実行時間、ノードの説明、実行コストを含む、ノードごとのリアルタイムな CLI 進捗状況を確認できます。
• ジェネレーターのコア修正：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",
)

コンテキスト分離

ノードが受け取るのは、宣言した依存関係からの出力のみです。信頼できない Web データを処理するノードから、認証情報や、関係のないパイプライン分岐の出力にアクセスすることはできません。

デッドレターキュー

ノードがすべての再試行を使い切ると、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()

(注：Orchestrator に checkpoint_dir を直接渡す方法は、v0.2.0 で非推奨になりました。代わりに 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：保証された構造化出力

任意の LLM 呼び出しを Pydantic スキーマ検証でラップします。失敗した場合は、エラーをプロンプトに挿入して自動的に再試行します。

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"
  }
}

無料の Groq API キーは console.groq.com/keys で取得できます。所要時間は 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 の呼び出しによってノードを追加、削除、並べ替えることはできません。そのため、次に何を実行するかを LLM が決定するエージェントフレームワークとは異なり、DagPipe は機密データにも安全に使用できます。

意味的な出力コントラクト (v0.2.0+)

スキーマ検証に加えて、ノードに意味的なアサーションを宣言できます。構造上は有効でも論理的に誤っているデータを検出し、再試行プロンプトに挿入して、自己修正を促します。

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

❓ よくある質問

🤝 貢献

Issue や PR を歓迎します。バグを見つけた場合は、エラー出力と Python のバージョンを添えて Issue を作成してください。機能を追加したい場合は、コードを書く前に設計について合意できるよう、まず Issue を作成してください。

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 によって作られました