visualize_errors.py - master - PalkinEgor/log_analyzer

visualize_errors.py
509 строк24 KB
PalkinEgor
init commit
17 июн 2026, 01:47
17 июн 2026, 01:4743a6a30
import argparse
import json
from pathlib import Path
from typing import Dict, List, Optional, Tuple

import matplotlib.pyplot as plt
import pandas as pd
import seaborn as sns

CATEGORY_MAPPING_FILENAME = "category_mapping.json"
BENCHMARK_COMPARISON_DIR = "benchmark_comparison"


def load_category_mapping(results_dir: Path) -> Optional[Dict[str, str]]:
    mapping_path = results_dir / CATEGORY_MAPPING_FILENAME
    if not mapping_path.exists():
        return None

    with open(mapping_path, "r", encoding="utf-8") as f:
        mapping = json.load(f)

    print(f"Loaded category mapping from {mapping_path} ({len(mapping)} entries).")
    return mapping


def map_categories(categories: List[str], mapping: Optional[Dict[str, str]]) -> List[str]:
    if not mapping:
        return categories

    mapped: List[str] = []
    seen: set = set()
    for category in categories:
        new_category = mapping.get(category, category)
        if new_category not in seen:
            mapped.append(new_category)
            seen.add(new_category)
    return mapped


def discover_benchmark_dirs(logs_dir: Path) -> Dict[str, Path]:
    benchmark_dirs: Dict[str, Path] = {}
    for child in sorted(logs_dir.iterdir()):
        if child.is_dir() and list(child.rglob("*.json")):
            benchmark_dirs[child.name] = child

    if benchmark_dirs:
        return benchmark_dirs

    if list(logs_dir.rglob("*.json")):
        return {logs_dir.name: logs_dir}

    return {}


def extract_meta(log_data: dict, benchmark: str) -> Tuple[str, str]:
    meta = log_data.get("metadata", {})
    if meta.get("category"):
        return meta.get("category", "unknown"), meta.get("subset", "unknown")

    if benchmark == "supergpqa":
        meta_category = meta.get("prompt_type") or meta.get("discipline") or "unknown"
        meta_subset = meta.get("subfield") or meta.get("field") or "unknown"
        return meta_category, meta_subset

    return meta.get("category", "unknown"), meta.get("subset", "unknown")


def load_error_map(judge_data: list, category_mapping: Optional[Dict[str, str]]) -> Dict[str, List[str]]:
    error_map: Dict[str, List[str]] = {}
    for item in judge_data:
        item_id = item.get("item_id")
        categories = item.get("categories", [])
        if item_id and categories:
            error_map[item_id] = map_categories(categories, category_mapping)
    return error_map


def load_records_from_logs(
    logs_dir: Path,
    error_map: Dict[str, List[str]],
) -> Tuple[List[dict], List[dict]]:
    benchmark_dirs = discover_benchmark_dirs(logs_dir)
    if not benchmark_dirs:
        print(f"No benchmark logs found in {logs_dir}.")
        return [], []

    all_records: List[dict] = []
    error_records: List[dict] = []

    for benchmark, benchmark_dir in benchmark_dirs.items():
        print(f"Loading logs for benchmark '{benchmark}' from {benchmark_dir}...")
        log_files = list(benchmark_dir.rglob("*.json"))
        for log_file in log_files:
            try:
                with open(log_file, "r", encoding="utf-8") as f:
                    log_data = json.load(f)

                item_id = log_data.get("item_id")
                if not item_id:
                    continue

                meta_category, meta_subset = extract_meta(log_data, benchmark)
                is_error = item_id in error_map

                base_record = {
                    "item_id": item_id,
                    "benchmark": benchmark,
                    "meta_category": meta_category,
                    "meta_subset": meta_subset,
                    "response_time": log_data.get("response_time", 0.0),
                    "input_tokens": log_data.get("input_tokens", 0),
                    "output_tokens": log_data.get("output_tokens", 0),
                    "is_error": is_error,
                    "status": "Error" if is_error else "Correct",
                }
                all_records.append(base_record)

                if is_error:
                    for cat in error_map[item_id]:
                        error_rec = base_record.copy()
                        error_rec["error_category"] = cat
                        error_records.append(error_rec)
            except Exception as exc:
                print(f"Error reading {log_file}: {exc}")

    return all_records, error_records


