import os import gradio as gr import requests import json import pandas as pd from dotenv import load_dotenv from agent import AgentDeedee # Desabilita a autenticação OAuth para testes locais os.environ["GRADIO_ANALYTICS_ENABLED"] = "False" # Carrega variáveis de ambiente load_dotenv() # --- Constantes --- DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" OUTPUT_JSONL = "answers.jsonl" def carregar_prompt_sistema(): """Carrega o prompt do sistema a partir do arquivo""" try: with open("system_prompt.txt", "r", encoding="utf-8") as f: return f.read() except Exception as e: print(f"Erro ao carregar prompt do sistema: {e}") return "Você é a Deedee, um assistente especializado em responder perguntas." def salvar_jsonl(respostas_payload): """Salva as respostas no formato JSONL exatamente como esperado pelo sistema de avaliação""" with open(OUTPUT_JSONL, 'w', encoding='utf-8') as f: for resposta in respostas_payload: # Remove o prefixo "FINAL ANSWER: " se existir model_answer = resposta.get("model_answer", "") if model_answer.startswith("FINAL ANSWER: "): model_answer = model_answer[14:] # Remove os primeiros 14 caracteres # Cria uma entrada no formato correto entrada = { "task_id": resposta["task_id"], "model_answer": model_answer # Opcionalmente: "reasoning_trace": "Processo de raciocínio para esta resposta" } json.dump(entrada, f, ensure_ascii=False) f.write('\n') print(f"Respostas salvas em {OUTPUT_JSONL}") def processar_pergunta(pergunta, task_id): """ Processa uma pergunta usando o agente Deedee e retorna a resposta """ try: # Inicializa o agente agente = AgentDeedee() # Processa a pergunta resposta = agente(pergunta) # Remove o prefixo "FINAL ANSWER: " se existir if resposta.startswith("FINAL ANSWER: "): resposta = resposta[14:] return { "task_id": task_id, "model_answer": resposta } except Exception as e: print(f"Erro ao processar pergunta {task_id}: {e}") return { "task_id": task_id, "model_answer": "Erro ao processar a pergunta" } def executar_enviar_tudo(profile: gr.OAuthProfile | None): """ Carrega as respostas do arquivo answers.jsonl e as envia para avaliação. """ if not profile: print("Usuário não está logado.") return "Por favor, faça login no Hugging Face com o botão.", None nome_usuario = profile.username.strip() print(f"Usuário logado: {nome_usuario}") # Obtém o SPACE_ID para enviar o link para o código space_id = os.getenv("SPACE_ID", f"{nome_usuario}/Final_Assignment_Template") codigo_agente = f"https://huggingface.co/spaces/{space_id}/tree/main" print(f"Usando URL do código do agente: {codigo_agente}") api_url = DEFAULT_API_URL url_envio = f"{api_url}/submit" # Carrega as respostas do arquivo answers.jsonl try: respostas_payload = [] registro_resultados = [] with open(OUTPUT_JSONL, 'r', encoding='utf-8') as f: for line in f: if line.strip(): entry = json.loads(line.strip()) # Garantir que não há prefixo "FINAL ANSWER: " na resposta model_answer = entry.get("model_answer", "") if model_answer.startswith("FINAL ANSWER: "): model_answer = model_answer[14:] entry["model_answer"] = model_answer respostas_payload.append(entry) registro_resultados.append({ "ID da Tarefa": entry["task_id"], "Resposta Enviada": entry["model_answer"] }) print(f"Carregadas {len(respostas_payload)} respostas do arquivo {OUTPUT_JSONL}") if not respostas_payload: return f"Erro: O arquivo {OUTPUT_JSONL} está vazio ou não contém respostas válidas.", None except Exception as e: print(f"Erro ao carregar respostas do arquivo {OUTPUT_JSONL}: {e}") return f"Erro ao carregar respostas: {e}", None # Envia Respostas try: dados_envio = { "username": nome_usuario, "agent_code": codigo_agente, "answers": respostas_payload } print(f"Enviando {len(respostas_payload)} respostas...") resposta = requests.post(url_envio, json=dados_envio, timeout=120) resposta.raise_for_status() resultado = resposta.json() status = f"Envio bem-sucedido!\nUsuário: {nome_usuario}\nPontuação: {resultado.get('score', 'N/A')}%" return status, pd.DataFrame(registro_resultados) except Exception as e: msg_erro = f"Falha no envio: {str(e)}" if hasattr(e, 'response') and hasattr(e.response, 'text'): msg_erro += f"\nResposta: {e.response.text}" print(msg_erro) return msg_erro, pd.DataFrame(registro_resultados) # --- Constroi Interface Gradio usando Blocks --- with gr.Blocks() as demo: gr.Markdown("# Deedee - Agente de Avaliação") gr.Markdown( """ **Instruções:** 1. Configure sua chave API do Together AI no arquivo .env 2. Clique em 'Testar Agente' para testar o agente com uma pergunta de exemplo 3. Para envio completo, faça login com sua conta Hugging Face e clique em 'Executar Avaliação' """ ) # Interface para teste local with gr.Tab("Teste Local"): pergunta_teste = gr.Textbox(label="Pergunta de teste", lines=2, placeholder="Digite uma pergunta para testar o agente...") resposta_teste = gr.Textbox(label="Resposta do agente", lines=5, interactive=False) botao_teste = gr.Button("Testar Agente") # Função para testar o agente localmente def testar_agente(pergunta): try: agente = AgentDeedee() resposta = agente(pergunta) return resposta except Exception as e: return f"Erro ao processar pergunta: {str(e)}" botao_teste.click( fn=testar_agente, inputs=pergunta_teste, outputs=resposta_teste ) # Interface para envio completo (apenas para ambiente Hugging Face Space) with gr.Tab("Envio para Avaliação"): gr.Markdown(""" **Nota:** Esta funcionalidade está disponível apenas quando executada em um ambiente Hugging Face Space. Para enviar respostas para avaliação: 1. Faça o upload deste código para um Space no Hugging Face 2. Configure as variáveis de ambiente necessárias 3. Execute o aplicativo no ambiente do Space """) # Botão desabilitado para ambiente local gr.Button("Executar Avaliação (Desabilitado em ambiente local)", interactive=False) if __name__ == "__main__": demo.launch()