Update app.py

app.py

@@ -391,16 +391,105 @@ def _build_detail_view(results, tasks):
     return CSS + items
 
 # ════════════════════════════════════════════════════════════════
-# PART 10: 메인 평가 루프
+# PART 10: 병렬 평가 엔진 (기둥별 동시 실행)
 # ════════════════════════════════════════════════════════════════
 
+from concurrent.futures import ThreadPoolExecutor, as_completed
+
+def _eval_single_task(task, run_id, api_key, eval_model, judge_model, state):
+    """단일 과제 평가 (모델호출 + Judge채점). 워커 스레드에서 실행."""
+    try:
+        # Step 1: 피평가 모델 호출
+        model_response = call_llm(task.prompt, api_key=api_key, model=eval_model)
+
+        if model_response.startswith("[API_ERROR]"):
+            _save_result(run_id, task.task_id, model_response, "{}", 0)
+            with state["lock"]:
+                state["done"] += 1
+                state["errors"].append(task.task_id)
+            return task.task_id, {"response": model_response, "judge": "{}", "score": 0}
+
+        # Step 2: Judge 채점
+        judge_prompt = build_judge_prompt(task, model_response)
+        judge_raw = call_llm(judge_prompt, system=JUDGE_SYSTEM, api_key=api_key,
+                             model=judge_model, temperature=0.3)
+
+        rubric_keys = list(task.scoring_rubric.keys())
+        judge_data = parse_judge_response(judge_raw, rubric_keys)
+        weighted = compute_weighted_score(judge_data["scores"], task.scoring_rubric)
+
+        judge_json = json.dumps(judge_data, ensure_ascii=False)
+        _save_result(run_id, task.task_id, model_response, judge_json, weighted)
+
+        with state["lock"]:
+            state["done"] += 1
+            info = PILLAR_INFO.get(task.pillar, {})
+            state["active"].append(f'{info.get("icon","")} {task.task_id}')
+            if len(state["active"]) > 10:
+                state["active"] = state["active"][-10:]
+
+        return task.task_id, {"response": model_response, "judge": judge_json, "score": weighted}
+
+    except Exception as e:
+        with state["lock"]:
+            state["done"] += 1
+            state["errors"].append(f"{task.task_id}: {str(e)[:80]}")
+        _save_result(run_id, task.task_id, f"[ERROR] {e}", "{}", 0)
+        return task.task_id, {"response": f"[ERROR] {e}", "judge": "{}", "score": 0}
+
+
+def _parallel_progress_html(state, total):
+    """병렬 실행 진행 상태 HTML"""
+    done = state["done"]
+    pct = min(int(done / max(total, 1) * 100), 100)
+    active = state.get("active", [])
+    errors = state.get("errors", [])
+
+    # 기둥별 진행 상태 바
+    pillar_bars = ""
+    for p, info in PILLAR_INFO.items():
+        p_total = state["pillar_total"].get(p, 0)
+        p_done = state["pillar_done"].get(p, 0)
+        if p_total == 0: continue
+        p_pct = min(int(p_done / p_total * 100), 100)
+        c = "#4caf50" if p_pct == 100 else ("#1976d2" if p_pct > 0 else "#e0e0e0")
+        pillar_bars += f'''<div style="display:flex;align-items:center;gap:8px;margin:3px 0;">
+            <span style="width:100px;font-size:0.85em">{info["icon"]} {info["name"]}</span>
+            <div style="flex:1;background:#e0e0e0;border-radius:6px;height:14px;overflow:hidden">
+                <div style="width:{p_pct}%;height:100%;background:{c};border-radius:6px;transition:width .3s"></div>
+            </div>
+            <span style="width:60px;font-size:0.82em;text-align:right;color:{c}">{p_done}/{p_total}</span>
+        </div>'''
+
+    out = f'''<div style="margin:8px 0;">
+    <div style="display:flex;justify-content:space-between;font-size:0.95em;margin-bottom:6px;">
+        <span>⚡ <b>병렬 평가 진행 중</b> — {done}/{total} 완료</span>
+        <span style="font-weight:700">{pct}%</span>
+    </div>
+    <div class="progress-bar"><div class="progress-fill" style="width:{pct}%"></div></div>
+    <div style="margin-top:8px;">{pillar_bars}</div>'''
+
+    if active:
+        tags = " ".join([f'<span style="background:#e3f2fd;padding:2px 6px;border-radius:4px;font-size:0.78em;">{a}</span>' for a in active[-8:]])
+        out += f'<div style="margin-top:8px;">🔄 최근 완료: {tags}</div>'
+
+    if errors:
+        err_html = " · ".join([f"⚠️{html.escape(e[:30])}" for e in errors[-5:]])
+        out += f'<div style="color:#c62828;margin-top:6px;font-size:0.8em;">{err_html}</div>'
+
+    out += '</div>'
+    return out
+
+
 def run_evaluation(api_key, eval_model, judge_model, pillar_filter, diff_filter,
-                   max_tasks, fresh_start, progress=gr.Progress()):
+                   max_tasks, n_workers, fresh_start, progress=gr.Progress()):
+    """메인 평가 — 기둥별 병렬 실행"""
     api_key = api_key.strip() or os.getenv("FIREWORKS_API_KEY", "")
     if not api_key:
         yield "❌ API Key를 ���력하세요.", "", "", "", None
         return
 
