Agents en graphe

Construisez des flux de travail complexes et avec état à l’aide d’une orchestration de type LangGraph avec une intégration native de ADK-Rust.

Vue d’ensemble

GraphAgent vous permet de définir des flux de travail sous forme de graphes orientés comportant des nœuds et des arêtes, avec notamment :

• AgentNode : encapsuler des agents LLM en tant que nœuds de graphe avec des mappeurs d’entrée/sortie personnalisés
• Flux de travail cycliques : prise en charge native des boucles et du raisonnement itératif (modèle ReAct)
• Routage conditionnel : routage dynamique des arêtes en fonction de l’état
• Gestion de l’état : état typé avec des réducteurs (remplacement, ajout, somme, personnalisé)
• Points de contrôle : état persistant pour la tolérance aux pannes et la participation humaine
• Diffusion en continu : plusieurs modes de diffusion (valeurs, mises à jour, messages, débogage)

La crate adk-graph fournit une orchestration de flux de travail de type LangGraph pour créer des flux de travail d’agents complexes et avec état. Elle apporte des fonctionnalités de flux de travail fondées sur des graphes à l’écosystème ADK-Rust, tout en conservant une compatibilité complète avec le système d’agents de ADK.

Principaux avantages :

• Conception visuelle des flux de travail : définir une logique complexe sous forme de graphes intuitifs de nœuds et d’arêtes
• Exécution parallèle : plusieurs nœuds peuvent s’exécuter simultanément pour de meilleures performances
• Persistance de l’état : points de contrôle intégrés pour la tolérance aux pannes et la participation humaine
• Intégration de LLM : prise en charge native de l’encapsulation des agents ADK en tant que nœuds de graphe
• Routage flexible : arêtes statiques, routage conditionnel et prise de décision dynamique

Choisir entre les agents de flux de travail et le graphe

ADK-Rust prend en charge deux façons d’orchestrer. Aucune ne remplace l’autre, et toutes deux sont maintenues.

Privilégiez les agents de workflow lorsque la structure du travail est connue et linéaire : ils sont plus courts à écrire et il n’est pas nécessaire de gérer un schéma d’état. Optez pour le graphe lorsque le flux de contrôle dépend des résultats ou lorsqu’une exécution doit survivre au processus.

Ils se composent

Les trois agents de workflow implémentent Agent, et AgentNode encapsule n’importe quel Agent, de sorte qu’un agent de workflow est un nœud de graphe :

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 implémente également Agent, l’inverse est donc aussi vrai : un graphe peut être un sous-agent d’un SequentialAgent. Utilisez le graphe pour la partie qui nécessite des embranchements ou de la persistance, et les agents de workflow pour les parties qui n’en ont pas besoin.

Ce que vous allez construire

Dans ce guide, vous allez créer un pipeline de traitement de texte qui exécute en parallèle la traduction et la synthèse :

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

Concepts clés :

• Nœuds - Unités de traitement qui effectuent le travail (agents LLM, fonctions ou logique personnalisée)
• Arêtes - Flux de contrôle entre les nœuds (connexions statiques ou routage conditionnel)
• État - Données partagées qui circulent dans le graphe et persistent entre les nœuds
• Exécution parallèle - Plusieurs nœuds peuvent s’exécuter simultanément pour de meilleures performances

Comprendre les composants principaux

🔧 Nœuds : les exécutants C’est dans les nœuds que le travail réel est effectué. Chaque nœud peut :

• AgentNode : encapsuler un agent LLM pour traiter le langage naturel
• Nœud de fonction : exécuter du code Rust personnalisé pour le traitement des données
• Nœuds intégrés : utiliser une logique prédéfinie, comme des compteurs ou des validateurs

Considérez les nœuds comme des exécutants spécialisés sur une chaîne de montage : chacun possède une tâche et une expertise spécifiques.

🔀 Arêtes : le contrôle du flux
Les arêtes déterminent la manière dont l’exécution se déplace dans votre graphe :

• Arêtes statiques : connexions directes (A → B → C)
• Arêtes conditionnelles : routage dynamique basé sur l’état (if sentiment == "positive" → positive_handler)
• Arêtes parallèles : plusieurs chemins à partir d’un nœud (START → [translator, summarizer])

Les arêtes sont comme des feux de circulation et des panneaux routiers qui dirigent le flux de travail.

💾 État : la mémoire partagée
L’état est un magasin clé-valeur dans lequel tous les nœuds peuvent lire et écrire :

• Données d’entrée : informations initiales transmises au graphe
• Résultats intermédiaires : la sortie d’un nœud devient l’entrée d’un autre
• Sortie finale : le résultat terminé une fois tout le traitement effectué

