Update app.py

"""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

app.py

@@ -1,13 +1,17 @@
-"""HF Agents Course 최종 과제 - Gemini 또는 Hugging Face 선택형 경량 버전
+"""HF Agents Course 최종 과제 - 다단계 에이전트 버전 (Gemini 또는 Hugging Face)
 
-설계 원칙: 문항당 LLM 호출 1회.
-  웹 검색(호출 없음) -> 검색 결과 + 첨부 파일을 한 번에 Gemini에 전달 -> 최종 답만 받음
+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
 
@@ -38,40 +42,51 @@ except ImportError:
 # ---------------------------------------------------------------------------
 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.json"
+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 "2"))  # LLM 호출 간 최소 간격(초)
-HF_ROUTER = "https://router.huggingface.co/v1"
+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()]
-MAX_RETRY = int(os.getenv("MAX_RETRY", "3"))             # 429 재시도 횟수
-FETCH_PAGES = int(os.getenv("FETCH_PAGES", "2"))         # 본문을 읽을 검색 결과 수
-PAGE_CHARS = int(os.getenv("PAGE_CHARS", "5000"))        # 페이지당 최대 글자 수
-LLM_QUERY = os.getenv("LLM_QUERY", "0") == "1"           # 1이면 검색어도 Gemini로 생성(호출 2배)
-GEMINI_MODELS = os.getenv("GEMINI_MODELS", "")           # 선택: 쉼표로 구분한 모델 이름 목록
+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>
 
-PROMPT = """You are answering a GAIA-style benchmark question. The answer is graded by EXACT MATCH.
+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:
-- Use the provided web context and attached file. If the context is insufficient, use your own knowledge and reason carefully.
-- Give ONLY the answer value: no explanation, no trailing period, no units unless asked.
+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), no abbreviations unless asked.
-- End your reply with one final line exactly in the form:
-FINAL ANSWER: <answer>
+- Strings: no leading articles (a, an, the) and no abbreviations unless asked.
+- Follow any formatting instruction inside the question exactly.
 
