Writing the workflow
The contract of src/workflow.py: the Workflow class, what each method receives and returns, the Outcome object, solver details and how answers are rechecked.
Updated 13 November 2026 View as markdown
On this page
A workflow is one Python class that subclasses qsolve.Workflow, named by the entrypoint in qsolve.yaml. Qsolve builds it from a validated input, runs every declared method and recomputes each objective with your evaluate. The manifest side is in The manifest. This page is the guided tour; every name, parameter and exception is in the exact reference, The Workflow class, and the rest of the API is mapped in the SDK reference.
The class
from qsolve import Context, Estimate, InvalidSolution, MethodUnavailable, Outcome, Workflow, WorkflowError
class MyWorkflow(Workflow):
def __init__(self, input, params) # parse and validate; raise WorkflowError; call super().__init__
def size(self) -> int # qubits or variables
def quantum(self, ctx) -> Outcome # the method with role quantum
def reference(self, ctx) -> Outcome # the method with role reference
def classical_<id>(self, ctx) -> Outcome # one per method with role classical
def evaluate(self, solution) -> float # the objective of a solution; lower is better
def answer(self, outcome) -> dict # the domain answer, validated by outputs.schema
def estimate(self, target) -> Estimate # optional: circuits and shots to expect
def summary(self, quantum, reference) -> str # optional: one sentence for the result page
You never call these methods yourself: Qsolve calls them. quantum, reference, evaluate, answer and size are required (the base class raises NotImplementedError for them). A declared method with no matching class method fails with "method 'x' is declared in the manifest but the workflow has no classical_x method". In classical_<id>, dashes in the manifest id become underscores: exhaustive-or-annealing would be classical_exhaustive_or_annealing. The roles quantum and reference always use those plain names.
What you receive
__init__(self, input, params)
input is the user's input as a dictionary, already validated against schemas/input.json. params holds every manifest parameter: defaults, with any overrides applied and checked. Call super().__init__(input, params), which stores them as self.input and self.params.
Parse the input once here and raise WorkflowError("message") for anything the user got wrong. The user sees the message as written.
if len(set(self.nodes)) != len(self.nodes):
raise WorkflowError("node names must be unique")
A ValueError, KeyError, TypeError or IndexError from __init__ becomes "Could not set up the problem from this input: ...". Any other exception is a crash.
size(self) -> int
The number of qubits or variables. Qsolve compares it with the manifest's envelope and rejects the input above max_variables or max_qubits, with the envelope's notes in the message.
ctx (a Context)
| Field | Type | Purpose |
|---|---|---|
seed |
int | The run's seed. Use it for all randomness. |
target |
str | The target string, for example sim-cpu or fake:fez@1.6. |
backend |
Qiskit BackendV2 |
Runs circuits. See Backends and targets. |
sampler |
Qiskit SamplerV2 |
Runs circuits and returns counts. |
estimator |
Qiskit EstimatorV2 |
Measures expectation values. |
emit(type, **data) |
function | Sends a progress event, for example ctx.emit("train_step", step=3, value=-1.2, best=-1.4). Pass the type positionally. Never affects the result and costs nothing when nobody listens. |
report_mitigation(circuits, distribution, methods=(), kept=None, steps=()) |
function | Declares mitigated output. See Error mitigation and fidelity. |
Only the quantum method may use the backend. The ctx given to reference and classical_* has a backend that raises QuantumNotAllowed, a subclass of WorkflowError, as soon as one of its run calls is made. The fields are specified in The Workflow class; sampler, estimator and backend in Backends and the Qiskit bridge.
What you return
Outcome
Every method returns an Outcome:
Outcome(solution, objective, stderr=None, samples=None, details={})
| Field | Meaning |
|---|---|
solution |
JSON-compatible data in whatever form your evaluate accepts. Use None for a result that cannot be re-evaluated, such as a measured energy or one taken from a classical library; it is then labelled reported instead of recomputed. |
objective |
The value the method reports for its solution. Lower is better. Must be finite. |
stderr |
One standard deviation, when the value is a measurement. Must be finite and not negative. |
samples |
For a sampling method: a list of {"solution": ..., "p": probability}, optionally with a "label" string. Give all of them. |
details |
A free-form dictionary kept with the run (default: an empty one). |
For a maximisation problem, return the negative: max-cut returns the negative cut weight and labels its objective "Negative cut weight". Your answer() turns it back into a cut weight for people.
evaluate(self, solution) -> float
The objective of a solution, lower is better. Qsolve calls it on every method's solution and on each entry of samples, so keep it exact and fast. Raise InvalidSolution("reason") for a solution that breaks a rule of the problem.
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
The answer people read, shaped by schemas/output.json. Qsolve calls it on the quantum method's outcome, which must succeed, and on every other method's outcome so the answers can be shown side by side. Every method's answer is validated against the output schema. A mismatch in the quantum answer fails the run with "Workflow produced an invalid answer: ...". For a classical method (or the reference), a failing answer() or a mismatch marks only that method unavailable, with the reason (answer() failed: ... or its answer breaks the output schema: ...) and the run continues; if it is the reference, the run fails because there is nothing to compare with.
estimate(self, target) -> Estimate
Optional but strongly recommended. It returns what a run will send to the target:
Estimate(
qubits=5,
circuits_on_target=1,
shots_on_target=4000,
classical_cost_class="low", # "low", "medium" or "high": local CPU work
runtime_s=None, # predicted seconds for the whole run, or None
notes=[],
)
The estimate becomes the cost quote shown before a run. Without one, Qsolve assumes one circuit and params["shots"] shots and marks the estimate as a default; estimate_honest then passes with a note. With one, it is compared with what the first benchmark case sent: circuits and shots must each be within a factor of two. A runtime_s above the manifest's resources.timeout_s makes the run refused. The helpers the examples use to predict simulation time (statevector_seconds, simulator_threads, S_PER_TRANSPILE) are in qsolve.sim; see Metering and cost.
summary(self, quantum, reference) -> str
Optional. One sentence for the result page, truncated to 500 characters. The default reads "QAOA (quantum approximate optimisation): -10; Exhaustive search: -10 (accuracy 100.00%)."
Writing the quantum method
Build circuits with Qiskit, train however you like, and send what you want measured through ctx. Local statevector training is not metered. There are three ways to send circuits, all Qiskit-native:
# 1. the Sampler: bit strings and counts
counts = ctx.sampler.run([(circuit, parameter_values, shots)]).result()[0].data.meas.get_counts()
# 2. the Estimator: expectation values, no mitigation, precision sets the shots
evs = ctx.estimator.run([(circuit, observable)], precision=0.02).result()[0].data.evs
# 3. the backend: plain Qiskit run()
counts = ctx.backend.run(circuit, shots=2000).result().get_counts()
Every circuit you send is copied, recorded and measured; all circuits of one sampler.run with the same shot count are sent as one batch (on a quantum computer, one job and one queue wait). Circuits need measurements for the sampler and the backend. Send circuits as you built them; Qsolve transpiles for the device. Each pub result carries per-circuit metadata in pub_result.metadata["qsolve"][k], including dispatch (the index to cite in report_mitigation) and, on noise-model and hardware targets, readout errors and two_qubit_gates. This is the final-sampling step from the scaffold: This is the final-sampling step from the scaffold:
measured = qc.copy()
measured.measure_all()
counts = ctx.sampler.run([(measured, theta, shots)]).result()[0].data.meas.get_counts()
samples = [
{"solution": [int(c) for c in reversed(bits)], "p": n / shots}
for bits, n in counts.items()
]
best = min(samples, key=lambda s: self.evaluate(s["solution"]))
return Outcome(
solution=best["solution"],
objective=self.evaluate(best["solution"]),
samples=sorted(samples, key=lambda s: -s["p"])[:10],
details={"theta": theta.tolist(), "layers": p},
)
Qiskit puts qubit 0 on the right of a bitstring, so the code reverses them to put variable 0 first. A quantum method that sends no circuit fails validation: "quantum method qaoa sent no circuit to the backend".
Writing the classical methods
reference is the method you are judged against. Use the strongest standard method that is practical, such as an exact solver where the size allows, and describe it honestly in the manifest. Add any number of classical_<id> methods for comparison.
Raise MethodUnavailable("why") from a method that does not apply to an input; it is shown as unavailable with that reason and the run continues. A classical_* method that raises anything else, a WorkflowError included, is also shown as unavailable ("failed: ..."), and a case where that happens fails validation: use MethodUnavailable for "does not apply" and keep real errors out. If the quantum method or the reference raises, the run fails (a WorkflowError reaches the user as written; any other exception is a crash). The reference must produce a value: otherwise the run fails with "the quantum method or the reference produced no result".
A classical method with no solution to re-evaluate, such as an energy from a chemistry library, returns solution=None and can raise MethodUnavailable when the number does not exist for an input. This is from molecular-ground-state, where the references are classical energies computed before the run:
def _reference_energy(self, kind, name):
energy = self.chemset.reference(kind)
if energy is None:
raise MethodUnavailable(f"{name} was not computed for this molecule")
return Outcome(solution=None, objective=energy)
def classical_hartree_fock(self, ctx):
return self._reference_energy("hf", "Hartree-Fock")
How Qsolve checks your answer
Qsolve does not trust a method's own number. For each method:
- If it returned a solution, Qsolve calls
evaluate(solution)and uses that value as the objective, labelled recomputed. - If your reported
objectivediffers from the recomputed value by more than one millionth (relative, with a floor of 1), the method is flagged with the claimed number and fails validation: "method qaoa claimed X, its solution gives Y". - If
evaluateraisesInvalidSolutionor returns a non-finite value, the method is shown as unavailable. - If it returned
solution=None, its objective is accepted but labelled reported. - For methods with
samples, Qsolve evaluates every sample, reports the share that is valid and the share on the baseline's answer, and shows the five most frequent outcomes. - The headline accuracy and verdict come from your manifest's metric kind, using the recomputed values. See Benchmarks.
Re-evaluation shows a number is consistent with its solution, not that your evaluate is correct. Test it.
Reporting solver details
Put diagnostics a reader might want in the quantum method's details: trained angles, layer counts, per-system numbers, anything JSON-compatible.
return Outcome(run.solution, run.energy, samples=run.samples, details=run.details)
Qsolve keeps them with the run under solver, converting NumPy arrays and scalars to plain JSON. Details larger than 200,000 bytes serialised are replaced by a note giving their size. They are not part of the result hash.
For an error_rate metric, details must contain n_test, the number of test cases, so the 95% interval can be computed.
Testing it with pytest
qsolve init creates tests/test_workflow.py. It calls the same function the platform runs:
from qsolve import run_workflow
out = run_workflow(str(ROOT), inp, params={"layers": 2}, seed=1)
assert out.result.metrics is not None and out.result.metrics.matches
run_workflow returns a RunOutput with a result and a provenance; fields are in Results and events. Use estimate_run to test your estimate() and qsolve.benchmark.run_benchmarks to run the gates from Python (Runner and manifest, Testing, linting and the sandbox). To test one method in isolation, build the class yourself and call evaluate and answer directly; the methods that need a Context are best tested through run_workflow.
Determinism
The same input, parameters and seed must give the same result. The reproducible gate runs your first benchmark case twice and compares result hashes with floats rounded to a grid of 1e-6. Use ctx.seed for every random number generator. Runs use one BLAS thread so summation order is fixed, and results that differ only at floating-point noise (relative 1e-8) pass.
Next
- Input schema reference and Output schema reference, or the overview in Input and output schemas
- Parameters and The manifest
- SDK reference for every public name
- Lint codes for what
qsolve lintflags (imports the sandbox allows, for example) - Backends and targets
- Examples for complete working classes