图代理

使用基于 LangGraph 风格的编排与原生 ADK-Rust 集成，构建复杂的有状态工作流。

概述

GraphAgent 允许你将工作流定义为包含节点和边的有向图，并支持：

• AgentNode：将 LLM 代理封装为图节点，并使用自定义输入/输出映射器
• 循环工作流：原生支持循环和迭代推理（ReAct 模式）
• 条件路由：根据状态动态路由边
• 状态管理：使用归约器管理类型化状态（覆盖、追加、求和、自定义）
• 检查点：持久化状态，以实现容错和人在回路
• 流式传输：多种流模式（值、更新、消息、调试）

adk-graph crate 提供 LangGraph 风格的工作流编排，用于构建复杂的有状态代理工作流。它将基于图的工作流能力引入 ADK-Rust 生态系统，同时保持与 ADK 代理系统的完全兼容。

主要优势：

• 可视化工作流设计：将复杂逻辑定义为直观的节点和边图
• 并行执行：多个节点可以同时运行，以提升性能
• 状态持久化：内置检查点功能，以实现容错和人在回路
• LLM 集成：原生支持将 ADK 代理封装为图节点
• 灵活路由：静态边、条件路由和动态决策

在工作流代理和图之间进行选择

ADK-Rust 支持两种编排方式。二者互不取代，并且都会 持续维护。

当工作形态已知且是线性的：工作流代理写起来更简短，也无需维护状态架构时，优先使用工作流代理。当控制流取决于结果，或一次运行必须持续超过进程生命周期时，则应使用图。

它们可以组合

这三个工作流代理实现了 Agent，而 AgentNode 会包装任意 Agent，因此工作流代理就是一个图节点：

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 也实现了 Agent，因此反过来也成立：图可以成为 SequentialAgent 的子代理。将需要分支或持久性的部分交给图处理，而不需要这些能力的部分则使用工作流代理。

你将构建的内容

在本指南中，你将创建一个文本处理流水线，并行运行翻译和摘要任务：

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

核心概念：

• 节点 - 执行工作的处理单元（LLM 代理、函数或自定义逻辑）
• 边 - 节点之间的控制流（静态连接或条件路由）
• 状态 - 在图中流动并在节点之间持久保存的共享数据
• 并行执行 - 多个节点可以同时运行，以提升性能

了解核心组件

🔧 节点：工作单元 节点是实际执行工作的地方。每个节点都可以：

• AgentNode：包装一个 LLM 代理来处理自然语言
• 函数节点：执行自定义 Rust 代码来处理数据
• 内置节点：使用预定义的逻辑，例如计数器或验证器

可以将节点想象成装配线上的专业工作单元——每个节点都有特定的任务和专长。

🔀 边：流程控制
边决定执行如何在图中流动：

• 静态边：直接连接（A → B → C）
• 条件边：基于状态的动态路由（if sentiment == "positive" → positive_handler）
• 并行边：从一个节点出发的多条路径（START → [translator, summarizer]）

边就像交通信号灯和道路标志，用于指引工作流。

💾 状态：共享内存
状态是一个所有节点都可以读取和写入的键值存储：

• 输入数据：馈送到图中的初始信息
• 中间结果：一个节点的输出成为另一个节点的输入
• 最终输出：完成所有处理后的结果

状态就像一块共享白板，节点可以在上面留下信息供其他节点使用。

⚡ 并行执行：速度提升
当多条边离开一个节点时，这些目标节点会同时运行：

• 更快的处理：独立任务同时运行
• 资源效率：更充分地利用 CPU 和 I/O
• 可扩展性：处理更复杂的工作流，而不会出现线性减速

这就像让多个工作人员同时处理一项工作中的不同部分，而不是排队等待。

---

快速开始

1. 创建项目

cargo new graph_demo
cd graph_demo

将依赖项添加到 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"

使用你的 API 密钥创建 .env：

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

2. 并行处理示例

下面是一个并行处理文本的完整可运行示例：

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

示例输出：

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

图执行的工作原理

整体概览

