Final_Assignment_Template

Sleeping

Update app.py
a212dc5 verified 24 days ago
32.4 kB
"""HF Agents Course 최종 과제 - 다단계 에이전트 버전 (Gemini 또는 Hugging Face)
LLM이 한 단계에 하나의 도구를 골라 쓰고(검색/위키/웹페이지/파이썬/유튜브 자막), 결과를 보고 다음 단계를 정합니다.
최대 MAX_STEPS 단계 안에 최종 답을 내도록 제한해 비용을 통제합니다.
"""
import base64
import json
import mimetypes
import os
import re
import subprocess
import sys
import tempfile
import threading
import time
from collections import deque

import gradio as gr
import pandas as pd
import requests

# ZeroGPU 하드웨어에서 시작 오류를 막기 위한 더미 함수 (CPU 하드웨어에서는 무시됨)
try:
    import spaces

    @spaces.GPU
    def _gpu_placeholder():
        return "ok"
except ImportError:
    pass

try:
    from ddgs import DDGS
except ImportError:
    try:
        from duckduckgo_search import DDGS
    except ImportError:
        DDGS = None

# ---------------------------------------------------------------------------
# 설정 (Space Variables/Secrets로 조정)
# ---------------------------------------------------------------------------
SCORING_API = "https://agents-course-unit4-scoring.hf.space"
GEMINI_API = "https://generativelanguage.googleapis.com/v1beta"
HF_ROUTER = "https://router.huggingface.co/v1"
FILES_DIR = "/tmp/task_files"
CACHE_PATH = "/tmp/answers_cache_v4.json"
os.makedirs(FILES_DIR, exist_ok=True)

PROVIDER = (os.getenv("PROVIDER", "") or ("gemini" if os.getenv("GEMINI_API_KEY") else "hf")).lower()  # gemini | hf
MIN_INTERVAL = float(os.getenv("MIN_INTERVAL", "7" if PROVIDER == "gemini" else "1"))  # LLM 호출 간 최소 간격(초)
MAX_RETRY = int(os.getenv("MAX_RETRY", "3"))                  # 429 재시도 횟수
MAX_STEPS = int(os.getenv("MAX_STEPS", "6"))                  # 문항당 최대 단계(=최대 LLM 호출 수)
QUESTION_TIMEOUT = int(os.getenv("QUESTION_TIMEOUT", "240"))  # 문항당 목표 제한 시간(초). 넘으면 바로 답을 내게 함
OBS_CHARS = int(os.getenv("OBS_CHARS", "6000"))               # 도구 결과 1회 최대 글자 수
PY_TIMEOUT = int(os.getenv("PY_TIMEOUT", "25"))               # 파이썬 실행 제한 시간(초)
GEMINI_MODELS = os.getenv("GEMINI_MODELS", "")                # 선택: 쉼표로 구분한 Gemini 모델 목록
HF_MODELS = [m.strip() for m in (os.getenv("HF_MODELS") or os.getenv("HF_MODEL") or "Qwen/Qwen2.5-72B-Instruct").split(",") if m.strip()]
UA = {"User-Agent": "Mozilla/5.0 (agents-course-runner)"}

AGENT_PROMPT = """You are a careful research agent solving a GAIA-style benchmark question. The final answer is graded by EXACT MATCH.
You work in steps. In each turn reply with EXACTLY ONE action, in this format and nothing else:
Thought: <short reasoning>
Action: <search|wiki|fetch|python|youtube|final>
Input: <input for the action>
Actions:
- search: Input = a web search query. Returns titles, URLs and snippets.
- wiki: Input = "Page Title" or "Page Title | keyword". Returns the English Wikipedia article text including tables; with a keyword you also get the passages around it.
- fetch: Input = a URL, or "URL | keyword". Returns the page text.
- python: Input = Python code (may span several lines). Use print() to output results. pandas, openpyxl, re, json and math are available. Use it for ANY arithmetic, counting, sorting, string manipulation, and for reading attached files.
- youtube: Input = a YouTube URL. Returns the transcript if available.
- final: Input = ONLY the final answer value.
Rules for the final answer:
- Output only the answer value: no explanation, no trailing period, no units unless asked.
- Numbers: digits only, no thousands separators, no currency or percent symbols unless asked.
- Lists: comma-separated, in the order the question asks.
- Strings: no leading articles (a, an, the) and no abbreviations unless asked.
- Follow any formatting instruction inside the question exactly.
Be efficient: you have at most {max_steps} steps. Prefer verifying facts with tools over guessing. If a tool result is not helpful, change your approach instead of repeating it. If you cannot find the answer, give your best reasoned answer rather than leaving it empty."""

