Skip to content

This feature is currently in Preview.

Solve a decision problem

Solve a decision problem by calling Problem.solve() with a solver backend. This guide covers solving for feasibility, solving optimally, checking statuses, and diagnosing infeasible problems.

When you call Problem.solve(), PyRel runs a solve job in the RelationalAI Native App in Snowflake. Even though you call solve() from Python, the backend solve happens in Snowflake.

At a high level, a solve looks like this:

  1. PyRel translates your decision variables, constraints, and objective into a solve job.
  2. PyRel submits the solve job to the RelationalAI (RAI) Native App for Snowflake.
  3. The Native App runs the prescriptive reasoner service, which calls the backend solver.
  4. PyRel imports the results and exposes them as attributes on the Problem.

The following diagram combines the reasoner architecture with the job workflow:

Local Python processRAI Native App in SnowflakeModel objectProblem objectSolve jobPrescriptive reasoner serviceBackend solver(HiGHS / MiniZinc / ...) submit solve jobrun job(provisions a reasonerif needed)translate + solvesolution + logsupdate job statusimport resultsand metadata

Most beginner issues are formulation issues. Use these checks below as a lightweight pre-solve checklist:

  1. Check you actually created variables

    Problem.num_variables() should report a value greater than 0:

    from relationalai.semantics import Float, Integer, Model
    from relationalai.semantics.reasoners.prescriptive import Problem
    from relationalai.semantics.std import aggregates as agg
    m = Model("ShiftAssignment")
    # Declare the model's schema
    Worker = m.Concept("Worker", identify_by={"id": Integer})
    Shift = m.Concept("Shift", identify_by={"id": Integer})
    Worker.available_shifts = m.Relationship(f"{Worker} is available for {Shift}")
    Worker.cost_for_shift = m.Relationship(f"{Worker} working {Shift} has cost {Float:cost}")
    Shift.required_workers = m.Property(f"{Shift} requires {Integer:n} workers")
    49 collapsed lines
    # Define base facts.
    workers = m.data([
    {"id": 1, "name": "Alice"},
    {"id": 2, "name": "Bob"},
    {"id": 3, "name": "Chen"},
    ])
    shifts = m.data([
    {"id": 10, "name": "Morning"},
    {"id": 20, "name": "Evening"},
    ])
    availability = m.data([
    {"worker_id": 1, "shift_id": 10},
    {"worker_id": 2, "shift_id": 10},
    {"worker_id": 2, "shift_id": 20},
    {"worker_id": 3, "shift_id": 20},
    ])
    costs = m.data([
    {"worker_id": 1, "shift_id": 10, "cost": 9.0},
    {"worker_id": 2, "shift_id": 10, "cost": 10.0},
    {"worker_id": 2, "shift_id": 20, "cost": 8.0},
    {"worker_id": 3, "shift_id": 20, "cost": 11.0},
    ])
    required = m.data([
    {"shift_id": 10, "required_workers": 1},
    {"shift_id": 20, "required_workers": 1},
    ])
    m.define(
    Worker.new(workers.to_schema()),
    Shift.new(shifts.to_schema()),
    )
    m.define(
    Worker.lookup(id=availability.worker_id).available_shifts(
    Shift.lookup(id=availability.shift_id)
    )
    )
    m.define(
    Worker.lookup(id=costs.worker_id).cost_for_shift(
    Shift.lookup(id=costs.shift_id), costs.cost
    )
    )
    m.define(
    Shift.lookup(id=required.shift_id).required_workers(
    required.required_workers
    )
    )
    p = Problem(m, Float)
    25 collapsed lines
    # Declare a decision-variable relationship.
    Worker.x_assign = m.Relationship(
    f"{Worker} is assigned to {Shift} if {Float:assigned}"
    )
    # Define `Worker.x_assign` as a decision variable to be solved for,
    # scoped to available worker-shift pairs, and with binary values of 0 or 1.
    x = Float.ref("x")
    p.solve_for(
    Worker.x_assign(Shift, x),
    populate=True,
    name=["assign", Worker.id, Shift.id],
    where=[Worker.available_shifts(Shift)],
    type="bin",
    lower=0,
    upper=1,
    )
    # Each shift must be covered by the required number of workers.
    assigned_per_shift = agg.sum(x).where(Worker.x_assign(Shift, x)).per(Shift)
    p.satisfy(m.require(assigned_per_shift == Shift.required_workers))
    # Worker capacity: each worker is assigned to at most one shift.
    assigned_per_worker = agg.sum(x).where(Worker.x_assign(Shift, x)).per(Worker)
    p.satisfy(m.require(assigned_per_worker <= 1))
    m.select(p.num_variables()).inspect()
  2. Check that constraints have been added

    If your problem has constraints, then Problem.num_constraints() should report a value greater than 0:

    m.select(p.num_constraints()).inspect()
  3. Spot check the entire problem formulation

    Use Problem.display() to get a complete description of the decision problem:

    p.display()

If you don’t add an objective to Problem, the solver treats the problem as a feasibility problem. It finds any solution that satisfies all solution constraints without optimizing for anything in particular.

Pass a backend name to Problem.solve() to solve the problem with a specific solver:

