调度 Agent 开发实战

本教程演示如何从零开发一个经济调度（Economic Dispatch, ED）Agent，使其在满足机组上下限、旋转备用与网络约束的前提下，最小化系统总发电成本。本 Agent 还将集成短期负荷预测、AGC（自动发电控制）与安全约束校验，形成一个可在调度主站实际运行的智能体。

经济调度问题概述

经济调度是在已知负荷需求与机组参数的条件下，确定各台发电机的有功出力，使总发电成本最小。每台发电机的成本曲线通常用二次函数近似：

F_i(P_i) = a_i · P_i² + b_i · P_i + c_i

约束条件：

• 功率平衡：Σ P_i = P_load + P_loss
• 机组上下限：P_min,i ≤ P_i ≤ P_max,i
• 旋转备用：Σ (P_max,i - P_i) ≥ Reserve_required
• 网络约束：支路潮流不越限（通过潮流计算校验）

经典解法是 lambda 迭代法（等微增率原则）：当所有在线机组的微增率 λ_i = 2·a_i·P_i + b_i 相等时，总成本最小。

准备工作

完成 首个 Agent 教程，了解 Agent 生命周期与消息总线基础。本教程涉及以下 crate：

新建二进制 crate 并添加依赖：

[package]
name = "dispatch-agent-tutorial"
version = "0.1.0"
edition = "2021"

[dependencies]
eneros-agent = { path = "../eneros/crates/eneros-agent" }
eneros-powerflow = { path = "../eneros/crates/eneros-powerflow" }
eneros-core = { path = "../eneros/crates/eneros-core" }
eneros-eventbus = { path = "../eneros/crates/eneros-eventbus" }
eneros-tool = { path = "../eneros/crates/eneros-tool" }
eneros-reasoning = { path = "../eneros/crates/eneros-reasoning" }
eneros-network = { path = "../eneros/crates/eneros-network" }
eneros-memory = { path = "../eneros/crates/eneros-memory" }
eneros-gateway = { path = "../eneros/crates/eneros-gateway" }
async-trait = "0.1"
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"

步骤 1：定义机组成本曲线

eneros_agent::GeneratorCostCurve 已封装了二次成本模型与微增率计算：

use eneros_agent::GeneratorCostCurve;

let generators = vec![
    GeneratorCostCurve {
        gen_id: "G1".into(),       // Slack 母线上的机组
        a: 0.004,                   // $/MW²
        b: 12.0,                    // $/MW
        c: 240.0,                   // $
        p_min_mw: 50.0,
        p_max_mw: 200.0,
    },
    GeneratorCostCurve {
        gen_id: "G2".into(),
        a: 0.0035,
        b: 13.5,
        c: 200.0,
        p_min_mw: 20.0,
        p_max_mw: 80.0,
    },
    GeneratorCostCurve {
        gen_id: "G3".into(),
        a: 0.005,
        b: 11.0,
        c: 180.0,
        p_min_mw: 30.0,
        p_max_mw: 100.0,
    },
    GeneratorCostCurve {
        gen_id: "G6".into(),
        a: 0.006,
        b: 14.0,
        c: 120.0,
        p_min_mw: 10.0,
        p_max_mw: 50.0,
    },
    GeneratorCostCurve {
        gen_id: "G8".into(),
        a: 0.0045,
        b: 13.0,
        c: 100.0,
        p_min_mw: 10.0,
        p_max_mw: 35.0,
    },
];

// 验证成本曲线计算
let g1 = &generators[0];
println!("G1 @ 150MW: 成本 = ${:.2}", g1.cost_at(150.0));
println!("G1 @ 150MW: 微增率 = ${:.4}/MWh", g1.incremental_cost(150.0));

输出示例：

G1 @ 150MW: 成本 = $3150.00
G1 @ 150MW: 微增率 = $13.2000/MWh

步骤 2：实现负荷预测工具

调度 Agent 需要在每个调度周期开始时预测未来 5-15 分钟的负荷。eneros-agent 已提供 Holt-Winters 与指数平滑两种算法：

