Function test_math_hf_data_processor

tests/unit/data/test_data_processor.py:124–160

(tokenizer_name, dataset_cls)

    ],
)
def test_math_hf_data_processor(tokenizer_name, dataset_cls):
    # Initialize dataset
    data = dataset_cls()
    task_name = data.task_name
    # Setup tokenizer
    tokenizer = get_tokenizer(
        TokenizerConfig(
            name=tokenizer_name,
            chat_template="default",
        )
    )

    # Configure task specification
    math_task_spec = TaskDataSpec(
        task_name=task_name,
        prompt_file=f"{os.path.dirname(abspath)}/../../../examples/prompts/cot.txt",
        system_prompt_file=None,
    )

    task_data_processors: dict[str, tuple[TaskDataSpec, TaskDataProcessFnCallable]] = (
        defaultdict(lambda: (math_task_spec, math_hf_data_processor))
    )
    task_data_processors[task_name] = (math_task_spec, math_hf_data_processor)

    dataset = AllTaskProcessedDataset(
        dataset=data.dataset,
        tokenizer=tokenizer,
        default_task_data_spec=math_task_spec,
        task_data_processors=task_data_processors,
        max_seq_length=128,
    )

    # Test that the first item can be retrieved when the BOS token assertion passes
    first_item = dataset[0]
    assert first_item is not None
    assert "message_log" in first_item
    assert len(first_item["message_log"]) > 0


def test_math_hf_data_processor_without_prompt():