Evaluation Pipeline#

Optimization often requires preprocessing steps before an objective or constraint can be computed. For example, calculating process performance may involve simulating the process, determining fractionation times under purity constraints, and then computing yield and productivity from the fractionation result.

CADET-Process represents these steps as a directed acyclic graph (DAG) of evaluators. Each evaluator is a named callable that receives the output of upstream evaluators and produces a named result. Objectives and constraints declare which evaluator outputs they require; the pipeline ensures each evaluator runs exactly once per evaluation, regardless of how many objectives or constraints depend on it.

../../_images/single_objective_evaluators.svg

Evaluation Objects#

Optimization variables usually represent attributes of a Process, such as model parameter values or event times, but any Python object with gettable and settable attributes can serve as an evaluation object.

To associate variables with an evaluation object, add it to the optimization problem first.

optimization_problem.add_evaluation_object(process)

Multiple evaluation objects can be added, which allows simultaneous optimization of multiple operating conditions. When adding variables, specify which evaluation objects the variable targets and the path to the attribute.

optimization_problem.add_variable(
    'var_0',
    evaluation_objects=[process],
    parameter_path='flow_sheet.column.total_porosity',
    lb=0, ub=1,
)

RangedParameter(name='var_0', parameter_type=float, lb=0, ub=1)

By default, a variable targets all evaluation objects. If no path is provided, the variable name is used as the path.

optimization_problem.add_variable('flow_sheet.column.total_porosity', lb=0, ub=1)

RangedParameter(name='flow_sheet.column.total_porosity', parameter_type=float, lb=0, ub=1)

Multiple evaluation objects with different variable associations:

../../_images/multiple_evaluation_objects.svg

optimization_problem.add_evaluation_object(process_a)
optimization_problem.add_evaluation_object(process_b)
optimization_problem.add_variable('flow_sheet.column.total_porosity')
optimization_problem.add_variable('flow_sheet.column.length', evaluation_objects=[process_a])

RangedParameter(name='flow_sheet.column.length', parameter_type=float, lb=-inf, ub=inf)

Evaluators#

To register a preprocessing step, use add_evaluator(). Any callable can be an evaluator; its first argument receives the input (the evaluation object, or the output of an upstream evaluator) and it returns a result that downstream nodes consume.

def evaluator(x):
    return x**2

optimization_problem.add_evaluator(evaluator)

To wire an objective to this evaluator, pass it via the requires argument on add_objective().

def objective(result):
    return result + 1

optimization_problem.add_objective(objective, requires=[evaluator])

When evaluating objectives, the evaluator runs first and its output is passed to the objective.

optimization_problem.evaluate_objectives(2)

array([[5.]])

Shared evaluator outputs#

When multiple objectives or constraints depend on the same evaluator, the evaluator runs once and its result is shared. This is structural, not a cache accident: the pipeline knows the graph and executes each node exactly once.

def simulate(x):
    print(f"simulate called with {x}")
    return {"yield": x * 0.8, "pressure": x * 1.2}

optimization_problem.add_evaluator(simulate)

def yield_objective(sim_result):
    return sim_result["yield"]

def pressure_constraint(sim_result):
    return sim_result["pressure"]

optimization_problem.add_objective(yield_objective, requires=[simulate])
optimization_problem.add_nonlinear_constraint(pressure_constraint, requires=[simulate])

print("objectives:", optimization_problem.evaluate_objectives(5))
print("constraints:", optimization_problem.evaluate_nonlinear_constraints(5))

simulate called with [5.]
objectives: [[4.]]
simulate called with [5.]
constraints: [[6.]]

Note that simulate is called once: both the objective and the constraint receive the same result.

Chaining evaluators#

Evaluators can be chained: the output of one feeds into the next. Only declare the immediate upstream dependency; transitive dependencies are resolved automatically.

def simulate(x):
    return {"chromatogram": x * 2}

def fractionate(sim_result):
    return {"yield": sim_result["chromatogram"] * 0.9}

optimization_problem.add_evaluator(simulate)
optimization_problem.add_evaluator(fractionate)

def compute_yield(frac_result):
    return frac_result["yield"]

optimization_problem.add_objective(compute_yield, requires=[simulate, fractionate])
optimization_problem.evaluate_objectives(3)

array([[5.4]])

Caching#

The evaluation pipeline caches intermediate results so that repeated evaluations at the same parameter vector are free. This is particularly useful during gradient approximation, where the same point may be evaluated multiple times.

By default, an in-memory LRU cache is used. To persist results across runs or share them between parallel workers, pass a cache_directory when creating the optimization problem.

optimization_problem = OptimizationProblem('cached', cache_directory='/tmp/my_cache')

The disk cache stores results as pickled files and survives process restarts. For most interactive workflows, the default in-memory cache is sufficient.