SDK: the Workflow class

Module qsolve.workflow. Everything on this page is importable from the top-level package: from qsolve import Workflow, Outcome, Context, WorkflowError, InvalidSolution, MethodUnavailable, QuantumNotAllowed. For a guided walk-through read Writing the workflow; this page is the exact reference. Back to the SDK reference.

Workflow

class Workflow:
    def __init__(self, input: dict[str, Any], params: dict[str, Any]) -> None
    def size(self) -> int
    def quantum(self, ctx: Context) -> Outcome
    def reference(self, ctx: Context) -> Outcome
    def evaluate(self, solution: Any) -> float
    def answer(self, outcome: Outcome) -> dict[str, Any]
    def estimate(self, target: str) -> Estimate | None
    def summary(self, quantum: Outcome, reference: Outcome) -> str | None

You subclass Workflow and name the class in the manifest's entrypoint (package.module:ClassName). The SDK calls every method; you never call them. The base class implements only __init__, estimate and summary. The other five raise NotImplementedError until you override them.

In addition to these, every method of role classical that you declare in the manifest is a method named classical_<id>(self, ctx) -> Outcome, where <id> is the manifest method id with each dash turned into an underscore (simulated-annealing is classical_simulated_annealing). The base class does not define these; the runner looks them up by name.

A declared method whose class method is missing fails the run with method 'x-y' is declared in the manifest but the workflow has no classical_x_y method.

__init__(self, input, params)

Returns nothing. The base implementation stores them as self.input and self.params; call super().__init__(input, params) first. Side effects: none in the base class. Called once per run, by the runner, before any method runs.

Raise WorkflowError("message") for anything the user got wrong that a schema cannot express. The platform shows the message to the user as written. A ValueError, KeyError, TypeError or IndexError raised here becomes WorkflowError("Could not set up the problem from this input: <original>"); any other exception is a crash.

def __init__(self, input, params):
    super().__init__(input, params)
    self.nodes = input["nodes"]
    if len(set(self.nodes)) != len(self.nodes):
        raise WorkflowError("node names must be unique")

Gotcha: do the expensive work in the method that needs it, not here. estimate_run also builds your class (to call size and estimate), so a slow constructor slows every price quote.

size(self) -> int

The number of qubits or variables. The runner compares it with the manifest envelope straight after construction: if it is above envelope.max_variables or envelope.max_qubits the run is refused with Problem has N variables; this workflow supports at most M (<envelope.notes>). No credits are spent. Raises NotImplementedError if you do not override it.

quantum(self, ctx) -> Outcome

The method with role quantum. It is the only method whose ctx can send circuits. Build circuits in Qiskit, train however you like, and run what you want measured with ctx.sampler.run(...), ctx.estimator.run(...) or ctx.backend.run(...). Local statevector simulation that does not go through ctx is not metered.

Exceptions: a WorkflowError raised here is the deliberate "this input cannot be solved" and reaches the user as written. MethodUnavailable makes the run fail with the quantum method or the reference produced no result: <why>. Any other exception is a crash: the platform reports it and bills what was used. The method must send at least one circuit through ctx, or validation fails (quantum method <id> sent no circuit to the backend).

reference(self, ctx) -> Outcome

The method with role reference: the classical answer the quantum result is judged against. Its ctx refuses circuits. Same exception rules as quantum; the reference must produce a value or the run fails.

classical_<id>(self, ctx) -> Outcome

One per classical method. Its ctx refuses circuits. Raise MethodUnavailable("why") when the method does not apply to an input: it is shown as unavailable with that reason and the run continues. Any other exception, WorkflowError included, is caught and the method is shown as unavailable with failed: <ExceptionType>: <message>, but a method in that state fails the validation case (see check_quality in Testing, linting and the sandbox). Only MethodUnavailable is a clean "not applicable".

evaluate(self, solution) -> float

Returns the objective of that solution; lower is better. The SDK calls it on every method's solution and on every entry of samples, to recompute the number each method claimed. Keep it exact and fast. Raise InvalidSolution("reason") for a solution that breaks a rule of the problem. Any other exception is turned into WorkflowError("evaluate() failed on a method's solution: <Type>: <msg>") and fails the run. A non-finite return value marks that method unavailable. See How Qsolve checks your answer.

def evaluate(self, solution):
    x = np.asarray(solution, dtype=float)
    if x.shape != (self.n,) or not np.isin(x, (0, 1)).all():
        raise InvalidSolution("a solution is one 0 or 1 per variable")
    return float(x @ self.Q @ x)

answer(self, outcome) -> dict[str, Any]

Turns an Outcome into the domain answer people read, shaped by schemas/output.json (Output schema reference). Called on the quantum method's outcome (it must succeed; the result is validated against the output schema, and a failure or mismatch fails the run with answer() failed on the quantum result: ... or Workflow produced an invalid answer: ...) and on every other method's outcome so answers can be shown side by side (a failure there is ignored and that method simply has no answer).

estimate(self, target) -> Estimate | None

Optional. The base class returns None. Return an Estimate of what a run will send to the target; it becomes the price quote. With None, estimate_run substitutes one circuit and params["shots"] shots (or 0 if there is no shots parameter) and marks it by_default. The runner sets by_default itself, and adds the fixed run overhead to a runtime_s you give. Called by estimate_run and qsolve estimate; the estimate_honest gate compares it with what the first benchmark case really sent. Details in Metering and cost.

