| import os |
| import json |
| import random |
| from typing import Any, Dict, Callable, List |
| from .benchmark import Benchmark |
| from .measures import exact_match_score, f1_score, acc_score |
| from ..core.logging import logger |
| from ..core.module_utils import load_json |
| from datasets import load_dataset |
|
|
| |
| WORFBENCH_FILES_MAP = { |
| "train": "worfbench_train.json", |
| "test": "worfbench_test.json" |
| } |
| VALID_WORFBENCH_FILES = list(WORFBENCH_FILES_MAP.values()) |
|
|
| def evaluate_workflow_sequence(prediction: List[Any], ground_truth: List[Any]) -> float: |
| """Evaluate F1 score for sequence workflow.""" |
| from .measures import f1_score |
| return f1_score(prediction=prediction, ground_truth=ground_truth) |
|
|
| def evaluate_workflow_graph(prediction: Dict[str, Any], ground_truth: Dict[str, Any]) -> float: |
| """Evaluate F1 score for graph workflow.""" |
| pred_nodes = set(prediction.get("nodes", [])) |
| true_nodes = set(ground_truth.get("nodes", [])) |
| pred_edges = set(tuple(edge) for edge in prediction.get("edges", [])) |
| true_edges = set(tuple(edge) for edge in ground_truth.get("edges", [])) |
| |
| node_precision = len(pred_nodes & true_nodes) / len(pred_nodes) if pred_nodes else 0 |
| node_recall = len(pred_nodes & true_nodes) / len(true_nodes) if true_nodes else 0 |
| edge_precision = len(pred_edges & true_edges) / len(pred_edges) if pred_edges else 0 |
| edge_recall = len(pred_edges & true_edges) / len(true_edges) if true_edges else 0 |
| |
| node_f1 = 2 * (node_precision * node_recall) / (node_precision + node_recall) if (node_precision + node_recall) > 0 else 0 |
| edge_f1 = 2 * (edge_precision * edge_recall) / (edge_precision + edge_recall) if (edge_precision + edge_recall) > 0 else 0 |
| |
| return (node_f1 + edge_f1) / 2 |
|
|
| def download_worfbench_data(dataset: str, save_folder: str) -> None: |
| """ |
| Download WorfBench dataset from Hugging Face. |
| |
| Args: |
| dataset (str): Dataset name ("worfbench"). |
| save_folder (str): Directory to save data. |
| """ |
| datasets_map = { |
| "train": {"repo_id": "zjunlp/WorFBench_train", "filename": "worfbench_train.json", "split": "train"}, |
| "test": {"repo_id": "zjunlp/WorFBench_test", "filename": "worfbench_test.json", "split": "test"} |
| } |
| |
| os.makedirs(save_folder, exist_ok=True) |
| for split, info in datasets_map.items(): |
| repo_id = info["repo_id"] |
| filename = info["filename"] |
| dataset_split = info["split"] |
| save_path = os.path.join(save_folder, filename) |
| |
| if not os.path.exists(save_path): |
| logger.info(f"Downloading {split} split of {dataset} from {repo_id}...") |
| try: |
| |
| ds = load_dataset(repo_id, split=dataset_split) |
| |
| data = [item for item in ds] |
| with open(save_path, 'w', encoding='utf-8') as f: |
| json.dump(data, f, ensure_ascii=False, indent=2) |
| logger.info(f"Successfully downloaded and saved {filename} to {save_path}") |
| except Exception as e: |
| logger.error(f"Failed to download or save {filename}: {e}") |
| raise |
| else: |
| logger.info(f"File {save_path} already exists, skipping download.") |
|
|
| class WorfBench(Benchmark): |
| """ |
| WorfBench evaluation class for assessing LLM agents on complex workflow generation tasks. |
| Assumed data structure: |
| { |
| "id": str, |
| "task": str, |
| "context": list of dicts (e.g., [{"title": str, "content": list of str}]), |
| "expected_output": str or dict (sequence or graph), |
| "type": str, |
| "level": str |
| } |
| """ |
| def __init__(self, path: str = None, mode: str = "test", **kwargs): |
| path = os.path.expanduser(path or "~/.worfbench/data") |
| super().__init__(name=type(self).__name__, path=path, mode=mode, **kwargs) |
|
|
| def _load_data_from_file(self, file_name: str) -> Dict: |
| if file_name is None: |
| return None |
| file_path = os.path.join(self.path, file_name) |
| if not os.path.exists(file_path): |
| download_worfbench_data(dataset="worfbench", save_folder=self.path) |
| if not os.path.exists(file_path): |
| logger.error(f"File {file_path} still does not exist after download attempt!") |
| return None |
| logger.info(f"Loading WorfBench data from {file_path} ...") |
| data = load_json(path=file_path, type="json") |
| if data is None: |
| logger.error(f"Failed to load data from {file_path}") |
| return None |
| return data |
|
|
| def _load_data(self) -> None: |
| if self.mode in ["train", "dev"]: |
| self._train_data = self._load_data_from_file(file_name=WORFBENCH_FILES_MAP["train"]) |
| if self.mode == "dev": |
| if self._train_data: |
| random.seed(42) |
| keys = list(self._train_data.keys()) |
| n_dev = len(self._train_data[keys[0]]) // 10 or 1 |
| indices = list(range(len(self._train_data[keys[0]]))) |
| random.shuffle(indices) |
| self._train_data = {k: [v[i] for i in indices[:n_dev]] for k, v in self._train_data.items()} |
| if self.mode == "test": |
| self._test_data = self._load_data_from_file(file_name=WORFBENCH_FILES_MAP["test"]) |
|
|
| def _get_label(self, example: Dict) -> Any: |
| return example.get("expected_output", "") |
|
|
| def _get_id(self, example: Dict) -> Any: |
| return example.get("id", "") |
|
|
| def evaluate(self, prediction: Any, label: Any) -> Dict: |
| if isinstance(prediction, list) and isinstance(label, list): |
| f1 = evaluate_workflow_sequence(prediction, label) |
| elif isinstance(prediction, dict) and isinstance(label, dict): |
| f1 = evaluate_workflow_graph(prediction, label) |
| else: |
| f1 = f1_score(prediction=str(prediction), ground_truth=str(label)) |
| em = exact_match_score(prediction=prediction, ground_truth=label) |
| acc = acc_score(prediction=prediction, ground_truths=[label]) |
| return {"em": em, "f1": f1, "acc": acc} |
|
|
| async def async_evaluate(self, graph: Callable, example: Dict) -> float: |
| task = example.get("task", "") |
| context = "\n".join( |
| f"{ctx.get('title', '')}: {' '.join(ctx.get('content', []))}" |
| for ctx in example.get("context", []) |
| if isinstance(ctx, dict) |
| ) |
| inputs = f"Task: {task}\nContext: {context}\nGenerate workflow:\nAnswer:" |
| try: |
| generated_workflow = await graph(inputs) |
| except Exception as e: |
| logger.error(f"Error generating workflow: {e}") |
| generated_workflow = "" |
| label = self._get_label(example) |
| metrics = self.evaluate(prediction=generated_workflow, label=label) |
| return metrics["f1"] |
