Agentes de Grafo

Crie fluxos de trabalho complexos e com estado usando orquestração no estilo LangGraph com integração nativa ao ADK-Rust.

Visão geral

GraphAgent permite definir fluxos de trabalho como grafos direcionados com nós e arestas, oferecendo suporte a:

• AgentNode: Encapsule agentes LLM como nós do grafo com mapeadores personalizados de entrada/saída
• Fluxos de Trabalho Cíclicos: Suporte nativo a loops e raciocínio iterativo (padrão ReAct)
• Roteamento Condicional: Roteamento dinâmico de arestas com base no estado
• Gerenciamento de Estado: Estado tipado com redutores (substituição, acréscimo, soma, personalizado)
• Criação de Pontos de Verificação: Estado persistente para tolerância a falhas e interação humana no processo
• Streaming: Vários modos de fluxo (valores, atualizações, mensagens, depuração)

O crate adk-graph fornece orquestração de fluxos de trabalho no estilo LangGraph para criar fluxos de trabalho de agentes complexos e com estado. Ele traz recursos de fluxos de trabalho baseados em grafos para o ecossistema ADK-Rust, mantendo compatibilidade total com o sistema de agentes do ADK.

Principais benefícios:

• Design Visual de Fluxos de Trabalho: Defina lógicas complexas como grafos intuitivos de nós e arestas
• Execução Paralela: Vários nós podem ser executados simultaneamente para obter melhor desempenho
• Persistência de Estado: Criação integrada de pontos de verificação para tolerância a falhas e interação humana no processo
• Integração com LLM: Suporte nativo para encapsular agentes ADK como nós do grafo
• Roteamento Flexível: Arestas estáticas, roteamento condicional e tomada de decisões dinâmica

Escolhendo entre os Agentes de Fluxo de Trabalho e o Grafo

ADK-Rust oferece duas formas de orquestrar. Nenhuma substitui a outra, e ambas são mantidas.

Prefira os agentes de fluxo de trabalho quando a estrutura do trabalho for conhecida e linear: eles são mais curtos de escrever e não há um esquema de estado para manter. Opte pelo grafo quando o fluxo de controle depender dos resultados ou quando uma execução precisar sobreviver ao processo.

Eles são combináveis

Os três agentes de fluxo de trabalho implementam Agent, e AgentNode encapsula qualquer Agent, portanto, um agente de fluxo de trabalho é um nó do grafo:

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 também implementa Agent, portanto, o inverso também é válido: um grafo pode ser um subagente de um SequentialAgent. Use o grafo para a parte que precisa de ramificação ou durabilidade, e os agentes de fluxo de trabalho para as partes que não precisam.

O que você vai criar

Neste guia, você criará um Pipeline de Processamento de Texto que executa tradução e sumarização em paralelo:

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

Conceitos principais:

• Nós - Unidades de processamento que realizam tarefas (agentes LLM, funções ou lógica personalizada)
• Arestas - Fluxo de controle entre nós (conexões estáticas ou roteamento condicional)
• Estado - Dados compartilhados que fluem pelo grafo e persistem entre os nós
• Execução paralela - Vários nós podem ser executados simultaneamente para melhorar o desempenho

Entendendo os componentes principais

🔧 Nós: os trabalhadores É nos nós que o trabalho efetivamente acontece. Cada nó pode:

• AgentNode: encapsular um agente LLM para processar linguagem natural
• Nó de função: executar código Rust personalizado para processamento de dados
• Nós integrados: usar lógica predefinida, como contadores ou validadores

Pense nos nós como trabalhadores especializados em uma linha de montagem — cada um tem uma tarefa e uma especialidade específicas.

🔀 Arestas: o controle do fluxo As arestas determinam como a execução se move pelo seu grafo:

• Arestas estáticas: conexões diretas (A → B → C)
• Arestas condicionais: roteamento dinâmico com base no estado (if sentiment == "positive" → positive_handler)
• Arestas paralelas: vários caminhos a partir de um nó (START → [translator, summarizer])

As arestas são como sinais de trânsito e placas rodoviárias que direcionam o fluxo de trabalho.

