By Deepak Bagada, CEO at SaaSNext & Principal AI Architect.

Build a Power-Aware AI Workload Scheduler with LangGraph

In August 2026, the single biggest constraint on the AI industry is no longer GPUs. It is megawatts. When Brookfield and NextEra unveiled their $100 billion AI campus in Kentucky — a buildout anchored by 2 GW of new gas-fired generation and 2.6 GW of grid-scale batteries — they were betting on something most engineering teams still haven't internalized: in the agent era, compute follows power, not the other way around.

That flips your scheduling mental model. A training run that costs ₹14 lakh at the 6 p.m. peak can cost ₹4.5 lakh in the 3 a.m. trough — or run at near-zero carbon intensity, or both. A batch inference pipeline can be shifted by an hour to ride a solar/wind dip on the ISO's real-time curve. And a rogue agent swarm that trips a facility capacity alarm should be killed at the breaker, not after a human reads a Slack alert.

This guide builds a production-grade Power-Aware AI Workload Scheduling workflow in LangGraph: a stateful graph that watches live energy prices and grid carbon intensity, matches them against your compute inventory, optimizes cost, holds workloads for cheap windows, gates expensive or dirty runs behind human approval, and — most importantly — carries a power-shutdown safety rail that brings training down safely when the grid says no. The design is deliberately grounded in the operational reality of the Kentucky build: firm gas when you need it, batteries to shave the spikes, and an orchestration layer that treats energy as a first-class scheduling resource, not an afterthought.

If you're newer to multi-agent orchestration, start at our AI workflows hub, and for the MCP tool servers that feed these price and carbon watchers, see the MCP directory. For what changes week to week in the data-center buildout race, our latest AI news page is the place to keep up.

Why Energy Is the New Scheduling Currency

Three facts define the energy ceiling on the agent era in 2026:

1. Prices swing like crypto. Wholesale grid prices in many ISOs swing 8-12x between peak and trough. On a windy night, real-time prices can even go negative — the grid pays you to draw power.
2. Carbon intensity is time-dependent. On the same grid, marginal CO₂ per kWh can drop below 100 g/kWh on sunny afternoons and spike past 700 g/kWh on peak evening ramps as gas peakers fire up.
3. Interruptible capacity is now a product. Facilities like the Kentucky campus sell interruptible power at a discount: you get cheap firm-ish energy, but you must curtail instantly when the facility or grid demands. A workload orchestrator that cannot shed load on demand is disqualifying — and the largest fleets now hard-require it.

The result: "when" you run a workload has become as important as "what" and "where." A scheduler that is power-blind is leaving 40-60% of energy cost — and most of the carbon debt — on the table.

The Workflow at a Glance

flowchart TD
    A[START] --> B[energy_price_watcher]
    B --> C[carbon_intensity_fetcher]
    C --> D[inventory_reader]
    D --> E[cost_optimizer_router]
    E --"best window within budget + carbon OK"--> F[schedule_release]
    E --"cheap window available later"--> G[hold_for_cheap_window]
    E --"high cost or high carbon"--> H[approval_gate]
    H --"approved"--> F
    H --"denied / pending"--> I[escalate_to_human]
    F --> J{power_shutdown_rail}
    I --> K[HOLD_STATE]
    G --> K
    J --"curtailment demanded"--> L[graceful_curtail]
    J --"capacity OK"--> M[END]
    L --> M
    K --"re-enter on next cycle"--> B

The graph is a superstep loop: it never "finishes" so much as it parks workloads in a hold state and re-enters on the next signal tick (every 5-15 minutes in production). This is the same pattern utility-scale fleets use to decide which of hundreds of queued jobs run this hour.

What the Graph Tracks

The scheduler is only as good as its state. We model three signal planes:

Every workload carries three scheduling metadata fields: flexibility (can it start up to N hours late without hurting the business?), criticality (can it be preempted?), and carbon budget (max gCO₂eq/kWh this run may tolerate). The cost-optimization router turns those into a go/hold/gate decision.

Building Node by Node

Node 1: energy_price_watcher

Polls the wholesale price feed, normalizes to a local currency and a target duration, and attaches the 24-hour forecast curve. The key detail: the watcher writes both the current price and the forecast so the optimizer can look forward, not just at the instant.

