Graph-Agenten

Erstellen Sie komplexe, zustandsbehaftete Workflows mithilfe der Orchestrierung im LangGraph-Stil mit nativer ADK-Rust-Integration.

Übersicht

GraphAgent ermöglicht es Ihnen, Workflows als gerichtete Graphen mit Knoten und Kanten zu definieren, die Folgendes unterstützen:

• AgentNode: LLM-Agenten als Graphknoten mit benutzerdefinierten Mappern für Eingaben und Ausgaben einbinden
• Zyklische Workflows: Native Unterstützung für Schleifen und iteratives Schlussfolgern (ReAct-Muster)
• Bedingtes Routing: Dynamisches Kanten-Routing auf Grundlage des Zustands
• Zustandsverwaltung: Typisierter Zustand mit Reduzierern (überschreiben, anhängen, summieren, benutzerdefiniert)
• Checkpointing: Persistenter Zustand für Fehlertoleranz und Human-in-the-Loop
• Streaming: Mehrere Streaming-Modi (Werte, Aktualisierungen, Nachrichten, Debugging)

Der adk-graph-Crate stellt eine Workflow-Orchestrierung im LangGraph-Stil zum Erstellen komplexer, zustandsbehafteter Agenten-Workflows bereit. Er erweitert das ADK-Rust-Ökosystem um graphbasierte Workflow-Funktionen und bewahrt dabei die vollständige Kompatibilität mit dem Agentensystem von ADK.

Wichtige Vorteile:

• Visuelles Workflow-Design: Komplexe Logik als intuitive Graphen aus Knoten und Kanten definieren
• Parallele Ausführung: Mehrere Knoten können gleichzeitig ausgeführt werden, um die Leistung zu verbessern
• Persistenz des Zustands: Integriertes Checkpointing für Fehlertoleranz und Human-in-the-Loop
• LLM-Integration: Native Unterstützung für das Einbinden von ADK-Agenten als Graphknoten
• Flexibles Routing: Statische Kanten, bedingtes Routing und dynamische Entscheidungsfindung

Auswahl zwischen den Workflow-Agenten und dem Graphen

ADK-Rust unterstützt zwei Möglichkeiten zur Orchestrierung. Keine ersetzt die andere, und beide werden weiter gepflegt.

Bevorzugen Sie Workflow-Agenten, wenn die Struktur der Arbeit bekannt und linear ist: Sie sind kürzer zu schreiben, und es muss kein Zustandsschema gepflegt werden. Verwenden Sie den Graphen, wenn der Kontrollfluss von Ergebnissen abhängt oder wenn ein Lauf den Prozess überdauern muss.

Sie lassen sich kombinieren

Die drei Workflow-Agenten implementieren Agent, und AgentNode umschließt jedes Agent, sodass ein Workflow-Agent ein Graphknoten ist:

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 implementiert ebenfalls Agent, daher gilt auch die Umkehrung: Ein Graph kann ein Sub-Agent eines SequentialAgent sein. Verwenden Sie den Graphen für den Teil, der Verzweigungen oder Dauerhaftigkeit benötigt, und die Workflow-Agenten für die Teile, die dies nicht tun.

Was Sie erstellen werden

In diesem Leitfaden erstellen Sie eine Textverarbeitungspipeline, die Übersetzung und Zusammenfassung parallel ausführt:

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

Schlüsselkonzepte:

• Knoten – Verarbeitungseinheiten, die Arbeit ausführen (LLM-Agenten, Funktionen oder benutzerdefinierte Logik)
• Kanten – Kontrollfluss zwischen Knoten (statische Verbindungen oder bedingtes Routing)
• Zustand – Gemeinsam genutzte Daten, die durch den Graphen fließen und zwischen Knoten erhalten bleiben
• Parallele Ausführung – Mehrere Knoten können gleichzeitig ausgeführt werden, um die Leistung zu verbessern

Die Kernkomponenten verstehen