L’état agit comme un tableau blanc partagé sur lequel les nœuds peuvent laisser des informations à l’intention des autres.

⚡ Exécution parallèle : le gain de vitesse
Lorsque plusieurs arêtes partent d’un nœud, les nœuds cibles s’exécutent simultanément :

• Traitement plus rapide : les tâches indépendantes s’exécutent en même temps
• Efficacité des ressources : meilleure utilisation du CPU et des entrées-sorties
• Évolutivité : prise en charge de workflows plus complexes sans ralentissement linéaire

C’est comme disposer de plusieurs travailleurs qui s’occupent simultanément de différentes parties d’un travail au lieu d’attendre leur tour.

---

Démarrage rapide

1. Créer votre projet

cargo new graph_demo
cd graph_demo

Ajoutez les dépendances à 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"

Créez .env avec votre clé API :

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

2. Exemple de traitement parallèle

Voici un exemple complet et fonctionnel qui traite du texte en parallèle :

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

Résultat de l’exemple :

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

Fonctionnement de l’exécution d’un graphe

Vue d’ensemble

Les agents de graphe s’exécutent en super-étapes : tous les nœuds prêts s’exécutent en parallèle, puis le graphe attend qu’ils soient tous terminés avant de passer à l’étape suivante :

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 ✅

Flux de l’état à travers les nœuds

Chaque nœud peut lire l’état partagé et y écrire :

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

Ce qui le rend efficace

---

Routage conditionnel avec la classification LLM

Créez des systèmes de routage intelligents dans lesquels LLMs déterminent le chemin d’exécution :

Visuel : routage basé sur le sentiment

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

Exemple de code complet

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

Flux de l’exemple :

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

Modèle ReAct : raisonnement + action

Créez des agents capables d’utiliser des outils de manière itérative pour résoudre des problèmes complexes :

Visuel : cycle ReAct

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

Exemple complet de 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(())
}

Flux de l’exemple :

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

Encapsule tout ADK Agent (généralement LlmAgent) en tant que nœud de graphe :

Ce que voit l’agent

Un agent au sein d’un graphe s’exécute dans un contexte dérivé de l’invocation qui a démarré le graphe ; il se comporte donc de la même manière qu’en dehors de celui-ci. Lorsqu’un Runner invoque un GraphAgent, l’identité et les services de l’appelant sont automatiquement transmis :

Un graphe invoqué directement — graph.invoke(state, ExecutionConfig::new("thread")) — n’a aucune invocation à hériter. C’est le mode autonome : le nœud reçoit user_id = "graph_user", app_name = "graph_app", la branche main, aucun secret et aucune mémoire. Il s’agit d’un mode délibéré pour exécuter un graphe en dehors d’un Runner, et non d’une solution de repli à utiliser en production.

Pour faire le lien manuellement — par exemple lorsque vous pilotez un graphe depuis votre propre exécuteur — transmettez explicitement l’invocation :

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

Remarque : le nœud s’exécute toujours dans sa propre session de graphe en mémoire ; l’historique des conversations de l’agent à l’intérieur d’un nœud est donc limité à ce nœud, au lieu d’être ajouté à la session de l’appelant.

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

Nœuds de fonction

Fonctions asynchrones simples qui traitent l’état :

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

Types d’arêtes

Arêtes statiques

Connexions directes entre les nœuds :

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

Arêtes conditionnelles

Routage dynamique fondé sur l’état :

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

Routage depuis l’intérieur d’un nœud

Une arête conditionnelle fixe ses cibles lors de la construction du graphe. NodeOutput::with_goto ne le fait pas : un nœud écrit l’état et nomme ses successeurs lors de la même étape, et il peut nommer n’importe quel nœud du graphe, y compris un nœud vers lequel il n’a aucune arête.

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

La frontière produite par un goto fait l’objet d’un point de contrôle comme toute autre frontière ; ainsi, une exécution mise en pause reprend dans le nœud choisi par le goto.

Remarque : utilisez add_conditional_edges lorsque les cibles possibles sont connues lors de la construction du graphe — les arêtes apparaissent alors dans un diagramme généré. Utilisez un goto lorsque le choix appartient au nœud.

Assistants de routage

Utilisez les routeurs intégrés pour les schémas courants :

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

Exécution parallèle

Plusieurs arêtes issues d’un même nœud s’exécutent en parallèle :

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 s’exécute une seule fois, après l’arrivée des trois branches. Rien dans le code ci-dessus ne le demande : lors de la compilation, un nœud ayant plusieurs arêtes directes entrantes est automatiquement différé. Les branches de longueurs différentes se rejoignent donc correctement sans configuration.

