relationalai.experimental.solvers.operators()

Signature

operators() -> Operators

Enables expression-based syntax for writing solver constraints and objectives. Used as a context manager to override standard Python operators (like ==, >=, +, etc.) so they return SolverExpression objects instead dsl.Expression objects that filter entities and properties.

Use this when building solver logic that involves arithmetic or logical operations between solver variables.

Returns

An Operators object.

Example

Use a with solvers.operators(): block to build constraints with standard arithmetic or comparison syntax:

regional_water_allocation.py

import relationalai as rai
from relationalai.experimental import solvers

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

67 collapsed lines
# Declare water distribution stage type.
WaterSource = model.Type("WaterSource")
Region = model.Type("Region")
Consumer = model.Type("Consumer")
Consumer.region.declare()  # Region where the consumer is located.
Consumer.demand.declare()  # Water demand of the consumer.

# Declare Pipeline type to connect stages.
Pipeline = model.Type("Pipeline")
Pipeline.source.declare()
Pipeline.destination.declare()
Pipeline.capacity.declare()

# Define water sources.
with model.rule(dynamic=True):
    WaterSource.add(name="Reservoir")

# Define regions.
for region in [
    {"id": 1, "name": "North"},
    {"id": 2, "name": "South"},
    {"id": 3, "name": "East"},
    {"id": 4, "name": "West"},
]:
    with model.rule():
        Region.add(id=region["id"]).set(name=region["name"])

# Define consumers.
for consumer in [
    {"id": 1, "region_id": 1, "name": "Residential", "demand": 150.0},
    {"id": 2, "region_id": 1, "name": "Industrial", "demand": 60.0},
    {"id": 3, "region_id": 1, "name": "Farms", "demand": 40.0},
    {"id": 4, "region_id": 2, "name": "Residential", "demand": 80.0},
    {"id": 5, "region_id": 2, "name": "Industrial", "demand": 140.0},
    {"id": 6, "region_id": 2, "name": "Farms", "demand": 50.0},
    {"id": 7, "region_id": 3, "name": "Residential", "demand": 90.0},
    {"id": 8, "region_id": 3, "name": "Industrial", "demand": 180.0},
    {"id": 9, "region_id": 4, "name": "Residential", "demand": 40.0},
    {"id": 10, "region_id": 4, "name": "Industrial", "demand": 30.0},
    {"id": 11, "region_id": 4, "name": "Farms", "demand": 200.0},
]:
    with model.rule():
        region = Region(id=consumer["region_id"])
        Consumer.add(id=consumer["id"]).set(
                name=consumer["name"],
                region=region,
                demand=consumer["demand"]
        )

# Define pipelines from the reservoir to each region.
for region_id, capacity in [(1, 260.0), (2, 300.0), (3, 200.0), (4, 220.0)]:
    with model.rule():
        source = WaterSource(name="Reservoir")
        region = Region(id=region_id)
        Pipeline.add(source=source, destination=region).set(capacity=capacity)

# Define pipelines between consumers and their regions.
for consumer_id, capacity in [
    (1, 150.0), (2, 60.0), (3, 40.0),   # North
    (4, 80.0), (5, 140.0), (6, 50.0),   # South
    (7, 90.0), (8, 180.0),              # East
    (9, 40.0), (10, 30.0), (11, 200.0)  # West
]:
    with model.rule():
        consumer = Consumer(id=consumer_id)
        region = consumer.region
        Pipeline.add(source=region, destination=consumer).set(capacity=capacity)

# Create a solver model.
solver_model = solvers.SolverModel(model)

# Specify variables, constraints, and objective.
# Define continuous variables for each pipeline to represent it's flow.
with model.rule():
    pipe = Pipeline()
    solver_model.variable(
        pipe,
        name_args=["pipe", pipe.source.name, pipe.destination.name],
        lower=0,
        upper=pipe.capacity,
    )

# Define a constraint to ensure flow conservation in each region.
with solvers.operators():
    region = Region()
    flow_in = solvers.sum(Pipeline(destination=region), per=[region])
    flow_out = solvers.sum(Pipeline(source=region), per=[region])
    solver_model.constraint(
        flow_in == flow_out,
        name_args=["preserve_flow", region.name]
    )

In a solvers.operators() context, all infix operators are overridden to build solver expressions instead of filtering entities. If you need to filter entities, do so in a model.rule() context and then define the constraint in a nested solvers.operators() context:

regional_water_allocation.py

import relationalai as rai
from relationalai.experimental import solvers

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

91 collapsed lines
# Declare water distribution stage type.
WaterSource = model.Type("WaterSource")
Region = model.Type("Region")
Consumer = model.Type("Consumer")
Consumer.region.declare()  # Region where the consumer is located.
Consumer.demand.declare()  # Water demand of the consumer.

# Declare Pipeline type to connect stages.
Pipeline = model.Type("Pipeline")
Pipeline.source.declare()
Pipeline.destination.declare()
Pipeline.capacity.declare()

# Define water sources.
with model.rule(dynamic=True):
    WaterSource.add(name="Reservoir")

# Define regions.
for region in [
    {"id": 1, "name": "North"},
    {"id": 2, "name": "South"},
    {"id": 3, "name": "East"},
    {"id": 4, "name": "West"},
]:
    with model.rule():
        Region.add(id=region["id"]).set(name=region["name"])

# Define consumers.
for consumer in [
    {"id": 1, "region_id": 1, "name": "Residential", "demand": 150.0},
    {"id": 2, "region_id": 1, "name": "Industrial", "demand": 60.0},
    {"id": 3, "region_id": 1, "name": "Farms", "demand": 40.0},
    {"id": 4, "region_id": 2, "name": "Residential", "demand": 80.0},
    {"id": 5, "region_id": 2, "name": "Industrial", "demand": 140.0},
    {"id": 6, "region_id": 2, "name": "Farms", "demand": 50.0},
    {"id": 7, "region_id": 3, "name": "Residential", "demand": 90.0},
    {"id": 8, "region_id": 3, "name": "Industrial", "demand": 180.0},
    {"id": 9, "region_id": 4, "name": "Residential", "demand": 40.0},
    {"id": 10, "region_id": 4, "name": "Industrial", "demand": 30.0},
    {"id": 11, "region_id": 4, "name": "Farms", "demand": 200.0},
]:
    with model.rule():
        region = Region(id=consumer["region_id"])
        Consumer.add(id=consumer["id"]).set(
                name=consumer["name"],
                region=region,
                demand=consumer["demand"]
        )

# Define pipelines from the reservoir to each region.
for region_id, capacity in [(1, 260.0), (2, 300.0), (3, 200.0), (4, 220.0)]:
    with model.rule():
        source = WaterSource(name="Reservoir")
        region = Region(id=region_id)
        Pipeline.add(source=source, destination=region).set(capacity=capacity)

# Define pipelines between consumers and their regions.
for consumer_id, capacity in [
    (1, 150.0), (2, 60.0), (3, 40.0),   # North
    (4, 80.0), (5, 140.0), (6, 50.0),   # South
    (7, 90.0), (8, 180.0),              # East
    (9, 40.0), (10, 30.0), (11, 200.0)  # West
]:
    with model.rule():
        consumer = Consumer(id=consumer_id)
        region = consumer.region
        Pipeline.add(source=region, destination=consumer).set(capacity=capacity)

# Create a solver model.
solver_model = solvers.SolverModel(model)

# Specify variables, constraints, and objective.
# Define continuous variables for each pipeline to represent it's flow.
with model.rule():
    pipe = Pipeline()
    solver_model.variable(
        pipe,
        name_args=["pipe", pipe.source.name, pipe.destination.name],
        lower=0,
        upper=pipe.capacity,
    )

# Define a constraint to ensure flow conservation in each region.
with solvers.operators():
    region = Region()
    flow_in = solvers.sum(Pipeline(destination=region), per=[region])
    flow_out = solvers.sum(Pipeline(source=region), per=[region])
    solver_model.constraint(
        flow_in == flow_out,
        name_args=["preserve_flow", region.name]
    )

# Define a constraint satisfy at least half of high-demand consumers' demand.
with model.rule():
    consumer = Consumer()
    consumer.demand >= 100.0  # Filter for high-demand consumers.
    pipe = Pipeline(source=consumer.region, destination=consumer)
    with solvers.operators():
        solver_model.constraint(
            pipe >= .5 * consumer.demand,
            name_args=["satisfy_consumer", consumer.id]
        )

Resource

See the Define Solver Constraints section of the Working with Solver Models guide to learn more about solver expressions.