グラフエージェント

LangGraphスタイルのオーケストレーションとネイティブなADK-Rust統合を使用して、複雑でステートフルなワークフローを構築します。

概要

GraphAgentでは、ノードとエッジを持つ有向グラフとしてワークフローを定義でき、以下をサポートします。

• AgentNode: カスタム入出力マッパーを使用してLLMエージェントをグラフノードとしてラップ
• 循環ワークフロー: ループと反復的推論をネイティブにサポート（ReActパターン）
• 条件付きルーティング: 状態に基づく動的なエッジルーティング
• 状態管理: リデューサー（上書き、追加、合計、カスタム）を備えた型付き状態
• チェックポイント: 障害耐性とヒューマン・イン・ザ・ループのための永続的な状態
• ストリーミング: 複数のストリームモード（値、更新、メッセージ、デバッグ）

adk-graphクレートは、複雑でステートフルなエージェントワークフローを構築するためのLangGraphスタイルのワークフローオーケストレーションを提供します。グラフベースのワークフロー機能をADK-Rustエコシステムにもたらしながら、ADKのエージェントシステムとの完全な互換性を維持します。

主な利点:

• 視覚的なワークフロー設計: 複雑なロジックを直感的なノードとエッジのグラフとして定義
• 並列実行: パフォーマンス向上のため、複数のノードを同時に実行可能
• 状態の永続化: 障害耐性とヒューマン・イン・ザ・ループのための組み込みチェックポイント
• LLM統合: ADKエージェントをグラフノードとしてラップするネイティブサポート
• 柔軟なルーティング: 静的エッジ、条件付きルーティング、動的な意思決定

ワークフローエージェントとグラフの選択

ADK-Rustはオーケストレーションに2つの方法をサポートしています。どちらか一方が他方に取って代わるものではなく、両方とも 維持されています。

作業の形が既知で線形の場合は、ワークフローエージェントを優先してください。ワークフローエージェントは記述量が少なく、維持する状態スキーマもないためです。制御フローが結果に依存する場合や、実行をプロセス終了後も継続させる必要がある場合は、グラフを使用してください。

それらは組み合わせられます

3つのワークフローエージェントは Agent を実装し、AgentNode は任意の Agent をラップするため、ワークフローエージェントはグラフノードになります。

use adk_agent::SequentialAgent;
use adk_graph::node::AgentNode;
use std::sync::Arc;

let pipeline = Arc::new(SequentialAgent::new("pipeline", vec![extract, validate]));
let node = AgentNode::new(pipeline as Arc<dyn adk_core::Agent>);
// `node` now goes into a StateGraph like any other node.

GraphAgent も Agent を実装するため、逆も成り立ちます。グラフは SequentialAgent のサブエージェントになれます。分岐や永続性が必要な部分にはグラフを使用し、そうでない部分にはワークフローエージェントを使用してください。

作成するもの

このガイドでは、翻訳と要約を並列で実行する テキスト処理パイプライン を作成します。

                        ┌─────────────────────┐
       User Input       │                     │
      ────────────────▶ │       START         │
                        │                     │
                        └──────────┬──────────┘
                                   │
                   ┌───────────────┴───────────────┐
                   │                               │
                   ▼                               ▼
        ┌──────────────────┐            ┌──────────────────┐
        │   TRANSLATOR     │            │   SUMMARIZER     │
        │                  │            │                  │
        │  🇫🇷 French       │            │  📝 One sentence │
        │     Translation  │            │     Summary      │
        └─────────┬────────┘            └─────────┬────────┘
                  │                               │
                  └───────────────┬───────────────┘
                                  │
                                  ▼
                        ┌─────────────────────┐
                        │      COMBINE        │
                        │                     │
                        │  📋 Merge Results   │
                        └──────────┬──────────┘
                                   │
                                   ▼
                        ┌─────────────────────┐
                        │        END          │
                        │                     │
                        │   ✅ Complete       │
                        └─────────────────────┘

主な概念:

• ノード - 作業を実行する処理単位（LLM エージェント、関数、またはカスタムロジック）
• エッジ - ノード間の制御フロー（静的な接続または条件付きルーティング）
• 状態 - グラフ内を流れ、ノード間で保持される共有データ
• 並列実行 - 複数のノードを同時に実行してパフォーマンスを向上させる仕組み

コアコンポーネントを理解する

🔧 ノード: 作業担当 ノードで実際の作業が行われます。各ノードは次の処理を実行できます。

• AgentNode: LLM エージェントをラップして自然言語を処理する
• 関数ノード: データ処理用のカスタム Rust コードを実行する
• 組み込みノード: カウンターやバリデーターなど、あらかじめ定義されたロジックを使用する

ノードは組立ラインの専門作業員のようなものだと考えてください。それぞれに固有の役割と専門知識があります。

🔀 エッジ：フロー制御

エッジは、グラフ内で実行がどのように進むかを決定します。

• 静的エッジ：直接接続（A → B → C）
• 条件付きエッジ：状態に基づく動的ルーティング（if sentiment == "positive" → positive_handler）
• 並列エッジ：1つのノードから複数のパス（START → [translator, summarizer]）

エッジは、作業の流れを指示する信号機や道路標識のようなものです。

💾 状態：共有メモリ

状態は、すべてのノードが読み書きできるキーと値のストアです。

• 入力データ：グラフに供給される初期情報
• 中間結果：あるノードの出力が別のノードの入力になる
• 最終出力：すべての処理が完了した後の完成した結果

状態は、ノードが他のノードに利用してもらう情報を残せる共有ホワイトボードのように機能します。

⚡ 並列実行：速度の向上

1つのノードから複数のエッジが出ている場合、それらの対象ノードは同時に実行されます。

• 高速処理：独立したタスクを同時に実行
• リソース効率：CPU と I/O をより有効に活用
• スケーラビリティ：直線的な速度低下なしに、より複雑なワークフローを処理

これは、順番を待つのではなく、複数の作業者が仕事の異なる部分に同時に取り組むようなものです。

---

クイックスタート

1. プロジェクトを作成する

cargo new graph_demo
cd graph_demo

Cargo.toml に依存関係を追加します。

[dependencies]
adk-graph = { version = "2.1.0", features = ["sqlite"] }
adk-agent = "2.1.0"
adk-model = "2.1.0"
adk-core = "2.1.0"
tokio = { version = "1", features = ["full"] }
dotenvy = "0.15"
serde_json = "1.0"

API キーを使用して .env を作成します。

echo 'GOOGLE_API_KEY=your-api-key' > .env

2. 並列処理の例

以下は、テキストを並列処理する完全に動作する例です。