from relationalai.semantics import Float, Integer, Model
from relationalai.semantics.reasoners.prescriptive import Problem
from relationalai.semantics.std import aggregates as agg
m = Model("ShiftAssignment")
# Declare the model's schema
Worker = m.Concept("Worker", identify_by={"id": Integer})
Shift = m.Concept("Shift", identify_by={"id": Integer})
Worker.available_shifts = m.Relationship(f"{Worker} is available for {Shift}")
Worker.cost_for_shift = m.Relationship(f"{Worker} working {Shift} has cost {Float:cost}")
Shift.required_workers = m.Property(f"{Shift} requires {Integer:n} workers")
# Define base facts.
68 collapsed lines
workers = m.data([
{"id": 1, "name": "Alice"},
{"id": 2, "name": "Bob"},
{"id": 3, "name": "Chen"},
])
shifts = m.data([
{"id": 10, "name": "Morning"},
{"id": 20, "name": "Evening"},
])
availability = m.data([
{"worker_id": 1, "shift_id": 10},
{"worker_id": 2, "shift_id": 10},
{"worker_id": 2, "shift_id": 20},
{"worker_id": 3, "shift_id": 20},
])
costs = m.data([
{"worker_id": 1, "shift_id": 10, "cost": 9.0},
{"worker_id": 2, "shift_id": 10, "cost": 10.0},
{"worker_id": 2, "shift_id": 20, "cost": 8.0},
{"worker_id": 3, "shift_id": 20, "cost": 11.0},
])
required = m.data([
{"shift_id": 10, "required_workers": 1},
{"shift_id": 20, "required_workers": 1},
])
m.define(
Worker.new(workers.to_schema()),
Shift.new(shifts.to_schema()),
)
m.define(
Worker.lookup(id=availability.worker_id).available_shifts(
Shift.lookup(id=availability.shift_id)
)
)
m.define(
Worker.lookup(id=costs.worker_id).cost_for_shift(
Shift.lookup(id=costs.shift_id), costs.cost
)
)
m.define(
Shift.lookup(id=required.shift_id).required_workers(
required.required_workers
)
)
p = Problem(m, Float)
# Declare a decision-variable relationship.
Worker.x_assign = m.Relationship(
f"{Worker} is assigned to {Shift} if {Float:assigned}"
)
# Define `Worker.x_assign` as a decision variable to be solved for,
# scoped to available worker-shift pairs, and with binary values of 0 or 1.
x = Float.ref("x")
p.solve_for(
Worker.x_assign(Shift, x),
populate=True,
name=["assign", Worker.id, Shift.id],
where=[Worker.available_shifts(Shift)],
type="bin",
lower=0,
upper=1,
)
# Each shift must be covered by the required number of workers.
assigned_per_shift = agg.sum(x).where(Worker.x_assign(Shift, x)).per(Shift)
p.satisfy(m.require(assigned_per_shift == Shift.required_workers))
# Worker capacity: each worker is assigned to at most one shift.
assigned_per_worker = agg.sum(x).where(Worker.x_assign(Shift, x)).per(Worker)
p.satisfy(m.require(assigned_per_worker <= 1))
p.solve("highs")
print(p.solve_info().termination_status)
  • No objective is added to the Problem, so this is a feasibility solve.
  • p.solve("highs") calls the HiGHS solver to solve the problem.
  • p.solve_info().termination_status reports the solver’s termination status, which indicates whether a feasible solution was found or not.
  • Calling solve() before declaring any variables is an error.
  • Use "minizinc" when you have a pure discrete feasibility problem and you want a constraint-programming solver. In that case, create the problem with Problem(m, Integer).