Node 2: carbon_intensity_fetcher

Polls marginal carbon intensity. This node matters more than most teams think: corporate reporting, ESG attestations, and EU-style disclosure rules increasingly require actual, timestamped intensity — not annual averages. The fetcher also returns the forecast so the scheduler can wait for a cleaner window.

Node 3: inventory_reader

Queries the compute inventory and facility state: free GPU capacity, total facility headroom, current facility load, and battery state of charge (if your site has storage). On the Kentucky-style campus, battery SoC is a real scheduling input: run from battery during the price spike, recharge during the trough.

Node 4: cost_optimizer_router

The decision core. It evaluates every candidate start time over the forecast horizon, computes projected energy + carbon cost per candidate window, and returns the optimal plan. Routing rules:

• If the optimal window starts now and projected cost is under budget and carbon is under budget → release the workload.
• If a materially cheaper/cleaner window exists within the workload's flexibility → hold for that window.
• If cost or carbon exceeds the workload's budget ceiling → route to the approval gate (never auto-run an over-budget job).

Node 5: approval_gate

A human-in-the-loop gate. Any workload whose projected run exceeds the configured cost or carbon ceiling pauses and emits a structured approval request (with the exact ₹/carbon figures and the alternative windows). No code path releases a job past this gate without an explicit approved decision in state.

Node 6: power_shutdown_rail

The safety rail, checked after release and continuously while the job runs. It consumes facility curtailment signals, battery SoC floor, and grid emergency flags. If any trip condition fires, it does not hard-kill: it runs a graceful curtail that checkpoints training, drains in-flight inference queues, and parks the workload in the hold state. The rail guarantees the campus operator's interruptible-power contract — curtail within seconds — without destroying your training state.

The Multi-File Implementation

A production deployment splits cleanly across five files. This is the exact shape we run for batch training and inference fleets.

# .env — keep this out of git, one env per region/site
ENERGY_PRICE_URL=https://api.hubprice.example/v1/market/rt
ENERGY_PRICE_API_KEY=phk_live_xxxxx
CARBON_URL=https://api.carbonintensity.example/v1/intensity
CARBON_API_KEY=wtt_live_xxxxx
INVENTORY_URL=http://cluster-mgmt.internal:9100/api/inventory
FACILITY_URL=http://scada-gw.internal:9200/api/load

CURRENCY=INR
FACILITY_MAX_MW=200
BATTERY_MW=2600
CURTAIL_LATENCY_S=10

APPROVER_SLACK_CHANNEL=#power-approvals
BUDGET_INR_PER_HOUR=150000
CARBON_CAP_GCO2=450
HOLD_REENTRY_SECONDS=300
MAX_APPROVAL_WAIT_SECONDS=3600

# ============================================================
# schemas.py
# ============================================================
from __future__ import annotations
import datetime as dt
from enum import Enum
from typing import Optional
from pydantic import BaseModel, Field


class WorkloadKind(str, Enum):
    TRAINING = "training"
    BATCH_INFERENCE = "batch_inference"
    EVALUATION = "evaluation"


class Workload(BaseModel):
    """A unit of compute the scheduler may place in time."""
    id: str
    kind: WorkloadKind
    gpu_hours: int
    gpu_type: str = "H100"
    est_power_kw: float = Field(..., gt=0)
    flexibility_hours: int = Field(2, ge=0)   # how late it may start
    criticality: str = Field("normal", pattern="^(normal|high|preemptible)$")
    max_cost_inr: float = Field(..., gt=0)
    max_carbon_gco2: int = Field(600, ge=0)


class EnergySignal(BaseModel):
    fetched_at: dt.datetime
    price_inr_per_mwh: float
    forecast: list[dict] = Field(default_factory=list)  # [{ts, price_inr_per_mwh}]


class CarbonSignal(BaseModel):
    fetched_at: dt.datetime
    intensity_gco2_per_kwh: float
    forecast: list[dict] = Field(default_factory=list)


class ComputeInventory(BaseModel):
    free_gpus: int
    facility_load_mw: float
    battery_soc_pct: float
    curtail_demanded: bool = False


class CostProjection(BaseModel):
    best_window: dict            # {start_ts, end_ts, price, carbon}
    projected_cost_inr: float
    projected_carbon: float
    savings_vs_now_pct: float


