from __future__ import annotations import csv import hashlib import json import re import ssl import sys import urllib.request from datetime import datetime, timezone from decimal import Decimal, ROUND_HALF_UP, InvalidOperation from pathlib import Path from typing import Any BASE = Path("frame_audit_001_bfcl") RAW = BASE / "data" / "raw" PROCESSED = BASE / "data" / "processed" OUTPUTS = BASE / "outputs" META = RAW / "metadata" USER_AGENT = "FrameAudit001-BFCLSnapshot/1.0" CSV_FILES = [ "data_overall.csv", "data_agentic.csv", "data_format_sensitivity.csv", "data_live.csv", "data_multi_turn.csv", "data_non_live.csv", ] DOWNLOADS = [ *[ { "kind": "summary_csv", "name": filename, "url": f"https://gorilla.cs.berkeley.edu/{filename}", "dest": RAW / filename, } for filename in CSV_FILES ], { "kind": "metadata", "name": "bfcl_leaderboard.html", "url": "https://gorilla.cs.berkeley.edu/leaderboard", "dest": META / "bfcl_leaderboard.html", }, { "kind": "metadata", "name": "bfcl_v4_score_composition.html", "url": "https://gorilla.cs.berkeley.edu/blogs/15_bfcl_v4_web_search.html", "dest": META / "bfcl_v4_score_composition.html", }, { "kind": "metadata", "name": "bfcl_readme.md", "url": "https://raw.githubusercontent.com/ShishirPatil/gorilla/main/berkeley-function-call-leaderboard/README.md", "dest": META / "bfcl_readme.md", }, { "kind": "metadata", "name": "bfcl_changelog.md", "url": "https://raw.githubusercontent.com/ShishirPatil/gorilla/main/berkeley-function-call-leaderboard/CHANGELOG.md", "dest": META / "bfcl_changelog.md", }, { "kind": "archive_index", "name": "bfcl_result_archive_root_index.json", "url": "https://api.github.com/repos/HuanzhiMao/BFCL-Result/contents?per_page=100", "dest": META / "bfcl_result_archive_root_index.json", }, { "kind": "archive_index", "name": "bfcl_result_archive_score_index_2025-12-16.json", "url": "https://api.github.com/repos/HuanzhiMao/BFCL-Result/contents/2025-12-16/score?per_page=100", "dest": META / "bfcl_result_archive_score_index_2025-12-16.json", }, { "kind": "archive_index", "name": "bfcl_result_archive_result_index_2025-12-16.json", "url": "https://api.github.com/repos/HuanzhiMao/BFCL-Result/contents/2025-12-16/result?per_page=100", "dest": META / "bfcl_result_archive_result_index_2025-12-16.json", }, ] TEXT_COLUMNS = {"Model", "Model Link", "Organization", "License"} BASELINE_FIELDS = [ "Overall Acc", "Web Search Acc", "Memory Acc", "Multi Turn Acc", "Live Acc", "Non-Live AST Acc", "Irrelevance Detection", ] def utc_now() -> str: return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") def sha256_bytes(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def request_bytes(url: str) -> tuple[bytes, dict[str, str]]: req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT}) ctx = ssl._create_unverified_context() with urllib.request.urlopen(req, context=ctx, timeout=60) as resp: data = resp.read() headers = {k: v for k, v in resp.headers.items()} return data, headers def dnum(value: Any, missing_as_zero: bool = False) -> Decimal | None: if value is None: return Decimal("0") if missing_as_zero else None text = str(value).strip() if text in {"", "N/A", "NA", "null", "None"}: return Decimal("0") if missing_as_zero else None text = text.replace("%", "").replace("$", "").replace(",", "") try: return Decimal(text) except InvalidOperation: return None def q2(value: Decimal) -> Decimal: return value.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) def load_csv(path: Path) -> list[dict[str, str]]: with path.open("r", encoding="utf-8-sig", newline="") as f: return list(csv.DictReader(f)) def csv_quality(path: Path) -> dict[str, Any]: rows = load_csv(path) columns = list(rows[0].keys()) if rows else [] missing_by_col = {col: 0 for col in columns} malformed: list[str] = [] for idx, row in enumerate(rows, start=2): for col in columns: value = row.get(col) if value is None or str(value).strip() in {"", "N/A", "NA", "null", "None"}: missing_by_col[col] += 1 continue if col not in TEXT_COLUMNS and dnum(value) is None: malformed.append(f"row {idx} column {col}: {value!r}") missing_summary = "; ".join(f"{k}={v}" for k, v in missing_by_col.items() if v) return { "row_count": len(rows), "columns": columns, "missing_summary": missing_summary or "none", "malformed_summary": "; ".join(malformed[:20]) + ("; ..." if len(malformed) > 20 else "") if malformed else "none", "missing_by_col": missing_by_col, "malformed_count": len(malformed), } def json_quality(path: Path) -> dict[str, Any]: data = json.loads(path.read_text(encoding="utf-8")) if isinstance(data, list): keys: set[str] = set() for item in data: if isinstance(item, dict): keys.update(item.keys()) return { "row_count": len(data), "columns": sorted(keys), "missing_summary": "not evaluated for JSON index", "malformed_summary": "none", } if isinstance(data, dict): return { "row_count": 1, "columns": sorted(data.keys()), "missing_summary": "not evaluated for JSON object", "malformed_summary": "none", } return { "row_count": "", "columns": [], "missing_summary": "unexpected JSON root", "malformed_summary": "unexpected JSON root", } def non_tabular_quality(path: Path) -> dict[str, Any]: text = path.read_text(encoding="utf-8", errors="replace") return { "row_count": "N/A", "columns": [], "missing_summary": f"not applicable; non-tabular file with {text.count(chr(10)) + 1} lines", "malformed_summary": "not applicable", } def parse_model_date(model: str) -> str: candidates: list[str] = [] for y, m, d in re.findall(r"(20\d{2})[-_/](\d{2})[-_/](\d{2})", model): candidates.append(f"{y}-{m}-{d}") for ymd in re.findall(r"(20\d{6})", model): candidates.append(f"{ymd[0:4]}-{ymd[4:6]}-{ymd[6:8]}") valid: list[str] = [] for candidate in candidates: try: datetime.strptime(candidate, "%Y-%m-%d") valid.append(candidate) except ValueError: pass return max(valid) if valid else "" def parse_snapshot_metadata(leaderboard_html: str) -> dict[str, str]: text = re.sub(r"<[^>]+>", " ", leaderboard_html) text = re.sub(r"\s+", " ", text) last_updated = "" match = re.search(r"Last Updated:\s*(20\d{2}-\d{2}-\d{2})", text, re.I) if match: last_updated = match.group(1) package = "" match = re.search(r"bfcl-eval==[0-9.]+", text) if match: package = match.group(0) commit = "" match = re.search(r"\b([0-9a-f]{7})\b", text) if match: commit = match.group(1) return { "leaderboard_last_updated": last_updated or "not_found_in_downloaded_html", "bfcl_eval_package": package or "not_found_in_downloaded_html", "bfcl_eval_commit": commit or "not_found_in_downloaded_html", } def reconstruct(row: dict[str, str]) -> tuple[Decimal, Decimal, Decimal, list[str]]: missing = [field for field in BASELINE_FIELDS if dnum(row.get(field)) is None] web = dnum(row.get("Web Search Acc"), missing_as_zero=True) memory = dnum(row.get("Memory Acc"), missing_as_zero=True) agentic = (web + memory) / Decimal("2") score = ( Decimal("0.40") * agentic + Decimal("0.30") * dnum(row.get("Multi Turn Acc"), missing_as_zero=True) + Decimal("0.10") * dnum(row.get("Live Acc"), missing_as_zero=True) + Decimal("0.10") * dnum(row.get("Non-Live AST Acc"), missing_as_zero=True) + Decimal("0.10") * dnum(row.get("Irrelevance Detection"), missing_as_zero=True) ) return score, q2(score), agentic, missing def make_archive_index() -> Path: score_items = json.loads((META / "bfcl_result_archive_score_index_2025-12-16.json").read_text(encoding="utf-8")) result_items = json.loads((META / "bfcl_result_archive_result_index_2025-12-16.json").read_text(encoding="utf-8")) score_dirs = {item["name"]: item for item in score_items if item.get("type") == "dir"} result_dirs = {item["name"]: item for item in result_items if item.get("type") == "dir"} names = sorted(set(score_dirs) | set(result_dirs)) out = META / "bfcl_result_archive_index_2025-12-16.csv" with out.open("w", encoding="utf-8", newline="") as f: writer = csv.DictWriter( f, fieldnames=[ "archive_snapshot", "model_variant_directory", "has_score_dir", "has_result_dir", "score_api_url", "result_api_url", "score_html_url", "result_html_url", ], ) writer.writeheader() for name in names: s = score_dirs.get(name, {}) r = result_dirs.get(name, {}) writer.writerow( { "archive_snapshot": "2025-12-16", "model_variant_directory": name, "has_score_dir": str(name in score_dirs).lower(), "has_result_dir": str(name in result_dirs).lower(), "score_api_url": s.get("url", ""), "result_api_url": r.get("url", ""), "score_html_url": s.get("html_url", ""), "result_html_url": r.get("html_url", ""), } ) return out def write_manifest(download_records: list[dict[str, Any]], qualities: dict[str, dict[str, Any]], latest_model_date: str, snapshot: dict[str, str]) -> Path: manifest = RAW / "BFCL_SNAPSHOT_MANIFEST.csv" fieldnames = [ "local_path", "file_name", "file_kind", "source_url", "retrieval_datetime_utc", "http_last_modified", "http_etag", "byte_size", "sha256", "row_count", "column_names", "latest_model_update_represented", "leaderboard_last_updated", "bfcl_eval_package", "bfcl_eval_commit", "missing_or_malformed_fields", ] with manifest.open("w", encoding="utf-8", newline="") as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() for record in download_records: q = qualities[record["name"]] writer.writerow( { "local_path": str(record["dest"]), "file_name": record["name"], "file_kind": record["kind"], "source_url": record["url"], "retrieval_datetime_utc": record["retrieved_at"], "http_last_modified": record["headers"].get("Last-Modified", ""), "http_etag": record["headers"].get("ETag", ""), "byte_size": record["byte_size"], "sha256": record["sha256"], "row_count": q["row_count"], "column_names": "; ".join(q["columns"]), "latest_model_update_represented": latest_model_date, "leaderboard_last_updated": snapshot["leaderboard_last_updated"], "bfcl_eval_package": snapshot["bfcl_eval_package"], "bfcl_eval_commit": snapshot["bfcl_eval_commit"], "missing_or_malformed_fields": f"missing: {q['missing_summary']}; malformed: {q['malformed_summary']}", } ) return manifest def write_baseline(rows: list[dict[str, str]], snapshot: dict[str, str], data_overall_sha: str, retrieved_at: str, csv_last_modified: str) -> tuple[Path, dict[str, Any]]: out = PROCESSED / "bfcl_v4_baseline.csv" baseline_columns = [ "model_identifier", "official_rank", "published_overall_score", "reconstructed_overall_score", "reconstructed_overall_score_unrounded", "absolute_discrepancy", "reconstructed_agentic_component", "web_search_acc", "memory_acc", "multi_turn_acc", "live_acc", "non_live_ast_acc", "hallucination_irrelevance_detection", "relevance_detection", "non_live_simple_ast", "non_live_multiple_ast", "non_live_parallel_ast", "non_live_parallel_multiple_ast", "live_simple_ast", "live_multiple_ast", "live_parallel_ast", "live_parallel_multiple_ast", "multi_turn_base", "multi_turn_miss_func", "multi_turn_miss_param", "multi_turn_long_context", "web_search_base", "web_search_no_snippet", "memory_kv", "memory_vector", "memory_recursive_summarization", "format_sensitivity_max_delta", "format_sensitivity_standard_deviation", "total_cost_usd", "latency_mean_s", "latency_standard_deviation_s", "latency_95th_percentile_s", "complete_baseline_components", "complete_cost_latency_fields", "has_format_sensitivity_scores", "missing_required_baseline_fields", "model_update_date_detected", "model_link", "organization", "license", "leaderboard_last_updated", "bfcl_eval_package", "bfcl_eval_commit", "archive_snapshot", "data_overall_sha256", "data_overall_retrieval_datetime_utc", "data_overall_http_last_modified", ] max_diff = Decimal("0") exact_count = 0 within_count = 0 eligible = 0 missing_required_cells = 0 rows_out: list[dict[str, str]] = [] for row in rows: published = dnum(row.get("Overall Acc")) if published is None or not row.get("Model"): continue eligible += 1 recon_unrounded, recon, agentic, missing = reconstruct(row) diff = abs(recon - published) max_diff = max(max_diff, diff) if diff == 0: exact_count += 1 if diff <= Decimal("0.01"): within_count += 1 missing_required_cells += len(missing) def fmt_decimal(value: Decimal | None) -> str: if value is None: return "" return str(value.normalize()) def val(field: str) -> str: parsed = dnum(row.get(field)) return fmt_decimal(parsed) rows_out.append( { "model_identifier": row.get("Model", ""), "official_rank": row.get("Rank", ""), "published_overall_score": val("Overall Acc"), "reconstructed_overall_score": str(recon), "reconstructed_overall_score_unrounded": str(recon_unrounded), "absolute_discrepancy": str(diff), "reconstructed_agentic_component": str(q2(agentic)), "web_search_acc": val("Web Search Acc"), "memory_acc": val("Memory Acc"), "multi_turn_acc": val("Multi Turn Acc"), "live_acc": val("Live Acc"), "non_live_ast_acc": val("Non-Live AST Acc"), "hallucination_irrelevance_detection": val("Irrelevance Detection"), "relevance_detection": val("Relevance Detection"), "non_live_simple_ast": val("Non-Live Simple AST"), "non_live_multiple_ast": val("Non-Live Multiple AST"), "non_live_parallel_ast": val("Non-Live Parallel AST"), "non_live_parallel_multiple_ast": val("Non-Live Parallel Multiple AST"), "live_simple_ast": val("Live Simple AST"), "live_multiple_ast": val("Live Multiple AST"), "live_parallel_ast": val("Live Parallel AST"), "live_parallel_multiple_ast": val("Live Parallel Multiple AST"), "multi_turn_base": val("Multi Turn Base"), "multi_turn_miss_func": val("Multi Turn Miss Func"), "multi_turn_miss_param": val("Multi Turn Miss Param"), "multi_turn_long_context": val("Multi Turn Long Context"), "web_search_base": val("Web Search Base"), "web_search_no_snippet": val("Web Search No Snippet"), "memory_kv": val("Memory KV"), "memory_vector": val("Memory Vector"), "memory_recursive_summarization": val("Memory Recursive Summarization"), "format_sensitivity_max_delta": val("Format Sensitivity Max Delta"), "format_sensitivity_standard_deviation": val("Format Sensitivity Standard Deviation"), "total_cost_usd": val("Total Cost ($)"), "latency_mean_s": val("Latency Mean (s)"), "latency_standard_deviation_s": val("Latency Standard Deviation (s)"), "latency_95th_percentile_s": val("Latency 95th Percentile (s)"), "complete_baseline_components": str(not missing).lower(), "complete_cost_latency_fields": str( all(dnum(row.get(field)) is not None for field in ["Total Cost ($)", "Latency Mean (s)", "Latency Standard Deviation (s)", "Latency 95th Percentile (s)"]) ).lower(), "has_format_sensitivity_scores": str( dnum(row.get("Format Sensitivity Max Delta")) is not None and dnum(row.get("Format Sensitivity Standard Deviation")) is not None ).lower(), "missing_required_baseline_fields": "; ".join(missing), "model_update_date_detected": parse_model_date(row.get("Model", "")), "model_link": row.get("Model Link", ""), "organization": row.get("Organization", ""), "license": row.get("License", ""), "leaderboard_last_updated": snapshot["leaderboard_last_updated"], "bfcl_eval_package": snapshot["bfcl_eval_package"], "bfcl_eval_commit": snapshot["bfcl_eval_commit"], "archive_snapshot": "2025-12-16", "data_overall_sha256": data_overall_sha, "data_overall_retrieval_datetime_utc": retrieved_at, "data_overall_http_last_modified": csv_last_modified, } ) with out.open("w", encoding="utf-8", newline="") as f: writer = csv.DictWriter(f, fieldnames=baseline_columns) writer.writeheader() writer.writerows(rows_out) stats = { "eligible_model_variants": eligible, "exact_count": exact_count, "within_one_hundredth_count": within_count, "max_discrepancy": str(max_diff), "missing_required_cells": missing_required_cells, "reproduced": eligible > 0 and missing_required_cells == 0 and max_diff <= Decimal("0.01"), } return out, stats def write_report(stats: dict[str, Any], qualities: dict[str, dict[str, Any]]) -> Path: report = OUTPUTS / "BFCL_BASELINE_RECONSTRUCTION_REPORT.md" reproduced_text = "YES - reproduced to the official two-decimal precision" if stats["reproduced"] else "NO" required_issue = "none" if stats["missing_required_cells"] == 0 else f"{stats['missing_required_cells']} missing required baseline cells" format_missing = qualities["data_overall.csv"]["missing_by_col"].get("Format Sensitivity Max Delta", 0) report.write_text( f"""# BFCL Baseline Reconstruction Report Status: baseline reconstruction only. No buyer-profile weights were applied. No paid API calls were made. ## Result - Published aggregate formula reproduced: {reproduced_text} - Models included: {stats['eligible_model_variants']} - Maximum discrepancy: {stats['max_discrepancy']} percentage points - Exact two-decimal matches: {stats['exact_count']} - Matches within 0.01 percentage points: {stats['within_one_hundredth_count']} - Required baseline missing-data issue: {required_issue} ## Formula Used Published BFCL V4 overall score was reconstructed from downloaded `data_overall.csv` fields as: `Agentic = (Web Search Acc + Memory Acc) / 2` `Overall Acc = 0.40 * Agentic + 0.30 * Multi Turn Acc + 0.10 * Live Acc + 0.10 * Non-Live AST Acc + 0.10 * Irrelevance Detection` This follows the documented V4 composition: Agentic 40%, Multi-Turn 30%, Live 10%, Non-Live 10%, and Hallucination Measurement 10%. The downloaded public CSV exposes hallucination measurement in the aggregate through `Irrelevance Detection`; substituting `Relevance Detection` does not reproduce the published aggregate. ## Treatment Of N/A And Missing Categories BFCL documentation states that unevaluated categories appear as `N/A` and summary columns treat unevaluated categories as zero during calculation. The reconstruction used zero for missing required formula fields, but the current downloaded `data_overall.csv` had no missing required formula fields among eligible rows. Format sensitivity was not included in the reconstructed aggregate because BFCL documents it as non-scoring/no-score-impact. In `data_overall.csv`, `Format Sensitivity Max Delta` is missing for {format_missing} rows; this is expected for many FC/native tool-call variants and does not affect the baseline formula. Cost and latency were retained in the processed baseline but were not used in the published aggregate reconstruction. ## Discrepancies All eligible rows reproduced within 0.01 percentage points. The small nonzero differences are consistent with using rounded public component columns to reconstruct a rounded published aggregate. ## Reliability The baseline is reliable enough for sensitivity analysis if subsequent audit steps retain the frozen raw CSVs and use the same missing-data rules. Buyer-profile weighting should not begin until the frozen manifest and processed baseline are treated as immutable inputs. """, encoding="utf-8", ) return report def main() -> int: for directory in [RAW, PROCESSED, OUTPUTS, META]: directory.mkdir(parents=True, exist_ok=True) download_records: list[dict[str, Any]] = [] for spec in DOWNLOADS: data, headers = request_bytes(spec["url"]) spec["dest"].parent.mkdir(parents=True, exist_ok=True) spec["dest"].write_bytes(data) download_records.append( { **spec, "retrieved_at": utc_now(), "headers": headers, "byte_size": len(data), "sha256": sha256_bytes(data), } ) http_meta = { "created_at_utc": utc_now(), "downloads": [ { "name": r["name"], "kind": r["kind"], "url": r["url"], "local_path": str(r["dest"]), "retrieved_at": r["retrieved_at"], "byte_size": r["byte_size"], "sha256": r["sha256"], "headers": r["headers"], } for r in download_records ], } (META / "bfcl_download_http_metadata.json").write_text(json.dumps(http_meta, indent=2, sort_keys=True), encoding="utf-8") qualities: dict[str, dict[str, Any]] = {} for r in download_records: path = r["dest"] if path.suffix.lower() == ".csv": qualities[r["name"]] = csv_quality(path) elif path.suffix.lower() == ".json": qualities[r["name"]] = json_quality(path) else: qualities[r["name"]] = non_tabular_quality(path) overall_rows = load_csv(RAW / "data_overall.csv") latest_model_date = max((parse_model_date(row.get("Model", "")) for row in overall_rows), default="") latest_model_date = latest_model_date or "not_detected" leaderboard_html = (META / "bfcl_leaderboard.html").read_text(encoding="utf-8", errors="replace") snapshot = parse_snapshot_metadata(leaderboard_html) snapshot["latest_model_update_represented"] = latest_model_date snapshot["archive_snapshot"] = "2025-12-16" (META / "bfcl_snapshot_metadata.json").write_text(json.dumps(snapshot, indent=2, sort_keys=True), encoding="utf-8") archive_index = make_archive_index() manifest = write_manifest(download_records, qualities, latest_model_date, snapshot) data_overall_record = next(r for r in download_records if r["name"] == "data_overall.csv") baseline, stats = write_baseline( overall_rows, snapshot, data_overall_record["sha256"], data_overall_record["retrieved_at"], data_overall_record["headers"].get("Last-Modified", ""), ) report = write_report(stats, qualities) if not stats["reproduced"]: print(json.dumps({"status": "failed", "reason": "baseline_not_reproduced", "stats": stats}, indent=2)) return 2 summary = { "downloaded_files": [r["name"] for r in download_records], "generated_files": [ str(manifest), str(baseline), str(report), str(META / "bfcl_download_http_metadata.json"), str(META / "bfcl_snapshot_metadata.json"), str(archive_index), ], "model_variants": stats["eligible_model_variants"], "baseline_reproduced": stats["reproduced"], "maximum_discrepancy": stats["max_discrepancy"], "missing_required_baseline_cells": stats["missing_required_cells"], "format_sensitivity_missing_rows": qualities["data_overall.csv"]["missing_by_col"].get("Format Sensitivity Max Delta", 0), "audit_can_proceed": True, } print(json.dumps(summary, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())