def _save_figure(output_path: Path) -> None:
    plt.tight_layout()
    plt.savefig(output_path, dpi=300)
    plt.close()


def plot_absolute_graphs(df_all: pd.DataFrame, df_err: pd.DataFrame, output_dir: Path, title_suffix: str = "") -> None:
    output_dir.mkdir(parents=True, exist_ok=True)
    suffix = f" ({title_suffix})" if title_suffix else ""

    if not df_err.empty:
        plt.figure(figsize=(12, 8))
        error_counts = df_err["error_category"].value_counts().reset_index()
        error_counts.columns = ["error_category", "count"]
        ax = sns.barplot(data=error_counts, y="error_category", x="count", hue="error_category", legend=False, palette="viridis")
        plt.title(f"Распределение ошибок по категориям{suffix}", fontsize=16)
        plt.xlabel("Количество ошибок", fontsize=14)
        plt.ylabel("Категория ошибки", fontsize=14)
        max_count = error_counts["count"].max()
        for i, value in enumerate(error_counts["count"]):
            ax.text(value + max_count * 0.01, i, str(value), color="black", va="center", fontweight="bold")
        _save_figure(output_dir / "error_counts.png")

        plt.figure(figsize=(14, 10))
        cross_tab = pd.crosstab(df_err["error_category"], df_err["meta_category"])
        sns.heatmap(cross_tab, annot=True, fmt="d", cmap="YlGnBu", linewidths=0.5)
        plt.title(f"Связь категорий ошибок и мета-категорий{suffix}", fontsize=16)
        plt.xlabel("Мета-категория", fontsize=14)
        plt.ylabel("Категория ошибки", fontsize=14)
        _save_figure(output_dir / "error_vs_meta_category.png")

        top_subsets = df_err["meta_subset"].value_counts().nlargest(20).index
        df_top_subsets = df_err[df_err["meta_subset"].isin(top_subsets)]
        if not df_top_subsets.empty:
            plt.figure(figsize=(16, 12))
            cross_tab_subset = pd.crosstab(df_top_subsets["error_category"], df_top_subsets["meta_subset"])
            sns.heatmap(cross_tab_subset, annot=True, fmt="d", cmap="Purples", linewidths=0.5)
            plt.title(f"Связь категорий ошибок и подмножеств (топ-20){suffix}", fontsize=16)
            plt.xlabel("Подмножество", fontsize=14)
            plt.ylabel("Категория ошибки", fontsize=14)
            _save_figure(output_dir / "error_vs_meta_subset.png")

    if df_all.empty:
        return

    plt.figure(figsize=(8, 8))
    status_counts = df_all["status"].value_counts()
    plt.pie(status_counts, labels=status_counts.index, autopct="%1.1f%%", startangle=90, colors=["#2ecc71", "#e74c3c"])
    plt.title(f"Доля успешных ответов и ошибок{suffix}", fontsize=16)
    _save_figure(output_dir / "overall_error_rate_pie.png")

    plt.figure(figsize=(12, 8))
    error_rates = df_all.groupby("meta_category")["is_error"].mean().reset_index()
    error_rates["is_error"] = error_rates["is_error"] * 100
    error_rates = error_rates.sort_values(by="is_error", ascending=False)
    ax = sns.barplot(data=error_rates, x="is_error", y="meta_category", hue="meta_category", legend=False, palette="magma")
    plt.title(f"Доля ошибок по мета-категориям (%){suffix}", fontsize=16)
    plt.xlabel("Процент ошибок (%)", fontsize=14)
    plt.ylabel("Мета-категория", fontsize=14)
    for i, value in enumerate(error_rates["is_error"]):
        ax.text(value + 0.5, i, f"{value:.1f}%", color="black", va="center")
    _save_figure(output_dir / "error_rate_by_category.png")

    plt.figure(figsize=(10, 6))
    sns.boxplot(data=df_all, x="status", y="response_time", hue="status", legend=False, palette=["#2ecc71", "#e74c3c"])
    plt.title(f"Сравнение времени ответа: Успех vs Ошибка{suffix}", fontsize=16)
    plt.xlabel("Статус ответа", fontsize=14)
    plt.ylabel("Время ответа (сек)", fontsize=14)
    if not df_all["response_time"].isnull().all() and df_all["response_time"].max() > 0:
        plt.ylim(0, df_all["response_time"].quantile(0.99) * 1.2)
    _save_figure(output_dir / "response_time_comparison.png")

    plt.figure(figsize=(10, 6))
    sns.boxplot(data=df_all, x="status", y="output_tokens", hue="status", legend=False, palette=["#2ecc71", "#e74c3c"])
    plt.title(f"Сравнение количества сгенерированных токенов: Успех vs Ошибка{suffix}", fontsize=16)
    plt.xlabel("Статус ответа", fontsize=14)
    plt.ylabel("Количество токенов на выходе", fontsize=14)
    if not df_all["output_tokens"].isnull().all() and df_all["output_tokens"].max() > 0:
        plt.ylim(0, df_all["output_tokens"].quantile(0.99) * 1.2)
    _save_figure(output_dir / "output_tokens_comparison.png")