-Question:
-{question}
-
-{extra}"""
+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=12)
+_events = deque(maxlen=14)
 
 
 def log(msg: str):
@@ -84,9 +99,6 @@ def recent_log() -> str:
     return "\n".join(_events)
 
 
-# ---------------------------------------------------------------------------
-# 캐시
-# ---------------------------------------------------------------------------
 def load_cache() -> dict:
     try:
         with open(CACHE_PATH, "r", encoding="utf-8") as f:
@@ -103,29 +115,36 @@ def save_cache(task_id: str, answer: str):
 
 
 # ---------------------------------------------------------------------------
-# Gemini REST 호출 (라이브러리 없이 requests만 사용)
+# 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 = requests.get(f"{GEMINI_API}/models", headers={"x-goog-api-key": api_key()}, params={"pageSize": 200}, timeout=30)
     r.raise_for_status()
-    names = []
-    for m in r.json().get("models", []):
-        if "generateContent" in m.get("supportedGenerationMethods", []):
-            names.append(m["name"].replace("models/", ""))
-    return names
+    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")
@@ -133,7 +152,6 @@ _candidates: list[str] = []
 
 
 def candidate_models() -> list[str]:
-    """GEMINI_MODELS가 있으면 그 목록, 없으면 키로 쓸 수 있는 모델에서 자동 선택."""
     global _candidates
     if _candidates:
         return _candidates
@@ -144,23 +162,11 @@ def candidate_models() -> list[str]:
     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"자동 선택된 모델 후보: {_candidates}")
+    log(f"자동 선택된 Gemini 모델 후보: {_candidates}")
     return _candidates
 
 
-_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 call_gemini(parts: list[dict]) -> tuple[str, str]:
-    """(응답 텍스트, 사용한 모델)을 반환. 모든 후보가 실패하면 LLMUnavailable."""
     if not api_key():
         raise LLMUnavailable("GEMINI_API_KEY가 설정되지 않았습니다.")
     last_err = ""
@@ -171,33 +177,24 @@ def call_gemini(parts: list[dict]) -> tuple[str, str]:
                 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},
-                    },
+                    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")
-                )
+                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:
-                # ��일 한도 소진 또는 한도 0이면 재시도해도 소용없으므로 다음 모델로
                 if "PerDay" in body or "limit: 0" in body:
                     log(f"{model}: 일일 한도 소진 또는 한도 0 -> 다음 모델")
                     break
@@ -211,25 +208,16 @@ def call_gemini(parts: list[dict]) -> tuple[str, str]:
             if r.status_code >= 500:
                 time.sleep(5)
                 continue
-            break  # 400, 403, 404 등은 이 모델 포기
+            break
     raise LLMUnavailable(last_err or "사용 가능한 모델이 없습니다.")
 
 
-# ---------------------------------------------------------------------------
-# Hugging Face Inference Providers (OpenAI 호환 라우터)
-# ---------------------------------------------------------------------------
-def hf_token() -> str:
-    return os.getenv("HF_TOKEN", "")
-
-
 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)
+        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):
@@ -239,7 +227,7 @@ def call_hf(parts: list[dict]) -> tuple[str, str]:
                     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": 1024, "temperature": 0},
+                          "max_tokens": 1500, "temperature": 0},
                     timeout=120,
                 )
             except requests.RequestException as e:
@@ -256,9 +244,9 @@ def call_hf(parts: list[dict]) -> tuple[str, str]:
             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])
+                raise LLMUnavailable("HF 크레딧 소진(402). " + body[:150])
             if r.status_code in (401, 403):
-                raise LLMUnavailable(f"HF_TOKEN 인증/권한 오류({r.status_code}). 토큰의 Inference Providers 권한을 확인하세요.")
+                raise LLMUnavailable(f"HF_TOKEN 인증/권한 오류({r.status_code}). Inference Providers 권한을 확인하세요.")
             if r.status_code == 429:
                 if attempt == MAX_RETRY:
                     break
@@ -269,7 +257,7 @@ def call_hf(parts: list[dict]) -> tuple[str, str]:
             if r.status_code >= 500:
                 time.sleep(5)
                 continue
-            break  # 400, 404 등은 이 모델 포기
+            break
     raise LLMUnavailable(last_err or "사용 가능한 모델이 없습니다.")
 
 
@@ -278,39 +266,148 @@ def call_llm(parts: list[dict]) -> tuple[str, str]:
 
 
 # ---------------------------------------------------------------------------
-# 웹 검색 + 페이지 읽기 (Gemini 호출 없음)
+# 도구 구현
 # ---------------------------------------------------------------------------
 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 = re.sub(r"&nbsp;|&#160;", " ", text)
-    return re.sub(r"\s+", " ", text).strip()
+    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 web_context(query: str) -> str:
+def tool_search(query: str) -> str:
     if DDGS is None:
-        return ""
+        return "검색 도구를 사용할 수 없습니다."
     try:
         results = list(DDGS().text(query, max_results=6))
     except Exception as e:  # noqa: BLE001
-        log(f"검색 실패: {type(e).__name__}")
-        return ""
-    chunks = []
-    for i, res in enumerate(results, start=1):
-        chunks.append(f"[{i}] {res.get('title', '')}\n{res.get('href', '')}\n{res.get('body', '')}")
-    for res in results[:FETCH_PAGES]:
-        url = res.get("href")
-        if not url or url.lower().endswith((".pdf", ".zip", ".mp4")):
-            continue
+        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:
-            page = requests.get(url, timeout=15, headers={"User-Agent": "Mozilla/5.0"})
-            if page.ok and "text/html" in page.headers.get("content-type", ""):
-                chunks.append(f"--- PAGE: {url} ---\n{html_to_text(page.text)[:PAGE_CHARS]}")
-        except requests.RequestException:
-            continue
-    return "\n\n".join(chunks)
+            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}
+
 # ---------------------------------------------------------------------------
 # 첨부 파일
 # ---------------------------------------------------------------------------
@@ -330,67 +427,88 @@ def download_file(task_id: str, file_name: str):
 
 
 def file_to_parts(path: str) -> tuple[str, list[dict]]:
-    """(프롬프트에 붙일 텍스트, Gemini에 직접 넣을 inline_data 파트들)"""
+    """(미리보기 텍스트, 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 = mimetypes.guess_type(path)[0] or "application/octet-stream"
-            if lower.endswith(".mp3"):
-                mime = "audio/mp3"
+                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 "(첨부 파일은 함께 전달됨)", [{"inline_data": {"mime_type": mime, "data": data}}]
+            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}\n{df.to_string(max_rows=300)}" for n, df in sheets.items())[:30000], []
+            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"):
-            return pd.read_csv(path).to_string(max_rows=300)[:30000], []
+            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()[:30000], []
+            return f.read()[:12000], []
     except Exception as e:  # noqa: BLE001
         return f"(파일 읽기 실패: {e})", []
 
 
 # ---------------------------------------------------------------------------
-# 한 문항 풀이
+# 에이전트 루프
 # ---------------------------------------------------------------------------
-def parse_final(text: str) -> str:
-    m = re.findall(r"FINAL ANSWER:\s*(.+)", text, flags=re.IGNORECASE)
-    ans = m[-1] if m else (text.strip().splitlines() or [""])[-1]
-    return ans.strip().strip("`").strip().rstrip(".").strip()
+def clean_answer(ans: str) -> str:
+    ans = (ans.strip().splitlines() or [""])[0]
+    return ans.strip().strip("`").strip().strip('"').strip().rstrip(".").strip()
 
 
-def make_search_query(question: str) -> str:
-    if not LLM_QUERY:
-        return question[:300]
-    try:
-        text, _ = call_llm([{"text": "Write ONE concise web search query (max 12 words) that would help answer "
-                                         "this question. Reply with the query only.\n\n" + question}])
-        return text.strip().splitlines()[0][:200]
-    except LLMUnavailable:
-        return question[:300]
+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] = []
-    extra_text = ""
-
+    file_note = ""
     path = download_file(item["task_id"], item.get("file_name", ""))
     if path:
