多智能体协作

EnerOS 内置完整的 Agent 运行时，提供 7 种领域 Agent 开箱即用，并通过 Orchestrator 实现编排式协作。Agent 之间通过类型化消息总线通信，所有决策受内核约束引擎约束。

「Agent-as-Grid-Node」是 EnerOS 的核心设计哲学之一：每个 Agent 在内核中被建模为电网的一个节点，拥有明确的管辖范围、权限边界与电气位置，可与其他节点进行拓扑感知的协作。

架构总览

┌─────────────────────────────────────────────────────┐
│  Orchestrator (Pipeline / DAG 编排)                 │
│  ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐      │
│  │ Fcst │→│Disp  │→│Optim │→│Maint │→│Coord │      │
│  └──────┘ └──────┘ └──────┘ └──────┘ └──────┘      │
└──────────────────┬──────────────────────────────────┘
                   │ 类型化消息总线
┌──────────────────┴──────────────────────────────────┐
│  Agent 运行时 (eneros-agent)                        │
│  ┌─────────┐ ┌─────────┐ ┌─────────┐               │
│  │ Memory  │ │ Tools   │ │ Reason  │               │
│  └─────────┘ └─────────┘ └─────────┘               │
│  ┌─────────┐ ┌─────────┐ ┌─────────┐               │
│  │ Tenant  │ │ Quota   │ │ Audit   │               │
│  └─────────┘ └─────────┘ └─────────┘               │
└──────────────────┬──────────────────────────────────┘
                   │ 系统调用
┌──────────────────┴──────────────────────────────────┐
│  Power-Native Kernel (拓扑/潮流/约束/设备)          │
└─────────────────────────────────────────────────────┘

7 种领域 Agent

Agent 优先级与抢占

Agent 优先级映射到内核调度器，高优先级 Agent 可抢占低优先级：

Agent 定义

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 {
            // 获取当前电网状态
            let state = ctx.snapshot().await?;

            // 计算经济调度
            let decision = self.economic_dispatch(&state)?;

            // 通过约束引擎校验（必经环节）
            let decision = ctx.constraints().validate_or_project(decision)?;

            // 执行决策（自动经过安全网关）
            ctx.execute(decision.clone()).await?;
            self.last_dispatch = Some(decision);

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

Agent trait 完整方法

Orchestrator 编排

Orchestrator 负责调度多个 Agent 协作完成复杂任务，支持 Pipeline（线性）、DAG（有向无环图）与 Workflow（条件分支）三种编排模式：

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

// 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 模式：复杂依赖

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

编排模式对比

通信机制

Agent 之间通过类型化消息总线通信，支持三种模式：

发布订阅

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

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

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

// DispatchAgent 订阅
let mut rx = ctx.subscribe::<LoadForecastUpdated>().await?;
while let Some(msg) = rx.recv().await {
    println!("母线 {} 预测更新，置信度 {:.0}%", msg.bus_id, msg.confidence * 100.0);
    self.re_dispatch(ctx).await?;
}

请求响应

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

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

// CoordinationAgent 发起请求
let resp: QueryReserveResponse = ctx
    .request(QueryReserveRequest { bus_id: 1 })
    .await?;

println!("可用备用: {:.1} MW", resp.mw);

共享状态

通过内核提供的 CRDT 状态存储实现最终一致性：

// 写入共享状态（CRDT 自动合并）
ctx.shared().insert("net_demand", 50.0).await?;
ctx.shared().increment("dispatch_count", 1).await?;

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

// 订阅变化
let mut rx = ctx.shared().watch::<f64>("net_demand").await?;
while let Some(new_val) = rx.recv().await {
    println!("net_demand 变更: {:.2}", new_val);
}

通信模式对比

记忆与工具

每个 Agent 拥有独立的短期记忆（窗口）与长期记忆（向量库），并可调用注册的工具集：

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

// 短期记忆：滑动窗口
let stm = ctx.memory().short_term();
stm.push(observation).await?;
let recent: Vec<_> = stm.last_n(100).await?;

// 长期记忆：向量库
let ltm = ctx.memory().long_term();
ltm.store(case_study).await?;
let similar = ltm.search(&query, 10).await?;
for case in &similar {
    println!("相似度 {:.2}: {}", case.score, case.summary);
}

// 调用工具
let result = ctx.tools().call("load_flow", args).await?;
let pf_result: PowerFlowResult = result.into();

内置工具集

Agent 生命周期

use eneros_agent::{AgentState, Health};

// 状态机
// Created → Initialized → Running → Paused → Stopped
//                ↓            ↓
//             Failed      Healthy

match agent.state() {
    AgentState::Running => { /* 正常运行 */ }
    AgentState::Paused => { /* 已暂停，可恢复 */ }
    AgentState::Failed => {
        let err = agent.last_error();
        log::error!("Agent 失败: {:?}", err);
        agent.restart().await?;
    }
    AgentState::Stopped => { /* 已停止 */ }
}

// 健康检查
let health = agent.health();
println!("健康: {:?}, 上次心跳: {:?}", health.status, health.last_heartbeat);

多 Agent 协作示例：故障自愈

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

// 故障自愈工作流
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?;

性能指标

测试环境：4 核 / 8GB / Ubuntu 22.04。

配置参数

eneros.toml 中 Agent 相关配置：

[agent]
# 单节点最大 Agent 数
max_agents = 10000
# 默认消息队列大小
message_queue_size = 1024
# 默认短期记忆窗口
short_term_window = 1000
# 长期记忆向量库维度
vector_dim = 768
# 心跳超时（秒）
heartbeat_timeout_secs = 30
# 自动重启失败 Agent
auto_restart = true
# 最大重启次数
max_restarts = 5
# Orchestrator 默认模式
default_orchestration = "pipeline"

与其他能力的关系

相关文档

• Agent 智能进阶 — LLM 集成与推理引擎
• 物理约束决策 — Agent 决策必经约束校验
• 安全守卫 — Agent 命令的安全网关
• 实时双执行域 — 实时类 Agent 运行环境
• EnerOS 简介 — Agent-as-Grid-Node 设计哲学