use eneros_agent::{LoadForecastAgent, HoltWintersParams, SmoothingMethod};

/// 创建一个短期负荷预测 Agent
fn build_forecast_agent() -> LoadForecastAgent {
    LoadForecastAgent::new("forecast-001", "ShortTermForecast")
        .with_method(SmoothingMethod::HoltWinters(HoltWintersParams {
            alpha: 0.4,         // 水平滑子系数
            beta: 0.1,          // 趋势平滑系数
            gamma: 0.2,         // 季节性平滑系数
            season_length: 96,  // 日内 96 点（15 分钟采样）
        }))
        .with_history_window(7 * 96)  // 7 天历史
}

#[tokio::test]
async fn test_forecast() {
    let mut agent = build_forecast_agent();
    // 喂入历史负荷数据（模拟 7 天 96 点/日）
    let base_load = 259.0;  // IEEE 14 总负荷 MW
    for t in 0..(7 * 96) {
        let hour = (t / 4) % 24;
        let daily_pattern = match hour {
            0..=5 => 0.7,
            6..=8 => 0.9,
            9..=11 => 1.0,
            12..=13 => 1.05,
            14..=17 => 1.0,
            18..=21 => 1.15,  // 晚高峰
            _ => 0.85,
        };
        let noise = 0.02 * ((t as f64).sin() * 5.0);
        let load = base_load * daily_pattern + noise;
        agent.ingest_sample(load).await.unwrap();
    }

    let forecast = agent.forecast_steps(4).await.unwrap();  // 预测未来 4 步（1 小时）
    println!("未来 1 小时负荷预测:");
    for (i, v) in forecast.iter().enumerate() {
        println!("  t+{}: {:.2} MW", (i + 1) * 15, v);
    }
}

输出示例：

未来 1 小时负荷预测:
  t+15: 271.34 MW
  t+30: 278.91 MW
  t+45: 283.12 MW
  t+60: 279.45 MW

步骤 3：实现经济调度算法

eneros_agent::economic_dispatch 已实现 lambda 迭代法。下面演示如何调用并自定义扩展：

use eneros_agent::{economic_dispatch, EconomicDispatchResult, GeneratorCostCurve};

/// 执行经济调度，含旋转备用校验
fn dispatch_with_reserve(
    generators: &[GeneratorCostCurve],
    load_mw: f64,
    reserve_requirement_mw: f64,
) -> Result<EconomicDispatchResult, String> {
    // 1. 检查总容量是否满足负荷 + 备用
    let total_capacity: f64 = generators.iter().map(|g| g.p_max_mw).sum();
    let total_min: f64 = generators.iter().map(|g| g.p_min_mw).sum();

    if total_capacity < load_mw + reserve_requirement_mw {
        return Err(format!(
            "容量不足: 可用 {} MW, 需求 {} MW (load={} + reserve={})",
            total_capacity, load_mw + reserve_requirement_mw,
            load_mw, reserve_requirement_mw
        ));
    }

    if total_min > load_mw {
        return Err(format!(
            "最小出力 {} MW 超过负荷 {} MW", total_min, load_mw
        ));
    }

    // 2. 调用内置经济调度
    let result = economic_dispatch(generators, load_mw);

    // 3. 旋转备用校验
    let total_reserve: f64 = result.gen_outputs.iter()
        .zip(generators.iter())
        .map(|((_, p), g)| g.p_max_mw - p)
        .sum();

    if total_reserve < reserve_requirement_mw {
        return Err(format!(
            "备用不足: 可用 {} MW, 需要 {} MW", total_reserve, reserve_requirement_mw
        ));
    }

    Ok(result)
}

// 测试
let result = dispatch_with_reserve(&generators, 259.0, 30.0).unwrap();
println!("=== 经济调度结果 ===");
println!("总负荷: {:.2} MW", result.total_load_mw);
println!("总发电: {:.2} MW", result.total_generation_mw);
println!("总成本: ${:.2}/h", result.total_cost);
println!("机组出力:");
for (gen_id, p) in &result.gen_outputs {
    println!("  {}: {:.2} MW", gen_id, p);
}

