Swarms for Energy

import os
from swarms_client import SwarmsClient

client = SwarmsClient(api_key=os.environ["SWARMS_API_KEY"])

# A GroupChat: three specialists discuss one grid event over several turns.
# The output is the transcript plus each agent's final position, for the
# control room to read. No agent dispatches anything.
event = (
    "17:05 local. Net load ramping at 3.1 GW per hour, about 40 minutes earlier "
    "than the day-ahead forecast. Solar dropping under cloud in the west zone. "
    "Transformer T-14 at the Westgate substation top-oil 92 C, rating 105 C. "
    "Real-time price 3x the day-ahead clear. Import path at 94 percent of limit. "
    "Battery fleet at 88 percent state of charge."
)

response = client.swarms.run(
    name="Evening Ramp Review",
    description="Grid, asset and market agents discuss a ramp event and recommend options.",
    swarm_type="GroupChat",
    task=(
        "Discuss the event below. Agree on what is happening, what it means for "
        "equipment and for price, and list the options the operator could take "
        "with the trade-offs of each. Do not issue instructions; the control room decides.\n\n"
        + event
    ),
    agents=[
        {
            "agent_name": "Grid Ops Analyst",
            "description": "Load, generation, ramps and reserve margins.",
            "system_prompt": (
                "You are a grid operations analyst in a balancing authority. Explain the "
                "ramp, the reserve position and which resources can respond in time. "
                "Speak in short turns and respond to what the other analysts say."
            ),
            "model_name": "gpt-5.4",
            "max_loops": 1,
        },
        {
            "agent_name": "Asset Engineer",
            "description": "Transformer, line and plant thermal limits.",
            "system_prompt": (
                "You are a substation asset engineer. For each piece of equipment named, "
                "state how long it can run at the described loading and what you would "
                "want the operator to avoid. Disagree with the others when the data says so."
            ),
            "model_name": "claude-sonnet-5-5",  # careful with limits and units
            "max_loops": 1,
        },
        {
            "agent_name": "Market Analyst",
            "description": "Real-time and day-ahead prices, imports and bids.",
            "system_prompt": (
                "You are a power market analyst. Explain the price move, whether it is "
                "likely to persist through the ramp, and what each option costs or earns. "
                "Keep every number tied to the data in the event."
            ),
            "model_name": "gemini-2.5-pro",
            "max_loops": 1,
        },
    ],
    max_loops=3,  # three rounds of discussion
)

print(response)

curl -X POST https://api.swarms.world/v1/swarm/completions \
  -H "x-api-key: $SWARMS_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Evening Ramp Review",
    "description": "Grid, asset and market agents discuss a ramp event and recommend options.",
    "swarm_type": "GroupChat",
    "task": "Discuss the event and list the options the operator could take, with trade-offs. Do not issue instructions; the control room decides. Event: 17:05 local, net load ramping 3.1 GW/h, 40 minutes early vs forecast. Solar dropping in the west zone. Transformer T-14 top-oil 92 C, rating 105 C. Real-time price 3x day-ahead. Import path at 94 percent of limit. Battery fleet at 88 percent state of charge.",
    "agents": [
      {
        "agent_name": "Grid Ops Analyst",
        "system_prompt": "You are a grid operations analyst. Explain the ramp, the reserve position and which resources can respond in time. Short turns, respond to the others.",
        "model_name": "gpt-5.4",
        "max_loops": 1
      },
      {
        "agent_name": "Asset Engineer",
        "system_prompt": "You are a substation asset engineer. State how long each named asset can run at the described loading and what the operator should avoid.",
        "model_name": "claude-sonnet-5-5",
        "max_loops": 1
      },
      {
        "agent_name": "Market Analyst",
        "system_prompt": "You are a power market analyst. Explain the price move, whether it persists through the ramp, and what each option costs or earns.",
        "model_name": "gemini-2.5-pro",
        "max_loops": 1
      }
    ],
    "max_loops": 3
  }'

# The response "output" holds the discussion and each agent's final position.
# "usage" has input_tokens, output_tokens and total_tokens for the whole chat.

// Storage dispatch review as a MajorityVoting swarm: three forecasters each
// propose a charge and discharge plan from the same forecast, and the API
// returns the majority plan with the dissent. The storage operator reviews
// it before anything is scheduled.
const SWARMS_API = "https://api.swarms.world/v1/swarm/completions"

type Forecast = { date: string; priceByHour: number[]; netLoadByHour: number[]; socPct: number; capacityMwh: number }

const forecaster = (name: string, model: string, focus: string) => ({
  agent_name: name,
  system_prompt:
    "You plan a grid battery's day. Focus: " + focus + " " +
    "Return a 24-row table of hour, action (CHARGE, HOLD, DISCHARGE) and MW, then two sentences on why. " +
    "The plan is a recommendation for the storage operator, who schedules it.",
  model_name: model,
  max_loops: 1,
})