def plot_benchmark_comparison_graphs(df_all: pd.DataFrame, df_err: pd.DataFrame, output_dir: Path) -> None:
    output_dir.mkdir(parents=True, exist_ok=True)
    benchmarks = sorted(df_all["benchmark"].unique())
    if len(benchmarks) < 2:
        print("Skipping benchmark comparison graphs: only one benchmark found.")
        return

    if not df_err.empty:
        category_share = (
            df_err.groupby(["benchmark", "error_category"]).size().reset_index(name="count")
        )
        totals = df_err.groupby("benchmark").size().reset_index(name="total")
        category_share = category_share.merge(totals, on="benchmark")
        category_share["share_pct"] = category_share["count"] / category_share["total"] * 100

        plt.figure(figsize=(14, max(8, len(category_share["error_category"].unique()) * 0.5)))
        pivot_share = category_share.pivot(index="error_category", columns="benchmark", values="share_pct").fillna(0)
        sns.heatmap(pivot_share, annot=True, fmt=".1f", cmap="YlGnBu", linewidths=0.5)
        plt.title("Доля ошибок по категориям внутри каждого бенчмарка (%)", fontsize=16)
        plt.xlabel("Бенчмарк", fontsize=14)
        plt.ylabel("Категория ошибки", fontsize=14)
        _save_figure(output_dir / "error_category_share_by_benchmark.png")

        cross_tab = pd.crosstab(df_err["benchmark"], df_err["error_category"], normalize="index") * 100
        plt.figure(figsize=(14, max(6, len(benchmarks) * 0.8)))
        sns.heatmap(cross_tab, annot=True, fmt=".1f", cmap="viridis", linewidths=0.5)
        plt.title("Распределение категорий ошибок по бенчмаркам (%, по строкам)", fontsize=16)
        plt.xlabel("Категория ошибки", fontsize=14)
        plt.ylabel("Бенчмарк", fontsize=14)
        _save_figure(output_dir / "error_counts.png")

        cross_meta = pd.crosstab(
            [df_err["benchmark"], df_err["error_category"]],
            df_err["meta_category"],
            normalize="index",
        ) * 100
        top_meta = df_err["meta_category"].value_counts().nlargest(15).index
        cross_meta = cross_meta.loc[:, cross_meta.columns.isin(top_meta)]
        if not cross_meta.empty:
            plt.figure(figsize=(16, max(10, len(cross_meta) * 0.35)))
            sns.heatmap(cross_meta, annot=True, fmt=".1f", cmap="YlGnBu", linewidths=0.5)
            plt.title("Связь категорий ошибок и мета-категорий (%, по строкам, топ-15 мета-категорий)", fontsize=16)
            plt.xlabel("Мета-категория", fontsize=14)
            plt.ylabel("Бенчмарк / категория ошибки", fontsize=14)
            _save_figure(output_dir / "error_vs_meta_category.png")

    benchmark_error_rates = df_all.groupby("benchmark")["is_error"].mean().reset_index()
    benchmark_error_rates["error_rate_pct"] = benchmark_error_rates["is_error"] * 100
    benchmark_error_rates = benchmark_error_rates.sort_values(by="error_rate_pct", ascending=False)

    plt.figure(figsize=(10, 6))
    ax = sns.barplot(
        data=benchmark_error_rates,
        x="error_rate_pct",
        y="benchmark",
        hue="benchmark",
        legend=False,
        palette="magma",
    )
    plt.title("Доля ошибок по бенчмаркам (%)", fontsize=16)
    plt.xlabel("Процент ошибок (%)", fontsize=14)
    plt.ylabel("Бенчмарк", fontsize=14)
    for i, value in enumerate(benchmark_error_rates["error_rate_pct"]):
        ax.text(value + 0.5, i, f"{value:.1f}%", color="black", va="center")
    _save_figure(output_dir / "overall_error_rate_by_benchmark.png")

    meta_error_rates = (
        df_all.groupby(["benchmark", "meta_category"])["is_error"].mean().reset_index()
    )
    meta_error_rates["error_rate_pct"] = meta_error_rates["is_error"] * 100
    top_meta_cats = (
        df_all.groupby("meta_category").size().nlargest(15).index
    )
    meta_error_rates = meta_error_rates[meta_error_rates["meta_category"].isin(top_meta_cats)]
    if not meta_error_rates.empty:
        pivot_rates = meta_error_rates.pivot(index="meta_category", columns="benchmark", values="error_rate_pct")
        plt.figure(figsize=(12, 10))
        sns.heatmap(pivot_rates, annot=True, fmt=".1f", cmap="Reds", linewidths=0.5)
        plt.title("Доля ошибок по мета-категориям и бенчмаркам (%, топ-15)", fontsize=16)
        plt.xlabel("Бенчмарк", fontsize=14)
        plt.ylabel("Мета-категория", fontsize=14)
        _save_figure(output_dir / "error_rate_by_category.png")

    n_benchmarks = len(benchmarks)
    fig, axes = plt.subplots(1, n_benchmarks, figsize=(5 * n_benchmarks, 6), sharey=True)
    if n_benchmarks == 1:
        axes = [axes]
    for ax, benchmark in zip(axes, benchmarks):
        subset = df_all[df_all["benchmark"] == benchmark]
        sns.boxplot(data=subset, x="status", y="response_time", hue="status", legend=False, ax=ax, palette=["#2ecc71", "#e74c3c"])
        ax.set_title(benchmark, fontsize=14)
        ax.set_xlabel("Статус ответа")
        if ax is axes[0]:
            ax.set_ylabel("Время ответа (сек)")
        else:
            ax.set_ylabel("")
        if not subset["response_time"].isnull().all() and subset["response_time"].max() > 0:
            ax.set_ylim(0, subset["response_time"].quantile(0.99) * 1.2)
    fig.suptitle("Сравнение времени ответа по бенчмаркам", fontsize=16)
    _save_figure(output_dir / "response_time_comparison.png")

    fig, axes = plt.subplots(1, n_benchmarks, figsize=(5 * n_benchmarks, 6), sharey=True)
    if n_benchmarks == 1:
        axes = [axes]
    for ax, benchmark in zip(axes, benchmarks):
        subset = df_all[df_all["benchmark"] == benchmark]
        sns.boxplot(data=subset, x="status", y="output_tokens", hue="status", legend=False, ax=ax, palette=["#2ecc71", "#e74c3c"])
        ax.set_title(benchmark, fontsize=14)
        ax.set_xlabel("Статус ответа")
        if ax is axes[0]:
            ax.set_ylabel("Количество токенов на выходе")
        else:
            ax.set_ylabel("")
        if not subset["output_tokens"].isnull().all() and subset["output_tokens"].max() > 0:
            ax.set_ylim(0, subset["output_tokens"].quantile(0.99) * 1.2)
    fig.suptitle("Сравнение количества токенов по бенчмаркам", fontsize=16)
    _save_figure(output_dir / "output_tokens_comparison.png")