Deux détails découlent de la manière dont le décompte est effectué :

Graphes cycliques (motif ReAct)

Construisez des agents de raisonnement itératif avec des cycles :

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

Superviseur multi-agents

Acheminez les tâches vers des agents spécialisés :

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

Gestion de l’état

Schéma d’état avec des réducteurs

Contrôlez la manière dont les mises à jour de l’état sont fusionnées :

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

Types de réducteurs

Ordre des mises à jour

Les nœuds d’une même super-étape s’exécutent en parallèle et terminent dans un ordre quelconque, selon la durée de leur travail. Leurs mises à jour d’état sont appliquées dans l’ordre des noms de nœuds, et non dans l’ordre de fin des nœuds.

L’ordre est important lorsqu’un réducteur n’est pas commutatif. Append construit un tableau, l’ordre fait donc partie du résultat ; un réducteur Custom peut également être sensible à l’ordre. Le tri par nom de nœud rend l’exécution reproductible : le même graphe et la même entrée produisent le même état, quels que soient les délais d’une dépendance lente.

Lorsque plusieurs canaux sont écrits par un même nœud, ils sont appliqués dans l’ordre des noms de canaux. Les canaux n’interagissent pas ; cela n’a donc d’importance que pour la lecture d’une trace.

Création de points de contrôle

Activez la persistance de l’état pour la tolérance aux pannes et la supervision humaine :

En mémoire (développement)

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

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

Ce qu’enregistre un point de contrôle

Un point de contrôle stocke l’état accumulé, le numéro de l’étape et la frontière — les nœuds qui doivent encore être exécutés. Il est écrit après la progression de la frontière ; ainsi, la reprise ne réexécute jamais un nœud déjà terminé et n’applique jamais ses mises à jour deux fois. Une exécution qui se termine enregistre une frontière vide ; la reprise d’un fil d’exécution terminé renvoie donc l’état final au lieu de redémarrer le graphe.

Dans deux cas, la frontière enregistrée est intentionnellement celle qui était en cours d’exécution plutôt que la suivante, car le nœud interrompu n’a pas encore produit ses mises à jour et doit être exécuté à nouveau lors de la reprise :

Les exécutions en flux créent des points de contrôle selon le même calendrier que les exécutions bloquantes, y compris lorsqu’une interruption met fin au flux ; une pause avec intervention humaine peut donc être reprise dans l’un ou l’autre mode d’exécution.

Historique des points de contrôle (voyage dans le temps)

Lectures uniquement. TimeTravelHandle::state_history(from, to) renvoie l’état qui a été stocké à chaque étape faisant l’objet d’un point de contrôle. Il n’exécute rien — aucun nœud ne s’exécute, aucun événement n’est régénéré et aucun effet secondaire ne se répète. Pour réexécuter à partir d’un point de l’historique, utilisez fork_at pour créer une branche à partir de ce point de contrôle et invoquer le graphe sur le fil dérivé. La méthode s’appelait auparavant replay et était documentée comme réexécutant le graphe, ce qu’elle n’a jamais fait.

Les points de contrôle permettent également une reprise durable : si l’exécution d’un graphe échoue ou si le processus redémarre, l’exécution reprend à partir du dernier point de contrôle persistant au lieu de recommencer. Utilisez SqliteCheckpointer (la fonctionnalité sqlite) pour une persistance résistante aux défaillances. MemoryCheckpointer conserve les points de contrôle dans le processus ; ils ne survivent donc pas à un redémarrage. Ce sont les deux backends fournis par cette crate ; implémentez le trait Checkpointer pour tout autre backend.

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

Intervention humaine

Mettez l’exécution en pause pour obtenir une approbation humaine à l’aide d’interruptions dynamiques :

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

Interruptions statiques

Utilisez interrupt_before ou interrupt_after pour les points de pause obligatoires :

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

Exécution en flux

Diffusez les événements au fur et à mesure de l’exécution du graphe :

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

Modes de flux

Messages lit les jetons depuis Node::execute_stream au fur et à mesure de leur production. Dans ce mode, chaque nœud s’exécute une seule fois par super-étape : le nœud signale ses mises à jour d’état dans le flux sous la forme d’un événement StreamEvent::Updates, et l’exécuteur les applique au lieu d’exécuter le nœud une seconde fois pour les recueillir. Cela est particulièrement important pour AgentNode, où une seconde exécution entraînerait un second appel de modèle facturé par nœud.

