The RelationalAI Prescriptive Reasoner

The RelationalAI (RAI) Native App’s Prescriptive Reasoner helps you solve problems like scheduling, routing, pricing, and resource allocation directly in Snowflake using industry-grade solvers. Mixed-integer linear/quadratic optimization, continuous nonlinear optimization, and constraint programming are currently supported.

How It Works

Create a SolverModel from your RAI model.

First, define a RAI model. Then, create a new SolverModel instance from the model object:

import relationalai as rai
from relationalai.experimental import solvers


# Create a RAI model.
model = rai.Model("WeeklyShiftAssignment")

# Declare entity types and properties.

# Entity types.
Employee = model.Type("Employee")
Shift = model.Type("Shift")
Day = model.Type("Day")
Available = model.Type("Available")
Scenario = model.Type("Scenario")  # All possible employee–shift–day combinations.
Assignment = model.Type("Assignment")  # A Scenario that is assigned.

48 collapsed lines
# Properties.
Employee.name.declare()
Shift.name.declare()
Shift.capacity.declare()
Available.employee.declare()
Available.day.declare()
Scenario.employee.declare()
Scenario.shift.declare()
Scenario.day.declare()


# Define sample data.

# Employees.
with model.rule(dynamic=True):
    for name in ["Alice", "Bob", "Carol", "Dave", "Eve"]:
        Employee.add(name=name)

# Shifts.
with model.rule():
    Shift.add(name="Morning").set(capacity=2)
    Shift.add(name="Evening").set(capacity=3)

# Days of the week.
with model.rule(dynamic=True):
    for name in ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"]:
        Day.add(name=name)

# Employee-day availability.
with model.rule(dynamic=True):
    # Alice works weekdays only
    for name in ["Mon", "Tue", "Wed", "Thu", "Fri"]:
        Available.add(employee=Employee(name="Alice"), day=Day(name=name))
    # Bob works all days
    Available.add(employee=Employee(name="Bob"), day=Day())
    # Carol works weekends only
    for name in ["Sat", "Sun"]:
        Available.add(employee=Employee(name="Carol"), day=Day(name=name))
    # Dave works Mon/Wed/Fri
    for name in ["Mon", "Wed", "Fri"]:
        Available.add(employee=Employee(name="Dave"), day=Day(name=name))
    # Eve works Tue/Thu/Sat
    for name in ["Tue", "Thu", "Sat"]:
        Available.add(employee=Employee(name="Eve"), day=Day(name=name))

# All possible employee–shift–day combinations
with model.rule():
    Scenario.add(employee=Employee(), shift=Shift(), day=Day())

# Create a SolverModel instance from the model.
solver_model = solvers.SolverModel(model)

Declare variables, constraints, and the objective function.

Use the SolverModel object’s .variable() and .constraint() methods to declare variables and constraints, and the .max_objective() or .min_objective() methods to define the objective function:

import relationalai as rai
from relationalai.experimental import solvers

66 collapsed lines

# Create a RAI model.
model = rai.Model("WeeklyShiftAssignment")

# Declare entity types and properties.

# Entity types.
Employee = model.Type("Employee")
Shift = model.Type("Shift")
Day = model.Type("Day")
Available = model.Type("Available")
Scenario = model.Type("Scenario")  # All possible employee–shift–day combinations.
Assignment = model.Type("Assignment")  # A Scenario that is assigned.

# Properties.
Employee.name.declare()
Shift.name.declare()
Shift.capacity.declare()
Available.employee.declare()
Available.day.declare()
Scenario.employee.declare()
Scenario.shift.declare()
Scenario.day.declare()


# Define sample data.

# Employees.
with model.rule(dynamic=True):
    for name in ["Alice", "Bob", "Carol", "Dave", "Eve"]:
        Employee.add(name=name)

# Shifts.
with model.rule():
    Shift.add(name="Morning").set(capacity=2)
    Shift.add(name="Evening").set(capacity=3)

# Days of the week.
with model.rule(dynamic=True):
    for name in ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"]:
        Day.add(name=name)

# Employee-day availability.
with model.rule(dynamic=True):
    # Alice works weekdays only
    for name in ["Mon", "Tue", "Wed", "Thu", "Fri"]:
        Available.add(employee=Employee(name="Alice"), day=Day(name=name))
    # Bob works all days
    Available.add(employee=Employee(name="Bob"), day=Day())
    # Carol works weekends only
    for name in ["Sat", "Sun"]:
        Available.add(employee=Employee(name="Carol"), day=Day(name=name))
    # Dave works Mon/Wed/Fri
    for name in ["Mon", "Wed", "Fri"]:
        Available.add(employee=Employee(name="Dave"), day=Day(name=name))
    # Eve works Tue/Thu/Sat
    for name in ["Tue", "Thu", "Sat"]:
        Available.add(employee=Employee(name="Eve"), day=Day(name=name))