use adk_agent::LlmAgentBuilder;
use adk_graph::{
    agent::GraphAgent,
    edge::{END, START},
    node::{AgentNode, ExecutionConfig, NodeOutput},
    state::State,
};
use adk_model::GeminiModel;
use serde_json::json;
use std::sync::Arc;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    dotenvy::dotenv().ok();
    let api_key = std::env::var("GOOGLE_API_KEY")?;
    let model = Arc::new(GeminiModel::new(&api_key, "gemini-3.7-flash")?);

    // Create specialized LLM agents
    let translator_agent = Arc::new(
        LlmAgentBuilder::new("translator")
            .description("Translates text to French")
            .model(model.clone())
            .instruction("Translate the input text to French. Only output the translation.")
            .build()?,
    );

    let summarizer_agent = Arc::new(
        LlmAgentBuilder::new("summarizer")
            .description("Summarizes text")
            .model(model.clone())
            .instruction("Summarize the input text in one sentence.")
            .build()?,
    );

    // Wrap agents as graph nodes with input/output mappers
    let translator_node = AgentNode::new(translator_agent)
        .with_input_mapper(|state| {
            let text = state.get("input").and_then(|v| v.as_str()).unwrap_or("");
            adk_core::Content::new("user").with_text(text)
        })
        .with_output_mapper(|events| {
            let mut updates = std::collections::HashMap::new();
            for event in events {
                if let Some(content) = event.content() {
                    let text: String = content.parts.iter()
                        .filter_map(|p| p.text())
                        .collect::<Vec<_>>()
                        .join("");
                    if !text.is_empty() {
                        updates.insert("translation".to_string(), json!(text));
                    }
                }
            }
            updates
        });

    let summarizer_node = AgentNode::new(summarizer_agent)
        .with_input_mapper(|state| {
            let text = state.get("input").and_then(|v| v.as_str()).unwrap_or("");
            adk_core::Content::new("user").with_text(text)
        })
        .with_output_mapper(|events| {
            let mut updates = std::collections::HashMap::new();
            for event in events {
                if let Some(content) = event.content() {
                    let text: String = content.parts.iter()
                        .filter_map(|p| p.text())
                        .collect::<Vec<_>>()
                        .join("");
                    if !text.is_empty() {
                        updates.insert("summary".to_string(), json!(text));
                    }
                }
            }
            updates
        });

    // Build the graph with parallel execution
    let agent = GraphAgent::builder("text_processor")
        .description("Processes text with translation and summarization in parallel")
        .channels(&["input", "translation", "summary", "result"])
        .node(translator_node)
        .node(summarizer_node)
        .node_fn("combine", |ctx| async move {
            let translation = ctx.get("translation").and_then(|v| v.as_str()).unwrap_or("N/A");
            let summary = ctx.get("summary").and_then(|v| v.as_str()).unwrap_or("N/A");

            let result = format!(
                "=== Processing Complete ===\n\n\
                French Translation:\n{}\n\n\
                Summary:\n{}",
                translation, summary
            );

            Ok(NodeOutput::new().with_update("result", json!(result)))
        })
        // Parallel execution: both nodes start simultaneously
        .edge(START, "translator")
        .edge(START, "summarizer")
        .edge("translator", "combine")
        .edge("summarizer", "combine")
        .edge("combine", END)
        .build()?;

    // Execute the graph
    let mut input = State::new();
    input.insert("input".to_string(), json!("AI is transforming how we work and live."));

    let result = agent.invoke(input, ExecutionConfig::new("thread-1")).await?;
    println!("{}", result.get("result").and_then(|v| v.as_str()).unwrap_or(""));

    Ok(())
}

出力例：

=== Processing Complete ===

French Translation:
L'IA transforme notre façon de travailler et de vivre.

Summary:
AI is revolutionizing work and daily life through technological transformation.

グラフ実行の仕組み

全体像

グラフエージェントは スーパーステップ で実行されます。準備ができたすべてのノードが並列に実行され、グラフは次のステップに進む前に、すべてのノードの完了を待機します。

Step 1: START ──┬──▶ translator (running)
                └──▶ summarizer (running)
                
                ⏳ Wait for both to complete...
                
Step 2: translator ──┬──▶ combine (running)
        summarizer ──┘
        
                ⏳ Wait for combine to complete...
                
Step 3: combine ──▶ END ✅

ノード間の状態の流れ

各ノードは共有状態を読み書きできます。

┌─────────────────────────────────────────────────────────────────────┐
│ STEP 1: Initial state                                               │
├─────────────────────────────────────────────────────────────────────┤
│                                                                     │
│   State: { "input": "AI is transforming how we work" }             │
│                                                                     │
│                              ↓                                      │
│                                                                     │
│   ┌──────────────────┐              ┌──────────────────┐           │
│   │   translator     │              │   summarizer     │           │
│   │  reads "input"   │              │  reads "input"   │           │
│   └──────────────────┘              └──────────────────┘           │
│                                                                     │
└─────────────────────────────────────────────────────────────────────┘
                              ↓
┌─────────────────────────────────────────────────────────────────────┐
│ STEP 2: After parallel execution                                    │
├─────────────────────────────────────────────────────────────────────┤
│                                                                     │
│   State: {                                                          │
│     "input": "AI is transforming how we work",                     │
│     "translation": "L'IA transforme notre façon de travailler",    │
│     "summary": "AI is revolutionizing work through technology"     │
│   }                                                                 │
│                                                                     │
│                              ↓                                      │
│                                                                     │
│   ┌──────────────────────────────────────┐                         │
│   │           combine                    │                         │
│   │  reads "translation" + "summary"     │                         │
│   │  writes "result"                     │                         │
│   └──────────────────────────────────────┘                         │
│                                                                     │
└─────────────────────────────────────────────────────────────────────┘
                              ↓
┌─────────────────────────────────────────────────────────────────────┐
│ STEP 3: Final state                                                 │
├─────────────────────────────────────────────────────────────────────┤
│                                                                     │
│   State: {                                                          │
│     "input": "AI is transforming how we work",                     │
│     "translation": "L'IA transforme notre façon de travailler",    │
│     "summary": "AI is revolutionizing work through technology",    │
│     "result": "=== Processing Complete ===\n\nFrench..."          │
│   }                                                                 │
│                                                                     │
└─────────────────────────────────────────────────────────────────────┘

仕組みを支えるもの

---

LLM による条件付きルーティング

LLMs が実行経路を決定する、スマートなルーティングシステムを構築します。

ビジュアル：感情に基づくルーティング

                        ┌─────────────────────┐
       User Feedback    │                     │
      ────────────────▶ │    CLASSIFIER       │
                        │  🧠 Analyze tone    │
                        └──────────┬──────────┘
                                   │
                   ┌───────────────┼───────────────┐
                   │               │               │
                   ▼               ▼               ▼
        ┌──────────────────┐ ┌──────────────────┐ ┌──────────────────┐
        │   POSITIVE       │ │    NEGATIVE      │ │    NEUTRAL       │
        │                  │ │                  │ │                  │
        │  😊 Thank you!   │ │  😔 Apologize    │ │  😐 Ask more     │
        │     Celebrate    │ │     Help fix     │ │     questions    │
        └──────────────────┘ └──────────────────┘ └──────────────────┘

