Update app.py

import glob
import json
import os
import re
import threading
import time
from collections import deque
from concurrent.futures import ThreadPoolExecutor, wait, FIRST_COMPLETED

# litellm의 재시도 기능에 필요한 tenacity가 없으면 시작 시 자동 설치
try:
import tenacity # noqa: F401
except ImportError:
import subprocess
import sys

subprocess.run([sys.executable, "-m", "pip", "install", "tenacity"], check=False)

import gradio as gr
import pandas as pd
import requests
from smolagents import (
CodeAgent,
DuckDuckGoSearchTool,
InferenceClientModel,
LiteLLMModel,
VisitWebpageTool,
tool,
)

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

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

# ---------------------------------------------------------------------------
# 설정 (Space의 Variables에서 조정 가능)
# ---------------------------------------------------------------------------
DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
FILES_DIR = "/tmp/task_files"
CACHE_PATH = "/tmp/answers_cache.json"
os.makedirs(FILES_DIR, exist_ok=True)

WORKERS = int(os.getenv("WORKERS", "2")) # 동시 실행 문항 수
QUESTION_TIMEOUT = int(os.getenv("QUESTION_TIMEOUT", "300")) # 문항당 제한 시간(초)
MAX_STEPS = int(os.getenv("MAX_STEPS", "8")) # 에이전트 최대 단계
MODEL_TIMEOUT = int(os.getenv("MODEL_TIMEOUT", "90")) # LLM 1회 호출 제한(초)
HEARTBEAT = 3 # 화면 갱신 주기(초)

ANSWER_RULES = """You are a careful research and reasoning agent solving GAIA-style questions.
Rules:
1. Use the tools (web search, webpage visit, file reading, python code) whenever the
question needs facts you are not 100% sure about. Do not guess.
2. The final answer is graded by EXACT MATCH. Return ONLY the answer value, with no
explanation, no units unless asked, no trailing period, no "FINAL ANSWER:" prefix.
3. If the answer is a number: no thousands separators, no currency/percent symbols
unless the question asks for them.
4. If the answer is a list: comma-separated, in the order the question requests.
5. If the answer is a string: no articles (a, an, the) and no abbreviations unless
the question asks for them.
6. If a file path is given, read it with the read_local_file tool or with python code.
7. Be efficient: stop researching as soon as you have a reliable answer.
8. Finish by calling final_answer(...) with the answer only."""

# ---------------------------------------------------------------------------
# 이벤트 로그 (화면 상태창에 최근 기록을 보여주기 위함)
# ---------------------------------------------------------------------------
_events = deque(maxlen=10)
_active: dict = {} # task_id -> 시작 시각
_state_lock = threading.Lock()


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


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


# ---------------------------------------------------------------------------
# 캐시
# ---------------------------------------------------------------------------
_cache_lock = threading.Lock()


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_entry(task_id: str, answer: str):
with _cache_lock:
cache = load_cache()
cache[task_id] = answer
with open(CACHE_PATH, "w", encoding="utf-8") as f:
json.dump(cache, f, ensure_ascii=False)


# ---------------------------------------------------------------------------
# 도구
# ---------------------------------------------------------------------------
@tool
def read_local_file(path: str) -> str:
"""Reads a local file (txt, csv, json, py, md, xlsx) and returns its content as text.

Args:
path: Absolute or relative path of the file to read.
"""
try:
if not os.path.exists(path):
basename = os.path.basename(path)
matches = glob.glob(os.path.join(FILES_DIR, f"*{basename}"))
if matches:
path = matches[0]
lower = path.lower()
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=200)}" for n, df in sheets.items()
)
if lower.endswith(".csv"):
return pd.read_csv(path).to_string(max_rows=200)
with open(path, "r", encoding="utf-8", errors="ignore") as f:
return f.read()[:20000]
except Exception as e: # noqa: BLE001
return f"Could not read file: {e}"


# ---------------------------------------------------------------------------
# 모델 / 에이전트
# ---------------------------------------------------------------------------
def gemini_model_id() -> str:
return os.getenv("GEMINI_MODEL", "gemini-3.8-flash").replace("gemini/", "")