class SchedulerState(BaseModel):
    workload: Workload
    energy: Optional[EnergySignal] = None
    carbon: Optional[CarbonSignal] = None
    inventory: Optional[ComputeInventory] = None
    projection: Optional[CostProjection] = None
    decision: Optional[str] = None       # release | hold | gate | escalate
    approval: Optional[str] = None       # approved | denied | pending
    hold_until: Optional[dt.datetime] = None
    last_error: Optional[str] = None
    attempts: int = 0

# ============================================================
# tools.py — all external I/O, every call is idempotent + retryable
# ============================================================
from __future__ import annotations
import asyncio, os, time
import httpx
from schemas import EnergySignal, CarbonSignal, ComputeInventory

TIMEOUT_S = float(os.getenv("TOOL_TIMEOUT_S", "15"))

def backoff_delay(attempt: int) -> float:
    """Exponential backoff with jitter: 1,2,4,8,16… capped at 60s."""
    import random
    base = min(2 ** attempt, 60.0)
    return base * (0.5 + random.random() / 2)


async def with_retry(fn, *, retries: int = 4, timeout: float = TIMEOUT_S):
    """Run fn with exponential backoff; last resort returns a fallback signal."""
    for attempt in range(retries):
        try:
            async with httpx.AsyncClient(timeout=timeout) as client:
                return await fn(client)
        except (httpx.TimeoutException, httpx.ConnectError) as exc:
            if attempt == retries - 1:
                return None  # caller falls back to cached/last-known signal
            await asyncio.sleep(backoff_delay(attempt))
    return None


async def fetch_energy_price() -> EnergySignal | None:
    async def _fetch(client: httpx.AsyncClient) -> EnergySignal:
        resp = await client.get(
            os.environ["ENERGY_PRICE_URL"],
            headers={"Authorization": f"Bearer {os.environ['ENERGY_PRICE_API_KEY']}"},
            params={"currency": os.environ.get("CURRENCY", "INR")},
        )
        resp.raise_for_status()
        payload = resp.json()
        return EnergySignal(
            price_inr_per_mwh=float(payload["current"]["price_inr_per_mwh"]),
            forecast=payload.get("forecast", []),
        )

    return await with_retry(_fetch)


async def fetch_carbon_intensity() -> CarbonSignal | None:
    async def _fetch(client: httpx.AsyncClient) -> CarbonSignal:
        resp = await client.get(
            os.environ["CARBON_URL"],
            headers={"Authorization": f"Bearer {os.environ['CARBON_API_KEY']}"},
        )
        resp.raise_for_status()
        payload = resp.json()
        return CarbonSignal(
            intensity_gco2_per_kwh=float(payload["data"]["intensity"]["actual"]),
            forecast=payload["data"].get("forecast", []),
        )

    return await with_retry(_fetch)


async def fetch_inventory() -> ComputeInventory | None:
    async def _fetch(client: httpx.AsyncClient) -> ComputeInventory:
        inv = await client.get(os.environ["INVENTORY_URL"])
        fac = await client.get(os.environ["FACILITY_URL"])
        inv.raise_for_status(); fac.raise_for_status()
        i, f = inv.json(), fac.json()
        return ComputeInventory(
            free_gpus=int(i["free_gpus"]),
            facility_load_mw=float(f["load_mw"]),
            battery_soc_pct=float(f.get("battery_soc_pct", 100.0)),
            curtail_demanded=bool(f.get("curtail_demanded", False)),
        )

    return await with_retry(_fetch)

# ============================================================
# graph.py — the LangGraph state machine
# ============================================================
from __future__ import annotations
import os, datetime as dt
from typing import Optional
from langgraph.graph import StateGraph, START, END
from schemas import SchedulerState, CostProjection
import tools

BUDGET_INR = float(os.getenv("BUDGET_INR_PER_HOUR", "150000"))
CARBON_CAP = int(os.getenv("CARBON_CAP_GCO2", "450"))


async def energy_price_watcher(state: SchedulerState) -> dict:
    state.attempts += 1
    signal = await tools.fetch_energy_price()
    if signal is None:
        signal = state.energy  # fallback to last known — never block the graph
        if signal is None:
            return {"decision": "escalate", "last_error": "energy feed down, no cache"}
    return {"energy": signal}