图代理以超级步骤执行——所有已就绪的节点并行运行，然后图等待它们全部完成后再进入下一步：

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 ✅

状态在节点间的流动

每个节点都可以读取和写入共享状态：

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

它为何有效

---

使用 LLM 分类进行条件路由

构建智能路由系统，让 LLMs 决定执行路径：

可视化：基于情感的路由

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

完整示例代码

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

示例流程：

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

ReAct 模式：推理 + 行动

构建能够迭代使用工具来解决复杂问题的 agent：

可视化：ReAct 循环

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

完整的 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(())
}

示例流程：

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

将任意 ADK Agent（通常为 LlmAgent）封装为图节点：

agent 所看到的内容

图中的 agent 在启动该图的调用所派生的上下文中运行，因此其行为与在图外运行时相同。当 Runner 调用 GraphAgent 时，调用方的身份和服务会自动传递：

直接调用的图 — graph.invoke(state, ExecutionConfig::new("thread")) — 没有可继承的调用。这就是独立模式：节点会获得 user_id = "graph_user"、app_name = "graph_app"、分支 main，但没有密钥，也没有 内存。这是一种用于在 Runner 外部运行图的有意模式，而不是生产环境中应当采用的备用方案。

要手动连接——例如从你自己的执行器驱动图时——请显式传递调用：

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

**注意：**节点仍会在其自身的内存图会话中运行，因此节点内部的 agent 对话历史的作用域属于该节点，而不是追加到调用方的会话中。

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

函数节点

用于处理状态的简单异步函数：

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

边类型

静态边

节点之间的直接连接：

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

条件边

基于状态的动态路由：

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

从节点内部进行路由

条件边会在构建图时确定其目标。NodeOutput::with_goto 则不会：节点写入状态，并在同一步骤中指定其后继节点；它可以指定图中的任意节点， 包括与其没有边连接的节点。

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

goto 产生的前沿会像其他前沿一样被检查点保存，因此暂停的运行会恢复到 goto 所选择的节点。

注意： 当你构建图时就已知可能的目标时，请使用 add_conditional_edges——这样边就会出现在渲染后的图中。当选择权属于节点时，请使用 goto。

路由辅助工具

对于常见模式，请使用内置路由器：

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

并行执行

来自单个节点的多条边会并行执行：

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 在三个分支全部到达后只运行一次。上面的代码没有明确要求这一点：在编译时，具有多个直接入边的节点会自动延迟执行。因此，即使分支长度不同，也能正确汇合，无需额外配置。

由计数方式可得出以下两个细节：

循环图（ReAct 模式）

构建具有循环的迭代推理智能体：

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

多智能体监督器

将任务路由至专业智能体：

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

状态管理

带归约器的状态架构

控制状态更新的合并方式：

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

归约器类型

更新顺序

一个 super-step 中的节点会并发运行，并按照各自工作所需的时间以任意顺序完成。它们的状态更新会按节点名称顺序应用，而不是按节点完成的顺序应用。

只要 reducer 不满足交换律，顺序就很重要。Append 构建一个数组，因此顺序就是结果；Custom reducer 也可能对顺序敏感。按节点名称排序可以使运行具有可复现性：相同的图和相同的输入会得到相同的状态，而不受缓慢依赖项的执行时序影响。

如果一个节点写入多个通道，则这些通道会按通道名称顺序应用。通道之间不会相互影响，因此这只会影响跟踪记录的读取。

检查点

启用持久化状态，以实现容错和人在回路中的处理：

内存中（开发）

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（生产）

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

检查点记录的内容

检查点会存储累积的状态、步骤编号以及前沿——仍需运行的节点。它会在前沿推进后写入，因此恢复时不会重新执行已经完成的节点，也不会重复应用其更新。运行结束时会为一个空的前沿创建检查点，因此恢复已完成的线程时会返回最终状态，而不是重新启动图。

在以下两种情况下，检查点会特意记录当时正在执行的前沿，而不是下一个前沿，因为被中断的节点尚未生成其更新，恢复时必须再次运行：