def build_model():
gemini_key = os.getenv("GEMINI_API_KEY")
if gemini_key:
return LiteLLMModel(
model_id=f"gemini/{gemini_model_id()}",
api_key=gemini_key,
timeout=MODEL_TIMEOUT, # LLM 호출이 무한정 멈추지 않도록 제한
num_retries=2,
)
openai_key = os.getenv("OPENAI_API_KEY")
if openai_key:
return LiteLLMModel(
model_id=f"openai/{os.getenv('OPENAI_MODEL', 'gpt-4o-mini')}",
api_key=openai_key,
timeout=MODEL_TIMEOUT,
num_retries=2,
)
return InferenceClientModel(
model_id=os.getenv("HF_MODEL", "Qwen/Qwen2.5-Coder-32B-Instruct"),
token=os.getenv("HF_TOKEN"),
)


def build_agent() -> CodeAgent:
return CodeAgent(
tools=[DuckDuckGoSearchTool(), VisitWebpageTool(), read_local_file],
model=build_model(),
additional_authorized_imports=[
"pandas", "numpy", "re", "json", "math", "datetime",
"collections", "itertools", "statistics", "requests", "bs4",
],
max_steps=MAX_STEPS,
verbosity_level=1,
)


def clean_answer(answer) -> str:
text = str(answer).strip()
text = re.sub(r"^\s*final answer\s*:\s*", "", text, flags=re.IGNORECASE)
return text.strip().rstrip(".").strip()


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


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


# ---------------------------------------------------------------------------
# 문항 풀이 (캐시 + 시간 제한)
# ---------------------------------------------------------------------------
def solve_one(item: dict, api_url: str) -> str:
prompt = ANSWER_RULES + "\n\nQuestion:\n" + item["question"]
file_path = download_task_file(api_url, item["task_id"], item.get("file_name", ""))
if file_path:
prompt += f"\n\nAn attached file is available at: {file_path}"
return clean_answer(build_agent().run(prompt))


def solve_with_timeout(item: dict, api_url: str):
"""(답변, 성공 여부, 비고)를 반환합니다."""
task_id = item["task_id"]
short = task_id[:8]
cached = load_cache().get(task_id)
if cached:
log(f"캐시 사용 {short} -> {cached}")
return cached, True, "cache"

box: dict = {}

def target():
try:
box["answer"] = solve_one(item, api_url)
except Exception as e: # noqa: BLE001
box["error"] = f"{type(e).__name__}: {e}"[:300]

log(f"시작 {short}")
with _state_lock:
_active[task_id] = time.time()
t = threading.Thread(target=target, daemon=True)
t.start()
t.join(QUESTION_TIMEOUT)
with _state_lock:
_active.pop(task_id, None)

if t.is_alive():
log(f"시간 초과 {short} ({QUESTION_TIMEOUT}초)")
return "unknown", False, f"timeout {QUESTION_TIMEOUT}s"
if "error" in box:
log(f"오류 {short}: {box['error']}")
return "unknown", False, box["error"]

save_cache_entry(task_id, box["answer"])
log(f"완료 {short} -> {box['answer']}")
return box["answer"], True, ""


def iter_solve(questions: list[dict], api_url: str):
"""끝난 문항은 (item, answer, ok, note)로, 대기 중에는 주기적으로 None을 내보냅니다."""
valid = [q for q in questions if q.get("task_id") and q.get("question") is not None]
pool = ThreadPoolExecutor(max_workers=WORKERS)
try:
pending = {pool.submit(solve_with_timeout, q, api_url): q for q in valid}
while pending:
done, _ = wait(list(pending), timeout=HEARTBEAT, return_when=FIRST_COMPLETED)
if not done:
yield None
continue
for fut in done:
item = pending.pop(fut)
answer, ok, note = fut.result()
yield item, answer, ok, note
finally:
pool.shutdown(wait=False)


def make_df(rows: dict, order: list[str]) -> pd.DataFrame:
data = [rows[tid] for tid in order if ti

app.py

@@ -7,6 +7,15 @@ import time
 from collections import deque
 from concurrent.futures import ThreadPoolExecutor, wait, FIRST_COMPLETED
 
+# litellm의 재시도 기능에 필요한 tenacity가 없으면 시작 시 자동 설치
+try:
+    import tenacity  # noqa: F401
+except ImportError:
+    import subprocess
+    import sys
+
+    subprocess.run([sys.executable, "-m", "pip", "install", "tenacity"], check=False)
+
 import gradio as gr
 import pandas as pd
 import requests