+    # ── 과제 필터링 ──
     tasks = ALL_TASKS[:]
     if pillar_filter != "전체":
         tasks = [t for t in tasks if t.pillar == pillar_filter]
@@ -412,54 +501,100 @@ def run_evaluation(api_key, eval_model, judge_model, pillar_filter, diff_filter,
     if fresh_start:
         _clear_run(run_id)
 
+    # ── 기존 결과 복원 ──
     results = dict(_load_all(run_id))
     total = len(tasks)
-    done = sum(1 for t in tasks if t.task_id in results)
-
-    if done > 0 and not fresh_start:
-        yield (f"💾 체크포인트 복원: {done}/{total}. 이어서 진행.",
-               _build_progress_table(results, tasks), "", "", None)
-        time.sleep(0.5)
-
-    for i, task in enumerate(tasks):
-        if task.task_id in results:
-            continue
-
-        # Step 1: 피평가 모델 호출
-        progress((i + 0.3) / total, desc=f"[{i+1}/{total}] {task.task_id} 모델응답...")
-        yield (f"🤖 [{i+1}/{total}] {task.task_id} ({task.difficulty}) — 모델 응답 대기...",
-               _build_progress_table(results, tasks), "", "", None)
+    cached = sum(1 for t in tasks if t.task_id in results)
 
-        model_response = call_llm(task.prompt, api_key=api_key, model=eval_model)
+    # 미완료 과제만 추출
+    pending = [t for t in tasks if t.task_id not in results]
 
-        if model_response.startswith("[API_ERROR]"):
-            results[task.task_id] = {"response": model_response, "judge": "{}", "score": 0}
-            _save_result(run_id, task.task_id, model_response, "{}", 0)
-            yield (f"⚠️ {task.task_id} API 오류 — 다음 과제로.",
-                   _build_progress_table(results, tasks), "", "", None)
-            continue
-
-        # Step 2: Judge 채점
-        progress((i + 0.7) / total, desc=f"[{i+1}/{total}] {task.task_id} 채점...")
-        yield (f"⚖️ [{i+1}/{total}] {task.task_id} — Judge 채점 중...",
+    if cached > 0 and not fresh_start:
+        yield (f"💾 체크포인트 복원: {cached}/{total} 완료 — {len(pending)}개 남음",
                _build_progress_table(results, tasks), "", "", None)
+        time.sleep(0.5)
 
-        judge_prompt = build_judge_prompt(task, model_response)
-        judge_raw = call_llm(judge_prompt, system=JUDGE_SYSTEM, api_key=api_key,
-                             model=judge_model, temperature=0.3)
-
-        rubric_keys = list(task.scoring_rubric.keys())
-        judge_data = parse_judge_response(judge_raw, rubric_keys)
-        weighted = compute_weighted_score(judge_data["scores"], task.scoring_rubric)
-
-        judge_json = json.dumps(judge_data, ensure_ascii=False)
-        results[task.task_id] = {"response": model_response, "judge": judge_json, "score": weighted}
-        _save_result(run_id, task.task_id, model_response, judge_json, weighted)
-
-        done = sum(1 for t in tasks if t.task_id in results)
-        progress(done / total, desc=f"{done}/{total}")
+    if not pending:
+        # 전부 캐시 히트
+        pillar_scores = {}
+        for p in PILLAR_INFO:
+            pt = [t for t in tasks if t.pillar == p and t.task_id in results]
+            if pt: pillar_scores[p] = np.mean([results[t.task_id]["score"] for t in pt])
+        aether = calculate_aether_score(pillar_scores)
+        csv_str = generate_csv(results, eval_model)
+        csv_path = f"/tmp/aether_eval_{run_id}.csv"
+        with open(csv_path, "w", encoding="utf-8") as f: f.write(csv_str)
+        hf_status = upload_to_hf(csv_str, eval_model)
+        yield (f"🏁 전부 캐시! AETHER Score: {aether:.1f}",
+               _build_progress_table(results, tasks),
+               _build_final_summary(results, tasks, pillar_scores, aether, eval_model, hf_status),
+               _build_detail_view(results, tasks), csv_path)
+        return
 
-    # ── 최종 ──
+    # ── 기둥별 과제 그룹핑 (병렬 단위) ──
+    pillar_tasks = {}
+    for t in pending:
+        pillar_tasks.setdefault(t.pillar, []).append(t)
+
+    n_pillars = len(pillar_tasks)
+    n_workers = int(n_workers)
+
+    # 진행 상태 (스레드 안전)
+    state = {
+        "lock": threading.Lock(),
+        "done": 0,
+        "active": [],
+        "errors": [],
+        "pillar_total": {p: len(ts) for p, ts in pillar_tasks.items()},
+        "pillar_done": {p: 0 for p in pillar_tasks},
+    }
+
+    yield (CSS + f'<div style="background:#e8f5e9;padding:12px;border-radius:8px;margin:8px 0;">'
+           f'⚡ <b>병렬 평가 시작!</b> {len(pending)}개 과제 · {n_pillars}개 기둥 동시 · {n_workers}개 워커'
+           f'</div>', _build_progress_table(results, tasks), "", "", None)
+
+    # ── ThreadPoolExecutor 병렬 실행 ──
+    with ThreadPoolExecutor(max_workers=n_workers) as executor:
+        futures = {}
+        for task in pending:
+            fut = executor.submit(_eval_single_task, task, run_id, api_key,
+                                  eval_model, judge_model, state)
+            futures[fut] = task
+
+        completed = set()
+        while len(completed) < len(futures):
+            newly_done = []
+            for fut in futures:
+                if fut in completed: continue
+                if fut.done():
+                    completed.add(fut)
+                    newly_done.append(fut)
+
+            for fut in newly_done:
+                try:
+                    tid, data = fut.result()
+                    results[tid] = data
+                    # 기둥별 카운터 업데이트
+                    task_obj = futures[fut]
+                    with state["lock"]:
+                        state["pillar_done"][task_obj.pillar] = state["pillar_done"].get(task_obj.pillar, 0) + 1
+                except Exception as e:
+                    with state["lock"]:
+                        state["errors"].append(str(e)[:60])
+
+            # 진행 UI 업데이트
+            with state["lock"]:
+                done_now = cached + state["done"]
+                pct = min(int(done_now / total * 100), 100)
+                progress(done_now / total, desc=f"{done_now}/{total} ({pct}%)")
+                prog_html = CSS + _parallel_progress_html(state, len(pending))
+
+            yield (prog_html, _build_progress_table(results, tasks), "", "", None)
+
+            if len(completed) < len(futures):
+                time.sleep(1.0)
+
+    # ── 최종 결과 ──
     progress(1.0, desc="완료!")
 
     pillar_scores = {}
@@ -470,17 +605,22 @@ def run_evaluation(api_key, eval_model, judge_model, pillar_filter, diff_filter,
     aether = calculate_aether_score(pillar_scores)
 
     csv_str = generate_csv(results, eval_model)
-    csv_path = f"/tmp/aether_eval_{_make_run_id(eval_model)}.csv"
+    csv_path = f"/tmp/aether_eval_{run_id}.csv"
     with open(csv_path, "w", encoding="utf-8") as f:
         f.write(csv_str)
 
     hf_status = upload_to_hf(csv_str, eval_model)
 
+    n_err = len(state["errors"])
+    err_msg = f" (⚠️ {n_err}개 오류)" if n_err > 0 else ""
+    restore_msg = f" (💾 {cached}개 복원)" if cached > 0 else ""
+
     summary = _build_final_summary(results, tasks, pillar_scores, aether, eval_model, hf_status)
     table = _build_progress_table(results, tasks)
     detail = _build_detail_view(results, tasks)
 
-    yield (f"🏁 평가 완료! AETHER Score: {aether:.1f}", table, summary, detail, csv_path)
+    yield (f"🏁 평가 완료!{restore_msg}{err_msg} AETHER Score: {aether:.1f}",
+           table, summary, detail, csv_path)
 
 
 # ════════════════════════════════════════════════════════════════
@@ -526,11 +666,12 @@ def create_app():
             pillar_dd = gr.Dropdown(PILLAR_CHOICES, value="전체", label="기둥 필터", scale=2)
             diff_dd = gr.Dropdown(DIFF_CHOICES, value="전체", label="난이도 필터", scale=2)
             max_tasks = gr.Slider(1, 120, value=120, step=1, label="최대 과제 수", scale=2)
+            n_workers = gr.Slider(1, 20, value=10, step=1, label="⚡ 병렬 워커 수", scale=2)
 
         with gr.Row():
             start_btn = gr.Button("▶️ 평가 시작 (이어하기)", variant="primary", size="lg", scale=2)
             fresh_btn = gr.Button("🚀 새로 시작", variant="secondary", size="lg", scale=2)
-            gr.HTML('<p style="color:#888;font-size:0.8em;margin:auto 0;">▶️ 중단시 이어서 | 🚀 초기화후 재시작<br>결과→CSV→HF PRIVATE 자동 업로드</p>')
+            gr.HTML('<p style="color:#888;font-size:0.8em;margin:auto 0;">⚡ 기둥별 병렬 실행 — 5개 기둥 동시 평가<br>▶️ 중단시 이어서 | 🚀 초기화후 재시작 | CSV→HF PRIVATE</p>')
 
         with gr.Tabs():
             with gr.Tab("📊 진행"):
@@ -545,13 +686,13 @@ def create_app():
                 csv_file = gr.File(label="평가 결과 CSV")
 
         start_btn.click(
-            fn=lambda ak,em,jm,pf,df,mt: run_evaluation(ak,em,jm,pf,df,mt,False),
-            inputs=[api_key, eval_model, judge_model, pillar_dd, diff_dd, max_tasks],
+            fn=lambda ak,em,jm,pf,df,mt,nw: run_evaluation(ak,em,jm,pf,df,mt,nw,False),
+            inputs=[api_key, eval_model, judge_model, pillar_dd, diff_dd, max_tasks, n_workers],
             outputs=[progress_html, table_html, summary_html, detail_html, csv_file],
         )
         fresh_btn.click(
-            fn=lambda ak,em,jm,pf,df,mt: run_evaluation(ak,em,jm,pf,df,mt,True),
-            inputs=[api_key, eval_model, judge_model, pillar_dd, diff_dd, max_tasks],
+            fn=lambda ak,em,jm,pf,df,mt,nw: run_evaluation(ak,em,jm,pf,df,mt,nw,True),
+            inputs=[api_key, eval_model, judge_model, pillar_dd, diff_dd, max_tasks, n_workers],
             outputs=[progress_html, table_html, summary_html, detail_html, csv_file],
         )