Get started with a coding agent

This tutorial shows you how to use a coding agent and RelationalAI skills to turn raw CSV files into a working semantic model you can query and refine.

You’ll learn how to:

• Set up your environment and install the RAI agent skills with a coding agent.
• Build a semantic model from sample CSV files with the rai-build-starter-ontology skill.
• Use the rai-discovery skill to turn a business question into a prompt.
• Run the refined prompt to get real answers grounded in your model and data.

Before you start, make sure that:

• You have access to a Snowflake account with the RelationalAI (RAI) Native App installed.

  How do I know if the RAI Native App is installed?

  To check if the RAI Native App is installed in your Snowflake account:

  1. Log in to your Snowflake account.
  2. Run the following SQL command in a worksheet:

     SHOW APPLICATIONS;

  3. Check the output for an application named RELATIONALAI.
     • If you see it listed, the RAI Native App is installed.
     • If not, you need to install it.

  How do I install the RAI Native App?

  See Install the RAI Native App for Snowflake for instructions.

• You have a Snowflake user with the RAI_DEVELOPER database role.

  How do I check my roles?

  To check your roles in Snowflake:

  1. Log in to your Snowflake account.
  2. Run the following SQL command in a worksheet:

     SHOW GRANTS TO USER <your_username>;

  3. Check the output in the name column for the RAI_DEVELOPER role.
     • If you see it listed, you have the required role.
     • If not, contact your Snowflake account administrator to have the role assigned to you. See Set Up User Access for the RAI Native App for more information.

Install the RAI agent skills

Prompt

Copy and paste the following prompt into your coding agent to install the RAI agent skills for your preferred coding agent:

Install the RelationalAI Agent Skills (https://github.com/RelationalAI/rai-agent-skills) for this environment. Provide me with instructions on how to keep up-to-date and leverage them across projects. Once installed, print the skill inventory given in the README.

Manual steps

Use Vercel’s Skills CLI to install the RAI agent skills for your preferred coding agent:

npx skills add RelationalAI/rai-agent-skills --skill '*'

Tip

Accept the default options to install every skill in an .agents/ directory. This works with most coding agents, including Claude Code, OpenAI, and GitHub Copilot.

Set up your environment

1. Gather your Snowflake connection details.

   You will need:

   Detail	Where to find it
   Account	Log in to the Snowflake account that has the RAI Native App installed. Execute the following SQL command in a worksheet to find your account name: SELECT CURRENT_ORGANIZATION_NAME() || '-' || CURRENT_ACCOUNT_NAME() AS account_identifier;
   User	The Snowflake username you will use to connect. This user must have the RAI_DEVELOPER role.
   Password	The password you use to log in to Snowflake.
   Warehouse	The name of the Snowflake warehouse you want to use to execute SQL queries. Can be any existing X-SMALL or larger warehouse.

   Tip

   For real development work, use a programmatic access token (PAT) instead of username/password authentication. This avoids storing a password in an environment variable and provides better security and access control.

   See Configure programmatic access token authentication for details.

2. Install RAI and set up your project environment.

   Both the agent prompt and manual steps below make use of the uv package manager because it manages project environments and package installation with one tool. This makes it easier to keep your agent’s Python setup isolated from the rest of your machine.

   Prompt

   Copy and paste the following prompt into your coding agent to install RAI and set up your project environment using the rai-setup skill:

   /rai-setup Use uv (install if missing) to create a new project called my-first-rai-project using version of Python supported by RelationalAI. Then use uv to install the relationalai package in this new project environment. Try to connect to Snowflake using the external browser method. Ask me one item at a time for the required connection details. If external browser authentication fails because my Snowflake account doesn't support it, update the configuration to use username/password authentication instead. In that case, use a placeholder for the password and instruct me to edit the configuration file with the real password when you are done.

   What this does

   • Uses uv to install relationalai and create the project environment if needed.
   • Creates or updates raiconfig.yaml with your Snowflake connection settings.
   • Tries to connect to Snowflake using external browser authentication, which opens a browser window for you to log in with your Snowflake credentials.
   • If external browser authentication is not set up for your account, this falls back to username/password authentication and prompts you to edit the config file with your password.
   Manual steps

   1. Install uv if needed.

      Run the following in your terminal to check if you have uv installed:

      uv --version

      If you see a version number, you’re good to go! If you get a “command not found” error, you’ll need to install uv.

      Select the instructions for your operating system below to install uv.

      macOS / Linux

      Use the standalone installer:

      curl -LsSf https://astral.sh/uv/install.sh | sh

      Note

      If you use Homebrew on macOS, you can install it with brew install uv instead.

      Tip

      Accept the default options to install every skill in a .agents/ subfolder, which works with most coding agents, including Claude Code, OpenAI, and GitHub Copilot.

      Windows

      In PowerShell, run:

      powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

      Note

      If you use winget, you can install it with winget install --id=astral-sh.uv -e instead.

   2. Create a new project directory.

      In your terminal, run the following commands to create a new directory for this project and navigate into it:

      uv init my-first-rai-project
      cd my-first-rai-project

   3. Initialize your RAI project.

      Run the following command to create a new raiconfig.yaml in your project directory:

      uv run add relationalai
      uv run rai init

      Open the file and replace the placeholders with your Snowflake connection details. You should end up with something like this:

      # raiconfig.yaml — RelationalAI configuration
      # Uncomment and fill in the sections you need.
      
      active_profile: dev
      
      profile:                              # add more profiles (e.g. staging, prod) as needed
      dev:
          default_connection: snowflake   # change to: duckdb  or  local
      
          connections:
      
          # ── Snowflake ────────────────────────────────────────────────────────────
          snowflake:
              type: snowflake
              account: <MY_SNOWFLAKE_ACCOUNT>      # format: <org>-<account>, e.g. acme-prod123
              warehouse: <MY_SNOWFLAKE_WAREHOUSE>
              rai_app_name: relationalai    # name of the RAI app in your Snowflake account
              authenticator: username_password     # other options: username_password_mfa | externalbrowser | jwt | oauth_authorization_code | oauth | programmatic_access_token
              user: <MY_SNOWFLAKE_USER>
              password: "<MY_SNOWFLAKE_PASSWORD>"     # tip: use env var SNOWFLAKE_PASSWORD instead
              # role: MY_ROLE               # optional
              # database: MY_DATABASE       # optional
              # schema: PUBLIC              # optional
      
          # ── DuckDB ───────────────────────────────────────────────────────────────
          # duckdb:
          #   type: duckdb
          #   path: ":memory:"            # or a file path, e.g. /tmp/mydb.duckdb
      
          # ── Local RAI server ─────────────────────────────────────────────────────
          # local:
          #   type: local
          #   host: localhost
          #   port: 8010
      
          # ── Reasoners (optional) ─────────────────────────────────────────────────
          # reasoners:
          #   logic:
          #     name: my_reasoner           # reasoner instance name (Snowflake)
          #     size: HIGHMEM_X64_S         # reasoner size

3. Verify your connection with rai connect.

   In your terminal, run:

   uv run rai connect

   You should see a success message confirming that you are connected to Snowflake.

Related resources:

• Set up your environment
• Configure PyRel

Download sample data

This tutorial uses sample CSV files from the Energy Grid Planning template. The template models an ERCOT-style grid planning scenario for evaluating AI data center interconnection requests against existing capacity and possible infrastructure upgrades.

The sample data includes six CSV files covering:

• Substations
• Generators
• Transmission lines
• Demand forecasts
• Pending data center requests
• Candidate substation upgrades

Note

In production scenarios, data is typically loaded from Snowflake tables and views, not local CSV files. We use CSVs here to keep things simple and self-contained.

Follow these steps to download the sample data and add it to your project.

Prompt

Copy and paste the following prompt into your coding agent to download the sample data and place it in your project:

Download the Energy Grid Planning sample data from https://docs.relational.ai/templates/zips/v1/energy_grid_planning.zip into my current project.

Extract the zip, copy the `data/` directory from the extracted `energy_grid_planning` folder into my project root as `./data`, and then clean up the downloaded zip file and extracted folder.

Manual steps

Use the following steps to download the sample data and add it to your project:

1. Download and extract the sample data.

   In your terminal, run the following commands to download the sample data as a zip file and extract it:

   curl -O https://docs.relational.ai/templates/zips/v1/energy_grid_planning.zip
   unzip energy_grid_planning.zip

2. Move the data into your project.

   The previous step creates an energy_grid_planning directory with the CSV files in data/. Run the following commands to copy data/ into your project root and clean up the downloaded files:

   cp -R energy_grid_planning/data ./data
   rm energy_grid_planning.zip
   rm -rf energy_grid_planning

Now that you have some data, let’s use RelationalAI’s agent skills to build a semantic model and help you answer a complex business question.

Build a semantic model

RelationalAI semantic models are built using PyRel, a Python DSL for declaratively defining concepts, properties, relationships, and connecting them to your data.

Use the rai-build-starter-ontology skill to build a PyRel semantic model based on the sample data.

1. Invoke the rai-build-starter-ontology skill.

   Copy and paste the following prompt into your coding agent:

   /rai-build-starter-ontology Build an ontology for grid infrastructure planning from the CSVs in data/ covering substations, generators, transmission lines, demand forecasts, data center requests, and substation upgrades. Write the code to a file named energy_grid.py in the project root and execute it to run the validation queries.

2. Choose one-shot mode.

   The agent will prompt you to choose between one-shot and guided mode. For this tutorial, select one-shot mode so the agent generates the entire model in a single response.

   Tip

   When using this skill on your own data, use guided mode to ensure the generated model is aligned with your expectations and needs.

   Heads up

   Your agent may take 10–20 minutes to complete, depending on its settings.

What this does

• Analyzes the CSV files in /data to infer the domain model.
• Identifies the concepts, properties, and relationships to include in the PyRel semantic model.
• Writes a PyRel model with validation queries.

What you should see

• A complete energy_grid.py model with concepts, relationships, and validation queries.
• If the agent skips validation, run uv run energy_grid.py yourself.
• Iterate with your agent until all validation queries pass.

Note

When your agent executes energy_grid.py, a RelationalAI logic reasoner will be provisioned in the RAI Native App. It may take 5 minutes to the reasoner to start up.

What is a reasoner?

A reasoner is a compute resource managed by the RAI Native App on Snowpark Container Services that evaluates PyRel models and returns query results.

Sample output

Expand the section below for a sample output of what the generated energy_grid.py.

Sample output using OpenAI Codex (GPT5.4 Medium)

energy_grid.py

from pathlib import Path

import pandas as pd
from relationalai.semantics import Boolean, Float, Integer, Model, String
from relationalai.semantics import inspect
from relationalai.semantics.std import aggregates as aggs


DATA_DIR = Path(__file__).parent / "data"


def load_csv(name: str) -> pd.DataFrame:
    df = pd.read_csv(DATA_DIR / name)
    for col in df.columns:
        if df[col].dtype == object:
            values = set(df[col].dropna().astype(str).str.lower().unique())
            if values and values.issubset({"true", "false"}):
                df[col] = df[col].astype(str).str.lower().map({"true": True, "false": False})
    return df


substations_df = load_csv("substations.csv")
generators_df = load_csv("generators.csv")
transmission_lines_df = load_csv("transmission_lines.csv")
demand_forecasts_df = load_csv("demand_forecasts.csv")
data_center_requests_df = load_csv("data_center_requests.csv")
substation_upgrades_df = load_csv("substation_upgrades.csv")


model = Model("energy_grid_starter_ontology")

# Core grid concepts
Substation = model.Concept("Substation", identify_by={"id": String})
Substation.name = model.Property(f"{Substation} has {String:name}")
Substation.voltage_kv = model.Property(f"{Substation} has {Float:voltage_kv}")
Substation.max_capacity_mw = model.Property(f"{Substation} has {Float:max_capacity_mw}")
Substation.current_load_mw = model.Property(f"{Substation} has {Float:current_load_mw}")
Substation.latitude = model.Property(f"{Substation} has {Float:latitude}")
Substation.longitude = model.Property(f"{Substation} has {Float:longitude}")

Generator = model.Concept("Generator", identify_by={"id": String})
Generator.name = model.Property(f"{Generator} has {String:name}")
Generator.gen_type = model.Property(f"{Generator} has {String:gen_type}")
Generator.capacity_mw = model.Property(f"{Generator} has {Float:capacity_mw}")
Generator.min_output_mw = model.Property(f"{Generator} has {Float:min_output_mw}")
Generator.ramp_rate_mw_per_hr = model.Property(f"{Generator} has {Float:ramp_rate_mw_per_hr}")
Generator.startup_cost = model.Property(f"{Generator} has {Float:startup_cost}")
Generator.marginal_cost = model.Property(f"{Generator} has {Float:marginal_cost}")
Generator.min_up_time_hrs = model.Property(f"{Generator} has {Integer:min_up_time_hrs}")
Generator.min_down_time_hrs = model.Property(f"{Generator} has {Integer:min_down_time_hrs}")
Generator.emissions_rate = model.Property(f"{Generator} has {Float:emissions_rate}")
Generator.is_renewable = model.Property(f"{Generator} has {Boolean:is_renewable}")
Generator.substation = model.Property(f"{Generator} interconnects at {Substation:substation}")

TransmissionLine = model.Concept("TransmissionLine", identify_by={"id": String})
TransmissionLine.from_substation = model.Property(
    f"{TransmissionLine} originates at {Substation:from_substation}"
)
TransmissionLine.to_substation = model.Property(
    f"{TransmissionLine} terminates at {Substation:to_substation}"
)
TransmissionLine.capacity_mw = model.Property(f"{TransmissionLine} has {Float:capacity_mw}")
TransmissionLine.length_km = model.Property(f"{TransmissionLine} has {Float:length_km}")
TransmissionLine.impedance = model.Property(f"{TransmissionLine} has {Float:impedance}")
TransmissionLine.is_active = model.Property(f"{TransmissionLine} has {Boolean:is_active}")
TransmissionLine.maintenance_priority = model.Property(
    f"{TransmissionLine} has {String:maintenance_priority}"
)

DemandForecast = model.Concept("DemandForecast", identify_by={"id": String})
DemandForecast.substation = model.Property(f"{DemandForecast} forecasts for {Substation:substation}")
DemandForecast.forecast_period = model.Property(f"{DemandForecast} has {Integer:forecast_period}")
DemandForecast.predicted_load_mw = model.Property(f"{DemandForecast} has {Float:predicted_load_mw}")
DemandForecast.confidence = model.Property(f"{DemandForecast} has {Float:confidence}")
DemandForecast.includes_dc_growth = model.Property(f"{DemandForecast} has {Boolean:includes_dc_growth}")

DataCenterRequest = model.Concept("DataCenterRequest", identify_by={"id": String})
DataCenterRequest.name = model.Property(f"{DataCenterRequest} has {String:name}")
DataCenterRequest.hyperscaler = model.Property(f"{DataCenterRequest} has {String:hyperscaler}")
DataCenterRequest.requested_mw = model.Property(f"{DataCenterRequest} has {Float:requested_mw}")
DataCenterRequest.substation = model.Property(
    f"{DataCenterRequest} requests service from {Substation:substation}"
)
DataCenterRequest.annual_revenue_per_mw = model.Property(
    f"{DataCenterRequest} has {Float:annual_revenue_per_mw}"
)
DataCenterRequest.pue = model.Property(f"{DataCenterRequest} has {Float:pue}")
DataCenterRequest.is_ai_workload = model.Property(f"{DataCenterRequest} has {Boolean:is_ai_workload}")
DataCenterRequest.cooling_type = model.Property(f"{DataCenterRequest} has {String:cooling_type}")
DataCenterRequest.low_carbon_requirement_pct = model.Property(
    f"{DataCenterRequest} has {Float:low_carbon_requirement_pct}"
)
DataCenterRequest.queue_position = model.Property(f"{DataCenterRequest} has {Integer:queue_position}")
DataCenterRequest.status = model.Property(f"{DataCenterRequest} has {String:status}")

SubstationUpgrade = model.Concept("SubstationUpgrade", identify_by={"id": String})
SubstationUpgrade.substation = model.Property(
    f"{SubstationUpgrade} upgrades {Substation:substation}"
)
SubstationUpgrade.capacity_increase_mw = model.Property(
    f"{SubstationUpgrade} has {Float:capacity_increase_mw}"
)
SubstationUpgrade.cost_million = model.Property(f"{SubstationUpgrade} has {Float:cost_million}")
SubstationUpgrade.lead_time_months = model.Property(
    f"{SubstationUpgrade} has {Integer:lead_time_months}"
)
SubstationUpgrade.enables_low_carbon = model.Property(
    f"{SubstationUpgrade} has {Boolean:enables_low_carbon}"
)

# Derived planning metrics
Substation.total_generation_capacity_mw = model.Property(
    f"{Substation} has {Float:total_generation_capacity_mw}"
)
Substation.max_forecast_load_mw = model.Property(f"{Substation} has {Float:max_forecast_load_mw}")
Substation.pending_dc_request_mw = model.Property(f"{Substation} has {Float:pending_dc_request_mw}")
Substation.total_upgrade_capacity_mw = model.Property(
    f"{Substation} has {Float:total_upgrade_capacity_mw}"
)


def load_ontology() -> None:
    src = model.data(substations_df)
    model.define(
        Substation.new(
            id=src.ID,
            name=src.NAME,
            voltage_kv=src.VOLTAGE_KV,
            max_capacity_mw=src.MAX_CAPACITY_MW,
            current_load_mw=src.CURRENT_LOAD_MW,
            latitude=src.LATITUDE,
            longitude=src.LONGITUDE,
        )
    )

    src = model.data(generators_df)
    model.define(
        Generator.new(
            id=src.ID,
            name=src.NAME,
            gen_type=src.GEN_TYPE,
            capacity_mw=src.CAPACITY_MW,
            min_output_mw=src.MIN_OUTPUT_MW,
            ramp_rate_mw_per_hr=src.RAMP_RATE_MW_PER_HR,
            startup_cost=src.STARTUP_COST,
            marginal_cost=src.MARGINAL_COST,
            min_up_time_hrs=src.MIN_UP_TIME_HRS,
            min_down_time_hrs=src.MIN_DOWN_TIME_HRS,
            emissions_rate=src.EMISSIONS_RATE,
            is_renewable=src.IS_RENEWABLE,
            substation=Substation.filter_by(id=src.SUBSTATION_ID),
        )
    )

    src = model.data(transmission_lines_df)
    model.define(
        TransmissionLine.new(
            id=src.ID,
            from_substation=Substation.filter_by(id=src.FROM_SUBSTATION_ID),
            to_substation=Substation.filter_by(id=src.TO_SUBSTATION_ID),
            capacity_mw=src.CAPACITY_MW,
            length_km=src.LENGTH_KM,
            impedance=src.IMPEDANCE,
            is_active=src.IS_ACTIVE,
            maintenance_priority=src.MAINTENANCE_PRIORITY,
        )
    )

    src = model.data(demand_forecasts_df)
    model.define(
        DemandForecast.new(
            id=src.ID,
            substation=Substation.filter_by(id=src.SUBSTATION_ID),
            forecast_period=src.FORECAST_PERIOD,
            predicted_load_mw=src.PREDICTED_LOAD_MW,
            confidence=src.CONFIDENCE,
            includes_dc_growth=src.INCLUDES_DC_GROWTH,
        )
    )

    src = model.data(data_center_requests_df)
    model.define(
        DataCenterRequest.new(
            id=src.ID,
            name=src.NAME,
            hyperscaler=src.HYPERSCALER,
            requested_mw=src.REQUESTED_MW,
            substation=Substation.filter_by(id=src.SUBSTATION_ID),
            annual_revenue_per_mw=src.ANNUAL_REVENUE_PER_MW,
            pue=src.PUE,
            is_ai_workload=src.IS_AI_WORKLOAD,
            cooling_type=src.COOLING_TYPE,
            low_carbon_requirement_pct=src.LOW_CARBON_REQUIREMENT_PCT,
            queue_position=src.QUEUE_POSITION,
            status=src.STATUS,
        )
    )

    src = model.data(substation_upgrades_df)
    model.define(
        SubstationUpgrade.new(
            id=src.ID,
            substation=Substation.filter_by(id=src.SUBSTATION_ID),
            capacity_increase_mw=src.CAPACITY_INCREASE_MW,
            cost_million=src.COST_MILLION,
            lead_time_months=src.LEAD_TIME_MONTHS,
            enables_low_carbon=src.ENABLES_LOW_CARBON,
        )
    )

    model.define(
        Substation.total_generation_capacity_mw(
            aggs.sum(Generator.capacity_mw).where(Generator.substation(Substation)).per(Substation)
        )
    )
    model.define(
        Substation.max_forecast_load_mw(
            aggs.max(DemandForecast.predicted_load_mw)
            .where(DemandForecast.substation(Substation))
            .per(Substation)
        )
    )
    model.define(
        Substation.pending_dc_request_mw(
            aggs.sum(DataCenterRequest.requested_mw)
            .where(
                DataCenterRequest.substation(Substation),
                DataCenterRequest.status == "pending",
            )
            .per(Substation)
        )
    )
    model.define(
        Substation.total_upgrade_capacity_mw(
            aggs.sum(SubstationUpgrade.capacity_increase_mw)
            .where(SubstationUpgrade.substation(Substation))
            .per(Substation)
        )
    )


def print_schema_summary() -> None:
    schema = inspect.schema(model)
    concepts = [c for c in schema.concepts if not c.name.startswith("_")]
    print("\n=== ONTOLOGY SCHEMA SUMMARY ===")
    print(f"Concept count: {len(concepts)}")
    for concept in concepts:
        properties = ", ".join(sorted(p.name for p in concept.properties))
        print(f"- {concept.name}: {properties}")


def validation_queries() -> None:
    print("\n=== VALIDATION 1: TRANSMISSION TOPOLOGY ===")
    from_sub = Substation.ref("from_sub")
    to_sub = Substation.ref("to_sub")
    topology_df = (
        model.where(
            TransmissionLine.from_substation(from_sub),
            TransmissionLine.to_substation(to_sub),
        )
        .select(
            TransmissionLine.id.alias("line_id"),
            from_sub.id.alias("from_id"),
            from_sub.name.alias("from_name"),
            to_sub.id.alias("to_id"),
            to_sub.name.alias("to_name"),
            TransmissionLine.capacity_mw.alias("capacity_mw"),
            TransmissionLine.is_active.alias("is_active"),
            TransmissionLine.maintenance_priority.alias("maintenance_priority"),
        )
        .to_df()
        .sort_values(["from_id", "to_id"])
    )
    print(topology_df.head(10).to_string(index=False))

    print("\n=== VALIDATION 2: SUBSTATION CAPACITY PRESSURE ===")
    pressure_df = (
        model.select(
            Substation.id.alias("substation_id"),
            Substation.name.alias("substation_name"),
            Substation.current_load_mw.alias("current_load_mw"),
            Substation.max_capacity_mw.alias("max_capacity_mw"),
            Substation.total_generation_capacity_mw.alias("generation_capacity_mw"),
            Substation.max_forecast_load_mw.alias("max_forecast_load_mw"),
            Substation.pending_dc_request_mw.alias("pending_dc_request_mw"),
            Substation.total_upgrade_capacity_mw.alias("total_upgrade_capacity_mw"),
        )
        .to_df()
        .fillna(0.0)
    )
    pressure_df["headroom_now_mw"] = (
        pressure_df["max_capacity_mw"] - pressure_df["current_load_mw"]
    )
    pressure_df["headroom_vs_forecast_mw"] = (
        pressure_df["max_capacity_mw"] - pressure_df["max_forecast_load_mw"]
    )
    pressure_df["headroom_with_upgrades_mw"] = (
        pressure_df["max_capacity_mw"]
        + pressure_df["total_upgrade_capacity_mw"]
        - pressure_df["max_forecast_load_mw"]
    )
    pressure_df = pressure_df.sort_values(
        ["headroom_vs_forecast_mw", "pending_dc_request_mw"], ascending=[True, False]
    )
    print(pressure_df.head(12).to_string(index=False))

    print("\n=== VALIDATION 3: GENERATOR MIX BY SUBSTATION ===")
    mix_df = (
        model.where(Generator.substation(Substation))
        .select(
            Substation.id.alias("substation_id"),
            Substation.name.alias("substation_name"),
            Generator.id.alias("generator_id"),
            Generator.name.alias("generator_name"),
            Generator.gen_type.alias("gen_type"),
            Generator.capacity_mw.alias("capacity_mw"),
            Generator.is_renewable.alias("is_renewable"),
            Generator.marginal_cost.alias("marginal_cost"),
        )
        .to_df()
        .sort_values(["substation_id", "generator_id"])
    )
    print(mix_df.head(15).to_string(index=False))

    print("\n=== VALIDATION 4: DATA CENTER REQUESTS AND UPGRADE OPTIONS ===")
    request_df = (
        model.where(
            DataCenterRequest.substation(Substation),
            SubstationUpgrade.substation(Substation),
        )
        .select(
            DataCenterRequest.id.alias("request_id"),
            DataCenterRequest.name.alias("request_name"),
            DataCenterRequest.hyperscaler.alias("hyperscaler"),
            DataCenterRequest.requested_mw.alias("requested_mw"),
            DataCenterRequest.status.alias("status"),
            Substation.id.alias("substation_id"),
            Substation.name.alias("substation_name"),
            SubstationUpgrade.id.alias("upgrade_id"),
            SubstationUpgrade.capacity_increase_mw.alias("upgrade_capacity_mw"),
            SubstationUpgrade.cost_million.alias("upgrade_cost_million"),
            SubstationUpgrade.enables_low_carbon.alias("enables_low_carbon"),
        )
        .to_df()
        .sort_values(["request_id", "upgrade_cost_million"])
    )
    print(request_df.head(20).to_string(index=False))


def main() -> None:
    load_ontology()
    print_schema_summary()
    validation_queries()


if __name__ == "__main__":
    main()

Related resources:

• Understand how PyRel works
• Build a semantic model

Visualize the model’s schema

To help you understand the structure of the model you just built, use the rai-querying skill to visualize the ontology as a concept-relationship diagram and report row counts per concept.

Copy and paste the following prompt into your coding agent to invoke the skill:

/rai-querying Show the ontology as a concept-relationship diagram and report row counts per concept.

What this does

• Generates a concept-relationship diagram of your model so you can inspect its structure.

What to expect in response

• A diagram with concepts as boxes and relationships as labeled arrows.
• Some agents render it directly. Others return Mermaid code that you can render using a tool like Mermaid Live Editor.

Sample output

Expand the section below for a sample output of what the generated diagram might look like.

Sample output from OpenAI Codex (GPT5.4 Medium)

erDiagram
    Substation {
        String id PK
        String name
        Integer voltage_kv
        Integer max_capacity_mw
        Integer current_load_mw
        Float latitude
        Float longitude
    }

    Generator {
        String id PK
        String name
        String gen_type
        Integer capacity_mw
        Integer min_output_mw
        Integer ramp_rate_mw_per_hr
        Integer startup_cost
        Float marginal_cost
        Integer min_up_time_hrs
        Integer min_down_time_hrs
        Float emissions_rate
        Boolean is_renewable
    }

    TransmissionLine {
        String id PK
        Integer capacity_mw
        Integer length_km
        Float impedance
        String maintenance_priority
        Boolean is_active
    }

    DemandForecast {
        String id PK
        Integer forecast_period
        Float predicted_load_mw
        Float confidence
        Boolean includes_dc_growth
    }

    DataCenterRequest {
        String id PK
        String name
        String hyperscaler
        Integer requested_mw
        Integer annual_revenue_per_mw
        Float pue
        String cooling_type
        Float low_carbon_requirement_pct
        Integer queue_position
        String status
        Boolean is_ai_workload
    }

    SubstationUpgrade {
        String id PK
        Integer capacity_increase_mw
        Integer cost_million
        Integer lead_time_months
        Boolean enables_low_carbon
    }

    Substation ||--o{ Generator : "connected to (gen_substation)"
    Substation ||--o{ TransmissionLine : "originates from (tl_from)"
    Substation ||--o{ TransmissionLine : "terminates at (tl_to)"
    Substation ||--o{ DemandForecast : "forecasts load for (df_substation)"
    Substation ||--o{ DataCenterRequest : "is sited at (dc_substation)"
    Substation ||--o{ SubstationUpgrade : "upgrades (upg_substation)"

Diagram source code (Mermaid syntax)

```mermaid
erDiagram
    Substation {
        String id PK
        String name
        Integer voltage_kv
        Integer max_capacity_mw
        Integer current_load_mw
        Float latitude
        Float longitude
    }

    Generator {
        String id PK
        String name
        String gen_type
        Integer capacity_mw
        Integer min_output_mw
        Integer ramp_rate_mw_per_hr
        Integer startup_cost
        Float marginal_cost
        Integer min_up_time_hrs
        Integer min_down_time_hrs
        Float emissions_rate
        Boolean is_renewable
    }

    TransmissionLine {
        String id PK
        Integer capacity_mw
        Integer length_km
        Float impedance
        String maintenance_priority
        Boolean is_active
    }

    DemandForecast {
        String id PK
        Integer forecast_period
        Float predicted_load_mw
        Float confidence
        Boolean includes_dc_growth
    }

    DataCenterRequest {
        String id PK
        String name
        String hyperscaler
        Integer requested_mw
        Integer annual_revenue_per_mw
        Float pue
        String cooling_type
        Float low_carbon_requirement_pct
        Integer queue_position
        String status
        Boolean is_ai_workload
    }

    SubstationUpgrade {
        String id PK
        Integer capacity_increase_mw
        Integer cost_million
        Integer lead_time_months
        Boolean enables_low_carbon
    }

    Substation ||--o{ Generator : "connected to (gen_substation)"
    Substation ||--o{ TransmissionLine : "originates from (tl_from)"
    Substation ||--o{ TransmissionLine : "terminates at (tl_to)"
    Substation ||--o{ DemandForecast : "forecasts load for (df_substation)"
    Substation ||--o{ DataCenterRequest : "is sited at (dc_substation)"
    Substation ||--o{ SubstationUpgrade : "upgrades (upg_substation)"
```

Refine a question you want to answer

Use the rai-discovery skill to refine your questions and choose the right skills and prompts. In this grid planning scenario, the goal is to decide which requests to approve and upgrades to fund.

Copy and paste the following prompt into your coding agent to invoke the skill:

/rai-discovery I need to determine which datacenter requests to approve and which upgrades to fund at what budget level. What combination of skills/prompts can I use to answer these questions? The prompt should give me an end-end-answer and explain the results for a non-technical business user. Give me copy-pastable options as:

OPTION N: <NAME>

<Describe the option in a few sentences.>

```
[/skill1 [+ /skill2 ...]] <PROMPT>
 ```

**Benefits:**

- Benefit 1
- Benefit 2
- ...

**Trade-offs:**

- Trade-off 1
- Trade-off 2
- ...

What this does

• Checks whether your question is feasible with your model and data.
• Recommends which skills to use and how to combine them.
• Returns copy-paste prompts with the benefits and trade-offs of each option.

What to expect in response

• Expect several skill combinations for different solution paths.
• Common choices include:
  • rai-rules-authoring for rule-based reasoning
  • rai-prescriptive-problem-formulation for optimization
  • rai-graph-analysis for network analysis

Sample output

Expand the section below for a sample output of what the generated options might look like.

Sample output from OpenAI Codex (GPT5.4 Medium)

OPTION 1: Direct Investment Recommendation

This is the fastest path to the business decision you asked for. It goes straight to optimization using the ontology you already built, then asks for a business-facing explanation of which data center requests to approve, which upgrades to fund, and how the recommendation changes by budget level.

/rai-prescriptive-problem-formulation + /rai-prescriptive-results-interpretation Using the ontology in energy_grid.py, formulate and solve a budgeted grid planning optimization that decides which DataCenterRequest records to approve and which SubstationUpgrade records to fund. Evaluate multiple budget levels and return the recommended approvals and funded upgrades at each level. Optimize for business value using annual revenue, capacity feasibility, forecasted load, and upgrade costs, while respecting substation capacity and upgrade effects. Give me an end-to-end answer in business language for a non-technical executive: summarize the recommendation by budget level, explain why each approved request was selected, explain why rejected requests were deferred, identify the most constrained substations, and call out the trade-offs between low budget and high budget plans. Include a concise final recommendation for the best budget level.

Benefits:

• Fastest route to a concrete answer
• Produces a budget-by-budget recommendation, not just analysis
• Designed to explain the results for business users

Trade-offs:

• Focuses on optimization first, with less structural context on network risk
• May underemphasize resilience unless explicitly modeled in the objective or constraints

OPTION 2: Risk-Aware Approval Plan

This option adds structural grid risk before optimization. It is better when you want the final recommendation to consider not only revenue and capacity, but also whether approvals concentrate risk on important or fragile substations.

[/rai-graph-analysis + /rai-prescriptive-problem-formulation + /rai-prescriptive-results-interpretation] First analyze the transmission network in the energy_grid ontology to identify structurally critical substations and constrained corridors. Then use those graph outputs in a prescriptive optimization that decides which DataCenterRequest records to approve and which SubstationUpgrade records to fund across multiple budget levels. Optimize for business value while penalizing plans that overload or over-concentrate demand on structurally critical substations. Give me an end-to-end answer for a non-technical business audience: explain the approval and upgrade recommendation by budget level, describe the grid-risk trade-offs in plain English, identify which substations are strategic bottlenecks, and recommend the budget level that gives the best balance of growth, resilience, and upgrade spend.

Benefits:

• Better for resilience-aware planning
• Makes the explanation stronger for executives worried about concentration risk
• Connects topology risk to funding choices

Trade-offs:

• More complex than a direct optimization
• May take longer to build and explain
• Requires graph outputs to be incorporated cleanly into the optimization

OPTION 3: Policy-Constrained Approval Plan

This option is best when you want the recommendation to reflect explicit business policies such as low-carbon requirements, queue fairness, or approval rules before optimizing. It combines rule enforcement with optimization and then translates the result into business language.

[/rai-rules-authoring + /rai-prescriptive-problem-formulation + /rai-prescriptive-results-interpretation] Using the energy_grid ontology, first define business rules that classify each DataCenterRequest and SubstationUpgrade for policy fit, including low-carbon requirements, queue status, and whether forecasted capacity and planned upgrades can support the request. Then formulate and solve an optimization that chooses which requests to approve and which upgrades to fund at multiple budget levels, using those rule outputs as eligibility constraints or penalties. Give me an end-to-end answer for a non-technical business user: for each budget level, show which requests should be approved, which upgrades should be funded, which requests are blocked by policy or capacity, and why. Finish with a plain-English recommendation for the best budget level and the business rationale behind it.

Benefits:

• Best when governance and policy compliance matter
• Makes approvals easier to defend to business stakeholders
• Separates “not allowed” from “not worth funding yet”

Trade-offs:

• More setup than a pure optimization
• Requires explicit rule choices up front
• Can make the model stricter and reduce feasible approvals

OPTION 4: Full Executive Scenario Pack

This is the most complete option. It is the best choice if you want one prompt to produce a decision memo style answer with scenarios, trade-offs, and an executive recommendation.

[/rai-discovery + /rai-graph-analysis + /rai-rules-authoring + /rai-prescriptive-problem-formulation + /rai-prescriptive-results-interpretation] Use the energy_grid ontology to produce an executive decision analysis for data center interconnection planning. Determine which DataCenterRequest records to approve and which SubstationUpgrade records to fund under multiple budget levels. Use graph analysis to identify structural bottlenecks, rules to classify policy and low-carbon fit, and prescriptive optimization to choose the best portfolio of approvals and upgrades at each budget level. Return an end-to-end answer in business language for a non-technical audience: include an executive summary, a comparison of budget scenarios, the recommended approvals and upgrades at each level, the reasons behind the choices, the major risks and bottlenecks, the requests that should be deferred, and a final recommended budget level with clear business justification.

Benefits:

• Most complete business-facing answer
• Covers feasibility, resilience, policy, and optimization in one flow
• Best for presentation-ready output

Trade-offs:

• Heaviest option
• More moving parts means more time and more room for iteration
• Overkill if you only need a quick recommendation

If you want one default, use OPTION 1 for speed or OPTION 4 for the strongest executive-ready answer.

Answer your question

Pick one of the options from the previous step. Then paste its prompt into your coding agent to run the required skills.

For example, Option 2 in the previous step’s sample output shows how graph and prescriptive can be used together to answer the question in a more nuanced way:

[/rai-graph-analysis + /rai-prescriptive-problem-formulation + /rai-prescriptive-results-interpretation] First analyze the transmission network in the energy_grid ontology to identify structurally critical substations and constrained corridors. Then use those graph outputs in a prescriptive optimization that decides which DataCenterRequest records to approve and which SubstationUpgrade records to fund across multiple budget levels. Optimize for business value while penalizing plans that overload or over-concentrate demand on structurally critical substations. Give me an end-to-end answer for a non-technical business audience: explain the approval and upgrade recommendation by budget level, describe the grid-risk trade-offs in plain English, identify which substations are strategic bottlenecks, and recommend the budget level that gives the best balance of growth, resilience, and upgrade spend.

Heads up

This prompt may take your agent 5–10 minutes to complete.

What this does

• Adds a criticality score to each Substation in energy_grid.py using RAI’s graph reasoning capabilities.
• Uses those scores in a prescriptive optimization problem.
• Runs the optimization using a prescriptive reasoner in the RAI Native App.
• Summarizes which data centers to approve, which upgrades to fund, and why.

What to expect in response

• A second Python script in addition to energy_grid.py with a name like dc_approval_optimizer.py or energy_grid_investment_plan.py.
• A results summary in your chat showing explaining the recommended approvals and upgrades.

Sample output

Expand the section below for a sample output of what the generated code and results might look like.

Sample output from OpenAI Codex (GPT5.4 Medium)

Summary

I ran the full analysis in energy_grid_investment_plan.py against your current ontology.

Recommendation The best balance of growth, resilience, and spend is the $300M budget level.

At $300M, the plan approves 6 requests totaling 1,530 MW, funds 4 upgrades, and delivers about $213.1M in net annual value after risk penalties and annualized upgrade cost. It captures most of the attractive growth opportunities, but stops before the plan becomes too concentrated on the grid’s three biggest bottlenecks: Dallas-Fort Worth, Houston Ship Channel, and San Antonio Metro.

If the business wants the most conservative plan, choose $200M. If it wants maximum growth and is comfortable taking much more concentration risk, choose $600M.

Budget Outcomes

• $200M: approve 4 requests, 930 MW, fund 3 upgrades, net annual value about $132.2M. Lowest risk concentration.
• $300M: approve 6 requests, 1,530 MW, fund 4 upgrades, net annual value about $213.1M. Best balance.
• $400M: approve 7 requests, 1,900 MW, fund 5 upgrades, net annual value about $262.8M. Good growth, but Houston becomes a major new concentration point.
• $500M: approve 8 requests, 2,180 MW, fund 6 upgrades, net annual value about $300.4M. Higher return, but risk concentration climbs further.
• $600M: approve 9 requests, 2,430 MW, fund 7 upgrades, net annual value about $333.2M. Best pure growth result, but the portfolio is heavily concentrated on critical substations.

What The Model Is Saying The main strategic bottlenecks are:

• Dallas-Fort Worth: already forecast to be over capacity by about 100 MW, and also the most structurally critical node in the transmission network.
• Houston Ship Channel: almost no forecast headroom left, so large approvals there require major upgrades.
• San Antonio Metro: not as tight as Dallas or Houston, but still structurally important and connected to constrained corridors.

The most constrained corridors tied to those bottlenecks are the Dallas-Abilene links and the San Antonio links to Corpus Christi and Waco. In plain English: these are the paths where disruption or congestion is most likely to ripple through the grid.

That is why the model favors:

• Early approvals for CoreWeave Austin GPU, Crusoe Permian DC, and Oracle Coastal DC
• Adding Google Metroplex DC and Lambda Labs DFW once Dallas upgrades are funded
• Deferring the Houston pair until a higher budget
• Deferring xAI Colossus Texas at every budget level because Dallas is the most stressed node, and the Google + Lambda combination gives better risk-adjusted value per MW than a single very large Dallas request

Business Interpretation $300M is the point where the portfolio still looks disciplined. You get a meaningful step-up from $200M, but you avoid the sharper jump in dependence on critical substations that starts at $400M and accelerates after that. The extra spend above $300M does buy more growth, but each additional $100M produces smaller incremental value while pushing more load into the parts of the network that are hardest to absorb safely.

If you want, I can turn this into a one-page executive memo or export the per-budget approval and upgrade lists as a table.

Python Code

from __future__ import annotations

from pathlib import Path

import pandas as pd
from relationalai.semantics import Boolean, Float, Integer, String
from relationalai.semantics.reasoners.graph import Graph
from relationalai.semantics.reasoners.prescriptive import Problem
from relationalai.semantics.std import aggregates as aggs

from energy_grid import (
    DATA_DIR,
    DataCenterRequest,
    DemandForecast,
    Generator,
    Substation,
    SubstationUpgrade,
    TransmissionLine,
    load_ontology,
    model,
)


BUDGET_LEVELS = [200.0, 300.0, 400.0, 500.0, 600.0]
AMORTIZATION_YEARS = 20
TOP_CRITICAL_COUNT = 3


def build_graph_outputs() -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
    graph = Graph(
        model,
        directed=False,
        weighted=False,
        node_concept=Substation,
        aggregator="sum",
    )

    line_ref = TransmissionLine.ref()
    model.define(graph.Edge.new(src=line_ref.from_substation, dst=line_ref.to_substation)).where(
        line_ref.is_active == True
    )

    sub_ref = Substation.ref("sub")
    btwn = Float.ref("btwn")
    deg = Float.ref("deg")
    eig = Float.ref("eig")
    community_label = Integer.ref("community")
    component_ref = Substation.ref("component")

    betweenness = graph.betweenness_centrality()
    degree = graph.degree_centrality()
    eigenvector = graph.eigenvector_centrality()
    community = graph.louvain()
    wcc = graph.weakly_connected_component()

    centrality_df = (
        model.select(
            Substation.id.alias("substation_id"),
            Substation.name.alias("substation_name"),
            Substation.current_load_mw.alias("current_load_mw"),
            Substation.max_capacity_mw.alias("max_capacity_mw"),
            Substation.max_forecast_load_mw.alias("max_forecast_load_mw"),
        )
        .to_df()
        .fillna(0.0)
    )

    btwn_df = model.where(betweenness(sub_ref, btwn)).select(
        sub_ref.id.alias("substation_id"),
        btwn.alias("betweenness"),
    ).to_df()
    deg_df = model.where(degree(sub_ref, deg)).select(
        sub_ref.id.alias("substation_id"),
        deg.alias("degree_centrality"),
    ).to_df()
    eig_df = model.where(eigenvector(sub_ref, eig)).select(
        sub_ref.id.alias("substation_id"),
        eig.alias("eigenvector_centrality"),
    ).to_df()
    community_df = model.where(community(sub_ref, community_label)).select(
        sub_ref.id.alias("substation_id"),
        community_label.alias("community"),
    ).to_df()
    component_df = model.where(wcc(sub_ref, component_ref)).select(
        sub_ref.id.alias("substation_id"),
        component_ref.id.alias("component_id"),
    ).to_df()

    centrality_df = (
        centrality_df.merge(btwn_df, on="substation_id", how="left")
        .merge(deg_df, on="substation_id", how="left")
        .merge(eig_df, on="substation_id", how="left")
        .merge(community_df, on="substation_id", how="left")
        .merge(component_df, on="substation_id", how="left")
        .fillna(0.0)
    )
    centrality_df["betweenness_rank"] = centrality_df["betweenness"].rank(
        method="dense", ascending=False
    )
    centrality_df["degree_rank"] = centrality_df["degree_centrality"].rank(
        method="dense", ascending=False
    )
    centrality_df["eigenvector_rank"] = centrality_df["eigenvector_centrality"].rank(
        method="dense", ascending=False
    )
    centrality_df["combined_rank_score"] = (
        centrality_df["betweenness_rank"]
        + centrality_df["degree_rank"]
        + centrality_df["eigenvector_rank"]
    )
    centrality_df = centrality_df.sort_values(
        ["combined_rank_score", "betweenness", "degree_centrality"],
        ascending=[True, False, False],
    ).reset_index(drop=True)
    centrality_df["critical_rank"] = range(1, len(centrality_df) + 1)
    centrality_df["is_structurally_critical"] = (
        centrality_df["critical_rank"] <= TOP_CRITICAL_COUNT
    )
    centrality_df["forecast_headroom_mw"] = (
        centrality_df["max_capacity_mw"] - centrality_df["max_forecast_load_mw"]
    )

    from_sub = Substation.ref("from_sub")
    to_sub = Substation.ref("to_sub")
    topology_df = (
        model.where(
            TransmissionLine.from_substation(from_sub),
            TransmissionLine.to_substation(to_sub),
        )
        .select(
            TransmissionLine.id.alias("line_id"),
            TransmissionLine.capacity_mw.alias("capacity_mw"),
            TransmissionLine.maintenance_priority.alias("maintenance_priority"),
            from_sub.id.alias("from_id"),
            from_sub.name.alias("from_name"),
            to_sub.id.alias("to_id"),
            to_sub.name.alias("to_name"),
        )
        .to_df()
    )

    priority_weight = {"high": 1.0, "medium": 0.65, "low": 0.35}
    critical_ids = set(
        centrality_df.loc[centrality_df["is_structurally_critical"], "substation_id"]
    )
    topology_df["touches_critical"] = topology_df["from_id"].isin(critical_ids) | topology_df[
        "to_id"
    ].isin(critical_ids)
    topology_df["priority_weight"] = topology_df["maintenance_priority"].map(priority_weight).fillna(
        0.5
    )
    median_capacity = float(topology_df["capacity_mw"].median())
    topology_df["corridor_risk_score"] = topology_df["priority_weight"] * (
        median_capacity / topology_df["capacity_mw"].clip(lower=1.0)
    )
    corridor_df = topology_df[topology_df["touches_critical"]].copy()
    corridor_df = corridor_df.sort_values(
        ["corridor_risk_score", "capacity_mw"], ascending=[False, True]
    ).reset_index(drop=True)

    exposure_rows: list[dict[str, float | str]] = []
    for _, row in corridor_df.iterrows():
        exposure_rows.append(
            {"substation_id": row["from_id"], "corridor_exposure_score": row["corridor_risk_score"]}
        )
        exposure_rows.append(
            {"substation_id": row["to_id"], "corridor_exposure_score": row["corridor_risk_score"]}
        )
    exposure_df = pd.DataFrame(exposure_rows)
    if len(exposure_df) > 0:
        exposure_df = (
            exposure_df.groupby("substation_id", as_index=False)["corridor_exposure_score"]
            .sum()
            .sort_values("corridor_exposure_score", ascending=False)
        )
    else:
        exposure_df = pd.DataFrame(columns=["substation_id", "corridor_exposure_score"])

    centrality_df = centrality_df.merge(exposure_df, on="substation_id", how="left").fillna(
        {"corridor_exposure_score": 0.0}
    )

    return centrality_df, corridor_df, topology_df


def bind_risk_metrics(centrality_df: pd.DataFrame) -> pd.DataFrame:
    Substation.critical_rank = model.Property(f"{Substation} has {Integer:critical_rank}")
    Substation.is_structurally_critical = model.Property(
        f"{Substation} has {Boolean:is_structurally_critical}"
    )
    Substation.corridor_exposure_score = model.Property(
        f"{Substation} has {Float:corridor_exposure_score}"
    )
    Substation.risk_multiplier = model.Property(f"{Substation} has {Float:risk_multiplier}")
    DataCenterRequest.risk_penalty_value = model.Property(
        f"{DataCenterRequest} has {Float:risk_penalty_value}"
    )
    DataCenterRequest.risk_adjusted_value = model.Property(
        f"{DataCenterRequest} has {Float:risk_adjusted_value}"
    )

    max_corridor = max(float(centrality_df["corridor_exposure_score"].max()), 1e-6)
    max_rank = max(int(centrality_df["critical_rank"].max()), 1)
    centrality_df = centrality_df.copy()
    centrality_df["criticality_norm"] = (
        (max_rank + 1 - centrality_df["critical_rank"]) / max_rank
    )
    centrality_df["corridor_norm"] = centrality_df["corridor_exposure_score"] / max_corridor
    centrality_df["risk_multiplier"] = (
        0.12 * centrality_df["criticality_norm"] + 0.08 * centrality_df["corridor_norm"]
    )

    substation_src = model.data(
        centrality_df[
            [
                "substation_id",
                "critical_rank",
                "is_structurally_critical",
                "corridor_exposure_score",
                "risk_multiplier",
            ]
        ].rename(columns={"substation_id": "ID"})
    )
    sub_match = Substation.filter_by(id=substation_src.ID)
    model.define(sub_match.critical_rank(substation_src.CRITICAL_RANK))
    model.define(sub_match.is_structurally_critical(substation_src.IS_STRUCTURALLY_CRITICAL))
    model.define(sub_match.corridor_exposure_score(substation_src.CORRIDOR_EXPOSURE_SCORE))
    model.define(sub_match.risk_multiplier(substation_src.RISK_MULTIPLIER))

    request_df = pd.read_csv(DATA_DIR / "data_center_requests.csv")
    request_df = request_df.merge(
        centrality_df[["substation_id", "risk_multiplier"]],
        left_on="SUBSTATION_ID",
        right_on="substation_id",
        how="left",
    ).fillna({"risk_multiplier": 0.0})
    request_df["gross_annual_value"] = (
        request_df["REQUESTED_MW"].astype(float) * request_df["ANNUAL_REVENUE_PER_MW"].astype(float)
    )
    request_df["risk_penalty_value"] = (
        request_df["gross_annual_value"] * request_df["risk_multiplier"].astype(float)
    )
    request_df["risk_adjusted_value"] = (
        request_df["gross_annual_value"] - request_df["risk_penalty_value"]
    )

    req_src = model.data(
        request_df[["ID", "risk_penalty_value", "risk_adjusted_value"]]
    )
    req_match = DataCenterRequest.filter_by(id=req_src.ID)
    model.define(req_match.risk_penalty_value(req_src.RISK_PENALTY_VALUE))
    model.define(req_match.risk_adjusted_value(req_src.RISK_ADJUSTED_VALUE))
    return request_df


def solve_budget_scenarios() -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
    InvestmentLevel = model.Concept("InvestmentLevel", identify_by={"name": String})
    InvestmentLevel.budget_cap = model.Property(f"{InvestmentLevel} has {Float:budget_cap}")

    inv_df = pd.DataFrame(
        [{"name": f"${int(b)}M", "budget_cap": b} for b in BUDGET_LEVELS]
    )
    inv_src = model.data(inv_df)
    model.define(InvestmentLevel.new(inv_src.to_schema()))

    DataCenterRequest.x_approve = model.Property(
        f"{DataCenterRequest} in {InvestmentLevel} has {Float:x_approve}"
    )
    SubstationUpgrade.x_upgrade = model.Property(
        f"{SubstationUpgrade} in {InvestmentLevel} has {Float:x_upgrade}"
    )

    x_a = Float.ref("xa")
    x_u = Float.ref("xu")
    problem = Problem(model, Float)

    problem.solve_for(
        DataCenterRequest.x_approve(InvestmentLevel, x_a),
        type="bin",
        name=["approve", InvestmentLevel.name, DataCenterRequest.id],
    )
    problem.solve_for(
        SubstationUpgrade.x_upgrade(InvestmentLevel, x_u),
        type="bin",
        name=["upgrade", InvestmentLevel.name, SubstationUpgrade.id],
    )

    x_a_cap = Float.ref("xa_cap")
    x_u_cap = Float.ref("xu_cap")
    dc_ref = DataCenterRequest.ref()
    upg_ref = SubstationUpgrade.ref()
    scenario_ref = InvestmentLevel.ref()
    sub_ref = Substation.ref()

    problem.satisfy(
        model.where(
            dc_ref.x_approve(scenario_ref, x_a_cap),
            upg_ref.x_upgrade(scenario_ref, x_u_cap),
            dc_ref.substation(sub_ref),
            upg_ref.substation(sub_ref),
        ).require(
            sub_ref.max_capacity_mw - (sub_ref.max_forecast_load_mw | sub_ref.current_load_mw)
            + aggs.sum(x_u_cap * upg_ref.capacity_increase_mw)
            .where(upg_ref.substation(sub_ref))
            .per(sub_ref, scenario_ref)
            >= aggs.sum(x_a_cap * dc_ref.requested_mw)
            .where(dc_ref.substation(sub_ref))
            .per(sub_ref, scenario_ref)
        )
    )

    problem.satisfy(
        model.where(
            SubstationUpgrade.x_upgrade(InvestmentLevel, x_u),
        ).require(
            aggs.sum(x_u * SubstationUpgrade.cost_million).per(InvestmentLevel)
            <= InvestmentLevel.budget_cap
        )
    )

    problem.maximize(
        aggs.sum(x_a * DataCenterRequest.risk_adjusted_value).where(
            DataCenterRequest.x_approve(InvestmentLevel, x_a)
        )
        - aggs.sum((x_u * SubstationUpgrade.cost_million * 1_000_000.0) / AMORTIZATION_YEARS).where(
            SubstationUpgrade.x_upgrade(InvestmentLevel, x_u)
        )
    )

    problem.solve("highs", time_limit_sec=120)
    si = problem.solve_info()
    if si.termination_status != "OPTIMAL":
        raise RuntimeError(f"Optimization did not solve optimally: {si.termination_status}")

    approve_val = Float.ref("approve")
    approved_df = (
        model.where(
            DataCenterRequest.x_approve(InvestmentLevel, approve_val),
            approve_val > 0.5,
            DataCenterRequest.substation(Substation),
        )
        .select(
            InvestmentLevel.name.alias("level"),
            InvestmentLevel.budget_cap.alias("budget"),
            DataCenterRequest.id.alias("request_id"),
            DataCenterRequest.name.alias("request_name"),
            DataCenterRequest.hyperscaler.alias("hyperscaler"),
            DataCenterRequest.requested_mw.alias("requested_mw"),
            DataCenterRequest.annual_revenue_per_mw.alias("annual_revenue_per_mw"),
            DataCenterRequest.risk_penalty_value.alias("risk_penalty_value"),
            DataCenterRequest.risk_adjusted_value.alias("risk_adjusted_value"),
            Substation.id.alias("substation_id"),
            Substation.name.alias("substation_name"),
            Substation.critical_rank.alias("critical_rank"),
            Substation.is_structurally_critical.alias("is_structurally_critical"),
        )
        .to_df()
    )

    upgrade_val = Float.ref("upgrade")
    upgrades_df = (
        model.where(
            SubstationUpgrade.x_upgrade(InvestmentLevel, upgrade_val),
            upgrade_val > 0.5,
            SubstationUpgrade.substation(Substation),
        )
        .select(
            InvestmentLevel.name.alias("level"),
            InvestmentLevel.budget_cap.alias("budget"),
            SubstationUpgrade.id.alias("upgrade_id"),
            SubstationUpgrade.capacity_increase_mw.alias("capacity_increase_mw"),
            SubstationUpgrade.cost_million.alias("cost_million"),
            SubstationUpgrade.enables_low_carbon.alias("enables_low_carbon"),
            Substation.id.alias("substation_id"),
            Substation.name.alias("substation_name"),
        )
        .to_df()
    )

    summary_rows = []
    for budget in BUDGET_LEVELS:
        level = f"${int(budget)}M"
        a = approved_df[approved_df["level"] == level].copy()
        u = upgrades_df[upgrades_df["level"] == level].copy()
        gross_revenue = (
            a["requested_mw"].astype(float) * a["annual_revenue_per_mw"].astype(float)
        ).sum()
        risk_penalty = a["risk_penalty_value"].astype(float).sum()
        adjusted_revenue = a["risk_adjusted_value"].astype(float).sum()
        upgrade_cost = u["cost_million"].astype(float).sum()
        annualized_upgrade_cost = (upgrade_cost * 1_000_000.0) / AMORTIZATION_YEARS
        critical_mw = a.loc[a["is_structurally_critical"] == True, "requested_mw"].astype(float).sum()
        total_mw = a["requested_mw"].astype(float).sum()
        summary_rows.append(
            {
                "level": level,
                "budget": budget,
                "approved_requests": len(a),
                "approved_mw": total_mw,
                "gross_revenue": gross_revenue,
                "risk_penalty": risk_penalty,
                "adjusted_revenue": adjusted_revenue,
                "upgrade_cost_million": upgrade_cost,
                "annualized_upgrade_cost": annualized_upgrade_cost,
                "net_annual_value": adjusted_revenue - annualized_upgrade_cost,
                "critical_mw": critical_mw,
                "critical_share": critical_mw / total_mw if total_mw else 0.0,
                "funded_upgrades": len(u),
                "added_capacity_mw": u["capacity_increase_mw"].astype(float).sum(),
            }
        )
    summary_df = pd.DataFrame(summary_rows).sort_values("budget").reset_index(drop=True)
    summary_df["marginal_value_per_million"] = summary_df["net_annual_value"].diff() / summary_df[
        "budget"
    ].diff()

    if len(summary_df) >= 3:
        best_idx = 1
        best_jump = float("-inf")
        for i in range(1, len(summary_df) - 1):
            prev = summary_df.loc[i, "marginal_value_per_million"]
            nxt = summary_df.loc[i + 1, "marginal_value_per_million"]
            if pd.notna(prev) and pd.notna(nxt):
                jump = prev - nxt
                if jump > best_jump:
                    best_jump = jump
                    best_idx = i
        summary_df["is_recommended_budget"] = False
        summary_df.loc[best_idx, "is_recommended_budget"] = True
    else:
        summary_df["is_recommended_budget"] = False
        if len(summary_df) > 0:
            summary_df.loc[summary_df["net_annual_value"].idxmax(), "is_recommended_budget"] = True

    return summary_df, approved_df, upgrades_df


def print_report(
    centrality_df: pd.DataFrame,
    corridor_df: pd.DataFrame,
    request_df: pd.DataFrame,
    summary_df: pd.DataFrame,
    approved_df: pd.DataFrame,
    upgrades_df: pd.DataFrame,
) -> None:
    print("=== GRID RISK OVERVIEW ===")
    component_count = centrality_df["component_id"].nunique()
    community_count = centrality_df["community"].nunique()
    print(f"Network components: {component_count}")
    print(f"Detected communities: {community_count}")
    print("\nStructurally critical substations:")
    cols = [
        "critical_rank",
        "substation_name",
        "forecast_headroom_mw",
        "betweenness",
        "degree_centrality",
        "corridor_exposure_score",
    ]
    print(
        centrality_df.loc[centrality_df["is_structurally_critical"], cols]
        .to_string(index=False)
    )

    print("\nMost constrained corridors touching critical substations:")
    if len(corridor_df) == 0:
        print("No constrained corridors identified.")
    else:
        print(
            corridor_df[
                [
                    "line_id",
                    "from_name",
                    "to_name",
                    "capacity_mw",
                    "maintenance_priority",
                    "corridor_risk_score",
                ]
            ]
            .head(6)
            .to_string(index=False)
        )

    print("\nRequest risk adjustments:")
    print(
        request_df[
            [
                "ID",
                "NAME",
                "SUBSTATION_ID",
                "REQUESTED_MW",
                "gross_annual_value",
                "risk_penalty_value",
                "risk_adjusted_value",
            ]
        ]
        .sort_values("risk_penalty_value", ascending=False)
        .head(10)
        .to_string(index=False)
    )

    print("\n=== BUDGET SCENARIOS ===")
    print(
        summary_df[
            [
                "level",
                "budget",
                "approved_requests",
                "approved_mw",
                "gross_revenue",
                "risk_penalty",
                "upgrade_cost_million",
                "net_annual_value",
                "critical_share",
                "funded_upgrades",
            ]
        ].to_string(index=False)
    )

    for _, row in summary_df.iterrows():
        level = row["level"]
        print(f"\n--- {level} ---")
        level_approved = approved_df[approved_df["level"] == level].copy()
        level_upgrades = upgrades_df[upgrades_df["level"] == level].copy()
        print(
            f"Approved requests: {int(row['approved_requests'])} | "
            f"Approved load: {float(row['approved_mw']):.0f} MW | "
            f"Net annual value: ${float(row['net_annual_value']):,.0f}"
        )
        if len(level_approved) > 0:
            print(
                level_approved[
                    [
                        "request_name",
                        "hyperscaler",
                        "requested_mw",
                        "substation_name",
                        "risk_adjusted_value",
                    ]
                ].to_string(index=False)
            )
        if len(level_upgrades) > 0:
            print("Funded upgrades:")
            print(
                level_upgrades[
                    [
                        "upgrade_id",
                        "substation_name",
                        "capacity_increase_mw",
                        "cost_million",
                        "enables_low_carbon",
                    ]
                ].to_string(index=False)
            )


def main() -> None:
    load_ontology()
    centrality_df, corridor_df, _ = build_graph_outputs()
    request_df = bind_risk_metrics(centrality_df)
    summary_df, approved_df, upgrades_df = solve_budget_scenarios()
    print_report(
        centrality_df,
        corridor_df,
        request_df,
        summary_df,
        approved_df,
        upgrades_df,
    )


if __name__ == "__main__":
    main()

Wrap up

You now know how to use RelationalAI’s agent skills to build a semantic model, inspect its structure, and apply it to a real business question.

Note

The logic and prescriptive reasoners that were created by running the prompts in this example will be suspended automatically after 30 minutes of inactivity.

By completing this tutorial, you saw that:

• RelationalAI can model your business domain semantically.
• Reasoners support graph analysis, rules, and optimization without complex data pipelines or moving data into separate systems.
• Agent skills can build and refine a semantic model from prompts.
• You can use skills to answer real business questions with answers grounded in your data and semantic model.

Where to go next

Use RAI with coding agents Learn more about how to integrate RelationalAI into your coding agent workflows to build more powerful AI applications.

Meet PyRel Build a semantic model step-by-step using PyRel to learn about its core concepts and capabilities.

Understand how PyRel works Dive deeper into PyRel with our in-depth guides on modeling, querying, reasoning, and more.

Start from a template Explore our template library to see how you can apply RelationalAI to common use cases across industries.