Agent Coordination

Why Multi-Agent Systems?

Complex problems often require multiple specialized agents working together:

Code generation: Architect designs, Developer implements, Reviewer validates
Research: Searcher finds sources, Analyst synthesizes, Writer drafts
Customer support: Classifier routes, Specialist handles, QA verifies

MoFA provides 7 built-in coordination patterns for multi-agent collaboration.

Coordination Architecture

Coordinator Trait

All coordination patterns implement the Coordinator trait:

mofa-kernel/src/agent/components/coordinator.rs

#[async_trait]
pub trait Coordinator: Send + Sync {
    /// Dispatch task to agents
    async fn dispatch(
        &self,
        task: Task,
        ctx: &AgentContext,
    ) -> AgentResult<Vec<DispatchResult>>;
    
    /// Aggregate results from multiple agents
    async fn aggregate(
        &self,
        results: Vec<AgentOutput>,
    ) -> AgentResult<AgentOutput>;
    
    /// Get coordination pattern
    fn pattern(&self) -> CoordinationPattern;
    
    /// Select agents for task
    async fn select_agents(
        &self,
        task: &Task,
        ctx: &AgentContext,
    ) -> AgentResult<Vec<String>>;
}

Coordination Patterns

Sequential Execution: Output of one agent becomes input to the next.Use Cases:

Data pipelines (fetch → process → store)
Multi-step reasoning (plan → execute → verify)
Content generation (research → draft → edit)

Example:

use mofa_kernel::agent::components::coordinator::{
    Task, CoordinationPattern
};

// Create task
let task = Task::new(
    "task-1",
    "Research Rust async programming"
)
.with_type(TaskType::Analysis);

// Sequential pattern: research → summarize → format
let pattern = CoordinationPattern::Sequential;
let coordinator = SequentialCoordinator::new(vec![
    "researcher".to_string(),
    "summarizer".to_string(),
    "formatter".to_string(),
]);

let results = coordinator.dispatch(task, &ctx).await?;
let final_output = coordinator.aggregate(
    results.into_iter().filter_map(|r| r.output).collect()
).await?;

Parallel Execution: All agents execute simultaneously, results aggregated.Use Cases:

Multiple perspectives (get opinions from different experts)
Redundancy (run same task on multiple agents, pick best)
Independent subtasks (parallel data fetching)

Example:

let pattern = CoordinationPattern::Parallel;

// All agents analyze the same code from different angles
let coordinator = ParallelCoordinator::new(
    vec!["security_analyst", "performance_analyst", "style_checker"],
    AggregationStrategy::CollectAll,
);

let results = coordinator.dispatch(task, &ctx).await?;
// Returns: {"results": [...], "count": 3}

Hierarchical (Supervision): Supervisor delegates to workers and reviews results.Use Cases:

Quality control (workers execute, supervisor validates)
Complex task decomposition (supervisor breaks down, assigns)
Routing (supervisor decides which specialist to use)

Example:

let pattern = CoordinationPattern::Hierarchical {
    supervisor_id: "lead_developer".to_string(),
};

// Supervisor delegates code review tasks
let coordinator = HierarchicalCoordinator::new(
    "lead_developer",
    vec!["junior_dev_1", "junior_dev_2"],
);

let results = coordinator.dispatch(task, &ctx).await?;

Consensus: Agents negotiate until agreement is reached.Use Cases:

Critical decisions (medical diagnosis, financial planning)
Conflict resolution (multiple valid approaches)
Distributed consensus (Byzantine fault tolerance)

Example:

let pattern = CoordinationPattern::Consensus {
    threshold: 0.75,  // 75% agreement required
};

// Agents must reach consensus on architecture decision
let coordinator = ConsensusCoordinator::new(
    vec!["architect_1", "architect_2", "architect_3"],
    0.75,
);

let results = coordinator.dispatch(task, &ctx).await?;

Debate: Agents alternate proposing and critiquing solutions.Use Cases:

Iterative improvement (writing, code review)
Red team / blue team (security analysis)
Socratic method (teaching, learning)