Solve an optimization problem by adding an objective and then calling Problem.solve():

  1. Define an objective

    Add an objective expression involving at least one decision variable and use Problem.minimize() or Problem.maximize() to set the optimization direction:

    from relationalai.semantics import Float, Integer, Model
    from relationalai.semantics.reasoners.prescriptive import Problem
    from relationalai.semantics.std import aggregates as agg
    m = Model("ShiftAssignment")
    # Declare the model's schema
    Worker = m.Concept("Worker", identify_by={"id": Integer})
    Shift = m.Concept("Shift", identify_by={"id": Integer})
    Worker.available_shifts = m.Relationship(f"{Worker} is available for {Shift}")
    Worker.cost_for_shift = m.Relationship(f"{Worker} working {Shift} has cost {Float:cost}")
    Shift.required_workers = m.Property(f"{Shift} requires {Integer:n} workers")
    49 collapsed lines
    # Define base facts.
    workers = m.data([
    {"id": 1, "name": "Alice"},
    {"id": 2, "name": "Bob"},
    {"id": 3, "name": "Chen"},
    ])
    shifts = m.data([
    {"id": 10, "name": "Morning"},
    {"id": 20, "name": "Evening"},
    ])
    availability = m.data([
    {"worker_id": 1, "shift_id": 10},
    {"worker_id": 2, "shift_id": 10},
    {"worker_id": 2, "shift_id": 20},
    {"worker_id": 3, "shift_id": 20},
    ])
    costs = m.data([
    {"worker_id": 1, "shift_id": 10, "cost": 9.0},
    {"worker_id": 2, "shift_id": 10, "cost": 10.0},
    {"worker_id": 2, "shift_id": 20, "cost": 8.0},
    {"worker_id": 3, "shift_id": 20, "cost": 11.0},
    ])
    required = m.data([
    {"shift_id": 10, "required_workers": 1},
    {"shift_id": 20, "required_workers": 1},
    ])
    m.define(
    Worker.new(workers.to_schema()),
    Shift.new(shifts.to_schema()),
    )
    m.define(
    Worker.lookup(id=availability.worker_id).available_shifts(
    Shift.lookup(id=availability.shift_id)
    )
    )
    m.define(
    Worker.lookup(id=costs.worker_id).cost_for_shift(
    Shift.lookup(id=costs.shift_id), costs.cost
    )
    )
    m.define(
    Shift.lookup(id=required.shift_id).required_workers(
    required.required_workers
    )
    )
    # Create a decision problem.
    p = Problem(m, Float)
    25 collapsed lines
    # Declare a decision-variable relationship.
    Worker.x_assign = m.Relationship(
    f"{Worker} is assigned to {Shift} if {Float:assigned}"
    )
    # Define `Worker.x_assign` as a decision variable to be solved for,
    # scoped to available worker-shift pairs, and with binary values of 0 or 1.
    x = Float.ref("x")
    p.solve_for(
    Worker.x_assign(Shift, x),
    populate=True,
    name=["assign", Worker.id, Shift.id],
    where=[Worker.available_shifts(Shift)],
    type="bin",
    lower=0,
    upper=1,
    )
    # Each shift must be covered by the required number of workers.
    assigned_per_shift = agg.sum(x).where(Worker.x_assign(Shift, x)).per(Shift)
    p.satisfy(m.require(assigned_per_shift == Shift.required_workers))
    # Worker capacity: each worker is assigned to at most one shift.
    assigned_per_worker = agg.sum(x).where(Worker.x_assign(Shift, x)).per(Worker)
    p.satisfy(m.require(assigned_per_worker <= 1))
    # Minimize the total cost of assigned shifts.
    cost = Float.ref("cost")
    p.minimize(
    agg.sum(cost * x).where(
    Worker.x_assign(Shift, x),
    Worker.cost_for_shift(Shift, cost),
    )
    )
  2. Solve and check the result

    Call Problem.solve() and inspect the termination status before you treat the solution as optimal:

    p.solve("highs")
    info = p.solve_info()
    print(info.termination_status)
    print(info.objective_value)
  • p.minimize(...) adds a minimization objective.
  • agg.sum(cost * x).where(...) defines the expression to minimize.
  • p.solve("highs") calls the HiGHS solver to solve the problem.
  • p.solve_info().termination_status tells you whether the solver proved optimality (or stopped for another reason).
  • p.solve_info().objective_value reports the objective value for the returned solution.
  • Objective expressions must reference at least one decision variable. If the objective is constant, it is rejected.
  • You can also add a maximization objective with p.maximize(...).
  • Use one objective per solve. If you need to balance competing goals, combine them into one objective expression or solve separate scenarios.
  • If you set a time limit, a feasible (but not proven optimal) solution can still be useful. Check p.solve_info().termination_status before treating the result as optimal.

After a solve, you can inspect additional solver metadata on the Problem. This is useful when you need to report solve time, confirm which solver version ran, or quickly summarize the result.

Call p.solve_info() and use .display() on the result to print a short, human-readable summary:

from relationalai.semantics import Float, Integer, Model
from relationalai.semantics.reasoners.prescriptive import Problem
from relationalai.semantics.std import aggregates as agg
m = Model("ShiftAssignment")
# Declare the model's schema
Worker = m.Concept("Worker", identify_by={"id": Integer})
Shift = m.Concept("Shift", identify_by={"id": Integer})
Worker.available_shifts = m.Relationship(f"{Worker} is available for {Shift}")
Worker.cost_for_shift = m.Relationship(f"{Worker} working {Shift} has cost {Float:cost}")
Shift.required_workers = m.Property(f"{Shift} requires {Integer:n} workers")
# Define base facts.
77 collapsed lines
workers = m.data([
{"id": 1, "name": "Alice"},
{"id": 2, "name": "Bob"},
{"id": 3, "name": "Chen"},
])
shifts = m.data([
{"id": 10, "name": "Morning"},
{"id": 20, "name": "Evening"},
])
availability = m.data([
{"worker_id": 1, "shift_id": 10},
{"worker_id": 2, "shift_id": 10},
{"worker_id": 2, "shift_id": 20},
{"worker_id": 3, "shift_id": 20},
])
costs = m.data([
{"worker_id": 1, "shift_id": 10, "cost": 9.0},
{"worker_id": 2, "shift_id": 10, "cost": 10.0},
{"worker_id": 2, "shift_id": 20, "cost": 8.0},
{"worker_id": 3, "shift_id": 20, "cost": 11.0},
])
required = m.data([
{"shift_id": 10, "required_workers": 1},
{"shift_id": 20, "required_workers": 1},
])
m.define(
Worker.new(workers.to_schema()),
Shift.new(shifts.to_schema()),
)
m.define(
Worker.lookup(id=availability.worker_id).available_shifts(
Shift.lookup(id=availability.shift_id)
)
)
m.define(
Worker.lookup(id=costs.worker_id).cost_for_shift(
Shift.lookup(id=costs.shift_id), costs.cost
)
)
m.define(
Shift.lookup(id=required.shift_id).required_workers(
required.required_workers
)
)
# Create a decision problem.
p = Problem(m, Float)
# Declare a decision-variable relationship.
Worker.x_assign = m.Relationship(
f"{Worker} is assigned to {Shift} if {Float:assigned}"
)
# Define `Worker.x_assign` as a decision variable to be solved for,
# scoped to available worker-shift pairs, and with binary values of 0 or 1.
x = Float.ref("x")
p.solve_for(
Worker.x_assign(Shift, x),
populate=True,
name=["assign", Worker.id, Shift.id],
where=[Worker.available_shifts(Shift)],
type="bin",
lower=0,
upper=1,
)
# Each shift must be covered by the required number of workers.
assigned_per_shift = agg.sum(x).where(Worker.x_assign(Shift, x)).per(Shift)
p.satisfy(m.require(assigned_per_shift == Shift.required_workers))
# Worker capacity: each worker is assigned to at most one shift.
assigned_per_worker = agg.sum(x).where(Worker.x_assign(Shift, x)).per(Worker)
p.satisfy(m.require(assigned_per_worker <= 1))
# Minimize the total cost of assigned shifts.
cost = Float.ref("cost")
p.minimize(
agg.sum(cost * x).where(
Worker.x_assign(Shift, x),
Worker.cost_for_shift(Shift, cost),
)
)
p.solve("highs")
# Display a summary of the solve results
info = p.solve_info()
info.display()
# Or inspect specific metadata attributes
print("status:", info.termination_status)
print("objective:", info.objective_value) # None if the solve has no objective
print("solve time (sec):", info.solve_time_sec)
print("solver version:", info.solver_version)

