Multi-Agent Collaboration

EnerOS has a built-in complete Agent runtime, providing 7 domain Agents out of the box and enabling orchestrated collaboration through the Orchestrator. Agents communicate via a typed message bus, and all decisions are subject to the kernel constraint engine.

“Agent-as-Grid-Node” is one of EnerOS’s core design philosophies: each Agent is modeled as a node in the grid within the kernel, with a clear jurisdiction, permission boundary, and electrical position, enabling topology-aware collaboration with other nodes.

Architecture Overview

┌─────────────────────────────────────────────────────┐
│  Orchestrator (Pipeline / DAG orchestration)         │
│  ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐      │
│  │ Fcst │→│Disp  │→│Optim │→│Maint │→│Coord │      │
│  └──────┘ └──────┘ └──────┘ └──────┘ └──────┘      │
└──────────────────┬──────────────────────────────────┘
                   │ Typed Message Bus
┌──────────────────┴──────────────────────────────────┐
│  Agent Runtime (eneros-agent)                        │
│  ┌─────────┐ ┌─────────┐ ┌─────────┐               │
│  │ Memory  │ │ Tools   │ │ Reason  │               │
│  └─────────┘ └─────────┘ └─────────┘               │
│  ┌─────────┐ ┌─────────┐ ┌─────────┐               │
│  │ Tenant  │ │ Quota   │ │ Audit   │               │
│  └─────────┘ └─────────┘ └─────────┘               │
└──────────────────┬──────────────────────────────────┘
                   │ System Calls
┌──────────────────┴──────────────────────────────────┐
│  Power-Native Kernel (Topology/Load Flow/Constraint/Equipment) │
└─────────────────────────────────────────────────────┘

7 Domain Agents

Agent Priority and Preemption

Agent priorities are mapped to the kernel scheduler. High-priority Agents can preempt low-priority ones:

Agent Definition

use eneros_agent::{Agent, AgentContext, AgentResult};
use std::time::Duration;

struct DispatchAgent {
    interval: Duration,
    last_dispatch: Option<DispatchDecision>,
}

impl DispatchAgent {
    fn new() -> Self {
        Self {
            interval: Duration::from_secs(300),
            last_dispatch: None,
        }
    }

    fn economic_dispatch(&self, state: &GridState) -> AgentResult<DispatchDecision> {
        let mut decision = DispatchDecision::new();
        for gen in &state.generators {
            let target = self.dispatch_gen(gen, state)?;
            decision.set_generator(gen.id, target);
        }
        Ok(decision)
    }

    fn dispatch_gen(&self, gen: &Generator, state: &GridState) -> AgentResult<f64> {
        let cost = gen.cost_curve;
        let load = state.total_load;
        let share = gen.capacity / state.total_capacity;
        Ok(load * share * (1.0 - cost.alpha))
    }
}

impl Agent for DispatchAgent {
    fn name(&self) -> &str { "dispatch" }

    fn priority(&self) -> i32 { 50 }

    async fn run(&mut self, ctx: &mut AgentContext) -> AgentResult<()> {
        loop {
            // Get the current grid state
            let state = ctx.snapshot().await?;

            // Compute economic dispatch
            let decision = self.economic_dispatch(&state)?;

            // Validate through the constraint engine (mandatory)
            let decision = ctx.constraints().validate_or_project(decision)?;

            // Execute the decision (automatically passes through the safety gateway)
            ctx.execute(decision.clone()).await?;
            self.last_dispatch = Some(decision);

            ctx.sleep(self.interval).await;
        }
    }
}

Agent Trait Complete Methods

Orchestrator Orchestration

The Orchestrator schedules multiple Agents to collaborate on complex tasks, supporting three orchestration modes: Pipeline (linear), DAG (directed acyclic graph), and Workflow (conditional branching):

use eneros_agent::orchestrator::{Orchestrator, Pipeline, Stage, DagBuilder};

// Pipeline mode: linear pipeline
let pipeline = Pipeline::new()
    .stage(Stage::parallel()
        .agent(ForecastAgent::load())
        .agent(ForecastAgent::renewable()))
    .stage(Stage::single(DispatchAgent::new()))
    .stage(Stage::single(OptimalAgent::new()))
    .stage(Stage::parallel()
        .agent(MaintenanceAgent::new())
        .agent(CoordinationAgent::new()));

let mut orch = Orchestrator::new(pipeline);
orch.run(&mut ctx).await?;

DAG Mode: Complex Dependencies

use eneros_agent::orchestrator::{DagBuilder, NodeId};

let dag = DagBuilder::new()
    .node("forecast_load", ForecastAgent::load())
    .node("forecast_pv", ForecastAgent::renewable())
    .node("dispatch", DispatchAgent::new())
    .node("opf", OptimalAgent::new())
    .node("maint", MaintenanceAgent::new())
    .edge("forecast_load", "dispatch")
    .edge("forecast_pv", "dispatch")
    .edge("dispatch", "opf")
    .edge("opf", "maint")
    .build();

let mut orch = Orchestrator::new(dag);
orch.run(&mut ctx).await?;

Orchestration Mode Comparison

Communication Mechanism

