import gradio as gr import datasets import huggingface_hub import sys import os import time from pathlib import Path import pandas as pd from transformers.pipelines import TextClassificationPipeline HF_REPO_ID = 'HF_REPO_ID' HF_SPACE_ID = 'SPACE_ID' HF_WRITE_TOKEN = 'HF_WRITE_TOKEN' theme = gr.themes.Soft( primary_hue="green", ) def check_model(model_id): try: task = huggingface_hub.model_info(model_id).pipeline_tag except Exception: return None, None try: from transformers import pipeline ppl = pipeline(task=task, model=model_id) return model_id, ppl except Exception as e: return model_id, e def check_dataset(dataset_id, dataset_config="default", dataset_split="test"): try: configs = datasets.get_dataset_config_names(dataset_id) except Exception: # Dataset may not exist return None, dataset_config, dataset_split if dataset_config not in configs: # Need to choose dataset subset (config) return dataset_id, configs, dataset_split ds = datasets.load_dataset(dataset_id, dataset_config) if isinstance(ds, datasets.DatasetDict): # Need to choose dataset split if dataset_split not in ds.keys(): return dataset_id, None, list(ds.keys()) elif not isinstance(ds, datasets.Dataset): # Unknown type return dataset_id, None, None return dataset_id, dataset_config, dataset_split def text_classificaiton_match_label_case_unsensative(id2label_mapping, label): for model_label in id2label_mapping.keys(): if model_label.upper() == label.upper(): return model_label, label def text_classification_map_model_and_dataset_labels(id2label, dataset_features): id2label_mapping = {id2label[k]: None for k in id2label.keys()} for feature in dataset_features.values(): if not isinstance(feature, datasets.ClassLabel): continue if len(feature.names) != len(id2label_mapping.keys()): continue # Try to match labels for label in feature.names: if label in id2label_mapping.keys(): model_label = label else: # Try to find case unsensative model_label, label = text_classificaiton_match_label_case_unsensative(id2label_mapping, label) id2label_mapping[model_label] = label return id2label_mapping def try_validate(model_id, dataset_id, dataset_config, dataset_split): # Validate model m_id, ppl = check_model(model_id=model_id) if m_id is None: gr.Warning(f'Model "{model_id}" is not accessible. Please set your HF_TOKEN if it is a private model.') return dataset_config, dataset_split, gr.update(interactive=False), gr.update(visible=False), gr.update(visible=False) if isinstance(ppl, Exception): gr.Warning(f'Failed to load "{model_id} model": {ppl}') return dataset_config, dataset_split, gr.update(interactive=False), gr.update(visible=False), gr.update(visible=False) # Validate dataset d_id, config, split = check_dataset(dataset_id=dataset_id, dataset_config=dataset_config, dataset_split=dataset_split) dataset_ok = False if d_id is None: gr.Warning(f'Dataset "{dataset_id}" is not accessible. Please set your HF_TOKEN if it is a private dataset.') elif isinstance(config, list): gr.Warning(f'Dataset "{dataset_id}" does not have "{dataset_config}" config. Please choose a valid config.') config = gr.update(choices=config, value=config[0]) elif isinstance(split, list): gr.Warning(f'Dataset "{dataset_id}" does not have "{dataset_split}" split. Please choose a valid split.') split = gr.update(choices=split, value=split[0]) else: dataset_ok = True if not dataset_ok: return config, split, gr.update(interactive=False), gr.update(visible=False), gr.update(visible=False) # TODO: Validate column mapping by running once prediction_result = {} id2label_df = None if isinstance(ppl, TextClassificationPipeline): # Retrieve all labels id2label_mapping = {} try: results = ppl({"text": "Test"}, top_k=None) prediction_result = { result["label"]: result["score"] for result in results } except Exception as e: # Pipeline is not executable pass # We assume dataset is ok here ds = datasets.load_dataset(d_id, config)[split] try: id2label = ppl.model.config.id2label id2label_mapping = text_classification_map_model_and_dataset_labels(ppl.model.config.id2label, ds.features) id2label_df = pd.DataFrame({ "ID": [i for i in id2label.keys()], "Model labels": [id2label[label] for label in id2label.keys()], "Dataset labels": [id2label_mapping[id2label[label]] for label in id2label.keys()], }) except AttributeError: # Dataset does not have features pass del ppl gr.Info("Model and dataset validations passed. Your can submit the evaluation task.") return config, split, gr.update(interactive=True), gr.update(value=prediction_result, visible=True), gr.update(value=id2label_df, visible=True) def try_submit(m_id, d_id, config, split, local): if local: if "cicd" not in sys.path: sys.path.append("cicd") from giskard_cicd.loaders import HuggingFaceLoader from giskard_cicd.pipeline.runner import PipelineRunner from automation import create_discussion_detailed supported_loaders = { "huggingface": HuggingFaceLoader(), } runner = PipelineRunner(loaders=supported_loaders) runner_kwargs = { "loader_id": "huggingface", "model": m_id, "dataset": d_id, "scan_config": None, "dataset_split": split, "dataset_config": config, } eval_str = f"[{m_id}]<{d_id}({config}, {split} set)>" start = time.time() print(f"Start local evaluation on {eval_str}") report = runner.run(**runner_kwargs) # TODO: Publish it with given repo id/model id if os.environ.get(HF_REPO_ID) or os.environ.get(HF_SPACE_ID) and os.environ.get(HF_WRITE_TOKEN): rendered_report = report.to_markdown(template="github") repo = os.environ.get(HF_REPO_ID) or os.environ.get(HF_SPACE_ID) create_discussion_detailed(repo, m_id, d_id, config, split, os.environ.get(HF_WRITE_TOKEN), rendered_report) # Cache locally rendered_report = report.to_html() output_dir = Path(f"output/{m_id}/{d_id}/{config}/{split}/") output_dir.mkdir(parents=True, exist_ok=True) with open(output_dir / "report.html", "w") as f: print(f'Writing to {output_dir / "report.html"}') f.write(rendered_report) print(f"Finished local evaluation on {eval_str}: {time.time() - start:.2f}s") with gr.Blocks(theme=theme) as iface: with gr.Row(): with gr.Column(): model_id_input = gr.Textbox( label="Hugging Face model id", placeholder="cardiffnlp/twitter-roberta-base-sentiment-latest", ) # TODO: Add supported model pairs: Text Classification - text-classification model_type = gr.Dropdown( label="Hugging Face model type", choices=[ ("Auto-detect", 0), ("Text Classification", 1), ], value=0, ) run_local = gr.Checkbox(value=True, label="Run in this Space") example_labels = gr.Label(label='Model pipeline test prediction result', visible=False) with gr.Column(): dataset_id_input = gr.Textbox( label="Hugging Face dataset id", placeholder="tweet_eval", ) dataset_config_input = gr.Dropdown( label="Hugging Face dataset subset", choices=[ "default", ], allow_custom_value=True, value="default", ) dataset_split_input = gr.Dropdown( label="Hugging Face dataset split", choices=[ "test", ], allow_custom_value=True, value="test", ) id2label_mapping_dataframe = gr.DataFrame(visible=False) with gr.Row(): validate_btn = gr.Button("Validate model and dataset", variant="primary") run_btn = gr.Button( "Submit evaluation task", variant="primary", interactive=False, ) validate_btn.click( try_validate, inputs=[ model_id_input, dataset_id_input, dataset_config_input, dataset_split_input, ], outputs=[ dataset_config_input, dataset_split_input, run_btn, example_labels, id2label_mapping_dataframe, ], ) run_btn.click( try_submit, inputs=[ model_id_input, dataset_id_input, dataset_config_input, dataset_split_input, run_local, ], ) iface.queue(max_size=20) iface.launch()