The solve_info() summary describes the solve as a whole. To inspect individual parts of your formulation instead, query the problem’s variables, constraints, and objectives directly. There are two ways to reach them:

To limit how long the solver runs, pass time_limit_sec with a value in seconds to Problem.solve():

from relationalai.semantics import Float, Integer, Model
from relationalai.semantics.reasoners.prescriptive import Problem
from relationalai.semantics.std import aggregates as agg
m = Model("ShiftAssignment")
88 collapsed lines
# Declare the model's schema
Worker = m.Concept("Worker", identify_by={"id": Integer})
Shift = m.Concept("Shift", identify_by={"id": Integer})
Worker.available_shifts = m.Relationship(f"{Worker} is available for {Shift}")
Worker.cost_for_shift = m.Relationship(f"{Worker} working {Shift} has cost {Float:cost}")
Shift.required_workers = m.Property(f"{Shift} requires {Integer:n} workers")
# Define base facts.
workers = m.data([
{"id": 1, "name": "Alice"},
{"id": 2, "name": "Bob"},
{"id": 3, "name": "Chen"},
])
shifts = m.data([
{"id": 10, "name": "Morning"},
{"id": 20, "name": "Evening"},
])
availability = m.data([
{"worker_id": 1, "shift_id": 10},
{"worker_id": 2, "shift_id": 10},
{"worker_id": 2, "shift_id": 20},
{"worker_id": 3, "shift_id": 20},
])
costs = m.data([
{"worker_id": 1, "shift_id": 10, "cost": 9.0},
{"worker_id": 2, "shift_id": 10, "cost": 10.0},
{"worker_id": 2, "shift_id": 20, "cost": 8.0},
{"worker_id": 3, "shift_id": 20, "cost": 11.0},
])
required = m.data([
{"shift_id": 10, "required_workers": 1},
{"shift_id": 20, "required_workers": 1},
])
m.define(
Worker.new(workers.to_schema()),
Shift.new(shifts.to_schema()),
)
m.define(
Worker.lookup(id=availability.worker_id).available_shifts(
Shift.lookup(id=availability.shift_id)
)
)
m.define(
Worker.lookup(id=costs.worker_id).cost_for_shift(
Shift.lookup(id=costs.shift_id), costs.cost
)
)
m.define(
Shift.lookup(id=required.shift_id).required_workers(
required.required_workers
)
)
p = Problem(m, Float)
Worker.x_assign = m.Relationship(
f"{Worker} is assigned to {Shift} if {Float:assigned}"
)
x = Float.ref("x")
p.solve_for(
Worker.x_assign(Shift, x),
populate=True,
name=["assign", Worker.id, Shift.id],
where=[Worker.available_shifts(Shift)],
type="bin",
lower=0,
upper=1,
)
assigned_per_shift = agg.sum(x).where(Worker.x_assign(Shift, x)).per(Shift)
p.satisfy(m.require(assigned_per_shift == Shift.required_workers))
assigned_per_worker = agg.sum(x).where(Worker.x_assign(Shift, x)).per(Worker)
p.satisfy(m.require(assigned_per_worker <= 1))
cost = Float.ref("cost")
p.minimize(
agg.sum(cost * x).where(
Worker.x_assign(Shift, x),
Worker.cost_for_shift(Shift, cost),
)
)
p.solve(
"highs",
time_limit_sec=60,
)
info = p.solve_info()
print("status:", info.termination_status)
print("objective:", info.objective_value)
  • time_limit_sec=60 limits the solver to run for at most 60 seconds.
  • p.solve_info().termination_status indicates whether the solve completed, hit the time limit, or failed.
  • When a time limit is reached, the returned solution can be feasible but not proven optimal.
  • time_limit_sec is a solver-independent option that works with any backend.

You can set optimality gap tolerances to accept a near-optimal solution and stop the solver earlier. Use:

  • relative_gap_tolerance to stop when the relative optimality gap is below a threshold.
  • absolute_gap_tolerance to stop when the absolute optimality gap is below a threshold.

Pass these as arguments to Problem.solve():