输出示例：

=== 经济调度结果 ===
总负荷: 259.00 MW
总发电: 259.00 MW
总成本: $3287.45/h
机组出力:
  G1: 146.23 MW
  G2: 70.18 MW
  G3: 28.59 MW
  G6: 10.00 MW
  G8: 10.00 MW

自定义 lambda 迭代法（含网损补偿）

如果需要在调度算法中纳入网损补偿（B 系数法），可自行实现：

/// 含网损补偿的经济调度（B 系数法）
fn economic_dispatch_with_losses(
    generators: &[GeneratorCostCurve],
    load_mw: f64,
    b_coefficients: &[Vec<f64>],  // 网损系数矩阵 B
    tolerance: f64,
    max_iter: usize,
) -> Result<EconomicDispatchResult, String> {
    let n = generators.len();
    let mut lambda = 13.0;
    let mut p: Vec<f64> = generators.iter()
        .map(|g| (g.p_min_mw + g.p_max_mw) / 2.0)
        .collect();

    for iter in 0..max_iter {
        // 1. 计算网损 P_loss = P^T · B · P
        let mut p_loss = 0.0;
        for i in 0..n {
            for j in 0..n {
                p_loss += p[i] * b_coefficients[i][j] * p[j];
            }
        }

        // 2. 等微增率分配（含网损微增率 L_i = 2·Σ B_ij·P_j）
        let mut total_gen = 0.0;
        for i in 0..n {
            let l_i = 2.0 * (0..n).map(|j| b_coefficients[i][j] * p[j]).sum::<f64>();
            let penalty_factor = 1.0 / (1.0 - l_i);
            let g = &generators[i];
            let p_new = ((lambda - g.b) / (2.0 * g.a)) * penalty_factor;
            p[i] = p_new.clamp(g.p_min_mw, g.p_max_mw);
            total_gen += p[i];
        }

        // 3. 收敛判定
        let mismatch = total_gen - load_mw - p_loss;
        if mismatch.abs() < tolerance {
            let gen_outputs: Vec<(String, f64)> = generators.iter()
                .zip(p.iter())
                .map(|(g, &pg)| (g.gen_id.clone(), pg))
                .collect();
            let total_cost: f64 = gen_outputs.iter()
                .zip(generators.iter())
                .map(|((_, p), g)| g.cost_at(*p))
                .sum();
            return Ok(EconomicDispatchResult {
                gen_outputs,
                total_cost,
                total_generation_mw: total_gen,
                total_load_mw: load_mw,
            });
        }

        // 4. 更新 lambda
        let step = 0.05;
        if mismatch > 0.0 { lambda -= step; } else { lambda += step; }
    }

    Err("Lambda 迭代未收敛".into())
}

步骤 4：定义 Agent 骨架

继承 Agent trait，组合负荷预测与经济调度能力：

use eneros_agent::{Agent, AgentAction, AgentType, AgentContext, GeneratorCostCurve};
use eneros_core::{AuthorityLevel, Jurisdiction, Result, ZoneId};
use eneros_eventbus::Event;
use std::time::Duration;

pub struct EconomicDispatchAgent {
    id: String,
    name: String,
    jurisdiction: Jurisdiction,
    generators: Vec<GeneratorCostCurve>,
    reserve_requirement_mw: f64,
    tick_interval: Duration,
    forecast_agent: LoadForecastAgent,
    last_dispatch: Option<EconomicDispatchResult>,
}

impl EconomicDispatchAgent {
    pub fn new(id: &str, name: &str, zone_ids: Vec<ZoneId>) -> Self {
        Self {
            id: id.to_string(),
            name: name.to_string(),
            jurisdiction: Jurisdiction::for_zones(zone_ids),
            generators: Vec::new(),
            reserve_requirement_mw: 30.0,
            tick_interval: Duration::from_secs(300),  // 5 分钟
            forecast_agent: build_forecast_agent(),
            last_dispatch: None,
        }
    }

    pub fn with_generators(mut self, gens: Vec<GeneratorCostCurve>) -> Self {
        self.generators = gens;
        self
    }

