Multi-Agent Coordination

Overview

This example demonstrates a complete multi-agent system built on MoFA’s microkernel architecture. It showcases:

Master-worker coordination pattern
LLM-powered intelligent task assignment
Worker agents with different specialties (Analyst, Coder, Writer)
Runtime agent registration and lifecycle management
Message bus communication between agents
Real-time workload balancing

What You’ll Learn

Implementing MoFAAgent trait for custom agents
Using SimpleRuntime for agent lifecycle management
LLM-based decision making for task routing
Agent capability matching and workload optimization
Message bus pattern for inter-agent communication

Prerequisites

Rust 1.75 or higher
OpenAI API key
Understanding of async patterns and message passing

Architecture

Source Code

use mofa_sdk::kernel::{AgentConfig, MoFAAgent, AgentState};
use mofa_sdk::llm::LLMClient;
use std::sync::Arc;

pub struct MasterAgent {
    id: String,
    name: String,
    state: AgentState,
    llm_client: Arc<LLMClient>,
    worker_states: Arc<RwLock<HashMap<String, WorkerState>>>,
}

impl MasterAgent {
    pub fn new(config: AgentConfig, llm_client: Arc<LLMClient>) -> Self {
        // Implementation...
    }

    /// Register a worker with specialty
    pub async fn register_worker(
        &self, 
        worker_id: String, 
        specialty: WorkerSpecialty
    ) {
        let state = WorkerState {
            specialty,
            load: 0,
            tasks_completed: 0,
            tasks_failed: 0,
        };
        self.worker_states.write().await.insert(worker_id, state);
    }

    /// Use LLM to intelligently assign tasks
    async fn assign_task_with_llm(
        &self, 
        task: &TaskRequest
    ) -> Result<String, Box<dyn std::error::Error>> {
        let workers_map = self.worker_states.read().await;
        
        // Build worker description
        let worker_info: Vec<String> = workers_map
            .iter()
            .map(|(id, state)| {
                format!(
                    "- {}: {} (load: {}, completed: {})",
                    id, state.specialty, state.load, state.tasks_completed
                )
            })
            .collect();

        // Ask LLM to select best worker
        let prompt = format!(
            "Available workers:\n{}\n\n\
             Task: {}\nPriority: {:?}\n\n\
             Select the best worker based on specialty and load.",
            worker_info.join("\n"), task.content, task.priority
        );

        let response = self.llm_client
            .chat()
            .system("You are a task dispatcher")
            .user(&prompt)
            .temperature(0.3)
            .send()
            .await?;

        let worker_id = self.parse_worker_id(
            response.content().unwrap_or(""),
            &workers_map
        ).await?;

        Ok(worker_id)
    }
}

Running the Example

Set API Key

export OPENAI_API_KEY="your-api-key"

Run All Scenarios

cd examples/multi_agent_coordination
cargo run

Run Specific Scenario

# Code review
cargo run -- --scenario=code-review

# Documentation generation
cargo run -- --scenario=doc-generation

# Problem diagnosis
cargo run -- --scenario=diagnosis

Demo Scenarios

Code Review
Documentation
Diagnosis

Scenario: Analyze code for security and performance issues

let code_snippet = r#"
fn process_data(input: &str) -> String {
    let mut result = String::new();
    for ch in input.chars() {
        if ch.is_ascii() {
            result.push(ch.to_ascii_uppercase());
        }
    }
    result
}
"#;

let task = TaskRequest {
    task_id: "review_001".to_string(),
    content: format!("Analyze this code: {}", code_snippet),
    priority: TaskPriority::High,
};

Master: Assigns to Analyst worker based on capabilitiesAnalyst: Reviews code and provides detailed feedback

Scenario: Generate API documentation

let api_definition = r#"
API: User Management
Endpoints:
- POST /users - Create new user
- GET /users/:id - Get user by ID
- PUT /users/:id - Update user
- DELETE /users/:id - Delete user
"#;

let task = TaskRequest {
    task_id: "doc_001".to_string(),
    content: format!("Create docs for: {}", api_definition),
    priority: TaskPriority::High,
};

Master: Routes to Writer workerWriter: Generates comprehensive documentation

Scenario: Diagnose error logs

let error_log = r#"
[ERROR] 2024-01-15 10:23:45
Thread: main
Panic: called `Result::unwrap()` on an `Err` value
"#;

let task = TaskRequest {
    task_id: "diag_001".to_string(),
    content: format!("Diagnose: {}", error_log),
    priority: TaskPriority::Critical,
};

Master: Assigns to Analyst based on priorityAnalyst: Provides root cause analysis

Expected Output

======================================================================
 MoFA Multi-Agent Coordination - Microkernel Architecture
======================================================================

LLM Provider initialized
SimpleRuntime initialized
Master Agent registered to runtime
Worker Pool initialized with 3 workers