from relationalai.semantics import Float, Integer, Model
from relationalai.semantics.reasoners.prescriptive import Problem
from relationalai.semantics.std import aggregates as agg
m = Model("ShiftAssignment")
88 collapsed lines
# Declare the model's schema
Worker = m.Concept("Worker", identify_by={"id": Integer})
Shift = m.Concept("Shift", identify_by={"id": Integer})
Worker.available_shifts = m.Relationship(f"{Worker} is available for {Shift}")
Worker.cost_for_shift = m.Relationship(f"{Worker} working {Shift} has cost {Float:cost}")
Shift.required_workers = m.Property(f"{Shift} requires {Integer:n} workers")
# Define base facts.
workers = m.data([
{"id": 1, "name": "Alice"},
{"id": 2, "name": "Bob"},
{"id": 3, "name": "Chen"},
])
shifts = m.data([
{"id": 10, "name": "Morning"},
{"id": 20, "name": "Evening"},
])
availability = m.data([
{"worker_id": 1, "shift_id": 10},
{"worker_id": 2, "shift_id": 10},
{"worker_id": 2, "shift_id": 20},
{"worker_id": 3, "shift_id": 20},
])
costs = m.data([
{"worker_id": 1, "shift_id": 10, "cost": 9.0},
{"worker_id": 2, "shift_id": 10, "cost": 10.0},
{"worker_id": 2, "shift_id": 20, "cost": 8.0},
{"worker_id": 3, "shift_id": 20, "cost": 11.0},
])
required = m.data([
{"shift_id": 10, "required_workers": 1},
{"shift_id": 20, "required_workers": 1},
])
m.define(
Worker.new(workers.to_schema()),
Shift.new(shifts.to_schema()),
)
m.define(
Worker.lookup(id=availability.worker_id).available_shifts(
Shift.lookup(id=availability.shift_id)
)
)
m.define(
Worker.lookup(id=costs.worker_id).cost_for_shift(
Shift.lookup(id=costs.shift_id), costs.cost
)
)
m.define(
Shift.lookup(id=required.shift_id).required_workers(
required.required_workers
)
)
p = Problem(m, Float)
Worker.x_assign = m.Relationship(
f"{Worker} is assigned to {Shift} if {Float:assigned}"
)
x = Float.ref("x")
p.solve_for(
Worker.x_assign(Shift, x),
populate=True,
name=["assign", Worker.id, Shift.id],
where=[Worker.available_shifts(Shift)],
type="bin",
lower=0,
upper=1,
)
assigned_per_shift = agg.sum(x).where(Worker.x_assign(Shift, x)).per(Shift)
p.satisfy(m.require(assigned_per_shift == Shift.required_workers))
assigned_per_worker = agg.sum(x).where(Worker.x_assign(Shift, x)).per(Worker)
p.satisfy(m.require(assigned_per_worker <= 1))
cost = Float.ref("cost")
p.minimize(
agg.sum(cost * x).where(
Worker.x_assign(Shift, x),
Worker.cost_for_shift(Shift, cost),
)
)
p.solve(
"highs",
relative_gap_tolerance=0.01,
)
info = p.solve_info()
print("status:", info.termination_status)
print("objective:", info.objective_value)
  • relative_gap_tolerance=0.01 asks the solver to stop once the relative optimality gap is below 1%. absolute_gap_tolerance works similarly, but with an absolute gap threshold.
  • p.solve_info().termination_status is printed to check the status before accepting the solution.
  • Optimality gap tolerances are supported by some solvers, but not all. If a solver does not support the specified tolerance, it is ignored.

Solver backends have many options that you can tune to improve solve time or influence the returned solution. You can pass backend-specific options as keyword arguments to Problem.solve() and they will be passed through to the solver.

For example, you can pass Gurobi parameters as keyword arguments when you solve with "gurobi":

from relationalai.semantics import Float, Integer, Model
from relationalai.semantics.reasoners.prescriptive import Problem
from relationalai.semantics.std import aggregates as agg
m = Model("ShiftAssignment")
88 collapsed lines
# Declare the model's schema
Worker = m.Concept("Worker", identify_by={"id": Integer})
Shift = m.Concept("Shift", identify_by={"id": Integer})
Worker.available_shifts = m.Relationship(f"{Worker} is available for {Shift}")
Worker.cost_for_shift = m.Relationship(f"{Worker} working {Shift} has cost {Float:cost}")
Shift.required_workers = m.Property(f"{Shift} requires {Integer:n} workers")
# Define base facts.
workers = m.data([
{"id": 1, "name": "Alice"},
{"id": 2, "name": "Bob"},
{"id": 3, "name": "Chen"},
])
shifts = m.data([
{"id": 10, "name": "Morning"},
{"id": 20, "name": "Evening"},
])
availability = m.data([
{"worker_id": 1, "shift_id": 10},
{"worker_id": 2, "shift_id": 10},
{"worker_id": 2, "shift_id": 20},
{"worker_id": 3, "shift_id": 20},
])
costs = m.data([
{"worker_id": 1, "shift_id": 10, "cost": 9.0},
{"worker_id": 2, "shift_id": 10, "cost": 10.0},
{"worker_id": 2, "shift_id": 20, "cost": 8.0},
{"worker_id": 3, "shift_id": 20, "cost": 11.0},
])
required = m.data([
{"shift_id": 10, "required_workers": 1},
{"shift_id": 20, "required_workers": 1},
])
m.define(
Worker.new(workers.to_schema()),
Shift.new(shifts.to_schema()),
)
m.define(
Worker.lookup(id=availability.worker_id).available_shifts(
Shift.lookup(id=availability.shift_id)
)
)
m.define(
Worker.lookup(id=costs.worker_id).cost_for_shift(
Shift.lookup(id=costs.shift_id), costs.cost
)
)
m.define(
Shift.lookup(id=required.shift_id).required_workers(
required.required_workers
)
)
p = Problem(m, Float)
Worker.x_assign = m.Relationship(
f"{Worker} is assigned to {Shift} if {Float:assigned}"
)
x = Float.ref("x")
p.solve_for(
Worker.x_assign(Shift, x),
populate=True,
name=["assign", Worker.id, Shift.id],
where=[Worker.available_shifts(Shift)],
type="bin",
lower=0,
upper=1,
)
assigned_per_shift = agg.sum(x).where(Worker.x_assign(Shift, x)).per(Shift)
p.satisfy(m.require(assigned_per_shift == Shift.required_workers))
assigned_per_worker = agg.sum(x).where(Worker.x_assign(Shift, x)).per(Worker)
p.satisfy(m.require(assigned_per_worker <= 1))
cost = Float.ref("cost")
p.minimize(
agg.sum(cost * x).where(
Worker.x_assign(Shift, x),
Worker.cost_for_shift(Shift, cost),
)
)
p.solve(
"gurobi",
MIPFocus=1,
Presolve=2,
Threads=0,
)
  • Raw solver parameters must be int, float, str, or bool values. This means they must be plain scalar configuration values, not PyRel objects like concept types or expression values. It also means you cannot pass structured values like lists or dictionaries.
  • PyRel emits a warning when you pass any backend-specific parameters. It does this because raw solver parameters may not be portable across solvers and can change meaning across backends.

