adk-graph

Graph-based workflow orchestration for Rust Agent Development Kit (ADK-Rust) agents, inspired by LangGraph.

Overview

adk-graph provides a powerful way to build complex, stateful agent workflows using a graph-based approach. It brings LangGraph-style capabilities to the Rust ADK ecosystem while maintaining full compatibility with ADK's agent system, callbacks, and streaming infrastructure.

Features

• Graph-Based Workflows: Define agent workflows as directed graphs with nodes and edges
• AgentNode: Wrap LLM agents as graph nodes with custom input/output mappers
• Cyclic Support: Native support for loops and iterative reasoning (ReAct pattern)
• Conditional Routing: Dynamic edge routing based on state
• Fan-out / fan-in: parallel branches run concurrently in a super-step. A node with more than one incoming direct edge is deferred automatically, so it runs once after its branches arrive — branches of unequal length join correctly with no configuration. mark_deferred sets a fan_in_timeout or an n-of-m min_predecessors quorum
• State Management: Typed state with reducers (overwrite, append, sum, custom)
• Checkpointing: Persistent state after each step (memory, SQLite)
• Durable Resume: Automatically resume from the last checkpoint after a crash — skips already-completed nodes
• Human-in-the-Loop: Interrupt before/after nodes, dynamic interrupts. A pause is resumable, including one raised inside a subgraph
• Subgraphs: run a compiled graph as a node with mapped channels (SubgraphNode). A pause inside pauses the parent, and a channel mapping that names a channel neither side declares fails when the parent compiles
• Routing from inside a node: NodeOutput::with_goto names successors and replaces the node's declared edges; AgentNode::with_goto_mapper routes on what the agent answered; with_goto_parent hands control to a node of the parent graph
• Reliability: per-node retry with capped backoff and jitter, a concurrency bound, per-node timeouts, graph-wide NodeDefaults, and with_node_error_handler to recover once a retry budget is spent
• Invoking a node directly: ctx.run_node_with runs a node the graph has no edge to, sized from state, and records it so a resume does not pay for it twice
• As a tool: NodeTool exposes a graph or a single node through the Tool trait, so an LlmAgent can call a whole graph
• Bounded growth: RetentionPolicy limits how many checkpoints a thread keeps, by count or age
• Strictness, opt-in: with_strict_channels fails the run when a node writes a channel the schema does not declare
• Streaming: Multiple stream modes (values, updates, messages, debug)
• ADK Integration: Full callback support, works with existing runners
• Functional API (feature: functional): Write workflows as async functions with automatic checkpointing
• Typed State Reducers: ReducedValue, UntrackedValue, MessagesValue containers
• State Schema Validation: Type-level validation at workflow boundaries
• Proc Macros: #[entrypoint] and #[task] for zero-boilerplate workflow definition

Architecture

              ┌─────────────────────────────────────────┐
              │              Agent Trait                │
              │  (name, description, run, sub_agents)   │
              └────────────────┬────────────────────────┘
                               │
       ┌───────────────────────┼───────────────────────┐
       │                       │                       │
┌──────▼──────┐      ┌─────────▼─────────┐   ┌─────────▼─────────┐
│  LlmAgent   │      │   GraphAgent      │   │  RealtimeAgent    │
│ (text-based)│      │ (graph workflow)  │   │  (voice-based)    │
└─────────────┘      └───────────────────┘   └───────────────────┘

Quick Start

Add to your 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"

Basic Graph with AgentNode

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

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

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

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

    // Create AgentNodes with input/output mappers
    let translator_node = AgentNode::new(translator)
        .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)
        .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 graph with parallel execution
    let agent = GraphAgent::builder("text_processor")
        .description("Translates and summarizes text in parallel")
        .channels(&["input", "translation", "summary"])
        .node(translator_node)
        .node(summarizer_node)
        .edge(START, "translator")
        .edge(START, "summarizer")  // Both start in parallel
        .edge("translator", END)
        .edge("summarizer", END)
        .build()?;

    // Execute
    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!("Translation: {}", result.get("translation").and_then(|v| v.as_str()).unwrap_or(""));
    println!("Summary: {}", result.get("summary").and_then(|v| v.as_str()).unwrap_or(""));

    Ok(())
}

