from smolagents import CodeAgent, DuckDuckGoSearchTool, VisitWebpageTool, InferenceClientModel, load_tool, tool import datetime import os import requests import pytz import pandas as pd import gradio as gr import spaces from tools.final_answer import FinalAnswerTool # Esse Space está configurado com hardware ZeroGPU, que exige pelo menos uma # função decorada com @spaces.GPU detectável na inicialização — mesmo que o # agente em si não use GPU (ele só chama a API de inferência remota). Essa # função "dummy" só existe pra satisfazer essa checagem de startup. @spaces.GPU def _warmup_zerogpu(): return "ok" # --------------------------------------------------------------------------- # Suas ferramentas (mantidas do seu app.py original) # --------------------------------------------------------------------------- @tool def my_custom_tool(arg1: str, arg2: int) -> str: """A tool that does nothing yet Args: arg1: the first argument arg2: the second argument """ return "What magic will you build ?" @tool def get_current_time_in_timezone(timezone: str) -> str: """A tool that fetches the current local time in a specified timezone. Args: timezone: A string representing a valid timezone (e.g., 'America/New_York'). """ try: tz = pytz.timezone(timezone) local_time = datetime.datetime.now(tz).strftime("%Y-%m-%d %H:%M:%S") return f"The current local time in {timezone} is: {local_time}" except Exception as e: return f"Error fetching time for timezone '{timezone}': {str(e)}" final_answer = FinalAnswerTool() # Modelo: mantive o seu. Se estiver "overloaded" com frequência nas perguntas # do GAIA (que costumam exigir mais raciocínio), considere trocar por algo # como 'Qwen/Qwen2.5-72B-Instruct' ou 'meta-llama/Llama-3.3-70B-Instruct'. model = InferenceClientModel( max_tokens=2096, temperature=0.5, model_id='Qwen/Qwen2.5-Coder-32B-Instruct', token=os.environ.get("HF_TOKEN"), custom_role_conversions=None, ) # Tool de imagem: não é útil para responder perguntas do GAIA (que exige # respostas em texto/número), então deixei comentada para não gastar tempo # de carregamento à toa. Descomente se quiser usá-la em outro contexto. # image_generation_tool = load_tool("agents-course/text-to-image", trust_remote_code=True) # Nota: o prompts.yaml original do template não é compatível com esta # versão mais nova do smolagents (falta a seção 'final_answer'). Por isso # não carregamos mais um prompt_templates customizado — o CodeAgent usa # os templates padrão da própria lib, que já funcionam bem. agent = CodeAgent( model=model, tools=[ final_answer, DuckDuckGoSearchTool(), VisitWebpageTool(), get_current_time_in_timezone, my_custom_tool, ], max_steps=8, verbosity_level=1, planning_interval=None, name=None, description=None, ) GAIA_SYSTEM_SUFFIX = ( "\n\nVocê está sendo avaliado no benchmark GAIA. Pense passo a passo e " "use as ferramentas quando precisar de informação externa ou cálculo. " "Sua resposta final (a que você passar para o final_answer) deve ser " "um número, ou a string mais curta possível, ou uma lista separada por " "vírgulas, conforme a pergunta pedir — sem unidades, artigos ou " "explicações extras, a menos que peçam explicitamente." ) class GaiaAgentWrapper: """Adapta o seu `agent` (CodeAgent) para a interface esperada pelo fluxo de submissão do Unit 4: __call__(question) -> str""" def __init__(self, api_url: str | None = None): self.api_url = api_url def __call__(self, question: str, task: dict | None = None) -> str: prompt = question + GAIA_SYSTEM_SUFFIX file_path = None if task is not None and task.get("file_name") and self.api_url: file_path = self._maybe_download_task_file(task) if file_path: prompt += ( f"\n\nHá um arquivo anexado a esta tarefa, salvo em: " f"{file_path}\nAbra/analise-o com código Python se for " f"relevante para responder." ) try: result = agent.run(prompt) except Exception as e: print(f"Erro ao rodar o agente: {e}") return "ERRO" finally: if file_path and os.path.exists(file_path): os.remove(file_path) return str(result).strip() def _maybe_download_task_file(self, task: dict) -> str | None: import tempfile task_id = task["task_id"] try: resp = requests.get(f"{self.api_url}/files/{task_id}", timeout=30) resp.raise_for_status() except Exception as e: print(f"Falha ao baixar arquivo do task {task_id}: {e}") return None suffix = os.path.splitext(task.get("file_name", ""))[1] tmp = tempfile.NamedTemporaryFile(delete=False, suffix=suffix) tmp.write(resp.content) tmp.close() return tmp.name # --------------------------------------------------------------------------- # Fluxo de submissão do Unit 4 (busca perguntas -> roda agente -> submete) # --------------------------------------------------------------------------- DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" def run_and_submit_all(profile: gr.OAuthProfile | None): space_id = os.getenv("SPACE_ID") if profile: username = profile.username print(f"Usuário logado: {username}") else: return "Por favor, faça login no Hugging Face com o botão acima.", None api_url = DEFAULT_API_URL questions_url = f"{api_url}/questions" submit_url = f"{api_url}/submit" gaia_agent = GaiaAgentWrapper(api_url=api_url) agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main" print(f"Agent code: {agent_code}") try: response = requests.get(questions_url, timeout=15) response.raise_for_status() questions_data = response.json() if not questions_data: return "Nenhuma pergunta recebida da API.", None print(f"{len(questions_data)} perguntas recebidas.") except Exception as e: return f"Erro ao buscar perguntas: {e}", None results_log = [] answers_payload = [] for item in questions_data: task_id = item.get("task_id") question_text = item.get("question") if not task_id or question_text is None: continue try: submitted_answer = gaia_agent(question_text, task=item) except Exception as e: submitted_answer = f"AGENT ERROR: {e}" answers_payload.append({"task_id": task_id, "submitted_answer": submitted_answer}) results_log.append({ "Task ID": task_id, "Question": question_text, "Submitted Answer": submitted_answer, }) if not answers_payload: return "O agente não produziu nenhuma resposta a ser submetida.", pd.DataFrame(results_log) submission_data = { "username": username.strip(), "agent_code": agent_code, "answers": answers_payload, } print(f"Enviando {len(answers_payload)} respostas para: {submit_url}") try: response = requests.post(submit_url, json=submission_data, timeout=60) response.raise_for_status() result_data = response.json() final_status = ( f"Submissão concluída!\n" f"Usuário: {result_data.get('username')}\n" f"Pontuação: {result_data.get('score', 'N/A')}% " f"({result_data.get('correct_count', '?')}/{result_data.get('total_attempted', '?')} corretas)\n" f"Mensagem: {result_data.get('message', '')}" ) return final_status, pd.DataFrame(results_log) except requests.exceptions.HTTPError as e: detail = e.response.text[:500] return f"Falha na submissão: {e.response.status_code} - {detail}", pd.DataFrame(results_log) except Exception as e: return f"Falha na submissão: {e}", pd.DataFrame(results_log) # --------------------------------------------------------------------------- # Interface: aba de chat (seu agente original) + aba de submissão do GAIA # --------------------------------------------------------------------------- with gr.Blocks() as demo: gr.Markdown("# Meu Agente — GAIA / Unit 4") gr.Markdown( "Faça login, clique em rodar, e o agente vai responder todas as " "perguntas do GAIA e submeter automaticamente." ) gr.LoginButton() run_button = gr.Button("Rodar avaliação e submeter respostas") status_output = gr.Textbox(label="Status da submissão", lines=5, interactive=False) results_table = gr.DataFrame(label="Perguntas e respostas", wrap=True) run_button.click(fn=run_and_submit_all, outputs=[status_output, results_table]) if __name__ == "__main__": print("Iniciando o Space...") try: _warmup_zerogpu() except Exception as e: print(f"Aviso: warmup do ZeroGPU falhou (não é crítico): {e}") demo.launch(debug=True, share=False)