""" Inference Comparator - 推論結果比較アプリ 問題文: public_150.json と 複数の推論結果: inference.json を比較するGradioアプリ。 フォーマット不正の行は黄色でハイライト表示する。 """ import gradio as gr import json import re import difflib import html from pathlib import Path from typing import Optional, Tuple, List, Dict import yaml import toml import xml.etree.ElementTree as ET import pandas as pd import io import base64 # 定数 PUBLIC_DEFAULT_PATH = Path(__file__).parent / "public_150.json" HIGHLIGHT_COLOR = "#fff3b0" # 黄色ハイライト # カスタムCSS/JS(外部ファイルから読み込み) CSS_PATH = Path(__file__).parent / "styles.css" JS_PATH = Path(__file__).parent / "scripts.js" def esc(x: str) -> str: """HTMLエスケープ""" return html.escape(x or "") def count_tokens(text: str) -> int: """ 簡易的なトークン数カウント(文字数/4で近似) より正確にはtiktokenを使用するが、依存を減らすため簡易版 算出ロジック: - 入力テキストの文字数を4で割った整数値を返す - 英語テキストでは約4文字≒1トークンが一般的な近似 - 日本語テキストでは1文字≒1-2トークンのため、この近似は過小評価となる - 正確なトークン数が必要な場合はtiktoken等のトークナイザーを使用すること """ return len(text) // 4 if text else 0 def extract_content(text: str, output_type: str) -> Tuple[Optional[str], str]: """ テキストからコードフェンスを除去してコンテンツを抽出する。 """ text = text.strip() fence_pattern = r'```(?:\w+)?\s*\n?(.*?)```' fence_match = re.search(fence_pattern, text, re.DOTALL | re.IGNORECASE) if fence_match: return fence_match.group(1).strip(), "fence" return text, "raw" def validate_format(text: str, output_type: str) -> Tuple[bool, str]: """output_typeに応じたフォーマット検証""" content, _ = extract_content(text, output_type) output_type = output_type.upper() try: if output_type == 'JSON': json.loads(content) elif output_type == 'YAML': yaml.safe_load(content) elif output_type == 'TOML': toml.loads(content) elif output_type == 'XML': ET.fromstring(content) elif output_type == 'CSV': if not content.strip(): raise ValueError("Empty CSV") pd.read_csv(io.StringIO(content)) else: return False, f"Unknown format: {output_type}" return True, "" except Exception as e: return False, str(e) def load_public_data() -> Dict[str, Dict]: """public_150.jsonを読込み、task_id -> metadata のマップを返す""" if not PUBLIC_DEFAULT_PATH.exists(): raise FileNotFoundError( f"public_150.json not found at {PUBLIC_DEFAULT_PATH}. " "Please ensure it is placed in the same directory as app.py." ) with open(PUBLIC_DEFAULT_PATH, 'r', encoding='utf-8') as f: data = json.load(f) return { str(item['task_id']): { 'task_name': item.get('task_name', ''), 'output_type': item.get('output_type', ''), 'query': item.get('query', '') } for item in data } def load_inference_file(file) -> Tuple[str, Dict[str, str], Optional[str]]: """inferenceファイルを読込み""" if file is None: return "", {}, None filename = Path(file.name).name try: with open(file.name, 'r', encoding='utf-8') as f: data = json.load(f) return filename, { str(item['task_id']): item.get('generation', '') for item in data }, None except json.JSONDecodeError as e: return filename, {}, f"JSONパースエラー: {e}" except Exception as e: return filename, {}, f"読込みエラー: {e}" def generate_summary_html( public_index: Dict[str, Dict], inference_names: List[str], inference_maps: List[Dict[str, str]], task_name_filter: str = "", output_type_filter: str = "", sort_column: str = "", sort_order: str = "", invalid_only: bool = False, ) -> str: """一覧HTMLテーブルを生成(フィルタ・ソート対応)""" rows_data = [] task_ids = sorted(public_index.keys()) for tid in task_ids: meta = public_index[tid] task_name = meta['task_name'].strip() # trim output_type = meta['output_type'].strip() # trim query = meta.get('query', '') # フィルタリング if task_name_filter and task_name_filter != task_name: continue if output_type_filter and output_type_filter != output_type: continue # 各inferenceの結果を収集 inf_results = [] any_invalid = False total_tokens = 0 for inf_map in inference_maps: gen = inf_map.get(tid, "") is_valid, error_msg = validate_format(gen, output_type) if not is_valid: any_invalid = True tokens = count_tokens(gen) total_tokens += tokens # 推論結果全文を保存(モーダル表示用) inf_results.append({ 'valid': is_valid, 'tokens': tokens, 'error': error_msg if not is_valid else '', 'generation': gen, # 全文保存 }) # queryのトークン数を計算 query_tokens = count_tokens(query) rows_data.append({ 'tid': tid, 'task_name': task_name, 'output_type': output_type, 'query': query, 'query_tokens': query_tokens, 'any_invalid': any_invalid, 'inf_results': inf_results, 'total_tokens': total_tokens, }) # フォーマット不正のレコードのみ表示フィルタ if invalid_only: rows_data = [r for r in rows_data if r['any_invalid']] # ソート処理 if sort_column and sort_order: reverse = (sort_order == "降順") # queryでソート if sort_column == "query": rows_data.sort( key=lambda x: x['query_tokens'], reverse=reverse ) else: # 特定のinferenceカラムでソート try: col_idx = int(sort_column) rows_data.sort( key=lambda x: x['inf_results'][col_idx]['tokens'] if col_idx < len(x['inf_results']) else 0, reverse=reverse ) except (ValueError, IndexError): pass # HTML生成 rows = [] for row_idx, row in enumerate(rows_data): bg_color = HIGHLIGHT_COLOR if row['any_invalid'] else "#ffffff" tid_disp = esc(row['tid']) task_name_disp = esc(row['task_name']) output_type_disp = esc(row['output_type']) # queryは最初の50文字を表示(長い場合は省略) query_raw = row['query'] query_len = len(query_raw) query_preview = query_raw[:50] + ('...' if query_len > 50 else '') query_disp = esc(query_preview) query_full = esc(query_raw[:200]) # title属性用は200文字まで # Base64エンコードでqueryを安全にコピー query_b64 = base64.b64encode( query_raw.encode('utf-8') ).decode('ascii') row_html = f'
「📝 結果概要」タブで推論" "ファイルをアップロードしてください
") return ( "⚠️ 最低1つのinferenceファイルを" "アップロードしてください
", empty_choices, # task_dropdown empty_choices, # inf_a empty_choices, # inf_b empty_choices, # task_name_filter empty_choices, # output_type_filter empty_choices, # sort_column gr.update(interactive=False), # sort_order (disable) [], [], {}, no_file_msg, # eval_result_html ) try: public_index = load_public_data() except FileNotFoundError as e: empty_choices = gr.update( choices=[], value=None, allow_custom_value=True ) no_file_msg = ("「📝 結果概要」タブで推論" "ファイルをアップロードしてください
") return ( f"⚠️ {str(e)}
", empty_choices, empty_choices, empty_choices, empty_choices, empty_choices, empty_choices, gr.update(interactive=False), [], [], {}, no_file_msg, # eval_result_html ) inference_names = [] inference_maps = [] load_errors = [] for file in files: name, inf_map, error = load_inference_file(file) if error: load_errors.append(f"{esc(name)}: {esc(error)}") else: inference_names.append(name) inference_maps.append(inf_map) if load_errors: error_html = ( "「📝 結果概要」タブで推論" "ファイルをアップロードしてください
") return ( error_html + "⚠️ 有効なinference" "ファイルが必要です
", empty_choices, empty_choices, empty_choices, empty_choices, empty_choices, empty_choices, gr.update(interactive=False), # sort_order (disable) [], [], {}, no_file_msg, # eval_result_html ) summary_html = error_html + generate_summary_html( public_index, inference_names, inference_maps, task_name_filter=task_name_filter_val or "", output_type_filter=output_type_filter_val or "", sort_column=sort_column_val or "", sort_order=sort_order_val or "", invalid_only=invalid_only_val, ) # task_id選択肢(検索用にtask_nameも表示) task_choices = [] for tid in sorted(public_index.keys()): meta = public_index[tid] label = f"{tid[:20]}... | {meta['task_name']}" task_choices.append((label, tid)) # フィルター用の選択肢 task_names = sorted(set(m['task_name'] for m in public_index.values())) output_types = sorted(set(m['output_type'] for m in public_index.values())) # ソートカラム選択肢(queryとinferenceファイル) sort_cols = [("query", "query")] sort_cols.extend([(f"{i}: {name}", str(i)) for i, name in enumerate(inference_names)]) # 詳細比較用のデフォルト値 inf_a_default = inference_names[0] if inference_names else None inf_b_default = (inference_names[1] if len(inference_names) > 1 else inference_names[0] if inference_names else None) # ローカル評価結果を生成 eval_html = run_local_evaluation( inference_names, inference_maps, public_index ) return ( summary_html, gr.update(choices=task_choices, value=task_choices[0][1] if task_choices else None), gr.update(choices=inference_names, value=inf_a_default), gr.update(choices=inference_names, value=inf_b_default), gr.update(choices=[("なし", "")] + [(n, n) for n in task_names], value=""), gr.update(choices=[("なし", "")] + [(t, t) for t in output_types], value=""), gr.update(choices=[("なし", "")] + sort_cols, value=""), gr.update(interactive=False), # sort_order (初期はdisable) inference_names, inference_maps, public_index, eval_html, # eval_result_html ) def update_summary( task_name_filter: str, output_type_filter: str, sort_column: str, sort_order: str, invalid_only: bool, inference_names: List[str], inference_maps: List[Dict[str, str]], public_index: Dict[str, Dict], ): """一覧をフィルタ・ソートして更新""" if not public_index: return "データが読み込まれていません
" return generate_summary_html( public_index, inference_names, inference_maps, task_name_filter=task_name_filter or "", output_type_filter=output_type_filter or "", sort_column=sort_column or "", sort_order=sort_order or "", invalid_only=invalid_only, ) def update_detail( task_id: str, inf_a_name: str, inf_b_name: str, inference_names: List[str], inference_maps: List[Dict[str, str]], public_index: Dict[str, Dict], ): """詳細表示を更新""" if not task_id or not inf_a_name or not inf_b_name: return "", "", "", "", "", "" meta = public_index.get(task_id, {}) query = meta.get('query', 'N/A') query_tokens = count_tokens(query) meta_md = f""" ### Task Information - **task_id**: `{task_id}` - **task_name**: {meta.get('task_name', 'N/A')} - **output_type**: {meta.get('output_type', 'N/A')} ### Query ({query_tokens} tokens){html.escape(query)}
"""
idx_a = (inference_names.index(inf_a_name)
if inf_a_name in inference_names else 0)
idx_b = (inference_names.index(inf_b_name)
if inf_b_name in inference_names else 1)
gen_a = (inference_maps[idx_a].get(task_id, "")
if idx_a < len(inference_maps) else "")
gen_b = (inference_maps[idx_b].get(task_id, "")
if idx_b < len(inference_maps) else "")
tokens_a = count_tokens(gen_a)
tokens_b = count_tokens(gen_b)
label_a = f"### Inference A: {inf_a_name} ({tokens_a} tokens)"
label_b = f"### Inference B: {inf_b_name} ({tokens_b} tokens)"
diff_html = generate_colored_diff(gen_a, gen_b)
return meta_md, gen_a, gen_b, diff_html, label_a, label_b
def evaluate_single_inference(
inf_name: str,
inf_map: Dict[str, str],
public_index: Dict[str, Dict],
) -> Dict:
"""
単一の推論ファイルを評価し、統計情報を返す。
"""
from collections import defaultdict
stats = defaultdict(lambda: {"total": 0, "valid": 0})
errors = []
for tid, meta in public_index.items():
target_fmt = meta.get('output_type', '').upper()
generated_text = inf_map.get(tid, '')
if not target_fmt:
continue
# extract_contentを使ってコードフェンス除去
content, _ = extract_content(generated_text, target_fmt)
is_valid, error_msg = validate_format(generated_text, target_fmt)
stats[target_fmt]["total"] += 1
if is_valid:
stats[target_fmt]["valid"] += 1
else:
errors.append({
"task_id": tid,
"format": target_fmt,
"error": error_msg,
"generated_snippet": generated_text[:100]
})
# 全体統計を計算
total_valid = sum(d["valid"] for d in stats.values())
total_count = sum(d["total"] for d in stats.values())
overall_rate = (total_valid / total_count) * 100 if total_count > 0 else 0
return {
"name": inf_name,
"stats": dict(stats),
"errors": errors,
"total_valid": total_valid,
"total_count": total_count,
"overall_rate": overall_rate,
}
def generate_eval_report_html(eval_result: Dict) -> str:
"""
単一評価結果からHTMLレポートを生成。
"""
stats = eval_result["stats"]
errors = eval_result["errors"]
total_valid = eval_result["total_valid"]
total_count = eval_result["total_count"]
overall_rate = eval_result["overall_rate"]
format_rows = []
for fmt, data in sorted(stats.items()):
rate = (data["valid"] / data["total"]) * 100 if data["total"] > 0 else 0
# プログレスバーの色と状態
if rate >= 80:
bar_color = "#28a745"
status = "✓"
status_color = "#28a745"
elif rate >= 60:
bar_color = "#ffc107"
status = "△"
status_color = "#ffc107"
else:
bar_color = "#dc3545"
status = "✗"
status_color = "#dc3545"
bar_width = int(rate)
format_rows.append(f'''
| Format | Task ID | Error |
|---|
| Format | Progress | Rate | Count | Status |
|---|
{judgment}
「📝 結果概要」タブで推論ファイル" "(inference.json)をアップロードしてください
") try: # 全ファイルを評価 all_results = [] for i, inf_map in enumerate(inference_maps): inf_name = inference_names[i] if i < len(inference_names) \ else f"inference_{i}.json" result = evaluate_single_inference( inf_name, inf_map, public_index ) all_results.append(result) # ソート処理 if sort_order == "スコア降順": all_results.sort(key=lambda x: x["overall_rate"], reverse=True) elif sort_order == "スコア昇順": all_results.sort(key=lambda x: x["overall_rate"], reverse=False) # else: アップロード順(デフォルト) # 単一ファイルの場合はシンプル表示 if len(all_results) == 1: report_html = generate_eval_report_html(all_results[0]) return f'''ファイル: {esc(all_results[0]["name"])}
{report_html}
※ あくまで参考値であり、
LBでの評価と相関があるとは限りません。
詳細な評価基準が公表されていないため、
フォーマットに沿っているかのみを判定しています。
{len(all_results)}件のファイルを評価しました。 各ファイル名をクリックして結果を展開/折りたたみできます。
{''.join(accordion_items)}
※ あくまで参考値であり、
LBでの評価と相関があるとは限りません。
詳細な評価基準が公表されていないため、
フォーマットに沿っているかのみを判定しています。
⚠️ JSONパースエラー: {e}
" except Exception as e: return f"⚠️ エラー: {e}
" def swap_inferences(inf_a: str, inf_b: str): """Inference A と B の選択を入れ替える""" return inf_b, inf_a def load_custom_css() -> str: """styles.cssを読込む""" if CSS_PATH.exists(): with open(CSS_PATH, 'r', encoding='utf-8') as f: return f.read() return "" def get_custom_js() -> str: """scripts.jsを読込む""" if JS_PATH.exists(): with open(JS_PATH, 'r', encoding='utf-8') as f: return f.read() return "" def update_sort_order_visibility(sort_column: str): """ソートカラムに応じてソート順の有効/無効を切り替え""" if sort_column and sort_column != "": return gr.update(interactive=True) else: return gr.update(interactive=False) def update_eval_results( sort_order: str, inference_names: List[str], inference_maps: List[Dict[str, str]], public_index: Dict[str, Dict], ) -> str: """ローカル評価結果をソートして更新""" return run_local_evaluation( inference_names, inference_maps, public_index, sort_order ) # アプリ構築 with gr.Blocks(title="Inference Comparator") as app: gr.Markdown(""" # 🔍 Inference Comparator テストデータ(public_150.json)への推論結果(inference.json)を比較するツール。 **使い方:** 1. 推論ファイル(inference.json)をアップロードする(1ファイル以上必要、複数ファイル選択可) 2. 一覧タブで、推論結果の概要を確認できる(黄色: フォーマット不正を含む) 3. 詳細比較タブで、複数の推論結果を個別比較できる(2ファイル以上必要) """) # State inference_names_state = gr.State([]) inference_maps_state = gr.State([]) public_index_state = gr.State({}) file_input = gr.File( label="📂 推論ファイル(inference.json)のアップロード", file_count="multiple", file_types=[".json"], ) with gr.Tabs(): with gr.TabItem("📝 結果概要"): with gr.Row(): task_name_filter = gr.Dropdown( label="🔍 task_name フィルター", choices=[], value="", allow_custom_value=True, interactive=True, ) output_type_filter = gr.Dropdown( label="🔍 output_type フィルター", choices=[], value="", allow_custom_value=True, interactive=True, ) invalid_only_checkbox = gr.Checkbox( label="⚠️ フォーマット不正のレコードのみ表示", value=False, interactive=True, ) with gr.Row(): sort_column = gr.Dropdown( label="📊 ソートカラム", choices=[], value="", allow_custom_value=True, interactive=True, ) sort_order = gr.Radio( label="ソート順", choices=["昇順", "降順"], value="降順", interactive=False, # 初期状態はdisable ) summary_html = gr.HTML( value="ファイルをアップロードしてください
" ) with gr.TabItem("🔎 詳細比較"): # Task IDドロップダウン(検索機能付き) task_dropdown = gr.Dropdown( label="🔍 Task ID を選択(検索可能: ID / Task Name)", choices=[], interactive=True, allow_custom_value=False, filterable=True, # 検索機能を有効化 ) with gr.Row(): with gr.Column(scale=5): inf_a_dropdown = gr.Dropdown( label="Inference A", choices=[], interactive=True, ) with gr.Column(scale=0, min_width=60): swap_btn = gr.Button( "⇄", size="sm", ) with gr.Column(scale=5): inf_b_dropdown = gr.Dropdown( label="Inference B", choices=[], interactive=True, ) meta_md = gr.Markdown("Task情報がここに表示されます") with gr.Row(): with gr.Column(): label_a = gr.Markdown("### Inference A") text_a = gr.Code( label="", language="json", lines=15, interactive=False, ) with gr.Column(): label_b = gr.Markdown("### Inference B") text_b = gr.Code( label="", language="json", lines=15, interactive=False, ) gr.Markdown("### Diff (A → B)") diff_output = gr.HTML( value="推論ファイルをアップロードすると評価結果が表示されます
" ) # イベントハンドラ - ファイル選択時に即座に処理(フィルター状態を維持) file_input.change( fn=process_files, inputs=[ file_input, task_name_filter, output_type_filter, sort_column, sort_order, invalid_only_checkbox, ], outputs=[ summary_html, task_dropdown, inf_a_dropdown, inf_b_dropdown, task_name_filter, output_type_filter, sort_column, sort_order, # 追加: ソート順の表示/非表示制御 inference_names_state, inference_maps_state, public_index_state, eval_result_html, # ローカル評価結果も更新 ] ) # フィルター・ソート変更時 filter_inputs = [ task_name_filter, output_type_filter, sort_column, sort_order, invalid_only_checkbox, inference_names_state, inference_maps_state, public_index_state, ] task_name_filter.change( fn=update_summary, inputs=filter_inputs, outputs=[summary_html], ) output_type_filter.change( fn=update_summary, inputs=filter_inputs, outputs=[summary_html], ) invalid_only_checkbox.change( fn=update_summary, inputs=filter_inputs, outputs=[summary_html], ) sort_column.change( fn=update_summary, inputs=filter_inputs, outputs=[summary_html], ) # ソートカラム変更時にソート順の表示/非表示を切り替え sort_column.change( fn=update_sort_order_visibility, inputs=[sort_column], outputs=[sort_order], ) sort_order.change( fn=update_summary, inputs=filter_inputs, outputs=[summary_html], ) # 詳細更新トリガー detail_inputs = [ task_dropdown, inf_a_dropdown, inf_b_dropdown, inference_names_state, inference_maps_state, public_index_state, ] detail_outputs = [meta_md, text_a, text_b, diff_output, label_a, label_b] task_dropdown.change( fn=update_detail, inputs=detail_inputs, outputs=detail_outputs ) inf_a_dropdown.change( fn=update_detail, inputs=detail_inputs, outputs=detail_outputs ) inf_b_dropdown.change( fn=update_detail, inputs=detail_inputs, outputs=detail_outputs ) # Inference A と B の入れ替えボタン swap_btn.click( fn=swap_inferences, inputs=[inf_a_dropdown, inf_b_dropdown], outputs=[inf_a_dropdown, inf_b_dropdown], ) # ローカル評価ソート変更時 eval_sort_dropdown.change( fn=update_eval_results, inputs=[ eval_sort_dropdown, inference_names_state, inference_maps_state, public_index_state, ], outputs=[eval_result_html], ) if __name__ == "__main__": # css, jsは Gradio 6.0 以降は launch() に指定 app.launch( theme=gr.themes.Soft(), css=load_custom_css(), js=get_custom_js(), )