그래프 에이전트

LangGraph 스타일 오케스트레이션과 기본 ADK-Rust 통합을 사용하여 복잡하고 상태를 유지하는 워크플로를 구축합니다.

개요

GraphAgent을 사용하면 노드와 엣지가 있는 방향성 그래프로 워크플로를 정의할 수 있으며, 다음을 지원합니다.

• AgentNode: 사용자 지정 입력/출력 매퍼를 사용하여 LLM 에이전트를 그래프 노드로 래핑
• 순환 워크플로: 루프 및 반복적 추론에 대한 기본 지원(ReAct 패턴)
• 조건부 라우팅: 상태에 기반한 동적 엣지 라우팅
• 상태 관리: 리듀서(overwrite, append, sum, custom)를 사용하는 형식이 지정된 상태
• 체크포인팅: 내결함성 및 인간 개입을 위한 영구 상태
• 스트리밍: 여러 스트림 모드(values, updates, messages, debug)

adk-graph 크레이트는 복잡하고 상태를 유지하는 에이전트 워크플로를 구축하기 위한 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 패턴: 추론 + 행동

복잡한 문제를 해결하기 위해 도구를 반복적으로 사용할 수 있는 에이전트를 구축합니다.

시각 자료: 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)을 그래프 노드로 래핑합니다.

에이전트가 보는 것

그래프 내부의 에이전트는 그래프를 시작한 호출에서 파생된 컨텍스트 아래에서 실행되므로, 그래프 외부에서 실행될 때와 동일하게 동작합니다. Runner이 GraphAgent을 호출하면 호출자의 ID와 서비스가 자동으로 전달됩니다:

직접 호출된 그래프 — 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());

참고: 노드는 여전히 자체 메모리 내 그래프 세션에서 실행되므로, 노드 내부의 에이전트 대화 기록은 호출자의 세션에 추가되지 않고 노드 단위로 범위가 지정됩니다.

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

리듀서 유형

업데이트 순서

하나의 슈퍼 스텝에 있는 노드는 동시에 실행되며, 각 작업에 걸리는 시간에 따라 서로 다른 순서로 완료됩니다. 노드의 상태 업데이트는 노드가 완료된 순서가 아니라 노드 이름 순서로 적용됩니다.

리듀서가 교환 법칙을 따르지 않는 경우 순서가 중요합니다. Append은 배열을 생성하므로 순서가 결과가 되며, Custom 리듀서도 순서에 민감할 수 있습니다. 노드 이름으로 정렬하면 실행을 재현할 수 있습니다. 느린 종속성의 실행 시점과 관계없이 동일한 그래프와 동일한 입력은 동일한 상태를 생성합니다.

하나의 노드가 여러 채널에 기록하는 경우, 채널 이름 순서로 적용됩니다. 채널은 서로 상호 작용하지 않으므로, 이는 추적을 읽을 때만 중요합니다.

체크포인트

내결함성과 휴먼 인 더 루프를 위해 영구 상태를 활성화합니다.

인메모리(개발)

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

체크포인트에 기록되는 내용

체크포인트에는 누적된 상태, 단계 번호, 그리고 프런티어(아직 실행해야 하는 노드)가 저장됩니다. 프런티어가 진행된 후 기록되므로, 재개할 때 이미 완료된 노드가 다시 실행되지 않으며 해당 노드의 업데이트가 두 번 적용되지도 않습니다. 완료된 실행은 빈 프런티어를 체크포인트로 저장하므로, 완료된 스레드를 재개하면 그래프를 다시 시작하는 대신 최종 상태를 반환합니다.

다음 두 경우에는 중단된 노드가 아직 업데이트를 생성하지 않았고 재개할 때 다시 실행해야 하므로, 의도적으로 다음 프런티어가 아니라 실행 중이던 프런티어를 체크포인트로 저장합니다:

스트리밍 실행은 블로킹 실행과 동일한 일정으로 체크포인트를 생성합니다. 인터럽트로 스트림이 종료되는 경우에도 마찬가지이므로, human-in-the-loop 일시 중지는 어느 실행 모드에서든 재개할 수 있습니다.

체크포인트 기록 (시간 여행)

읽기 전용입니다. TimeTravelHandle::state_history(from, to)은 각 체크포인트 단계에서 저장된 상태를 반환합니다. 아무것도 실행하지 않습니다. 노드가 실행되지 않고, 이벤트가 다시 생성되지 않으며, 부작용도 반복되지 않습니다. 기록의 특정 지점부터 다시 실행하려면 fork_at를 사용하여 해당 체크포인트에서 분기한 다음 분기된 스레드에서 그래프를 호출하세요. 이 메서드의 이전 이름은 replay이었으며 그래프를 다시 실행한다고 문서화되어 있었지만, 실제로는 그렇게 동작한 적이 없습니다.

체크포인트는 영구적 재개도 지원합니다. 그래프 실행이 중단되거나 프로세스가 재시작되면 처음부터 다시 시작하는 대신 마지막으로 영구 저장된 체크포인트부터 실행을 재개합니다. 충돌에 안전한 영속성을 위해 SqliteCheckpointer(sqlite 기능)을 사용하세요. MemoryCheckpointer은 프로세스 내에 체크포인트를 보관하므로 재시작 후에도 유지되지 않습니다. 이 두 가지가 이 크레이트에서 제공하는 백엔드이며, 그 외의 항목에는 Checkpointer 트레이트를 구현하세요.

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

Human-in-the-Loop

동적 인터럽트를 사용하여 human approval을 위해 실행을 일시 중지합니다.

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을 래핑하는 기본 execute_stream은 이 작업을 대신 수행합니다.

시간 초과 정책은 스트리밍된 실행 자체에 적용됩니다. 스트림의 경우 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()의 정책은 10회의 시도를 허용하며, 9회의 대기 시간은 합쳐서 약 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을 보고합니다.

그래프 전체 기본값

스무 개의 노드에 동일한 재시도 설정을 반복하면 누락으로 인해 쉽게 잘못 구성할 수 있습니다.

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)은 1로 설정되며 모든 체크포인트가 보존 기간을 지난 스레드도 하나의 체크포인트를 유지합니다.

기본적으로 비활성화되어 있으므로 기존 스레드는 전체 기록을 유지하고 시간 이동으로 모든 단계에 계속 도달할 수 있습니다. 스레드가 장기간 실행되고 먼 과거로 되돌릴 필요가 없다면 정책을 설정하세요.

선언되지 않은 채널 거부

스키마에 선언되지 않은 채널에는 알 수 없는 이름에 대한 대체 동작인 덮어쓰기 리듀서가 적용됩니다. 목록 채널을 선언한 그래프가 이름을 잘못 입력한 유사 채널에 기록하면 마지막 값만 유지되며 아무런 보고도 하지 않습니다.

let graph = graph.with_strict_channels();

선언되지 않은 채널에 기록하는 노드는 GraphError::UndeclaredChannel과 함께 런을 실패시키며, 노드와 채널의 이름을 표시합니다. 기본적으로 비활성화되어 있으며, 그래프가 채널을 전혀 선언하지 않은 경우에는 아무 동작도 하지 않습니다.

ADK 통합

GraphAgent은 ADK Agent 트레이트를 구현하므로 다음과 함께 사용할 수 있습니다.

• 실행기: 표준 실행에 adk-runner과 함께 사용
• 콜백: 실행 전/후 콜백을 완전히 지원
• 세션: 대화 기록에 adk-session와 함께 사용
• 스트리밍: 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는 작업을 스케줄러에 넘기고 리듀서를 통해 수집합니다. 또한 서브그래프에서 부모 그래프로 이동하는 홉에 해당하는 기능은 아직 여기에 없습니다.

---

이전: ← 멀티 에이전트 시스템 | 다음: 실시간 에이전트 →