Python Solver

Runs a custom Python function as the solver. Use this when the built-in declarative solvers cannot express the interaction patterns with the model you need.

Function Signature

Both def and async def are supported. The function must be named run_solver and accepts the following arguments:

from typing import Any

from latticeflow.core.dtypes import RawSample, SolverTrace


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

where:

  • sample is a dictionary representing the current sample (RawSample is an alias for dict[str, Any]).
  • model is the model object used for inference (with a predict method).
  • trace is a SolverTrace object that can be used to record the conversation and should be returned by the function.

Accessing task config

run_solver can optionally accept a task_config argument to receive the task configuration as a plain Python dict. Config values of type dataset are not included.

async def run_solver(
    sample: RawSample,
    model: Any,
    trace: SolverTrace,
    task_config: dict[str, Any],
) -> SolverTrace:
    system_prompt = task_config["system_prompt"]
    ...

Using models from config

To call a model inside the solver, declare it in the task’s config_spec with type model and access it via task_config. The model exposes a predict method:

# config_spec:
#   - type: model
#     key: translate_model
#   - type: string
#     key: target_language


from latticeflow.core.dtypes import Message

async def run_solver(
    sample: RawSample,
    model: Any,
    trace: SolverTrace,
    task_config: dict[str, Any],
) -> SolverTrace:
    translate_model = task_config["translate_model"]
    target_language = task_config["target_language"]
    response = await translate_model.predict([
        Message(role="system", content=f"Translate the following question to {target_language}."),
        Message(role="user", content=sample["question"]),
    ])
    translated_question = response.text
    ...

Examples

Example: Geography QA. A Python function builds the conversation trace manually - appending a system message, the user question from the dataset, calling the model, and recording the response.

Python Solver
# ...
definition:
  # ...
  solver:
    type: "python"
    run_solver_snippet: |
      async def run_solver(sample, model, trace):
          trace.append_system_message(
              "You are a helpful assistant. Answer with a single word or short phrase."
          )
          trace.append_user_message(sample["question"])
          response = await model.predict(trace.items)
          trace.add_model_response(response)
          return trace

For a sample {"question": "What is the capital of Japan?", "answer": "Tokyo"}, the solver builds and returns the following trace:

Turn Role Content
1 system You are a helpful assistant. Answer with a single word or short phrase.
2 user What is the capital of Japan?
3 assistant Tokyo

Configuration

Properties

postprocessor PythonPostprocessorTemplate

An optional postprocessor applied to solver outputs before scoring.

Default: None

type Literal “python” required

The type of the solver.

run_solver_snippet string required

The Python code snippet defining how to run the solver. It must define a run_solver function with the following API:

from typing import Any

from latticeflow.core.dtypes import RawSample, SolverTrace


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

where:

  • sample is a dictionary representing the current sample (RawSample is an alias for dict[str, Any]).
  • model is the model object used for inference (with a predict method).
  • trace is a SolverTrace object that can be used to record the conversation and should be returned by the function.