async def carbon_intensity_fetcher(state: SchedulerState) -> dict:
    signal = await tools.fetch_carbon_intensity()
    if signal is None:
        signal = state.carbon
        if signal is None:
            return {"decision": "escalate", "last_error": "carbon feed down, no cache"}
    return {"carbon": signal}


async def inventory_reader(state: SchedulerState) -> dict:
    inv = await tools.fetch_inventory()
    if inv is None:
        inv = state.inventory
        if inv is None:
            return {"decision": "escalate", "last_error": "inventory down"}
    if inv.curtail_demanded:
        return {"inventory": inv, "decision": "curtail"}
    return {"inventory": inv}


def score_window(w: dict, wl) -> float:
    """Lower is better. Cost dominates, carbon penalizes past cap."""
    cost = float(w["price_inr_per_mwh"]) * wl.est_power_kw * wl.gpu_hours / 1000.0
    carbon_penalty = max(0.0, float(w.get("carbon_gco2", 0)) - wl.max_carbon_gco2) / 100.0
    return cost + carbon_penalty


async def cost_optimizer_router(state: SchedulerState) -> dict:
    wl, energy, carbon = state.workload, state.energy, state.carbon
    if energy is None or carbon is None:
        return {"decision": "escalate", "last_error": "missing signals"}

    now = dt.datetime.utcnow()
    candidates = [{"ts": now, "price_inr_per_mwh": energy.price_inr_per_mwh,
                   "carbon_gco2": carbon.intensity_gco2_per_kwh}]
    for fc in energy.forecast[: max(wl.flexibility_hours, 1) * 4]:
        candidates.append({"ts": fc["ts"], "price_inr_per_mwh": fc["price_inr_per_mwh"],
                           "carbon_gco2": fc.get("carbon_gco2", carbon.intensity_gco2_per_kwh)})

    best = min(candidates, key=lambda c: score_window(c, wl))
    cost = best["price_inr_per_mwh"] * wl.est_power_kw * wl.gpu_hours / 1000.0
    ghg = best.get("carbon_gco2", carbon.intensity_gco2_per_kwh)

    projection = CostProjection(
        best_window=best,
        projected_cost_inr=cost,
        projected_carbon=ghg,
        savings_vs_now_pct=max(0.0, (energy.price_inr_per_mwh - best["price_inr_per_mwh"])
                               / energy.price_inr_per_mwh * 100),
    )

    if cost <= BUDGET_INR and ghg <= CARBON_CAP and best["ts"] <= now:
        return {"projection": projection, "decision": "release"}
    if best["ts"] > now and cost <= BUDGET_INR * 1.15:
        return {"projection": projection, "decision": "hold",
                "hold_until": best["ts"]}
    return {"projection": projection, "decision": "gate"}


async def approval_gate(state: SchedulerState) -> dict:
    # In production this posts to Slack/Teams and awaits a structured reply.
    if state.approval == "approved":
        return {"decision": "release"}
    if state.approval == "denied":
        return {"decision": "escalate", "last_error": "approval denied by human"}
    return {"decision": "escalate"}  # pending -> keep parked, do not release


async def schedule_release(state: SchedulerState) -> dict:
    print(f"[release] workload={state.workload.id} "
          f"cost=₹{state.projection.projected_cost_inr:,.0f} "
          f"carbon={state.projection.projected_carbon} gCO2/kWh")
    return {"decision": "released"}


async def hold_for_cheap_window(state: SchedulerState) -> dict:
    print(f"[hold] workload={state.workload.id} until {state.hold_until} — "
          f"saves {state.projection.savings_vs_now_pct:.0f}%")
    return {"decision": "held"}


async def power_shutdown_rail(state: SchedulerState) -> dict:
    if state.inventory and state.inventory.curtail_demanded:
        print(f"[curtail] {state.workload.id}: checkpointing + draining queues")
        return {"decision": "curtailed"}
    return {"decision": "ok"}


async def escalate_to_human(state: SchedulerState) -> dict:
    print(f"[escalate] {state.workload.id}: {state.last_error}")
    return {"decision": "escalated"}


def route_after_optimizer(state: SchedulerState) -> str:
    return {
        "release": "schedule_release",
        "hold": "hold_for_cheap_window",
        "gate": "approval_gate",
        "curtail": "power_shutdown_rail",
        "escalate": "escalate_to_human",
    }.get(state.decision, "escalate_to_human")


