Python Solver

The Python solver replaces the declarative solver types with a function you write. You drive the conversation yourself: build the messages, call the model, react to what comes back, and record everything into a trace that the scorers then read.

When to Use It

Reach for a Python solver only when the declarative types genuinely cannot express the interaction. They are easier to read, cheaper to review and validated ahead of time. Use python when the task needs:

• Control flow that depends on the model’s answer in a way the multi-turn loop builder cannot express - branching on parsed content, retrying with a different strategy, stopping on a computed condition.
• Errors as signal. The declarative solvers treat a failed model call as a task failure. A Python solver can catch the exception and record “the endpoint rejected this input” as the result - which is exactly how a guardrail or input-limit task works.
• Computation between turns, such as generating the prompt from the sample, calling an external tool, or measuring something about the response before continuing.
• Several models in one task, resolved through the task’s configuration specification.

If you only need to reshape the output after the model call, use a post-processor on a declarative solver instead - it is a much smaller change.

The Snippet Contract

The snippet must define a function called run_solver that returns the SolverTrace it was given. Both def and async def work, but since you will be awaiting model calls, async def is what you want.

from latticeflow.core.dtypes import RawSample, SolverTrace


async def run_solver(sample: RawSample, model, trace: SolverTrace) -> SolverTrace:
    ...
    return trace

The argument names are fixed, and only two signatures are accepted:

Signature	Use
run_solver(sample, model, trace)	The common case.
run_solver(sample, model, trace, task_config)	Also receives the task’s resolved configuration, including any models referenced by a model config parameter.

Important

The names must match exactly and the whole signature must be one of the two above. A snippet defining run_solver(sample, model_under_test, trace) is rejected before the task runs.

What You Get

sample is the current dataset row as a plain dict[str, Any].

model is the model under test, exposing a single method:

response = await model.predict(input, sample=None)

input is either a list of trace items (normally trace.items) or a raw request body for custom models. The returned ModelResponse has .raw_output (what the endpoint actually returned) and .items (the same response as trace items).

trace is an empty SolverTrace you append to as the conversation progresses:

Method	Effect
trace.append_system_message(text)	Add a system message.
trace.append_user_message(text)	Add a user message.
trace.add_model_response(response)	Record a ModelResponse returned by model.predict.
trace.append_custom_task_input_message(data)	Record an arbitrary JSON-serialisable dict in the trace, for the scorer to read back.
trace.items	The conversation so far, and what you pass to model.predict.

What You Must Return

The same trace, as a SolverTrace. Returning anything else fails the task with a validation error - including returning trace.items, a string, or None.

Example: Input Size Enforcement

This task checks whether an endpoint rejects an oversized prompt. The rejection is the desired behaviour, so the model call is wrapped in try/except and the outcome is written into the trace for the scorer.

tasks/input_size/solver.py

_FILLER = (
    "Lorem ipsum dolor sit amet, consectetur adipiscing elit. "
    "Sed do eiusmod tempor incididunt ut labore et dolore magna aliqua. "
)


async def run_solver(sample, model, trace):
    max_tokens = int("<< config.max_input_tokens >>")
    # Roughly 4 characters per token, plus 10% margin to reliably exceed the limit.
    target_chars = int(max_tokens * 4 * 1.1)
    oversized_text = (_FILLER * (target_chars // len(_FILLER) + 1))[:target_chars]

    trace.append_user_message(oversized_text)

    try:
        response = await model.predict(trace.items)
        trace.add_model_response(response)
        trace.append_custom_task_input_message(
            {
                "rejected": False,
                "reason": "Endpoint processed the oversized input without rejection.",
            }
        )
    except Exception as e:
        trace.append_custom_task_input_message(
            {
                "rejected": True,
                "reason": str(e) or repr(e),
                "exception_type": type(e).__name__,
            }
        )

    return trace

Wire it into the task with !include so the Python stays in its own file:

tasks/input_size/task.yaml

key: input_size_enforcement
display_name: "Input Size Enforcement"
description: "Checks that the endpoint rejects prompts above its documented input limit."
config_spec:
  - type: int
    key: max_input_tokens
    display_name: "Maximum Input Tokens"
  - type: dataset
    key: probe_dataset
    display_name: "Probe Dataset"
definition:
  dataset:
    key: "<< config.probe_dataset >>"
  solver:
    type: python
    run_solver_snippet: !include "./solver.py"
  scorers:
    - type: python
      compute_scores_snippet: !include "./scorer.py"
      metrics:
        - type: mean
          field: input_size_enforcement
          name: "Input Size Enforcement"

The matching scorer reads back what the solver recorded:

tasks/input_size/scorer.py

from __future__ import annotations

from typing import Any

from latticeflow.core.dtypes import RawSample
from latticeflow.core.dtypes import SolverOutput


def compute_scores(sample: RawSample, solver_output: SolverOutput) -> dict[str, Any]:
    outcome = solver_output.trace.items[-1].content
    return {
        "scores": {"input_size_enforcement": 1 if outcome["rejected"] else 0},
        "metadata": {"reason": outcome["reason"]},
    }

The payload passed to append_custom_task_input_message comes back as the item’s content, and because the solver appends it last, items[-1] is the outcome. The scorer returns the structured {"scores": ..., "metadata": ...} shape so that the human-readable reason travels with the score without becoming a metric - see Python Scorers.

Note

Note the << config.max_input_tokens >> in the solver snippet. Configuration parameters are substituted into snippets before they run, so the snippet is a template as well as code. Because the substitution is textual, keep the placeholder inside a string and convert it - int("<< config.max_input_tokens >>") - so the file stays valid Python for your editor and linter.

Runtime and Limits

The snippet runs server-side in a fixed Python runtime with a fixed set of libraries. It cannot install packages and it should not reach for the network beyond model.predict. See Python Snippets for the available libraries and the constraints that apply to every snippet type.

Field-level documentation is in the Python Solver reference.