🔧 Knoten: Die Arbeiter In den Knoten findet die eigentliche Arbeit statt. Jeder Knoten kann:

• AgentNode: Einen LLM-Agenten umschließen, um natürliche Sprache zu verarbeiten
• Funktionsknoten: Benutzerdefinierten Rust-Code zur Datenverarbeitung ausführen
• Integrierte Knoten: Vordefinierte Logik wie Zähler oder Validatoren verwenden

Stellen Sie sich Knoten als spezialisierte Arbeiter in einer Fertigungsstraße vor – jeder hat eine bestimmte Aufgabe und Fachkenntnis.

🔀 Kanten: Die Ablaufsteuerung
Kanten bestimmen, wie die Ausführung durch Ihren Graphen verläuft:

• Statische Kanten: Direkte Verbindungen (A → B → C)
• Bedingte Kanten: Dynamisches Routing basierend auf dem Zustand (if sentiment == "positive" → positive_handler)
• Parallele Kanten: Mehrere Pfade von einem Knoten (START → [translator, summarizer])

Kanten sind wie Verkehrssignale und Verkehrsschilder, die den Arbeitsablauf steuern.

💾 Zustand: Der gemeinsame Speicher
Der Zustand ist ein Schlüssel-Wert-Speicher, aus dem alle Knoten lesen und in den sie schreiben können:

• Eingabedaten: Anfangsinformationen, die dem Graphen zugeführt werden
• Zwischenergebnisse: Die Ausgabe eines Knotens wird zur Eingabe eines anderen
• Endausgabe: Das abgeschlossene Ergebnis nach der gesamten Verarbeitung

Der Zustand fungiert wie ein gemeinsames Whiteboard, auf dem Knoten Informationen hinterlassen können, die andere verwenden.

⚡ Parallele Ausführung: Der Geschwindigkeitsboost
Wenn mehrere Kanten von einem Knoten ausgehen, werden die Zielknoten gleichzeitig ausgeführt:

• Schnellere Verarbeitung: Unabhängige Aufgaben werden gleichzeitig ausgeführt
• Ressourceneffizienz: Bessere Auslastung von CPU und E/A
• Skalierbarkeit: Komplexere Workflows ohne lineare Verlangsamung bewältigen

Das ist so, als würden mehrere Mitarbeiter gleichzeitig verschiedene Teile einer Aufgabe bearbeiten, anstatt in einer Warteschlange zu warten.

---

Schnellstart

1. Erstellen Sie Ihr Projekt

cargo new graph_demo
cd graph_demo

Fügen Sie Abhängigkeiten zu Cargo.toml hinzu:

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

Erstellen Sie .env mit Ihrem API-Schlüssel:

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

2. Beispiel für parallele Verarbeitung

Hier ist ein vollständiges, funktionsfähiges Beispiel, das Text parallel verarbeitet:

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(())
}

Beispielausgabe:

=== 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.

Funktionsweise der Graphausführung

Das große Ganze

Graphagenten werden in Super-Schritten ausgeführt – alle bereiten Knoten laufen parallel, anschließend wartet der Graph, bis alle abgeschlossen sind, bevor der nächste Schritt beginnt:

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 ✅

Zustandsfluss durch Knoten

Jeder Knoten kann den gemeinsamen Zustand lesen und in ihn schreiben:

┌─────────────────────────────────────────────────────────────────────┐
│ 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..."          │
│   }                                                                 │
│                                                                     │
└─────────────────────────────────────────────────────────────────────┘

Was es funktionieren lässt

---

Bedingtes Routing mit LLM-Klassifizierung

Erstellen Sie intelligente Routing-Systeme, in denen LLMs den Ausführungspfad bestimmen:

Visualisierung: Sentiment-basiertes Routing

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

Vollständiger Beispielcode

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(())
}

Beispielablauf:

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-Muster: Schlussfolgern + Handeln

Erstellen Sie Agents, die Tools iterativ verwenden können, um komplexe Probleme zu lösen:

