Skip to content

base

base

Workflow protocol + evaluator-factory plumbing.

A workflow is a multi-stage orchestration of optimizer passes. Workflows consume:

  • an immutable :class:~q2mm.models.problem.OptimizationProblem,
  • an evaluator factory — a Callable[[ObjectivePlan], ObjectiveEvaluator] that builds a concrete executor (Python or JAX) for a compiled plan, and
  • a pre-configured optimizer.

They compile :class:~q2mm.objectives.plan.ObjectivePlan objects, drive the optimizer, and aggregate per-stage diagnostics into the one canonical :class:~q2mm.models.results.OptimizationResult. Workflows never construct backends or objectives directly, never mutate a force field, and never store a force-field snapshot in the result — callers materialize a force field explicitly via layout.replace(base_ff, result.final_params).

Workflow

Bases: Protocol

Multi-stage force field parameterization workflow.

name property 

name: str

The configured workflow's public identifier.

run 

run(problem: OptimizationProblem, make_evaluator: Callable[[ObjectivePlan], ObjectiveEvaluator], optimizer: _Optimizer, *, n_evals: int = 1, regularization: float = 0.0) -> OptimizationResult

Execute the workflow and return the canonical result.

Parameters:

Name Type Description Default
problem OptimizationProblem

The immutable optimization problem.

 required
make_evaluator Callable[[ObjectivePlan], ObjectiveEvaluator]

Factory that builds an executor from a plan.

 required
optimizer _Optimizer

Pre-configured optimizer.

 required
n_evals int

Configured-executor samples at the initial and final parameters (0 skips). Multi-stage workflows may sample different objective plans at the two endpoints.

 1
regularization float

Non-negative L2 penalty threaded into every :class:~q2mm.objectives.plan.ObjectivePlan the workflow compiles, so optimization and reporting share one objective.

 0.0
Source code in q2mm/workflows/base.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def run(
    self,
    problem: OptimizationProblem,
    make_evaluator: Callable[[ObjectivePlan], ObjectiveEvaluator],
    optimizer: _Optimizer,
    *,
    n_evals: int = 1,
    regularization: float = 0.0,
) -> OptimizationResult:
    """Execute the workflow and return the canonical result.

    Args:
        problem: The immutable optimization problem.
        make_evaluator: Factory that builds an executor from a plan.
        optimizer: Pre-configured optimizer.
        n_evals: Configured-executor samples at the initial and final
            parameters (``0`` skips). Multi-stage workflows may sample
            different objective plans at the two endpoints.
        regularization: Non-negative L2 penalty threaded into every
            :class:`~q2mm.objectives.plan.ObjectivePlan` the workflow
            compiles, so optimization and reporting share one objective.

    """
    ...

make_evaluator_factory 

make_evaluator_factory(backend: Backend, base_force_field: ForceField, *, settings: ObjectiveSettings) -> EvaluatorFactory

Close over explicit construction choices; each stage gets its own executor.

Source code in q2mm/workflows/base.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
def make_evaluator_factory(
    backend: Backend,
    base_force_field: ForceField,
    *,
    settings: ObjectiveSettings,
) -> EvaluatorFactory:
    """Close over explicit construction choices; each stage gets its own executor."""
    from q2mm.objectives import create_evaluator

    def factory(plan: ObjectivePlan) -> ObjectiveEvaluator:
        return create_evaluator(plan, backend=backend, force_field=base_force_field, settings=settings)

    return factory

build_workflow 

build_workflow(settings: WorkflowSettings) -> Workflow

Construct exactly the declared workflow, without policy resolution.

Source code in q2mm/workflows/base.py 
55
56
57
58
59
60
61
62
63
64
65
def build_workflow(settings: WorkflowSettings) -> Workflow:
    """Construct exactly the declared workflow, without policy resolution."""
    if isinstance(settings, SingleStageSettings):
        from q2mm.workflows.single_stage import SingleStageWorkflow

        return SingleStageWorkflow(settings)
    if isinstance(settings, MethodE2Settings):
        from q2mm.workflows.method_e2 import MethodE2Workflow

        return MethodE2Workflow(settings)
    raise TypeError(f"Unsupported workflow settings {type(settings).__name__}.")