流式运行与阻塞运行按照相同的计划创建检查点，包括中断结束流的情况，因此人机协同暂停可在任一执行模式下恢复。

检查点历史（时间旅行）

仅读取。 TimeTravelHandle::state_history(from, to) 返回每个已创建检查点的步骤中存储的状态。它不会执行任何操作——不会运行节点、不会重新生成事件，也不会重复任何副作用。若要从历史中的某个点重新运行，请使用 fork_at 为该检查点创建分支，并在分叉的线程上调用图。该方法以前名为 replay，文档中称其会重新执行图，但它实际上从未这样做。

检查点还支持持久化恢复——如果图执行崩溃或进程重启，执行会从最后一个已持久化的检查点恢复，而不是从头开始。使用 SqliteCheckpointer（sqlite 功能）实现崩溃安全的持久化。MemoryCheckpointer 将检查点保存在进程中，因此重启后不会保留。这两个是此 crate 提供的后端；如需其他后端，请实现 Checkpointer trait。

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

人机协同

使用动态中断暂停执行，以获取人工批准：

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

静态中断

使用 interrupt_before 或 interrupt_after 设置强制暂停点：

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

流式执行

在图执行时流式传输事件：

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

流模式

Messages 模式会在 Node::execute_stream 生成令牌时读取它们。 在此模式下，每个节点在每个超级步骤中仅运行一次：节点会将其状态 更新作为 StreamEvent::Updates 事件报告到流中，执行器会应用这些更新， 而不是再次执行节点来收集它们。这对于 AgentNode 尤其重要，因为 再次执行意味着每个节点会产生第二次计费的模型调用。

重要提示： 覆盖 execute_stream 的自定义 Node 必须生成一个携带其状态更新的 StreamEvent::Updates 事件。否则，该节点会在流中生成事件，但在 Messages 模式下不会贡献状态。默认的 execute_stream 会包装 execute，并会为你完成此操作。

超时策略适用于流式执行本身。对于流来说，idle_timeout 表示在时限内没有生成任何事件。

子图

已编译的图通过 SubgraphNode 作为另一个图的节点运行。内部图 保留自己的通道、边和中断门，并与其父图交换命名通道。

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

最后一行是值得了解的区别：两个架构在任何操作运行前都可用，因此，命名了任一方未声明的通道的映射无法到达某次运行，也不会表现为缺失值。完全不交换任何内容的子图也会以同样的方式被拒绝，因为它无法影响其父图。

恢复子图内的暂停

无需额外操作。在同一线程上再次调用父图会重新进入子图，子图会在 <parent thread>/<node name> 上找到自己的检查点，并从停止的位置继续执行。子图在暂停前已完成的工作不会重复执行；暂停发生在更深层级时也会以相同方式恢复——消息会命名它经过的每一层。

声明了中断门但未持有检查点保存器的子图，会在父图编译时被拒绝：它会从第一个节点重新进入，并为已完成的工作再次付费。

由于两个图都持有真实的检查点器，因此进程重启后仍能继续运行：一组仅共享数据库的新图对象可以恢复同一次运行。

将控制权交还给父图

子图中的节点可以结束自身所在的图，并指定承载它的图中的一个节点：

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

子图完成后会照常投影其输出通道，然后父图从 escalate 继续执行，而不是沿着子图节点声明的边继续。父图会验证目标，因为只有它知道自身的节点。

可靠性和成本控制

以下各项默认均处于关闭状态，因此在设置之前，图的行为与原来一致。

按节点重试

瞬时故障——例如速率限制或连接断开——否则会直接结束运行。

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

延迟按 backoff_factor 增长，上限为 max_delay，随后会应用抖动。没有策略的节点只运行一次，因此重试仍需显式启用。RetryPolicy::default() 中的策略允许尝试十次，九次休眠总计约 243 秒——如果调用方正在等待响应，应调低 max_attempts。

无论 retry_on 如何设置，中断都不会重试：暂停不是失败。尝试次数会被记录在检查点中，因此恢复运行时会继续使用剩余预算，而不是重新开始计算。

限制并发

