import html as html_module import markdown from common import METRIC_IDS, RETRIEVAL_RESULT, STATUS_PASS, STATUS_FAIL, STATUS_ERROR, STATUS_PARTIAL, STATUS_UNKNOWN, pascal_case_to_title from datetime import datetime, timezone from evaluator_resolver import EVALUATOR_REGISTRY def calculate_aggregate_statistics(results): """Calculate aggregate statistics across all evaluation results. Scans ALL results (not just the first) to discover which metrics were used, correctly handling per-prompt evaluator variation. Each metric reports prompts_evaluated (how many prompts it actually ran on) and total_prompts. """ if not results: return {} # Flatten: multi-turn threads contribute each turn as a separate item flat_results = [] for result in results: if result.get("type") == "multi_turn": for turn in result.get("turns", []): flat_results.append(turn) else: flat_results.append(result) # Discover all metric keys across all results all_metric_keys = set() for result in flat_results: all_metric_keys.update(result.get('results', {}).keys()) aggregates = {} for eval_name in sorted(all_metric_keys): display_name = pascal_case_to_title(eval_name) metric_id = METRIC_IDS.get(eval_name, eval_name) scores = [] pass_count = 0 fail_count = 0 error_count = 0 threshold_value = None prompts_evaluated = 0 for result in flat_results: parsed_data = result.get('results', {}).get(eval_name) if parsed_data is None: continue # This metric did not run for this prompt if not isinstance(parsed_data, dict): continue prompts_evaluated += 1 try: score = parsed_data.get(metric_id) result_status = parsed_data.get('result') threshold = parsed_data.get('threshold') if score is not None: scores.append(float(score)) if result_status: status = str(result_status).lower() if status == STATUS_PASS: pass_count += 1 elif status == STATUS_FAIL: fail_count += 1 elif status == STATUS_ERROR: error_count += 1 if threshold is not None and threshold_value is None: threshold_value = threshold except (ValueError, TypeError): continue # Surface evaluators that ran in any form — including those whose only # attempts errored. Suppressing error-only evaluators would hide them # from the aggregate report (SC-001). if scores or pass_count > 0 or fail_count > 0 or error_count > 0: avg_score = sum(scores) / len(scores) if scores else 0 # Per-evaluator pass rate is "agreement among completed evaluations" — # errors are surfaced separately as a count, not folded into the rate. total_evaluated = pass_count + fail_count pass_rate = (pass_count / total_evaluated * 100) if total_evaluated > 0 else 0 # Defensive fallback: if no per-entry threshold was recorded # (shouldn't happen — both successful and errored runtime entries # carry it — but guard against malformed input), use the registry # default. Evaluators with no registry default (e.g. ExactMatch) # legitimately have threshold=None. if threshold_value is None: registry_entry = EVALUATOR_REGISTRY.get(eval_name) if registry_entry is not None: threshold_value = registry_entry.default_threshold aggregates[display_name] = { 'total_prompts': len(flat_results), 'prompts_evaluated': prompts_evaluated, 'total_evaluated': total_evaluated, 'pass_count': pass_count, 'fail_count': fail_count, 'error_count': error_count, 'pass_rate': pass_rate, 'avg_score': avg_score, 'threshold': threshold_value, 'scores': scores } return aggregates def format_score(score): try: val = float(score) except (TypeError, ValueError): return score if val.is_integer(): return str(int(val)) s = f"{val:.3f}".rstrip('0').rstrip('.') return s or "0" # Evaluators whose ``reason`` is authored as Markdown and should be rendered as such. _MARKDOWN_REASON_EVALUATORS = frozenset({RETRIEVAL_RESULT}) def extract_metric_rows(entry): """ Build generic metric rows from evaluation results. Each row has: Metric, Result, Score, Threshold, Reason, IsReasonMarkdown. Omits metrics that did not run (None values) for this prompt. """ rows = [] def pick(d, candidates): for k in candidates: if k in d and d[k] not in (None, ''): return d[k] return '' results_container = entry.get('results', {}) if isinstance(entry, dict) else {} for eval_name, metric_obj in results_container.items(): if metric_obj is None: continue # Skip metrics that did not run for this prompt if not isinstance(metric_obj, dict): continue display_name = pascal_case_to_title(eval_name) metric_id = METRIC_IDS.get(eval_name, eval_name) # Candidate key patterns inside the parsed metric object score_val = pick(metric_obj, [metric_id]) result_val = pick(metric_obj, ['result']) threshold_val = pick(metric_obj, ['threshold']) # Errored entries carry the per-evaluator failure description in `error` # (e.g. "Evaluator failed: Connection timeout"). Surface it in the Reason # column so HTML readers see why the evaluator couldn't produce a result. if result_val == STATUS_ERROR: reason_val = metric_obj.get('error', '') # Prefix a machine-readable code (e.g. "emptyAgentResponse") so HTML # readers can distinguish the failure category at a glance. err_code = metric_obj.get('code') if err_code: reason_val = f'[{err_code}] {reason_val}' if reason_val else f'[{err_code}]' else: reason_val = pick(metric_obj, [f'{metric_id}_reason', 'reason']) rows.append({ 'Metric': display_name, 'Result': str(result_val).lower() if isinstance(result_val, str) else result_val, 'Score': format_score(score_val), 'Threshold': format_score(threshold_val), 'Reason': reason_val, 'IsReasonMarkdown': eval_name in _MARKDOWN_REASON_EVALUATORS and result_val != STATUS_ERROR, }) return rows _CHIP_CLASSES = { STATUS_PASS: "status-pass", STATUS_FAIL: "status-fail", STATUS_PARTIAL: "status-partial", STATUS_ERROR: "status-error", } def _chip_class(status): """Map a status value to its chip CSS class. Unknown statuses fall back to status-error.""" return _CHIP_CLASSES.get(status, "status-error") def classify_attempt(entry): """Return one of {pass, fail, partial, error} for an attempt or a thread. For an un-flattened multi-turn thread, returns the thread's overall_status. For a single-turn item or a per-turn entry (from a flattened thread), returns the entry's status — which is set authoritatively by the runner. """ if entry.get("type") == "multi_turn": return entry.get("summary", {}).get("overall_status", STATUS_UNKNOWN) return entry.get("status", STATUS_UNKNOWN) def _escape(text): """HTML-escape user-controlled content to prevent XSS.""" if text is None: return "" return html_module.escape(str(text)) _CELL_CLASSES = { STATUS_PASS: "cell-pass", STATUS_FAIL: "cell-fail", STATUS_ERROR: "cell-error", } def _render_metric_table(html, rows): """Append a metric-table block to ``html`` (no-op if rows is empty).""" if not rows: return html.append(' ') html.append(' ') for row in rows: result_val = str(row.get("Result", "")).lower() cell_class = _CELL_CLASSES.get(result_val) result_attr = f' class="{cell_class}"' if cell_class else "" reason = _escape(str(row.get("Reason", ""))) if row.get("IsReasonMarkdown"): reason_body = markdown.markdown(reason) else: # Plain reasons render verbatim. reason_body = reason.replace("\n", "
") reason_cell = f'
{reason_body}
' html.append( '' f'' f'{_escape(str(row.get("Result", "")))}' f'' f'' f'' '' ) html.append('
MetricResultScoreThresholdReason
{_escape(row.get("Metric", ""))}{_escape(str(row.get("Score", "")))}{_escape(str(row.get("Threshold", "")))}{reason_cell}
') def _render_response_table(html, response, expected_response): """Append the response pair as an accessible, Markdown-aware table.""" response_html = markdown.markdown(_escape(response)) expected_html = markdown.markdown(_escape(expected_response)) html.append(' ') html.append(' ') html.append(' ') html.append(' ') html.append(f' ') html.append(' ') html.append(' ') html.append(' ') html.append(f' ') html.append(' ') html.append(' ') html.append('
Response
{response_html}
Expected Response
{expected_html}
') def _render_tags(html, entry): """Append inline first-class ``tags`` chips if present. Tags are surfaced as lightweight inline chips so the HTML report echoes the same values carried in the JSON output. """ tags = entry.get("tags") if not tags: return chips = ''.join(f'{_escape(str(t))}' for t in tags) html.append(f'
Tags: {chips}
') def _render_diagnostics(html, entry): """Append a collapsible Diagnostics section for ``entry`` if present. Renders the internal ``_diagnostics`` container. Only the ``retrieval_executions`` category exists today; the loop tolerates future sibling categories by rendering retrieval executions when present and skipping unknown keys. No-op when diagnostics are absent. """ diagnostics = entry.get("_diagnostics") if isinstance(entry, dict) else None if not diagnostics: return executions = diagnostics.get("retrieval_executions") or [] html.append('
') html.append(' Retrieval Diagnostics') if not executions: html.append('