When you need to debug a formulation or share a solver-native model with someone, request a text representation during solve. To request a text representation, pass print_format to Problem.solve() and read the result from p.solve_info().printed_model:

from relationalai.semantics import Float, Integer, Model
from relationalai.semantics.reasoners.prescriptive import Problem
from relationalai.semantics.std import aggregates as agg
m = Model("ShiftAssignment")
88 collapsed lines
# Declare the model's schema
Worker = m.Concept("Worker", identify_by={"id": Integer})
Shift = m.Concept("Shift", identify_by={"id": Integer})
Worker.available_shifts = m.Relationship(f"{Worker} is available for {Shift}")
Worker.cost_for_shift = m.Relationship(f"{Worker} working {Shift} has cost {Float:cost}")
Shift.required_workers = m.Property(f"{Shift} requires {Integer:n} workers")
# Define base facts.
workers = m.data([
{"id": 1, "name": "Alice"},
{"id": 2, "name": "Bob"},
{"id": 3, "name": "Chen"},
])
shifts = m.data([
{"id": 10, "name": "Morning"},
{"id": 20, "name": "Evening"},
])
availability = m.data([
{"worker_id": 1, "shift_id": 10},
{"worker_id": 2, "shift_id": 10},
{"worker_id": 2, "shift_id": 20},
{"worker_id": 3, "shift_id": 20},
])
costs = m.data([
{"worker_id": 1, "shift_id": 10, "cost": 9.0},
{"worker_id": 2, "shift_id": 10, "cost": 10.0},
{"worker_id": 2, "shift_id": 20, "cost": 8.0},
{"worker_id": 3, "shift_id": 20, "cost": 11.0},
])
required = m.data([
{"shift_id": 10, "required_workers": 1},
{"shift_id": 20, "required_workers": 1},
])
m.define(
Worker.new(workers.to_schema()),
Shift.new(shifts.to_schema()),
)
m.define(
Worker.lookup(id=availability.worker_id).available_shifts(
Shift.lookup(id=availability.shift_id)
)
)
m.define(
Worker.lookup(id=costs.worker_id).cost_for_shift(
Shift.lookup(id=costs.shift_id), costs.cost
)
)
m.define(
Shift.lookup(id=required.shift_id).required_workers(
required.required_workers
)
)
p = Problem(m, Float)
Worker.x_assign = m.Relationship(
f"{Worker} is assigned to {Shift} if {Float:assigned}"
)
x = Float.ref("x")
p.solve_for(
Worker.x_assign(Shift, x),
populate=True,
name=["assign", Worker.id, Shift.id],
where=[Worker.available_shifts(Shift)],
type="bin",
lower=0,
upper=1,
)
assigned_per_shift = agg.sum(x).where(Worker.x_assign(Shift, x)).per(Shift)
p.satisfy(m.require(assigned_per_shift == Shift.required_workers))
assigned_per_worker = agg.sum(x).where(Worker.x_assign(Shift, x)).per(Worker)
p.satisfy(m.require(assigned_per_worker <= 1))
cost = Float.ref("cost")
p.minimize(
agg.sum(cost * x).where(
Worker.x_assign(Shift, x),
Worker.cost_for_shift(Shift, cost),
)
)
p.solve(
"highs",
print_format="lp", # Request a printed model representation in LP format
)
print(p.solve_info().printed_model)

You can also set print_only=True to print the model without solving:

from relationalai.semantics import Float, Integer, Model
from relationalai.semantics.reasoners.prescriptive import Problem
from relationalai.semantics.std import aggregates as agg
m = Model("ShiftAssignment")
88 collapsed lines
# Declare the model's schema
Worker = m.Concept("Worker", identify_by={"id": Integer})
Shift = m.Concept("Shift", identify_by={"id": Integer})
Worker.available_shifts = m.Relationship(f"{Worker} is available for {Shift}")
Worker.cost_for_shift = m.Relationship(f"{Worker} working {Shift} has cost {Float:cost}")
Shift.required_workers = m.Property(f"{Shift} requires {Integer:n} workers")
# Define base facts.
workers = m.data([
{"id": 1, "name": "Alice"},
{"id": 2, "name": "Bob"},
{"id": 3, "name": "Chen"},
])
shifts = m.data([
{"id": 10, "name": "Morning"},
{"id": 20, "name": "Evening"},
])
availability = m.data([
{"worker_id": 1, "shift_id": 10},
{"worker_id": 2, "shift_id": 10},
{"worker_id": 2, "shift_id": 20},
{"worker_id": 3, "shift_id": 20},
])
costs = m.data([
{"worker_id": 1, "shift_id": 10, "cost": 9.0},
{"worker_id": 2, "shift_id": 10, "cost": 10.0},
{"worker_id": 2, "shift_id": 20, "cost": 8.0},
{"worker_id": 3, "shift_id": 20, "cost": 11.0},
])
required = m.data([
{"shift_id": 10, "required_workers": 1},
{"shift_id": 20, "required_workers": 1},
])
m.define(
Worker.new(workers.to_schema()),
Shift.new(shifts.to_schema()),
)
m.define(
Worker.lookup(id=availability.worker_id).available_shifts(
Shift.lookup(id=availability.shift_id)
)
)
m.define(
Worker.lookup(id=costs.worker_id).cost_for_shift(
Shift.lookup(id=costs.shift_id), costs.cost
)
)
m.define(
Shift.lookup(id=required.shift_id).required_workers(
required.required_workers
)
)
p = Problem(m, Float)
Worker.x_assign = m.Relationship(
f"{Worker} is assigned to {Shift} if {Float:assigned}"
)
x = Float.ref("x")
p.solve_for(
Worker.x_assign(Shift, x),
populate=True,
name=["assign", Worker.id, Shift.id],
where=[Worker.available_shifts(Shift)],
type="bin",
lower=0,
upper=1,
)
assigned_per_shift = agg.sum(x).where(Worker.x_assign(Shift, x)).per(Shift)
p.satisfy(m.require(assigned_per_shift == Shift.required_workers))
assigned_per_worker = agg.sum(x).where(Worker.x_assign(Shift, x)).per(Worker)
p.satisfy(m.require(assigned_per_worker <= 1))
cost = Float.ref("cost")
p.minimize(
agg.sum(cost * x).where(
Worker.x_assign(Shift, x),
Worker.cost_for_shift(Shift, cost),
)
)
p.solve(
"highs",
print_only=True, # Do not run the solver, just print the model
print_format="lp",
)
print(p.solve_info().printed_model)

Supported print_format values are:

  • "moi"
  • "latex"
  • "mof"
  • "lp"
  • "mps"
  • "nl"

Check error details when termination status is not OPTIMAL

Section titled “Check error details when termination status is not OPTIMAL”

If the solver does not prove optimality, p.solve_info().termination_status tells you what happened. When results are imported, p.solve_info().error can include backend-provided diagnostic text that helps you decide what to try next.

Use the following pattern after p.solve(...) to check the status and print p.solve_info().error when it is available:

from relationalai.semantics import Float, Integer, Model
from relationalai.semantics.reasoners.prescriptive import Problem
from relationalai.semantics.std import aggregates as agg
m = Model("ShiftAssignment")
76 collapsed lines
Worker = m.Concept("Worker", identify_by={"id": Integer})
Shift = m.Concept("Shift", identify_by={"id": Integer})
Worker.available_shifts = m.Relationship(f"{Worker} is available for {Shift}")
Worker.cost_for_shift = m.Relationship(f"{Worker} working {Shift} has cost {Float:cost}")
Shift.required_workers = m.Property(f"{Shift} requires {Integer:n} workers")
workers = m.data([
{"id": 1, "name": "Alice"},
{"id": 2, "name": "Bob"},
])
shifts = m.data([
{"id": 10, "name": "Morning"},
])
availability = m.data([
{"worker_id": 1, "shift_id": 10},
{"worker_id": 2, "shift_id": 10},
])
costs = m.data([
{"worker_id": 1, "shift_id": 10, "cost": 9.0},
{"worker_id": 2, "shift_id": 10, "cost": 10.0},
])
required = m.data([
{"shift_id": 10, "required_workers": 1},
])
m.define(
Worker.new(workers.to_schema()),
Shift.new(shifts.to_schema()),
)
m.define(
Worker.lookup(id=availability.worker_id).available_shifts(
Shift.lookup(id=availability.shift_id)
)
)
m.define(
Worker.lookup(id=costs.worker_id).cost_for_shift(
Shift.lookup(id=costs.shift_id), costs.cost
)
)
m.define(
Shift.lookup(id=required.shift_id).required_workers(
required.required_workers
)
)
p = Problem(m, Float)
Worker.x_assign = m.Relationship(
f"{Worker} is assigned to {Shift} if {Float:assigned}"
)
x = Float.ref("x")
p.solve_for(
Worker.x_assign(Shift, x),
populate=True,
name=["assign", Worker.id, Shift.id],
where=[Worker.available_shifts(Shift)],
type="bin",
lower=0,
upper=1,
)
assigned_per_shift = agg.sum(x).where(Worker.x_assign(Shift, x)).per(Shift)
p.satisfy(m.require(assigned_per_shift == Shift.required_workers))
cost = Float.ref("cost")
p.minimize(
agg.sum(cost * x).where(
Worker.x_assign(Shift, x),
Worker.cost_for_shift(Shift, cost),
)
)
p.solve("highs")
info = p.solve_info()
print("termination status:", info.termination_status)
if info.termination_status != "OPTIMAL":
print("error:", info.error)
  • Only treat the solution as optimal when p.solve_info().termination_status == "OPTIMAL".
  • Use p.solve_info().error as best-effort diagnostic text when the solve returns a non-OPTIMAL status.
  • p.solve_info().error is a tuple of backend-provided diagnostic strings. It is empty (()) before a solve and whenever the solve reports no error.
  • A non-OPTIMAL status does not guarantee diagnostic text: error can still be empty.
  • If Problem.solve() raises before results import completes, rely on the raised exception details and retry with log_to_console=True to capture solver logs.