======================================================================
Scenario 1: Code Review Collaboration
======================================================================

Master: Assigned task 'review_001' to worker 'worker_001'
Worker Agent (worker_001): Processing task 'review_001'
Worker Agent (worker_001): Completed task 'review_001' in 2341ms

Final Statistics:
worker_001: 1 tasks completed, 0 failed, load 0
worker_002: 0 tasks completed, 0 failed, load 0
worker_003: 0 tasks completed, 0 failed, load 0

======================================================================
 Demo completed successfully!
======================================================================

Key Concepts

Master-Worker Pattern

The Master Agent:

Receives task requests
Analyzes worker capabilities and load
Uses LLM to make intelligent routing decisions
Tracks task completion and worker statistics

Worker Agents:

Specialize in specific domains
Process assigned tasks using LLM
Report status back to master
Maintain performance metrics

LLM-Based Task Routing

// Master uses LLM to decide worker assignment
let prompt = format!(
    "Available workers:\n{}\n\nTask: {}\n\n\
     Select the best worker based on:\
     1. Specialty match\n\
     2. Current workload\n\
     3. Past performance",
    worker_list, task.content
);

let response = llm_client.chat()
    .system("You are an intelligent task dispatcher")
    .user(&prompt)
    .send()
    .await?;

Runtime Management

// Register agents to runtime
let runtime = SimpleRuntime::new();

runtime.register_agent(
    metadata,
    config,
    agent_type
).await?;

// Broadcast events
runtime.broadcast(AgentEvent::Custom(
    "task_request",
    task_data
)).await?;

// Cleanup
runtime.stop_all().await?;

Advanced Features

Dynamic Worker Registration

Workers can be added/removed at runtime:

// Add new worker
master.register_worker(
    "worker_004".to_string(),
    WorkerSpecialty::Analyst
).await;

// Remove worker
master.unregister_worker("worker_001").await;

Load Balancing

Master tracks and balances workload:

let worker_id = workers
    .iter()
    .min_by_key(|(_, state)| state.load)
    .map(|(id, _)| id.clone())
    .unwrap();

Priority Queuing

Tasks are prioritized:

pub enum TaskPriority {
    Low,
    Medium,
    High,
    Critical,
}

Performance Tracking

Each worker maintains metrics:

pub struct WorkerState {
    pub specialty: WorkerSpecialty,
    pub load: usize,
    pub tasks_completed: usize,
    pub tasks_failed: usize,
}

Common Use Cases

Code Analysis

Distribute code review across specialized agents

Content Pipeline

Research → Writing → Editing workflow

Customer Support

Route tickets to specialized support agents

Data Processing

Parallel data transformation and analysis

Troubleshooting

Worker Overload

Problem: All tasks go to one workerSolution: Improve LLM prompt or add load balancing:

if state.load > MAX_LOAD {
    continue; // Skip overloaded workers
}

Task Timeout

Problem: Tasks take too longSolution: Add timeout and retry logic:

tokio::time::timeout(
    Duration::from_secs(30),
    worker.process_task(task)
).await??;

Communication Failure

Problem: Messages not reaching agentsSolution: Check runtime registration:

let agents = runtime.list_agents().await;
println!("Registered: {:?}", agents);

Next Steps

Secretary Agent

Add human-in-the-loop workflows

Workflow Orchestration

Build complex multi-stage workflows

Coordination Guide

Learn all coordination patterns

Runtime API

Runtime system reference

Basic Examples

Advanced Examples

Domain-Specific

Multi-Agent Coordination

Overview

What You’ll Learn

Prerequisites

Architecture

Source Code

Running the Example

Demo Scenarios

Expected Output

Key Concepts

Master-Worker Pattern

LLM-Based Task Routing

Runtime Management

Advanced Features

Common Use Cases

Code Analysis

Content Pipeline

Customer Support

Data Processing

Troubleshooting

Next Steps

Secretary Agent

Workflow Orchestration

Coordination Guide

Runtime API

Build docs developers (and LLMs) love

Basic Examples

Advanced Examples

Domain-Specific

​Overview

​What You’ll Learn

​Prerequisites

​Architecture

​Source Code

​Running the Example

​Demo Scenarios

​Expected Output

​Key Concepts

​Master-Worker Pattern

​LLM-Based Task Routing

​Runtime Management

​Advanced Features

​Common Use Cases

Code Analysis

Content Pipeline

Customer Support

Data Processing

​Troubleshooting

​Next Steps

Secretary Agent

Workflow Orchestration

Coordination Guide

Runtime API

Build docs developers (and LLMs) love

Overview

What You’ll Learn

Prerequisites

Architecture

Source Code

Running the Example

Demo Scenarios

Expected Output

Key Concepts

Master-Worker Pattern

LLM-Based Task Routing

Runtime Management

Advanced Features

Common Use Cases

Troubleshooting

Next Steps