Agents communicate via a typed message bus, supporting three modes:

Publish-Subscribe

use eneros_agent::Message;
use chrono::{DateTime, Utc};

#[derive(Message, Clone)]
struct LoadForecastUpdated {
    bus_id: u32,
    forecast: Vec<(DateTime<Utc>, f64)>,
    confidence: f64,
}

// ForecastAgent publishes
ctx.publish(LoadForecastUpdated {
    bus_id: 1,
    forecast: vec![(t1, 50.0), (t2, 55.0)],
    confidence: 0.95,
}).await?;

// DispatchAgent subscribes
let mut rx = ctx.subscribe::<LoadForecastUpdated>().await?;
while let Some(msg) = rx.recv().await {
    println!("Bus {} forecast updated, confidence {:.0}%", msg.bus_id, msg.confidence * 100.0);
    self.re_dispatch(ctx).await?;
}

Request-Response

#[derive(Message)]
struct QueryReserveRequest { bus_id: u32 }

#[derive(Message)]
struct QueryReserveResponse { mw: f64, available: bool }

// CoordinationAgent initiates the request
let resp: QueryReserveResponse = ctx
    .request(QueryReserveRequest { bus_id: 1 })
    .await?;

println!("Available reserve: {:.1} MW", resp.mw);

Shared State

Achieves eventual consistency through the CRDT state store provided by the kernel:

// Write to shared state (CRDT auto-merges)
ctx.shared().insert("net_demand", 50.0).await?;
ctx.shared().increment("dispatch_count", 1).await?;

// Read
let demand: f64 = ctx.shared().get("net_demand").await?;
let count: u64 = ctx.shared().get("dispatch_count").await?;

// Subscribe to changes
let mut rx = ctx.shared().watch::<f64>("net_demand").await?;
while let Some(new_val) = rx.recv().await {
    println!("net_demand changed: {:.2}", new_val);
}

Communication Mode Comparison

Memory and Tools

Each Agent has independent short-term memory (windowed) and long-term memory (vector store), and can call the registered tool set:

use eneros_agent::memory::{ShortTermMemory, LongTermMemory};

// Short-term memory: sliding window
let stm = ctx.memory().short_term();
stm.push(observation).await?;
let recent: Vec<_> = stm.last_n(100).await?;

// Long-term memory: vector store
let ltm = ctx.memory().long_term();
ltm.store(case_study).await?;
let similar = ltm.search(&query, 10).await?;
for case in &similar {
    println!("Similarity {:.2}: {}", case.score, case.summary);
}

// Call a tool
let result = ctx.tools().call("load_flow", args).await?;
let pf_result: PowerFlowResult = result.into();

Built-in Tool Set

Agent Lifecycle

use eneros_agent::{AgentState, Health};

// State machine
// Created → Initialized → Running → Paused → Stopped
//                ↓            ↓
//             Failed      Healthy

match agent.state() {
    AgentState::Running => { /* Running normally */ }
    AgentState::Paused => { /* Paused, can be resumed */ }
    AgentState::Failed => {
        let err = agent.last_error();
        log::error!("Agent failed: {:?}", err);
        agent.restart().await?;
    }
    AgentState::Stopped => { /* Stopped */ }
}

// Health check
let health = agent.health();
println!("Health: {:?}, last heartbeat: {:?}", health.status, health.last_heartbeat);

Multi-Agent Collaboration Example: Fault Self-Healing

use eneros_agent::orchestrator::{Workflow, Branch, Condition};

// Fault self-healing workflow
let fdir = Workflow::new()
    .step("detect", EmergencyAgent::new())
    .step("isolate", Step::with_action(|ctx| {
        ctx.tools().call("isolate_fault", fault_args).await
    }))
    .branch(
        Condition::on("isolate.success"),
        Branch::then("restore", EmergencyAgent::restoration())
            .then("verify", OptimalAgent::new()),
    )
    .branch(
        Condition::on("isolate.failed"),
        Branch::then("escalate", CoordinationAgent::new())
            .then("manual", MaintenanceAgent::new()),
    );

let mut orch = Orchestrator::new(fdir);
orch.run(&mut ctx).await?;

Performance Metrics

Test environment: 4 cores / 8GB / Ubuntu 22.04.

Configuration Parameters

Agent-related configuration in eneros.toml:

[agent]
# Maximum number of Agents per node
max_agents = 10000
# Default message queue size
message_queue_size = 1024
# Default short-term memory window
short_term_window = 1000
# Long-term memory vector store dimension
vector_dim = 768
# Heartbeat timeout (seconds)
heartbeat_timeout_secs = 30
# Automatically restart failed Agents
auto_restart = true
# Maximum restart attempts
max_restarts = 5
# Default Orchestrator mode
default_orchestration = "pipeline"

Relationship with Other Capabilities

Related Documentation

• Agent Intelligence Advanced — LLM integration and reasoning engine
• Physical Constraint Decision — Agent decisions must pass constraint validation
• Safety Guard — Safety gateway for Agent commands
• Real-Time Dual Execution Domain — Runtime environment for real-time Agents
• EnerOS Introduction — Agent-as-Grid-Node design philosophy