    pub fn with_reserve(mut self, reserve_mw: f64) -> Self {
        self.reserve_requirement_mw = reserve_mw;
        self
    }

    /// 从 AgentContext 获取当前系统总负荷
    fn get_total_load(&self, ctx: &AgentContext) -> f64 {
        let network = ctx.remote.network.read();
        if let Ok(pf_result) = network.solve() {
            let total_load: f64 = pf_result.bus_results.iter()
                .filter(|b| b.p_injection < 0.0)
                .map(|b| -b.p_injection)
                .sum();
            if total_load > 0.0 {
                return total_load * 100.0;  // p.u. → MW
            }
        }
        259.0  // 默认值（IEEE 14 总负荷）
    }
}

#[async_trait::async_trait]
impl Agent for EconomicDispatchAgent {
    fn id(&self) -> &str { &self.id }
    fn name(&self) -> &str { &self.name }
    fn agent_type(&self) -> AgentType { AgentType::Dispatcher }

    fn tick_interval(&self) -> Duration { self.tick_interval }

    fn authority_level(&self) -> AuthorityLevel {
        AuthorityLevel::Operator
    }

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

    async fn handle_event(&mut self, event: &Event, ctx: &AgentContext)
        -> Result<Vec<AgentAction>>
    {
        // 监听负荷突变事件
        if event.event_type == eneros_eventbus::event::EventType::LoadChange {
            let load_mw = self.get_total_load(ctx);
            let dispatch = self.dispatch(load_mw)?;
            self.last_dispatch = Some(dispatch.clone());
            return Ok(vec![AgentAction::PublishEvent(Event::new(
                eneros_eventbus::event::EventType::DispatchUpdate,
                &self.id,
                eneros_eventbus::event::EventPayload::Json(
                    serde_json::to_value(&dispatch).unwrap()
                ),
            ))]);
        }
        Ok(vec![AgentAction::NoOp])
    }

    async fn tick(&mut self, ctx: &AgentContext) -> Result<Vec<AgentAction>> {
        // 1. 预测下一时段负荷
        let current_load = self.get_total_load(ctx);
        self.forecast_agent.ingest_sample(current_load).await.ok();
        let forecast = self.forecast_agent.forecast_steps(1).await
            .unwrap_or_else(|_| vec![current_load]);
        let target_load = forecast.first().copied().unwrap_or(current_load);

        // 2. 经济调度
        let dispatch = self.dispatch(target_load)?;
        self.last_dispatch = Some(dispatch.clone());

        // 3. 发布调度结果
        let event = Event::new(
            eneros_eventbus::event::EventType::DispatchUpdate,
            &self.id,
            eneros_eventbus::event::EventPayload::Json(
                serde_json::to_value(&dispatch).unwrap()
            ),
        );

        Ok(vec![AgentAction::PublishEvent(event)])
    }
}

impl EconomicDispatchAgent {
    fn dispatch(&self, load_mw: f64) -> Result<EconomicDispatchResult> {
        dispatch_with_reserve(&self.generators, load_mw, self.reserve_requirement_mw)
            .map_err(|e| eneros_core::EnerOSError::Agent(e))
    }
}

步骤 5：注册安全约束

调度结果必须通过约束引擎校验后才能下发。将调度结果转换为 Command 并提交校验：

use eneros_gateway::command::{Command, CommandType, CommandPriority};

/// 将调度结果转换为机组出力命令，并提交约束校验
async fn validate_and_dispatch(
    dispatch: &EconomicDispatchResult,
    ctx: &AgentContext,
) -> Result<()> {
    let mut commands = Vec::new();

    for (gen_id, p_mw) in &dispatch.gen_outputs {
        let element_id: u64 = gen_id.trim_start_matches('G')
            .parse()
            .unwrap_or(0);
        let cmd = Command::new(
            CommandType::GeneratorSetpoint,
            element_id,
            CommandPriority::Normal,
            "ed-agent",
        )
        .with_parameter("p_mw", *p_mw);
        commands.push(cmd);
    }

    // 提交约束引擎校验
    let verdict = ctx.constraints.validate_batch(&commands).await?;

    if !verdict.passed {
        eprintln!("调度方案被约束引擎拒绝:");
        for violation in &verdict.violations {
            eprintln!("  - {}", violation);
        }
        // 触发安全回退：使用上一次可行解
        ctx.safety.fallback(&commands, verdict.clone()).await?;
        return Err(eneros_core::EnerOSError::Constraint(
            "调度方案违反约束".into()
        ));
    }

    // 校验通过，下发命令
    for cmd in commands {
        ctx.gateway.execute(cmd).await?;
    }

    println!("✓ 调度方案已下发 ({} 台机组)", dispatch.gen_outputs.len());
    Ok(())
}