Example:

let pattern = CoordinationPattern::Debate {
    max_rounds: 3,
};

// Two agents debate design approach
let coordinator = DebateCoordinator::new(
    vec!["optimist", "critic"],
    3,  // max rounds
);

let results = coordinator.dispatch(task, &ctx).await?;

MapReduce: Divide task, process in parallel, aggregate results.Use Cases:

Large document processing (split, analyze, combine)
Batch operations (process 1000s of records)
Distributed computation

Example:

let pattern = CoordinationPattern::MapReduce;

// Process large dataset in parallel
let coordinator = MapReduceCoordinator::new(
    vec!["worker_1", "worker_2", "worker_3"],
    |data| split_into_chunks(data, 3),  // Map function
    |results| merge_results(results),    // Reduce function
);

let results = coordinator.dispatch(task, &ctx).await?;

Voting: Agents independently decide, majority wins.Use Cases:

Classification (multiple models vote on label)
Content moderation (vote on whether to flag)
Ensemble methods (combine ML model predictions)

Example:

let pattern = CoordinationPattern::Voting;

// Multiple agents vote on content safety
let coordinator = VotingCoordinator::new(
    vec!["safety_1", "safety_2", "safety_3"],
    AggregationStrategy::Vote,
);

let results = coordinator.dispatch(task, &ctx).await?;

Task Definition

Tasks are the units of work passed to coordinators:

use mofa_kernel::agent::components::coordinator::{
    Task, TaskType, TaskPriority
};

let task = Task::new("task-1", "Analyze security vulnerabilities")
    .with_type(TaskType::Analysis)
    .with_priority(TaskPriority::High)
    .for_agent("security_expert")  // Optional: target specific agent
    .with_param("scope", json!("authentication"))
    .with_timeout(30_000);  // 30 seconds

Task Fields:

Field	Type	Description
`id`	String	Unique task identifier
`task_type`	TaskType	Category (Analysis, Generation, etc.)
`content`	String	Task description/prompt
`priority`	TaskPriority	Execution priority (Low, Normal, High, Urgent)
`target_agent`	Option<String>	Specific agent to use (optional)
`params`	HashMap	Additional parameters
`timeout_ms`	Option<u64>	Maximum execution time

Aggregation Strategies

How to combine results from multiple agents:

use mofa_kernel::agent::components::coordinator::AggregationStrategy;

let strategy = AggregationStrategy::Concatenate {
    separator: "\n---\n".to_string(),
};

let aggregated = aggregate_outputs(results, &strategy)?;

Concatenate
FirstSuccess
CollectAll
Vote
LLM Summarize

Join all outputs with a separator:

AggregationStrategy::Concatenate {
    separator: "\n\n".to_string(),
}

Result: "Output 1\n\nOutput 2\n\nOutput 3"

Return the first successful (non-error) output:

AggregationStrategy::FirstSuccess

Result: First agent’s successful output

Collect all outputs into a JSON array:

AggregationStrategy::CollectAll

Result:

{
  "results": ["Output 1", "Output 2", "Output 3"],
  "count": 3
}

Choose the most frequent output:

AggregationStrategy::Vote

Result: Output with most votes

Use LLM to summarize all outputs:

AggregationStrategy::LLMSummarize {
    prompt_template: "Summarize these analyses: {results}".to_string(),
}

Result: LLM-generated summary

Communication Patterns

Message Bus

Agents communicate via the microkernel message bus:

use mofa_kernel::bus::MessageBus;

// Send to specific agent
bus.send("agent-2", Message::text("Process this")).await?;

// Broadcast to all
bus.broadcast(Message::event("task_completed")).await?;

// Pub/sub topics
bus.subscribe("llm_responses").await?;
bus.publish("llm_requests", Message::json(request)).await?;

Agent Discovery

Find agents by capabilities:

use mofa_runtime::registry::AgentRegistry;

let registry = AgentRegistry::new();

// Find agents with specific capability
let llm_agents = registry.find_by_tag("llm").await?;

