Training Recipes

import verifiers as vf
from datasets import load_dataset

def load_environment(split: str = "train", num_examples: int = -1) -> vf.Environment:
    ds = load_dataset("openai/gsm8k", split=split)
    dataset = vf.Dataset.from_hf(ds, question_col="question", answer_col="answer")
    if num_examples > 0:
        dataset = dataset.select(range(num_examples))

    rubric = vf.MathRubric()
    return vf.SingleTurnEnv(
        dataset=dataset,
        rubric=rubric,
        system_prompt="Solve the problem step by step. Put your final answer in \\boxed{}.",
    )

import verifiers as vf
from datasets import Dataset

def load_environment() -> vf.Environment:
    dataset = Dataset.from_list([
        {
            "question": "Write a function `fibonacci(n)` that returns the nth Fibonacci number.",
            "info": '{"test_code": "assert fibonacci(0) == 0\\nassert fibonacci(1) == 1\\nassert fibonacci(10) == 55"}'
        },
        # ... more examples
    ])

    async def tests_pass(completion, info, state) -> float:
        code = completion[-1]["content"]
        test_code = info["test_code"]
        try:
            exec_result = state.get("exec_result", "")
            return 1.0 if "PASSED" in exec_result else 0.0
        except Exception:
            return 0.0

    rubric = vf.Rubric(funcs=[tests_pass])
    return vf.PythonEnv(
        dataset=dataset,
        rubric=rubric,
        max_turns=5,
    )

import verifiers as vf
from datasets import Dataset
import random

class WordGameEnv(vf.MultiTurnEnv):
    async def setup_state(self, state, **kwargs):
        state["target"] = state["info"]["target_word"]
        state["guesses"] = []
        return await super().setup_state(state, **kwargs)

    async def env_response(self, messages, state):
        guess = messages[-1]["content"].strip().lower()
        target = state["target"]
        state["guesses"].append(guess)

        if guess == target:
            state["won"] = True
            return [{"role": "user", "content": "Correct! You found the word."}]

        # Give hints: which letters are in the right position
        hints = []
        for i, (g, t) in enumerate(zip(guess, target)):
            if g == t:
                hints.append(f"Position {i+1}: correct")
            elif g in target:
                hints.append(f"Position {i+1}: wrong position, letter is in the word")
            else:
                hints.append(f"Position {i+1}: letter not in word")

        return [{"role": "user", "content": "\n".join(hints) + "\nGuess again."}]

    @vf.stop
    async def game_won(self, state):
        return state.get("won", False)


def load_environment() -> vf.Environment:
    words = ["apple", "brain", "cloud", "dance", "eagle"]
    dataset = Dataset.from_list([
        {"question": "Guess the 5-letter word. I'll give you hints after each guess.",
         "info": f'{{"target_word": "{w}"}}'} for w in words
    ])

    async def win_reward(state) -> float:
        if state.get("won"):
            return max(0.2, 1.0 - 0.15 * len(state["guesses"]))
        return 0.0

    rubric = vf.Rubric(funcs=[win_reward])
    return WordGameEnv(dataset=dataset, rubric=rubric, max_turns=8)

import verifiers as vf
from datasets import Dataset

async def web_search(query: str) -> str:
    """Search the web for information.

    Args:
        query: The search query to look up.

    Returns:
        Search results as text.
    """
    # your search implementation
    return await do_search(query)

async def calculate(expression: str) -> str:
    """Evaluate a mathematical expression.

    Args:
        expression: A math expression to evaluate (e.g. "2 + 2 * 3").

    Returns:
        The result of the evaluation.
    """
    try:
        return str(eval(expression))
    except Exception as e:
        return f"Error: {e}"

def load_environment() -> vf.Environment:
    dataset = Dataset.from_list([
        {
            "question": "What is the population of France divided by the population of Switzerland?",
            "answer": "approximately 8.3"
        },
        # ... more examples requiring tool use
    ])

    async def answer_quality(completion, answer, judge) -> float:
        verdict = await judge(completion, answer)
        return 1.0 if "correct" in verdict.lower() else 0.0

    rubric = vf.JudgeRubric(judge_model="gpt-4.1-mini")
    rubric.add_reward_func(answer_quality)

    return vf.ToolEnv(
        dataset=dataset,
        tools=[web_search, calculate],
        rubric=rubric,
        max_turns=10,
    )