Important : un Node personnalisé qui surcharge execute_stream doit produire un événement StreamEvent::Updates contenant ses mises à jour d’état. Sans cela, le nœud diffuse des événements, mais ne contribue à aucun état en mode Messages. Le execute_stream par défaut, qui encapsule execute, s’en charge pour vous.

Les politiques de délai d’expiration s’appliquent à l’exécution diffusée elle-même. Pour un flux, idle_timeout signifie qu’aucun événement n’a été produit dans le délai imparti.

Sous-graphes

Un graphe compilé s’exécute comme un nœud d’un autre graphe via SubgraphNode. Le graphe interne conserve ses propres canaux, arêtes et points d’interruption, et échange des canaux nommés avec son parent.

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

Cette dernière ligne constitue la différence à connaître : les deux schémas sont disponibles avant toute exécution, de sorte qu’un mappage nommant un canal qu’aucun des deux ne déclare ne peut atteindre une exécution ni se manifester comme une valeur absente. Un sous-graphe qui n’échange absolument rien est rejeté de la même manière, car il ne pourrait pas avoir d’incidence sur son parent.

Reprendre une pause à l’intérieur d’un sous-graphe

Rien de plus n’est nécessaire. Invoquer à nouveau le parent sur le même fil réintègre le sous-graphe, qui retrouve son propre point de contrôle sur <parent thread>/<node name> et reprend là où il s’était arrêté. Le travail terminé par le sous-graphe avant la pause n’est pas répété, et une pause survenue plusieurs niveaux plus bas reprend de la même manière — le message nomme chaque niveau qu’il a traversé.

Un sous-graphe qui déclare une barrière d’interruption, mais ne possède aucun point de contrôle, est rejeté lors de la compilation du parent : il réintégrerait son premier nœud et paierait une seconde fois le coût du travail déjà terminé.

Comme les deux graphes possèdent de véritables points de contrôle, cela survit à un redémarrage du processus : un nouvel ensemble d’objets de graphe partageant uniquement les bases de données reprend la même exécution.

Redonner le contrôle au parent

Un nœud au sein d’un sous-graphe peut terminer son propre graphe et désigner un nœud du graphe qui le contient :

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

Le sous-graphe s’achève et projette ses canaux de sortie comme d’habitude, puis le parent poursuit à escalate au lieu de suivre les arêtes déclarées du nœud du sous-graphe. Le parent valide la cible, car il est le seul à connaître ses propres nœuds.

Contrôles de fiabilité et de coûts

Chacun de ces contrôles est désactivé par défaut ; un graphe se comporte donc comme auparavant tant que vous n’en avez pas activé un.

Nouvelle tentative par nœud

Une défaillance temporaire — une limitation de débit, une connexion interrompue — met sinon fin à l’exécution.

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

Le délai augmente selon backoff_factor, est plafonné à max_delay, puis une temporisation aléatoire lui est appliquée. Un nœud sans politique ne s’exécute qu’une seule fois, afin que les nouvelles tentatives restent facultatives. Une politique de RetryPolicy::default() autorise dix tentatives, dont les neuf délais totalisent environ 243 secondes — réduisez max_attempts lorsqu’un appelant attend la réponse.

Une interruption ne fait jamais l’objet d’une nouvelle tentative, quels que soient les indications de retry_on : une pause n’est pas une défaillance. Le nombre de tentatives est enregistré dans le point de contrôle ; une exécution reprise poursuit donc le budget au lieu de le réinitialiser.

Limiter la concurrence

Une distribution largement ramifiée envoie toute sa frontière en une seule fois, ce qui peut épuiser un pool de connexions ou déclencher une limitation de débit du fournisseur.

let graph = graph.with_max_concurrency(4);

Les nœuds au-delà de la limite attendent qu’une place se libère. L’ordre d’admission correspond à la frontière triée par nom ; il ne dépend donc pas du moment d’exécution. Les invocations impératives d’enfants ne sont pas incluses dans ce budget, car un parent attend ses enfants tout en conservant sa propre place.

Délais d’expiration par nœud

Un TimeoutPolicy limite une tentative unique et, avec idle_timeout, définit combien de temps un nœud peut rester sans signaler de progression. Le dépassement de l’une ou l’autre valeur produit GraphError::NodeTimedOut, sur lequel une stratégie de nouvelle tentative peut ensuite agir.

Invoquer directement un nœud

Lorsque le nombre de sous-tâches provient de l’état plutôt que de la structure du graphe, un nœud peut invoquer lui-même un autre nœud :

use adk_graph::child::RunNodeOptions;

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