No retrieval activity captured.

') for ex in executions: if not isinstance(ex, dict): continue cap = _escape(ex.get("capability")) status = _escape(ex.get("status")) total = ex.get("totalResultCount") total_str = "" if total is None else f" · {_escape(total)} result(s)" # Each execution is its own collapsible block. html.append('
') html.append( f' {cap} ' f'{status}{total_str}' ) for q in ex.get("queries") or []: if not isinstance(q, dict): continue qs = _escape(q.get("queryString")) fe = q.get("filterExpression") fe_str = f' (filter: {_escape(fe)})' if fe else "" html.append(f'
Query: {qs}{fe_str}
') hits = ex.get("retrievalHits") or [] if hits: html.append(' ') html.append(' ') for hit in hits: if not isinstance(hit, dict): continue rank = _escape(hit.get("rank")) url = _escape(hit.get("webUrl")) extracts = hit.get("extracts") or [] text = _escape("\n\n".join( e.get("text", "") for e in extracts if isinstance(e, dict) and isinstance(e.get("text"), str) )) html.append( f' ' f'' ) html.append('
RankURLExtract
{rank}{url}
{text}
') html.append('
') html.append('
') def generate_html_report(results, agent_name=None, agent_id=None, cli_version=None, judge=None): aggregates = calculate_aggregate_statistics(results) # Flatten multi-turn threads for banner counts (consistent with aggregate stats) flat_items = [] for entry in results: if entry.get("type") == "multi_turn": flat_items.extend(entry.get("turns", [])) else: flat_items.append(entry) total_prompts = len(flat_items) counts = {STATUS_PASS: 0, STATUS_FAIL: 0, STATUS_PARTIAL: 0, STATUS_ERROR: 0} for item in flat_items: c = classify_attempt(item) if c in counts: counts[c] += 1 incomplete_count = counts[STATUS_PARTIAL] + counts[STATUS_ERROR] decisive_count = counts[STATUS_PASS] + counts[STATUS_FAIL] overall_pass_rate = (counts[STATUS_PASS] / decisive_count * 100) if decisive_count else 0 generated_utc = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") html = [ '', '', '', ' ', ' ', ' M365 Copilot Agents Evaluation Scores Report', ' ', '', '', '
', '

