Agentes de grafos

Construya flujos de trabajo complejos y con estado mediante la orquestación al estilo de LangGraph con integración nativa de ADK-Rust.

Descripción general

GraphAgent le permite definir flujos de trabajo como grafos dirigidos con nodos y aristas, compatibles con:

• AgentNode: Encapsule agentes LLM como nodos del grafo con mapeadores de entrada/salida personalizados
• Flujos de trabajo cíclicos: Compatibilidad nativa con bucles y razonamiento iterativo (patrón ReAct)
• Enrutamiento condicional: Enrutamiento dinámico de aristas basado en el estado
• Gestión del estado: Estado tipado con reductores (sobrescritura, adición, suma, personalizado)
• Puntos de control: Estado persistente para tolerancia a fallos y participación humana en el proceso
• Transmisión: Múltiples modos de transmisión (valores, actualizaciones, mensajes, depuración)

El crate adk-graph proporciona orquestación de flujos de trabajo al estilo de LangGraph para crear flujos de trabajo de agentes complejos y con estado. Incorpora capacidades de flujos de trabajo basados en grafos al ecosistema ADK-Rust, manteniendo la compatibilidad total con el sistema de agentes de ADK.

Ventajas principales:

• Diseño visual de flujos de trabajo: Defina lógica compleja como grafos intuitivos de nodos y aristas
• Ejecución en paralelo: Varios nodos pueden ejecutarse simultáneamente para mejorar el rendimiento
• Persistencia del estado: Puntos de control integrados para tolerancia a fallos y participación humana en el proceso
• Integración con LLM: Compatibilidad nativa para encapsular agentes ADK como nodos del grafo
• Enrutamiento flexible: Aristas estáticas, enrutamiento condicional y toma de decisiones dinámica

Elegir entre los agentes de flujo de trabajo y el grafo

ADK-Rust admite dos formas de realizar la orquestación. Ninguna sustituye a la otra y ambas se mantienen.

Prefiere los agentes de flujo de trabajo cuando la estructura del trabajo es conocida y lineal: son más breves de escribir y no hay ningún esquema de estado que mantener. Recurre al grafo cuando el flujo de control depende de los resultados o cuando una ejecución debe sobrevivir al proceso.

Se componen

Los tres agentes de flujo de trabajo implementan Agent, y AgentNode envuelve cualquier Agent, por lo que un agente de flujo de trabajo es un nodo del 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 también implementa Agent, por lo que se cumple lo contrario: un grafo puede ser un subagente de un SequentialAgent. Usa el grafo para la parte que necesita ramificación o durabilidad, y los agentes de flujo de trabajo para las partes que no las necesitan.

Lo que crearás

En esta guía, crearás una Canalización de procesamiento de texto que ejecuta la traducción y el resumen en paralelo:

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

Conceptos clave:

• Nodos: unidades de procesamiento que realizan tareas (agentes LLM, funciones o lógica personalizada)
• Aristas: flujo de control entre nodos (conexiones estáticas o enrutamiento condicional)
• Estado: datos compartidos que fluyen por el grafo y persisten entre nodos
• Ejecución en paralelo: varios nodos pueden ejecutarse simultáneamente para mejorar el rendimiento

Comprender los componentes principales

🔧 Nodos: los trabajadores Los nodos son donde se realiza el trabajo real. Cada nodo puede:

• AgentNode: envolver un agente LLM para procesar lenguaje natural
• Nodo de función: ejecutar código Rust personalizado para el procesamiento de datos
• Nodos integrados: usar lógica predefinida, como contadores o validadores

Piensa en los nodos como trabajadores especializados en una cadena de montaje: cada uno tiene una tarea y una especialización específicas.

🔀 Aristas: El control del flujo Las aristas determinan cómo se mueve la ejecución por tu grafo:

• Aristas estáticas: Conexiones directas (A → B → C)
• Aristas condicionales: Enrutamiento dinámico basado en el estado (if sentiment == "positive" → positive_handler)
• Aristas paralelas: Múltiples rutas desde un nodo (START → [translator, summarizer])

Las aristas son como semáforos y señales de tráfico que dirigen el flujo de trabajo.

💾 Estado: La memoria compartida El estado es un almacén de clave-valor del que todos los nodos pueden leer y en el que pueden escribir:

• Datos de entrada: Información inicial introducida en el grafo
• Resultados intermedios: La salida de un nodo se convierte en la entrada de otro
• Salida final: El resultado completado después de todo el procesamiento

El estado funciona como una pizarra compartida donde los nodos pueden dejar información para que otros la utilicen.

⚡ Ejecución paralela: El aumento de velocidad Cuando varias aristas salen de un nodo, los nodos de destino se ejecutan simultáneamente:

• Procesamiento más rápido: Las tareas independientes se ejecutan al mismo tiempo
• Eficiencia de los recursos: Mejor aprovechamiento de la CPU y de la E/S
• Escalabilidad: Permite gestionar flujos de trabajo más complejos sin una ralentización lineal

Es como tener varios trabajadores ocupándose simultáneamente de distintas partes de un trabajo, en lugar de esperar en una fila.

---

Inicio rápido

1. Crea tu proyecto

cargo new graph_demo
cd graph_demo

Añade dependencias 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"

Crea .env con tu clave API:

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

2. Ejemplo de procesamiento paralelo

Este es un ejemplo completo funcional que procesa texto en 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(())
}

Salida de ejemplo:

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

Cómo funciona la ejecución del grafo

La visión general

Los agentes del grafo se ejecutan en superpasos: todos los nodos listos se ejecutan en paralelo y, después, el grafo espera a que todos terminen antes del siguiente paso:

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 ✅

Flujo del estado a través de los nodos

Cada nodo puede leer del estado compartido y escribir en él:

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

Qué hace que funcione

---

Enrutamiento condicional con la clasificación de LLM

Construye sistemas de enrutamiento inteligentes donde LLMs decidan la ruta de ejecución:

Visual: Enrutamiento basado en el sentimiento

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

Código completo del ejemplo

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

Flujo del ejemplo:

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

Patrón de ReAct: Razonamiento + acción

Construye agentes que puedan usar herramientas iterativamente para resolver problemas complejos:

Visual: Ciclo de ReAct

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

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

Flujo del ejemplo:

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

Envuelve cualquier ADK Agent (normalmente LlmAgent) como un nodo del grafo:

Lo que ve el agente

Un agente dentro de un grafo se ejecuta bajo un contexto derivado de la invocación que inició el grafo, por lo que se comporta igual que fuera de él. Cuando un Runner invoca a un GraphAgent, la identidad y los servicios del invocador se transmiten automáticamente:

Un grafo invocado directamente — graph.invoke(state, ExecutionConfig::new("thread")) — no tiene ninguna invocación que heredar. Ese es el modo independiente: el nodo obtiene user_id = "graph_user", app_name = "graph_app", la rama main, ningún secreto ni memoria. Es un modo deliberado para ejecutar un grafo fuera de un Runner, no un recurso alternativo al que recurrir en producción.

Para establecer el puente manualmente —por ejemplo, al controlar un grafo desde tu propio ejecutor— pasa la invocación explícitamente:

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

Nota: el nodo sigue ejecutándose en su propia sesión de grafo en memoria, por lo que el historial de conversación del agente dentro de un nodo está limitado al nodo, en lugar de añadirse a la sesión del llamador.

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

Nodos de función

Funciones asíncronas simples que procesan el 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 aristas

Aristas estáticas

Conexiones directas entre nodos:

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

Aristas condicionales

Enrutamiento dinámico basado en el 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),
    ],
)

Enrutamiento desde dentro de un nodo

Una arista condicional fija sus destinos cuando se construye el grafo. NodeOutput::with_goto no lo hace: un nodo escribe el estado y nombra a sus sucesores en el mismo paso, y puede nombrar cualquier nodo del grafo, incluido uno con el que no tiene ninguna arista.

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 frontera que produce un goto se guarda como checkpoint igual que cualquier otra, por lo que una ejecución pausada se reanuda en el nodo que eligió el goto.

Nota: usa add_conditional_edges cuando los posibles destinos se conozcan al construir el grafo; las aristas aparecen entonces en un diagrama renderizado. Usa un goto cuando la elección corresponda al nodo.

Ayudantes de enrutamiento

Usa routers integrados para patrones comunes:

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

Ejecución en paralelo

