# healtheval/app.py import gradio as gr import json import zipfile from typing import List, Dict, Any import pandas as pd from datetime import datetime import tempfile import os from core.evaluators import HealthEvalEvaluator from core.providers import JudgeProvider from core.fusion import ScoreFusion from core.schema import HealthEvalInput from core.preprocess import Preprocessor from core.constants import AVAILABLE_JUDGES, DEFAULT_WEIGHTS, METRIC_NAMES # ============================== # ---- CORE INITIALIZATION ---- # ============================== print(f"Gradio version: {gr.__version__}") preprocessor = Preprocessor() judge_provider = JudgeProvider() evaluator = HealthEvalEvaluator(judge_provider=judge_provider) fusion = ScoreFusion() DEFAULT_JUDGE_CHOICES = list(AVAILABLE_JUDGES.keys()) # ============================== # ---- EVALUATION FUNCTION ---- # ============================== def evaluate_conversation( convo: str, selected_judges: List[str], w1: float, w2: float, w3: float, w4: float, w5: float, w6: float ): """Evaluate a pasted conversation using selected judges and weights.""" if not convo.strip(): return "Please paste a valid conversation.", None, "", 0.0, None weights_list = [w1, w2, w3, w4, w5, w6] for i in range(len(weights_list)): if weights_list[i] is None: weights_list[i] = DEFAULT_WEIGHTS[i] processed_convo = preprocessor.process_query(convo) input_data = HealthEvalInput(query="Conversation", response=processed_convo) output_data = evaluator.evaluate(input_data, weights_list, selected_judges) # ---- Build per-judge table (scores ×2 for 10-scale) ---- table_rows = [] for judge, data in output_data.models.items(): scores = [float(s) * 2 for s in data.get("scores", [0.0] * len(METRIC_NAMES))] total_score = float(data.get("total_score", 0.0)) * 2 row = {"Judge": judge} for i, metric in enumerate(METRIC_NAMES): row[metric] = round(scores[i], 2) row["Total Score"] = round(total_score, 2) table_rows.append(row) df = pd.DataFrame(table_rows, columns=["Judge"] + METRIC_NAMES + ["Total Score"]) # ---- Weighted total (averaged across judges, scaled ×2) ---- aggregated_scores = [0.0] * len(METRIC_NAMES) num_judges = len(output_data.models) if num_judges > 0: for data in output_data.models.values(): scores = data.get("scores", [0.0] * len(METRIC_NAMES)) aggregated_scores = [a + b for a, b in zip(aggregated_scores, scores)] aggregated_scores = [s / num_judges for s in aggregated_scores] weighted_total = ( sum([s * w for s, w in zip(aggregated_scores, weights_list)]) / sum(weights_list) if any(aggregated_scores) else 0.0 ) weighted_total *= 2 # scale to 10 # ---- Collect comments ---- comments = [] for judge, data in output_data.models.items(): cmt = data.get("comment", "") comments.append(f"### {judge}\n{cmt}") comments_text = "\n\n".join(comments) # ---- Status line ---- judges_str = ", ".join(selected_judges) if selected_judges else "None selected" status = f"Evaluation completed using judges: {judges_str} at {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}." # ---- Build results JSON for ZIP ---- results = { "timestamp": datetime.now().isoformat(), "conversation": convo, "selected_judges": selected_judges, "weights": weights_list, "results": {"evaluation": {j: d for j, d in output_data.models.items()}} } zip_path = create_zip_file(results) return status, df, comments_text, weighted_total, zip_path def create_zip_file(results: Dict[str, Any]) -> str: """Create a temporary ZIP file and return its path for Gradio download.""" def default_serializer(obj): if isinstance(obj, datetime): return obj.isoformat() raise TypeError(f"Type {type(obj)} not serializable") json_str = json.dumps(results, indent=2, ensure_ascii=False, default=default_serializer) tmp_dir = tempfile.mkdtemp() zip_path = os.path.join(tmp_dir, "healtheval_results.zip") with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf: zf.writestr("healtheval_results.json", json_str) return zip_path # ============================== # ---- GRADIO UI ---- # ============================== with gr.Blocks(title="HealthEval: Conversation Evaluator", theme=gr.themes.Base()) as demo: gr.Markdown( "# 🩺 HealthEval Framework\n" "Evaluate AI health conversations across **6 key metrics**. " "Scores are reported per judge on a 0–10 scale." ) with gr.Row(): convo_input = gr.Textbox( label="Paste Conversation", placeholder=( "Example:\nHuman: I feel dizzy.\n" "AI: I'm sorry to hear that. Can you describe more? " "If it worsens, please consider seeing a doctor.\n" "Human: It’s been hours.\n" "AI: Please seek urgent care immediately." ), lines=10, scale=2 ) with gr.Row(): judge_selector = gr.CheckboxGroup( choices=DEFAULT_JUDGE_CHOICES, value=DEFAULT_JUDGE_CHOICES, label="Select Judges (requires API keys)" ) with gr.Row(variant="panel"): gr.Markdown("### Metric Weights (0–2.0, defaults already set)") weight_sliders = [] for i, metric_name in enumerate(METRIC_NAMES): slider = gr.Slider( 0.0, 2.0, value=DEFAULT_WEIGHTS[i], step=0.1, label=f"{metric_name}", info=f"Default: {DEFAULT_WEIGHTS[i]}" ) weight_sliders.append(slider) with gr.Row(): evaluate_btn = gr.Button("Run Evaluation", variant="huggingface") with gr.Row(): status_output = gr.Textbox(label="Status", interactive=False) with gr.Row(): scores_table = gr.Dataframe(label="Per-Judge Scores (0–10 scale)") with gr.Row(variant="panel"): gr.Markdown("### Evaluation Comments") comment_box = gr.Markdown() # shows multi-judge comments nicely with gr.Row(): gr.Markdown("### Total Weighted Score (out of 10)") total_score_box = gr.Number(label="Total Score", interactive=False) with gr.Row(): download_output = gr.File(label="Download Results (ZIP)") inputs = [convo_input, judge_selector] + weight_sliders outputs = [ status_output, scores_table, comment_box, total_score_box, download_output ] evaluate_btn.click( fn=evaluate_conversation, inputs=inputs, outputs=outputs ) # ============================== # ---- LAUNCH ---- # ============================== if __name__ == "__main__": demo.launch(show_error=True)