الوكلاء البيانيون

أنشئ مسارات عمل معقدة وحاملة للحالة باستخدام تنسيق على نمط LangGraph مع تكامل أصلي مع ADK-Rust.

نظرة عامة

يتيح لك GraphAgent تعريف مسارات العمل كرسوم بيانية موجهة تحتوي على عقد وحواف، مع دعم ما يلي:

• AgentNode: تغليف وكلاء LLM كعقد بيانية باستخدام معيّنات إدخال/إخراج مخصصة
• مسارات العمل الدورية: دعم أصلي للحلقات والاستدلال التكراري (نمط ReAct)
• التوجيه الشرطي: توجيه ديناميكي للحواف استنادًا إلى الحالة
• إدارة الحالة: حالة مكتوبة النوع مع مخفِّضات (استبدال، إلحاق، جمع، مخصص)
• نقاط التحقق: حالة مستمرة لتحمل الأعطال والتفاعل مع الإنسان ضمن الحلقة
• البث: أوضاع بث متعددة (القيم، التحديثات، الرسائل، التصحيح)

توفر حزمة 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])

تشبه الحواف إشارات المرور وعلامات الطرق التي توجه تدفق العمل.

💾 الحالة: الذاكرة المشتركة
الحالة هي مخزن أزواج المفتاح والقيمة الذي يمكن لجميع العقد القراءة منه والكتابة فيه:

• بيانات الإدخال: المعلومات الأولية التي يتم تزويد الرسم البياني بها
• النتائج الوسيطة: يصبح ناتج إحدى العقد إدخالًا لعقدة أخرى
• الناتج النهائي: النتيجة المكتملة بعد انتهاء كل عمليات المعالجة

تعمل الحالة مثل لوح أبيض مشترك يمكن للعقد ترك معلومات عليه لاستخدامها من قِبل العقد الأخرى.

⚡ التنفيذ المتوازي: تعزيز السرعة
عندما تغادر حواف متعددة عقدةً ما، تعمل العقد المستهدفة في الوقت نفسه:

• معالجة أسرع: تعمل المهام المستقلة في الوقت نفسه
• كفاءة الموارد: استخدام أفضل لوحدة المعالجة المركزية وعمليات الإدخال والإخراج
• قابلية التوسع: التعامل مع مهام سير عمل أكثر تعقيدًا دون تباطؤ خطي

يشبه ذلك وجود عدة عمال يتولون أجزاء مختلفة من مهمة في الوقت نفسه بدلًا من الانتظار في طابور.

---

البدء السريع

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"

أنشئ .env باستخدام مفتاح API الخاص بك:

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")) — لا يملك أي استدعاء يرثه. هذا هو الوضع المستقل: تحصل العقدة على 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);

ما الذي تسجله نقطة التحقق

تخزّن نقطة التحقق الحالة المتراكمة، ورقم الخطوة، والجبهة — أي العقد التي لا يزال يتعين تشغيلها. تُكتب نقطة التحقق بعد تقدّم الجبهة، ولذلك لا تؤدي الاستئناف مطلقًا إلى إعادة تنفيذ عقدة اكتمل تنفيذها، ولا إلى تطبيق تحديثاتها مرتين. ويخزّن التشغيل الذي يكتمل جبهةً فارغة، ولذلك يعيد استئناف سلسلة مكتملة الحالة النهائية بدلًا من إعادة تشغيل الرسم البياني.

تسجّل حالتان عمدًا الجبهة التي كانت قيد التنفيذ بدلًا من الجبهة التالية، لأن العقدة المتوقفة لم تُنتج تحديثاتها بعد، ويجب تشغيلها مرة أخرى عند الاستئناف:

تُنشئ عمليات التنفيذ المتدفقة نقاط تحقق وفق الجدول نفسه الذي تتبعه عمليات التنفيذ الحاجزة، بما في ذلك عند إنهاء مقاطعةٍ للتدفق، لذا يمكن استئناف الإيقاف المؤقت ضمن نمط الإنسان في الحلقة في أيٍّ من نمطي التنفيذ.

سجل نقاط التحقق (السفر عبر الزمن)

للقراءة فقط. 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);
}

الإنسان في الحلقة

أوقف التنفيذ مؤقتًا للحصول على موافقة بشرية باستخدام المقاطعات الديناميكية:

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

الصف الأخير هو الفرق الذي يستحق المعرفة: كلا المخططين متاح قبل تشغيل أي شيء، لذا فإن تعيينًا يسمّي قناة لا يعلنها أيّ من الطرفين لا يمكنه الوصول إلى تشغيل وإظهار قيمة مفقودة. ويُرفَض الرسم البياني الفرعي الذي لا يتبادل أي شيء على الإطلاق بالطريقة نفسها، لأنه لن يتمكن من التأثير في الرسم البياني الأب.

استئناف توقّف مؤقت داخل رسم بياني فرعي

