ग्राफ एजेंट

LangGraph-शैली ऑर्केस्ट्रेशन के साथ native ADK-Rust integration का उपयोग करके जटिल, stateful workflows बनाएँ।

अवलोकन

GraphAgent आपको nodes और edges वाले directed graphs के रूप में workflows परिभाषित करने की सुविधा देता है, जिसमें निम्नलिखित का समर्थन है:

• AgentNode: LLM agents को custom input/output mappers के साथ graph nodes के रूप में लपेटें
• Cyclic Workflows: loops और iterative reasoning (ReAct pattern) के लिए native support
• Conditional Routing: state के आधार पर dynamic edge routing
• State Management: reducers (overwrite, append, sum, custom) के साथ typed state
• Checkpointing: fault tolerance और human-in-the-loop के लिए persistent state
• Streaming: multiple stream modes (values, updates, messages, debug)

adk-graph crate जटिल, stateful agent workflows बनाने के लिए LangGraph-शैली workflow orchestration प्रदान करता है। यह ADK-Rust ecosystem में graph-based workflow capabilities लाता है और साथ ही ADK के agent system के साथ पूर्ण compatibility बनाए रखता है।

मुख्य लाभ:

• Visual Workflow Design: जटिल logic को intuitive node-and-edge graphs के रूप में परिभाषित करें
• Parallel Execution: बेहतर performance के लिए multiple nodes एक साथ चल सकते हैं
• State Persistence: fault tolerance और human-in-the-loop के लिए built-in checkpointing
• LLM Integration: ADK agents को graph nodes के रूप में लपेटने के लिए native support
• Flexible Routing: static edges, conditional routing, और dynamic decision making

Workflow Agents और Graph के बीच चयन

ADK-Rust orchestration के दो तरीके support करता है। इनमें से कोई भी दूसरे का स्थान नहीं लेता, और दोनों maintain किए जाते हैं।

जब कार्य का स्वरूप ज्ञात और रैखिक हो, तो workflow agents को प्राथमिकता दें: इन्हें लिखना छोटा होता है और बनाए रखने के लिए कोई state schema नहीं होता। जब control flow परिणामों पर निर्भर हो, या जब किसी run को process से अधिक समय तक जारी रहना हो, तब graph का उपयोग करें।

ये संयोजित होते हैं

तीनों workflow agents Agent को लागू करते हैं, और AgentNode किसी भी Agent को wrap करता है, इसलिए workflow agent एक graph node होता है:

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 को लागू करता है, इसलिए इसका उल्टा भी सही है: कोई graph, SequentialAgent का sub-agent हो सकता है। जिस भाग में branching या durability की आवश्यकता हो, उसके लिए graph का उपयोग करें और जिन भागों में इनकी आवश्यकता न हो, उनके लिए workflow agents का उपयोग करें।

आप क्या बनाएँगे

इस गाइड में, आप एक Text Processing Pipeline बनाएँगे, जो translation और summarization को समानांतर रूप से चलाएगी:

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

मुख्य अवधारणाएँ:

• Nodes - ऐसे processing units जो कार्य करते हैं (LLM agents, functions, या custom logic)
• Edges - nodes के बीच control flow (static connections या conditional routing)
• State - साझा data जो graph में प्रवाहित होता है और nodes के बीच बना रहता है
• Parallel Execution - बेहतर performance के लिए कई nodes एक साथ चल सकते हैं

Core Components को समझना

🔧 Nodes: Workers Nodes वह स्थान हैं जहाँ वास्तविक कार्य होता है। प्रत्येक node:

• AgentNode: प्राकृतिक भाषा को process करने के लिए किसी LLM agent को wrap करता है
• Function Node: data processing के लिए custom Rust code चलाता है
• Built-in Nodes: counters या validators जैसी पूर्वनिर्धारित logic का उपयोग करता है

Nodes को assembly line में काम करने वाले विशेषज्ञ workers की तरह समझें - प्रत्येक का एक विशिष्ट कार्य और विशेषज्ञता होती है।

🔀 किनारे: प्रवाह नियंत्रण

किनारे यह निर्धारित करते हैं कि आपके ग्राफ़ में निष्पादन कैसे आगे बढ़ता है:

• स्थिर किनारे: प्रत्यक्ष कनेक्शन (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 को इनवोक करता है, तो कॉलर की पहचान और सेवाएँ स्वचालित रूप से आगे पहुंचा दी जाती हैं:

सीधे लागू किया गया एक ग्राफ — graph.invoke(state, ExecutionConfig::new("thread")) — को विरासत में पाने के लिए कोई invocation नहीं होता। यह standalone mode है: नोड को user_id = "graph_user", app_name = "graph_app", शाखा main, कोई secrets और कोई memory नहीं मिलता। यह किसी Runner के बाहर ग्राफ चलाने के लिए एक जानबूझकर चुना गया मोड है, production में अपनाने के लिए कोई fallback नहीं।

मैन्युअल रूप से जोड़ने के लिए — उदाहरण के लिए, जब आप अपने executor से ग्राफ चला रहे हों — invocation को स्पष्ट रूप से पास करें:

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

नोट: नोड अभी भी अपनी अलग in-memory graph session में चलता है, इसलिए नोड के भीतर agent conversation history, caller की session में जोड़ने के बजाय, नोड तक सीमित रहती है।

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

Function Nodes

सरल async functions जो state को प्रोसेस करते हैं:

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

Static Edges

नोड्स के बीच सीधे कनेक्शन:

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

Conditional Edges

state के आधार पर dynamic routing:

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

Routing From Inside a Node

Conditional edge, ग्राफ बनाए जाने पर अपने targets निर्धारित करता है। NodeOutput::with_goto ऐसा नहीं करता: कोई नोड state लिखता है और उसी step में अपने successors के नाम देता है, और वह ग्राफ के किसी भी नोड का नाम दे सकता है, जिसमें वह नोड भी शामिल है जिससे उसका कोई edge नहीं है।

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 द्वारा उत्पन्न frontier को भी किसी अन्य frontier की तरह checkpoint किया जाता है, इसलिए रुका हुआ run उस node में फिर से शुरू होता है जिसे goto ने चुना था।

नोट: जब graph बनाते समय संभावित targets ज्ञात हों, तो add_conditional_edges का उपयोग करें — तब edges रेंडर किए गए diagram में दिखाई देते हैं। जब चुनाव node पर निर्भर हो, तो goto का उपयोग करें।

Router Helpers

सामान्य पैटर्न के लिए अंतर्निहित routers का उपयोग करें:

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

Parallel Execution

एकल node से निकलने वाले कई edges समानांतर रूप से execute होते हैं:

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 तीनों branches के आने के बाद एक बार चलता है। ऊपर दिए गए code में ऐसा करने के लिए कुछ भी नहीं कहा गया है: एक से अधिक incoming direct edge वाले node को compile time पर स्वचालित रूप से defer किया जाता है। इसलिए अलग-अलग लंबाई वाली branches बिना किसी configuration के सही ढंग से join होती हैं।

गणना किए जाने के तरीके से दो बातें निकलती हैं:

चक्रीय ग्राफ़ (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 में nodes समवर्ती रूप से चलते हैं और उनका कार्य जितना समय लेता है, उसके अनुसार किसी भी क्रम में समाप्त होते हैं। उनके state updates node-name क्रम में लागू किए जाते हैं, nodes के समाप्त होने के क्रम में नहीं।

क्रम तब महत्वपूर्ण होता है जब reducer commutative न हो। Append एक array बनाता है, इसलिए क्रम ही परिणाम होता है; Custom reducer भी क्रम-संवेदी हो सकता है। node name के अनुसार sorting करने से run पुनरुत्पाद्य बनता है: समान graph और समान input से slow dependency के timing की परवाह किए बिना समान state प्राप्त होती है।

जब एक node द्वारा कई channels में लिखा जाता है, तो वे channel-name क्रम में लागू किए जाते हैं। Channels एक-दूसरे से interact नहीं करते, इसलिए यह केवल trace पढ़ने के लिए महत्वपूर्ण है।

Checkpointing

fault tolerance और human-in-the-loop के लिए persistent state सक्षम करें:

In-Memory (Development)

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

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

Checkpoint क्या रिकॉर्ड करता है

एक checkpoint संचित state, step number और frontier — अर्थात् वे nodes जिन्हें अभी चलना बाकी है — को संग्रहीत करता है। इसे frontier के आगे बढ़ने के बाद लिखा जाता है, इसलिए resume करने पर पहले से पूर्ण हो चुका node दोबारा execute नहीं होता और उसके updates दो बार लागू नहीं होते। समाप्त होने वाला run एक खाली frontier का checkpoint बनाता है, इसलिए पूर्ण thread को resume करने पर graph फिर से शुरू होने के बजाय final state लौटती है।

दो स्थितियों में अगले frontier के बजाय जानबूझकर उस frontier का checkpoint बनाया जाता है जो executing था, क्योंकि बाधित node ने अभी अपने updates उत्पन्न नहीं किए हैं और resume होने पर उसे फिर से चलना आवश्यक है:

स्ट्रीम किए गए रन, ब्लॉकिंग रन के समान शेड्यूल पर चेकपॉइंट होते हैं, जिसमें वह स्थिति भी शामिल है जब कोई इंटरप्ट स्ट्रीम को समाप्त करता है, इसलिए मानव-इन-द-लूप विराम किसी भी निष्पादन मोड में फिर से शुरू किया जा सकता है।

चेकपॉइंट इतिहास (समय यात्रा)

केवल पढ़ता है। TimeTravelHandle::state_history(from, to) प्रत्येक चेकपॉइंट किए गए चरण पर संग्रहीत स्थिति लौटाता है। यह कुछ भी निष्पादित नहीं करता — कोई नोड नहीं चलता, कोई इवेंट फिर से जनरेट नहीं होता और कोई साइड इफ़ेक्ट दोबारा नहीं होता। इतिहास में किसी बिंदु से फिर से चलाने के लिए, उस चेकपॉइंट की शाखा बनाने हेतु fork_at का उपयोग करें और फोर्क किए गए थ्रेड पर ग्राफ़ को इनवोक करें। इस मेथड का नाम पहले replay था और इसे ग्राफ़ को फिर से निष्पादित करने वाला बताया गया था, जबकि इसने ऐसा कभी नहीं किया।

चेकपॉइंट टिकाऊ पुनरारंभ भी सक्षम करते हैं — यदि ग्राफ़ का निष्पादन क्रैश हो जाता है या प्रोसेस फिर से शुरू होता है, तो निष्पादन शुरू से आरंभ होने के बजाय अंतिम स्थायी चेकपॉइंट से फिर शुरू होता है। क्रैश-सुरक्षित स्थायित्व के लिए SqliteCheckpointer (sqlite सुविधा) का उपयोग करें। MemoryCheckpointer चेकपॉइंट को प्रोसेस में रखता है, इसलिए वे रीस्टार्ट होने के बाद सुरक्षित नहीं रहते। ये दोनों बैकएंड इस क्रेट में उपलब्ध हैं; किसी अन्य चीज़ के लिए 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 के लिए सबसे अधिक महत्वपूर्ण है, जहाँ दूसरी बार चलने का अर्थ प्रत्येक नोड के लिए मॉडल कॉल का दूसरा बिल लगना होगा।

महत्वपूर्ण: एक कस्टम Node, जो execute_stream को ओवरराइड करता है, उसे एक 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()?;

वह अंतिम पंक्ति जानने योग्य अंतर है: दोनों स्कीमा कुछ भी चलने से पहले उपलब्ध होते हैं, इसलिए ऐसा मैपिंग जो किसी ऐसे चैनल का नाम लेता है जिसे कोई भी पक्ष घोषित नहीं करता, किसी रन तक पहुँचकर अनुपस्थित मान के रूप में सामने नहीं आ सकता। ऐसा उपग्राफ जो बिल्कुल भी कुछ आदान-प्रदान नहीं करता, उसी तरह अस्वीकार कर दिया जाता है, क्योंकि वह अपने पैरेंट को प्रभावित नहीं कर सकता।

उपग्राफ के अंदर रुके हुए कार्य को फिर से शुरू करना

कुछ अतिरिक्त आवश्यक नहीं है। उसी थ्रेड पर पैरेंट को फिर से invoke करने से उपग्राफ में दोबारा प्रवेश होता है, जो <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 चाहे कुछ भी कहे, इंटरप्ट का कभी पुनःप्रयास नहीं किया जाता: विराम विफलता नहीं है। प्रयासों की संख्या चेकपॉइंट की जाती है, इसलिए फिर से शुरू किया गया रन बजट को जारी रखता है, उसे पुनः आरंभ नहीं करता।

समवर्तीता की सीमा निर्धारित करना

एक विस्तृत fan-out अपने पूरे फ्रंटियर को एक साथ डिस्पैच करता है, जिससे कनेक्शन पूल समाप्त हो सकता है या प्रदाता की दर सीमा सक्रिय हो सकती है।

let graph = graph.with_max_concurrency(4);

सीमा से आगे के नोड किसी स्लॉट के लिए प्रतीक्षा करते हैं। प्रवेश क्रम नाम के अनुसार सॉर्ट किया गया फ्रंटियर होता है, इसलिए यह समय-निर्धारण पर निर्भर नहीं करता। इम्पेरेटिव चाइल्ड इनवोकेशन इस बजट से बाहर होते हैं, क्योंकि पैरेंट अपने स्लॉट को धारण करते हुए अपने चिल्ड्रन का इंतज़ार करता है।

प्रति-नोड टाइमआउट

एक TimeoutPolicy किसी एक प्रयास की सीमा तय करता है और idle_timeout के साथ यह निर्धारित करता है कि कोई नोड प्रगति की सूचना दिए बिना कितनी देर तक चल सकता है। इनमें से किसी भी सीमा के पार जाने पर GraphError::NodeTimedOut मिलता है, जिस पर retry policy आगे कार्रवाई कर सकती है।

किसी नोड को सीधे invoke करना

जब sub-tasks की संख्या graph के आकार के बजाय state से आती है, तो कोई नोड स्वयं किसी अन्य नोड को invoke कर सकता है:

use adk_graph::child::RunNodeOptions;

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

लक्ष्य के लिए किसी edge की आवश्यकता नहीं होती। पूरा किया गया प्रत्येक child <parent>/<child>@<run_id> के अंतर्गत दर्ज किया जाता है, इसलिए resumed run दर्ज किए गए उत्तर को लौटाता है, child को दोबारा execute नहीं करता — यह तब महत्वपूर्ण है जब child के लिए model call की लागत होती है।

नोड caching

किसी नोड पर cache_policy उसके परिणाम को नोड के नाम और वर्तमान state के आधार पर key करता है, साथ में एक वैकल्पिक TTL भी हो सकता है, इसलिए अपरिवर्तित input होने पर कार्य छोड़ दिया जाता है। इसके लिए node-cache feature आवश्यक है; Redis-backed store redis-cache के पीछे उपलब्ध है।

Delta checkpoints

delta feature पूरे state के बजाय super-steps के बीच का अंतर संग्रहीत करता है, जो तब महत्वपूर्ण होता है जब state बड़ा हो और steps बहुत अधिक हों।

Time travel

time-travel feature के साथ, graph.time_travel(thread_id)? किसी thread के checkpoint history पर एक handle लौटाता है: steps की सूची बनाएं, किसी एक step पर state पढ़ें, या किसी checkpoint से नया thread branch करने के लिए fork_at करें। यह call Result लौटाता है, क्योंकि प्रत्येक operation checkpoints को पढ़ता है; इसलिए बिना checkpointer वाला graph panic करने के बजाय GraphError::CheckpointError रिपोर्ट करता है।

पूरे graph के defaults

बीस nodes में एक ही retry दोहराने पर किसी node में इसे छोड़ देना आसान है।

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

प्रति-node value को हमेशा प्राथमिकता मिलती है। NodeDefaults में timeout और failure handler भी शामिल होता है।

नोड की विफलता से recovery

जब किसी नोड का retry budget समाप्त हो जाता है, तो कोई handler run समाप्त करने के बजाय हुई घटना को दर्ज कर सकता है और recovery node का नाम दे सकता है:

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 को लागू करता है, इसलिए यह इनके साथ काम करता है:

• रनर: मानक निष्पादन के लिए 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 कार्य को शेड्यूलर को सौंपता है और रिड्यूसर के माध्यम से उसे एकत्र करता है। और सबग्राफ से उसके पैरेंट में जाने वाले हॉप का अभी यहां कोई समकक्ष नहीं है।

---

पिछला: ← मल्टी-एजेंट सिस्टम | अगला: रियलटाइम एजेंट →