# All possible employee–shift–day combinations
with model.rule():
    Scenario.add(employee=Employee(), shift=Shift(), day=Day())

# Create a SolverModel instance from the model.
solver_model = solvers.SolverModel(model)


# Define solver variables.

# Scenario assignment: binary variable for each employee–shift–day combination.
with model.rule():
    scenario = Scenario()
    solver_model.variable(
        scenario,
        type="zero_one",
        name_args=["assigned", scenario.employee.name, scenario.day.name, scenario.shift.name],
    )

# Define Solver Constraints

# Unavailable employee–day pairs cannot be assigned.
with model.rule():
    scenario = Scenario()
    with model.not_found():
        Available(employee=scenario.employee, day=scenario.day)
    solver_model.constraint(
        solvers.not_(scenario),
        name_args=["unavailable", scenario.employee.name, scenario.day.name]
    )

40 collapsed lines
# If an employee works the morning shift, they can't work the evening shift.
with model.rule():
    works_morning = Scenario(shift=Shift(name="Morning"))
    works_evening = Scenario(
        shift=Shift(name="Evening"),
        employee=works_morning.employee,
        day=works_morning.day
    )
    solver_model.constraint(
        solvers.implies(works_morning, solvers.not_(works_evening))
    )

# At least one of Alice or Bob works Morning on Monday.
with model.rule():
    morning, monday = Shift(name="Morning"), Day(name="Mon")
    alice_works = Scenario(employee=Employee(name="Alice"), shift=morning, day=monday)
    bob_works = Scenario(employee=Employee(name="Bob"),   shift=morning, day=monday)
    solver_model.constraint(solvers.or_(alice_works, bob_works))

# If Alice works in the morning, she doesn't work in the evening. Otherwise, she does work in the evening.
with model.rule():
    alice = Employee(name="Alice")
    day = Available(employee=alice).day
    works_morning = Scenario(employee=alice, shift=Shift(name="Morning"), day=day)
    works_evening = Scenario(employee=alice, shift=Shift(name="Evening"), day=day)
    solver_model.constraint(
        solvers.if_then_else(works_morning, solvers.not_(works_evening), works_evening),
        name_args=["alice_shift", day.name]
    )

# Limit the number of employees assigned to each shift to its capacity.
with solvers.operators():
    scenario = Scenario()
    shift, day = scenario.shift, scenario.day
    num_shifts_per_day = solvers.count(scenario, per=[shift, day])
    solver_model.constraint(
        num_shifts_per_day <= shift.capacity,
        name_args=["shift_cap", shift.name, day.name]
    )


# Define solver objective.

# Maximize total assignments across the week.
with model.rule():
    scenario = Scenario()
    solver_model.max_objective(solvers.sum(scenario))

Use a Solver to solve the model.

Create a Solver instance and pass it to the SolverModel object’s .solve() method to find the optimal solution:

import relationalai as rai
from relationalai.experimental import solvers

142 collapsed lines

# Create a RAI model.
model = rai.Model("WeeklyShiftAssignment")

# Declare entity types and properties.

# Entity types.
Employee = model.Type("Employee")
Shift = model.Type("Shift")
Day = model.Type("Day")
Available = model.Type("Available")
Scenario = model.Type("Scenario")  # All possible employee–shift–day combinations.
Assignment = model.Type("Assignment")  # A Scenario that is assigned.

# Properties.
Employee.name.declare()
Shift.name.declare()
Shift.capacity.declare()
Available.employee.declare()
Available.day.declare()
Scenario.employee.declare()
Scenario.shift.declare()
Scenario.day.declare()


# Define sample data.

# Employees.
with model.rule(dynamic=True):
    for name in ["Alice", "Bob", "Carol", "Dave", "Eve"]:
        Employee.add(name=name)

# Shifts.
with model.rule():
    Shift.add(name="Morning").set(capacity=2)
    Shift.add(name="Evening").set(capacity=3)

# Days of the week.
with model.rule(dynamic=True):
    for name in ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"]:
        Day.add(name=name)

# Employee-day availability.
with model.rule(dynamic=True):
    # Alice works weekdays only
    for name in ["Mon", "Tue", "Wed", "Thu", "Fri"]:
        Available.add(employee=Employee(name="Alice"), day=Day(name=name))
    # Bob works all days
    Available.add(employee=Employee(name="Bob"), day=Day())
    # Carol works weekends only
    for name in ["Sat", "Sun"]:
        Available.add(employee=Employee(name="Carol"), day=Day(name=name))
    # Dave works Mon/Wed/Fri
    for name in ["Mon", "Wed", "Fri"]:
        Available.add(employee=Employee(name="Dave"), day=Day(name=name))
    # Eve works Tue/Thu/Sat
    for name in ["Tue", "Thu", "Sat"]:
        Available.add(employee=Employee(name="Eve"), day=Day(name=name))