def plot_aggregated_relative_graphs(df_all: pd.DataFrame, df_err: pd.DataFrame, output_dir: Path) -> None:
    """Aggregated graphs across all benchmarks with relative (percentage) metrics."""
    output_dir.mkdir(parents=True, exist_ok=True)
    suffix = " (все бенчмарки)"

    if not df_err.empty:
        error_counts = df_err["error_category"].value_counts().reset_index()
        error_counts.columns = ["error_category", "count"]
        total_errors = error_counts["count"].sum()
        error_counts["share_pct"] = error_counts["count"] / total_errors * 100

        plt.figure(figsize=(12, 8))
        ax = sns.barplot(
            data=error_counts,
            y="error_category",
            x="share_pct",
            hue="error_category",
            legend=False,
            palette="viridis",
        )
        plt.title(f"Доля ошибок по категориям{suffix}", fontsize=16)
        plt.xlabel("Доля от всех ошибок (%)", fontsize=14)
        plt.ylabel("Категория ошибки", fontsize=14)
        max_pct = error_counts["share_pct"].max()
        for i, row in error_counts.iterrows():
            ax.text(
                row["share_pct"] + max_pct * 0.01,
                i,
                f"{row['share_pct']:.1f}%",
                color="black",
                va="center",
                fontweight="bold",
            )
        _save_figure(output_dir / "error_counts.png")

        cross_tab = pd.crosstab(df_err["error_category"], df_err["meta_category"], normalize="index") * 100
        plt.figure(figsize=(14, 10))
        sns.heatmap(cross_tab, annot=True, fmt=".1f", cmap="YlGnBu", linewidths=0.5)
        plt.title(f"Связь категорий ошибок и мета-категорий{suffix} (%, по строкам)", fontsize=16)
        plt.xlabel("Мета-категория", fontsize=14)
        plt.ylabel("Категория ошибки", fontsize=14)
        _save_figure(output_dir / "error_vs_meta_category.png")

        top_subsets = df_err["meta_subset"].value_counts().nlargest(20).index
        df_top_subsets = df_err[df_err["meta_subset"].isin(top_subsets)]
        if not df_top_subsets.empty:
            cross_tab_subset = pd.crosstab(
                df_top_subsets["error_category"],
                df_top_subsets["meta_subset"],
                normalize="index",
            ) * 100
            plt.figure(figsize=(16, 12))
            sns.heatmap(cross_tab_subset, annot=True, fmt=".1f", cmap="Purples", linewidths=0.5)
            plt.title(f"Связь категорий ошибок и подмножеств (топ-20){suffix} (%, по строкам)", fontsize=16)
            plt.xlabel("Подмножество", fontsize=14)
            plt.ylabel("Категория ошибки", fontsize=14)
            _save_figure(output_dir / "error_vs_meta_subset.png")

    if df_all.empty:
        return

    plt.figure(figsize=(8, 8))
    status_counts = df_all["status"].value_counts()
    plt.pie(status_counts, labels=status_counts.index, autopct="%1.1f%%", startangle=90, colors=["#2ecc71", "#e74c3c"])
    plt.title(f"Доля успешных ответов и ошибок{suffix}", fontsize=16)
    _save_figure(output_dir / "overall_error_rate_pie.png")

    error_rates = df_all.groupby("meta_category")["is_error"].mean().reset_index()
    error_rates["error_rate_pct"] = error_rates["is_error"] * 100
    error_rates = error_rates.sort_values(by="error_rate_pct", ascending=False)
    ax = sns.barplot(
        data=error_rates,
        x="error_rate_pct",
        y="meta_category",
        hue="meta_category",
        legend=False,
        palette="magma",
    )
    plt.title(f"Доля ошибок по мета-категориям{suffix}", fontsize=16)
    plt.xlabel("Процент ошибок (%)", fontsize=14)
    plt.ylabel("Мета-категория", fontsize=14)
    for i, value in enumerate(error_rates["error_rate_pct"]):
        ax.text(value + 0.5, i, f"{value:.1f}%", color="black", va="center")
    _save_figure(output_dir / "error_rate_by_category.png")

    plt.figure(figsize=(10, 6))
    sns.boxplot(data=df_all, x="status", y="response_time", hue="status", legend=False, palette=["#2ecc71", "#e74c3c"])
    plt.title(f"Сравнение времени ответа: Успех vs Ошибка{suffix}", fontsize=16)
    plt.xlabel("Статус ответа", fontsize=14)
    plt.ylabel("Время ответа (сек)", fontsize=14)
    if not df_all["response_time"].isnull().all() and df_all["response_time"].max() > 0:
        plt.ylim(0, df_all["response_time"].quantile(0.99) * 1.2)
    _save_figure(output_dir / "response_time_comparison.png")

    plt.figure(figsize=(10, 6))
    sns.boxplot(data=df_all, x="status", y="output_tokens", hue="status", legend=False, palette=["#2ecc71", "#e74c3c"])
    plt.title(f"Сравнение количества сгенерированных токенов: Успех vs Ошибка{suffix}", fontsize=16)
    plt.xlabel("Статус ответа", fontsize=14)
    plt.ylabel("Количество токенов на выходе", fontsize=14)
    if not df_all["output_tokens"].isnull().all() and df_all["output_tokens"].max() > 0:
        plt.ylim(0, df_all["output_tokens"].quantile(0.99) * 1.2)
    _save_figure(output_dir / "output_tokens_comparison.png")


