Solver Post-processor

The solver post-processor transforms a single solver output into multiple individually-scored rows. After the solver runs, the post-processor receives the original sample and the full solver output and returns a list of new (sample, solver output) pairs - one per scored row. Each pair then flows through the scorer pipeline independently.

When to Use It

Sometimes one model call produces several things you want to score separately, but the solver hands you a single output. Scoring that combined output as one row hides which parts passed and which failed. The post-processor splits it so each part becomes its own row in the evidence, with its own score.

Reach for it when a single solver output needs to be broken into smaller outputs that are viewed and scored independently, such as:

• Multiple answers in one response - the model answers several questions in a single turn and you want a pass/fail per question, not per turn.
• A batch of items - one call returns a list (extracted entities, generated examples, classified rows) and each item should be graded on its own.
• Several gradable fields in a structured output - one response carries multiple parts (e.g. a summary and a title) that each need a different scorer or ground truth.

If instead you need to reshape the interaction during the model call - branching on the answer, retrying, calling tools between turns - use a Python solver, not a post-processor.

The Running Example

The rest of this page builds one example end to end: a task that asks a model several geography questions in a single turn and scores each answer on its own.

Each dataset sample holds a list of questions and their expected answers:

{"sample_id": "japan", "questions": ["What is the capital city of Japan?", "On which continent is Japan located?"], "targets": ["Tokyo", "Asia"]}

The solver sends all of a sample’s questions in one prompt, so the model replies with every answer in a single response (separated by ---). Without a post-processor this whole response is one row, so japan gets a single pass/fail even though it covers two questions. The post-processor splits that response into one (sample, output) pair per question, giving japan.0 (capital) and japan.1 (continent) as separate scored rows.

The following sections wire this up: the task definition declares the solver and post-processor, then the snippet implements the split.

Task Definition

Add a postprocessor block inside the solver definition. Set type to "python" and provide the Python code via postprocess_snippet. Like every other snippet, it runs server-side in a fixed runtime - see Python Snippets for the available libraries.

display_name: "Multi QA Checker"
key: "multi-qa-checker"
description: >
  Evaluates a model that answers multiple questions in a single turn.
  The post-processor splits the combined answer into individual scored rows.
config_spec:
  - type: "model"
    key: "judge_model"
    display_name: "Judge Model"
  - type: "dataset"
    key: "dataset_key"
    display_name: "Dataset"
    default_value: "multi-qa"
definition:
  dataset:
    key: "<< config.dataset_key >>"
  solver:
    type: "single_turn_solver"
    input_builder:
      type: "chat_completion"
      input_messages:
        - role: "system"
          content: "You are a helpful assistant."
        - role: "user"
          content: >
            Respond to all of these questions and separate the answers with '---':
            {{ '\n---\n'.join(sample.questions) }}

            Answer with a single word and no explanations.
    postprocessor:
      type: "python"
      postprocess_snippet: !include "./postprocessor.py"
  scorers:
    - type: "string_equals"
      ground_truth: "{{ sample.target }}"

The postprocessor block is supported for all solver types.

See the Tasks CLI reference for the full task specification.

Writing the Postprocess Snippet

The snippet must define a postprocess function with this signature:

def postprocess(
    sample: RawSample,
    solver_output: SolverOutput,
) -> list[tuple[RawSample, SolverOutput]]:
    ...

Both def and async def are supported.

• sample: the original dataset row as a dictionary.
• solver_output: the output produced by the solver for that sample (a SolverTrace, SingleSolverOutput, GroupedSolverTrace, or GroupedSolverOutput).
• Return value: a list of (new_sample, new_solver_output) pairs. Each pair becomes one scored row in the evidence.

Sample IDs

Each returned sample dict may optionally include a "sample_id" key. If you omit it , LF AI Platform generates unique IDs automatically by appending an index to the original sample ID (e.g. "japan.0", "japan.1"). The original sample ID is always stored under "original_sample_id" in each new sample dict so you can trace results back to their source.

Available types

The following types from latticeflow.core.dtypes are available in the snippet:

Type	Description
RawSample	dict[str, Any] alias for a dataset row
SolverOutput	Union of all solver output types
SolverTrace	Open Responses solver output with a structured Trace
SingleSolverOutput	Legacy solver output (messages + model output)
GroupedSolverTrace	Multiple SolverTrace objects from a grouped solver
GroupedSolverOutput	Multiple SingleSolverOutput objects from a grouped solver

Example

The following snippet handles a solver that answers several questions in one response, separated by ---. It splits the response and reconstructs one SolverTrace per question:

from latticeflow.core.dtypes import ChatCompletionModelOutput
from latticeflow.core.dtypes import ChatCompletionModelOutputChoice
from latticeflow.core.dtypes import ChatCompletionOutputMessage
from latticeflow.core.dtypes import Message
from latticeflow.core.dtypes import MessageRole
from latticeflow.core.dtypes import MessageStatus
from latticeflow.core.dtypes import ModelResponse
from latticeflow.core.dtypes import OutputTextContent
from latticeflow.core.dtypes import RawSample
from latticeflow.core.dtypes import SolverOutput
from latticeflow.core.dtypes import SolverTrace
from latticeflow.core.dtypes import Trace


def postprocess(
    sample: RawSample, solver_output: SolverOutput
) -> list[tuple[RawSample, SolverOutput]]:
    # Split the full model answer into individual answers.
    model_response = solver_output.output.choices[0].message.content
    answers = [answer.strip() for answer in model_response.split("---")]
    if len(answers) != len(sample["questions"]):
        raise ValueError(
            f"Expected {len(sample['questions'])} answers, but got {len(answers)}. "
            f"Answer:\n{model_response}"
        )

    # Construct one (sample, solver output) pair per question.
    postprocessed_outputs = []
    for question, target, answer in zip(
        sample["questions"], sample["targets"], answers
    ):
        trace = SolverTrace(trace=Trace.from_items([]), raw_outputs=[])
        trace.append_user_message(question)
        trace.add_model_response(
            ModelResponse(
                raw_output=ChatCompletionModelOutput(
                    choices=[
                        ChatCompletionModelOutputChoice(
                            message=ChatCompletionOutputMessage(
                                role="assistant", content=answer
                            )
                        )
                    ]
                ),
                items=[
                    Message(
                        id="",
                        status=MessageStatus.completed,
                        role=MessageRole.assistant,
                        content=[OutputTextContent(text=answer)],
                    )
                ],
            )
        )
        # Preserve the raw direct I/O from the original solver call.
        trace.direct_ios = solver_output.direct_ios
        postprocessed_outputs.append(({"question": question, "target": target}, trace))

    return postprocessed_outputs

The full dataset (introduced above) has one sample per country, each with its list of questions and targets:

{"sample_id": "japan", "questions": ["What is the capital city of Japan?", "On which continent is Japan located?"], "targets": ["Tokyo", "Asia"]}
{"sample_id": "austria", "questions": ["What is the capital city of Austria?", "On which continent is Austria located?"], "targets": ["Vienna", "Europe"]}

With 4 dataset samples and 2 questions each, the evaluation will contain 8 scored rows in the evidence (e.g. japan.0, japan.1, austria.0, austria.1, …). Each row is scored independently by the string_equals scorer against its individual target.

See the full runnable example for the complete task definition, dataset, and evaluation run configuration.