# All possible employee–shift–day combinations
with model.rule():
    Scenario.add(employee=Employee(), shift=Shift(), day=Day())

# Create a SolverModel instance from the model.
solver_model = solvers.SolverModel(model)


# Define solver variables.

# Scenario assignment: binary variable for each employee–shift–day combination.
with model.rule():
    scenario = Scenario()
    solver_model.variable(
        scenario,
        type="zero_one",
        name_args=["assigned", scenario.employee.name, scenario.day.name, scenario.shift.name],
    )


# Define Solver Constraints

# Unavailable employee–day pairs cannot be assigned.
with model.rule():
    scenario = Scenario()
    with model.not_found():
        Available(employee=scenario.employee, day=scenario.day)
    solver_model.constraint(
        solvers.not_(scenario),
        name_args=["unavailable", scenario.employee.name, scenario.day.name]
    )

# If an employee works the morning shift, they can't work the evening shift.
with model.rule():
    works_morning = Scenario(shift=Shift(name="Morning"))
    works_evening = Scenario(
        shift=Shift(name="Evening"),
        employee=works_morning.employee,
        day=works_morning.day
    )
    solver_model.constraint(
        solvers.implies(works_morning, solvers.not_(works_evening))
    )

# At least one of Alice or Bob works Morning on Monday.
with model.rule():
    morning, monday = Shift(name="Morning"), Day(name="Mon")
    alice_works = Scenario(employee=Employee(name="Alice"), shift=morning, day=monday)
    bob_works = Scenario(employee=Employee(name="Bob"),   shift=morning, day=monday)
    solver_model.constraint(solvers.or_(alice_works, bob_works))

# If Alice works in the morning, she doesn't work in the evening. Otherwise, she does work in the evening.
with model.rule():
    alice = Employee(name="Alice")
    day = Available(employee=alice).day
    works_morning = Scenario(employee=alice, shift=Shift(name="Morning"), day=day)
    works_evening = Scenario(employee=alice, shift=Shift(name="Evening"), day=day)
    solver_model.constraint(
        solvers.if_then_else(works_morning, solvers.not_(works_evening), works_evening),
        name_args=["alice_shift", day.name]
    )

# Limit the number of employees assigned to each shift to its capacity.
with solvers.operators():
    scenario = Scenario()
    shift, day = scenario.shift, scenario.day
    num_shifts_per_day = solvers.count(scenario, per=[shift, day])
    solver_model.constraint(
        num_shifts_per_day <= shift.capacity,
        name_args=["shift_cap", shift.name, day.name]
    )


# Define solver objective.

# Maximize total assignments across the week.
with model.rule():
    scenario = Scenario()
    solver_model.max_objective(solvers.sum(scenario))


# Solve the solver model.


# Create a Solver instance.
solver = solvers.Solver("minizinc")

# Solve the model.
solver_model.solve(solver)

Augment your model using the solution.

After solving the model, you can use the SolverModel object’s .value() method to use solution values in logic that drives new entities or properties:

import relationalai as rai
from relationalai.experimental import solvers

151 collapsed lines

# Create a RAI model.
model = rai.Model("WeeklyShiftAssignment")

# Declare entity types and properties.

# Entity types.
Employee = model.Type("Employee")
Shift = model.Type("Shift")
Day = model.Type("Day")
Available = model.Type("Available")
Scenario = model.Type("Scenario")  # All possible employee–shift–day combinations.
Assignment = model.Type("Assignment")  # A Scenario that is assigned.

# Properties.
Employee.name.declare()
Shift.name.declare()
Shift.capacity.declare()
Available.employee.declare()
Available.day.declare()
Scenario.employee.declare()
Scenario.shift.declare()
Scenario.day.declare()


# Define sample data.

# Employees.
with model.rule(dynamic=True):
    for name in ["Alice", "Bob", "Carol", "Dave", "Eve"]:
        Employee.add(name=name)

# Shifts.
with model.rule():
    Shift.add(name="Morning").set(capacity=2)
    Shift.add(name="Evening").set(capacity=3)

# Days of the week.
with model.rule(dynamic=True):
    for name in ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"]:
        Day.add(name=name)