完全なサンプルコード

use adk_agent::LlmAgentBuilder;
use adk_graph::{
    edge::{END, Router, START},
    graph::StateGraph,
    node::{AgentNode, ExecutionConfig},
    state::State,
};
use adk_model::GeminiModel;
use serde_json::json;
use std::sync::Arc;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    dotenvy::dotenv().ok();
    let api_key = std::env::var("GOOGLE_API_KEY")?;
    let model = Arc::new(GeminiModel::new(&api_key, "gemini-3.7-flash")?);

    // Create classifier agent
    let classifier_agent = Arc::new(
        LlmAgentBuilder::new("classifier")
            .description("Classifies text sentiment")
            .model(model.clone())
            .instruction(
                "You are a sentiment classifier. Analyze the input text and respond with \
                ONLY one word: 'positive', 'negative', or 'neutral'. Nothing else.",
            )
            .build()?,
    );

    // Create response agents for each sentiment
    let positive_agent = Arc::new(
        LlmAgentBuilder::new("positive")
            .description("Handles positive feedback")
            .model(model.clone())
            .instruction(
                "You are a customer success specialist. The customer has positive feedback. \
                Express gratitude, reinforce the positive experience, and suggest ways to \
                share their experience. Be warm and appreciative. Keep response under 3 sentences.",
            )
            .build()?,
    );

    let negative_agent = Arc::new(
        LlmAgentBuilder::new("negative")
            .description("Handles negative feedback")
            .model(model.clone())
            .instruction(
                "You are a customer support specialist. The customer has a complaint. \
                Acknowledge their frustration, apologize sincerely, and offer help. \
                Be empathetic. Keep response under 3 sentences.",
            )
            .build()?,
    );

    let neutral_agent = Arc::new(
        LlmAgentBuilder::new("neutral")
            .description("Handles neutral feedback")
            .model(model.clone())
            .instruction(
                "You are a customer service representative. The customer has neutral feedback. \
                Ask clarifying questions to better understand their needs. Be helpful and curious. \
                Keep response under 3 sentences.",
            )
            .build()?,
    );

    // Create AgentNodes with mappers
    let classifier_node = AgentNode::new(classifier_agent)
        .with_input_mapper(|state| {
            let text = state.get("feedback").and_then(|v| v.as_str()).unwrap_or("");
            adk_core::Content::new("user").with_text(text)
        })
        .with_output_mapper(|events| {
            let mut updates = std::collections::HashMap::new();
            for event in events {
                if let Some(content) = event.content() {
                    let text: String = content.parts.iter()
                        .filter_map(|p| p.text())
                        .collect::<Vec<_>>()
                        .join("")
                        .to_lowercase()
                        .trim()
                        .to_string();
                    
                    let sentiment = if text.contains("positive") { "positive" }
                        else if text.contains("negative") { "negative" }
                        else { "neutral" };
                    
                    updates.insert("sentiment".to_string(), json!(sentiment));
                }
            }
            updates
        });

    // Response nodes (similar pattern for each)
    let positive_node = AgentNode::new(positive_agent)
        .with_input_mapper(|state| {
            let text = state.get("feedback").and_then(|v| v.as_str()).unwrap_or("");
            adk_core::Content::new("user").with_text(text)
        })
        .with_output_mapper(|events| {
            let mut updates = std::collections::HashMap::new();
            for event in events {
                if let Some(content) = event.content() {
                    let text: String = content.parts.iter()
                        .filter_map(|p| p.text())
                        .collect::<Vec<_>>()
                        .join("");
                    updates.insert("response".to_string(), json!(text));
                }
            }
            updates
        });

    // Build graph with conditional routing
    let graph = StateGraph::with_channels(&["feedback", "sentiment", "response"])
        .add_node(classifier_node)
        .add_node(positive_node)
        // ... add negative_node and neutral_node similarly
        .add_edge(START, "classifier")
        .add_conditional_edges(
            "classifier",
            Router::by_field("sentiment"),  // Route based on sentiment field
            [
                ("positive", "positive"),
                ("negative", "negative"),
                ("neutral", "neutral"),
            ],
        )
        .add_edge("positive", END)
        .add_edge("negative", END)
        .add_edge("neutral", END)
        .compile()?;

    // Test with different feedback
    let mut input = State::new();
    input.insert("feedback".to_string(), json!("Your product is amazing! I love it!"));

    let result = graph.invoke(input, ExecutionConfig::new("feedback-1")).await?;
    println!("Sentiment: {}", result.get("sentiment").and_then(|v| v.as_str()).unwrap_or(""));
    println!("Response: {}", result.get("response").and_then(|v| v.as_str()).unwrap_or(""));

    Ok(())
}

サンプルフロー：

Input: "Your product is amazing! I love it!"
       ↓
Classifier: "positive"
       ↓
Positive Agent: "Thank you so much for the wonderful feedback! 
                We're thrilled you love our product. 
                Would you consider leaving a review to help others?"

ReAct パターン：推論 + 実行

複雑な問題を解決するために、ツールを反復的に使用できるエージェントを構築します。

ビジュアル：ReAct サイクル

                        ┌─────────────────────┐
       User Question    │                     │
      ────────────────▶ │      REASONER       │
                        │  🧠 Think + Act     │
                        └──────────┬──────────┘
                                   │
                                   ▼
                        ┌─────────────────────┐
                        │   Has tool calls?   │
                        │                     │
                        └──────────┬──────────┘
                                   │
                   ┌───────────────┴───────────────┐
                   │                               │
                   ▼                               ▼
        ┌──────────────────┐            ┌──────────────────┐
        │       YES        │            │        NO        │
        │                  │            │                  │
        │  🔄 Loop back    │            │  ✅ Final answer │
        │     to reasoner  │            │      END         │
        └─────────┬────────┘            └──────────────────┘
                  │
                  └─────────────────┐
                                    │
                                    ▼
                        ┌─────────────────────┐
                        │      REASONER       │
                        │  🧠 Think + Act     │
                        │   (next iteration)  │
                        └─────────────────────┘

完全なReActサンプル