Varias aristas desde un mismo nodo se ejecutan en 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 se ejecuta una vez, después de que llegan las tres ramas. Nada en el código anterior lo solicita: un nodo con más de una arista directa entrante se pospone automáticamente durante la compilación. Por lo tanto, las ramas de distinta longitud se unen correctamente sin configuración.

De cómo se obtiene el recuento se derivan dos detalles:

Grafos cíclicos (ReAct Pattern)

Construye agentes de razonamiento iterativo con 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

Enruta las tareas a 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()?;

Gestión del estado

Esquema de estado con reductores

Controla cómo se combinan las actualizaciones del estado:

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 reductores

Orden de actualización

Los nodos de un superpaso se ejecutan simultáneamente y terminan en el orden que requiere su trabajo. Sus actualizaciones de estado se aplican en orden de nombre de nodo, no en el orden en que terminan los nodos.

El orden importa siempre que un reductor no sea conmutativo. Append construye un array, por lo que el orden es el resultado; un reductor Custom también puede depender del orden. Ordenar por nombre de nodo hace que una ejecución sea reproducible: el mismo grafo y la misma entrada producen el mismo estado, independientemente del momento en que termine una dependencia lenta.

Cuando un nodo escribe en varios canales, estos se aplican en orden de nombre de canal. Los canales no interactúan, por lo que esto solo importa al leer una traza.

Puntos de control

Habilita el estado persistente para la tolerancia a fallos y la intervención humana:

En memoria (desarrollo)

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 (producción)

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

Qué registra un punto de control

Un punto de control almacena el estado acumulado, el número de paso y la frontera —los nodos que aún deben ejecutarse—. Se escribe después de que la frontera avanza, por lo que al reanudar nunca se vuelve a ejecutar un nodo que ya terminó ni se aplican dos veces sus actualizaciones. Una ejecución que termina registra un punto de control con una frontera vacía, por lo que reanudar un hilo completado devuelve el estado final en lugar de reiniciar el grafo.

En dos casos, el punto de control registra deliberadamente la frontera que estaba ejecutándose en lugar de la siguiente, porque el nodo interrumpido aún no ha producido sus actualizaciones y debe ejecutarse de nuevo al reanudar:

Las ejecuciones en streaming crean puntos de control con el mismo calendario que las ejecuciones bloqueantes, incluso cuando una interrupción finaliza el flujo, por lo que una pausa con intervención humana se puede reanudar en cualquiera de los dos modos de ejecución.

Historial de puntos de control (viaje en el tiempo)

Solo lectura. TimeTravelHandle::state_history(from, to) devuelve el estado que se almacenó en cada paso con punto de control. No ejecuta nada: no se ejecuta ningún nodo, no se regenera ningún evento ni se repite ningún efecto secundario. Para volver a ejecutar desde un punto del historial, usa fork_at para crear una bifurcación a partir de ese punto de control e invoca el grafo en el hilo bifurcado. Anteriormente, el método se llamaba replay y se documentaba como si volviera a ejecutar el grafo, pero nunca lo hizo.

Los puntos de control también permiten una reanudación duradera: si una ejecución del grafo falla o el proceso se reinicia, la ejecución se reanuda desde el último punto de control persistido en lugar de comenzar de nuevo. Usa SqliteCheckpointer (la funcionalidad sqlite) para una persistencia segura frente a fallos. MemoryCheckpointer mantiene los puntos de control en el proceso, por lo que no sobreviven a un reinicio. Esos son los dos backends que incluye este crate; implementa el trait Checkpointer para cualquier otra opción.

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

Intervención humana

Pausa la ejecución para obtener aprobación humana mediante interrupciones 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);
    }
}

Interrupciones estáticas

Usa interrupt_before o interrupt_after para establecer puntos de pausa obligatorios:

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

Ejecución en streaming

Transmite eventos a medida que se ejecuta el grafo:

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

Messages mode lee tokens de Node::execute_stream a medida que se producen. Cada nodo se ejecuta una sola vez por superpaso en este modo: el nodo informa de sus actualizaciones de estado en el flujo como un evento StreamEvent::Updates, y el ejecutor aplica esas actualizaciones en lugar de ejecutar el nodo una segunda vez para recopilarlas. Esto es especialmente importante para AgentNode, donde una segunda ejecución implicaría una segunda llamada facturable al modelo por nodo.