Conditional Routing with LLM Classification

use adk_graph::{edge::Router, node::NodeOutput};

// Create a classifier agent
let classifier = Arc::new(
    LlmAgentBuilder::new("classifier")
        .model(model.clone())
        .instruction("Classify the sentiment as 'positive', 'negative', or 'neutral'. Reply with one word only.")
        .build()?
);

let classifier_node = AgentNode::new(classifier)
    .with_input_mapper(|state| {
        let msg = state.get("message").and_then(|v| v.as_str()).unwrap_or("");
        adk_core::Content::new("user").with_text(&format!("Classify: {}", msg))
    })
    .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 sentiment = if text.contains("positive") { "positive" }
                    else if text.contains("negative") { "negative" }
                    else { "neutral" };

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

// Build conditional routing graph
let graph = StateGraph::with_channels(&["message", "sentiment", "response"])
    .add_node(classifier_node)
    .add_node(positive_handler_node)
    .add_node(negative_handler_node)
    .add_node(neutral_handler_node)
    .add_edge(START, "classifier")
    .add_conditional_edges(
        "classifier",
        Router::by_field("sentiment"),  // Route based on "sentiment" field
        [
            ("positive", "positive_handler"),
            ("negative", "negative_handler"),
            ("neutral", "neutral_handler"),
        ],
    )
    .add_edge("positive_handler", END)
    .add_edge("negative_handler", END)
    .add_edge("neutral_handler", END)
    .compile()?;

Human-in-the-Loop with Risk Assessment

use adk_graph::{checkpoint::MemoryCheckpointer, error::GraphError};

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

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

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

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

        if risk == "high" || risk == "medium" {
            // Interrupt for human approval
            return Ok(NodeOutput::interrupt_with_data(
                "Human approval required",
                json!({"risk_level": risk, "action": "Set 'approved' to true to continue"})
            ));
        }

        // Auto-approve low risk
        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 result = graph.invoke(input, ExecutionConfig::new("task-001")).await;

match result {
    Err(GraphError::Interrupted(interrupt)) => {
        println!("Paused: {}", interrupt.interrupt);

        // Human reviews and approves...
        graph.update_state("task-001", [("approved".to_string(), json!(true))]).await?;

        // Resume
        let final_result = graph.invoke(State::new(), ExecutionConfig::new("task-001")).await?;
    }
    Ok(result) => println!("Completed: {:?}", result),
    Err(e) => println!("Error: {}", e),
}

ReAct Agent with Tools

use adk_core::Part;
use adk_tool::FunctionTool;

// 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(Arc::new(FunctionTool::new("search", "Search for info", |_ctx, args| async move {
            Ok(json!({"result": "Search results..."}))
        })))
        .tool(Arc::new(FunctionTool::new("calculator", "Calculate", |_ctx, args| async move {
            Ok(json!({"result": "42"}))
        })))
        .build()?
);

