import spaces import os import re import time import gradio as gr import pandas as pd import requests from smolagents import ( CodeAgent, DuckDuckGoSearchTool, InferenceClientModel, LiteLLMModel, VisitWebpageTool, tool, ) # --------------------------------------------------------------------------- # 상수 # --------------------------------------------------------------------------- DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" FILES_DIR = "/tmp/task_files" os.makedirs(FILES_DIR, exist_ok=True) SLEEP_BETWEEN_QUESTIONS = float(os.getenv("SLEEP_BETWEEN_QUESTIONS", "2")) 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. Finish by calling final_answer(...) with the answer only. """ # --------------------------------------------------------------------------- # 도구 # --------------------------------------------------------------------------- @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 path of the file to read. """ try: lower = path.lower() if lower.endswith((".xlsx", ".xls")): sheets = pd.read_excel(path, sheet_name=None) parts = [] for name, df in sheets.items(): parts.append(f"## Sheet: {name}\n{df.to_string(max_rows=200)}") return "\n\n".join(parts) 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: return f"Could not read file: {e}" # --------------------------------------------------------------------------- # 에이전트 # --------------------------------------------------------------------------- def build_model(): gemini_key = os.getenv("GEMINI_API_KEY") if gemini_key: model_id = os.getenv("GEMINI_MODEL", "gemini-2.5-flash") print(f"[model] Gemini: {model_id}") return LiteLLMModel(model_id=f"gemini/{model_id}", api_key=gemini_key) openai_key = os.getenv("OPENAI_API_KEY") if openai_key: model_id = os.getenv("OPENAI_MODEL", "gpt-4o-mini") print(f"[model] OpenAI: {model_id}") return LiteLLMModel(model_id=f"openai/{model_id}", api_key=openai_key) model_id = os.getenv("HF_MODEL", "Qwen/Qwen2.5-Coder-32B-Instruct") print(f"[model] HF Inference: {model_id}") return InferenceClientModel(model_id=model_id, token=os.getenv("HF_TOKEN")) class BasicAgent: def __init__(self): print("BasicAgent 초기화 중...") self.agent = 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=int(os.getenv("MAX_STEPS", "12")), verbosity_level=1, ) @staticmethod def _clean(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 __call__(self, question: str, file_path: str | None = None) -> str: prompt = ANSWER_RULES + "\n\nQuestion:\n" + question if file_path: prompt += f"\n\nAn attached file is available at: {file_path}" result = self.agent.run(prompt) return self._clean(result) # --------------------------------------------------------------------------- # 채점 서버 통신 유틸 # --------------------------------------------------------------------------- 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) -> str | None: 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: print(f"파일 다운로드 실패 ({task_id}): {e}") return None def run_questions(agent: BasicAgent, questions: list[dict], api_url: str): answers_payload, results_log = [], [] for i, item in enumerate(questions, start=1): task_id = item.get("task_id") question_text = item.get("question") if not task_id or question_text is None: continue print(f"\n===== [{i}/{len(questions)}] task_id={task_id} =====") file_path = download_task_file(api_url, task_id, item.get("file_name", "")) try: answer = agent(question_text, file_path) except Exception as e: print(f"에이전트 오류: {e}") answer = f"AGENT ERROR: {e}" answers_payload.append({"task_id": task_id, "submitted_answer": answer}) results_log.append( {"Task ID": task_id, "Question": question_text, "Submitted Answer": answer} ) time.sleep(SLEEP_BETWEEN_QUESTIONS) return answers_payload, results_log # --------------------------------------------------------------------------- # Gradio 콜백 # --------------------------------------------------------------------------- def run_and_submit_all(profile: gr.OAuthProfile | None): space_id = os.getenv("SPACE_ID") if not profile: return "먼저 Hugging Face 로그인 버튼으로 로그인해 주세요.", None username = profile.username api_url = DEFAULT_API_URL if not space_id: return "SPACE_ID를 찾을 수 없습니다. Space 안에서 실행해야 합니다.", None agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main" try: agent = BasicAgent() except Exception as e: return f"에이전트 초기화 오류: {e}", None try: questions = fetch_questions(api_url) except Exception as e: return f"질문 가져오기 오류: {e}", None answers_payload, results_log = run_questions(agent, questions, api_url) results_df = pd.DataFrame(results_log) if not answers_payload: return "제출할 답변이 없습니다.", results_df submission = { "username": username.strip(), "agent_code": agent_code, "answers": answers_payload, } try: resp = requests.post(f"{api_url}/submit", json=submission, timeout=120) resp.raise_for_status() r = resp.json() status = ( "제출 성공!\n" f"사용자: {r.get('username')}\n" f"점수: {r.get('score', 'N/A')}% " f"({r.get('correct_count', '?')}/{r.get('total_attempted', '?')} 정답)\n" f"메시지: {r.get('message', '')}" ) return status, results_df except requests.exceptions.HTTPError as e: detail = e.response.text[:500] if e.response is not None else str(e) return f"제출 실패: {detail}", results_df except Exception as e: return f"제출 중 오류: {e}", results_df def run_test_only(n_questions: int): api_url = DEFAULT_API_URL try: agent = BasicAgent() questions = fetch_questions(api_url)[: int(n_questions)] except Exception as e: return f"준비 중 오류: {e}", None _, results_log = run_questions(agent, questions, api_url) return f"테스트 완료: {len(results_log)}개 질문 실행 (제출하지 않음)", pd.DataFrame(results_log) # --------------------------------------------------------------------------- # UI # --------------------------------------------------------------------------- with gr.Blocks() as demo: gr.Markdown("# Basic Agent Evaluation Runner") gr.Markdown( """ **Instructions:** 1. 이 Space를 복제(Duplicate)한 뒤 코드를 수정해 에이전트 로직을 정의합니다. 2. 아래 버튼으로 Hugging Face에 로그인합니다. (HF 사용자명으로 제출됩니다) 3. 먼저 '테스트 실행'으로 소수 문항을 확인하고, 문제없으면 'Run Evaluation & Submit All Answers'를 누릅니다. """ ) gr.LoginButton() with gr.Row(): test_n = gr.Number(value=2, label="테스트할 문항 수 (제출 안 함)", precision=0) test_button = gr.Button("테스트 실행 (제출 안 함)") run_button = gr.Button("Run Evaluation & Submit All Answers") status_output = gr.Textbox(label="Run Status / Submission Result", lines=6, interactive=False) results_table = gr.DataFrame(label="Questions and Agent Answers", wrap=True) 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]) @spaces.GPU def dummy(): return "ok" if __name__ == "__main__": demo.launch(debug=True, share=False)