# # app.py # """ # Gradio application entrypoint for Hugging Face Spaces. # """ # import os # import tempfile # import pandas as pd # import gradio as gr # from evaluator import evaluate_dataframe # from synthetic_data import generate_synthetic_dataset # # Helper to save uploaded file to local temp path (gradio File gives a NamedTemporaryFile-like object) # def save_uploaded(file_obj): # if not file_obj: # return None # # file_obj can be a dictionary or a file-like object depending on Gradio version # try: # path = file_obj.name # return path # except Exception: # # fallback: write bytes to temp file # data = file_obj.read() # suffix = ".csv" if file_obj.name.endswith(".csv") else ".json" # fd, tmp = tempfile.mkstemp(suffix=suffix) # with os.fdopen(fd, "wb") as f: # f.write(data) # return tmp # def load_file_to_df(path): # if path is None: # return None # # Try CSV # try: # if path.endswith(".csv"): # return pd.read_csv(path) # # JSONL # try: # return pd.read_json(path, lines=True) # except ValueError: # return pd.read_json(path) # except Exception as e: # # As last resort, raise # raise e # def run_evaluation(file_obj): # # If no file provided, use synthetic demo # if file_obj is None: # df = generate_synthetic_dataset(num_agents=3, num_samples=12) # else: # path = save_uploaded(file_obj) # df = load_file_to_df(path) # # Ensure required columns exist; otherwise, attempt to map common alternatives # if df is None: # return None, "No data loaded", None # # Try to normalize column names # cols = {c.lower(): c for c in df.columns} # # rename common variants # rename_map = {} # for k in ["prompt", "response", "task", "agent", "reference"]: # if k not in cols: # # try variants # if k == "reference": # for alt in ["answer", "ground_truth", "ref"]: # if alt in cols: # rename_map[cols[alt]] = k # break # else: # for alt in [k, k.capitalize(), k.upper()]: # if alt.lower() in cols: # rename_map[cols[alt.lower()]] = k # if rename_map: # df = df.rename(columns=rename_map) # metrics_df, images, leaderboard = evaluate_dataframe(df) # # Prepare gallery (list of image file paths). Gradio Gallery accepts list of image paths or PIL images. # gallery_items = [p for (p, caption) in images] # captions = [caption for (p, caption) in images] # # Save a CSV report for download # out_csv = "/tmp/eval_results.csv" # metrics_df.to_csv(out_csv, index=False) # return (gallery_items, captions), metrics_df, leaderboard # # Build Gradio UI # with gr.Blocks() as demo: # gr.Markdown("# Agentic Evaluation Framework") # gr.Markdown( # "Upload a CSV/JSON/JSONL with columns: `prompt,response,task,agent,reference` (reference optional). " # "If no file is uploaded, a small synthetic demo will run." # ) # with gr.Row(): # file_input = gr.File(label="Upload CSV / JSON / JSONL (optional)", file_types=[".csv", ".json", ".jsonl"]) # run_btn = gr.Button("Run Evaluation") # download_report = gr.File(label="Download CSV Report") # # โœ… Fixed Gallery (removed .style, added columns=2) # gallery = gr.Gallery( # label="Visualization Outputs", # columns=2, # height="auto" # ) # table = gr.Dataframe(headers=None, label="Per-example Metrics (detailed)") # leaderboard = gr.Dataframe(headers=None, label="Leaderboard (Avg Final Score per Agent & Task)") # def on_run(file_in): # (gallery_items, captions), metrics_df, lb = run_evaluation(file_in) # # Save gallery captions mapping into a simple list of tuples for Gradio gallery (path, caption) # gallery_display = [] # for i, p in enumerate(gallery_items): # caption = captions[i] if i < len(captions) else "" # gallery_display.append((p, caption)) # return gallery_display, metrics_df, lb # run_btn.click(fn=on_run, inputs=[file_input], outputs=[gallery, table, leaderboard]) # gr.Markdown("## Usage tips\n- Columns: `prompt,response,task,agent,reference` (case-insensitive). " # "- `reference` can be empty but accuracy/hallucination will be weaker.\n" # "- Visualization images are available in the Gallery and a CSV report is downloadable.") # demo.launch() # app.py # """ # Gradio application entrypoint for Hugging Face Spaces. # """ # import os # import tempfile # import pandas as pd # import gradio as gr # from evaluation import evaluate_dataframe # โœ… updated import # from synthetic_data import generate_synthetic_dataset # # Helper to save uploaded file # def save_uploaded(file_obj): # if not file_obj: # return None # try: # return file_obj.name # except Exception: # data = file_obj.read() # suffix = ".csv" if file_obj.name.endswith(".csv") else ".json" # fd, tmp = tempfile.mkstemp(suffix=suffix) # with os.fdopen(fd, "wb") as f: # f.write(data) # return tmp # def load_file_to_df(path): # if path is None: # return None # try: # if path.endswith(".csv"): # return pd.read_csv(path) # try: # return pd.read_json(path, lines=True) # except ValueError: # return pd.read_json(path) # except Exception as e: # raise e # def run_evaluation(file_obj): # if file_obj is None: # df = generate_synthetic_dataset(num_agents=3, num_samples=12) # else: # path = save_uploaded(file_obj) # df = load_file_to_df(path) # if df is None: # return None, "No data loaded", None # # Normalize column names # cols = {c.lower(): c for c in df.columns} # rename_map = {} # for k in ["prompt", "response", "task", "agent", "reference"]: # if k not in cols: # if k == "reference": # for alt in ["answer", "ground_truth", "ref"]: # if alt in cols: # rename_map[cols[alt]] = k # break # else: # for alt in [k, k.capitalize(), k.upper()]: # if alt.lower() in cols: # rename_map[cols[alt.lower()]] = k # if rename_map: # df = df.rename(columns=rename_map) # metrics_df, images, leaderboard = evaluate_dataframe(df) # gallery_items = [p for (p, caption) in images] # captions = [caption for (p, caption) in images] # out_csv = "/tmp/eval_results.csv" # metrics_df.to_csv(out_csv, index=False) # return (gallery_items, captions), metrics_df, leaderboard # # Build Gradio UI # with gr.Blocks() as demo: # gr.Markdown("# Agentic Evaluation Framework") # gr.Markdown( # "Upload a CSV/JSON/JSONL with columns: `prompt,response,task,agent,reference`. " # "If no file is uploaded, a synthetic demo will run." # ) # with gr.Row(): # file_input = gr.File(label="Upload CSV/JSON/JSONL", file_types=[".csv", ".json", ".jsonl"]) # run_btn = gr.Button("Run Evaluation") # download_report = gr.File(label="Download CSV Report") # gallery = gr.Gallery(label="Visualization Outputs", columns=2, height="auto") # table = gr.Dataframe(headers=None, label="Per-example Metrics (detailed)") # leaderboard = gr.Dataframe(headers=None, label="Leaderboard (Avg Score per Agent & Task)") # def on_run(file_in): # (gallery_items, captions), metrics_df, lb = run_evaluation(file_in) # gallery_display = [(p, captions[i] if i < len(captions) else "") for i, p in enumerate(gallery_items)] # return gallery_display, metrics_df, lb # run_btn.click(fn=on_run, inputs=[file_input], outputs=[gallery, table, leaderboard]) # gr.Markdown("## Tips\n- Columns: `prompt,response,task,agent,reference` (case-insensitive). " # "- `reference` optional.\n- Download CSV report after evaluation.") # demo.launch() # app.py (patch) # import gradio as gr # import os # import pandas as pd # from evaluator import evaluate_dataframe # import json # def load_file_to_df(path): # """ # Robust loader: # - try pandas auto-detection (sep=None) # - if result is a single column, re-read as TSV # - fallback to JSON/JSONL # """ # if path is None: # return None # p = str(path) # try: # df = pd.read_csv(p, sep=None, engine="python") # if df.shape[1] == 1: # # probably tab-separated but .csv extension # df = pd.read_csv(p, sep="\t") # return df # except Exception: # pass # try: # return pd.read_json(p, lines=True) # except Exception: # return pd.read_json(p) # def save_uploaded(file_obj): # """Return a filesystem path for the uploaded file object.""" # if not file_obj: # return None # if isinstance(file_obj, str) and os.path.exists(file_obj): # return file_obj # return None # def run_evaluation(file_obj): # path = save_uploaded(file_obj) # if not path: # return "No file uploaded", None, None # df = load_file_to_df(path) # if df is None or df.empty: # return "File could not be parsed", None, None # metrics_df, images, leaderboard = evaluate_dataframe(df) # # Save CSV report # csv_path = "/tmp/eval_report.csv" # metrics_df.to_csv(csv_path, index=False) # # Prepare outputs # plots = [gr.Image(img[0], label=img[1]) for img in images] # leaderboard_table = gr.DataFrame(value=leaderboard) # return csv_path, plots, leaderboard_table # with gr.Blocks() as demo: # gr.Markdown("# ๐Ÿงช Agentic Evaluation Framework") # file_upload = gr.File( # label="Upload CSV", type="filepath", file_types=[".csv", ".tsv", ".json", ".jsonl"] # ) # run_button = gr.Button("Run Evaluation") # download = gr.File(label="Download CSV Report") # plots_output = gr.Group() # leaderboard_output = gr.DataFrame() # run_button.click( # fn=run_evaluation, # inputs=file_upload, # outputs=[download, plots_output, leaderboard_output], # ) # if __name__ == "__main__": # demo.launch() # app.py """ Gradio UI for the agentic evaluation framework. Usage: python app.py Then open the local Gradio URL. The app allows CSV/JSON upload, column mapping, optional judge model selection, and shows results: leaderboard, radar chart, heatmap, and a downloadable CSV. """ import os import json import tempfile import pandas as pd import gradio as gr from evaluator import Evaluator # default judge models (open-source) JUDGE_MODEL_CHOICES = [ "None", "google/flan-t5-base", "google/flan-t5-large" # if you have GPU & memory ] def read_uploaded_file(file_obj): # file_obj is a temporary file-like object from Gradio name = file_obj.name try: if name.endswith(".csv"): return pd.read_csv(name) elif name.endswith(".json") or name.endswith(".jsonl"): try: return pd.read_json(name) except ValueError: # try line-delimited json return pd.read_json(name, lines=True) else: # try csv fallback return pd.read_csv(name) except Exception as e: raise def auto_map_columns(df: pd.DataFrame): # Try to auto detect column names cols = [c.lower() for c in df.columns] mapping = {} def find(candidates): for cand in candidates: if cand in cols: return df.columns[cols.index(cand)] return None mapping["prompt"] = find(["prompt","instruction","input","query","task"]) mapping["response"] = find(["response","output","answer","assistant","reply"]) mapping["agent"] = find(["agent","model","model_name","system"]) mapping["reference"] = find(["reference","target","ground_truth","gold","label"]) mapping["task_type"] = find(["task_type","type","category"]) mapping["metadata"] = find(["metadata","meta","info"]) return mapping def run_full_evaluation(uploaded_file, prompt_col, response_col, agent_col, reference_col, judge_model_choice): # Load df = read_uploaded_file(uploaded_file) # Build mapping mapping = { "prompt": prompt_col, "response": response_col, "agent": agent_col, "reference": reference_col } # Instantiate evaluator judge = None if judge_model_choice in (None, "None") else judge_model_choice evaluator = Evaluator(judge_model_name=judge) # Evaluate per_example = evaluator.evaluate_dataframe(df, mapping, progress=False) agg = evaluator.aggregate_agent_scores(per_example, use_judge_if_available=True) # Visuals leaderboard_fig = evaluator.plot_leaderboard(agg, top_k=20) heatmap_fig = evaluator.plot_metric_heatmap(agg) radar_fig = evaluator.plot_radar(agg, agents_to_plot=agg["agent"].head(5).tolist()) # Return objects: leaderboard table (pandas), and figures return agg, leaderboard_fig, heatmap_fig, radar_fig, per_example # Build Gradio UI with gr.Blocks(title="Agentic Evaluation Framework") as demo: gr.Markdown("# Agentic Evaluation Framework") gr.Markdown("Upload a CSV or JSON containing columns like `prompt, response, agent, reference, metadata`. " "Map columns below if auto-detected mapping is incorrect. Choose an open-source judge model if you'd like LLM-as-a-judge scoring.") with gr.Row(): file_input = gr.File(label="Upload CSV / JSON (columns: prompt, response, agent, reference(optional))") sample_button = gr.Button("Use sample dataset") with gr.Row(): col_prompt = gr.Dropdown(label="Prompt column", choices=[], value=None) col_resp = gr.Dropdown(label="Response column", choices=[], value=None) col_agent = gr.Dropdown(label="Agent column", choices=[], value=None) col_ref = gr.Dropdown(label="Reference column (optional)", choices=[], value=None) with gr.Row(): judge_choice = gr.Dropdown(JUDGE_MODEL_CHOICES, label="LLM judge model (optional)", value="None") run_btn = gr.Button("Run Evaluation", variant="primary") gr.HTML("Note: Using a judge model may require substantial RAM / GPU depending on the model selected.") # Outputs with gr.Tab("Leaderboard"): leaderboard_table = gr.Dataframe(headers=["agent","n_examples","instruction_following","hallucination","assumption_control","coherence","accuracy","overall_score"], interactive=False) leaderboard_plot = gr.Plot() with gr.Tab("Per-metric heatmap"): heatmap_plot = gr.Plot() with gr.Tab("Radar"): radar_plot = gr.Plot() with gr.Tab("Raw per-example"): per_example_table = gr.Dataframe(interactive=False) download_btn = gr.Button("Download per-example CSV") # sample data creation def get_sample_tempfile(): sample = pd.DataFrame([ {"task_type":"summarize","prompt":"Summarize the paragraph about climate change.","agent":"agentA","response":"Climate change warms the planet due to CO2.", "reference":"Climate change is caused by greenhouse gas emissions, especially CO2."}, {"task_type":"qa","prompt":"Who wrote '1984'?", "agent":"agentB", "response":"George Orwell wrote 1984.", "reference":"George Orwell"}, {"task_type":"qa","prompt":"What is the capital of France?", "agent":"agentA","response":"Berlin", "reference":"Paris"}, ]) tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".csv") sample.to_csv(tmp.name, index=False) return tmp.name def load_file_and_populate(file_obj): # file_obj : gradio file object or local path path = file_obj.name if hasattr(file_obj, "name") else file_obj df = pd.read_csv(path) if path.endswith(".csv") else pd.read_json(path, lines=True) cols = list(df.columns) # set choices for dropdowns return gr.update(choices=cols, value=cols[0] if cols else None), gr.update(choices=cols, value=cols[1] if len(cols)>1 else None), gr.update(choices=cols, value=cols[2] if len(cols)>2 else None), gr.update(choices=cols, value=cols[3] if len(cols)>3 else None) # Callbacks def on_sample_click(): tmp = get_sample_tempfile() # populate dropdowns return load_file_and_populate(open(tmp, "rb")) sample_button.click(fn=on_sample_click, inputs=None, outputs=[col_prompt, col_resp, col_agent, col_ref]) def on_file_uploaded(file_obj): # read df to get columns and auto map if file_obj is None: return gr.update(), gr.update(), gr.update(), gr.update() path = file_obj.name try: df = pd.read_csv(path) if path.endswith(".csv") else pd.read_json(path, lines=True) except Exception: df = pd.read_csv(path) cols = list(df.columns) # auto map mapping = {} low = [c.lower() for c in cols] def find(cands): for cand in cands: if cand in low: return cols[low.index(cand)] return None mapping['prompt'] = find(['prompt','instruction','input','query','task']) or cols[0] mapping['response'] = find(['response','output','answer','assistant','reply']) or cols[1] if len(cols)>1 else cols[0] mapping['agent'] = find(['agent','model','model_name','system']) or cols[2] if len(cols)>2 else cols[0] mapping['reference'] = find(['reference','target','ground_truth','gold','label']) or None return gr.update(choices=cols, value=mapping['prompt']), gr.update(choices=cols, value=mapping['response']), gr.update(choices=cols, value=mapping['agent']), gr.update(choices=cols, value=mapping['reference']) file_input.change(fn=on_file_uploaded, inputs=[file_input], outputs=[col_prompt, col_resp, col_agent, col_ref]) def on_run(uploaded_file, prompt_col, response_col, agent_col, reference_col, judge_choice): # Safety: ensure inputs present if uploaded_file is None: return gr.update(), gr.update(), gr.update(), gr.update(), gr.update() agg, lb_fig, hm_fig, rd_fig, per_example = run_full_evaluation(uploaded_file, prompt_col, response_col, agent_col, reference_col, judge_choice) # update outputs return agg, lb_fig, hm_fig, rd_fig, per_example run_btn.click(fn=on_run, inputs=[file_input, col_prompt, col_resp, col_agent, col_ref, judge_choice], outputs=[leaderboard_table, leaderboard_plot, heatmap_plot, radar_plot, per_example_table]) def download_per_example(per_example_df): # per_example_df is a pandas DataFrame from the Gradio widget df = per_example_df tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".csv") df.to_csv(tmp.name, index=False) return tmp.name download_btn.click(fn=download_per_example, inputs=[per_example_table], outputs=[gr.File(label="Download per-example CSV")]) if __name__ == "__main__": demo.launch()