Visualisierung: ReAct-Zyklus

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

Vollständiges ReAct-Beispiel

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(())
}

Beispielablauf:

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

Verpackt jedes ADK Agent (typischerweise LlmAgent) als Graphknoten:

Was der Agent sieht

Ein Agent innerhalb eines Graphen wird unter einem Kontext ausgeführt, der aus dem Aufruf abgeleitet ist, durch den der Graph gestartet wurde, und verhält sich daher genauso wie außerhalb eines Graphen. Wenn ein Runner ein GraphAgent aufruft, werden die Identität und Services des Aufrufers automatisch weitergegeben:

Ein direkt aufgerufener Graph – graph.invoke(state, ExecutionConfig::new("thread")) –
hat keine Invocation, die er erben kann. Das ist der Standalone-Modus: Der Knoten erhält user_id = "graph_user", app_name = "graph_app", den Branch main, keine Secrets und keinen Speicher. Dieser Modus ist dafür gedacht, einen Graphen außerhalb einer Runner auszuführen, und kein Fallback, auf den man in der Produktion zurückgreifen sollte.

Um manuell eine Verbindung herzustellen – beispielsweise wenn ein Graph über einen eigenen Executor gesteuert wird –, übergeben Sie die Invocation explizit:

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

Hinweis: Der Knoten läuft weiterhin in seiner eigenen In-Memory-Graph-Session, daher ist der Gesprächsverlauf des Agenten innerhalb eines Knotens auf diesen Knoten beschränkt, statt an die Session des Aufrufers angehängt zu werden.

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
    });

Funktionsknoten

Einfache asynchrone Funktionen, die den Zustand verarbeiten:

.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))
})

Kantentypen

Statische Kanten

Direkte Verbindungen zwischen Knoten:

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

Bedingte Kanten

Dynamisches Routing basierend auf dem Zustand:

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

Routing innerhalb eines Knotens

Eine bedingte Kante legt ihre Ziele beim Erstellen des Graphen fest. NodeOutput::with_goto tut dies nicht: Ein Knoten schreibt den Zustand und benennt seine Nachfolger im selben Schritt, und er kann jeden Knoten im Graphen benennen, einschließlich eines Knotens, zu dem keine Kante von ihm führt.

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]))
}

Die von einem goto erzeugte Frontier wird wie jede andere als Checkpoint gespeichert, sodass ein pausierter Lauf in dem vom goto ausgewählten Knoten fortgesetzt wird.

Hinweis: Verwenden Sie add_conditional_edges, wenn die möglichen Ziele beim Erstellen des Graphen bekannt sind — die Kanten erscheinen dann in einem gerenderten Diagramm. Verwenden Sie ein goto, wenn die Auswahl zum Knoten gehört.

Router-Hilfsfunktionen

Verwenden Sie integrierte Router für gängige Muster:

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

Parallele Ausführung

Mehrere Kanten von einem einzelnen Knoten werden parallel ausgeführt:

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 wird einmal ausgeführt, nachdem alle drei Zweige eingetroffen sind. Nichts im obigen Code fordert dies ausdrücklich an: Ein Knoten mit mehr als einer eingehenden direkten Kante wird beim Kompilieren automatisch zurückgestellt. Zweige unterschiedlicher Länge werden daher ohne Konfiguration korrekt zusammengeführt.

Aus der Art und Weise, wie die Anzahl ermittelt wird, ergeben sich zwei Details:

Zyklische Graphen (ReAct-Muster)

Erstelle iterative Reasoning-Agenten mit Zyklen:

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

Supervisor für mehrere Agenten

Leite Aufgaben an spezialisierte Agenten weiter:

// 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()?;

Zustandsverwaltung

Zustandsschema mit Reducern

Steuere, wie Zustandsaktualisierungen zusammengeführt werden:

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-Typen

Aktualisierungsreihenfolge