// Find by multiple criteria
let requirements = AgentRequirements::builder()
    .tag("code-generation")
    .supports_streaming(true)
    .build();

let agents = registry.find_matching(&requirements).await?;

Priority Scheduling

Tasks are executed based on priority:

pub enum TaskPriority {
    Low = 0,
    Normal = 1,
    High = 2,
    Urgent = 3,
}

Scheduler Behavior:

Higher priority tasks preempt lower priority
Same priority uses FIFO queue
Urgent tasks interrupt running tasks (if interruptible)

let urgent_task = Task::new("security-alert", "Analyze breach")
    .with_priority(TaskPriority::Urgent);

coordinator.dispatch(urgent_task, &ctx).await?;
// Interrupts current tasks and executes immediately

Secretary Agent Pattern

MoFA includes a specialized Secretary Agent for human-in-the-loop workflows: Features:

🧠 Autonomous task planning and decomposition
🔄 Intelligent agent scheduling
👤 Human intervention at key nodes
📊 Full process observability
🔁 Closed-loop feedback

Secretary Agent Guide

Learn how to build human-in-the-loop workflows

Real-World Example

Scenario: Code review system

use mofa_kernel::agent::components::coordinator::*;

// Sequential: Static analysis → Security scan → Style check
let analysis_task = Task::new(
    "code-review-1",
    "Review authentication.rs"
);

let sequential = SequentialCoordinator::new(vec![
    "static_analyzer",
    "security_scanner",
    "style_checker",
]);

let analysis_results = sequential.dispatch(analysis_task, &ctx).await?;

// Parallel: Multiple reviewers give feedback simultaneously
let review_task = Task::new(
    "code-review-2",
    "Get expert opinions on design"
);

let parallel = ParallelCoordinator::new(
    vec!["senior_dev_1", "senior_dev_2", "architect"],
    AggregationStrategy::CollectAll,
);

let review_results = parallel.dispatch(review_task, &ctx).await?;

// Hierarchical: Lead decides if changes needed
let decision_task = Task::new(
    "code-review-3",
    "Approve or request changes"
);

let hierarchical = HierarchicalCoordinator::new(
    "tech_lead",
    vec!["reviewer_1", "reviewer_2"],
);

let final_decision = hierarchical.dispatch(decision_task, &ctx).await?;

Performance Considerations

Coordination Overhead: Each coordination pattern adds latency:

Sequential: Sum of all agent latencies
Parallel: Max of all agent latencies + aggregation
Debate: Multiple rounds × agent latency

Design workflows to minimize unnecessary coordination.

Next Steps

Workflow Engine

Build complex stateful workflows

Examples

See coordination patterns in action

Get Started

Core Concepts

Building Agents

Plugin Development

Multi-Agent Systems

Advanced Features

Language Bindings

Agent Coordination

Why Multi-Agent Systems?

Coordination Architecture

Coordinator Trait

Coordination Patterns

Task Definition

Aggregation Strategies

Communication Patterns

Message Bus

Agent Discovery

Priority Scheduling

Secretary Agent Pattern

Secretary Agent Guide

Real-World Example

Performance Considerations

Next Steps

Workflow Engine

Examples

Build docs developers (and LLMs) love

Get Started

Core Concepts

Building Agents

Plugin Development

Multi-Agent Systems

Advanced Features

Language Bindings

Documentation Index

​Why Multi-Agent Systems?

​Coordination Architecture

​Coordinator Trait

​Coordination Patterns

​Task Definition

​Aggregation Strategies

​Communication Patterns

​Message Bus

​Agent Discovery

​Priority Scheduling

​Secretary Agent Pattern

Secretary Agent Guide

​Real-World Example

​Performance Considerations

​Next Steps

Workflow Engine

Examples

Build docs developers (and LLMs) love

Why Multi-Agent Systems?

Coordination Architecture

Coordinator Trait

Coordination Patterns

Task Definition

Aggregation Strategies

Communication Patterns

Message Bus

Agent Discovery

Priority Scheduling

Secretary Agent Pattern

Real-World Example

Performance Considerations

Next Steps