use adk_agent::LlmAgentBuilder;
use adk_core::{Part, Tool};
use adk_graph::{
    edge::{END, START},
    graph::StateGraph,
    node::{AgentNode, ExecutionConfig, NodeOutput},
    state::State,
};
use adk_model::GeminiModel;
use adk_tool::FunctionTool;
use serde_json::json;
use std::sync::Arc;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    dotenvy::dotenv().ok();
    let api_key = std::env::var("GOOGLE_API_KEY")?;
    let model = Arc::new(GeminiModel::new(&api_key, "gemini-3.7-flash")?);

    // Create tools
    let weather_tool = Arc::new(FunctionTool::new(
        "get_weather",
        "Get the current weather for a location. Takes a 'location' parameter (city name).",
        |_ctx, args| async move {
            let location = args.get("location").and_then(|v| v.as_str()).unwrap_or("unknown");
            Ok(json!({
                "location": location,
                "temperature": "72°F",
                "condition": "Sunny",
                "humidity": "45%"
            }))
        },
    )) as Arc<dyn Tool>;

    let calculator_tool = Arc::new(FunctionTool::new(
        "calculator",
        "Perform mathematical calculations. Takes an 'expression' parameter (string).",
        |_ctx, args| async move {
            let expr = args.get("expression").and_then(|v| v.as_str()).unwrap_or("0");
            let result = match expr {
                "2 + 2" => "4",
                "10 * 5" => "50",
                "100 / 4" => "25",
                "15 - 7" => "8",
                _ => "Unable to evaluate",
            };
            Ok(json!({ "result": result, "expression": expr }))
        },
    )) as Arc<dyn Tool>;

    // Create reasoner agent with tools
    let reasoner_agent = Arc::new(
        LlmAgentBuilder::new("reasoner")
            .description("Reasoning agent with tools")
            .model(model.clone())
            .instruction(
                "You are a helpful assistant with access to tools. Use tools when needed to answer questions. \
                When you have enough information, provide a final answer without using more tools.",
            )
            .tool(weather_tool)
            .tool(calculator_tool)
            .build()?,
    );

    // Create reasoner node that detects tool usage
    let reasoner_node = AgentNode::new(reasoner_agent)
        .with_input_mapper(|state| {
            let question = state.get("question").and_then(|v| v.as_str()).unwrap_or("");
            adk_core::Content::new("user").with_text(question)
        })
        .with_output_mapper(|events| {
            let mut updates = std::collections::HashMap::new();
            let mut has_tool_calls = false;
            let mut response = String::new();

            for event in events {
                if let Some(content) = event.content() {
                    for part in &content.parts {
                        match part {
                            Part::FunctionCall { .. } => {
                                has_tool_calls = true;
                            }
                            Part::Text { text } => {
                                response.push_str(text);
                            }
                            _ => {}
                        }
                    }
                }
            }

            updates.insert("has_tool_calls".to_string(), json!(has_tool_calls));
            updates.insert("response".to_string(), json!(response));
            updates
        });

    // Build ReAct graph with cycle
    let graph = StateGraph::with_channels(&["question", "has_tool_calls", "response", "iteration"])
        .add_node(reasoner_node)
        .add_node_fn("counter", |ctx| async move {
            let i = ctx.get("iteration").and_then(|v| v.as_i64()).unwrap_or(0);
            Ok(NodeOutput::new().with_update("iteration", json!(i + 1)))
        })
        .add_edge(START, "counter")
        .add_edge("counter", "reasoner")
        .add_conditional_edges(
            "reasoner",
            |state| {
                let has_tools = state.get("has_tool_calls").and_then(|v| v.as_bool()).unwrap_or(false);
                let iteration = state.get("iteration").and_then(|v| v.as_i64()).unwrap_or(0);

                // Safety limit
                if iteration >= 5 { return END.to_string(); }

                if has_tools {
                    "counter".to_string()  // Loop back for more reasoning
                } else {
                    END.to_string()  // Done - final answer
                }
            },
            [("counter", "counter"), (END, END)],
        )
        .compile()?
        .with_recursion_limit(10);

    // Test the ReAct agent
    let mut input = State::new();
    input.insert("question".to_string(), json!("What's the weather in Paris and what's 15 + 25?"));

    let result = graph.invoke(input, ExecutionConfig::new("react-1")).await?;
    println!("Final answer: {}", result.get("response").and_then(|v| v.as_str()).unwrap_or(""));
    println!("Iterations: {}", result.get("iteration").and_then(|v| v.as_i64()).unwrap_or(0));

    Ok(())
}

サンプルフロー：

Question: "What's the weather in Paris and what's 15 + 25?"

Iteration 1:
  Reasoner: "I need to get weather info and do math"
  → Calls get_weather(location="Paris") and calculator(expression="15 + 25")
  → has_tool_calls = true → Loop back

Iteration 2:
  Reasoner: "Based on the results: Paris is 72°F and sunny, 15 + 25 = 40"
  → No tool calls → has_tool_calls = false → END

Final Answer: "The weather in Paris is 72°F and sunny with 45% humidity. 
              And 15 + 25 equals 40."

AgentNode

任意のADK Agent（通常はLlmAgent）をグラフノードとしてラップします。

エージェントから見えるもの

グラフ内のエージェントは、グラフを開始した呼び出しから導出されたコンテキストで実行されるため、グラフ外で実行される場合と同じように動作します。Runner がGraphAgentを呼び出すと、呼び出し元の識別情報とサービスは自動的に引き継がれます。

グラフを直接呼び出した場合 — graph.invoke(state, ExecutionConfig::new("thread")) — 継承する呼び出しはありません。これがスタンドアロンモードです。ノードには user_id = "graph_user"、app_name = "graph_app"、ブランチ main が渡されますが、シークレットも メモリもありません。これは Runner の外部でグラフを実行するための意図的な モードであり、本番環境で安易に使用するフォールバックではありません。

手動で橋渡しする場合 — たとえば独自の executor からグラフを駆動する場合 — は、 呼び出しを明示的に渡します。

let config = ExecutionConfig::new(ctx.session_id()).with_parent_context(ctx.clone());

注: ノードは引き続き独自のインメモリグラフセッション上で実行されるため、ノード内の エージェントの会話履歴は呼び出し元のセッションに追加されず、ノード単位にスコープされます。

let node = AgentNode::new(llm_agent)
    .with_input_mapper(|state| {
        // Transform graph state to agent input Content
        let text = state.get("input").and_then(|v| v.as_str()).unwrap_or("");
        adk_core::Content::new("user").with_text(text)
    })
    .with_output_mapper(|events| {
        // Transform agent events to state updates
        let mut updates = std::collections::HashMap::new();
        for event in events {
            if let Some(content) = event.content() {
                let text: String = content.parts.iter()
                    .filter_map(|p| p.text())
                    .collect::<Vec<_>>()
                    .join("");
                updates.insert("output".to_string(), json!(text));
            }
        }
        updates
    });

関数ノード

状態を処理する単純な非同期関数:

.node_fn("process", |ctx| async move {
    let input = ctx.state.get("input").unwrap();
    let output = process_data(input).await?;
    Ok(NodeOutput::new().with_update("output", output))
})

エッジの種類

静的エッジ

ノード間の直接的な接続:

.edge(START, "first_node")
.edge("first_node", "second_node")
.edge("second_node", END)

条件付きエッジ

状態に基づく動的なルーティング:

.conditional_edge(
    "router",
    |state| {
        match state.get("next").and_then(|v| v.as_str()) {
            Some("research") => "research_node".to_string(),
            Some("write") => "write_node".to_string(),
            _ => END.to_string(),
        }
    },
    [
        ("research_node", "research_node"),
        ("write_node", "write_node"),
        (END, END),
    ],
)

ノード内部からのルーティング

条件付きエッジでは、グラフの構築時にターゲットが固定されます。NodeOutput::with_goto ではそうなりません。ノードは状態を書き込み、同じステップで後続ノードを指定します。また、 エッジで接続されていないノードを含め、グラフ内の任意のノードを指定できます。

use adk_graph::node::NodeOutput;
use serde_json::json;

// The node decides where control goes, from what it just computed.
async fn triage(ctx: &adk_graph::node::NodeContext) -> adk_graph::error::Result<NodeOutput> {
    let amount = ctx.get("amount").and_then(|v| v.as_f64()).unwrap_or(0.0);
    let next = if amount > 10_000.0 { "escalate" } else { "auto_approve" };
    Ok(NodeOutput::new().with_update("risk", json!(next)).with_goto([next]))
}

goto が生成するフロンティアも他と同様にチェックポイント化されるため、一時停止した実行は goto が選択したノードから再開されます。

注: グラフの構築時点で可能なターゲットが判明している場合は add_conditional_edges を使用してください。この場合、エッジはレンダリングされた図に表示されます。選択をノードに委ねる場合は goto を使用してください。

ルーターのヘルパー

一般的なパターンには、組み込みルーターを使用します。

use adk_graph::edge::Router;

// Route based on a state field value
.conditional_edge("classifier", Router::by_field("sentiment"), [
    ("positive", "positive_handler"),
    ("negative", "negative_handler"),
    ("neutral", "neutral_handler"),
])

// Route based on boolean field
.conditional_edge("check", Router::by_bool("approved"), [
    ("true", "execute"),
    ("false", "reject"),
])

// Limit iterations
.conditional_edge("loop", Router::max_iterations("count", 5), [
    ("continue", "process"),
    ("done", END),
])

並列実行

1つのノードから複数のエッジがある場合、それらは並列に実行されます。

let agent = GraphAgent::builder("parallel_processor")
    .channels(&["input", "translation", "summary", "analysis"])
    .node(translator_node)
    .node(summarizer_node)
    .node(analyzer_node)
    .node(combiner_node)
    // All three start simultaneously
    .edge(START, "translator")
    .edge(START, "summarizer")
    .edge(START, "analyzer")
    // Wait for all to complete before combining
    .edge("translator", "combiner")
    .edge("summarizer", "combiner")
    .edge("analyzer", "combiner")
    .edge("combiner", END)
    .build()?;

combiner は、3つのブランチがすべて到達した後に 1回だけ 実行されます。上記のコードでは、そのための指定は何も行っていません。複数の直接エッジから入ってくるノードは、コンパイル時に自動的に遅延されます。そのため、長さの異なるブランチも設定なしで正しく結合されます。

カウントの取得方法から、次の2つの点が導かれます。

循環グラフ（ReAct パターン）

循環を持つ反復的な推論エージェントを構築します。

use adk_core::Part;

// Create agent with tools
let reasoner = Arc::new(
    LlmAgentBuilder::new("reasoner")
        .model(model)
        .instruction("Use tools to answer questions. Provide final answer when done.")
        .tool(search_tool)
        .tool(calculator_tool)
        .build()?
);

let reasoner_node = AgentNode::new(reasoner)
    .with_input_mapper(|state| {
        let question = state.get("question").and_then(|v| v.as_str()).unwrap_or("");
        adk_core::Content::new("user").with_text(question)
    })
    .with_output_mapper(|events| {
        let mut updates = std::collections::HashMap::new();
        let mut has_tool_calls = false;
        let mut response = String::new();

        for event in events {
            if let Some(content) = event.content() {
                for part in &content.parts {
                    match part {
                        Part::FunctionCall { name, .. } => {
                            has_tool_calls = true;
                        }
                        Part::Text { text } => {
                            response.push_str(text);
                        }
                        _ => {}
                    }
                }
            }
        }

        updates.insert("has_tool_calls".to_string(), json!(has_tool_calls));
        updates.insert("response".to_string(), json!(response));
        updates
    });

// Build graph with cycle
let react_agent = StateGraph::with_channels(&["question", "has_tool_calls", "response", "iteration"])
    .add_node(reasoner_node)
    .add_node_fn("counter", |ctx| async move {
        let i = ctx.get("iteration").and_then(|v| v.as_i64()).unwrap_or(0);
        Ok(NodeOutput::new().with_update("iteration", json!(i + 1)))
    })
    .add_edge(START, "counter")
    .add_edge("counter", "reasoner")
    .add_conditional_edges(
        "reasoner",
        |state| {
            let has_tools = state.get("has_tool_calls").and_then(|v| v.as_bool()).unwrap_or(false);
            let iteration = state.get("iteration").and_then(|v| v.as_i64()).unwrap_or(0);

            // Safety limit
            if iteration >= 5 { return END.to_string(); }

            if has_tools {
                "counter".to_string()  // Loop back
            } else {
                END.to_string()  // Done
            }
        },
        [("counter", "counter"), (END, END)],
    )
    .compile()?
    .with_recursion_limit(10);

マルチエージェント・スーパーバイザー

タスクを専門エージェントに振り分けます。

// Create supervisor agent
let supervisor = Arc::new(
    LlmAgentBuilder::new("supervisor")
        .model(model.clone())
        .instruction("Route tasks to: researcher, writer, or coder. Reply with agent name only.")
        .build()?
);

let supervisor_node = AgentNode::new(supervisor)
    .with_output_mapper(|events| {
        let mut updates = std::collections::HashMap::new();
        for event in events {
            if let Some(content) = event.content() {
                let text: String = content.parts.iter()
                    .filter_map(|p| p.text())
                    .collect::<Vec<_>>()
                    .join("")
                    .to_lowercase();

                let next = if text.contains("researcher") { "researcher" }
                    else if text.contains("writer") { "writer" }
                    else if text.contains("coder") { "coder" }
                    else { "done" };

                updates.insert("next_agent".to_string(), json!(next));
            }
        }
        updates
    });