# ---------------------------------------------------------------------------
# 로그 / 캐시
# ---------------------------------------------------------------------------
_events = deque(maxlen=14)


def log(msg: str):
    line = f"[{time.strftime('%H:%M:%S')}] {msg}"
    print(line, flush=True)
    _events.append(line)


def recent_log() -> str:
    return "\n".join(_events)


def load_cache() -> dict:
    try:
        with open(CACHE_PATH, "r", encoding="utf-8") as f:
            return json.load(f)
    except Exception:  # noqa: BLE001
        return {}


def save_cache(task_id: str, answer: str):
    cache = load_cache()
    cache[task_id] = answer
    with open(CACHE_PATH, "w", encoding="utf-8") as f:
        json.dump(cache, f, ensure_ascii=False)


# ---------------------------------------------------------------------------
# LLM 호출 (Gemini REST / Hugging Face 라우터)
# ---------------------------------------------------------------------------
class LLMUnavailable(Exception):
    pass


_last_call = 0.0


def throttle():
    global _last_call
    wait = MIN_INTERVAL - (time.time() - _last_call)
    if wait > 0:
        time.sleep(wait)
    _last_call = time.time()


def api_key() -> str:
    return os.getenv("GEMINI_API_KEY", "")


def hf_token() -> str:
    return os.getenv("HF_TOKEN", "")


def list_models() -> list[str]:
    r = requests.get(f"{GEMINI_API}/models", headers={"x-goog-api-key": api_key()}, params={"pageSize": 200}, timeout=30)
    r.raise_for_status()
    return [m["name"].replace("models/", "") for m in r.json().get("models", [])
            if "generateContent" in m.get("supportedGenerationMethods", [])]


_BAD = ("image", "tts", "audio", "live", "embed", "vision", "robotics", "computer", "imagen", "veo")
_candidates: list[str] = []


def candidate_models() -> list[str]:
    global _candidates
    if _candidates:
        return _candidates
    if GEMINI_MODELS.strip():
        _candidates = [m.strip().replace("models/", "") for m in GEMINI_MODELS.split(",") if m.strip()]
        return _candidates
    names = [n for n in list_models() if n.startswith("gemini") and not any(b in n for b in _BAD)]
    lite = [n for n in names if "flash-lite" in n]
    flash = [n for n in names if "flash" in n and "flash-lite" not in n]
    _candidates = (lite[:2] + flash[:2]) or names[:3]
    log(f"자동 선택된 Gemini 모델 후보: {_candidates}")
    return _candidates


def call_gemini(parts: list[dict]) -> tuple[str, str]:
    if not api_key():
        raise LLMUnavailable("GEMINI_API_KEY가 설정되지 않았습니다.")
    last_err = ""
    for model in candidate_models():
        for attempt in range(MAX_RETRY + 1):
            throttle()
            try:
                r = requests.post(
                    f"{GEMINI_API}/models/{model}:generateContent",
                    headers={"x-goog-api-key": api_key(), "Content-Type": "application/json"},
                    json={"contents": [{"role": "user", "parts": parts}],
                          "generationConfig": {"temperature": 0, "maxOutputTokens": 2048}},
                    timeout=120,
                )
            except requests.RequestException as e:
                last_err = f"{model}: 네트워크 오류 {e}"
                time.sleep(5)
                continue
            if r.status_code == 200:
                cand = (r.json().get("candidates") or [{}])[0]
                text = "".join(p.get("text", "") for p in cand.get("content", {}).get("parts", []) if not p.get("thought"))
                if text.strip():
                    return text, model
                last_err = f"{model}: 빈 응답({cand.get('finishReason')})"
                break
            body = r.text
            last_err = f"{model}: HTTP {r.status_code} {body[:180].replace(chr(10), ' ')}"
            if r.status_code == 429:
                if "PerDay" in body or "limit: 0" in body:
                    log(f"{model}: 일일 한도 소진 또는 한도 0 -> 다음 모델")
                    break
                if attempt == MAX_RETRY:
                    break
                m = re.search(r'"retryDelay":\s*"(\d+(?:\.\d+)?)s"', body)
                wait = min(float(m.group(1)) + 1, 60) if m else 15 * (attempt + 1)
                log(f"{model}: 429, {int(wait)}초 후 재시도 ({attempt + 1}/{MAX_RETRY})")
                time.sleep(wait)
                continue
            if r.status_code >= 500:
                time.sleep(5)
                continue
            break
    raise LLMUnavailable(last_err or "사용 가능한 모델이 없습니다.")


