examples/Unsloth/code_evaluator.py

"""
Simple example of a code evaluator that executes the generated code and checks if it produces the
correct output. Very unsafe, use with caution. Only for demonstration purposes.
"""

from lynxkite_core import ops
from lynxkite_graph_analytics.core import (
    Bundle,
    ENV,
)
from lynxkite_llm_training.llm_evaluation import LLMEvaluator

from tqdm import tqdm
import pandas as pd

op = ops.op_registration(ENV, "LLM Training", "Evaluation")


class CodeEvaluator(LLMEvaluator):
    def __init__(self):
        pass

    def evaluation(
        self,
        prompts: list[str],
        completions: list[str] | list[list[dict]],
        ground_truths: list[dict],
        dataset: list[dict] | None = None,
        path: str = "",
    ) -> dict[str, float] | pd.DataFrame:
        assert dataset is not None, "dataset is required for CodeEvaluator"
        results = []
        for prompt, completion, ground_truth, data in tqdm(
            zip(prompts, completions, ground_truths, dataset), total=len(prompts)
        ):
            code = completion
            code_env = {}
            is_correct = False
            if ground_truth == code:
                is_correct = True
                continue

            exec(ground_truth, code_env)
            try:
                # It raises an assertion error if the test fails
                if "test_execution" in code_env:
                    code_env["test_execution"](code)
                    is_correct = True
                else:
                    is_correct = False
            except Exception as e:
                is_correct = False
                exception_message = str(e)
            finally:
                results.append(
                    {
                        "prompt": prompt,
                        "completion": completion,
                        "reference": data["reference_code"],
                        "is_correct": is_correct,
                        "exception": exception_message if not is_correct else None,
                    }
                )

        results = pd.DataFrame(results)
        print(f"Accuracy: {results.is_correct.mean()}")
        return results


@op("Define code evaluator")
def define_code_evaluator_op(
    bundle: Bundle,
    *,
    save_as: str = "code_evaluator",
) -> Bundle:
    b = bundle.copy()
    evaluator = CodeEvaluator()
    b.other[save_as] = evaluator
    return b


@op("Filter out data")
def filter_out_data_op(
    bundle: Bundle,
    *,
    dataset_name: str = "training_dataset",
    save_as: str = "filtered_dataset",
) -> Bundle:
    b = bundle.copy()
    dataset_df: pd.DataFrame = b.dfs[dataset_name]
    # keep only rows where metadata field 'library' is not 'Tensorflow'
    filtered_df = dataset_df[
        dataset_df["metadata"].apply(lambda x: x.get("library") != "Tensorflow")
    ]
    b.dfs[save_as] = filtered_df.reset_index(drop=True)

    return b