eneros-agent

eneros-agent 提供 Agent 运行时与内置领域 Agent，将 Agent 作为电网一等节点纳入操作系统调度。Agent 通过内核 API 直接访问拓扑、潮流、时序与设备，无需重复建模。EnerOS 内置四类领域 Agent（DispatchAgent / SelfHealingAgent / TradingAgent / PlanningAgent），同时提供 Agent trait 供用户自定义。

职责描述

eneros-agent 承担以下核心职责：

• Agent 生命周期：注册 → 启动 → 暂停 → 恢复 → 停止，状态机化管理
• 领域 Agent 实现：调度、自愈、交易、规划四类内置 Agent
• 多智能体协作：通过事件总线（eneros-eventbus）通信，Orchestrator 编排任务分解与汇总
• 工具调用：Agent 通过 eneros-tool 提供的工具引擎执行内核 API（潮流、约束、控制）
• 记忆与推理：集成 eneros-memory 短期/长期记忆与 eneros-reasoning 推理引擎（LLM）
• 电网位置感知：每个 Agent 在拓扑中拥有明确的电气位置与管辖范围
• 权限边界：基于角色与管辖范围的访问控制，越权操作经网关拒绝

关键类型与接口

Agent trait

use async_trait::async_trait;
use eneros_core::{AgentId, Result};
use eneros_topology::NetworkGraph;
use eneros_timeseries::TimeSeriesEngine;
use eneros_eventbus::EventBus;
use eneros_tool::ToolRegistry;