M365 Copilot Agents Evaluation Report

', ] metadata_items = [] if agent_name: metadata_items.append(f'Agent Name: {_escape(agent_name)}') if agent_id: metadata_items.append(f'Agent ID: {_escape(agent_id)}') if cli_version: metadata_items.append(f'CLI Version: {_escape(cli_version)}') if judge: metadata_items.append(f'Judge: {_escape(judge)}') if metadata_items: html.append(f'

{" | ".join(metadata_items)}

') html.append('
') html.append(f'
Total{total_prompts}
') html.append(f'
Passed{counts[STATUS_PASS]}
') html.append(f'
Failed{counts[STATUS_FAIL]}
') html.append(f'
Incomplete{incomplete_count}
') html.append(f'
Pass Rate{overall_pass_rate:.1f}%
') html.append('
') html.append('
') html.append('

Aggregate Evaluator Statistics

') if aggregates: for metric_name, stats in aggregates.items(): pass_rate = stats.get('pass_rate', 0) prompts_evaluated = stats.get('prompts_evaluated', stats.get('total_evaluated', 0)) html.append('
') avg_score = stats.get('avg_score', 0) threshold_val = stats.get('threshold') threshold_str = "N/A" if threshold_val is None else str(threshold_val) error_count = stats.get('error_count', 0) error_clause = f' / {error_count} error' if error_count else '' html.append( f'
{_escape(metric_name)}' f'{pass_rate:.1f}% ({stats.get("pass_count", 0)} pass / {stats.get("fail_count", 0)} fail{error_clause}, {prompts_evaluated}/{total_prompts} prompts)' f' · Avg Score: {avg_score:.2f} · Threshold: {_escape(threshold_str)}
' ) html.append('
'.format(pass_rate, _escape(metric_name))) html.append(f'
') html.append('
') else: html.append('