La cible n’a besoin d’aucune arête. Chaque enfant terminé est enregistré sous <parent>/<child>@<run_id> ; ainsi, une exécution reprise renvoie la réponse enregistrée au lieu d’exécuter à nouveau l’enfant — ce qui est important lorsque l’enfant nécessite un appel à un modèle.

Mise en cache des nœuds

cache_policy sur un nœud indexe son résultat par le nom du nœud et l’état actuel, avec un TTL facultatif ; une entrée inchangée permet donc d’éviter le traitement. Nécessite la fonctionnalité node-cache ; un stockage basé sur Redis est disponible derrière redis-cache.

Points de contrôle différentiels

La fonctionnalité delta stocke la différence entre les super-étapes plutôt que l’état complet, ce qui est important lorsque l’état est volumineux et que les étapes sont nombreuses.

Voyage dans le temps

Avec la fonctionnalité time-travel, graph.time_travel(thread_id)? renvoie une référence vers l’historique des points de contrôle d’un fil d’exécution : lister les étapes, lire l’état à une étape donnée ou fork_at un point de contrôle pour créer un nouveau fil d’exécution à partir de celui-ci. L’appel renvoie Result, car chaque opération lit des points de contrôle ; ainsi, un graphe sans mécanisme de points de contrôle signale GraphError::CheckpointError au lieu de provoquer une panique.

Valeurs par défaut pour tout le graphe

Répéter la même stratégie de nouvelle tentative sur vingt nœuds peut facilement entraîner des erreurs par omission.

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

Une valeur définie par nœud prévaut toujours. NodeDefaults inclut également un délai d’expiration et un gestionnaire d’échec.

Récupération après l’échec d’un nœud

Une fois le budget de nouvelles tentatives d’un nœud épuisé, un gestionnaire peut enregistrer ce qui s’est produit et désigner un nœud de récupération au lieu de terminer l’exécution :

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

Le retour de Err termine l’exécution comme auparavant. Une interruption n’atteint jamais un gestionnaire, car une pause n’est pas un échec.

Limiter la croissance des points de contrôle

Un thread accumule un point de contrôle par super-étape. Une exécution qui dure plusieurs jours croît donc sans limite, ce qui augmente le stockage et ralentit 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)),
    );

L’élagage a lieu après chaque enregistrement, de sorte que le coût reste proportionnel à l’exécution et qu’aucune tâche externe ne soit nécessaire. Le point de contrôle le plus récent n’est jamais supprimé, quelle que soit la stratégie, car c’est celui chargé lors de la reprise — keep_last(0) est défini sur un, et un thread dont tous les points de contrôle dépassent la limite d’âge en conserve un.

Désactivé par défaut, afin qu’un thread existant conserve tout son historique et que le voyage dans le temps puisse toujours atteindre chaque étape. Définissez une stratégie lorsqu’un thread est de longue durée et que vous n’avez pas besoin de revenir très loin en arrière.

Rejeter les canaux non déclarés

Un canal que le schéma ne déclare pas utilise le réducteur de remplacement, car il s’agit du comportement de secours pour un nom inconnu. Un graphe qui a déclaré un canal de liste, puis écrit dans un nom presque identique, ne conserve que la dernière valeur et ne signale rien.

let graph = graph.with_strict_channels();

Un nœud qui écrit dans un canal non déclaré échoue ensuite l’exécution avec GraphError::UndeclaredChannel, en indiquant le nœud et le canal. Désactivé par défaut, et sans effet lorsqu’un graphe ne déclare aucun canal.

Intégration de ADK

GraphAgent implémente le trait ADK Agent, et fonctionne donc avec :

• Exécuteur : À utiliser avec adk-runner pour une exécution standard
• Rappels : Prise en charge complète des rappels avant/après
• Sessions : Fonctionne avec adk-session pour l’historique des conversations
• Diffusion en continu : Renvoie 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

Exemples

Exemples de graphes validés dans ce dépôt :

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

La galerie complète de graphes avec une intégration réelle de LLM se trouve dans adk-playground.

Comparaison avec LangGraph

Deux différences méritent d’être énoncées clairement. run_node_with renvoie le résultat de l’agent enfant directement et l’enregistre, de sorte qu’une exécution reprise ne paie pas à nouveau pour un agent enfant terminé, tandis que Send transmet le travail au planificateur et le récupère au moyen d’un réducteur. Par ailleurs, un saut d’un sous-graphe vers son parent n’a pas encore d’équivalent ici.

---

Précédent : ← Systèmes multi-agents | Suivant : Agents en temps réel →