Spaces:

agents-course
/

Final_Assignment_Template

Running

App Files Files Community

633

Final_Assignment_Template / app.py

omega21

Update app.py

4267a75 verified 24 days ago

Raw History Blame

10.1 kB

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)

	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"^\sfinal 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)