-        file_text, extra_parts = file_to_parts(path)
-        extra_text += f"ATTACHED FILE ({os.path.basename(path)}):\n{file_text}\n\n"
+        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")
 
-    ctx = web_context(make_search_query(question))
-    if ctx:
-        extra_text += f"WEB CONTEXT (may be incomplete or irrelevant):\n{ctx}\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]}")
 
-    prompt = PROMPT.format(question=question, extra=extra_text[:60000])
-    text, model = call_llm([{"text": prompt}] + extra_parts)
-    return parse_final(text), model
+    # 마지막 응답에서 답을 최대한 건져냄
+    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} / 단계초과"
 
 
 # ---------------------------------------------------------------------------
@@ -406,49 +524,64 @@ def fetch_questions() -> list[dict]:
 
 
 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"],
-    )
+    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)를 반환."""
+    """문항을 순서대로 풀며 진행 상황을 계속 내보냅니다. 마지막에 (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 = 0
-    aborted = False
-    log(f"{label} 시작: {len(order)}문항 (문항당 LLM 호출 {2 if LLM_QUERY else 1}회)")
+    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
-        head = f"{label}: {i - 1}/{len(order)} 완료, {i}번째 문항 처리 중 (경과 {int(time.time() - start)}초)"
-        yield f"{head}\n--- 최근 기록 ---\n{recent_log()}", make_df(rows, order)
+        yield status(i), make_df(rows, order)
 
         if tid in cache:
             answer, ok, note = cache[tid], True, "cache"
             log(f"캐시 사용 {tid[:8]}")
         else:
-            try:
-                answer, model = solve(item)
+            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)
-                ok, note = True, model
-                fails = 0
                 log(f"완료 {tid[:8]} -> {answer}")
-            except LLMUnavailable as e:
-                answer, ok, note = "unknown", False, str(e)[:250]
+            elif "llm" in box:
+                answer, ok, note = "unknown", False, box["llm"][:250]
                 fails += 1
                 log(f"LLM 실패 {tid[:8]}: {note}")
-            except Exception as e:  # noqa: BLE001
-                answer, ok, note = "unknown", False, f"{type(e).__name__}: {e}"[:250]
+            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}
+        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회 실패하여 중단합니다. (한도/크레딧 문제 가능성)")
@@ -479,18 +612,16 @@ def run_and_submit_all(profile: gr.OAuthProfile | None):
     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,
-    }
+    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()
@@ -516,23 +647,21 @@ def run_test_only(n_questions: int):
 
 
 def diagnose():
-    lines = [f"PROVIDER={PROVIDER}, MIN_INTERVAL={MIN_INTERVAL}초, LLM_QUERY={LLM_QUERY}"]
+    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}")
+        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", [])]
-            check = ", ".join(f"{m}={'목록에 있음' if m in ids else '목록에 없음'}" for m in HF_MODELS)
-            lines.append(f"1) 라우터 제공 모델 {len(ids)}개 / {check}")
+            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:
-            names = [n for n in list_models() if n.startswith("gemini")][:25]
-            lines.append("1) 이 키로 사용 가능한 모델(일부): " + ", ".join(names))
+            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]}")
@@ -544,12 +673,13 @@ def diagnose():
         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 + ["", "웹 검색 테스트 중..."])
-    try:
-        ctx = web_context("Hugging Face")
-        lines.append(f"3) 웹 검색: {'정상' if ctx else '결과 없음'} ({len(ctx)}자)")
-    except Exception as e:  # noqa: BLE001
-        lines.append(f"3) 웹 검색: 실패 - {e}")
+    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)
 
 
@@ -567,12 +697,12 @@ def clear_cache():
 # UI
 # ---------------------------------------------------------------------------
 with gr.Blocks() as demo:
-    gr.Markdown("# Basic Agent Evaluation Runner (Gemini / Hugging Face 선택형 경량 버전)")
+    gr.Markdown("# Basic Agent Evaluation Runner (다단계 에이전트 버전)")
     gr.Markdown(
         """**사용 순서**
-1. '연결 진단'으로 모델과 호출 가능 여부를 확인합니다.
+1. '연결 진단'으로 모델과 도구(검색, 위키, 파이썬)가 정상인지 확인합니다.
 2. 로그인 후 '테스트 실행'(1~2문항)으로 확인합니다. (제출 안 함)
-3. 문제없으면 'Run Evaluation & Submit All Answers'를 누릅니다."""
+3. 문제없으면 '캐시 초기화' 후 'Run Evaluation & Submit All Answers'를 누릅니다."""
     )
     gr.LoginButton()
     with gr.Row():
@@ -581,7 +711,7 @@ with gr.Blocks() as demo:
         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=14, interactive=False)
+    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])