Importante: un Node personalizado que sobrescriba execute_stream debe producir un evento StreamEvent::Updates que contenga sus actualizaciones de estado. Sin él, el nodo transmite eventos, pero no contribuye ningún estado en el modo Messages. El execute_stream predeterminado, que envuelve execute, lo hace por usted.

Las políticas de tiempo de espera se aplican a la ejecución transmitida en sí. Para un flujo, idle_timeout significa que no se produjo ningún evento dentro del límite.

Subgrafos

Un grafo compilado se ejecuta como nodo de otro mediante SubgraphNode. El grafo interno mantiene sus propios canales, aristas y puertas de interrupción, e intercambia canales con nombre con su grafo principal.

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

Esa última fila es la diferencia que conviene conocer: ambos esquemas están disponibles antes de que se ejecute nada, por lo que una asignación que nombre un canal que ninguno de los dos declare no puede llegar a una ejecución ni manifestarse como un valor ausente. Un subgrafo que no intercambie nada en absoluto se rechaza de la misma manera, porque no podría afectar a su elemento principal.

Reanudar una pausa dentro de un subgrafo

No se necesita nada adicional. Invocar de nuevo al elemento principal en el mismo hilo vuelve a entrar en el subgrafo, que encuentra su propio punto de control en <parent thread>/<node name> y continúa desde donde se detuvo. El trabajo que el subgrafo terminó antes de la pausa no se repite, y una pausa varios niveles más abajo se reanuda de la misma manera: el mensaje nombra cada nivel por el que pasó.

Un subgrafo que declara una puerta de interrupción, pero no tiene ningún punto de control, se rechaza cuando el elemento principal se compila: volvería a entrar en su primer nodo y pagaría por su trabajo terminado una segunda vez.

Dado que ambos grafos contienen checkpointers reales, esto sobrevive al reinicio de un proceso: un conjunto nuevo de objetos de grafo que comparten únicamente las bases de datos reanuda la misma ejecución.

Devolver el control al elemento principal

Un nodo dentro de un subgrafo puede finalizar su propio grafo y nombrar un nodo del grafo que lo contiene:

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

El subgrafo finaliza y proyecta sus canales de salida como de costumbre; después, el elemento principal continúa en escalate en lugar de seguir los bordes declarados del nodo del subgrafo. El elemento principal valida el destino, porque es el único lado que conoce sus propios nodos.

Controles de fiabilidad y costes

Cada una de estas opciones está desactivada de forma predeterminada, por lo que un grafo se comporta como antes de configurar cualquiera de ellas.

Reintentos por nodo

Un fallo transitorio —un límite de velocidad o una conexión interrumpida— finalizaría de otro modo la ejecución.

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

El retraso aumenta según backoff_factor, está limitado por max_delay y, después, se le aplica fluctuación aleatoria. Un nodo sin política se ejecuta una vez, por lo que los reintentos siguen siendo opcionales. Una política de RetryPolicy::default() permite diez intentos, cuyos nueve tiempos de espera suman aproximadamente 243 segundos; reduzca max_attempts cuando una persona que llama esté esperando la respuesta.

Una interrupción nunca se reintenta, independientemente de lo que indique retry_on: una pausa no es un fallo. El recuento de intentos se guarda en el checkpoint, por lo que una ejecución reanudada continúa utilizando el presupuesto restante en lugar de reiniciarlo.

Limitación de la concurrencia

Una distribución amplia envía todo su frente de trabajo a la vez, lo que puede agotar un grupo de conexiones o activar el límite de velocidad de un proveedor.

let graph = graph.with_max_concurrency(4);

Los nodos que superan el límite esperan a que haya un espacio disponible. El orden de admisión es el frente de trabajo ordenado por nombre, por lo que no depende de la temporización. Las invocaciones imperativas de elementos secundarios quedan fuera de este presupuesto, porque el elemento principal espera a sus elementos secundarios mientras mantiene su propio espacio.

Tiempos de espera por nodo

Un TimeoutPolicy limita un intento individual y, junto con idle_timeout, determina cuánto tiempo puede pasar un nodo sin informar de progreso. Superar cualquiera de los dos produce GraphError::NodeTimedOut, sobre el que una política de reintentos puede actuar.

Invocar un nodo directamente

Cuando el número de subtareas proviene del estado en lugar de la estructura del grafo, un nodo puede invocar otro nodo por sí mismo:

use adk_graph::child::RunNodeOptions;

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

El destino no necesita ninguna arista. Cada elemento secundario completado se registra en <parent>/<child>@<run_id>, de modo que una ejecución reanudada devuelve la respuesta registrada en lugar de ejecutar el elemento secundario de nuevo, lo que es importante cuando este requiere una llamada a un modelo.

Almacenamiento en caché de nodos

cache_policy en un nodo identifica su resultado mediante el nombre del nodo y el estado actual, con un TTL opcional, de modo que una entrada sin cambios omite el trabajo. Requiere la funcionalidad node-cache; hay disponible un almacén respaldado por Redis mediante redis-cache.

Puntos de control diferenciales

La funcionalidad delta almacena la diferencia entre superpasos en lugar del estado completo, lo que es importante cuando el estado es grande y hay muchos pasos.

Viaje en el tiempo

Con la funcionalidad time-travel, graph.time_travel(thread_id)? devuelve un controlador sobre el historial de puntos de control de un hilo: permite enumerar los pasos, leer el estado en uno de ellos o fork_at un punto de control para bifurcar un hilo nuevo a partir de él. La llamada devuelve Result porque toda operación lee puntos de control, por lo que un grafo sin gestor de puntos de control informa de GraphError::CheckpointError en lugar de provocar un pánico.

Valores predeterminados para todo el grafo

Repetir el mismo reintento en veinte nodos puede fallar fácilmente por omisión.

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

Un valor por nodo siempre tiene prioridad. NodeDefaults también incluye un tiempo de espera y un controlador de fallos.

Recuperarse de un fallo de nodo

Una vez agotado el presupuesto de reintentos de un nodo, un controlador puede registrar lo ocurrido y designar un nodo de recuperación en lugar de finalizar la ejecución:

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

Devolver Err finaliza la ejecución como antes. Una interrupción nunca llega a un controlador, porque una pausa no es un fallo.

Limitar el crecimiento de los puntos de control

Un hilo acumula un punto de control por superpaso. Por lo tanto, una ejecución que dura días crece sin límite, lo que aumenta el almacenamiento y ralentiza 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)),
    );

La poda se produce después de cada guardado, por lo que el coste se mantiene proporcional a la ejecución y no se necesita ningún trabajo externo. El punto de control más reciente nunca se descarta, independientemente de lo que indique la política, porque es el que carga una reanudación — keep_last(0) se establece en uno, y un hilo cuyos puntos de control hayan superado el límite de antigüedad conserva uno.

Está desactivado de forma predeterminada, por lo que un hilo existente conserva todo su historial y el viaje en el tiempo aún puede llegar a cada paso. Establece una política cuando un hilo tenga una larga duración y no necesites retroceder mucho.

Rechazar canales no declarados

Un canal que el esquema no declara utiliza el reductor de sobrescritura, porque ese es el valor alternativo para un nombre desconocido. Un grafo que declaró un canal de lista y luego escribió un nombre parecido conserva solo el último valor y no informa de nada.

let graph = graph.with_strict_channels();

Un nodo que escribe en un canal no declarado hace que la ejecución falle con GraphError::UndeclaredChannel, indicando el nodo y el canal. Está desactivado de forma predeterminada y no tiene efecto cuando un grafo no declara ningún canal.

Integración con ADK

GraphAgent implementa el trait ADK Agent, por lo que funciona con:

• Ejecutor: úsalo con adk-runner para la ejecución estándar
• Callbacks: compatibilidad completa con callbacks anteriores y posteriores
• Sesiones: funciona con adk-session para el historial de conversaciones
• Streaming: devuelve 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

Ejemplos

Ejemplos de grafos validados en este repositorio:

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 galería completa de grafos con integración real de LLM se encuentra en adk-playground.

Comparación con LangGraph

Vale la pena expresar claramente dos diferencias. run_node_with devuelve el resultado del elemento secundario en línea y lo registra, por lo que una ejecución reanudada no vuelve a pagar por un elemento secundario completado, mientras que Send entrega el trabajo al programador y lo recopila mediante un reductor. Y un salto desde un subgrafo a su elemento principal aún no tiene aquí un equivalente.

---

Anterior: ← Sistemas multiagente | Siguiente: Agentes en tiempo real →