// Build supervisor graph
let graph = StateGraph::with_channels(&["task", "next_agent", "research", "content", "code"])
    .add_node(supervisor_node)
    .add_node(researcher_node)
    .add_node(writer_node)
    .add_node(coder_node)
    .add_edge(START, "supervisor")
    .add_conditional_edges(
        "supervisor",
        Router::by_field("next_agent"),
        [
            ("researcher", "researcher"),
            ("writer", "writer"),
            ("coder", "coder"),
            ("done", END),
        ],
    )
    // Agents report back to supervisor
    .add_edge("researcher", "supervisor")
    .add_edge("writer", "supervisor")
    .add_edge("coder", "supervisor")
    .compile()?;

状態管理

Reducer を使用した状態スキーマ

状態の更新をマージする方法を制御します。

let schema = StateSchema::builder()
    .channel("current_step")                    // Overwrite (default)
    .list_channel("messages")                   // Append to list
    .channel_with_reducer("count", Reducer::Sum) // Sum values
    .channel_with_reducer("data", Reducer::Custom(Arc::new(|old, new| {
        // Custom merge logic
        merge_json(old, new)
    })))
    .build();

let agent = GraphAgent::builder("stateful")
    .state_schema(schema)
    // ... nodes and edges
    .build()?;

Reducer の種類

更新順序

1つのスーパーステップ内のノードは並行して実行され、処理にかかる時間に応じて異なる順序で完了します。状態の更新は、ノードの完了順ではなく、ノード名の順序で適用されます。

この順序は、リデューサーが可換でない場合に重要です。Append は配列を構築するため、順序が結果になります。Custom のリデューサーも順序に依存する可能性があります。ノード名で並べ替えることで、実行を再現可能にできます。同じグラフと同じ入力であれば、遅い依存先のタイミングに関係なく、同じ状態になります。

1つのノードが複数のチャネルに書き込む場合、それらはチャネル名の順序で適用されます。チャネルは相互に作用しないため、これはトレースを読み取る場合にのみ重要です。

チェックポイント

フォールトトレランスとヒューマン・イン・ザ・ループを実現するには、永続的な状態を有効にします。

インメモリ（開発環境）

use adk_graph::checkpoint::MemoryCheckpointer;

let checkpointer = Arc::new(MemoryCheckpointer::new());

let graph = StateGraph::with_channels(&["task", "result"])
    // ... nodes and edges
    .compile()?
    .with_checkpointer_arc(checkpointer.clone());

SQLite（本番環境）

use adk_graph::checkpoint::SqliteCheckpointer;

let checkpointer = SqliteCheckpointer::new("checkpoints.db").await?;

let graph = StateGraph::with_channels(&["task", "result"])
    // ... nodes and edges
    .compile()?
    .with_checkpointer(checkpointer);

チェックポイントに記録される内容

チェックポイントには、蓄積された状態、ステップ番号、そしてフロンティア（まだ実行する必要があるノード）が保存されます。フロンティアが進んだ後に書き込まれるため、再開時にすでに完了したノードが再実行されることも、更新が二重に適用されることもありません。完了した実行では空のフロンティアがチェックポイントに保存されるため、完了済みのスレッドを再開するとグラフが最初から再実行されるのではなく、最終状態が返されます。

次の2つのケースでは、次に実行するフロンティアではなく、実行中だったフロンティアを意図的にチェックポイントに保存します。これは、中断されたノードがまだ更新を生成しておらず、再開時にもう一度実行する必要があるためです。

ストリーム実行はブロッキング実行と同じスケジュールでチェックポイントを作成します。割り込みによってストリームが終了した場合も同様であるため、human-in-the-loop による一時停止はどちらの実行モードでも再開できます。

チェックポイント履歴（タイムトラベル）

読み取り専用です。 TimeTravelHandle::state_history(from, to) は、各チェックポイント化されたステップで 保存された 状態を返します。何も実行されず、ノードも実行されず、イベントも再生成されず、副作用も繰り返されません。履歴内の地点から再実行するには、fork_at を使用してそのチェックポイントから分岐し、フォークされたスレッド上でグラフを呼び出します。このメソッドは以前は replay という名前で、グラフを再実行すると説明されていましたが、実際にはそのような動作は一度も行っていません。

チェックポイントを使用すると 永続的な再開 も可能になります。グラフの実行がクラッシュした場合やプロセスが再起動した場合、実行は最初からやり直すのではなく、最後に永続化されたチェックポイントから再開されます。クラッシュに安全な永続化には SqliteCheckpointer（sqlite 機能）を使用します。MemoryCheckpointer はプロセス内にチェックポイントを保持するため、再起動後も存続しません。この 2 つがこのクレートに付属するバックエンドです。それ以外のものを使用する場合は、Checkpointer トレイトを実装してください。

// List all checkpoints for a thread
let checkpoints = checkpointer.list("thread-id").await?;
for cp in checkpoints {
    println!("Step {}: {:?}", cp.step, cp.state.get("status"));
}

// Load a specific checkpoint
if let Some(checkpoint) = checkpointer.load_by_id(&checkpoint_id).await? {
    println!("State at step {}: {:?}", checkpoint.step, checkpoint.state);
}

human-in-the-loop

動的な割り込みを使用して、人間による承認のために実行を一時停止します。

use adk_graph::{error::GraphError, node::NodeOutput};

// Planner agent assesses risk
let planner_node = AgentNode::new(planner_agent)
    .with_output_mapper(|events| {
        let mut updates = std::collections::HashMap::new();
        for event in events {
            if let Some(content) = event.content() {
                let text: String = content.parts.iter()
                    .filter_map(|p| p.text())
                    .collect::<Vec<_>>()
                    .join("");

                // Extract risk level from LLM response
                let risk = if text.to_lowercase().contains("risk: high") { "high" }
                    else if text.to_lowercase().contains("risk: medium") { "medium" }
                    else { "low" };

                updates.insert("plan".to_string(), json!(text));
                updates.insert("risk_level".to_string(), json!(risk));
            }
        }
        updates
    });

// Review node with dynamic interrupt
let graph = StateGraph::with_channels(&["task", "plan", "risk_level", "approved", "result"])
    .add_node(planner_node)
    .add_node(executor_node)
    .add_node_fn("review", |ctx| async move {
        let risk = ctx.get("risk_level").and_then(|v| v.as_str()).unwrap_or("low");
        let approved = ctx.get("approved").and_then(|v| v.as_bool());

        // Already approved - continue
        if approved == Some(true) {
            return Ok(NodeOutput::new());
        }

        // High/medium risk - interrupt for approval
        if risk == "high" || risk == "medium" {
            return Ok(NodeOutput::interrupt_with_data(
                &format!("{} RISK: Human approval required", risk.to_uppercase()),
                json!({
                    "plan": ctx.get("plan"),
                    "risk_level": risk,
                    "action": "Set 'approved' to true to continue"
                })
            ));
        }

        // Low risk - auto-approve
        Ok(NodeOutput::new().with_update("approved", json!(true)))
    })
    .add_edge(START, "planner")
    .add_edge("planner", "review")
    .add_edge("review", "executor")
    .add_edge("executor", END)
    .compile()?
    .with_checkpointer_arc(checkpointer.clone());