Knoten in einem Super-Schritt werden gleichzeitig ausgeführt und schließen ihre Arbeit in der jeweils benötigten Zeit ab. Ihre Zustandsaktualisierungen werden in Knoten-Namensreihenfolge angewendet, nicht in der Reihenfolge, in der die Knoten fertig werden.

Die Reihenfolge ist relevant, sobald ein Reducer nicht kommutativ ist. Append erstellt ein Array, daher ist die Reihenfolge das Ergebnis; auch ein Custom-Reducer kann von der Reihenfolge abhängen. Die Sortierung nach Knotennamen macht eine Ausführung reproduzierbar: Derselbe Graph und dieselbe Eingabe liefern unabhängig vom Timing einer langsamen Abhängigkeit denselben Zustand.

Wenn mehrere Kanäle von einem Knoten beschrieben werden, werden sie in Kanal-Namensreihenfolge angewendet. Kanäle interagieren nicht miteinander, daher ist dies nur für das Lesen eines Traces relevant.

Checkpointing

Aktiviere persistenten Zustand für Fehlertoleranz und Human-in-the-Loop:

Im Speicher (Entwicklung)

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 (Produktion)

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

Was ein Checkpoint aufzeichnet

Ein Checkpoint speichert den akkumulierten Zustand, die Schrittanzahl und die Frontier — die Knoten, die noch ausgeführt werden müssen. Er wird geschrieben, nachdem die Frontier fortgeschritten ist. Dadurch wird beim Fortsetzen niemals ein Knoten erneut ausgeführt, der bereits abgeschlossen wurde, und seine Aktualisierungen werden niemals doppelt angewendet. Bei einer abgeschlossenen Ausführung wird eine leere Frontier als Checkpoint gespeichert. Dadurch gibt das Fortsetzen eines abgeschlossenen Threads den endgültigen Zustand zurück, anstatt den Graphen neu zu starten.

In zwei Fällen wird absichtlich die Frontier gespeichert, die gerade ausgeführt wurde, und nicht die nächste, da der unterbrochene Knoten seine Aktualisierungen noch nicht erzeugt hat und beim Fortsetzen erneut ausgeführt werden muss:

Gestreamte Ausführungen werden nach demselben Zeitplan wie blockierende Ausführungen mit Prüfpunkten versehen, auch wenn eine Unterbrechung den Stream beendet, sodass eine Pause mit Beteiligung eines Menschen in beiden Ausführungsmodi fortgesetzt werden kann.

Prüfpunktverlauf (Zeitreise)

Nur Lesezugriffe. TimeTravelHandle::state_history(from, to) gibt den Zustand zurück, der bei jedem Schritt mit Prüfpunkt gespeichert wurde. Es wird nichts ausgeführt – kein Knoten wird ausgeführt, kein Ereignis wird neu erzeugt und kein Seiteneffekt wiederholt. Um die Ausführung ab einem Punkt im Verlauf erneut zu starten, verwenden Sie fork_at, um von diesem Prüfpunkt einen Zweig zu erstellen, und rufen Sie den Graphen auf dem abgezweigten Thread auf. Die Methode hieß zuvor replay und wurde als erneute Ausführung des Graphen dokumentiert, was sie nie getan hat.

Prüfpunkte ermöglichen außerdem eine dauerhafte Wiederaufnahme – wenn eine Graphausführung abstürzt oder der Prozess neu gestartet wird, wird die Ausführung am letzten persistent gespeicherten Prüfpunkt fortgesetzt, anstatt von vorne zu beginnen. Verwenden Sie SqliteCheckpointer (die Funktion sqlite) für absturzsichere Persistenz. MemoryCheckpointer hält Prüfpunkte im Prozess, sodass sie einen Neustart nicht überleben. Dies sind die beiden Backends, die dieses Crate bereitstellt; implementieren Sie das Trait Checkpointer für alle anderen.

// 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);
}

Mensch-in-der-Schleife

