Train MoE Volt/VAR Agents
Train decentralized agents to control voltage on a simulated radial feeder.
Introduction
30 Second Summary
Power demand and solar output can pull the same electricity network in opposite directions. A decision made at one connection can shift voltages elsewhere.
In this project, you will build three decentralized mixture-of-experts policies for Volt/VAR control on a four-bus radial feeder. Each agent acts from local measurements after cooperative training with one shared AC power-flow reward.
What You'll Build
You'll run swarm_voltage_control.py to produce a terminal benchmark alongside a results.png chart showing how the learned swarm behaves in high-solar and high-load conditions.
By the end of this project, you'll have:
- A zero-reactive-power baseline that reveals how one fixed action behaves under high-solar and high-load conditions.
- An expert inspection mode that prints every agent's gate weights plus reactive-power action probabilities.
- A trained decentralized swarm that produces a matched benchmark of voltage deviation, voltage-band violation, active line loss, reward, plus expert routing.
- Secret Mission: Stress-test the trained policies on an unseen mixed solar-load regime to judge whether the expert gates interpolate beyond their training conditions.
Are there any prerequisites?
You should be comfortable with reinforcement learning, power-system concepts, and Python syntax. Basic familiarity with PyTorch modules plus policy gradients will help you focus on the experiment.
Before We Start
This checkpoint commits you to an experiment where three inverter agents each choose a discrete reactive-power setpoint using only local voltage, load, plus solar observations. One cooperative reward scores feeder voltage deviation, voltage-band violation, active line loss, plus reactive-power effort.
Set Up the Windows Training Environment
This experiment needs a compatible Python interpreter. Package drift can turn later power-flow failures into environment failures that are hard to diagnose.
A project-specific virtual environment inside Visual Studio Code pins PyTorch, pandapower, and Matplotlib to exact versions. That isolation keeps the training run reproducible.
An early import check catches package problems before you build the feeder. A CUDA check reports the hardware path available to PyTorch.
In this step, get ready to:
- Verify Python 3.11 or later.
- Prepare an isolated workspace for the feeder code.
- Establish a verified dependency stack with CUDA detection.
Verify Python 3.11 or later
The pinned package set needs Python 3.11 or later. Checking the interpreter first prevents an incompatible environment from reaching the feeder code.
- Press the Windows key to open Windows search.
- Enter Visual Studio Code in the search field.
- Press Enter to start Visual Studio Code.
- Select View from the top menu.
- Select Terminal.
- Check the installed Python version by running this command:
python --versionWhat Does This Check Prove?
The output identifies the interpreter that your terminal reaches. That same interpreter creates the project environment in the next substep.
✔️ I see version 3.11 or higher
That check clears the first compatibility risk. Your displayed version meets the minimum of 3.11.
ⓧ I see an older version
An older interpreter can create an environment that rejects the pinned packages. Install a current Python release before continuing.
- Download a current release from the official Python Windows downloads page.
- Run the downloaded Windows installer.
- Accept the installer prompt that adds Python's command directory to PATH.
- Return to Visual Studio Code after the installer finishes.
- Select Terminal from the top menu.
- Select New Terminal.
- Repeat the version check shown above.
You should now see Python 3.11 or a later release.
Still Seeing the Older Release?
- Press the Windows key.
- Enter Manage App Execution Aliases.
- Press Enter.
- Select the newly installed Python release for python.exe.
- Repeat the version check shown above.
If the old version remains active, help me connect my terminal to the newer Python installation.
ⓧ Command not found
The terminal cannot reach a Python installation. Install a current release before creating the project environment.
- Download a current release from the official Python Windows downloads page.
- Run the downloaded Windows installer.
- Accept the installer prompt that adds Python's command directory to PATH.
- Return to Visual Studio Code after the installer finishes.
- Select Terminal from the top menu.
- Select New Terminal.
- Repeat the version check shown above.
You should now see Python 3.11 or a later release.
Still Missing the Python Command?
Restart Visual Studio Code so its terminal loads the updated PATH. Repeat the version check in a fresh terminal.
If the command is still unavailable, help me make Python available in my Windows terminal.
Create the project workspace
The project needs one named folder for its source file plus its dependency manifest. Keeping the virtual environment in this folder makes every later command use the same package set.
- Create the project folder and starter files by running these PowerShell commands:
Set-Location -Path ([Environment]::GetFolderPath("Desktop"))
New-Item -Path "volt-var-swarm" -ItemType Directory
Set-Location -Path ".\volt-var-swarm"
New-Item -Path "requirements.txt","swarm_voltage_control.py" -ItemType FileWhat Do These Commands Create?
- The first Set-Location moves the terminal to your Windows Desktop.
- The first New-Item creates the volt-var-swarm folder.
- The second Set-Location moves the terminal into that folder.
- The final New-Item creates requirements.txt plus the empty swarm_voltage_control.py source file.
- Review the PowerShell output.
You should see entries for the new folder plus both starter files.
Having Trouble Creating the Workspace?
Confirm that the terminal profile is PowerShell. Check whether volt-var-swarm already exists on your Desktop before repeating the commands.
If the folder or files still fail to appear, help me create the Windows project workspace safely.
- Open the new volt-var-swarm folder in Visual Studio Code by running this command:
code .What Does This Command Do?
The command asks Visual Studio Code to open the terminal's current folder as a workspace. The editor can now keep the source file plus its environment together.
- Check the Explorer sidebar in the project window.
You should see requirements.txt plus swarm_voltage_control.py inside volt-var-swarm.
Folder Not Opening in Visual Studio Code?
- Select File from the top menu.
- Select Open Folder.
- Choose volt-var-swarm from your Desktop.
If Visual Studio Code still cannot open the folder, help me locate and open my volt-var-swarm workspace.
- Select View from the top menu in the project window.
- Select Terminal.
- Create the virtual environment by running this command:
python -m venv .venvWhy Use a Virtual Environment?
The .venv directory gives this project its own Python interpreter plus package location. Later commands call that interpreter directly to avoid relying on a global installation.
- Check the Explorer sidebar again.
You should now see a .venv directory beside the two starter files.
Virtual Environment Missing?
Check that the terminal path ends with volt-var-swarm. Repeat the environment command from that folder.
If .venv still does not appear, help me diagnose the virtual environment creation.
Pin dependencies and verify imports
A requirements file records the exact dependency versions behind the experiment. Every learner can now install the same numerical stack.
- Select requirements.txt in the Explorer sidebar.
- Add the pinned dependencies by pasting this exact content:
torch==2.14.0
pandapower==3.5.4
matplotlib==3.11.1What Do These Packages Provide?
- PyTorch 2.14.0 provides the policy networks plus the optimization tools.
- pandapower 3.5.4 provides the balanced AC power-flow model.
- Matplotlib 3.11.1 creates the final benchmark chart.
- Press Ctrl+S to save requirements.txt.
Package installation can take a few minutes while the pinned wheels download. A long stream of terminal output is expected.
- Upgrade pip inside the environment by running these commands:
.\.venv\Scripts\python.exe -m pip install --upgrade pip
.\.venv\Scripts\python.exe -m pip install -r requirements.txtWhat Do These Install Commands Do?
- The first line updates pip inside .venv.
- The second line reads requirements.txt to install every pinned package into the same environment.
A successful installation returns control to the PowerShell prompt after processing all three packages.
Package Installation Failing?
Confirm that the earlier version check reported Python 3.11 or later. Check that your internet connection can reach package downloads.
If installation still stops early, help me diagnose the pinned dependency installation.
Before you run the final check, what CUDA status do you expect to see? Keep your prediction in mind.
- Verify the imports plus the available CUDA status by running this command:
.\.venv\Scripts\python.exe -c "import torch, pandapower, matplotlib; print('Imports OK'); print('CUDA:', torch.cuda.is_available())"What Does the Final Check Prove?
The imports prove that the pinned libraries load through the environment's interpreter. The CUDA result reports whether PyTorch can access a compatible accelerator.
You'll see Imports OK followed by either CUDA: True or CUDA: False. Both CUDA values are valid because the full project supports CPU execution.
Imports Not Completing?
A missing import means the dependency installation did not finish inside .venv. Repeat the installation commands before running the final check again.
If the environment interpreter cannot start, confirm that .venv exists in volt-var-swarm.
If the check still fails, help me diagnose the environment verification output.
✔️ Awesome, I've got everything!
Great. Double check that requirements.txt is saved plus swarm_voltage_control.py remains empty.
ⓧ I'd like to double check the full code
Your requirements.txt file should contain exactly these three lines:
torch==2.14.0
pandapower==3.5.4
matplotlib==3.11.1Your swarm_voltage_control.py file should still be empty. It is ready for the radial-feeder implementation.
That's the setup hurdle cleared. Next, you'll turn the empty source file into a radial feeder with a visible zero-reactive-power baseline.
Build the Feeder and Test the Baseline
Your Python environment is ready. Now the experiment needs a physical grid where each inverter action changes measurable voltage behavior.
A zero-reactive-power controller gives you a fixed reference across high-solar and high-load conditions. Running that reference first reveals whether one fixed action can handle both regimes.
In this step, get ready to:
- Build a four-bus radial feeder with three inverter locations.
- Sample operating regimes from local load and solar conditions.
- Run the zero-reactive-power baseline across both training regimes.
Build the radial feeder topology
The feeder uses pandapower to represent buses and electrical devices. AC power flow turns those network definitions into bus voltages and line losses.
- Switch back to swarm_voltage_control.py in the editor.
- Add the project imports by pasting this code at the top of the file:
import argparse
import random
from statistics import fmean
import matplotlib.pyplot as plt
import torch
from torch import nn
from pandapower.create import (
create_bus,
create_empty_network,
create_ext_grid,
create_line_from_parameters,
create_load,
create_sgen,
)
from pandapower.run import runppWhat Do These Imports Provide?
- The pandapower creation helpers define every physical component in the feeder.
- The runpp function solves the balanced AC power flow.
- PyTorch supplies tensors for local observations and the device selection used by later policies.
- Matplotlib and fmean are ready for the benchmark you build later.
- Save swarm_voltage_control.py.
- Check the editor for import diagnostics.
You should see the imports without unresolved-module warnings.
Seeing Unresolved Imports?
Confirm that the editor is still using the Python interpreter from the existing .venv environment.
If the packages remain unresolved, help me check the virtual environment and imports.
The constants define the agents and their shared action space. The reward weights turn physical feeder results into one cooperative score.
- Add the feeder constants below the imports by pasting this code:
NUM_AGENTS = 3
REGIMES = ("high_solar", "high_load")
Q_ACTIONS = (-0.45, 0.0, 0.45)
VOLTAGE_WEIGHT = 400.0
VIOLATION_WEIGHT = 120.0
LOSS_WEIGHT = 4.0
Q_WEIGHT = 0.05
DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu")
random.seed(7)
torch.manual_seed(7)How Is the Experiment Defined?
- NUM_AGENTS fixes the swarm at three inverter controllers.
- REGIMES names the two conditions used by the main experiment.
- Q_ACTIONS maps each discrete action ID to one reactive-power setpoint.
- DEVICE uses CUDA when it is available. CPU remains a supported path.
- The fixed random seeds make repeated runs easier to compare.
- Save swarm_voltage_control.py.
- Check that the editor recognizes DEVICE and each reward weight.
You should see each constant defined once below the imports.
Seeing a Name Warning?
Check that the constants sit below every import. Confirm that torch is spelled exactly as shown.
If a constant still appears unresolved, help me compare the imports and constants.
A radial feeder has one path from the slack bus to each downstream bus. This model connects four 10 kV buses with three successive lines.
- Add the first part of RadialFeeder below the random seeds by pasting this code:
class RadialFeeder:
def __init__(self):
self.net = create_empty_network(sn_mva=1.0)
self.buses = [
create_bus(self.net, vn_kv=10.0, name=f"Bus {index}")
for index in range(NUM_AGENTS + 1)
]
create_ext_grid(self.net, bus=self.buses[0], vm_pu=1.0, name="Grid")
for index in range(NUM_AGENTS):
create_line_from_parameters(
self.net,
from_bus=self.buses[index],
to_bus=self.buses[index + 1],
length_km=2.0,
r_ohm_per_km=0.45,
x_ohm_per_km=0.35,
c_nf_per_km=0.0,
max_i_ka=0.4,
name=f"Line {index + 1}",
)How Is the Feeder Connected?
- create_empty_network creates the 1.0 MVA network container.
- The bus comprehension creates Bus 0 through Bus 3.
- create_ext_grid makes Bus 0 the 1.0 per-unit slack reference.
- Each line joins one bus to the next. That sequence preserves the radial structure.
- Save swarm_voltage_control.py.
- Check the class indentation in the editor.
You should see the line-creation loop nested inside __init__.
Seeing an Indentation Error?
Keep the line loop inside __init__. Keep every line parameter inside create_line_from_parameters.
If the nesting is unclear, help me repair the RadialFeeder indentation.
Each downstream bus needs a local load and a PV generator. Their table IDs let later methods update the same devices for every sampled scenario.
- Add the load and PV setup inside __init__ after the line-creation loop by pasting this code:
self.load_ids = []
self.sgen_ids = []
for index, bus in enumerate(self.buses[1:]):
self.load_ids.append(
create_load(
self.net,
bus=bus,
p_mw=0.5,
q_mvar=0.15,
name=f"Load {index + 1}",
)
)
self.sgen_ids.append(
create_sgen(
self.net,
bus=bus,
p_mw=0.5,
q_mvar=0.0,
name=f"PV {index + 1}",
)
)
self.loads = [0.0] * NUM_AGENTS
self.solar = [0.0] * NUM_AGENTSWhy Store the Element IDs?
Pandapower stores loads and static generators in tables. The saved IDs point each agent to its own row when an operating point changes.
The initial values define valid devices. The sampled scenario replaces their active and reactive values before each power-flow solve.
- Save swarm_voltage_control.py.
- Check the editor outline for RadialFeeder.
You should see one class with a complete __init__ method. The method should contain three lines and three load-PV pairs.
Missing a Load or PV Pair?
Confirm that the device loop iterates over self.buses[1:]. That slice excludes the slack bus.
If the lists are nested incorrectly, help me check the load and PV creation loop.
Model operating regimes and local observations
Each episode begins with a randomized operating point. The reserved mixed regime stays out of the main training set so it can support a later generalization test.
- Add _sample_operating_point below __init__ by pasting this code:
def _sample_operating_point(self, regime):
if regime == "high_solar":
loads = [random.uniform(0.35, 0.55) for _ in range(NUM_AGENTS)]
solar = [random.uniform(0.95, 1.25) for _ in range(NUM_AGENTS)]
elif regime == "high_load":
loads = [random.uniform(0.85, 1.15) for _ in range(NUM_AGENTS)]
solar = [random.uniform(0.05, 0.25) for _ in range(NUM_AGENTS)]
elif regime == "mixed":
loads = [random.uniform(0.55, 0.95) for _ in range(NUM_AGENTS)]
solar = [random.uniform(0.45, 0.95) for _ in range(NUM_AGENTS)]
else:
raise ValueError(f"Unknown regime: {regime}")
return loads, solarHow Do the Regimes Differ?
- high_solar samples generation above the local load range.
- high_load samples heavy demand with little local generation.
- mixed samples overlapping load and solar ranges for the reserved stress test.
- An unknown regime raises an error immediately. That prevents a mistyped scenario from producing misleading data.
- Save swarm_voltage_control.py.
- Check that all three regime branches align inside the method.
You should see one return path that supplies three load values and three solar values.
Seeing an Unreachable Branch?
Check the if and elif indentation. Keep the final return outside every regime branch.
If the branches still look wrong, help me fix the operating-point sampler.
The backward-forward sweep solver is designed for radial networks. The flat initialization gives each randomized scenario a consistent starting point.
- Add _solve below _sample_operating_point by pasting this code:
def _solve(self):
runpp(
self.net,
algorithm="bfsw",
calculate_voltage_angles=False,
init="flat",
)Why Use the Radial Solver?
The bfsw algorithm selects pandapower's backward-forward sweep method. It fits this feeder because every downstream bus follows one radial path from the grid.
Voltage angles are unnecessary for the local observations and reward metrics used here. The solve stays focused on the quantities the swarm needs.
- Save swarm_voltage_control.py.
- Check that _solve aligns with _sample_operating_point inside the class.
You should see both methods at the same class indentation level.
Solver Method Nested Incorrectly?
Align the def _solve line with def _sample_operating_point. Keep the runpp call one level deeper.
If the method remains nested, help me repair the class method indentation.
A reset applies one sampled scenario before solving the uncontrolled feeder. Each agent then receives only its own initial voltage and local power values.
- Add reset below _solve by pasting this code:
def reset(self, regime):
self.loads, self.solar = self._sample_operating_point(regime)
for index in range(NUM_AGENTS):
load_id = self.load_ids[index]
sgen_id = self.sgen_ids[index]
self.net.load.at[load_id, "p_mw"] = self.loads[index]
self.net.load.at[load_id, "q_mvar"] = 0.32 * self.loads[index]
self.net.sgen.at[sgen_id, "p_mw"] = self.solar[index]
self.net.sgen.at[sgen_id, "q_mvar"] = 0.0
self._solve()
observations = []
for index, bus in enumerate(self.buses[1:]):
voltage = float(self.net.res_bus.at[bus, "vm_pu"])
observations.append(
torch.tensor(
[
20.0 * (voltage - 1.0),
self.loads[index] / 1.5,
self.solar[index] / 1.5,
],
dtype=torch.float32,
device=DEVICE,
)
)
return observationsWhat Goes Into a Local Observation?
- The first value scales that agent's initial voltage displacement from 1.0 per unit.
- The second value scales that agent's local active load.
- The third value scales that agent's local solar generation.
- No observation includes another agent's measurements. That boundary creates decentralized execution.
- Save swarm_voltage_control.py.
- Check that the observation tensor contains exactly three values.
You should see voltage first. Local load should follow it. Local solar output should appear last.
Observation Tensor Misaligned?
Confirm that the tensor is created inside the loop over self.buses[1:]. Keep the return outside that loop.
If an index is unresolved, help me trace the local observation indices.
The environment applies all three action IDs before the next power-flow solve. This preserves simultaneous control across the feeder.
- Add the first part of step below reset by pasting this code:
def step(self, action_ids):
q_setpoints = [Q_ACTIONS[action_id] for action_id in action_ids]
for sgen_id, q_mvar in zip(self.sgen_ids, q_setpoints):
self.net.sgen.at[sgen_id, "q_mvar"] = q_mvar
self._solve()
voltages = [
float(self.net.res_bus.at[bus, "vm_pu"])
for bus in self.buses[1:]
]
line_loss_mw = float(self.net.res_line["pl_mw"].sum())
voltage_squared_error = sum((voltage - 1.0) ** 2 for voltage in voltages)
violation_pu = sum(
max(0.0, 0.95 - voltage) + max(0.0, voltage - 1.05)
for voltage in voltages
)
q_effort = sum(abs(q_mvar) for q_mvar in q_setpoints)
reward = -(
VOLTAGE_WEIGHT * voltage_squared_error
+ VIOLATION_WEIGHT * violation_pu
+ LOSS_WEIGHT * line_loss_mw
+ Q_WEIGHT * q_effort
)How Does One Action Step Work?
Each action ID selects one value from Q_ACTIONS. The loop writes every selected setpoint before _solve runs.
The shared reward penalizes voltage displacement and target-band violation. It also penalizes active line loss and reactive-power effort.
- Save swarm_voltage_control.py.
- Check that the reward remains inside step.
You should see one negative weighted sum after all four physical terms are calculated.
Reward Terms Outside the Method?
Keep every metric line indented under def step. Keep the four weighted terms inside the reward parentheses.
If the expression is incomplete, help me repair the shared reward calculation.
The metrics dictionary keeps the physical results separate from the scalar reward. That makes the baseline easy to inspect without interpreting the training objective alone.
- Complete step by adding this code below the reward calculation:
metrics = {
"min_voltage": min(voltages),
"max_voltage": max(voltages),
"max_deviation": max(abs(voltage - 1.0) for voltage in voltages),
"violation_pu": violation_pu,
"loss_mw": line_loss_mw,
"reward": reward,
}
return reward, metricsWhy Return Reward and Metrics?
The reward supplies one cooperative learning signal. The metrics expose the voltage range and physical costs that produced that signal.
Returning both keeps training and evaluation attached to the same power-flow result.
- Save swarm_voltage_control.py.
- Check that return reward, metrics is the final line of step.
You should see six named metrics returned with the shared reward.
Metrics Not Returned?
Align metrics with the earlier reward assignment. Align the return with the metrics dictionary.
If the method still returns no values, help me check the end of RadialFeeder.step.
Run the zero-Q baseline
The baseline selects action ID 1 for every agent. That ID maps to a zero reactive-power setpoint in Q_ACTIONS.
- Add baseline_demo below the RadialFeeder class by pasting this code:
def baseline_demo(env):
random.seed(21)
print("Zero-reactive-power baseline")
for regime in REGIMES:
env.reset(regime)
_, metrics = env.step([1] * NUM_AGENTS)
print(
f"{regime:>10} | voltage {metrics['min_voltage']:.4f} to "
f"{metrics['max_voltage']:.4f} pu | loss {metrics['loss_mw']:.4f} MW | "
f"violation {metrics['violation_pu']:.4f} pu"
)What Does the Baseline Measure?
The fixed seed makes this demonstration repeatable. Each regime still receives its own randomized load and solar sample.
Every agent receives action ID 1. The printed row exposes the resulting voltage range and active line loss.
- Save swarm_voltage_control.py.
- Check that the print statement remains inside the regime loop.
You should see one formatted result row prepared for each value in REGIMES.
Only One Regime Prepared?
Keep env.reset and the print call inside the for regime in REGIMES loop.
If the output loop is malformed, help me repair baseline_demo.
The command-line flag keeps this baseline isolated from future training modes. The main path creates one feeder before passing it to the demonstration.
- Add argument parsing and the main path below baseline_demo by pasting this code:
def parse_args():
parser = argparse.ArgumentParser()
parser.add_argument("--baseline", action="store_true")
return parser.parse_args()
def main():
args = parse_args()
env = RadialFeeder()
print(f"Training device: {DEVICE}")
if args.baseline:
baseline_demo(env)
return
if __name__ == "__main__":
main()How Does Baseline Mode Start?
parse_args detects the --baseline flag. main creates the feeder and reports the selected training device.
Baseline mode runs the fixed controller before returning. That keeps this check separate from the learned controller added later.
- Save swarm_voltage_control.py.
- Check the editor for syntax diagnostics across the full file.
You should see the completed script without syntax errors.
Main Path Not Recognized?
Keep parse_args and main outside the class. Confirm that the final guard uses __name__ exactly.
If the script structure is unclear, help me check the baseline main path.
✔️ Awesome, I've got everything!
Great. Double check that swarm_voltage_control.py is saved before you run the baseline.
ⓧ I'd like to double check the full code
Compare your complete swarm_voltage_control.py file with this reference:
import argparse
import random
from statistics import fmean
import matplotlib.pyplot as plt
import torch
from torch import nn
from pandapower.create import (
create_bus,
create_empty_network,
create_ext_grid,
create_line_from_parameters,
create_load,
create_sgen,
)
from pandapower.run import runpp
NUM_AGENTS = 3
REGIMES = ("high_solar", "high_load")
Q_ACTIONS = (-0.45, 0.0, 0.45)
VOLTAGE_WEIGHT = 400.0
VIOLATION_WEIGHT = 120.0
LOSS_WEIGHT = 4.0
Q_WEIGHT = 0.05
DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu")
random.seed(7)
torch.manual_seed(7)
class RadialFeeder:
def __init__(self):
self.net = create_empty_network(sn_mva=1.0)
self.buses = [
create_bus(self.net, vn_kv=10.0, name=f"Bus {index}")
for index in range(NUM_AGENTS + 1)
]
create_ext_grid(self.net, bus=self.buses[0], vm_pu=1.0, name="Grid")
for index in range(NUM_AGENTS):
create_line_from_parameters(
self.net,
from_bus=self.buses[index],
to_bus=self.buses[index + 1],
length_km=2.0,
r_ohm_per_km=0.45,
x_ohm_per_km=0.35,
c_nf_per_km=0.0,
max_i_ka=0.4,
name=f"Line {index + 1}",
)
self.load_ids = []
self.sgen_ids = []
for index, bus in enumerate(self.buses[1:]):
self.load_ids.append(
create_load(
self.net,
bus=bus,
p_mw=0.5,
q_mvar=0.15,
name=f"Load {index + 1}",
)
)
self.sgen_ids.append(
create_sgen(
self.net,
bus=bus,
p_mw=0.5,
q_mvar=0.0,
name=f"PV {index + 1}",
)
)
self.loads = [0.0] * NUM_AGENTS
self.solar = [0.0] * NUM_AGENTS
def _sample_operating_point(self, regime):
if regime == "high_solar":
loads = [random.uniform(0.35, 0.55) for _ in range(NUM_AGENTS)]
solar = [random.uniform(0.95, 1.25) for _ in range(NUM_AGENTS)]
elif regime == "high_load":
loads = [random.uniform(0.85, 1.15) for _ in range(NUM_AGENTS)]
solar = [random.uniform(0.05, 0.25) for _ in range(NUM_AGENTS)]
elif regime == "mixed":
loads = [random.uniform(0.55, 0.95) for _ in range(NUM_AGENTS)]
solar = [random.uniform(0.45, 0.95) for _ in range(NUM_AGENTS)]
else:
raise ValueError(f"Unknown regime: {regime}")
return loads, solar
def _solve(self):
runpp(
self.net,
algorithm="bfsw",
calculate_voltage_angles=False,
init="flat",
)
def reset(self, regime):
self.loads, self.solar = self._sample_operating_point(regime)
for index in range(NUM_AGENTS):
load_id = self.load_ids[index]
sgen_id = self.sgen_ids[index]
self.net.load.at[load_id, "p_mw"] = self.loads[index]
self.net.load.at[load_id, "q_mvar"] = 0.32 * self.loads[index]
self.net.sgen.at[sgen_id, "p_mw"] = self.solar[index]
self.net.sgen.at[sgen_id, "q_mvar"] = 0.0
self._solve()
observations = []
for index, bus in enumerate(self.buses[1:]):
voltage = float(self.net.res_bus.at[bus, "vm_pu"])
observations.append(
torch.tensor(
[
20.0 * (voltage - 1.0),
self.loads[index] / 1.5,
self.solar[index] / 1.5,
],
dtype=torch.float32,
device=DEVICE,
)
)
return observations
def step(self, action_ids):
q_setpoints = [Q_ACTIONS[action_id] for action_id in action_ids]
for sgen_id, q_mvar in zip(self.sgen_ids, q_setpoints):
self.net.sgen.at[sgen_id, "q_mvar"] = q_mvar
self._solve()
voltages = [
float(self.net.res_bus.at[bus, "vm_pu"])
for bus in self.buses[1:]
]
line_loss_mw = float(self.net.res_line["pl_mw"].sum())
voltage_squared_error = sum((voltage - 1.0) ** 2 for voltage in voltages)
violation_pu = sum(
max(0.0, 0.95 - voltage) + max(0.0, voltage - 1.05)
for voltage in voltages
)
q_effort = sum(abs(q_mvar) for q_mvar in q_setpoints)
reward = -(
VOLTAGE_WEIGHT * voltage_squared_error
+ VIOLATION_WEIGHT * violation_pu
+ LOSS_WEIGHT * line_loss_mw
+ Q_WEIGHT * q_effort
)
metrics = {
"min_voltage": min(voltages),
"max_voltage": max(voltages),
"max_deviation": max(abs(voltage - 1.0) for voltage in voltages),
"violation_pu": violation_pu,
"loss_mw": line_loss_mw,
"reward": reward,
}
return reward, metrics
def baseline_demo(env):
random.seed(21)
print("Zero-reactive-power baseline")
for regime in REGIMES:
env.reset(regime)
_, metrics = env.step([1] * NUM_AGENTS)
print(
f"{regime:>10} | voltage {metrics['min_voltage']:.4f} to "
f"{metrics['max_voltage']:.4f} pu | loss {metrics['loss_mw']:.4f} MW | "
f"violation {metrics['violation_pu']:.4f} pu"
)
def parse_args():
parser = argparse.ArgumentParser()
parser.add_argument("--baseline", action="store_true")
return parser.parse_args()
def main():
args = parse_args()
env = RadialFeeder()
print(f"Training device: {DEVICE}")
if args.baseline:
baseline_demo(env)
return
if __name__ == "__main__":
main()How to Use This Reference
Compare the file structure and indentation with your saved version. Keep requirements.txt unchanged.
Before you run it, predict whether the two regimes will produce the same voltage range.
- Run the zero-reactive-power baseline in the existing Visual Studio Code terminal with this command:
.\.venv\Scripts\python.exe swarm_voltage_control.py --baselineWhat Should You See?
The first line reports the selected training device. The next heading identifies the zero-reactive-power baseline.
You should see one high_solar row and one high_load row. Each row should include a bus-voltage range and active line loss.
Each row should also include voltage-band violation. The exact values come from the seeded operating points.
You made the control problem visible. The same zero-Q action produces different physical metrics when the load and solar regime changes.
That shortfall is intentional. A fixed action has no mechanism for adapting its response to local operating conditions.
Baseline Command Failing?
Confirm that the terminal is still inside the open project folder. Check that swarm_voltage_control.py is saved there.
If the traceback points to indentation, compare the affected method with the full-code reference. If the power flow fails, check every line and device value for a transcription error.
For targeted help, help me debug the baseline traceback.
Your feeder now turns local operating conditions into physical grid results. Next up, each agent gets a policy that can route its local observation through specialized experts.
Add Mixture-of-Experts Policies
Your baseline showed that one fixed reactive-power action produces different grid outcomes across the two operating regimes. The feeder now needs policies that can adapt their decisions to each agent's local measurements.
A mixture-of-experts policy gives each agent two specialized action networks. A learned softmax gate in PyTorch blends their outputs before the agent selects a reactive-power action.
In this step, get ready to:
- Define a two-expert policy with a learned routing gate.
- Create three separate policies for decentralized execution.
- Inspect routing weights and action probabilities for both operating regimes.
Define each agent's expert policy
Each agent receives three local inputs: its voltage signal, load level, and solar output. Its gate converts those measurements into two routing weights.
Each expert produces three action logits. The weighted mixture preserves one score for every reactive-power action in Q_ACTIONS.
- Return to the import group near the top of swarm_voltage_control.py.
- Add the action-distribution import below from torch import nn:
from torch.distributions import CategoricalWhat does this import do?
Categorical provides a discrete probability distribution over the three action logits. The cooperative training loop uses it to sample one action per agent.
- Find the end of the RadialFeeder class in swarm_voltage_control.py.
- Add the policy class plus its factory immediately before baseline_demo(env):
class MoEPolicy(nn.Module):
def __init__(self, observation_size=3, hidden_size=16, action_count=3):
super().__init__()
self.gate = nn.Sequential(
nn.Linear(observation_size, hidden_size),
nn.Tanh(),
nn.Linear(hidden_size, 2),
)
self.experts = nn.ModuleList(
[
nn.Sequential(
nn.Linear(observation_size, hidden_size),
nn.Tanh(),
nn.Linear(hidden_size, action_count),
)
for _ in range(2)
]
)
def forward(self, observation):
gate_weights = torch.softmax(self.gate(observation), dim=-1)
expert_logits = torch.stack(
[expert(observation) for expert in self.experts], dim=0
)
mixed_logits = (gate_weights.unsqueeze(-1) * expert_logits).sum(dim=0)
return mixed_logits, gate_weights
def build_policies():
return nn.ModuleList([MoEPolicy() for _ in range(NUM_AGENTS)]).to(DEVICE)How does the policy route an observation?
- The gate maps three local measurements to two routing scores.
- The two networks in experts independently produce logits for the three reactive-power actions.
- The forward() method normalizes the routing scores before blending the expert logits.
- The build_policies() factory registers three separate policies on the detected device.
- Save swarm_voltage_control.py.
- Confirm the editor lists MoEPolicy plus build_policies() as top-level symbols.
Policy symbols missing?
Check that MoEPolicy starts at the far-left indentation level. Make sure build_policies() also sits outside the class.
Compare the closing parentheses around self.gate plus self.experts if the editor cannot parse the class.
Help me check my MoE policy structure.
Inspect every policy by regime
Inspection reveals what an untrained policy computes from each local observation. It exposes the two routing weights plus the three resulting action probabilities without changing the feeder.
- Find the end of baseline_demo(env) in swarm_voltage_control.py.
- Add the inspection function immediately below it:
def inspect_policies(env, policies):
random.seed(22)
for regime in REGIMES:
observations = env.reset(regime)
print(f"\n{regime}")
with torch.no_grad():
for agent_index, observation in enumerate(observations):
logits, gate_weights = policies[agent_index](observation)
action_probabilities = torch.softmax(logits, dim=-1)
print(
f"agent {agent_index + 1}: gate="
f"{gate_weights.detach().cpu().tolist()} actions="
f"{action_probabilities.detach().cpu().tolist()}"
)What does the inspection reveal?
- The fixed random seed makes the sampled inspection scenarios reproducible.
- Each call to env.reset(regime) produces one local observation for every inverter.
- The torch.no_grad() context evaluates the policies without building a gradient graph.
- Each printed row exposes the gate mixture plus the probability assigned to every action.
- Save swarm_voltage_control.py.
- Confirm the editor lists inspect_policies() as a top-level function below baseline_demo(env).
Inspection function nested incorrectly?
Align def inspect_policies(env, policies): with def baseline_demo(env):. Every line inside the new function should use consistent four-space indentation.
Help me fix the inspection function.
Connect the inspection command
The command-line parser needs an inspection flag. The main path can then build the three policies before sending them through both regimes.
- Find the existing parse_args() function near the bottom of swarm_voltage_control.py.
- Replace the complete function with this updated version:
def parse_args():
parser = argparse.ArgumentParser()
parser.add_argument("--baseline", action="store_true")
parser.add_argument("--inspect", action="store_true")
return parser.parse_args()What changed in the parser?
The parser now recognizes the inspection mode while preserving the existing baseline mode. Supplying the new flag sets args.inspect to a Boolean value.
- Find the existing main() function below parse_args().
- Replace the complete function with this updated version:
def main():
args = parse_args()
env = RadialFeeder()
print(f"Training device: {DEVICE}")
if args.baseline:
baseline_demo(env)
return
policies = build_policies()
if args.inspect:
inspect_policies(env, policies)
returnHow does the inspection path work?
- The existing baseline branch still runs before any policies are created.
- The call to build_policies() creates one independent policy for each inverter.
- The inspection branch prints policy outputs before returning without training.
- Save swarm_voltage_control.py.
✔️ Awesome, I've got everything!
Your policy class, policy factory, inspection function, parser flag, and inspection branch are now saved in swarm_voltage_control.py.
ⓧ I'd like to double check the full code
Compare your cumulative swarm_voltage_control.py file with this version:
import argparse
import random
from statistics import fmean
import matplotlib.pyplot as plt
import torch
from torch import nn
from torch.distributions import Categorical
from pandapower.create import (
create_bus,
create_empty_network,
create_ext_grid,
create_line_from_parameters,
create_load,
create_sgen,
)
from pandapower.run import runpp
NUM_AGENTS = 3
REGIMES = ("high_solar", "high_load")
Q_ACTIONS = (-0.45, 0.0, 0.45)
VOLTAGE_WEIGHT = 400.0
VIOLATION_WEIGHT = 120.0
LOSS_WEIGHT = 4.0
Q_WEIGHT = 0.05
DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu")
random.seed(7)
torch.manual_seed(7)
class RadialFeeder:
def __init__(self):
self.net = create_empty_network(sn_mva=1.0)
self.buses = [
create_bus(self.net, vn_kv=10.0, name=f"Bus {index}")
for index in range(NUM_AGENTS + 1)
]
create_ext_grid(self.net, bus=self.buses[0], vm_pu=1.0, name="Grid")
for index in range(NUM_AGENTS):
create_line_from_parameters(
self.net,
from_bus=self.buses[index],
to_bus=self.buses[index + 1],
length_km=2.0,
r_ohm_per_km=0.45,
x_ohm_per_km=0.35,
c_nf_per_km=0.0,
max_i_ka=0.4,
name=f"Line {index + 1}",
)
self.load_ids = []
self.sgen_ids = []
for index, bus in enumerate(self.buses[1:]):
self.load_ids.append(
create_load(
self.net,
bus=bus,
p_mw=0.5,
q_mvar=0.15,
name=f"Load {index + 1}",
)
)
self.sgen_ids.append(
create_sgen(
self.net,
bus=bus,
p_mw=0.5,
q_mvar=0.0,
name=f"PV {index + 1}",
)
)
self.loads = [0.0] * NUM_AGENTS
self.solar = [0.0] * NUM_AGENTS
def _sample_operating_point(self, regime):
if regime == "high_solar":
loads = [random.uniform(0.35, 0.55) for _ in range(NUM_AGENTS)]
solar = [random.uniform(0.95, 1.25) for _ in range(NUM_AGENTS)]
elif regime == "high_load":
loads = [random.uniform(0.85, 1.15) for _ in range(NUM_AGENTS)]
solar = [random.uniform(0.05, 0.25) for _ in range(NUM_AGENTS)]
elif regime == "mixed":
loads = [random.uniform(0.55, 0.95) for _ in range(NUM_AGENTS)]
solar = [random.uniform(0.45, 0.95) for _ in range(NUM_AGENTS)]
else:
raise ValueError(f"Unknown regime: {regime}")
return loads, solar
def _solve(self):
runpp(
self.net,
algorithm="bfsw",
calculate_voltage_angles=False,
init="flat",
)
def reset(self, regime):
self.loads, self.solar = self._sample_operating_point(regime)
for index in range(NUM_AGENTS):
load_id = self.load_ids[index]
sgen_id = self.sgen_ids[index]
self.net.load.at[load_id, "p_mw"] = self.loads[index]
self.net.load.at[load_id, "q_mvar"] = 0.32 * self.loads[index]
self.net.sgen.at[sgen_id, "p_mw"] = self.solar[index]
self.net.sgen.at[sgen_id, "q_mvar"] = 0.0
self._solve()
observations = []
for index, bus in enumerate(self.buses[1:]):
voltage = float(self.net.res_bus.at[bus, "vm_pu"])
observations.append(
torch.tensor(
[
20.0 * (voltage - 1.0),
self.loads[index] / 1.5,
self.solar[index] / 1.5,
],
dtype=torch.float32,
device=DEVICE,
)
)
return observations
def step(self, action_ids):
q_setpoints = [Q_ACTIONS[action_id] for action_id in action_ids]
for sgen_id, q_mvar in zip(self.sgen_ids, q_setpoints):
self.net.sgen.at[sgen_id, "q_mvar"] = q_mvar
self._solve()
voltages = [
float(self.net.res_bus.at[bus, "vm_pu"])
for bus in self.buses[1:]
]
line_loss_mw = float(self.net.res_line["pl_mw"].sum())
voltage_squared_error = sum((voltage - 1.0) ** 2 for voltage in voltages)
violation_pu = sum(
max(0.0, 0.95 - voltage) + max(0.0, voltage - 1.05)
for voltage in voltages
)
q_effort = sum(abs(q_mvar) for q_mvar in q_setpoints)
reward = -(
VOLTAGE_WEIGHT * voltage_squared_error
+ VIOLATION_WEIGHT * violation_pu
+ LOSS_WEIGHT * line_loss_mw
+ Q_WEIGHT * q_effort
)
metrics = {
"min_voltage": min(voltages),
"max_voltage": max(voltages),
"max_deviation": max(abs(voltage - 1.0) for voltage in voltages),
"violation_pu": violation_pu,
"loss_mw": line_loss_mw,
"reward": reward,
}
return reward, metrics
class MoEPolicy(nn.Module):
def __init__(self, observation_size=3, hidden_size=16, action_count=3):
super().__init__()
self.gate = nn.Sequential(
nn.Linear(observation_size, hidden_size),
nn.Tanh(),
nn.Linear(hidden_size, 2),
)
self.experts = nn.ModuleList(
[
nn.Sequential(
nn.Linear(observation_size, hidden_size),
nn.Tanh(),
nn.Linear(hidden_size, action_count),
)
for _ in range(2)
]
)
def forward(self, observation):
gate_weights = torch.softmax(self.gate(observation), dim=-1)
expert_logits = torch.stack(
[expert(observation) for expert in self.experts], dim=0
)
mixed_logits = (gate_weights.unsqueeze(-1) * expert_logits).sum(dim=0)
return mixed_logits, gate_weights
def build_policies():
return nn.ModuleList([MoEPolicy() for _ in range(NUM_AGENTS)]).to(DEVICE)
def baseline_demo(env):
random.seed(21)
print("Zero-reactive-power baseline")
for regime in REGIMES:
env.reset(regime)
_, metrics = env.step([1] * NUM_AGENTS)
print(
f"{regime:>10} | voltage {metrics['min_voltage']:.4f} to "
f"{metrics['max_voltage']:.4f} pu | loss {metrics['loss_mw']:.4f} MW | "
f"violation {metrics['violation_pu']:.4f} pu"
)
def inspect_policies(env, policies):
random.seed(22)
for regime in REGIMES:
observations = env.reset(regime)
print(f"\n{regime}")
with torch.no_grad():
for agent_index, observation in enumerate(observations):
logits, gate_weights = policies[agent_index](observation)
action_probabilities = torch.softmax(logits, dim=-1)
print(
f"agent {agent_index + 1}: gate="
f"{gate_weights.detach().cpu().tolist()} actions="
f"{action_probabilities.detach().cpu().tolist()}"
)
def parse_args():
parser = argparse.ArgumentParser()
parser.add_argument("--baseline", action="store_true")
parser.add_argument("--inspect", action="store_true")
return parser.parse_args()
def main():
args = parse_args()
env = RadialFeeder()
print(f"Training device: {DEVICE}")
if args.baseline:
baseline_demo(env)
return
policies = build_policies()
if args.inspect:
inspect_policies(env, policies)
return
if __name__ == "__main__":
main()- Before running the inspection, predict how many gate weights plus action probabilities each agent should report.
- Run the policy inspection from the Visual Studio Code terminal with this command:
.\.venv\Scripts\python.exe swarm_voltage_control.py --inspectWhat should I see?
You'll see separate sections for high_solar plus high_load.
Agents 1 through 3 each print two gate weights. Each agent also prints three action probabilities.
You now have three separate MoE policies. Each policy exposes its routing mixture plus action distribution from local observations.
Inspection command failing?
If only the device line appears, check that the inspection branch sits inside main(). Confirm that its indentation matches the baseline branch.
If one regime stops before all three agents print, compare the tensor dimensions in MoEPolicy.forward() with the policy code above.
Help me debug the inspection output.
Your agents can now route local observations through separate expert policies. Next, one shared grid reward turns those independent decisions into cooperative learning.
Train the Swarm with Shared Rewards
Your three mixture-of-experts policies now reveal gate weights for each local observation. They also reveal the resulting action probabilities.
Each agent sees only its own voltage, load, and solar measurements. A shared reward supplies the system-wide coordination signal that those local observations cannot provide.
In this step, get ready to:
- Define the weighted training objective.
- Sample decentralized actions under one shared reward.
- Update all three policies through episode 500.
Define the shared objective
The existing feeder already combines voltage deviation, voltage-band violation, active line loss, and reactive-power effort into one reward. These constants now define the training horizon plus the strength of the routing and exploration objectives.
- In the Explorer sidebar, select swarm_voltage_control.py.
- Find the line containing Q_ACTIONS = (-0.45, 0.0, 0.45).
- Add the training constants below that line by pasting this code:
TRAIN_EPISODES = 500
VOLTAGE_WEIGHT = 400.0
VIOLATION_WEIGHT = 120.0
LOSS_WEIGHT = 4.0
Q_WEIGHT = 0.05
ROUTING_WEIGHT = 0.25
ENTROPY_WEIGHT = 0.01How do these weights shape training?
- TRAIN_EPISODES gives the policies 500 randomized operating points to learn from.
- VOLTAGE_WEIGHT gives squared voltage deviation the strongest physical penalty.
- VIOLATION_WEIGHT penalizes voltage that crosses the project target band.
- LOSS_WEIGHT prices active line loss into every shared reward.
- Q_WEIGHT discourages unnecessary reactive-power effort.
- ROUTING_WEIGHT guides each gate toward the expert associated with the sampled regime.
- ENTROPY_WEIGHT preserves action exploration during training.
- Save swarm_voltage_control.py.
- Confirm the seven training constants appear before the existing DEVICE assignment.
Seeing a constant-related error?
Check every constant name against the code above. Python treats capitalization as significant.
Keep each assignment on its own line. Make sure none of the values are inside quotation marks.
Help me check my training constants.
Build the cooperative training loop
The cooperative policy gradient samples one action from each policy before the feeder solves its power flow. Every policy receives the same advantage from the resulting shared reward.
- Find the end of the existing inspect_policies() function in swarm_voltage_control.py.
- Add the first part of train() below that function by pasting this code:
def train(env, policies):
optimizer = torch.optim.Adam(policies.parameters(), lr=0.01)
reward_baseline = 0.0
reward_history = []
for episode in range(1, TRAIN_EPISODES + 1):
regime = REGIMES[(episode - 1) % len(REGIMES)]
observations = env.reset(regime)
actions = []
log_probabilities = []
entropies = []
routing_losses = []
target_expert = 0 if regime == "high_solar" else 1
for agent_index, observation in enumerate(observations):
logits, gate_weights = policies[agent_index](observation)
distribution = Categorical(logits=logits)
action = distribution.sample()
actions.append(action.item())
log_probabilities.append(distribution.log_prob(action))
entropies.append(distribution.entropy())
routing_losses.append((gate_weights[target_expert] - 1.0) ** 2)How are decentralized actions sampled?
- torch.optim.Adam receives parameters from all three policies through one optimizer.
- REGIMES alternates the episodes between high solar and high load.
- Categorical(logits=logits) turns each agent's mixed logits into a distribution over the three reactive-power actions.
- distribution.sample() draws the action that the agent explores during this episode.
- target_expert associates expert 0 with high solar. Expert 1 receives the high-load target.
- Save swarm_voltage_control.py.
- Confirm the inner agent loop stores one action, one log probability, one entropy value, and one routing loss per agent.
Does the agent loop look misaligned?
Keep the inner loop indented inside the episode loop. Keep every sampling line indented inside the agent loop.
Help me fix the indentation in train().
The sampled actions now need to affect the feeder before learning begins. A moving reward baseline centers the shared return while entropy keeps the action distribution from collapsing too early.
- Find the line containing routing_losses.append((gate_weights[target_expert] - 1.0) ** 2).
- Complete train() below that line by pasting this code with the displayed indentation:
reward, _ = env.step(actions)
advantage = reward - reward_baseline
reward_baseline = 0.95 * reward_baseline + 0.05 * reward
policy_loss = -torch.stack(log_probabilities).sum() * advantage
entropy_bonus = torch.stack(entropies).sum()
routing_loss = torch.stack(routing_losses).mean()
loss = (
policy_loss
+ ROUTING_WEIGHT * routing_loss
- ENTROPY_WEIGHT * entropy_bonus
)
optimizer.zero_grad()
loss.backward()
optimizer.step()
reward_history.append(reward)
if episode % 50 == 0:
print(
f"episode {episode:>3} | mean reward "
f"{fmean(reward_history[-50:]):.4f}"
)
return reward_historyHow does the shared update work?
- env.step(actions) applies all three reactive-power actions before calculating one feeder reward.
- reward_baseline tracks a moving reference using the current reward.
- policy_loss multiplies the summed agent log probabilities by the shared advantage.
- entropy_bonus rewards broader action distributions during exploration.
- routing_loss nudges each local gate toward the regime-associated expert.
- loss.backward() computes gradients through every participating policy.
- optimizer.step() updates all registered policy parameters together.
- Save swarm_voltage_control.py.
- Confirm return reward_history aligns with the start of the episode loop.
Seeing a training-loop syntax problem?
Check the indentation around the multiline loss expression. Its three terms must remain inside the surrounding parentheses.
Check that return reward_history sits inside train() but outside the episode loop.
Help me debug the cooperative training loop.
Run the 500-episode training path
The training function needs a command-line route that skips inspection. The main path can then build the policies before starting the optimizer.
- Find the existing def parse_args(): section near the bottom of swarm_voltage_control.py.
- Replace the parser and main section with this code:
def parse_args():
parser = argparse.ArgumentParser()
parser.add_argument("--baseline", action="store_true")
parser.add_argument("--inspect", action="store_true")
parser.add_argument("--train-only", action="store_true")
return parser.parse_args()
def main():
args = parse_args()
env = RadialFeeder()
print(f"Training device: {DEVICE}")
if args.baseline:
baseline_demo(env)
return
policies = build_policies()
if args.inspect:
inspect_policies(env, policies)
return
train(env, policies)
if args.train_only:
return
if __name__ == "__main__":
main()What does this wiring do?
- --train-only selects the cooperative training path from the terminal.
- build_policies() still creates three separate policies on the detected device.
- train(env, policies) starts learning after the baseline and inspection branches have been checked.
- return ends train-only mode after the final progress update.
- Save swarm_voltage_control.py.
Before you run this, form a prediction about which episodes produce progress updates.
- Train all three policies by running this command:
.\.venv\Scripts\python.exe swarm_voltage_control.py --train-onlyWhat should I see?
You should first see the selected training device. You should then see a mean reward at episodes 50, 100, 150, and every further 50-episode interval through episode 500.
The exact reward values depend on the sampled actions and randomized operating points. A complete sequence through episode 500 proves that every policy participated in the shared update loop.
Training stopped before episode 500?
If Python reports a missing name, compare the training constants and function names with the code above. A spelling mismatch can stop the loop before its first update.
If the failure points to tensor operations, confirm that every policy still moves to DEVICE inside build_policies().
Help me diagnose my interrupted training run.
That is the coordination loop working: three decentralized policies now learn from one feeder-wide reward.
✔️ Awesome, I've got everything!
Excellent. Confirm that swarm_voltage_control.py is saved and that the final progress line reaches episode 500.
ⓧ I'd like to double check the full code
Compare your cumulative swarm_voltage_control.py file with this training-ready version.
import argparse
import random
from statistics import fmean
import matplotlib.pyplot as plt
import torch
from torch import nn
from torch.distributions import Categorical
from pandapower.create import (
create_bus,
create_empty_network,
create_ext_grid,
create_line_from_parameters,
create_load,
create_sgen,
)
from pandapower.run import runpp
NUM_AGENTS = 3
REGIMES = ("high_solar", "high_load")
Q_ACTIONS = (-0.45, 0.0, 0.45)
TRAIN_EPISODES = 500
VOLTAGE_WEIGHT = 400.0
VIOLATION_WEIGHT = 120.0
LOSS_WEIGHT = 4.0
Q_WEIGHT = 0.05
ROUTING_WEIGHT = 0.25
ENTROPY_WEIGHT = 0.01
DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu")
random.seed(7)
torch.manual_seed(7)
class RadialFeeder:
def __init__(self):
self.net = create_empty_network(sn_mva=1.0)
self.buses = [
create_bus(self.net, vn_kv=10.0, name=f"Bus {index}")
for index in range(NUM_AGENTS + 1)
]
create_ext_grid(self.net, bus=self.buses[0], vm_pu=1.0, name="Grid")
for index in range(NUM_AGENTS):
create_line_from_parameters(
self.net,
from_bus=self.buses[index],
to_bus=self.buses[index + 1],
length_km=2.0,
r_ohm_per_km=0.45,
x_ohm_per_km=0.35,
c_nf_per_km=0.0,
max_i_ka=0.4,
name=f"Line {index + 1}",
)
self.load_ids = []
self.sgen_ids = []
for index, bus in enumerate(self.buses[1:]):
self.load_ids.append(
create_load(
self.net,
bus=bus,
p_mw=0.5,
q_mvar=0.15,
name=f"Load {index + 1}",
)
)
self.sgen_ids.append(
create_sgen(
self.net,
bus=bus,
p_mw=0.5,
q_mvar=0.0,
name=f"PV {index + 1}",
)
)
self.loads = [0.0] * NUM_AGENTS
self.solar = [0.0] * NUM_AGENTS
def _sample_operating_point(self, regime):
if regime == "high_solar":
loads = [random.uniform(0.35, 0.55) for _ in range(NUM_AGENTS)]
solar = [random.uniform(0.95, 1.25) for _ in range(NUM_AGENTS)]
elif regime == "high_load":
loads = [random.uniform(0.85, 1.15) for _ in range(NUM_AGENTS)]
solar = [random.uniform(0.05, 0.25) for _ in range(NUM_AGENTS)]
elif regime == "mixed":
loads = [random.uniform(0.55, 0.95) for _ in range(NUM_AGENTS)]
solar = [random.uniform(0.45, 0.95) for _ in range(NUM_AGENTS)]
else:
raise ValueError(f"Unknown regime: {regime}")
return loads, solar
def _solve(self):
runpp(
self.net,
algorithm="bfsw",
calculate_voltage_angles=False,
init="flat",
)
def reset(self, regime):
self.loads, self.solar = self._sample_operating_point(regime)
for index in range(NUM_AGENTS):
load_id = self.load_ids[index]
sgen_id = self.sgen_ids[index]
self.net.load.at[load_id, "p_mw"] = self.loads[index]
self.net.load.at[load_id, "q_mvar"] = 0.32 * self.loads[index]
self.net.sgen.at[sgen_id, "p_mw"] = self.solar[index]
self.net.sgen.at[sgen_id, "q_mvar"] = 0.0
self._solve()
observations = []
for index, bus in enumerate(self.buses[1:]):
voltage = float(self.net.res_bus.at[bus, "vm_pu"])
observations.append(
torch.tensor(
[
20.0 * (voltage - 1.0),
self.loads[index] / 1.5,
self.solar[index] / 1.5,
],
dtype=torch.float32,
device=DEVICE,
)
)
return observations
def step(self, action_ids):
q_setpoints = [Q_ACTIONS[action_id] for action_id in action_ids]
for sgen_id, q_mvar in zip(self.sgen_ids, q_setpoints):
self.net.sgen.at[sgen_id, "q_mvar"] = q_mvar
self._solve()
voltages = [
float(self.net.res_bus.at[bus, "vm_pu"])
for bus in self.buses[1:]
]
line_loss_mw = float(self.net.res_line["pl_mw"].sum())
voltage_squared_error = sum((voltage - 1.0) ** 2 for voltage in voltages)
violation_pu = sum(
max(0.0, 0.95 - voltage) + max(0.0, voltage - 1.05)
for voltage in voltages
)
q_effort = sum(abs(q_mvar) for q_mvar in q_setpoints)
reward = -(
VOLTAGE_WEIGHT * voltage_squared_error
+ VIOLATION_WEIGHT * violation_pu
+ LOSS_WEIGHT * line_loss_mw
+ Q_WEIGHT * q_effort
)
metrics = {
"min_voltage": min(voltages),
"max_voltage": max(voltages),
"max_deviation": max(abs(voltage - 1.0) for voltage in voltages),
"violation_pu": violation_pu,
"loss_mw": line_loss_mw,
"reward": reward,
}
return reward, metrics
class MoEPolicy(nn.Module):
def __init__(self, observation_size=3, hidden_size=16, action_count=3):
super().__init__()
self.gate = nn.Sequential(
nn.Linear(observation_size, hidden_size),
nn.Tanh(),
nn.Linear(hidden_size, 2),
)
self.experts = nn.ModuleList(
[
nn.Sequential(
nn.Linear(observation_size, hidden_size),
nn.Tanh(),
nn.Linear(hidden_size, action_count),
)
for _ in range(2)
]
)
def forward(self, observation):
gate_weights = torch.softmax(self.gate(observation), dim=-1)
expert_logits = torch.stack(
[expert(observation) for expert in self.experts], dim=0
)
mixed_logits = (gate_weights.unsqueeze(-1) * expert_logits).sum(dim=0)
return mixed_logits, gate_weights
def build_policies():
return nn.ModuleList([MoEPolicy() for _ in range(NUM_AGENTS)]).to(DEVICE)
def baseline_demo(env):
random.seed(21)
print("Zero-reactive-power baseline")
for regime in REGIMES:
env.reset(regime)
_, metrics = env.step([1] * NUM_AGENTS)
print(
f"{regime:>10} | voltage {metrics['min_voltage']:.4f} to "
f"{metrics['max_voltage']:.4f} pu | loss {metrics['loss_mw']:.4f} MW | "
f"violation {metrics['violation_pu']:.4f} pu"
)
def inspect_policies(env, policies):
random.seed(22)
for regime in REGIMES:
observations = env.reset(regime)
print(f"\n{regime}")
with torch.no_grad():
for agent_index, observation in enumerate(observations):
logits, gate_weights = policies[agent_index](observation)
action_probabilities = torch.softmax(logits, dim=-1)
print(
f"agent {agent_index + 1}: gate="
f"{gate_weights.detach().cpu().tolist()} actions="
f"{action_probabilities.detach().cpu().tolist()}"
)
def train(env, policies):
optimizer = torch.optim.Adam(policies.parameters(), lr=0.01)
reward_baseline = 0.0
reward_history = []
for episode in range(1, TRAIN_EPISODES + 1):
regime = REGIMES[(episode - 1) % len(REGIMES)]
observations = env.reset(regime)
actions = []
log_probabilities = []
entropies = []
routing_losses = []
target_expert = 0 if regime == "high_solar" else 1
for agent_index, observation in enumerate(observations):
logits, gate_weights = policies[agent_index](observation)
distribution = Categorical(logits=logits)
action = distribution.sample()
actions.append(action.item())
log_probabilities.append(distribution.log_prob(action))
entropies.append(distribution.entropy())
routing_losses.append((gate_weights[target_expert] - 1.0) ** 2)
reward, _ = env.step(actions)
advantage = reward - reward_baseline
reward_baseline = 0.95 * reward_baseline + 0.05 * reward
policy_loss = -torch.stack(log_probabilities).sum() * advantage
entropy_bonus = torch.stack(entropies).sum()
routing_loss = torch.stack(routing_losses).mean()
loss = (
policy_loss
+ ROUTING_WEIGHT * routing_loss
- ENTROPY_WEIGHT * entropy_bonus
)
optimizer.zero_grad()
loss.backward()
optimizer.step()
reward_history.append(reward)
if episode % 50 == 0:
print(
f"episode {episode:>3} | mean reward "
f"{fmean(reward_history[-50:]):.4f}"
)
return reward_history
def parse_args():
parser = argparse.ArgumentParser()
parser.add_argument("--baseline", action="store_true")
parser.add_argument("--inspect", action="store_true")
parser.add_argument("--train-only", action="store_true")
return parser.parse_args()
def main():
args = parse_args()
env = RadialFeeder()
print(f"Training device: {DEVICE}")
if args.baseline:
baseline_demo(env)
return
policies = build_policies()
if args.inspect:
inspect_policies(env, policies)
return
train(env, policies)
if args.train_only:
return
if __name__ == "__main__":
main()Your swarm now turns local observations into coordinated policy updates. Next, you will benchmark those learned decisions against the zero-reactive-power baseline.
Benchmark the Learned Controller
Your three mixture-of-experts policies now learn from one shared grid reward. Training output shows that optimization is happening. It does not show whether the learned controller improves the feeder's physical behavior.
This benchmark tests the zero-reactive-power baseline beside the greedy MoE controller on matched random scenarios. It measures voltage deviation, voltage-band violation, active line loss, reward, and expert routing.
In this step, get ready to:
- Evaluate both controllers on matched high-solar and high-load samples.
- Print averaged grid metrics for each controller.
- Plot the physical metrics beside the learned routing behavior.
Add matched evaluation
A fair comparison gives both controllers the same sequence of randomized operating points. Each MoE policy then uses its highest mixed logit to choose a deterministic action for the benchmark.
- In `swarm_voltage_control.py`, find `TRAIN_EPISODES = 500`.
- Add this evaluation sample count directly below it:
EVAL_SAMPLES = 40Why Use 40 Samples?
Each controller faces 40 randomized samples from each regime. Averaging those samples makes the comparison less dependent on one operating point.
- In `swarm_voltage_control.py`, add `greedy_actions()` directly below `train()` by pasting this function:
def greedy_actions(policies, observations):
actions = []
expert_zero_weights = []
with torch.no_grad():
for agent_index, observation in enumerate(observations):
logits, gate_weights = policies[agent_index](observation)
action_values = logits.detach().cpu().tolist()
action = max(
range(len(action_values)),
key=action_values.__getitem__,
)
actions.append(action)
expert_zero_weights.append(float(gate_weights[0].item()))
return actions, expert_zero_weightsWhat Does This Code Do?
- `torch.no_grad()` disables gradient tracking because evaluation does not update the policies.
- Each policy receives only its own agent's local observation.
- `max()` selects the action with the highest mixed logit.
- `expert_zero_weights` records how strongly each gate routes toward expert 0.
- Save `swarm_voltage_control.py`.
- Confirm the script still loads by running the existing inspection mode:
.\.venv\Scripts\python.exe swarm_voltage_control.py --inspectWhat Does This Check Prove?
The inspection output confirms that Python can parse the new function without disrupting the existing policy path. You should still see three agents under both training regimes.
Inspection Mode Stopped Working?
Check that `greedy_actions()` starts at the left edge of the file. Make sure its indented lines use one consistent indentation level.
Compare every closing parenthesis around `max()` with the snippet.
Ask for help with the exact traceback if the script still stops. Help me debug my greedy_actions function.
The raw results from many scenarios need one summary row. The next helper averages every metric returned by `RadialFeeder.step()`.
- Add `summarize()` directly below `greedy_actions()` by pasting this function:
def summarize(rows):
return {
key: fmean(row[key] for row in rows)
for key in rows[0]
}How Does the Summary Work?
The dictionary comprehension visits every metric key in the first result row. `fmean()` calculates that metric's arithmetic mean across all sampled scenarios.
- Save `swarm_voltage_control.py`.
- Check the updated script by running inspection mode again:
.\.venv\Scripts\python.exe swarm_voltage_control.py --inspectWhat Should You See?
You should see the same gate weights and action probabilities as before. That output confirms the summary helper has not broken the runnable script.
Seeing a Syntax Error?
Check the braces around the returned dictionary. Confirm that `for key in rows[0]` stays inside those braces.
Use the traceback line number to compare the failing line with the snippet. Help me fix my summarize function.
The evaluator resets the random seed before testing each controller. This gives the baseline and MoE controller matching scenario sequences.
- Add `evaluate()` directly below `summarize()` by pasting this function:
def evaluate(env, policies, include_mixed=False):
regimes = list(REGIMES)
if include_mixed:
regimes.append("mixed")
summary = {"baseline": {}, "moe": {}}
gate_summary = {}
for controller in ("baseline", "moe"):
random.seed(2026)
for regime in regimes:
rows = []
gate_values = []
for _ in range(EVAL_SAMPLES):
observations = env.reset(regime)
if controller == "baseline":
actions = [1] * NUM_AGENTS
else:
actions, gates = greedy_actions(policies, observations)
gate_values.extend(gates)
_, metrics = env.step(actions)
rows.append(metrics)
summary[controller][regime] = summarize(rows)
if controller == "moe":
gate_summary[regime] = fmean(gate_values)
return regimes, summary, gate_summaryHow Does Matched Evaluation Work?
- `random.seed(2026)` restarts the scenario sequence before each controller is tested.
- The baseline assigns action ID `1` to every agent. That action maps to zero reactive power.
- The MoE controller uses `greedy_actions()` for deterministic evaluation.
- The returned summaries contain averaged physical metrics plus mean expert-0 routing.
- Save `swarm_voltage_control.py`.
- Confirm the evaluation function parses by running inspection mode:
.\.venv\Scripts\python.exe swarm_voltage_control.py --inspectWhat Does This Check Confirm?
The familiar inspection report confirms that the larger evaluation function loads successfully. The default workflow will call it after the reporting functions are ready.
Evaluation Function Not Loading?
Check that `summary[controller][regime]` remains inside the regime loop. Confirm that the final `return` aligns with the first `for` statement.
Inspect the traceback for a missing bracket around `summary` or `gate_summary`. Help me debug my evaluate function.
Report and plot the benchmark
A benchmark becomes useful when its physical metrics are easy to compare. The terminal report preserves exact values. A Matplotlib figure makes the controller differences and routing pattern visible.
- Add `print_results()` directly below `evaluate()` by pasting this function:
def print_results(regimes, summary, gate_summary):
print("\nController benchmark")
for controller in ("baseline", "moe"):
for regime in regimes:
metrics = summary[controller][regime]
print(
f"{controller:>8} | {regime:>10} | max deviation "
f"{metrics['max_deviation']:.4f} pu | violation "
f"{metrics['violation_pu']:.4f} pu | loss "
f"{metrics['loss_mw']:.4f} MW | reward "
f"{metrics['reward']:.4f}"
)
print("\nMean expert-0 gate weight")
for regime in regimes:
print(f"{regime:>10}: {gate_summary[regime]:.3f}")What Will the Report Contain?
Each controller and regime receives one row for maximum voltage deviation, voltage-band violation, active line loss, and reward. A separate section prints the MoE controller's mean expert-0 gate weight.
- Save `swarm_voltage_control.py`.
- Check the reporting function by running inspection mode:
.\.venv\Scripts\python.exe swarm_voltage_control.py --inspectWhat Does the Output Confirm?
You should still see the complete inspection output for all three agents. This confirms that the formatted strings in `print_results()` parse correctly.
Formatted String Causing an Error?
Check the quotation marks around each metric key. Confirm that each continued formatted string remains inside the surrounding `print()` call.
Share the traceback without changing the metric names. Help me fix my print_results function.
The chart uses three side-by-side panels. The first panel compares maximum voltage deviation for the fixed baseline and learned controller.
- Add the first part of `plot_results()` directly below `print_results()` by pasting this code:
def plot_results(regimes, summary, gate_summary):
x_positions = list(range(len(regimes)))
width = 0.36
left = [value - width / 2 for value in x_positions]
right = [value + width / 2 for value in x_positions]
fig, axes = plt.subplots(1, 3, figsize=(14, 4))
axes[0].bar(
left,
[summary["baseline"][regime]["max_deviation"] for regime in regimes],
width=width,
label="Zero Q",
)
axes[0].bar(
right,
[summary["moe"][regime]["max_deviation"] for regime in regimes],
width=width,
label="MoE swarm",
)
axes[0].set_title("Maximum voltage deviation")
axes[0].set_ylabel("Per unit")
axes[0].set_xticks(x_positions, regimes)
axes[0].legend()How Is the First Panel Arranged?
`left` and `right` offset the bars around each regime label. This places the baseline beside the MoE result for direct comparison.
- Save `swarm_voltage_control.py`.
- Confirm the first plotting section parses by running inspection mode:
.\.venv\Scripts\python.exe swarm_voltage_control.py --inspectWhat Does This Run Verify?
The agent inspection should complete normally. That result confirms the first plotting section has valid indentation and balanced brackets.
Plot Function Not Parsing?
Check that both `axes[0].bar()` calls remain indented inside `plot_results()`. Confirm that each list comprehension closes before `width=width`.
Use the first traceback line inside `plot_results()` as your starting point. Help me debug the first chart panel.
The second panel repeats the paired comparison for active line loss. This connects the controller's voltage behavior to another physical cost.
- Inside `plot_results()`, add this code directly below `axes[0].legend()`:
axes[1].bar(
left,
[summary["baseline"][regime]["loss_mw"] for regime in regimes],
width=width,
label="Zero Q",
)
axes[1].bar(
right,
[summary["moe"][regime]["loss_mw"] for regime in regimes],
width=width,
label="MoE swarm",
)
axes[1].set_title("Active line loss")
axes[1].set_ylabel("MW")
axes[1].set_xticks(x_positions, regimes)
axes[1].legend()What Does the Second Panel Show?
The second panel uses the same regime positions and controller offsets. Its bar heights come from the averaged `loss_mw` metric.
- Save `swarm_voltage_control.py`.
- Confirm the second panel parses by running inspection mode:
.\.venv\Scripts\python.exe swarm_voltage_control.py --inspectWhat Should Remain Unchanged?
The existing inspection report should still list every agent under both regimes. This confirms that the second chart panel has not introduced a syntax error.
Second Panel Breaking the Script?
Confirm that every line starts with four spaces because it belongs inside `plot_results()`. Check that `loss_mw` matches the metric key returned by `RadialFeeder.step()`.
Keep the two bar calls separate while debugging. Help me fix the active-line-loss panel.
The third panel visualizes the learned gate instead of a physical grid metric. Saving happens before the chart window opens, so `results.png` is written even while the window remains visible.
- Inside `plot_results()`, add this final code directly below `axes[1].legend()`:
axes[2].bar(
x_positions,
[gate_summary[regime] for regime in regimes],
)
axes[2].set_title("Expert-0 gate weight")
axes[2].set_ylabel("Mean weight")
axes[2].set_xticks(x_positions, regimes)
axes[2].set_ylim(0.0, 1.0)
fig.savefig("results.png", dpi=160, bbox_inches="tight")
plt.show()How Is Routing Visualized?
Each routing bar shows the average weight assigned to expert 0 across all agents and evaluation samples. The fixed vertical range makes weights directly comparable between regimes.
`fig.savefig()` writes the three-panel figure to `results.png`. `plt.show()` displays the same figure for immediate inspection.
- Save `swarm_voltage_control.py`.
- Confirm the complete plotting function parses by running inspection mode:
.\.venv\Scripts\python.exe swarm_voltage_control.py --inspectWhat Does This Final Syntax Check Prove?
You should see the inspection output complete without a traceback. The reporting and plotting functions are now ready to receive benchmark results.
Final Plot Section Failing?
Check that `fig.savefig()` and `plt.show()` remain inside `plot_results()`. Confirm that the output filename is exactly `results.png`.
Check the traceback before changing any plotting values. Help me debug the completed plot_results function.
Run the complete experiment
The final workflow trains fresh policies before benchmarking them. The default run evaluates only the two regimes used during training.
- In `parse_args()`, find `parser.add_argument("--train-only", action="store_true")`.
- Add this argument directly below it:
parser.add_argument("--stress-test", action="store_true")Why Keep an Optional Evaluation Switch?
The evaluator accepts an optional request for an additional regime. The default benchmark leaves that option disabled, so this step still compares only `high_solar` and `high_load`.
- Save `swarm_voltage_control.py`.
- Confirm the parser change by running inspection mode:
.\.venv\Scripts\python.exe swarm_voltage_control.py --inspectWhat Does This Parser Check Show?
The script should recognize the existing inspection flag and print the policy details. This confirms that the new parser argument did not disrupt the current modes.
Argument Parser Rejecting the Command?
Check that the new line stays inside `parse_args()`. Confirm that it appears before `return parser.parse_args()`.
Compare the quotation marks and argument name with the snippet. Help me fix my parser change.
The default path currently stops after training. The final block passes the trained policies into evaluation, prints the summaries, and produces the chart.
- In `main()`, find the `if args.train_only:` block.
- Add this workflow directly below its `return` line:
regimes, summary, gate_summary = evaluate(
env,
policies,
include_mixed=args.stress_test,
)
print_results(regimes, summary, gate_summary)
plot_results(regimes, summary, gate_summary)How Does the Default Workflow Finish?
`evaluate()` generates the matched summaries after training. `print_results()` writes the benchmark to the terminal. `plot_results()` saves and displays the three-panel figure.
- Save `swarm_voltage_control.py`.
Before you run the full experiment, which controller do you expect to have the lower voltage deviation in each regime?
- Run the complete training and benchmark workflow with this command:
.\.venv\Scripts\python.exe swarm_voltage_control.pyWhat Should You See?
The terminal prints a mean reward every 50 episodes. It then prints baseline and MoE rows for `high_solar` and `high_load`.
A chart window displays maximum voltage deviation, active line loss, and expert-0 gate weight. The same figure is saved as `results.png`.
That closes the experimental loop. You now have physical evidence for how the learned swarm behaves beside the fixed controller.
Benchmark Not Completing?
If training finishes without a benchmark table, confirm that the new workflow sits outside the `if args.train_only:` block. It should align with that `if` statement.
If the chart window does not open, check the project file list for `results.png`. The script saves the figure before asking Matplotlib to display it.
Use the final traceback line to identify whether evaluation, reporting, or plotting failed. Help me debug the full benchmark run.
- Compare the baseline and MoE rows for maximum voltage deviation.
- Compare the same rows for voltage-band violation.
- Compare the same rows for active line loss.
- Inspect how the expert-0 gate weight changes between the two regimes.
Interpret the Evidence
Evaluation is stochastic because training samples actions from learned distributions. Treat the table and chart as evidence from one reproducible run. Focus on whether the physical metrics and routing behavior support the controller design.
✔️ Awesome, I've got everything!
Great work. Confirm that the terminal contains four controller-regime rows and that `results.png` is saved in your project folder.
ⓧ I'd like to double check the full code
Compare your complete `swarm_voltage_control.py` file with this reference:
import argparse
import random
from statistics import fmean
import matplotlib.pyplot as plt
import torch
from torch import nn
from torch.distributions import Categorical
from pandapower.create import (
create_bus,
create_empty_network,
create_ext_grid,
create_line_from_parameters,
create_load,
create_sgen,
)
from pandapower.run import runpp
NUM_AGENTS = 3
REGIMES = ("high_solar", "high_load")
Q_ACTIONS = (-0.45, 0.0, 0.45)
TRAIN_EPISODES = 500
EVAL_SAMPLES = 40
VOLTAGE_WEIGHT = 400.0
VIOLATION_WEIGHT = 120.0
LOSS_WEIGHT = 4.0
Q_WEIGHT = 0.05
ROUTING_WEIGHT = 0.25
ENTROPY_WEIGHT = 0.01
DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu")
random.seed(7)
torch.manual_seed(7)
class RadialFeeder:
def __init__(self):
self.net = create_empty_network(sn_mva=1.0)
self.buses = [
create_bus(self.net, vn_kv=10.0, name=f"Bus {index}")
for index in range(NUM_AGENTS + 1)
]
create_ext_grid(self.net, bus=self.buses[0], vm_pu=1.0, name="Grid")
for index in range(NUM_AGENTS):
create_line_from_parameters(
self.net,
from_bus=self.buses[index],
to_bus=self.buses[index + 1],
length_km=2.0,
r_ohm_per_km=0.45,
x_ohm_per_km=0.35,
c_nf_per_km=0.0,
max_i_ka=0.4,
name=f"Line {index + 1}",
)
self.load_ids = []
self.sgen_ids = []
for index, bus in enumerate(self.buses[1:]):
self.load_ids.append(
create_load(
self.net,
bus=bus,
p_mw=0.5,
q_mvar=0.15,
name=f"Load {index + 1}",
)
)
self.sgen_ids.append(
create_sgen(
self.net,
bus=bus,
p_mw=0.5,
q_mvar=0.0,
name=f"PV {index + 1}",
)
)
self.loads = [0.0] * NUM_AGENTS
self.solar = [0.0] * NUM_AGENTS
def _sample_operating_point(self, regime):
if regime == "high_solar":
loads = [random.uniform(0.35, 0.55) for _ in range(NUM_AGENTS)]
solar = [random.uniform(0.95, 1.25) for _ in range(NUM_AGENTS)]
elif regime == "high_load":
loads = [random.uniform(0.85, 1.15) for _ in range(NUM_AGENTS)]
solar = [random.uniform(0.05, 0.25) for _ in range(NUM_AGENTS)]
elif regime == "mixed":
loads = [random.uniform(0.55, 0.95) for _ in range(NUM_AGENTS)]
solar = [random.uniform(0.45, 0.95) for _ in range(NUM_AGENTS)]
else:
raise ValueError(f"Unknown regime: {regime}")
return loads, solar
def _solve(self):
runpp(
self.net,
algorithm="bfsw",
calculate_voltage_angles=False,
init="flat",
)
def reset(self, regime):
self.loads, self.solar = self._sample_operating_point(regime)
for index in range(NUM_AGENTS):
load_id = self.load_ids[index]
sgen_id = self.sgen_ids[index]
self.net.load.at[load_id, "p_mw"] = self.loads[index]
self.net.load.at[load_id, "q_mvar"] = 0.32 * self.loads[index]
self.net.sgen.at[sgen_id, "p_mw"] = self.solar[index]
self.net.sgen.at[sgen_id, "q_mvar"] = 0.0
self._solve()
observations = []
for index, bus in enumerate(self.buses[1:]):
voltage = float(self.net.res_bus.at[bus, "vm_pu"])
observations.append(
torch.tensor(
[
20.0 * (voltage - 1.0),
self.loads[index] / 1.5,
self.solar[index] / 1.5,
],
dtype=torch.float32,
device=DEVICE,
)
)
return observations
def step(self, action_ids):
q_setpoints = [Q_ACTIONS[action_id] for action_id in action_ids]
for sgen_id, q_mvar in zip(self.sgen_ids, q_setpoints):
self.net.sgen.at[sgen_id, "q_mvar"] = q_mvar
self._solve()
voltages = [
float(self.net.res_bus.at[bus, "vm_pu"])
for bus in self.buses[1:]
]
line_loss_mw = float(self.net.res_line["pl_mw"].sum())
voltage_squared_error = sum((voltage - 1.0) ** 2 for voltage in voltages)
violation_pu = sum(
max(0.0, 0.95 - voltage) + max(0.0, voltage - 1.05)
for voltage in voltages
)
q_effort = sum(abs(q_mvar) for q_mvar in q_setpoints)
reward = -(
VOLTAGE_WEIGHT * voltage_squared_error
+ VIOLATION_WEIGHT * violation_pu
+ LOSS_WEIGHT * line_loss_mw
+ Q_WEIGHT * q_effort
)
metrics = {
"min_voltage": min(voltages),
"max_voltage": max(voltages),
"max_deviation": max(abs(voltage - 1.0) for voltage in voltages),
"violation_pu": violation_pu,
"loss_mw": line_loss_mw,
"reward": reward,
}
return reward, metrics
class MoEPolicy(nn.Module):
def __init__(self, observation_size=3, hidden_size=16, action_count=3):
super().__init__()
self.gate = nn.Sequential(
nn.Linear(observation_size, hidden_size),
nn.Tanh(),
nn.Linear(hidden_size, 2),
)
self.experts = nn.ModuleList(
[
nn.Sequential(
nn.Linear(observation_size, hidden_size),
nn.Tanh(),
nn.Linear(hidden_size, action_count),
)
for _ in range(2)
]
)
def forward(self, observation):
gate_weights = torch.softmax(self.gate(observation), dim=-1)
expert_logits = torch.stack(
[expert(observation) for expert in self.experts], dim=0
)
mixed_logits = (gate_weights.unsqueeze(-1) * expert_logits).sum(dim=0)
return mixed_logits, gate_weights
def build_policies():
return nn.ModuleList([MoEPolicy() for _ in range(NUM_AGENTS)]).to(DEVICE)
def baseline_demo(env):
random.seed(21)
print("Zero-reactive-power baseline")
for regime in REGIMES:
env.reset(regime)
_, metrics = env.step([1] * NUM_AGENTS)
print(
f"{regime:>10} | voltage {metrics['min_voltage']:.4f} to "
f"{metrics['max_voltage']:.4f} pu | loss {metrics['loss_mw']:.4f} MW | "
f"violation {metrics['violation_pu']:.4f} pu"
)
def inspect_policies(env, policies):
random.seed(22)
for regime in REGIMES:
observations = env.reset(regime)
print(f"\n{regime}")
with torch.no_grad():
for agent_index, observation in enumerate(observations):
logits, gate_weights = policies[agent_index](observation)
action_probabilities = torch.softmax(logits, dim=-1)
print(
f"agent {agent_index + 1}: gate="
f"{gate_weights.detach().cpu().tolist()} actions="
f"{action_probabilities.detach().cpu().tolist()}"
)
def train(env, policies):
optimizer = torch.optim.Adam(policies.parameters(), lr=0.01)
reward_baseline = 0.0
reward_history = []
for episode in range(1, TRAIN_EPISODES + 1):
regime = REGIMES[(episode - 1) % len(REGIMES)]
observations = env.reset(regime)
actions = []
log_probabilities = []
entropies = []
routing_losses = []
target_expert = 0 if regime == "high_solar" else 1
for agent_index, observation in enumerate(observations):
logits, gate_weights = policies[agent_index](observation)
distribution = Categorical(logits=logits)
action = distribution.sample()
actions.append(action.item())
log_probabilities.append(distribution.log_prob(action))
entropies.append(distribution.entropy())
routing_losses.append((gate_weights[target_expert] - 1.0) ** 2)
reward, _ = env.step(actions)
advantage = reward - reward_baseline
reward_baseline = 0.95 * reward_baseline + 0.05 * reward
policy_loss = -torch.stack(log_probabilities).sum() * advantage
entropy_bonus = torch.stack(entropies).sum()
routing_loss = torch.stack(routing_losses).mean()
loss = (
policy_loss
+ ROUTING_WEIGHT * routing_loss
- ENTROPY_WEIGHT * entropy_bonus
)
optimizer.zero_grad()
loss.backward()
optimizer.step()
reward_history.append(reward)
if episode % 50 == 0:
print(
f"episode {episode:>3} | mean reward "
f"{fmean(reward_history[-50:]):.4f}"
)
return reward_history
def greedy_actions(policies, observations):
actions = []
expert_zero_weights = []
with torch.no_grad():
for agent_index, observation in enumerate(observations):
logits, gate_weights = policies[agent_index](observation)
action_values = logits.detach().cpu().tolist()
action = max(
range(len(action_values)),
key=action_values.__getitem__,
)
actions.append(action)
expert_zero_weights.append(float(gate_weights[0].item()))
return actions, expert_zero_weights
def summarize(rows):
return {
key: fmean(row[key] for row in rows)
for key in rows[0]
}
def evaluate(env, policies, include_mixed=False):
regimes = list(REGIMES)
if include_mixed:
regimes.append("mixed")
summary = {"baseline": {}, "moe": {}}
gate_summary = {}
for controller in ("baseline", "moe"):
random.seed(2026)
for regime in regimes:
rows = []
gate_values = []
for _ in range(EVAL_SAMPLES):
observations = env.reset(regime)
if controller == "baseline":
actions = [1] * NUM_AGENTS
else:
actions, gates = greedy_actions(policies, observations)
gate_values.extend(gates)
_, metrics = env.step(actions)
rows.append(metrics)
summary[controller][regime] = summarize(rows)
if controller == "moe":
gate_summary[regime] = fmean(gate_values)
return regimes, summary, gate_summary
def print_results(regimes, summary, gate_summary):
print("\nController benchmark")
for controller in ("baseline", "moe"):
for regime in regimes:
metrics = summary[controller][regime]
print(
f"{controller:>8} | {regime:>10} | max deviation "
f"{metrics['max_deviation']:.4f} pu | violation "
f"{metrics['violation_pu']:.4f} pu | loss "
f"{metrics['loss_mw']:.4f} MW | reward "
f"{metrics['reward']:.4f}"
)
print("\nMean expert-0 gate weight")
for regime in regimes:
print(f"{regime:>10}: {gate_summary[regime]:.3f}")
def plot_results(regimes, summary, gate_summary):
x_positions = list(range(len(regimes)))
width = 0.36
left = [value - width / 2 for value in x_positions]
right = [value + width / 2 for value in x_positions]
fig, axes = plt.subplots(1, 3, figsize=(14, 4))
axes[0].bar(
left,
[summary["baseline"][regime]["max_deviation"] for regime in regimes],
width=width,
label="Zero Q",
)
axes[0].bar(
right,
[summary["moe"][regime]["max_deviation"] for regime in regimes],
width=width,
label="MoE swarm",
)
axes[0].set_title("Maximum voltage deviation")
axes[0].set_ylabel("Per unit")
axes[0].set_xticks(x_positions, regimes)
axes[0].legend()
axes[1].bar(
left,
[summary["baseline"][regime]["loss_mw"] for regime in regimes],
width=width,
label="Zero Q",
)
axes[1].bar(
right,
[summary["moe"][regime]["loss_mw"] for regime in regimes],
width=width,
label="MoE swarm",
)
axes[1].set_title("Active line loss")
axes[1].set_ylabel("MW")
axes[1].set_xticks(x_positions, regimes)
axes[1].legend()
axes[2].bar(
x_positions,
[gate_summary[regime] for regime in regimes],
)
axes[2].set_title("Expert-0 gate weight")
axes[2].set_ylabel("Mean weight")
axes[2].set_xticks(x_positions, regimes)
axes[2].set_ylim(0.0, 1.0)
fig.savefig("results.png", dpi=160, bbox_inches="tight")
plt.show()
def parse_args():
parser = argparse.ArgumentParser()
parser.add_argument("--baseline", action="store_true")
parser.add_argument("--inspect", action="store_true")
parser.add_argument("--train-only", action="store_true")
parser.add_argument("--stress-test", action="store_true")
return parser.parse_args()
def main():
args = parse_args()
env = RadialFeeder()
print(f"Training device: {DEVICE}")
if args.baseline:
baseline_demo(env)
return
policies = build_policies()
if args.inspect:
inspect_policies(env, policies)
return
train(env, policies)
if args.train_only:
return
regimes, summary, gate_summary = evaluate(
env,
policies,
include_mixed=args.stress_test,
)
print_results(regimes, summary, gate_summary)
plot_results(regimes, summary, gate_summary)
if __name__ == "__main__":
main()What Does This Reference Include?
This is the cumulative feeder, baseline, MoE policy, training, evaluation, reporting, plotting, parser, and main workflow. Use it to compare structure and indentation without replacing working sections blindly.
Secret mission
Stress-Test an Unseen Mixed Regime
Your policies learned from two operating extremes. Stress-test them on a mixed solar-load regime to measure physical performance and inspect whether the learned gates interpolate between experts.
Clean Up Your Resources
Clean Up Your Resources
Your experiment runs entirely on your Windows machine, so keeping its files creates no ongoing cost. Choose whether to keep the experiment available, pause your work, or remove its local resources.
Resources you used:
- Local .venv virtual environment containing the installed Python packages (no cost).
- Local requirements.txt dependency file containing the pinned package versions (no cost).
- Local swarm_voltage_control.py experiment script containing the feeder, policies, training loop, evaluation, and stress test (no cost).
- Local results.png benchmark chart containing the trained regimes and unseen mixed regime (no cost).
Keep everything running
No action needed. Choose this if you want to rerun the benchmark or continue investigating the trained swarm.
- Your .venv environment remains ready with the pinned packages installed.
- Your requirements.txt file preserves the package versions used for the experiment.
- Your swarm_voltage_control.py script remains available for future training runs.
- Your results.png chart remains available as evidence from the mixed-regime stress test.
Pause - I'll come back to this later
There are no cloud services to pause. A completed training run stops after you close the chart window.
- Close the open Matplotlib chart window if it is still visible.
- Leave the project resources in the currently open folder for your next session.
Delete - I don't want to use this again
Removing these files clears the experiment from your current project folder. Your Python installation remains available for future projects.
- Close any open Matplotlib chart window.
- Return to the file tree in Visual Studio Code from earlier.
- Select .venv, requirements.txt, swarm_voltage_control.py, and results.png together in the file tree.
- Use the selection's context menu to delete all four resources.
- Confirm the deletion if Windows asks for confirmation.
- Verify that none of the four resources remains in the file tree.
Nice Work!
Nice Work!
You made it! Your Volt/VAR swarm now coordinates three local mixture-of-experts policies through balanced AC power flow.
You've learned how to:
- Built a balanced radial feeder with three inverter control locations. Used a zero-reactive-power baseline to expose regime-specific voltage ranges plus active line losses.
- Implemented three decentralized MoE policies with learned softmax routing. Trained them from local observations using one shared cooperative reward.
- Produced benchmark evidence across matched high-solar plus high-load samples. Generated results.png to compare voltage deviation, active line loss, plus expert routing.
- Completed the Secret Mission by stress-testing the trained swarm on an unseen mixed regime. Compared its physical metrics plus gate behavior with both trained regimes.
Ready to quiz yourself?