💾 Estado: a memória compartilhada O estado é um armazenamento de chave-valor do qual todos os nós podem ler e no qual podem gravar:

• Dados de entrada: informações iniciais fornecidas ao grafo
• Resultados intermediários: a saída de um nó se torna a entrada de outro
• Saída final: o resultado concluído após todo o processamento

O estado funciona como um quadro branco compartilhado, onde os nós podem deixar informações para que outros as utilizem.

⚡ Execução paralela: o aumento de velocidade Quando várias arestas saem de um nó, os nós de destino são executados simultaneamente:

• Processamento mais rápido: tarefas independentes são executadas ao mesmo tempo
• Eficiência de recursos: melhor utilização de CPU e E/S
• Escalabilidade: lidar com fluxos de trabalho mais complexos sem uma desaceleração linear

É como ter vários trabalhadores executando simultaneamente diferentes partes de um trabalho, em vez de esperar na fila.

---

Início rápido

1. Crie seu projeto

cargo new graph_demo
cd graph_demo

Adicione dependências a 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"

Crie .env com sua chave API:

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

2. Exemplo de processamento paralelo

Este é um exemplo completo e funcional que processa texto em paralelo:

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

Saída do exemplo:

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

Como funciona a execução do grafo

Visão geral

Os agentes de grafo executam em superetapas: todos os nós prontos são executados em paralelo, e então o grafo espera que todos sejam concluídos antes da próxima etapa:

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 ✅

Fluxo do estado pelos nós

Cada nó pode ler e gravar no estado compartilhado:

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

O que faz isso funcionar

---

Roteamento condicional com classificação de LLM

Crie sistemas de roteamento inteligentes nos quais LLMs decidam o caminho de execução:

Visualização: roteamento baseado em sentimento

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

Código completo do exemplo

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

Fluxo do exemplo:

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

Padrão ReAct: raciocínio + ação

Crie agentes que possam usar ferramentas iterativamente para resolver problemas complexos:

Visualização: ciclo ReAct

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

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

Fluxo do exemplo:

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

Envolve qualquer ADK Agent (normalmente LlmAgent) como um nó do grafo:

O que o agente vê

Um agente dentro de um grafo é executado em um contexto derivado da invocação que iniciou o grafo, portanto, comporta-se da mesma forma que fora dele. Quando um Runner invoca um GraphAgent, a identidade e os serviços do chamador são transmitidos automaticamente:

Um grafo invocado diretamente — graph.invoke(state, ExecutionConfig::new("thread")) — não tem nenhuma invocação para herdar. Esse é o modo independente: o nó recebe user_id = "graph_user", app_name = "graph_app", a ramificação main, nenhum segredo e nenhuma memória. É um modo deliberado para executar um grafo fora de um Runner, não um recurso alternativo a ser usado em produção.

Para fazer a ponte manualmente — por exemplo, ao controlar um grafo a partir do seu próprio executor — passe a invocação explicitamente:

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

Observação: o nó ainda é executado em sua própria sessão de grafo na memória, portanto o histórico de conversas do agente dentro de um nó fica restrito ao nó, em vez de ser acrescentado à sessão do chamador.

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ós de função

Funções assíncronas simples que processam o estado:

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

Tipos de aresta

Arestas estáticas

Conexões diretas entre nós:

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

Arestas condicionais

Roteamento dinâmico com base no estado:

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

Roteamento a partir de dentro de um nó

Uma aresta condicional define seus destinos quando o grafo é construído. NodeOutput::with_goto não faz isso: um nó grava o estado e nomeia seus sucessores na mesma etapa, e pode nomear qualquer nó do grafo, inclusive um ao qual não tenha nenhuma aresta.

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

A fronteira que um goto produz recebe checkpoint como qualquer outra, portanto uma execução pausada é retomada no nó escolhido pelo goto.

Nota: use add_conditional_edges quando os possíveis destinos forem conhecidos ao criar o grafo — as arestas aparecerão então em um diagrama renderizado. Use um goto quando a escolha pertencer ao nó.

Auxiliares de roteamento

Use roteadores integrados para padrões comuns:

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

Execução paralela