لا حاجة إلى أي شيء إضافي. تؤدي إعادة استدعاء الأب في الخيط نفسه إلى إعادة الدخول إلى الرسم البياني الفرعي، الذي يعثر على نقطة التحقق الخاصة به في <parent thread>/<node name> ويتابع من حيث توقّف. لا يتكرر العمل الذي أنجزه الرسم البياني الفرعي قبل التوقّف المؤقت، كما يُستأنف التوقّف المؤقت على بُعد عدة مستويات بالطريقة نفسها — إذ تسمّي الرسالة كل مستوى مرّت عبره.

يُرفَض الرسم البياني الفرعي الذي يعلن بوابة مقاطعة لكنه لا يملك أداة لحفظ نقاط التحقق عند تجميع الأب: إذ سيعيد الدخول عند عقدته الأولى ويدفع تكلفة العمل الذي أنجزه مرة ثانية.

لأن كلا الرسمين البيانيين يحتفظان بآليات تحقق فعلية، فإن ذلك يستمر عبر إعادة تشغيل العملية: إذ تستأنف مجموعة جديدة من كائنات الرسم البياني، لا تتشارك سوى قواعد البيانات، التشغيل نفسه.

إعادة التحكم إلى الرسم البياني الأصل

يمكن لعقدة داخل رسم بياني فرعي إنهاء رسمها البياني الخاص وتسمية عقدة في الرسم البياني الذي يحتويها:

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

ينتهي الرسم البياني الفرعي ويعرض قنوات مخرجاته كالمعتاد، ثم يواصل الأصل التنفيذ عند escalate بدلاً من اتباع الحواف المعلنة لعقدة الرسم البياني الفرعي. يتحقق الأصل من الهدف، لأنه الطرف الوحيد الذي يعرف عقده الخاصة.

عناصر التحكم في الموثوقية والتكلفة

كل عنصر من هذه العناصر معطّل افتراضياً، لذا يتصرف الرسم البياني كما كان قبل ضبط أي منها.

إعادة المحاولة لكل عقدة

يؤدي الفشل المؤقت — مثل تجاوز حد المعدل أو انقطاع الاتصال — إلى إنهاء التشغيل بخلاف ذلك.

use adk_graph::retry::{RetryOn, RetryPolicy};
use std::time::Duration;

let graph = graph.with_node_retry(
    "call_model",
    RetryPolicy::new(3)
        .with_initial_delay(Duration::from_millis(500))
        .with_max_delay(Duration::from_secs(8))
        .with_backoff_factor(2.0)
        .with_retry_on(RetryOn::Any),
);

تزداد فترة التأخير بمقدار backoff_factor، وتُحدّ عند max_delay، ثم يُطبَّق عليها التذبذب. تعمل العقدة من دون سياسة مرة واحدة، لذا تظل إعادة المحاولة اختيارية. تسمح سياسة من RetryPolicy::default() بعشر محاولات، يبلغ مجموع فترات نومها التسع نحو 243 ثانية — فاخفض max_attempts عندما يكون المستدعي في انتظار الإجابة.

لا يُعاد أبداً تشغيل المقاطعة، مهما قالت retry_on: فالإيقاف المؤقت ليس فشلاً. ويُحفظ عدد المحاولات في نقطة التحقق، لذلك يواصل التشغيل المستأنف استهلاك الميزانية بدلاً من إعادة بدئها.

تحديد التزامن

يُرسل التوزيع واسع الانتشار كامل واجهته الأمامية دفعة واحدة، مما قد يستنفد مجموعة اتصالات أو يؤدي إلى تجاوز حد معدل المزوّد.

let graph = graph.with_max_concurrency(4);

تنتظر العقد التي تتجاوز الحد الأقصى الحصول على خانة. ويكون ترتيب القبول هو الواجهة الأمامية مرتبة حسب الاسم، لذا لا يعتمد على التوقيت. وتُستثنى الاستدعاءات الإجرائية للعُقد الفرعية من هذه الميزانية، لأن الأصل ينتظر عُقده الفرعية أثناء احتفاظه بخانته الخاصة.

المهلات الزمنية لكل عقدة

يحدّ TimeoutPolicy محاولةً واحدة، ويحدد مع idle_timeout المدة التي يمكن للعقدة أن تظل خلالها دون الإبلاغ عن التقدّم. يؤدي تجاوز أيٍّ منهما إلى GraphError::NodeTimedOut، التي يمكن لسياسة إعادة المحاولة التعامل معها بعد ذلك.

استدعاء عقدة مباشرةً

عندما يأتي عدد المهام الفرعية من الحالة بدلاً من شكل الرسم البياني، يمكن لعقدة أن تستدعي عقدة أخرى بنفسها:

use adk_graph::child::RunNodeOptions;

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

لا تحتاج العقدة المستهدفة إلى حافة. يُسجَّل كل عنصر فرعي مكتمل ضمن <parent>/<child>@<run_id>، ولذلك يعيد التشغيل المستأنف الإجابة المسجَّلة بدلاً من تنفيذ العنصر الفرعي مرة أخرى — وهذا مهم عندما يتطلب العنصر الفرعي استدعاء نموذج.

التخزين المؤقت للعقد