An INFEASIBLE status means the problem’s constraints can’t all be true at the same time. This could be because something in the model’s data violates one of the problem’s constraints, or because the constraints themselves are incompatible.

PyRel provides an API for performing conflict analysis to help you understand why a problem is INFEASIBLE:

  1. Add conflict=True to your Problem.solve() call

    p.solve("highs", conflict=True)
  2. Rerun from the beginning

    Rerun your Python program, or in a notebook, restart the kernel and run all cells again. Conflict analysis needs a Problem that was created with conflict=True from the start. Calling Problem.solve() again on a Problem that already ran a plain solve raises a ValueError, because conflict analysis uses a different result schema than a plain solve. Restarting and rerunning builds a fresh Problem with conflict analysis enabled.

  3. Check the conflict analysis result

    Read p.solve_info().conflict_status:

    print(p.solve_info().conflict_status)

    Use the following table to interpret the output:

    StatusWhat it meansWhat to do
    CONFLICT_FOUNDThe solver found a conflicting set.Continue with the rest of the steps in this section to diagnose the conflict.
    NO_CONFLICT_EXISTSThe solver found no conflicting set.Check termination_status. This result normally means the solved problem is feasible and does not need conflict analysis.
    NOT_SUPPORTEDThe solver doesn’t support conflict analysis.Choose a backend that supports conflict analysis, or review the formulation manually.
    FAILEDConflict analysis didn’t complete.Check the value of p.solve_info().error for more information.
  4. Check which constraints are in conflict

    Query Problem.Constraint to list every named constraint in the conflicting set:

    if p.solve_info().conflict_status == "CONFLICT_FOUND":
    m.select(
    p.Constraint.name.alias("constraint_name"),
    ).where(
    p.Constraint.in_conflict,
    ).inspect()

    Match each returned name to the corresponding Problem.satisfy() call and review the input data used by that constraint. Give each constraint a distinct name so the output identifies the relevant part of your formulation.

  5. Check decision variable limit conflicts

    Query Problem.Variable for lower- and upper-limit conflicts separately to see which limit each variable contributes to the conflicting set:

    if p.solve_info().conflict_status == "CONFLICT_FOUND":
    print("Lower limits in conflict:")
    m.select(
    p.Variable.name.alias("variable_name"),
    ).where(
    p.Variable.lower_in_conflict,
    ).inspect()
    print("Upper limits in conflict:")
    m.select(
    p.Variable.name.alias("variable_name"),
    ).where(
    p.Variable.upper_in_conflict,
    ).inspect()

    Compare the returned names with the lower and upper arguments in the corresponding Problem.solve_for() calls. Then check whether those limits are consistent with the conflicting constraints and input data.

  6. Check for integrality conflicts

    For integer and binary variables, query Problem.Variable with .integrality_in_conflict to check whether requiring a whole-number value contributes to the conflict:

    if p.solve_info().conflict_status == "CONFLICT_FOUND":
    m.select(
    p.Variable.name.alias("variable_name"),
    ).where(
    p.Variable.integrality_in_conflict,
    ).inspect()

    Compare the returned names with the type arguments in the corresponding Problem.solve_for() calls. Check whether allowing a continuous value would remove the conflict without changing the meaning of the decision.

  7. Review the conflicting formulation as a whole

    Use Problem.display() to display all variables and constraints so you can review the reported conflict in context:

    p.display()
  • Conflict analysis is a useful starting point for understanding infeasibility, but it may not reveal every issue.
  • If one p.satisfy() call creates constraints for many entities, use keyed_by to keep each result connected to its entity. See Connect constraints to entities for details.

When Problem.solve() raises an exception, the most useful thing you can do first is capture logs. To stream solver logs to your console, pass log_to_console=True.

In particular, Problem.solve() can raise:

  • ValueError if you call solve() before declaring any decision variables.
  • RuntimeError if the solve job fails, or if the Problem is in a degraded state.
  • TimeoutError if the solve job doesn’t complete within the timeout.

Here’s an example of how to capture logs and retry on failure:

from relationalai.semantics import Float, Model
from relationalai.semantics.reasoners.prescriptive import Problem
def build_problem(model: Model) -> Problem:
p = Problem(model, Float)
# Declare decision variables, add requirements,
# and optionally add an objective.
return p
m = Model("MyDecisionProblem")
p = build_problem(m)
try:
p.solve("highs", log_to_console=True)
except (RuntimeError, TimeoutError):
# Rebuild and retry so you don't reuse a partially-failed Problem instance.
p = build_problem(m)
p.solve("highs", log_to_console=True)
  • If you only want logs on failure, solve once without logs and then retry with log_to_console=True.
  • If a solve completes but you suspect it terminated early, inspect p.solve_info().termination_status and use the metadata fields in the previous section.

Use this table to troubleshoot common issues when solving a decision problem:

SymptomLikely causeWhat to try
Solve fails because no decision variables existYou didn’t call solve_for() or your where=[...] scope matches nothing.Declare at least one variable, then confirm the count is above 0 with m.select(p.num_variables()).inspect().
Solution is OPTIMAL but everything is 0You added an objective function but need a forcing constraint.Add a coverage or demand constraint like sum(x) == required or sum(x) >= demand.
The solve is slow or very largeVariable scope is too broad or missing a join.Tighten where=[...] and avoid Cartesian products.
The solve is infeasibleConstraints conflict, bounds are too tight, or joins don’t match real data.Add conflict=True to the solve call and rerun the program. In a notebook, restart the kernel and rerun from the beginning. Then inspect conflict markers, constraints with p.display(), and your input facts.