// Execute - may pause for approval
let thread_id = "task-001";
let result = graph.invoke(input, ExecutionConfig::new(thread_id)).await;

match result {
    Err(GraphError::Interrupted(interrupt)) => {
        println!("*** EXECUTION PAUSED ***");
        println!("Reason: {}", interrupt.interrupt);
        println!("Plan awaiting approval: {:?}", interrupt.state.get("plan"));

        // Human reviews and approves...

        // Update state with approval
        graph.update_state(thread_id, [("approved".to_string(), json!(true))]).await?;

        // Resume execution
        let final_result = graph.invoke(State::new(), ExecutionConfig::new(thread_id)).await?;
        println!("Final result: {:?}", final_result.get("result"));
    }
    Ok(result) => {
        println!("Completed without interrupt: {:?}", result);
    }
    Err(e) => {
        println!("Error: {}", e);
    }
}

静的な割り込み

必須の一時停止ポイントには interrupt_before または interrupt_after を使用します。

let graph = StateGraph::with_channels(&["task", "plan", "result"])
    .add_node(planner_node)
    .add_node(executor_node)
    .add_edge(START, "planner")
    .add_edge("planner", "executor")
    .add_edge("executor", END)
    .compile()?
    .with_interrupt_before(&["executor"]);  // Always pause before execution

ストリーミング実行

グラフの実行中にイベントをストリームします。

use futures::StreamExt;
use adk_graph::stream::StreamMode;

let stream = agent.stream(input, config, StreamMode::Updates);

while let Some(event) = stream.next().await {
    match event? {
        StreamEvent::NodeStart(name) => println!("Starting: {}", name),
        StreamEvent::Updates { node, updates } => {
            println!("{} updated state: {:?}", node, updates);
        }
        StreamEvent::NodeEnd(name) => println!("Completed: {}", name),
        StreamEvent::Done(state) => println!("Final state: {:?}", state),
        _ => {}
    }
}

ストリームモード

Messages モードは、生成されるにつれて Node::execute_stream からトークンを読み取ります。 このモードでは、各ノードはスーパーステップごとに 1 回だけ 実行されます。ノードは StreamEvent::Updates イベントとしてストリーム上に状態の更新を報告し、エグゼキューターは それらを適用します。更新を収集するためにノードをもう一度実行することはありません。これは AgentNode で特に重要です。2 回目の実行により、ノードごとにモデル呼び出しの 課金が 2 回発生することになるためです。

重要: execute_stream をオーバーライドするカスタム Node は、状態の更新を含む StreamEvent::Updates イベントを生成する必要があります。これがない場合、ノードはイベントを ストリームに送信しますが、Messages モードでは状態に何も反映されません。 execute をラップするデフォルトの execute_stream は、これを自動的に行います。

タイムアウトポリシーは、ストリーミング実行自体に適用されます。ストリームの場合、 idle_timeout は制限時間内にイベントが生成されなかったことを意味します。

サブグラフ

コンパイル済みグラフは、SubgraphNode を介して別のグラフのノードとして実行されます。内部グラフは 独自のチャネル、エッジ、割り込みゲートを保持し、名前付きチャネルを親グラフと 交換します。

use adk_graph::subgraph::SubgraphNode;
use std::sync::Arc;

let outer = StateGraph::with_channels(&["document", "size"])
    .add_node(
        SubgraphNode::new("measure_doc", Arc::new(inner))
            .with_input("document", "text")
            .with_output("length", "size"),
    )
    .add_edge(START, "measure_doc")
    .add_edge("measure_doc", END)
    .compile()?;

最後の行が知っておく価値のある違いです。どちらのスキーマも実行前に利用できるため、どちらの側も宣言していないチャネルを指定するマッピングが実行に到達して、値の欠落として表面化することはありません。何も交換しないサブグラフも同じように拒否されます。親に影響を与えられないためです。

サブグラフ内の一時停止からの再開

追加の対応は必要ありません。同じスレッドで親を再度呼び出すとサブグラフに再入し、サブグラフは <parent thread>/<node name> 上にある自身のチェックポイントを見つけて、停止した箇所から続行します。一時停止前にサブグラフが完了した処理は繰り返されません。また、数レベル下での一時停止も同じ方法で再開されます — メッセージには通過した各レベルの名前が示されます。

割り込みゲートを宣言しているもののチェックポインターを保持していないサブグラフは、親のコンパイル時に拒否されます。そうしないと最初のノードから再入し、完了済みの処理を二度目に実行することになるためです。

両方のグラフが実際のチェックポインタを保持しているため、これはプロセスの再起動後も維持されます。データベースだけを共有する新しいグラフオブジェクトの集合が、同じ実行を再開します。

親への制御の返却

サブグラフ内のノードは、自身のグラフを終了し、それを保持するグラフのノードを指定できます。

Ok(NodeOutput::new()
    .with_update("reason", json!("no confident answer"))
    .with_goto_parent(["escalate"]))

サブグラフは通常どおり処理を完了し、出力チャネルを反映します。その後、親はサブグラフノードに宣言されたエッジをたどるのではなく、escalate から処理を続行します。ターゲットを検証するのは、それ自身のノードを把握しているのが親側だけだからです。

信頼性とコストの制御

これらはすべてデフォルトでは無効であるため、いずれかを設定するまではグラフは以前と同じように動作します。

ノードごとの再試行

一時的な失敗（レート制限や接続切断など）が発生すると、それ以外の場合は実行が終了します。

use adk_graph::retry::{RetryOn, RetryPolicy};
use std::time::Duration;

let graph = graph.with_node_retry(
    "call_model",
    RetryPolicy::new(3)
        .with_initial_delay(Duration::from_millis(500))
        .with_max_delay(Duration::from_secs(8))
        .with_backoff_factor(2.0)
        .with_retry_on(RetryOn::Any),
);

遅延は backoff_factor によって増加し、max_delay で上限に達した後、ジッターが適用されます。ポリシーのないノードは1回だけ実行されるため、再試行は明示的に有効化した場合に限られます。RetryPolicy::default() のポリシーでは10回試行でき、その9回のスリープ時間の合計は約243秒になります。呼び出し元が応答を待機している場合は、max_attempts を小さくしてください。

retry_on の内容にかかわらず、中断が再試行されることはありません。停止は失敗ではないためです。試行回数はチェックポイントに保存されるため、再開された実行では予算を最初からやり直すのではなく、その続きから使用します。