Pausieren Sie die Ausführung zur menschlichen Genehmigung mithilfe dynamischer Unterbrechungen:

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);
    }
}

Statische Unterbrechungen

Verwenden Sie interrupt_before oder interrupt_after für obligatorische Pausenpunkte:

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

Gestreamte Ausführung

Streamen Sie Ereignisse, während der Graph ausgeführt wird:

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),
        _ => {}
    }
}

Streaming-Modi

Messages mode liest Token aus Node::execute_stream, sobald sie erzeugt werden. In diesem Modus wird jeder Knoten pro Super-Schritt einmal ausgeführt: Der Knoten meldet seine Zustandsaktualisierungen im Stream als StreamEvent::Updates-Ereignis, und der Executor wendet diese an, anstatt den Knoten ein zweites Mal auszuführen, um sie zu sammeln. Dies ist besonders für AgentNode relevant, da eine zweite Ausführung einen zweiten abgerechneten Modellaufruf pro Knoten bedeuten würde.

Wichtig: Ein benutzerdefiniertes Node, das execute_stream überschreibt, muss ein StreamEvent::Updates-Ereignis liefern, das seine Zustandsaktualisierungen enthält. Andernfalls streamt der Knoten zwar Ereignisse, trägt aber im Messages-Modus keinen Zustand bei. Das standardmäßige execute_stream, das execute umschließt, übernimmt dies für Sie.

Timeout-Richtlinien gelten für die gestreamte Ausführung selbst. Bei einem Stream bedeutet idle_timeout, dass innerhalb des Limits kein Ereignis erzeugt wurde.

Untergraphen

Ein kompiliertes Diagramm wird über SubgraphNode als Knoten eines anderen ausgeführt. Der innere Graph behält seine eigenen Kanäle, Kanten und Unterbrechungspunkte und tauscht benannte Kanäle mit seinem übergeordneten Graphen aus.

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()?;

Die letzte Zeile ist der entscheidende Unterschied: Beide Schemata sind verfügbar, bevor irgendetwas ausgeführt wird. Daher kann eine Zuordnung, die einen Kanal benennt, den keine der beiden Seiten deklariert, keine Ausführung erreichen und als fehlender Wert sichtbar werden. Ein Untergraph, der überhaupt nichts austauscht, wird aus demselben Grund abgelehnt, da er seinen übergeordneten Graphen nicht beeinflussen könnte.

Fortsetzen einer Pause innerhalb eines Untergraphen

Es ist nichts Zusätzliches erforderlich. Wird der übergeordnete Graph erneut im selben Thread aufgerufen, wird der Untergraph erneut betreten. Dieser findet seinen eigenen Checkpoint auf <parent thread>/<node name> und setzt die Ausführung an der Stelle fort, an der sie angehalten wurde. Arbeit, die der Untergraph vor der Pause abgeschlossen hat, wird nicht wiederholt, und eine Pause mehrere Ebenen tiefer wird auf dieselbe Weise fortgesetzt – die Nachricht nennt jede Ebene, die sie durchlaufen hat.

Ein Untergraph, der ein Unterbrechungstor deklariert, aber keinen Checkpointer besitzt, wird beim Kompilieren des übergeordneten Graphen abgelehnt: Er würde an seinem ersten Knoten erneut betreten und für seine bereits abgeschlossene Arbeit ein zweites Mal bezahlen.

Da beide Graphen echte Checkpointer enthalten, bleibt dies auch nach einem Neustart des Prozesses erhalten: Ein neuer Satz von Graphobjekten, die sich nur die Datenbanken teilen, setzt denselben Lauf fort.

Übergabe der Kontrolle an den übergeordneten Graphen

Ein Knoten innerhalb eines Subgraphen kann seinen eigenen Graphen beenden und einen Knoten des Graphen benennen, der ihn enthält:

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

Der Subgraph wird wie gewohnt beendet und projiziert seine Ausgabekanäle, anschließend setzt der übergeordnete Graph bei escalate fort, anstatt den vom Subgraph-Knoten deklarierten Kanten zu folgen. Der übergeordnete Graph validiert das Ziel, da nur er seine eigenen Knoten kennt.