builder = StateGraph(SchedulerState)
builder.add_node("energy_price_watcher", energy_price_watcher)
builder.add_node("carbon_intensity_fetcher", carbon_intensity_fetcher)
builder.add_node("inventory_reader", inventory_reader)
builder.add_node("cost_optimizer_router", cost_optimizer_router)
builder.add_node("approval_gate", approval_gate)
builder.add_node("schedule_release", schedule_release)
builder.add_node("hold_for_cheap_window", hold_for_cheap_window)
builder.add_node("power_shutdown_rail", power_shutdown_rail)
builder.add_node("escalate_to_human", escalate_to_human)

builder.add_edge(START, "energy_price_watcher")
builder.add_edge("energy_price_watcher", "carbon_intensity_fetcher")
builder.add_edge("carbon_intensity_fetcher", "inventory_reader")
builder.add_edge("inventory_reader", "cost_optimizer_router")
builder.add_conditional_edges("cost_optimizer_router", route_after_optimizer)
builder.add_edge("approval_gate", "schedule_release")      # only when approved
builder.add_edge("schedule_release", "power_shutdown_rail")
builder.add_edge("power_shutdown_rail", END)
builder.add_edge("hold_for_cheap_window", END)             # parked; re-enter next tick
builder.add_edge("escalate_to_human", END)

graph = builder.compile()

# ============================================================
# main.py — the long-running scheduler loop
# ============================================================
from __future__ import annotations
import asyncio, os
from schemas import SchedulerState, Workload, WorkloadKind
from graph import graph

HOLD_REENTRY_S = int(os.getenv("HOLD_REENTRY_SECONDS", "300"))


async def run_scheduler(state: SchedulerState) -> None:
    """Re-enter the graph every tick. Held workloads loop; released ones exit."""
    while True:
        result = await graph.ainvoke(state.model_dump())
        decision = result.get("decision")
        if decision in ("released", "curtailed"):
            return
        if decision == "escalated" and state.attempts > 5:
            return  # give up to a human queue
        await asyncio.sleep(HOLD_REENTRY_S)


async def main() -> None:
    workload = Workload(
        id="fine-tune-llama-ckpt-17",
        kind=WorkloadKind.TRAINING,
        gpu_hours=240,
        est_power_kw=48.0,
        flexibility_hours=6,
        max_cost_inr=1_200_000,
        max_carbon_gco2=350,
    )
    await run_scheduler(SchedulerState(workload=workload))
    print("scheduler cycle complete")


if __name__ == "__main__":
    asyncio.run(main())

Retry Rules & Error Handling

A power scheduler cannot silently drop signals, and it cannot guess when a feed is lying. The retry posture is:

The three rules that matter: node timeouts (every tool call is bounded — a hung price feed must not hold a training fleet hostage), degraded-mode fallbacks (if a signal is stale but present, the graph runs with the last-known value and stamps a stale: true warning; if the cache is empty, it escalates rather than guess), and no auto-approval on failure (if the approval gate can't reach the approver, the safe default is deny/hold, never run).

Cost & Decision Matrix

The router's practical decision surface, for a sample workload with 6 hours of flexibility:

Running This on a Kentucky-Style Campus

Translate the graph to the Brookfield/NextEra reality and the win multiplies. The 2.6 GW battery bank becomes a first-class inventory input: charge batteries at negative/near-zero price windows, and let the scheduler run workloads from stored energy during price spikes — the inventory_reader already carries battery_soc_pct for exactly this. The 2 GW gas fleet provides firm base capacity for critical, non-preemptible training; everything else rides the interruptible tariff and must survive a curtailment call. The power-shutdown rail is what makes the interruptible contract executable: it's the difference between "cheap power" and "cheap power you can't legally use when it matters."

Set your hold_reentry_seconds to match the ISO's 15-minute settlement interval, budget the carbon cap per run rather than per site, and watch your fleet's ₹/GPU-hour curve flatten — that's the metric every AI CFO in Bangalore and San Francisco is now staring at. When the agent era hits its power ceiling, the teams that treat scheduling as an energy problem, not a queueing problem, are the ones that keep training.