同時実行数の制限

多数へのファンアウトではフロンティア全体が一度にディスパッチされるため、接続プールを使い果たしたり、プロバイダーのレート制限に抵触したりする可能性があります。

let graph = graph.with_max_concurrency(4);

上限を超えたノードは、スロットが空くまで待機します。受付順序は名前でソートされたフロンティアとなるため、タイミングには依存しません。命令型の子呼び出しはこの予算の対象外です。これは、親が自身のスロットを保持したまま子の完了を待機するためです。

ノードごとのタイムアウト

TimeoutPolicy は単一の試行を制限し、idle_timeout と組み合わせて、ノードが進捗を報告せずにいられる時間を定めます。いずれかを超えると GraphError::NodeTimedOut になり、リトライポリシーがそれに対処できます。

ノードを直接呼び出す

サブタスクの数がグラフの形状ではなく状態から決まる場合、ノードは別のノードを自ら呼び出せます。

use adk_graph::child::RunNodeOptions;

let output = ctx
    .run_node_with("reviewer", json!({ "aspect": aspect }), RunNodeOptions::with_run_id(aspect))
    .await?;

対象ノードにエッジは必要ありません。完了した各子ノードは <parent>/<child>@<run_id> の下に記録されるため、再開された実行では子ノードを再度実行せず、記録された結果を返します。これは、子ノードの実行にモデル呼び出しのコストがかかる場合に重要です。

ノードキャッシュ

ノード上の cache_policy は、オプションの TTL とともに、ノード名と現在の状態をキーとして結果を保存するため、入力に変更がなければ処理を省略できます。node-cache 機能が必要です。Redis をバックエンドとするストアは redis-cache の背後で利用できます。

差分チェックポイント

delta 機能は、状態全体ではなくスーパーステップ間の差分を保存します。これは、状態が大きく、ステップ数が多い場合に重要です。

タイムトラベル

time-travel 機能を使用すると、graph.time_travel(thread_id)? はスレッドのチェックポイント履歴に対するハンドルを返します。ステップの一覧表示、ある時点の状態の読み取り、または fork_at によるチェックポイントからの新しいスレッドの分岐が可能です。この呼び出しは Result を返します。すべての操作でチェックポイントが読み取られるため、チェックポインターのないグラフではパニックを起こすのではなく GraphError::CheckpointError が報告されます。

グラフ全体のデフォルト値

20 個のノードで同じリトライ設定を繰り返すと、設定漏れによる誤りが起こりやすくなります。

use adk_graph::graph::NodeDefaults;

let graph = graph
    .with_node_defaults(NodeDefaults::new().with_retry(RetryPolicy::new(3)))
    .with_node_retry("critical", RetryPolicy::new(10));

ノードごとの値が常に優先されます。NodeDefaults にはタイムアウトと失敗ハンドラーも指定できます。

ノードの失敗からの復旧

ノードのリトライ予算を使い切った後でも、ハンドラーは実行を終了する代わりに、発生した内容を記録し、復旧ノードを指定できます。

let graph = graph.with_node_error_handler("charge", |node, error, _state| {
    Ok(NodeOutput::new()
        .with_update("status", json!(format!("{node} failed: {error}")))
        .with_goto(["compensate"]))
});

Errを返すと、これまでどおり実行が終了します。中断がハンドラーに到達することはありません。 なぜなら、一時停止は失敗ではないからです。

チェックポイントの増加を抑える

スレッドはスーパーステップごとに1つのチェックポイントを蓄積します。そのため、数日間 存続する実行では無制限に増加し、ストレージを消費してlistを遅くします。

use adk_graph::checkpoint::RetentionPolicy;
use std::time::Duration;

let graph = graph
    .with_checkpoint_retention(
        RetentionPolicy::keep_last(50).with_max_age(Duration::from_secs(7 * 24 * 3600)),
    );

プルーニングは各保存の後に行われるため、コストは実行に比例したままとなり、 外部ジョブも必要ありません。ポリシーの指定にかかわらず、最新のチェックポイントは決して破棄されません。 これは再開時に読み込まれるチェックポイントだからです — keep_last(0)は 1に設定され、すべてのチェックポイントが経過時間の上限を超えているスレッドでも1つが保持されます。

デフォルトでは無効になっているため、既存のスレッドは履歴全体を保持し、タイムトラベルによって すべてのステップに到達できます。スレッドが長期間存続し、遠い過去まで巻き戻す必要がない場合にポリシーを設定してください。

未宣言のチャネルを拒否する

スキーマで宣言されていないチャネルには、上書きリデューサーが適用されます。これは未知の名前に対する フォールバックだからです。リストチャネルを宣言したグラフが、よく似ているものの異なる名前に書き込んだ場合、 最後の値だけが保持され、何も報告されません。

let graph = graph.with_strict_channels();

未宣言のチャネルに書き込むノードがあると、実行はGraphError::UndeclaredChannelを伴って失敗し、 ノードとチャネルの名前が示されます。デフォルトでは無効であり、グラフがチャネルを1つも宣言していない場合は 何も起こりません。

ADKの統合

GraphAgentはADK Agentトレイトを実装しているため、以下と連携できます。

• Runner: 標準実行にはadk-runnerとともに使用
• Callbacks: before/afterコールバックを完全にサポート
• Sessions: 会話履歴にはadk-sessionと連携
• Streaming: ADK EventStreamを返す

use adk_runner::Runner;

let graph_agent = GraphAgent::builder("workflow")
    .before_agent_callback(|ctx| async {
        println!("Starting graph execution for session: {}", ctx.session_id());
        Ok(())
    })
    .after_agent_callback(|ctx, event| async {
        if let Some(content) = event.content() {
            println!("Graph completed with content");
        }
        Ok(())
    })
    // ... graph definition
    .build()?;

// GraphAgent implements Agent trait - use with Launcher or Runner
// See adk-runner README for Runner configuration

例

このリポジトリにある検証済みのグラフ例:

cargo run --manifest-path examples/tier_examples/standard/Cargo.toml --bin 11-standard-graph
cargo run --manifest-path examples/tier_examples/standard/Cargo.toml --bin 12-standard-sequential
cargo run --manifest-path examples/competitive_graph_resume/Cargo.toml

実際のLLM統合を含む完全なグラフギャラリーはadk-playgroundにあります。

LangGraphとの比較

2つの違いは、明確に述べておく価値があります。run_node_withは子の結果をインラインで返して記録するため、再開された実行では完了済みの子に対して再度コストが発生しません。一方、Sendは作業をスケジューラーに渡し、リデューサーを通じて収集します。また、サブグラフからその親へ移動する機能に相当するものは、ここではまだありません。

---

前へ: ← マルチエージェントシステム | 次へ: リアルタイムエージェント →