#[async_trait]
pub trait Agent: Send + Sync {
    fn id(&self) -> &AgentId;
    fn agent_type(&self) -> AgentType;
    fn jurisdiction(&self) -> &Jurisdiction;
    async fn run(&mut self, ctx: &mut AgentContext<'_>) -> AgentResult<()>;
    async fn on_event(&mut self, event: &Event) -> AgentResult<()>;
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AgentType {
    Dispatch,
    SelfHealing,
    Trading,
    Planning,
    Custom,
}

#[derive(Debug, Clone)]
pub struct Jurisdiction {
    pub buses: Vec<eneros_core::BusId>,
    pub branches: Vec<eneros_core::BranchId>,
    pub region: Option<String>,
    pub role: AgentRole,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AgentRole {
    Viewer,
    Operator,
    Planner,
    Administrator,
}

#[derive(Debug, Clone)]
pub enum AgentError {
    ToolFailed(String),
    ConstraintRejected(String),
    Unauthorized,
    Internal(String),
}

pub type AgentResult<T> = std::result::Result<T, AgentError>;

AgentContext

pub struct AgentContext<'a> {
    network: &'a NetworkGraph,
    timeseries: &'a TimeSeriesEngine,
    eventbus: &'a EventBus,
    tools: &'a ToolRegistry,
    memory: &'a mut eneros_memory::MemoryStore,
}

impl<'a> AgentContext<'a> {
    pub async fn get_network(&self) -> &NetworkGraph {
        self.network
    }

    pub async fn call_tool(&self, name: &str, args: &serde_json::Value) -> AgentResult<serde_json::Value> {
        self.tools.call(name, args).await.map_err(AgentError::ToolFailed)
    }

    pub async fn publish(&self, event: impl Into<Event>) {
        self.eventbus.publish(event.into()).await;
    }

    pub async fn remember(&mut self, key: &str, value: &serde_json::Value) -> AgentResult<()> {
        self.memory.store(key, value).await.map_err(|e| AgentError::Internal(e.to_string()))
    }
}

Orchestrator

use std::collections::HashMap;

pub struct Orchestrator {
    agents: HashMap<AgentId, Box<dyn Agent>>,
    schedule: Vec<AgentId>,
    tick_interval: std::time::Duration,
}

impl Default for Orchestrator {
    fn default() -> Self {
        Self {
            agents: HashMap::new(),
            schedule: Vec::new(),
            tick_interval: std::time::Duration::from_millis(100),
        }
    }
}

impl Orchestrator {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn register(&mut self, agent: Box<dyn Agent>) {
        let id = agent.id().clone();
        self.schedule.push(id.clone());
        self.agents.insert(id, agent);
    }

    pub fn unregister(&mut self, id: &AgentId) -> Option<Box<dyn Agent>> {
        self.schedule.retain(|x| x != id);
        self.agents.remove(id)
    }

    pub async fn step(&mut self, ctx: &mut AgentContext<'_>) -> AgentResult<()> {
        for id in self.schedule.clone() {
            if let Some(agent) = self.agents.get_mut(&id) {
                agent.run(ctx).await?;
            }
        }
        Ok(())
    }

    pub async fn run_forever(&mut self, mut ctx: AgentContext<'_>) -> AgentResult<()> {
        loop {
            self.step(&mut ctx).await?;
            tokio::time::sleep(self.tick_interval).await;
        }
    }

    pub fn agent_count(&self) -> usize {
        self.agents.len()
    }
}

内置领域 Agent

pub struct DispatchAgent {
    id: AgentId,
    jurisdiction: Jurisdiction,
    state: DispatchState,
}

#[derive(Debug, Clone)]
pub struct DispatchState {
    pub last_dispatch_mw: f64,
    pub last_tick: Option<chrono::DateTime<chrono::Utc>>,
}

impl DispatchAgent {
    pub fn new(id: impl Into<String>) -> Self {
        Self {
            id: AgentId::new(id),
            jurisdiction: Jurisdiction {
                buses: Vec::new(),
                branches: Vec::new(),
                region: None,
                role: AgentRole::Operator,
            },
            state: DispatchState {
                last_dispatch_mw: 0.0,
                last_tick: None,
            },
        }
    }

    pub fn with_jurisdiction(mut self, j: Jurisdiction) -> Self {
        self.jurisdiction = j;
        self
    }
}

#[async_trait]
impl Agent for DispatchAgent {
    fn id(&self) -> &AgentId {
        &self.id
    }

    fn agent_type(&self) -> AgentType {
        AgentType::Dispatch
    }

    fn jurisdiction(&self) -> &Jurisdiction {
        &self.jurisdiction
    }

    async fn run(&mut self, ctx: &mut AgentContext<'_>) -> AgentResult<()> {
        let net = ctx.get_network().await;
        let pf_result = ctx.call_tool("powerflow", &serde_json::json!({"network_id": "default"})).await?;
        let total_mw: f64 = pf_result["total_generation_mw"].as_f64().unwrap_or(0.0);
        self.state.last_dispatch_mw = total_mw;
        self.state.last_tick = Some(chrono::Utc::now());
        Ok(())
    }

    async fn on_event(&mut self, _event: &Event) -> AgentResult<()> {
        Ok(())
    }
}

pub struct SelfHealingAgent {
    id: AgentId,
    jurisdiction: Jurisdiction,
    isolation_strategy: IsolationStrategy,
}

#[derive(Debug, Clone, Copy)]
pub enum IsolationStrategy {
    FastIsolation,
    ConservativeIsolation,
}

impl SelfHealingAgent {
    pub fn new(id: impl Into<String>) -> Self {
        Self {
            id: AgentId::new(id),
            jurisdiction: Jurisdiction {
                buses: Vec::new(),
                branches: Vec::new(),
                region: None,
                role: AgentRole::Operator,
            },
            isolation_strategy: IsolationStrategy::FastIsolation,
        }
    }
}

核心 API

使用示例

注册并运行多 Agent

use eneros_agent::{Orchestrator, DispatchAgent, SelfHealingAgent, AgentContext};
use std::time::Duration;

let mut orch = Orchestrator::new();
orch.tick_interval = Duration::from_millis(100);

orch.register(Box::new(DispatchAgent::new("dispatch_01")));
orch.register(Box::new(SelfHealingAgent::new("selfheal_01")));

println!("已注册 Agent 数: {}", orch.agent_count());

let ctx = AgentContext::new(&network, &timeseries, &eventbus, &tools, &mut memory);
orch.run_forever(ctx).await?;

单步执行

let mut ctx = AgentContext::new(&network, &timeseries, &eventbus, &tools, &mut memory);

for step in 0..10 {
    println!("=== Step {} ===", step);
    orch.step(&mut ctx).await?;
}

自定义 Agent

use async_trait::async_trait;
use eneros_agent::{Agent, AgentType, AgentResult, AgentContext, Jurisdiction, AgentRole, AgentError};
use eneros_core::AgentId;

struct VoltageMonitorAgent {
    id: AgentId,
    jurisdiction: Jurisdiction,
}

impl VoltageMonitorAgent {
    fn new(id: impl Into<String>) -> Self {
        Self {
            id: AgentId::new(id),
            jurisdiction: Jurisdiction {
                buses: Vec::new(),
                branches: Vec::new(),
                region: None,
                role: AgentRole::Viewer,
            },
        }
    }
}

#[async_trait]
impl Agent for VoltageMonitorAgent {
    fn id(&self) -> &AgentId {
        &self.id
    }

    fn agent_type(&self) -> AgentType {
        AgentType::Custom
    }

    fn jurisdiction(&self) -> &Jurisdiction {
        &self.jurisdiction
    }

    async fn run(&mut self, ctx: &mut AgentContext<'_>) -> AgentResult<()> {
        let net = ctx.get_network().await;
        let violations = ctx.call_tool("check_constraints", &serde_json::json!({})).await?;

        if violations["count"].as_u64().unwrap_or(0) > 0 {
            ctx.publish(AlarmEvent::critical(violations.clone())).await;
            ctx.remember("last_violation", &violations).await?;
        }
        Ok(())
    }

    async fn on_event(&mut self, event: &Event) -> AgentResult<()> {
        match event {
            Event::TopologyChanged(_) => {
                tracing::info!("拓扑变更，重新校验");
            }
            _ => {}
        }
        Ok(())
    }
}

let mut orch = Orchestrator::new();
orch.register(Box::new(VoltageMonitorAgent::new("vm_01")));

带管辖范围与权限的 Agent

use eneros_agent::{DispatchAgent, Jurisdiction, AgentRole};
use eneros_core::ElementId;

let jurisdiction = Jurisdiction {
    buses: vec![ElementId::new("bus_1"), ElementId::new("bus_2")],
    branches: vec![ElementId::new("branch_1_2")],
    region: Some("east".into()),
    role: AgentRole::Operator,
};

let agent = DispatchAgent::new("dispatch_east")
    .with_jurisdiction(jurisdiction);

orch.register(Box::new(agent));

性能指标

测试环境：4 核 / 8GB / Ubuntu 22.04 / Rust 1.78 release build。

配置参数表

[agent] 配置段

AgentRole 权限矩阵

依赖关系

自身依赖

[dependencies]
eneros-core = { path = "../core", version = "0.47.0" }
eneros-topology = { path = "../topology", version = "0.47.0" }
eneros-powerflow = { path = "../powerflow", version = "0.47.0" }
eneros-constraint = { path = "../constraint", version = "0.47.0" }
eneros-memory = { path = "../memory", version = "0.47.0" }
eneros-tool = { path = "../tool", version = "0.47.0" }
eneros-reasoning = { path = "../reasoning", version = "0.47.0" }
eneros-eventbus = { path = "../eventbus", version = "0.47.0" }
async-trait = "0.1"
tokio = { version = "1", features = ["full"] }
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
tracing = "0.1"
chrono = { version = "0.4", features = ["serde"] }

被依赖

• eneros-gateway — 网关代理 Agent 指令下发
• eneros-dashboard — 仪表盘展示 Agent 状态
• eneros-edge — 边缘节点运行轻量 Agent
• eneros-aiops — AIOps 集成 Agent 决策

版本与兼容性

相关文档

• Agent-as-Grid-Node — 设计理念
• 多智能体协作 — 协作机制
• 首个 Agent — 快速上手
• eneros-tool — Agent 工具引擎
• eneros-reasoning — LLM 推理集成
• Crate 索引 — 完整列表