def call_hf(parts: list[dict]) -> tuple[str, str]:
    if not hf_token():
        raise LLMUnavailable("HF_TOKEN이 설정되지 않았습니다.")
    text = "\n".join(p["text"] for p in parts if "text" in p)
    if any("inline_data" in p for p in parts):
        text = "NOTE: An image/audio attachment exists but you cannot view it; answer from the other information.\n\n" + text
    last_err = ""
    for model in HF_MODELS:
        for attempt in range(MAX_RETRY + 1):
            throttle()
            try:
                r = requests.post(
                    f"{HF_ROUTER}/chat/completions",
                    headers={"Authorization": f"Bearer {hf_token()}", "Content-Type": "application/json"},
                    json={"model": model, "messages": [{"role": "user", "content": text}],
                          "max_tokens": 1500, "temperature": 0},
                    timeout=120,
                )
            except requests.RequestException as e:
                last_err = f"{model}: 네트워크 오류 {e}"
                time.sleep(5)
                continue
            if r.status_code == 200:
                content = (r.json().get("choices") or [{}])[0].get("message", {}).get("content") or ""
                content = re.sub(r"(?s)<think>.*?</think>", "", content).strip()
                if content:
                    return content, model
                last_err = f"{model}: 빈 응답"
                break
            body = r.text
            last_err = f"{model}: HTTP {r.status_code} {body[:180].replace(chr(10), ' ')}"
            if r.status_code == 402:
                raise LLMUnavailable("HF 크레딧 소진(402). " + body[:150])
            if r.status_code in (401, 403):
                raise LLMUnavailable(f"HF_TOKEN 인증/권한 오류({r.status_code}). Inference Providers 권한을 확인하세요.")
            if r.status_code == 429:
                if attempt == MAX_RETRY:
                    break
                wait = 15 * (attempt + 1)
                log(f"{model}: 429, {wait}초 후 재시도 ({attempt + 1}/{MAX_RETRY})")
                time.sleep(wait)
                continue
            if r.status_code >= 500:
                time.sleep(5)
                continue
            break
    raise LLMUnavailable(last_err or "사용 가능한 모델이 없습니다.")


def call_llm(parts: list[dict]) -> tuple[str, str]:
    return call_hf(parts) if PROVIDER == "hf" else call_gemini(parts)


# ---------------------------------------------------------------------------
# 도구 구현
# ---------------------------------------------------------------------------
def html_to_text(html: str) -> str:
    html = re.sub(r"(?is)<(script|style|noscript|svg|header|footer|nav)[^>]*>.*?</\1>", " ", html)
    html = re.sub(r"(?is)<sup[^>]*class=\"[^\"]*reference[^\"]*\"[^>]*>.*?</sup>", "", html)
    html = re.sub(r"(?i)</t[dh]>", " | ", html)
    html = re.sub(r"(?i)</tr>|<br\s*/?>|</p>|</li>|</h[1-6]>|</caption>", "\n", html)
    text = re.sub(r"(?s)<[^>]+>", " ", html)
    text = (text.replace("&nbsp;", " ").replace("&#160;", " ").replace("&amp;", "&")
            .replace("&lt;", "<").replace("&gt;", ">").replace("&quot;", '"').replace("&#39;", "'"))
    text = re.sub(r"[ \t\u00a0]+", " ", text)
    text = re.sub(r"(\s*\|\s*)+\n", "\n", text)
    text = re.sub(r"\n\s*\n+", "\n", text)
    return text.strip()