Zuverlässigkeits- und Kostenkontrollen

Alle diese Optionen sind standardmäßig deaktiviert, sodass sich ein Graph so verhält wie zuvor, solange keine davon aktiviert wurde.

Wiederholungen pro Knoten

Ein vorübergehender Fehler — eine Ratenbegrenzung, eine unterbrochene Verbindung — beendet andernfalls den Lauf.

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

Die Verzögerung wächst um backoff_factor, wird auf max_delay begrenzt und anschließend mit Jitter versehen. Ein Knoten ohne Richtlinie wird einmal ausgeführt, sodass Wiederholungen ausdrücklich aktiviert werden müssen. Eine Richtlinie aus RetryPolicy::default() erlaubt zehn Versuche, deren neun Wartezeiten sich auf etwa 243 Sekunden summieren — senken Sie max_attempts, wenn ein Aufrufer auf die Antwort wartet.

Eine Unterbrechung wird unabhängig davon, was retry_on besagt, niemals wiederholt: Eine Pause ist kein Fehler. Die Anzahl der Versuche wird im Checkpoint gespeichert, sodass ein fortgesetzter Lauf das verbleibende Budget verwendet, anstatt es neu zu starten.

Begrenzung der Nebenläufigkeit

Eine weit aufgefächerte Verteilung startet ihre gesamte Frontier auf einmal, wodurch ein Verbindungspool erschöpft oder eine Ratenbegrenzung des Anbieters ausgelöst werden kann.

let graph = graph.with_max_concurrency(4);

Knoten jenseits des Limits warten auf einen freien Platz. Die Reihenfolge der Aufnahme entspricht der nach Namen sortierten Frontier und hängt daher nicht vom Timing ab. Imperative Aufrufe von untergeordneten Knoten fallen nicht unter dieses Budget, da ein übergeordneter Knoten auf seine untergeordneten Knoten wartet, während er seinen eigenen Platz belegt.

Zeitüberschreitungen pro Knoten

Ein TimeoutPolicy begrenzt einen einzelnen Versuch und legt zusammen mit idle_timeout fest, wie lange ein Knoten ohne Fortschrittsmeldung auskommen darf. Wird einer der beiden Werte überschritten, ergibt sich GraphError::NodeTimedOut, worauf eine Wiederholungsrichtlinie reagieren kann.

Direktes Aufrufen eines Knotens

Wenn die Anzahl der Teilaufgaben aus dem Zustand statt aus der Struktur des Graphen stammt, kann ein Knoten selbst einen anderen Knoten aufrufen:

use adk_graph::child::RunNodeOptions;

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

Das Ziel benötigt keine Kante. Jedes abgeschlossene Kind wird unter <parent>/<child>@<run_id> aufgezeichnet, sodass ein fortgesetzter Lauf die aufgezeichnete Antwort zurückgibt, statt das Kind erneut auszuführen — was wichtig ist, wenn das Kind einen Modellaufruf verursacht.

Knotencaching

cache_policy für einen Knoten versieht dessen Ergebnis anhand des Knotennamens und des aktuellen Zustands mit einem Schlüssel, optional mit einem TTL, sodass unveränderte Eingaben die Arbeit überspringen. Erfordert das Feature node-cache; ein von Redis unterstützter Speicher ist über redis-cache verfügbar.

Delta-Checkpoints

Das Feature delta speichert die Differenz zwischen Super-Schritten statt des gesamten Zustands, was wichtig ist, wenn der Zustand groß ist und es viele Schritte gibt.

Zeitreise