let reasoner_node = AgentNode::new(reasoner)
    .with_output_mapper(|events| {
        let mut updates = std::collections::HashMap::new();
        let mut has_tool_calls = false;

        for event in events {
            if let Some(content) = event.content() {
                for part in &content.parts {
                    if let Part::FunctionCall { name, .. } = part {
                        has_tool_calls = true;
                    }
                }
            }
        }

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

// Build ReAct graph with cycle
let graph = StateGraph::with_channels(&["input", "has_tool_calls", "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);

            if iteration >= 5 { return END.to_string(); }  // Safety limit
            if has_tools { "counter".to_string() } else { END.to_string() }
        },
        [("counter", "counter"), (END, END)],
    )
    .compile()?
    .with_recursion_limit(10);

Node Types

AgentNode

Wraps any ADK Agent (typically LlmAgent) as a graph node:

let node = AgentNode::new(llm_agent)
    .with_input_mapper(|state| {
        // Transform graph state to agent input Content
        adk_core::Content::new("user").with_text(state.get("input").unwrap().as_str().unwrap())
    })
    .with_output_mapper(|events| {
        // Transform agent events to state updates
        let mut updates = HashMap::new();
        // ... extract data from events
        updates
    });

FunctionNode

Simple async functions for data processing:

.node_fn("process", |ctx| async move {
    let data = ctx.get("data").unwrap();
    let result = transform(data);
    Ok(NodeOutput::new().with_update("result", result))
})

SubgraphNode

Run a compiled graph as a node. The inner graph keeps its own channels, edges and interrupt gates, and exchanges named channels with its parent.

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

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

Channels both schemas declare under one name pass through; isolated() requires every exchange to be named. A pause inside pauses the parent and resumes without re-running finished work. A mapping naming a channel neither side declares fails at compile(), as does a subgraph with an interrupt gate but no checkpointer — that one would re-enter at its first node and pay for its work twice.

A node inside can hand control to its parent:

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

Routing From Inside a Node

A conditional edge fixes its targets when the graph is built. A goto does not, and may name any node in the graph:

// A plain node.
Ok(NodeOutput::new().with_update("risk", json!(level)).with_goto([next]))

// An LLM-backed node, routing on what it answered.
AgentNode::new(classifier)
    .with_output_mapper(category_from_events)
    .with_goto_mapper(|updates| match updates.get("category").and_then(|v| v.as_str()) {
        Some("refund") => Some(vec!["refund_desk".to_string()]),
        Some(_) => Some(vec!["general_desk".to_string()]),
        None => None,
    })

A goto replaces that node's declared edges. Naming END stops the branch, and an unknown name fails the run with GraphError::UnknownRouteTarget.

Reliability and Limits

Each of these is off by default, so an existing graph behaves as before.

use adk_graph::checkpoint::RetentionPolicy;
use adk_graph::graph::NodeDefaults;
use adk_graph::retry::RetryPolicy;

let graph = graph
    // Every node retries three times unless it says otherwise.
    .with_node_defaults(NodeDefaults::new().with_retry(RetryPolicy::new(3)))
    // This one gets ten.
    .with_node_retry("call_model", RetryPolicy::new(10))
    // At most four nodes at once, so a wide fan-out cannot trip a rate limit.
    .with_max_concurrency(4)
    // Recover instead of ending the run, once the retry budget is spent.
    .with_node_error_handler("charge", |node, error, _state| {
        Ok(NodeOutput::new()
            .with_update("status", json!(format!("{node}: {error}")))
            .with_goto(["compensate"]))
    })
    // Keep a long-lived thread from growing without bound.
    .with_checkpoint_retention(RetentionPolicy::keep_last(50));

An interrupt is never retried and never reaches an error handler: a pause is not a failure.

Invoking a Node Directly

When the number of sub-tasks comes from state rather than the graph's shape:

use adk_graph::child::RunNodeOptions;

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

The target needs no edge. Each completed child is recorded under <parent>/<child>@<run_id>, so a resumed run returns the recorded answer instead of executing the child again — which matters when the child costs a model call.

A Graph as a Tool

use adk_graph::tool::NodeTool;

let desk = NodeTool::for_graph(Arc::new(graph))
    .with_name("research_desk")
    .with_description("Ask a research question. Pass it as `topic`.");

let agent = LlmAgentBuilder::new("analyst").model(model).tool(Arc::new(desk)).build()?;

The parameter schema is derived from the graph's channels, so the tool description and the graph cannot drift apart. It reports itself long-running, so a graph that pauses travels the existing tool-confirmation path.

State Management

Channels and Reducers

let schema = StateSchema::builder()
    .channel("current")                           // Overwrite (default)
    .list_channel("messages")                     // Append to list
    .channel_with_reducer("count", Reducer::Sum)  // Sum values
    .build();

Checkpointing

// Memory (development)
let checkpointer = MemoryCheckpointer::new();

// SQLite (production)
let checkpointer = SqliteCheckpointer::new("state.db").await?;

// View checkpoint history
let checkpoints = checkpointer.list("thread-id").await?;
for cp in checkpoints {
    println!("Step {}: {:?}", cp.step, cp.state);
}

Durable Resume

When a checkpointer is configured, the executor automatically checks for existing checkpoints before starting. If a checkpoint exists for the thread ID, execution resumes from where it left off — skipping already-completed nodes:

use adk_graph::prelude::*;
use std::sync::Arc;

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

let graph = GraphAgent::builder("resilient_workflow")
    .description("Workflow that survives crashes")
    .node_fn("step_a", |_ctx| async move {
        Ok(NodeOutput::new().with_update("a_done", json!(true)))
    })
    .node_fn("step_b", |_ctx| async move {
        Ok(NodeOutput::new().with_update("b_done", json!(true)))
    })
    .edge(START, "step_a")
    .edge("step_a", "step_b")
    .edge("step_b", END)
    .checkpointer_arc(checkpointer)
    .build()?;

// First run — completes step_a, saves checkpoint, then crashes before step_b
// Second run — resumes from checkpoint, skips step_a, runs only step_b
let result = graph.invoke(State::new(), ExecutionConfig::new("my-thread")).await?;

When streaming, a StreamEvent::Resumed event is emitted to indicate execution was restored from a checkpoint.

Examples

Examples are in the adk-playground repo:

# Parallel LLM agents with callbacks
cargo run --example graph_agent

# Sequential multi-agent pipeline
cargo run --example graph_workflow

# LLM-based sentiment classification and routing
cargo run --example graph_conditional

# ReAct pattern with tools
cargo run --example graph_react

# Multi-agent supervisor
cargo run --example graph_supervisor

# Human-in-the-loop with risk assessment
cargo run --example graph_hitl

# Checkpointing and time travel
cargo run --example graph_checkpoint

Comparison with LangGraph

Feature Flags

License

Apache-2.0

Functional API

The Functional API (feature: functional) provides a higher-level programming model that lets you write agent workflows as normal async Rust functions with automatic checkpointing, typed state reducers, and interrupt/resume support.

Quick Start

[dependencies]
adk-graph = { version = "2.1.0", features = ["functional"] }

use std::sync::Arc;
use adk_graph::checkpoint::MemoryCheckpointer;
use adk_graph::functional::{TaskContext, ReducedValue, MessagesValue, ExecutionLog};
use adk_graph::state::StateSchema;

// Create a TaskContext with a checkpointer
let checkpointer = Arc::new(MemoryCheckpointer::new());
let schema = StateSchema::builder()
    .channel("status")
    .list_channel("results")
    .build();

// ReducedValue — append-only accumulator
let mut results: ReducedValue<String> = ReducedValue::new();
results.push("step 1 complete".to_string());
results.push("step 2 complete".to_string());
assert_eq!(results.len(), 2);

// MessagesValue — chat messages with deduplication
let mut messages = MessagesValue::new();
// Duplicate IDs are replaced (upsert semantics)

// ExecutionLog — resume-skip behavior
let mut log = ExecutionLog::new();
log.record_start("fetch_data");
log.record_completion("fetch_data", serde_json::json!({"ok": true}));
assert!(log.is_completed("fetch_data")); // Skip on resume

Features

Examples

cargo run --manifest-path examples/functional_workflow/Cargo.toml
cargo run --manifest-path examples/background_runs/Cargo.toml
cargo run --manifest-path examples/cron_scheduling/Cargo.toml

Part of ADK-Rust

This crate is part of the ADK-Rust framework for building AI agents in Rust.