约束引擎将检查以下规则：

步骤 6：注册并运行 Agent

将 Agent 注册到 Orchestrator，以 5 分钟周期运行：

use eneros_agent::{AgentOrchestrator, event_adapter::AgentEventHandler};
use eneros_agent::context::AgentContext;
use eneros_eventbus::EventBus;
use eneros_gateway::SafetyGateway;
use eneros_memory::InMemoryMemory;
use eneros_network::PowerNetwork;
use eneros_reasoning::RuleBasedEngine;
use eneros_tool::ToolEngine;
use parking_lot::RwLock;
use std::sync::Arc;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 1. 构建共享上下文
    let ctx = AgentContext::new(
        Arc::new(EventBus::new(64)),
        Arc::new(SafetyGateway::new(100)),
        Arc::new(RwLock::new(ToolEngine::new())),
        Arc::new(RwLock::new(PowerNetwork::from_ieee14())),
        Arc::new(InMemoryMemory::default()),
        Arc::new(RuleBasedEngine::new()),
    );

    // 2. 创建 Orchestrator
    let mut orchestrator = AgentOrchestrator::new(ctx.clone());

    // 3. 创建并注册 DispatchAgent
    let agent = EconomicDispatchAgent::new("ed-001", "EconomicDispatch", vec![1, 2])
        .with_generators(generators.clone())
        .with_reserve(30.0);

    let handler = AgentEventHandler::new_all_events(Box::new(agent));
    orchestrator.register_agent(handler);

    // 4. 注册负荷预测 Agent（独立运行，发布预测结果到事件总线）
    let forecast = build_forecast_agent();
    let forecast_handler = AgentEventHandler::new_all_events(Box::new(forecast));
    orchestrator.register_agent(forecast_handler);

    // 5. 启动调度循环
    println!("启动经济调度 Agent (周期=5min)...");
    for tick in 0..12 {  // 运行 1 小时（12 个 5 分钟周期）
        let results = orchestrator.tick_all().await?;
        println!("[tick {}] {} 个 Agent 响应", tick, results.len());
        tokio::time::sleep(Duration::from_secs(1)).await;  // 演示加速
    }

    Ok(())
}

输出示例：

启动经济调度 Agent (周期=5min)...
[tick 0] 2 个 Agent 响应
[tick 1] 2 个 Agent 响应
...
[tick 11] 2 个 Agent 响应

步骤 7：AGC 与频率控制

经济调度处理 5 分钟级的有功分配，而 AGC（自动发电控制）处理秒级的频率波动。eneros-agent 提供 calculate_ace 函数计算区域控制误差（ACE）：

use eneros_agent::calculate_ace;

/// AGC 子模块：根据频率偏差调整机组出力
fn agc_adjust(
    dispatch: &EconomicDispatchResult,
    frequency_hz: f64,
    nominal_hz: f64,
    k_gov: f64,  // 系统频率响应系数 (MW/0.1Hz)
) -> Vec<(String, f64)> {
    let ace = calculate_ace(frequency_hz, nominal_hz, k_gov);

    // ACE > 0: 系统频率偏低，需要增加出力
    // ACE < 0: 系统频率偏高，需要减少出力
    let total_p = dispatch.total_generation_mw;
    let adjustment = -ace;  // 反向调节

    // 按机组容量比例分配调节量
    dispatch.gen_outputs.iter()
        .map(|(gen_id, p)| {
            let share = p / total_p;
            let adjusted = p + adjustment * share;
            (gen_id.clone(), adjusted)
        })
        .collect()
}