Mit dem Feature time-travel gibt graph.time_travel(thread_id)? ein Handle über den Checkpoint-Verlauf eines Threads zurück: Schritte auflisten, den Zustand zu einem bestimmten Schritt lesen oder einen Checkpoint mit fork_at abzweigen, um daraus einen neuen Thread zu erstellen. Der Aufruf gibt Result zurück, weil jede Operation Checkpoints liest; daher meldet ein Graph ohne Checkpointer GraphError::CheckpointError, statt eine Panik auszulösen.

Graphweite Standardwerte

Dieselbe Wiederholung über zwanzig Knoten zu wiederholen, kann durch Auslassungen leicht fehleranfällig werden.

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

Ein Wert auf Knotenebene hat immer Vorrang. NodeDefaults enthält außerdem ein Timeout und einen Fehler-Handler.

Wiederherstellung nach einem Knotenfehler

Sobald das Wiederholungslimit eines Knotens aufgebraucht ist, kann ein Handler aufzeichnen, was geschehen ist, und einen Wiederherstellungsknoten benennen, statt den Lauf zu beenden:

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"]))
});

Die Rückgabe von Err beendet den Lauf wie zuvor. Eine Unterbrechung erreicht niemals einen Handler, weil eine Pause kein Fehler ist.

Wachstum von Checkpoints begrenzen

Ein Thread sammelt pro Super-Schritt einen Checkpoint. Ein Lauf, der tagelang besteht, wächst daher unbegrenzt, was Speicherplatz kostet und list verlangsamt.

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

Das Bereinigen erfolgt nach jedem Speichern, sodass die Kosten proportional zum Lauf bleiben und kein externer Job erforderlich ist. Der neueste Checkpoint wird niemals verworfen, unabhängig davon, was die Richtlinie vorgibt, weil er derjenige ist, den eine Fortsetzung lädt — keep_last(0) wird auf eins gesetzt, und ein Thread, dessen sämtliche Checkpoints älter als das Alterslimit sind, behält einen.

Standardmäßig deaktiviert, sodass ein bestehender Thread seine vollständige Historie behält und Zeitreisen weiterhin jeden Schritt erreichen können. Legen Sie eine Richtlinie fest, wenn ein Thread lange läuft und Sie nicht weit zurückspringen müssen.

Nicht deklarierte Kanäle ablehnen

Ein Kanal, den das Schema nicht deklariert, verwendet den Überschreib-Reducer, weil dieser der Fallback für einen unbekannten Namen ist. Ein Graph, der einen Listenkanal deklariert und dann in einen ähnlich klingenden Namen schreibt, behält nur den letzten Wert und meldet nichts.

let graph = graph.with_strict_channels();

Ein Knoten, der in einen nicht deklarierten Kanal schreibt, schlägt anschließend mit GraphError::UndeclaredChannel fehl und nennt dabei den Knoten und den Kanal. Standardmäßig deaktiviert und wirkungslos, wenn ein Graph überhaupt keine Kanäle deklariert.

ADK-Integration

GraphAgent implementiert das ADK-Trait Agent und funktioniert daher mit:

• Runner: Zusammen mit adk-runner für die Standardausführung verwenden
• Callbacks: Vollständige Unterstützung für Vorher-/Nachher-Callbacks
• Sessions: Funktioniert mit adk-session für den Gesprächsverlauf
• Streaming: Gibt ADK EventStream zurück

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

Beispiele

Validierte Graphbeispiele in diesem Repository:

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

Die vollständige Graphgalerie mit echter LLM-Integration befindet sich unter adk-playground.

Vergleich mit LangGraph

Zwei Unterschiede sollten klar benannt werden. run_node_with gibt das Ergebnis des untergeordneten Elements inline zurück und zeichnet es auf, sodass ein fortgesetzter Lauf nicht erneut für ein bereits abgeschlossenes untergeordnetes Element bezahlt, während Send die Arbeit an den Scheduler übergibt und sie über einen Reducer sammelt. Und ein Wechsel von einem Teilgraphen zu seinem übergeordneten Graphen hat hier noch kein Gegenstück.

---

Zurück: ← Multi-Agent Systems | Weiter: Realtime Agents →