Spaces:
Runtime error
Runtime error
File size: 2,819 Bytes
46480a9 cda6fbf 46480a9 cda6fbf | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 | """
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,
path: str = "",
) -> dict[str, float]:
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
|