// 测试
let dispatch = last_dispatch.unwrap();
let adjusted = agc_adjust(&dispatch, 49.95, 50.0, 100.0);
println!("AGC 调节后 (频率=49.95Hz, ACE={:.2} MW):", calculate_ace(49.95, 50.0, 100.0));
for (gen_id, p) in adjusted {
    println!("  {}: {:.2} MW", gen_id, p);
}

输出示例：

AGC 调节后 (频率=49.95Hz, ACE=5.00 MW):
  G1: 147.12 MW
  G2: 70.59 MW
  G3: 28.75 MW
  G6: 10.08 MW
  G8: 10.08 MW

步骤 8：测试与调试

单元测试

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_economic_dispatch_basic() {
        let gens = vec![
            GeneratorCostCurve {
                gen_id: "G1".into(), a: 0.01, b: 10.0, c: 100.0,
                p_min_mw: 10.0, p_max_mw: 100.0,
            },
            GeneratorCostCurve {
                gen_id: "G2".into(), a: 0.02, b: 12.0, c: 80.0,
                p_min_mw: 10.0, p_max_mw: 80.0,
            },
        ];
        let result = economic_dispatch(&gens, 100.0);
        assert!((result.total_generation_mw - 100.0).abs() < 0.1);
        assert!(result.total_cost > 0.0);
    }

    #[test]
    fn test_reserve_check() {
        let gens = vec![
            GeneratorCostCurve {
                gen_id: "G1".into(), a: 0.01, b: 10.0, c: 100.0,
                p_min_mw: 10.0, p_max_mw: 100.0,
            },
        ];
        // 容量 100, 负荷 80, 备用 30 → 不满足
        let result = dispatch_with_reserve(&gens, 80.0, 30.0);
        assert!(result.is_err());
    }

    #[test]
    fn test_agc_frequency_response() {
        let ace_low = calculate_ace(49.9, 50.0, 100.0);
        assert!(ace_low > 0.0);  // 频率低 → ACE > 0
        let ace_high = calculate_ace(50.1, 50.0, 100.0);
        assert!(ace_high < 0.0);  // 频率高 → ACE < 0
        let ace_normal = calculate_ace(50.0, 50.0, 100.0);
        assert!(ace_normal.abs() < 1e-6);
    }
}

集成测试

#[tokio::test]
async fn test_dispatch_agent_full_cycle() {
    let ctx = test_context();
    let mut agent = EconomicDispatchAgent::new("test-ed", "TestED", vec![1])
        .with_generators(vec![
            GeneratorCostCurve {
                gen_id: "G1".into(), a: 0.004, b: 12.0, c: 240.0,
                p_min_mw: 50.0, p_max_mw: 200.0,
            },
            GeneratorCostCurve {
                gen_id: "G2".into(), a: 0.0035, b: 13.5, c: 200.0,
                p_min_mw: 20.0, p_max_mw: 80.0,
            },
        ])
        .with_reserve(20.0);

    // 第一次 tick
    let actions = agent.tick(&ctx).await.unwrap();
    assert!(!actions.is_empty());
    assert!(agent.last_dispatch.is_some());

    let d = agent.last_dispatch.as_ref().unwrap();
    assert!((d.total_generation_mw - d.total_load_mw).abs() < 1.0);
}

调试技巧

# 启用详细日志
RUST_LOG=eneros_agent=debug,eneros_powerflow=info cargo run --release

# 查看 Agent 注册状态
eneros-cli agent list

# 订阅调度结果
eneros-cli events subscribe /topics/dispatch/setpoints

# 强制触发调度
eneros-cli agent invoke ed-001 tick

验证

按以下指标验证调度 Agent 的正确性：

下一步

• 多智能体协作 — 多个 Agent 协同决策
• 物理约束决策 — 约束引擎详解
• SCADA 数据采集集成 — 实时数据驱动调度
• 插件开发 — 将调度算法封装为可分发插件