export async function reviewStoragePlan(f: Forecast) {
  const res = await fetch(SWARMS_API, {
    method: "POST",
    headers: {
      "x-api-key": process.env.SWARMS_API_KEY!,
      "Content-Type": "application/json",
    },
    body: JSON.stringify({
      name: "Storage Dispatch Review",
      description: "Three independent plans for one battery day, majority wins.",
      swarm_type: "MajorityVoting",
      task:
        "Plan " + f.date + " for a " + f.capacityMwh + " MWh battery starting at " + f.socPct + " percent state of charge.\n" +
        "Day-ahead price by hour ($/MWh): " + f.priceByHour.join(", ") + "\n" +
        "Net load forecast by hour (MW): " + f.netLoadByHour.join(", "),
      agents: [
        forecaster("Price Forecaster", "gpt-5.4", "capture the price spread between the midday trough and the evening peak."),
        forecaster("Load Forecaster", "gemini-2.5-pro", "discharge where the net load ramp is steepest, charge where it is flattest."),
        forecaster("Degradation Reviewer", "claude-sonnet-5-5", "keep cycles and depth of discharge inside the warranty limits even if revenue is lower."),
      ],
      max_loops: 1,
    }),
  })

  if (!res.ok) throw new Error("Swarms API " + res.status + ": " + (await res.text()))
  const data = await res.json()
  return { plan: data.output as string, usage: data.usage }
}

const price = [38, 35, 33, 32, 33, 41, 58, 72, 61, 44, 29, 18, 12, 11, 15, 27, 52, 118, 164, 142, 96, 71, 55, 44]
const netLoad = [21, 20, 19.5, 19.2, 19.6, 21.5, 24, 26, 25, 22, 18, 15, 13.5, 13, 14, 17, 23, 29, 31, 30, 27, 25, 23, 22]

reviewStoragePlan({ date: "2026-11-08", priceByHour: price, netLoadByHour: netLoad, socPct: 42, capacityMwh: 400 }).then((r) => {
  console.log(r.plan)
  console.log(r.usage)
})

import os
import requests

BASE = "https://api.swarms.world"
HEADERS = {"x-api-key": os.environ["SWARMS_API_KEY"]}

# After an event, the operations manager wants to see what the agents were
# told, what each one said and what the run cost. Swarms Cloud keeps that for
# every run, and the same data is available over the API.
logs = requests.get(f"{BASE}/v1/swarm/logs", headers=HEADERS, timeout=60).json()
metrics = requests.get(f"{BASE}/v1/metrics/summary", headers=HEADERS, timeout=30).json()


def walk(node, key):
    """Yield every value stored under key anywhere in a nested JSON document."""
    if isinstance(node, dict):
        for k, v in node.items():
            if k == key:
                yield v
            yield from walk(v, key)
    elif isinstance(node, list):
        for item in node:
            yield from walk(item, key)


# Pull out every run of the event review swarm, with its token total, so the
# operations log can reference the run alongside the operator's own entries.
EVENT_SWARM = "Evening Ramp Review"
runs = []
for entry in walk(logs, "data"):
    if not isinstance(entry, dict) or entry.get("name") != EVENT_SWARM:
        continue
    total = sum(int(t) for t in walk(entry, "total_tokens") if isinstance(t, (int, float)))
    runs.append((entry.get("created_at", ""), total, entry))

runs.sort()
print(f"{len(runs)} runs of {EVENT_SWARM!r}")
for created, total, entry in runs[-10:]:
    print(f"{created:>24}  {total:>9,} tokens  swarm_type={entry.get('swarm_type')}")

# Token totals across all swarms on the key, for the monthly operations budget.
by_name: dict[str, int] = {}
for entry in walk(logs, "data"):
    if isinstance(entry, dict):
        total = sum(int(t) for t in walk(entry, "total_tokens") if isinstance(t, (int, float)))
        by_name[entry.get("name", "unnamed")] = by_name.get(entry.get("name", "unnamed"), 0) + total

for name, total in sorted(by_name.items(), key=lambda kv: -kv[1]):
    print(f"{name:32s} {total:>12,} tokens")

print("summary:", metrics)

from swarms import Agent, ConcurrentWorkflow

# The asset health inspection on the open-source swarms package, in your own
# process: four agents read the same fleet snapshot at the same time, each on
# one signal. pip install swarms, then set the provider keys for the models.

vibration = Agent(
    agent_name="Vibration Agent",
    system_prompt=(
        "You read vibration data for rotating plant (turbines, pumps, fans). For "
        "each asset give a rating GREEN, AMBER or RED from the RMS trend and any "
        "spectrum notes, and name the probable failure mode when not GREEN."
    ),
    model_name="gpt-4.1-mini",
    max_loops=1,
)

temperature = Agent(
    agent_name="Temperature Agent",
    system_prompt=(
        "You read winding, bearing and top-oil temperatures against ratings and "
        "ambient. Rate each asset GREEN, AMBER or RED and state the margin to its limit."
    ),
    model_name="claude-haiku-5-5",
    max_loops=1,
)

output = Agent(
    agent_name="Output Agent",
    system_prompt=(
        "You compare each asset's output with its expected output for the "
        "conditions (wind speed, irradiance, dispatch). Rate the shortfall and say "
        "whether it looks like curtailment, soiling, derate or a fault."
    ),
    model_name="gemini-2.5-flash",
    max_loops=1,
)

age = Agent(
    agent_name="Age Agent",
    system_prompt=(
        "You read service age, hours run and maintenance history. Rate each asset "
        "on overdue inspections and end-of-life risk, and list the next due task."
    ),
    model_name="deepseek-chat",
    max_loops=1,
)

inspection = ConcurrentWorkflow(
    name="Fleet Health Inspection",
    description="Four signals, four agents, one snapshot.",
    agents=[vibration, temperature, output, age],
    max_loops=1,
)

with open("fleet_snapshot_2026-11-07.txt", encoding="utf-8") as f:
    snapshot = f.read()

result = inspection.run(
    "Rate every asset in the snapshot below. The maintenance engineer will use "
    "the four ratings to build the inspection list; do not schedule anything.\n\n"
    + snapshot
)

print(result)