大规模扇出会一次性调度整个前沿，这可能耗尽连接池或触发提供方的速率限制。

let graph = graph.with_max_concurrency(4);

超过上限的节点会等待可用槽位。准入顺序是按名称排序后的前沿，因此不依赖时序。命令式子调用不计入此预算，因为父节点在持有自身槽位时会等待其子调用完成。

按节点超时

TimeoutPolicy 限制单次尝试，并与 idle_timeout 一起规定节点在不报告进度的情况下最长可以持续多久。超过任一限制都会产生 GraphError::NodeTimedOut，之后重试策略可以对其采取操作。

直接调用节点

当子任务的数量来自状态，而不是图的结构时，节点可以自行调用另一个节点：

use adk_graph::child::RunNodeOptions;

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

目标节点不需要边。每个已完成的子任务都会记录在 <parent>/<child>@<run_id> 下，因此恢复运行时会返回已记录的答案，而不是再次执行子任务——当子任务需要调用模型时，这一点十分重要。

节点缓存

在节点上使用 cache_policy 会根据节点名称和当前状态为其结果建立键，并可选使用 TTL，因此输入未发生变化时可以跳过相关工作。需要启用 node-cache 功能；基于 Redis 的存储可通过 redis-cache 使用。

增量检查点

delta 功能存储超级步骤之间的差异，而不是整个状态；当状态较大且步骤较多时，这一点很重要。

时间旅行

启用 time-travel 功能后，graph.time_travel(thread_id)? 会返回一个句柄，用于访问线程的检查点历史：列出步骤、读取某个步骤的状态，或通过 fork_at 从某个检查点分支出一个新线程。该调用返回 Result，因为每项操作都会读取检查点，因此没有检查点保存器的图会报告 GraphError::CheckpointError，而不是发生 panic。

图级默认值

在二十个节点中重复设置相同的重试策略，很容易因遗漏而出错。

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

节点级值始终优先。NodeDefaults 还包含超时设置和失败处理程序。

从节点失败中恢复

节点的重试额度耗尽后，处理程序可以记录发生的情况并指定一个恢复节点，而不是结束运行：

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

返回 Err 会像之前一样结束运行。中断永远不会到达处理程序，因为暂停并不是失败。

限制检查点增长

一个线程每个超级步骤都会累积一个检查点。因此，持续数天的运行会无限增长，从而增加存储开销并减慢 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)),
    );

每次保存后都会进行裁剪，因此开销始终与运行规模成正比，也不需要外部任务。最新的检查点永远不会被丢弃，无论策略如何，因为恢复时会加载它 — keep_last(0) 会被提升为 一，并且所有检查点都超过期限的线程也会保留一个检查点。

默认关闭，因此现有线程会保留其完整历史记录，时间旅行仍可到达每个步骤。当线程生命周期较长且不需要回溯太远时，请设置策略。

拒绝未声明的通道

架构未声明的通道会采用覆盖归约器，因为这是未知名称的回退方式。声明了列表通道的图随后写入一个相近但不匹配的名称时，只会保留最后一个值，也不会报告任何信息。

let graph = graph.with_strict_channels();

写入未声明通道的节点随后会使运行失败，并显示 GraphError::UndeclaredChannel，同时指出节点和通道。默认关闭；当图完全未声明任何通道时，该选项也不会产生作用。

ADK 集成

GraphAgent 实现了 ADK Agent trait，因此可与以下组件配合使用：

• Runner：与 adk-runner 一起用于标准执行
• Callbacks：完整支持执行前/执行后回调
• Sessions：与 adk-session 一起用于对话历史记录
• Streaming：返回 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

示例

此仓库中经过验证的图示例：

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

包含真实 LLM 集成的完整图示例集位于 adk-playground。

与 LangGraph 的比较

有两点差异值得明确说明。run_node_with 会内联返回子级结果并记录该结果，因此恢复运行时不会再次为已完成的子级付费；而 Send 则将工作交给调度器，并通过归约器收集结果。此外，从子图跳转到其父图目前还没有对应的实现。

---

上一页：← 多智能体系统 | 下一页：实时智能体 →