Múltiplas arestas a partir de um único nó são executadas em paralelo:

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 é executado uma vez, depois que todas as três ramificações chegam. Nada no código acima solicita isso: um nó com mais de uma aresta direta de entrada é automaticamente adiado no momento da compilação. Portanto, ramificações de comprimentos diferentes são unidas corretamente sem configuração.

Dois detalhes decorrem da forma como a contagem é realizada:

Grafos Cíclicos (Padrão ReAct)

Crie agentes de raciocínio iterativo com ciclos:

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 Multiagente

Encaminhe tarefas para agentes especialistas:

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

Gerenciamento de Estado

Esquema de Estado com Redutores

Controle como as atualizações de estado são mescladas:

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

Tipos de Redutor

Ordem de atualização

Os nós em uma única superetapa são executados simultaneamente e terminam na ordem que o trabalho deles exigir. As atualizações de estado são aplicadas na ordem do nome do nó, não na ordem em que os nós terminaram.

A ordem importa sempre que um redutor não é comutativo. Append cria um array, portanto a ordem é o resultado; um redutor Custom também pode ser sensível à ordem. A ordenação pelo nome do nó torna uma execução reproduzível: o mesmo grafo e a mesma entrada produzem o mesmo estado, independentemente do momento em que uma dependência lenta termina.

Quando vários canais são escritos por um nó, eles são aplicados na ordem do nome do canal. Os canais não interagem, portanto isso importa apenas para a leitura de um rastreamento.

Criação de pontos de verificação

Habilite o estado persistente para tolerância a falhas e participação humana no circuito:

Em memória (desenvolvimento)

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 (produção)

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

O que um ponto de verificação registra

Um ponto de verificação armazena o estado acumulado, o número da etapa e a fronteira — os nós que ainda precisam ser executados. Ele é gravado depois que a fronteira avança, portanto a retomada nunca executa novamente um nó que já foi concluído e nunca aplica suas atualizações duas vezes. Uma execução que termina cria um ponto de verificação com uma fronteira vazia, portanto retomar uma thread concluída retorna o estado final em vez de reiniciar o grafo.

Dois casos criam deliberadamente um ponto de verificação da fronteira que estava em execução, em vez da próxima, porque o nó interrompido ainda não produziu suas atualizações e precisa ser executado novamente ao retomar:

As execuções em streaming criam pontos de verificação na mesma programação que as execuções bloqueantes, inclusive quando uma interrupção encerra o stream, portanto uma pausa com participação humana pode ser retomada em qualquer um dos modos de execução.

Histórico de Pontos de Verificação (Viagem no Tempo)

Somente leitura. TimeTravelHandle::state_history(from, to) retorna o estado que foi armazenado em cada etapa com ponto de verificação. Ele não executa nada — nenhum nó é executado, nenhum evento é regenerado e nenhum efeito colateral se repete. Para executar novamente a partir de um ponto do histórico, use fork_at para criar um branch desse ponto de verificação e invoque o grafo na thread derivada. O método anteriormente se chamava replay e era documentado como reexecutando o grafo, o que ele nunca fez.

Os pontos de verificação também permitem a retomada durável — se a execução de um grafo falhar ou o processo for reiniciado, a execução será retomada a partir do último ponto de verificação persistido, em vez de começar novamente. Use SqliteCheckpointer (o recurso sqlite) para obter persistência segura contra falhas. MemoryCheckpointer mantém os pontos de verificação no processo, portanto eles não sobrevivem a uma reinicialização. Esses são os dois backends fornecidos por este crate; implemente a trait Checkpointer para qualquer outra opção.

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

Participação Humana

Pause a execução para obter aprovação humana usando interrupções dinâmicas:

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

Interrupções Estáticas

Use interrupt_before ou interrupt_after para pontos de pausa obrigatórios:

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

Execução em Streaming

Transmita eventos à medida que o grafo é executado:

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

Modos de Streaming

O modo Messages lê tokens de Node::execute_stream à medida que são produzidos. Cada nó é executado uma vez por superetapa nesse modo: o nó relata suas atualizações de estado no fluxo como um evento StreamEvent::Updates, e o executor aplica essas atualizações em vez de executar o nó uma segunda vez para coletá-las. Isso é mais importante para AgentNode, em que uma segunda execução significaria uma segunda chamada faturada ao modelo por nó.