def parse_args():
    parser = argparse.ArgumentParser(description="Visualize error taxonomy results")
    parser.add_argument(
        "--results_dir",
        type=str,
        default="judge_results/GigaChat-3-Ultra_20260604_024450",
        help="Path to the directory containing results_table.json",
    )
    parser.add_argument(
        "--logs_dir",
        type=str,
        default="test_logs_v2/benchmark_logs",
        help="Path to benchmark logs root (with subdirs per benchmark) or a single benchmark dir",
    )
    return parser.parse_args()


def main():
    args = parse_args()
    results_dir = Path(args.results_dir)
    logs_dir = Path(args.logs_dir)

    results_path = results_dir / "results_table.json"
    if not results_path.exists():
        print(f"Error: {results_path} does not exist.")
        return

    graphs_dir = results_dir / "graphs"
    graphs_dir.mkdir(parents=True, exist_ok=True)

    print(f"Loading error categories from {results_path}...")
    with open(results_path, "r", encoding="utf-8") as f:
        judge_data = json.load(f)

    category_mapping = load_category_mapping(results_dir)
    error_map = load_error_map(judge_data, category_mapping)
    print(f"Loaded {len(error_map)} error records from judge results.")

    if not logs_dir.exists():
        print(f"Error: {logs_dir} does not exist.")
        return

    all_records, error_records = load_records_from_logs(logs_dir, error_map)
    df_all = pd.DataFrame(all_records)
    df_err = pd.DataFrame(error_records)

    if df_all.empty:
        print("No records found in the logs directory.")
        return

    benchmarks = sorted(df_all["benchmark"].unique())
    print(f"Extracted {len(df_all)} total queries across {len(benchmarks)} benchmark(s): {', '.join(benchmarks)}")
    print(f"Error records with categories: {len(df_err)}")

    sns.set_theme(style="whitegrid", rc={"axes.unicode_minus": False})
    plt.rcParams["font.family"] = "sans-serif"

    for benchmark in benchmarks:
        benchmark_graphs_dir = graphs_dir / benchmark
        df_all_b = df_all[df_all["benchmark"] == benchmark]
        df_err_b = df_err[df_err["benchmark"] == benchmark]
        print(f"Generating absolute graphs for '{benchmark}' -> {benchmark_graphs_dir}")
        plot_absolute_graphs(df_all_b, df_err_b, benchmark_graphs_dir, title_suffix=benchmark)

    comparison_dir = graphs_dir / BENCHMARK_COMPARISON_DIR
    print(f"Generating benchmark comparison graphs -> {comparison_dir}")
    plot_benchmark_comparison_graphs(df_all, df_err, comparison_dir)

    print(f"Generating aggregated relative graphs -> {graphs_dir}")
    plot_aggregated_relative_graphs(df_all, df_err, graphs_dir)

    print(f"Visualizations saved to {graphs_dir}")


if __name__ == "__main__":
    main()