def windows(text: str, keyword: str, head: int = 1500, win: int = 1600, max_hits: int = 3) -> str:
    """키워드 주변 문단만 잘라 보여줍니다. 키워드가 없으면 앞부분을 보여줍니다."""
    if not keyword:
        return text[:OBS_CHARS]
    low, key = text.lower(), keyword.lower()
    spans, pos = [], 0
    while len(spans) < max_hits:
        i = low.find(key, pos)
        if i < 0:
            break
        spans.append((max(0, i - win // 4), min(len(text), i + win)))
        pos = i + win
    if not spans:
        return f"(키워드 '{keyword}'를 찾지 못함. 문서 앞부분:)\n" + text[:OBS_CHARS]
    out = [text[:head]] + [f"...\n{text[a:b]}" for a, b in spans]
    return "\n".join(out)[: OBS_CHARS * 2]


def tool_search(query: str) -> str:
    if DDGS is None:
        return "검색 도구를 사용할 수 없습니다."
    try:
        results = list(DDGS().text(query, max_results=6))
    except Exception as e:  # noqa: BLE001
        return f"검색 실패: {type(e).__name__}: {str(e)[:150]}"
    if not results:
        return "검색 결과 없음"
    return "\n\n".join(f"[{i}] {r.get('title', '')}\n{r.get('href', '')}\n{r.get('body', '')}"
                       for i, r in enumerate(results, 1))[:OBS_CHARS]


def _wiki_parse(title: str):
    r = requests.get("https://en.wikipedia.org/w/api.php", headers=UA, timeout=25,
                     params={"action": "parse", "page": title, "prop": "text", "format": "json",
                             "redirects": 1, "formatversion": 2})
    j = r.json()
    if "parse" in j:
        return j["parse"]["text"], j["parse"]["title"]
    return None


def tool_wiki(inp: str) -> str:
    title, _, kw = inp.partition("|")
    title, kw = title.strip(), kw.strip()
    try:
        got = _wiki_parse(title)
        if not got:
            r = requests.get("https://en.wikipedia.org/w/api.php", headers=UA, timeout=25,
                             params={"action": "query", "list": "search", "srsearch": title,
                                     "format": "json", "srlimit": 1})
            hits = r.json().get("query", {}).get("search", [])
            got = _wiki_parse(hits[0]["title"]) if hits else None
        if not got:
            return f"Wikipedia 문서를 찾지 못했습니다: {title}"
        html, real = got
        return f"[Wikipedia: {real}]\n" + windows(html_to_text(html), kw)
    except Exception as e:  # noqa: BLE001
        return f"Wikipedia 조회 실패: {type(e).__name__}: {str(e)[:150]}"


def tool_fetch(inp: str) -> str:
    url, _, kw = inp.partition("|")
    url, kw = url.strip(), kw.strip()
    if not url.startswith("http"):
        return "URL은 http로 시작해야 합니다."
    try:
        r = requests.get(url, headers=UA, timeout=25)
        ctype = r.headers.get("content-type", "")
        if "pdf" in ctype or url.lower().endswith(".pdf"):
            return "PDF는 지원하지 않습니다."
        if "html" not in ctype and "text" not in ctype and "json" not in ctype:
            return f"지원하지 않는 형식입니다: {ctype}"
        text = html_to_text(r.text) if "html" in ctype else r.text
        return windows(text, kw)
    except Exception as e:  # noqa: BLE001
        return f"페이지를 가져오지 못했습니다: {type(e).__name__}: {str(e)[:150]}"


def tool_python(code: str) -> str:
    code = re.sub(r"^```(?:python)?\s*\n?", "", code.strip())
    code = re.sub(r"\n?```\s*$", "", code)
    with tempfile.NamedTemporaryFile("w", suffix=".py", dir="/tmp", delete=False, encoding="utf-8") as f:
        f.write(code)
        path = f.name
    # 토큰 등 비밀 환경변수가 코드에 노출되지 않도록 최소 환경만 전달
    env = {"PATH": os.environ.get("PATH", ""), "HOME": "/tmp", "PYTHONIOENCODING": "utf-8"}
    try:
        r = subprocess.run([sys.executable, path], capture_output=True, text=True,
                           timeout=PY_TIMEOUT, cwd=FILES_DIR, env=env)
        out = r.stdout or ""
        if r.stderr:
            out += "\n[stderr]\n" + r.stderr[-1500:]
    except subprocess.TimeoutExpired:
        out = f"(실행 시간 초과 {PY_TIMEOUT}초)"
    except Exception as e:  # noqa: BLE001
        out = f"실행 오류: {e}"
    finally:
        try:
            os.remove(path)
        except OSError:
            pass
    return out[:OBS_CHARS] if out.strip() else "(출력 없음. 결과는 print()로 출력해야 합니다.)"


def tool_youtube(url: str) -> str:
    m = re.search(r"(?:v=|youtu\.be/)([\w-]{11})", url)
    if not m:
        return "유효한 YouTube URL이 아닙니다."
    try:
        from youtube_transcript_api import YouTubeTranscriptApi
    except ImportError:
        return "youtube-transcript-api가 설치되지 않았습니다."
    vid = m.group(1)
    try:
        try:
            items = YouTubeTranscriptApi.get_transcript(vid)
            text = " ".join(i.get("text", "") for i in items)
        except AttributeError:
            text = " ".join(getattr(s, "text", "") for s in YouTubeTranscriptApi().fetch(vid))
        return text[: OBS_CHARS * 2] or "자막이 비어 있습니다."
    except Exception as e:  # noqa: BLE001
        return f"자막을 가져오지 못했습니다(영상 내용은 볼 수 없음): {type(e).__name__}: {str(e)[:150]}"


TOOLS = {"search": tool_search, "wiki": tool_wiki, "fetch": tool_fetch, "python": tool_python, "youtube": tool_youtube}

# ---------------------------------------------------------------------------
# 첨부 파일
# ---------------------------------------------------------------------------
def download_file(task_id: str, file_name: str):
    if not file_name:
        return None
    try:
        r = requests.get(f"{SCORING_API}/files/{task_id}", timeout=60)
        r.raise_for_status()
        path = os.path.join(FILES_DIR, f"{task_id}_{file_name}")
        with open(path, "wb") as f:
            f.write(r.content)
        return path
    except Exception as e:  # noqa: BLE001
        log(f"파일 다운로드 실패: {e}")
        return None


def file_to_parts(path: str) -> tuple[str, list[dict]]:
    """(미리보기 텍스트, Gemini에 직접 넣을 inline_data 파트들)"""
    lower = path.lower()
    try:
        if lower.endswith((".png", ".jpg", ".jpeg", ".webp", ".mp3", ".wav", ".m4a")):
            if os.path.getsize(path) > 15 * 1024 * 1024:
                return "(파일이 너무 커서 생략됨)", []
            mime = "audio/mp3" if lower.endswith(".mp3") else (mimetypes.guess_type(path)[0] or "application/octet-stream")
            with open(path, "rb") as f:
                data = base64.b64encode(f.read()).decode()
            return "(이미지/오디오 파일. Gemini일 때만 내용을 볼 수 있음)", [{"inline_data": {"mime_type": mime, "data": data}}]
        if lower.endswith((".xlsx", ".xls")):
            sheets = pd.read_excel(path, sheet_name=None)
            return "\n\n".join(f"## Sheet {n} (shape {df.shape})\n{df.head(25).to_string()}" for n, df in sheets.items()), []
        if lower.endswith(".csv"):
            df = pd.read_csv(path)
            return f"shape {df.shape}\n{df.head(25).to_string()}", []
        with open(path, "r", encoding="utf-8", errors="ignore") as f:
            return f.read()[:12000], []
    except Exception as e:  # noqa: BLE001
        return f"(파일 읽기 실패: {e})", []


# ---------------------------------------------------------------------------
# 에이전트 루프
# ---------------------------------------------------------------------------
def clean_answer(ans: str) -> str:
    ans = (ans.strip().splitlines() or [""])[0]
    return ans.strip().strip("`").strip().strip('"').strip().rstrip(".").strip()


def parse_action(text: str):
    m = re.search(r"Action:\s*([A-Za-z_]+)\s*\n?\s*Input:\s*(.*)", text, flags=re.S | re.I)
    if m:
        inp = re.split(r"\n\s*Observation:", m.group(2))[0].strip()
        return m.group(1).lower(), inp
    m = re.findall(r"FINAL ANSWER:\s*(.+)", text, flags=re.I)
    if m:
        return "final", m[-1].strip()
    return None, ""


def solve(item: dict) -> tuple[str, str]:
    """(답변, 비고)를 반환합니다."""
    question = item["question"]
    extra_parts: list[dict] = []
    file_note = ""
    path = download_file(item["task_id"], item.get("file_name", ""))
    if path:
        preview, extra_parts = file_to_parts(path)
        file_note = (f"ATTACHED FILE path: {path}\nPreview:\n{preview[:6000]}\n"
                     "(The file can be read by the python action at that exact path.)\n\n")

    head = AGENT_PROMPT.replace("{max_steps}", str(MAX_STEPS)) + "\n\nQUESTION:\n" + question + "\n\n" + file_note
    history: list[str] = []
    start = time.time()
    model_used, last_text = "", ""

    for step in range(1, MAX_STEPS + 1):
        last = step == MAX_STEPS or (time.time() - start) > QUESTION_TIMEOUT
        prompt = head + "\n".join(history)
        prompt += ("\n\nThis is your LAST step: reply now with Action: final." if last
                   else "\n\nReply with your next step (one action).")
        text, model_used = call_llm([{"text": prompt}] + extra_parts)
        last_text = text
        action, inp = parse_action(text)
        log(f"  [{question[:28]}…] 단계 {step}: {action or '형식오류'} {inp[:50].replace(chr(10), ' ')}")

        if action == "final":
            return clean_answer(inp), f"{model_used} / {step}단계"
        if last:
            break
        if action not in TOOLS:
            history.append(f"[Step {step}]\n{text[:800]}\nObservation: Invalid format or action. Reply exactly with "
                           "Thought/Action/Input using one of: search, wiki, fetch, python, youtube, final.")
            continue
        obs = TOOLS[action](inp)
        history.append(f"[Step {step}]\nThought/Action: {action}\nInput: {inp[:1500]}\nObservation: {obs[:OBS_CHARS]}")

    # 마지막 응답에서 답을 최대한 건져냄
    m = re.findall(r"FINAL ANSWER:\s*(.+)", last_text, flags=re.I)
    fallback = clean_answer(m[-1]) if m else clean_answer(last_text.splitlines()[-1] if last_text.strip() else "unknown")
    return fallback or "unknown", f"{model_used} / 단계초과"


# ---------------------------------------------------------------------------
# 채점 서버
# ---------------------------------------------------------------------------
def fetch_questions() -> list[dict]:
    r = requests.get(f"{SCORING_API}/questions", timeout=30)
    r.raise_for_status()
    data = r.json()
    if not data:
        raise ValueError("질문 목록이 비어 있습니다.")
    return data


def make_df(rows: dict, order: list[str]) -> pd.DataFrame:
    return pd.DataFrame([rows[t] for t in order if t in rows],
                        columns=["Task ID", "Question", "Submitted Answer", "OK", "Note"])


def run_batch(questions: list[dict], label: str):
    """문항을 순서대로 풀며 진행 상황을 계속 내보냅니다. 마지막에 (rows, order, aborted)를 반환."""
    order = [q["task_id"] for q in questions if q.get("task_id")]
    rows: dict = {}
    cache = load_cache()
    start = time.time()
    fails, aborted = 0, False
    log(f"{label} 시작: {len(order)}문항, 모델 제공자={PROVIDER}, 문항당 최대 {MAX_STEPS}단계")

    def status(i: int) -> str:
        return (f"{label}: {len(rows)}/{len(order)} 완료, {i}번째 문항 처리 중 (경과 {int(time.time() - start)}초)\n"
                f"--- 최근 기록 ---\n{recent_log()}")

    for i, item in enumerate(questions, start=1):
        tid = item.get("task_id")
        if not tid or item.get("question") is None:
            continue
        yield status(i), make_df(rows, order)

        if tid in cache:
            answer, ok, note = cache[tid], True, "cache"
            log(f"캐시 사용 {tid[:8]}")
        else:
            box: dict = {}

            def work(it=item):
                try:
                    box["r"] = solve(it)
                except LLMUnavailable as e:
                    box["llm"] = str(e)
                except Exception as e:  # noqa: BLE001
                    box["err"] = f"{type(e).__name__}: {e}"

            th = threading.Thread(target=work, daemon=True)
            th.start()
            while th.is_alive():
                th.join(2)
                yield status(i), make_df(rows, order)

            if "r" in box:
                answer, note = box["r"]
                ok, fails = True, 0
                save_cache(tid, answer)
                log(f"완료 {tid[:8]} -> {answer}")
            elif "llm" in box:
                answer, ok, note = "unknown", False, box["llm"][:250]
                fails += 1
                log(f"LLM 실패 {tid[:8]}: {note}")
            else:
                answer, ok, note = "unknown", False, box.get("err", "알 수 없는 오류")[:250]
                log(f"오류 {tid[:8]}: {note}")

        rows[tid] = {"Task ID": tid, "Question": item["question"], "Submitted Answer": answer,
                     "OK": "O" if ok else "X", "Note": note}
        if fails >= 2:
            aborted = True
            log("LLM이 연속 2회 실패하여 중단합니다. (한도/크레딧 문제 가능성)")
            break

    yield f"{label} 종료 (경과 {int(time.time() - start)}초)\n--- 최근 기록 ---\n{recent_log()}", make_df(rows, order)
    return rows, order, aborted


# ---------------------------------------------------------------------------
# Gradio 콜백
# ---------------------------------------------------------------------------
def run_and_submit_all(profile: gr.OAuthProfile | None):
    space_id = os.getenv("SPACE_ID")
    if not profile:
        yield "먼저 Hugging Face 로그인 버튼으로 로그인해 주세요.", None
        return
    if not space_id:
        yield "SPACE_ID를 찾을 수 없습니다. Space 안에서 실행해야 합니다.", None
        return
    yield "질문을 가져오는 중...", None
    try:
        questions = fetch_questions()
    except Exception as e:  # noqa: BLE001
        yield f"질문 가져오기 오류: {e}", None
        return

    rows, order, aborted = yield from run_batch(questions, "본 실행")
    if aborted:
        yield ("LLM 한도/크레딧 문제로 중단했으므로 제출하지 않았습니다.\n"
               "푼 답은 저장되어 있어 해결 후 다시 누르면 이어서 진행합니다. "
               "(Space가 재시작되면 저장된 답은 사라집니다.)\n\n" + recent_log()), make_df(rows, order)
        return

    payload = [{"task_id": t, "submitted_answer": rows[t]["Submitted Answer"]} for t in order if t in rows]
    failed = sum(1 for r in rows.values() if r["OK"] == "X")
    yield f"풀이 완료 (실패 {failed}개). 제출 중...", make_df(rows, order)
    submission = {"username": profile.username.strip(),
                  "agent_code": f"https://huggingface.co/spaces/{space_id}/tree/main",
                  "answers": payload}
    try:
        r = requests.post(f"{SCORING_API}/submit", json=submission, timeout=120)
        r.raise_for_status()
        res = r.json()
        status = (f"제출 성공!\n사용자: {res.get('username')}\n"
                  f"점수: {res.get('score', 'N/A')}% ({res.get('correct_count', '?')}/{res.get('total_attempted', '?')} 정답)\n"
                  f"메시지: {res.get('message', '')}")
    except requests.exceptions.HTTPError as e:
        status = f"제출 실패: {e.response.text[:500] if e.response is not None else e}"
    except Exception as e:  # noqa: BLE001
        status = f"제출 중 오류: {e}"
    yield status, make_df(rows, order)


def run_test_only(n_questions: int):
    yield "질문을 가져오는 중...", None
    try:
        questions = fetch_questions()[: int(n_questions)]
    except Exception as e:  # noqa: BLE001
        yield f"준비 중 오류: {e}", None
        return
    yield from run_batch(questions, "테스트")


def diagnose():
    lines = [f"PROVIDER={PROVIDER}, MAX_STEPS={MAX_STEPS}, MIN_INTERVAL={MIN_INTERVAL}초, QUESTION_TIMEOUT={QUESTION_TIMEOUT}초"]
    yield "\n".join(lines + ["", "모델 확인 중..."])
    if PROVIDER == "hf":
        lines.append(f"HF_TOKEN: {'있음' if hf_token() else '없음'}, 모델: {HF_MODELS}")
        try:
            r = requests.get(f"{HF_ROUTER}/models", headers={"Authorization": f"Bearer {hf_token()}"}, timeout=30)
            r.raise_for_status()
            ids = [m.get("id") for m in r.json().get("data", [])]
            lines.append("1) 모델 목록: " + ", ".join(f"{m}={'있음' if m in ids else '목록에 없음'}" for m in HF_MODELS))
        except Exception as e:  # noqa: BLE001
            lines.append(f"1) 모델 목록 조회 실패(진행은 계속): {type(e).__name__}: {str(e)[:150]}")
    else:
        lines.append(f"GEMINI_API_KEY: {'있음' if api_key() else '없음'}")
        try:
            lines.append("1) 사용 가능한 Gemini 모델(일부): " + ", ".join([n for n in list_models() if n.startswith('gemini')][:25]))
            lines.append(f"   사용할 후보 순서: {candidate_models()}")
        except Exception as e:  # noqa: BLE001
            lines.append(f"1) 모델 목록 조회 실패: {type(e).__name__}: {str(e)[:200]}")
            yield "\n".join(lines)
            return
    yield "\n".join(lines + ["", "LLM 호출 테스트 중..."])
    try:
        text, model = call_llm([{"text": "Reply with the single word OK."}])
        lines.append(f"2) LLM 호출: 정상 (모델 {model}, 응답 '{text.strip()[:30]}')")
    except LLMUnavailable as e:
        lines.append(f"2) LLM 호출: 실패 - {str(e)[:300]}")
    yield "\n".join(lines + ["", "도구 테스트 중..."])
    s = tool_search("Hugging Face")
    lines.append(f"3) 웹 검색: {'정상' if len(s) > 100 else '실패/결과 부족'} ({len(s)}자)")
    w = tool_wiki("Mercedes Sosa | studio album")
    lines.append(f"4) 위키백과: {'정상' if 'Wikipedia:' in w[:30] else '실패'} ({len(w)}자) {'' if 'Wikipedia:' in w[:30] else w[:100]}")
    p = tool_python("print(sum(range(1, 11)))")
    lines.append(f"5) 파이썬 실행: {'정상' if p.strip() == '55' else '이상 - ' + p[:100]}")
    yield "\n".join(lines)


def clear_cache():
    global _candidates
    _candidates = []
    try:
        os.remove(CACHE_PATH)
        return "캐시와 모델 선택 정보를 초기화했습니다."
    except FileNotFoundError:
        return "삭제할 캐시가 없습니다. (모델 선택 정보는 초기화됨)"


# ---------------------------------------------------------------------------
# UI
# ---------------------------------------------------------------------------
with gr.Blocks() as demo:
    gr.Markdown("# Basic Agent Evaluation Runner (다단계 에이전트 버전)")
    gr.Markdown(
        """**사용 순서**
1. '연결 진단'으로 모델과 도구(검색, 위키, 파이썬)가 정상인지 확인합니다.
2. 로그인 후 '테스트 실행'(1~2문항)으로 확인합니다. (제출 안 함)
3. 문제없으면 '캐시 초기화' 후 'Run Evaluation & Submit All Answers'를 누릅니다."""
    )
    gr.LoginButton()
    with gr.Row():
        diag_button = gr.Button("연결 진단")
        test_n = gr.Number(value=2, label="테스트할 문항 수 (제출 안 함)", precision=0)
        test_button = gr.Button("테스트 실행 (제출 안 함)")
        cache_button = gr.Button("캐시 초기화")
    run_button = gr.Button("Run Evaluation & Submit All Answers")
    status_output = gr.Textbox(label="Run Status / Submission Result", lines=16, interactive=False)
    results_table = gr.DataFrame(label="Questions and Agent Answers", wrap=True)

    diag_button.click(fn=diagnose, outputs=[status_output])
    test_button.click(fn=run_test_only, inputs=[test_n], outputs=[status_output, results_table])
    run_button.click(fn=run_and_submit_all, outputs=[status_output, results_table])
    cache_button.click(fn=clear_cache, outputs=[status_output])

if __name__ == "__main__":
    demo.queue().launch(debug=True, share=False)