يضع cache_policy على العقدة مفتاحًا لنتيجتها باستخدام اسم العقدة والحالة الحالية، مع TTL اختياري، بحيث يتخطى الإدخال غير المتغير العمل. يتطلب ذلك الميزة node-cache؛ ويتوفر مخزن مدعوم من Redis عبر redis-cache.

نقاط تحقق الفروقات

تخزّن الميزة delta الفرق بين الخطوات الفائقة بدلاً من الحالة بأكملها، وهو أمر مهم عندما تكون الحالة كبيرة ويكون عدد الخطوات كثيرًا.

السفر عبر الزمن

مع الميزة time-travel، يعيد graph.time_travel(thread_id)? مقبضًا عبر سجل نقاط تحقق سلسلة محادثة: لسرد الخطوات، أو قراءة الحالة عند إحداها، أو fork_at نقطة تحقق لإنشاء سلسلة محادثة جديدة متفرعة منها. يعيد الاستدعاء Result لأن كل عملية تقرأ نقاط التحقق، ولذلك يُبلغ الرسم البياني الذي لا يملك أداة تحقق عن GraphError::CheckpointError بدلاً من التسبب في حالة ذعر.

الإعدادات الافتراضية على مستوى الرسم البياني

من السهل أن يؤدي تكرار إعادة المحاولة نفسها عبر عشرين عقدة إلى أخطاء بسبب السهو.

use adk_graph::graph::NodeDefaults;

let graph = graph
    .with_node_defaults(NodeDefaults::new().with_retry(RetryPolicy::new(3)))
    .with_node_retry("critical", RetryPolicy::new(10));

تتغلب قيمة خاصة بالعقدة دائمًا. تحمل NodeDefaults أيضًا مهلةً ومعالجًا للفشل.

التعافي من فشل عقدة

بعد استنفاد ميزانية إعادة المحاولة الخاصة بالعقدة، يمكن لمعالج أن يسجل ما حدث ويحدد عقدة تعافٍ بدلاً من إنهاء التشغيل:

let graph = graph.with_node_error_handler("charge", |node, error, _state| {
    Ok(NodeOutput::new()
        .with_update("status", json!(format!("{node} failed: {error}")))
        .with_goto(["compensate"]))
});

إرجاع Err ينهي التشغيل كما كان من قبل. لا يصل المقاطِع أبدًا إلى معالج، لأن الإيقاف المؤقت ليس فشلًا.

الحد من نمو نقاط التحقق

يُراكم الخيط نقطة تحقق واحدة لكل خطوة فائقة. ولذلك فإن التشغيل الذي يستمر لأيام ينمو بلا حدود، مما يستهلك مساحة تخزين ويبطئ list.

use adk_graph::checkpoint::RetentionPolicy;
use std::time::Duration;

let graph = graph
    .with_checkpoint_retention(
        RetentionPolicy::keep_last(50).with_max_age(Duration::from_secs(7 * 24 * 3600)),
    );

تحدث عملية التشذيب بعد كل عملية حفظ، لذلك تظل التكلفة متناسبة مع التشغيل ولا تكون هناك حاجة إلى مهمة خارجية. لا تُحذف أحدث نقطة تحقق أبدًا، بصرف النظر عما تقوله السياسة، لأنها النقطة التي يحمّلها الاستئناف — تُرفع قيمة keep_last(0) إلى واحد، ويحتفظ الخيط الذي تجاوزت جميع نقاط تحققه حدّ العمر بنقطة واحدة.

تكون هذه الميزة معطّلة افتراضيًا، لذلك يحتفظ الخيط الحالي بسجلّه الكامل، ولا يزال السفر عبر الزمن قادرًا على الوصول إلى كل خطوة. عيّن سياسة عندما يكون الخيط طويل العمر ولا تحتاج إلى الرجوع بعيدًا.

رفض القنوات غير المُعلنة

تستخدم القناة التي لا يُعلن عنها المخطط مُجمِّع الاستبدال، لأنه يمثل الخيار الاحتياطي للاسم المجهول. فإذا أعلن الرسم البياني عن قناة قائمة ثم كتب اسمًا قريبًا منها، فإنه يحتفظ بالقيمة الأخيرة فقط ولا يُبلغ عن أي شيء.

let graph = graph.with_strict_channels();

يفشل التشغيل بعد ذلك إذا كتبت عقدة إلى قناة غير مُعلنة، مع GraphError::UndeclaredChannel الذي يسمي العقدة والقناة. وتكون هذه الميزة معطّلة افتراضيًا، ولا يكون لها تأثير عندما لا يعلن الرسم البياني عن أي قنوات على الإطلاق.

تكامل ADK

يطبّق GraphAgent سمة ADK Agent، ولذلك يعمل مع:

• المشغّل: استخدمه مع 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 العمل إلى المجدول ويجمعه من خلال مُختزِل. أما الانتقال من رسم فرعي إلى أصله فلا يوجد له نظير هنا حتى الآن.

---

السابق: ← أنظمة متعددة الوكلاء | التالي: الوكلاء الآنيون →