No evaluator aggregates available.

') html.append('
') html.append('
') html.append('

Prompt Results

') html.append('
') for idx, entry in enumerate(results, 1): if idx > 1: html.append(' ') if entry.get("type") == "multi_turn": # Multi-turn thread card thread_name = _escape(entry.get("name", "Unnamed Thread")) summary = entry.get("summary", {}) thread_status = summary.get("overall_status", STATUS_UNKNOWN) html.append('
') html.append(f' Entry {idx}/{len(results)}') html.append(f' {thread_status.upper()}') html.append(f'

Thread {idx}: {thread_name}

') html.append(f'

{summary.get("turns_passed", 0)}/{summary.get("turns_total", 0)} turns passed

') for t_idx, turn in enumerate(entry.get("turns", []), 1): turn_status = turn.get("status", STATUS_UNKNOWN) html.append(f'
') html.append(f' {turn_status.upper()}') html.append(f' Turn {t_idx}: {_escape(turn.get("prompt", ""))}') turn_evaluators = turn.get('evaluators_ran', []) if turn_evaluators: badges = ''.join(f'{_escape(e)}' for e in turn_evaluators) html.append(f'

Evaluators: {badges}

') _render_response_table( html, turn.get("response", ""), turn.get("expected_response", ""), ) html.append( f'
' ) turn_error = turn.get("error") if turn_error: html.append( f'

' f'Error: {_escape(turn_error.get("message", ""))}

' ) _render_metric_table(html, extract_metric_rows(turn)) _render_diagnostics(html, turn) _render_tags(html, turn) html.append('
') html.append('
') _render_tags(html, entry) html.append('
') else: score_rows = extract_metric_rows(entry) item_status = classify_attempt(entry) html.append('
') html.append(f' Entry {idx}/{len(results)}') html.append(f' {item_status.upper()}') html.append(f'

Prompt {idx}: {_escape(entry.get("prompt", ""))}

') evaluators_ran = entry.get('evaluators_ran', []) if evaluators_ran: badges = ''.join(f'{_escape(e)}' for e in evaluators_ran) html.append(f'

Evaluators: {badges}

') _render_response_table( html, entry.get("response", ""), entry.get("expected_response", ""), ) html.append( f'
' ) item_error = entry.get('error') if item_error: html.append( f'

' f'Error: {_escape(item_error.get("message", ""))}

' ) _render_metric_table(html, score_rows) _render_diagnostics(html, entry) _render_tags(html, entry) html.append('
') html.append('
') if not results: html.append('

No prompt results found.

') html.append('
') html.append('
') html.append(f'

Generated by M365 Copilot Agents Evaluation CLI —

') html.append('
') html.append('') html.append('') return '\n'.join(html)