summary(self, quantum, reference) -> str | None

Optional. Return one sentence for the result page, or None for the default (<quantum title>: <value>; <reference title>: <value> (accuracy NN.NN%).). The runner truncates it to 500 characters. Called after all methods have run, with the quantum and reference Outcomes.

Outcome

@dataclass
class Outcome:
    solution: Any
    objective: float
    stderr: float | None = None
    samples: list[dict[str, Any]] | None = None
    details: dict[str, Any] = field(default_factory=dict)

Every method returns one.

Maximisation: return the negative value as objective and turn it back in answer().

return Outcome(
    solution=best["solution"],
    objective=self.evaluate(best["solution"]),
    samples=sorted(samples, key=lambda s: -s["p"])[:10],
    details={"theta": theta.tolist()},
)

Gotcha: samples should hold all your outcomes, not a hand-picked top few, or valid_fraction and p_reference describe only that subset.

Context

@dataclass
class Context:
    seed: int
    target: str
    backend: BackendV2
    sampler: BaseSamplerV2
    estimator: BaseEstimatorV2
    emit: Callable[..., None]
    report_mitigation: Callable[..., None]

Built by the runner and passed to every method; you never construct one except in tests. There are two flavours, made from the same inner backend: the quantum method's is live, and the one given to reference and every classical_<id> has a backend that refuses to run anything.

Using backend, sampler or estimator from a classical method raises QuantumNotAllowed. It is raised at the point of the run call.

ctx.emit(type, **data)

This is qsolve.events.emit. Sends a progress event to whoever is listening (the run page on the platform, the on_event callback of run_workflow). With no listener it does nothing and costs nothing. It never raises and never affects the result.

Gotchas: pass type positionally; a keyword type= clashes with the first parameter. A data field named t replaces the timestamp. The platform caps how many events it keeps.

ctx.emit("train_step", step=i, value=float(cost), best=float(best))

ctx.report_mitigation(circuits, distribution, methods=(), kept=None, steps=())

Declares the mitigated output of earlier circuits. Mitigation is your own post-processing, so Qsolve cannot see it; you report it, Qsolve recomputes its fidelity against the circuit's ideal output and labels it reported next to the raw figure it measured itself. The whole story, including what you may claim, is in Error mitigation and fidelity.

Returns None. Raises WorkflowError when circuits is empty or cites an index that was not sent (report_mitigation cites circuits [3], but 2 were sent so far ...), when the distribution is empty or has an invalid weight, or when kept is outside 0 to 1. Called by you, from the quantum method, after the circuits it covers were run. A report is only counted for circuits sent by the quantum method that share one ideal output.

Exceptions

InvalidSolution and MethodUnavailable are plain Exception subclasses, not WorkflowErrors. Catch WorkflowError and you also catch QuantumNotAllowed; do not swallow it inside a classical method, because that hides a violation the platform wants to see.

WorkflowError is also re-exported from qsolve.runner (it is in that module's __all__); it is the same class.

A complete minimal workflow

The test helper in the SDK's own tests uses this shape: a tiny QUBO solved by exhaustive search and by one sampling circuit.

import numpy as np
from qiskit import QuantumCircuit
from qsolve import Estimate, InvalidSolution, Outcome, Workflow, WorkflowError

class Tiny(Workflow):
    def __init__(self, input, params):
        super().__init__(input, params)
        self.Q = np.array(input["Q"], dtype=float)
        if self.Q.ndim != 2 or self.Q.shape[0] != self.Q.shape[1]:
            raise WorkflowError("Q must be square")
        self.n = len(self.Q)

    def size(self): return self.n

    def evaluate(self, solution):
        x = np.asarray(solution, dtype=float)
        if x.shape != (self.n,) or not np.isin(x, (0, 1)).all():
            raise InvalidSolution("one 0 or 1 per variable")
        return float(x @ self.Q @ x)

    def answer(self, outcome):
        return {"bits": [int(b) for b in outcome.solution], "objective": self.evaluate(outcome.solution)}

    def estimate(self, target):
        return Estimate(qubits=self.n, circuits_on_target=1, shots_on_target=int(self.params["shots"]))

    def reference(self, ctx):
        states = (np.arange(2**self.n)[:, None] >> np.arange(self.n)) & 1
        e = np.einsum("si,ij,sj->s", states, self.Q, states)
        return Outcome(solution=states[int(np.argmin(e))].tolist(), objective=float(e.min()))

    def classical_greedy(self, ctx):        # manifest method id: greedy, role: classical
        return self.reference(ctx)

    def quantum(self, ctx):
        qc = QuantumCircuit(self.n)
        qc.h(range(self.n)); qc.measure_all()
        shots = int(self.params["shots"])
        counts = ctx.sampler.run([(qc, None, shots)]).result()[0].data.meas.get_counts()
        best = min(([int(c) for c in reversed(k)] for k in counts), key=self.evaluate)
        return Outcome(solution=best, objective=self.evaluate(best))

The matching manifest, schemas and benchmark cases are in Your first workflow and The manifest.