Importante: um Node personalizado que substitui execute_stream deve produzir um evento StreamEvent::Updates contendo suas atualizações de estado. Sem isso, o nó transmite eventos, mas não contribui com estado no modo Messages. O execute_stream padrão, que encapsula execute, faz isso por você.

As políticas de tempo limite se aplicam à própria execução transmitida. Para um fluxo, idle_timeout significa que nenhum evento foi produzido dentro do limite.

Subgrafos

Um grafo compilado é executado como um nó de outro por meio de SubgraphNode. O grafo interno mantém seus próprios canais, arestas e portas de interrupção, e troca canais nomeados com seu pai.

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

Essa última linha é a diferença que vale a pena conhecer: ambos os esquemas estão disponíveis antes de qualquer execução, portanto um mapeamento que nomeie um canal que nenhum dos lados declara não pode chegar a uma execução e aparecer como um valor ausente. Um subgrafo que não troque absolutamente nada é rejeitado da mesma forma, pois não poderia afetar seu pai.

Retomando uma pausa dentro de um subgrafo

Nada mais é necessário. Invocar o pai novamente na mesma thread entra novamente no subgrafo, que encontra seu próprio checkpoint em <parent thread>/<node name> e continua de onde parou. O trabalho concluído pelo subgrafo antes da pausa não é repetido, e uma pausa ocorrida vários níveis abaixo é retomada da mesma forma — a mensagem nomeia cada nível pelo qual passou.

Um subgrafo que declara um ponto de interrupção, mas não possui nenhum armazenador de checkpoints, é rejeitado quando o pai é compilado: ele entraria novamente em seu primeiro nó e pagaria uma segunda vez pelo trabalho já concluído.

Como ambos os grafos mantêm checkpointers reais, isso sobrevive a uma reinicialização do processo: um novo conjunto de objetos de grafo que compartilha apenas os bancos de dados retoma a mesma execução.

Devolvendo o controle ao pai

Um nó dentro de um subgrafo pode encerrar o próprio grafo e nomear um nó do grafo que o contém:

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

O subgrafo termina e projeta seus canais de saída normalmente; em seguida, o pai continua em escalate em vez de seguir as arestas declaradas pelo nó do subgrafo. O pai valida o destino, pois somente ele conhece seus próprios nós.

Controles de confiabilidade e custo

Cada um destes recursos vem desativado por padrão, portanto um grafo se comporta como antes de você configurar qualquer um deles.

Nova tentativa por nó

Uma falha transitória — um limite de taxa ou uma conexão interrompida — caso contrário encerra a execução.

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

O atraso aumenta em backoff_factor, é limitado a max_delay e, em seguida, recebe jitter. Um nó sem política é executado uma vez, portanto as novas tentativas continuam sendo opcionais. Uma política de RetryPolicy::default() permite dez tentativas, cujas nove esperas totalizam cerca de 243 segundos — reduza max_attempts quando um chamador estiver aguardando a resposta.

Uma interrupção nunca é repetida, independentemente do que retry_on diga: uma pausa não é uma falha. A contagem de tentativas é armazenada no checkpoint, portanto uma execução retomada continua usando o orçamento existente em vez de reiniciá-lo.

Limitando a concorrência

Uma distribuição ampla despacha toda a sua fronteira de uma só vez, o que pode esgotar um pool de conexões ou acionar um limite de taxa do provedor.

let graph = graph.with_max_concurrency(4);

Os nós além do limite aguardam uma vaga. A ordem de admissão é a fronteira classificada por nome, portanto não depende do tempo. As invocações imperativas de filhos ficam fora desse orçamento, pois um pai aguarda seus filhos enquanto mantém sua própria vaga.

Tempos limite por nó

Um TimeoutPolicy limita uma única tentativa e, com idle_timeout, define por quanto tempo um nó pode ficar sem relatar progresso. Exceder qualquer um deles gera GraphError::NodeTimedOut, sobre o qual uma política de novas tentativas pode então agir.

