# adk-graph
Graph-based workflow orchestration for Rust Agent Development Kit (ADK-Rust) agents, inspired by LangGraph.
[](https://crates.io/crates/adk-graph)
[](https://docs.rs/adk-graph)
[](LICENSE)
## 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`:
```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
```rust
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
```rust
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
```rust
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
```rust
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:
```rust
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:
```rust
.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.
```rust
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:
```rust
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:
```rust
// 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.
```rust
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));
```
| Super-steps per run | 100 (`recursion_limit`) |
| Retry when no policy is attached | one attempt |
| `RetryPolicy::default()` | ten attempts, about 243s of backoff in total |
| Concurrency | the whole frontier |
| Checkpoints kept | every one, until a retention policy is set |
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:
```rust
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
```rust
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
```rust
let schema = StateSchema::builder()
.channel("current") // Overwrite (default)
.list_channel("messages") // Append to list
.channel_with_reducer("count", Reducer::Sum) // Sum values
.build();
```
## Checkpointing
```rust
// 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:
```rust
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](https://github.com/zavora-ai/adk-playground) repo:
```bash
# 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
| State Management | TypedDict + Reducers | StateSchema + Reducers |
| Execution Model | Pregel super-steps | Pregel super-steps |
| Checkpointing | Memory, SQLite, Postgres | Memory, SQLite |
| Human-in-Loop | interrupt_before/after | interrupt_before/after + dynamic |
| Streaming | 5 modes | 5 modes |
| Cycles | Native support | Native support |
| Type Safety | Python typing | Rust type system |
| LLM Integration | LangChain | AgentNode + ADK agents |
## Feature Flags
| `sqlite` | Enable SQLite checkpointer |
| `functional` | Functional API: TaskContext, typed reducers, schema validation, proc macros |
| `full` | Enable all features |
## 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
```toml
[dependencies]
adk-graph = { version = "2.1.0", features = ["functional"] }
```
```rust
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
| `TaskContext` | Runtime context for tasks (state, checkpointing, streaming, interrupts) |
| `ReducedValue<T>` | Append-only accumulator persisted across checkpoints |
| `UntrackedValue<T>` | Transient data excluded from checkpoints |
| `MessagesValue` | Chat messages with ID-based deduplication |
| `StateSchemaValidator` | Type validation for state and task output |
| `ExecutionLog` | Task completion tracking for resume-skip |
| `TypedReducer` | Custom merge strategies (Replace, Append, Merge) |
### Examples
```bash
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](https://adk-rust.com) framework for building AI agents in Rust.