# Employee-day availability.
with model.rule(dynamic=True):
    # Alice works weekdays only
    for name in ["Mon", "Tue", "Wed", "Thu", "Fri"]:
        Available.add(employee=Employee(name="Alice"), day=Day(name=name))
    # Bob works all days
    Available.add(employee=Employee(name="Bob"), day=Day())
    # Carol works weekends only
    for name in ["Sat", "Sun"]:
        Available.add(employee=Employee(name="Carol"), day=Day(name=name))
    # Dave works Mon/Wed/Fri
    for name in ["Mon", "Wed", "Fri"]:
        Available.add(employee=Employee(name="Dave"), day=Day(name=name))
    # Eve works Tue/Thu/Sat
    for name in ["Tue", "Thu", "Sat"]:
        Available.add(employee=Employee(name="Eve"), day=Day(name=name))

# All possible employee–shift–day combinations
with model.rule():
    Scenario.add(employee=Employee(), shift=Shift(), day=Day())

# Create a SolverModel instance from the model.
solver_model = solvers.SolverModel(model)


# Define solver variables.

# Scenario assignment: binary variable for each employee–shift–day combination.
with model.rule():
    scenario = Scenario()
    solver_model.variable(
        scenario,
        type="zero_one",
        name_args=["assigned", scenario.employee.name, scenario.day.name, scenario.shift.name],
    )


# Define Solver Constraints

# Unavailable employee–day pairs cannot be assigned.
with model.rule():
    scenario = Scenario()
    with model.not_found():
        Available(employee=scenario.employee, day=scenario.day)
    solver_model.constraint(
        solvers.not_(scenario),
        name_args=["unavailable", scenario.employee.name, scenario.day.name]
    )

# If an employee works the morning shift, they can't work the evening shift.
with model.rule():
    works_morning = Scenario(shift=Shift(name="Morning"))
    works_evening = Scenario(
        shift=Shift(name="Evening"),
        employee=works_morning.employee,
        day=works_morning.day
    )
    solver_model.constraint(
        solvers.implies(works_morning, solvers.not_(works_evening))
    )

# At least one of Alice or Bob works Morning on Monday.
with model.rule():
    morning, monday = Shift(name="Morning"), Day(name="Mon")
    alice_works = Scenario(employee=Employee(name="Alice"), shift=morning, day=monday)
    bob_works = Scenario(employee=Employee(name="Bob"),   shift=morning, day=monday)
    solver_model.constraint(solvers.or_(alice_works, bob_works))

# If Alice works in the morning, she doesn't work in the evening. Otherwise, she does work in the evening.
with model.rule():
    alice = Employee(name="Alice")
    day = Available(employee=alice).day
    works_morning = Scenario(employee=alice, shift=Shift(name="Morning"), day=day)
    works_evening = Scenario(employee=alice, shift=Shift(name="Evening"), day=day)
    solver_model.constraint(
        solvers.if_then_else(works_morning, solvers.not_(works_evening), works_evening),
        name_args=["alice_shift", day.name]
    )

# Limit the number of employees assigned to each shift to its capacity.
with solvers.operators():
    scenario = Scenario()
    shift, day = scenario.shift, scenario.day
    num_shifts_per_day = solvers.count(scenario, per=[shift, day])
    solver_model.constraint(
        num_shifts_per_day <= shift.capacity,
        name_args=["shift_cap", shift.name, day.name]
    )


# Define solver objective.

# Maximize total assignments across the week.
with model.rule():
    scenario = Scenario()
    solver_model.max_objective(solvers.sum(scenario))


# Solve the solver model.


# Create a Solver instance.
solver = solvers.Solver("minizinc")

# Solve the model.
solver_model.solve(solver)


# Use the solution to define Assignments.

# Define Assignments from Scenarios whose solution value is 1.
with model.rule():
    scenario = Scenario()
    solver_model.value(scenario) == 1
    scenario.set(Assignment)

# View Dave's assignments.
with model.query() as select:
    assignment = Assignment(employee=Employee(name="Dave"))
    response = select(assignment.day.name, assignment.shift.name)
print("Dave's assignments:")
print(response.results)

Dave's assignments:
  name    name2
0  Fri  Morning
1  Mon  Morning
2  Wed  Evening

Supported Solver Backends

RAI supports the following solver backends for performing prescriptive reasoning:

Gurobi Supports mixed-integer linear and quadratic programs, nonlinear programs, and more.

HiGHS Supports mixed-integer linear and quadratic programs.

Ipopt Supports continuous nonlinear optimization problems.

MiniZinc Supports constraint programming, discrete optimization problems, and more.

Limitations

The Prescriptive Reasoner is currently in Preview and has some limitations:

Only Int64 and Float64 values are supported in variables, constraints, and objectives. See Supported Data Types for details.
Only single-objective optimization is supported, even if the selected solver backend supports multi-objective problems.
Some advanced solver features, such as MIP callbacks, are not supported.