Invocando um nó diretamente

Quando o número de subtarefas vem do estado, e não da estrutura do grafo, um nó pode invocar outro nó por conta própria:

use adk_graph::child::RunNodeOptions;

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

O destino não precisa de uma aresta. Cada filho concluído é registrado em <parent>/<child>@<run_id>, portanto uma execução retomada retorna a resposta registrada em vez de executar o filho novamente — o que é importante quando o filho exige uma chamada de modelo.

Armazenamento em cache de nós

cache_policy em um nó identifica o resultado por nome do nó e estado atual, com um TTL opcional, portanto uma entrada inalterada ignora o trabalho. Requer o recurso node-cache; um armazenamento baseado em Redis está disponível por meio de redis-cache.

Pontos de verificação delta

O recurso delta armazena a diferença entre superetapas em vez do estado inteiro, o que é importante quando o estado é grande e há muitas etapas.

Viagem no tempo

Com o recurso time-travel, graph.time_travel(thread_id)? retorna um identificador sobre o histórico de pontos de verificação de uma thread: listar as etapas, ler o estado em uma delas ou fork_at um ponto de verificação para criar uma nova thread a partir dele. A chamada retorna Result porque toda operação lê pontos de verificação; portanto, um grafo sem um registrador de pontos de verificação informa GraphError::CheckpointError em vez de entrar em pânico.

Padrões para todo o grafo

Repetir a mesma nova tentativa em vinte nós é fácil de fazer incorretamente por omissão.

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

Um valor por nó sempre prevalece. NodeDefaults também inclui um tempo limite e um manipulador de falhas.

Recuperando-se de uma falha de nó

Quando o orçamento de novas tentativas de um nó se esgota, um manipulador pode registrar o que aconteceu e nomear um nó de recuperação em vez de encerrar a execução:

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

Retornar Err encerra a execução como antes. Uma interrupção nunca chega a um manipulador, porque uma pausa não é uma falha.

Limitando o crescimento dos checkpoints

Uma thread acumula um checkpoint por superetapa. Portanto, uma execução que permanece ativa por dias cresce indefinidamente, o que aumenta o custo de armazenamento e torna list mais lento.

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

A remoção ocorre após cada salvamento, portanto o custo permanece proporcional à execução e nenhum job externo é necessário. O checkpoint mais recente nunca é descartado, independentemente da política, porque é aquele que um reinício carrega — keep_last(0) é elevado para um, e uma thread cujo todos os checkpoints ultrapassaram o limite de idade mantém um.

Desativado por padrão, para que uma thread existente mantenha todo o seu histórico e a viagem no tempo ainda possa alcançar cada etapa. Defina uma política quando uma thread tiver longa duração e você não precisar retroceder muito.

Rejeitando canais não declarados

Um canal que o esquema não declara usa o redutor de substituição, porque esse é o fallback para um nome desconhecido. Um grafo que declarou um canal de lista e depois escreveu um nome quase correspondente mantém apenas o último valor e não relata nada.

let graph = graph.with_strict_channels();

Um nó que escreve em um canal não declarado falha a execução com GraphError::UndeclaredChannel, identificando o nó e o canal. Desativado por padrão e inerte quando um grafo não declara nenhum canal.

Integração com ADK

GraphAgent implementa o trait Agent de ADK, portanto funciona com:

• Executor: Use com adk-runner para execução padrão
• Callbacks: Suporte completo para callbacks antes/depois
• Sessões: Funciona com adk-session para o histórico da conversa
• Streaming: Retorna 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

Exemplos

Exemplos de grafos validados neste repositório:

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

A galeria completa de grafos com integração real de LLM está disponível em adk-playground.

Comparação com LangGraph

Vale destacar claramente duas diferenças. run_node_with retorna o resultado do filho em linha e o registra, portanto uma execução retomada não paga novamente por um filho concluído, enquanto Send encaminha o trabalho ao agendador e o coleta por meio de um redutor. Além disso, ainda não há aqui um equivalente para um salto de um subgrafo para seu pai.

---

Anterior: ← Sistemas multiagente | Próximo: Agentes em tempo real →