educ.ai_Qgemini / gemini.py
GalvaoFilho's picture
Upload 2 files
022a8aa verified
# app_refatorado.py
import os
import json
import logging
import gradio as gr
from dotenv import load_dotenv
from crewai import Agent, Task, Crew
# --- CONFIGURAÇÃO INICIAL ---
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
load_dotenv()
# --- CONSTANTES DE CONFIGURAÇÃO DOS AGENTES E TAREFAS ---
# Backstories dos Agentes
REFINADOR_BACKSTORY = "Você é um curador de conteúdo experiente que sabe como decompor assuntos complexos."
GERADOR_BACKSTORY = (
"Você é um sistema automatizado ultra eficiente. Sua única função é gerar listas de objetos JSON puras. "
"Você NUNCA escreve texto explicativo, comentários, ou qualquer palavra que não faça parte do JSON final. "
"Sua saída SEMPRE começa com '[' e termina com ']'."
)
AVALIADOR_BACKSTORY = "Você é um especialista em avaliação que fornece resumos úteis para ajudar o aluno a aprender."
# Descrição da Tarefa de Geração de Questões
GERAR_QUESTOES_DESC = """
Gere {num_questoes} questões de múltipla escolha sobre: '{topico}'.
**REGRAS DE FORMATAÇÃO DA SAÍDA (MUITO IMPORTANTE):**
- A sua resposta final deve ser APENAS o código JSON.
- NÃO inclua NENHUM texto, explicação, comentário, ou markdown (como ```json) antes ou depois do JSON.
- A sua saída DEVE começar com o caractere `[` e terminar com o caractere `]`.
**REGRAS DO CONTEÚDO DO JSON:**
1. A saída deve ser uma lista de objetos JSON.
2. Cada objeto DEVE ter as chaves: 'id', 'questao', 'opcoes', e 'resposta_correta'.
3. A chave 'opcoes' DEVE conter uma lista de 5 strings de texto.
4. As strings dentro de 'opcoes' NUNCA DEVEM estar vazias.
5. O valor de 'resposta_correta' DEVE ser a letra da opção correta (a, b, c, d, ou e).
"""
# --- INICIALIZAÇÃO DOS AGENTES ---
refinador_de_topico = Agent(
role="Especialista em Estruturação de Conteúdo Educacional",
goal="Dado um tópico amplo, sugerir 5 sub-tópicos específicos e bem definidos, ideais para a criação de um quiz.",
backstory=REFINADOR_BACKSTORY,
verbose=False,
allow_delegation=False,
)
gerador_questoes = Agent(
role="Gerador de Questões JSON",
goal="Criar um conjunto de questões de múltipla escolha com base em um tópico específico e um número definido.",
backstory=GERADOR_BACKSTORY,
verbose=False,
allow_delegation=False,
)
avaliador_desempenho = Agent(
role="Avaliador de Desempenho",
goal="Avaliar as respostas do aluno comparando-as com o gabarito original e fornecer um feedback rápido e encorajador.",
backstory=AVALIADOR_BACKSTORY,
verbose=False,
allow_delegation=False,
)
# --- FUNÇÕES AUXILIARES ---
def formatar_pergunta_para_exibicao(item_questao: dict, indice: int) -> str:
"""Formata uma única questão para ser exibida de forma legível no chatbot."""
texto_pergunta = f"**Questão {indice + 1}:** {item_questao['questao']}\n\n"
if 'opcoes' in item_questao and isinstance(item_questao['opcoes'], list):
letras = ['a', 'b', 'c', 'd', 'e']
for i, opcao in enumerate(item_questao['opcoes']):
if i < len(letras):
texto_pergunta += f" {letras[i]}) {opcao}\n"
return texto_pergunta
def _executar_crew(crew: Crew) -> str | None:
"""Executa uma crew e trata possíveis exceções da API."""
try:
resultado = crew.kickoff()
logging.info(f"Resultado bruto da Crew: {resultado.raw}")
return resultado.raw
except Exception as e:
logging.error(f"Ocorreu um erro ao executar a crew: {e}")
return None
# --- LÓGICA PRINCIPAL DO QUIZ (DIVIDIDA POR FASE) ---
def handle_inicio_quiz(topico_geral: str, historico_chat: list, estado_quiz: dict) -> tuple[dict, str]:
"""Lida com a primeira mensagem do usuário, refinando o tópico."""
historico_chat.append({"role": "assistant", "content": f"Excelente tópico! Buscando sub-tópicos dentro de **{topico_geral}**..."})
tarefa_refinar = Task(
description=f"Sugira 5 sub-tópicos específicos sobre '{topico_geral}' para um quiz. Apresente-os como uma lista numerada simples (ex: 1. Tópico A).",
expected_output="Uma lista de 5 sub-tópicos, cada um em uma nova linha.",
agent=refinador_de_topico
)
crew_refinamento = Crew(agents=[refinador_de_topico], tasks=[tarefa_refinar])
sugestoes_raw = _executar_crew(crew_refinamento)
if sugestoes_raw:
sugestoes_limpas = [s.split('.', 1)[-1].strip() for s in sugestoes_raw.split('\n') if s.strip()]
mensagem_para_aluno = f"Aqui estão algumas sugestões. **Escolha uma** (digitando o número ou o texto) ou defina um tópico ainda mais específico:\n\n{sugestoes_raw}"
historico_chat.append({"role": "assistant", "content": mensagem_para_aluno})
estado_quiz.update({
"fase": "gerando_questoes",
"topico_geral": topico_geral,
"sugestoes": sugestoes_limpas
})
else:
historico_chat.append({"role": "assistant", "content": "Desculpe, não consegui gerar sugestões. Tente outro tópico."})
return estado_quiz, historico_chat
def handle_geracao_questoes(mensagem_usuario: str, num_questoes: int, historico_chat: list, estado_quiz: dict) -> tuple[dict, str]:
"""Lida com a escolha do sub-tópico e gera as questões do quiz."""
try:
escolha_num = int(mensagem_usuario.strip())
topico_real = estado_quiz["sugestoes"][escolha_num - 1]
except (ValueError, IndexError):
topico_real = mensagem_usuario
historico_chat.append({"role": "assistant", "content": f"Ótimo! Preparando {int(num_questoes)} questões sobre **{topico_real}**. Aguarde um momento..."})
tarefa_gerar = Task(
description=GERAR_QUESTOES_DESC.format(num_questoes=int(num_questoes), topico=topico_real),
expected_output="Uma string de JSON puro, começando com '[' e terminando com ']'.",
agent=gerador_questoes
)
crew_geracao = Crew(agents=[gerador_questoes], tasks=[tarefa_gerar])
resultado_raw = _executar_crew(crew_geracao)
if not resultado_raw:
historico_chat.append({"role": "assistant", "content": "Desculpe, ocorreu um erro ao gerar as questões. Tente novamente."})
return {}, historico_chat # Reseta o estado
try:
inicio_json = resultado_raw.find('[')
fim_json = resultado_raw.rfind(']')
if inicio_json != -1 and fim_json != -1:
json_str_limpo = resultado_raw[inicio_json : fim_json + 1]
questoes = json.loads(json_str_limpo)
estado_quiz.update({
"fase": "respondendo",
"questoes": questoes,
"respostas_aluno": [],
"indice_atual": 0
})
primeira_pergunta = formatar_pergunta_para_exibicao(questoes[0], 0)
historico_chat.append({"role": "assistant", "content": primeira_pergunta})
else:
raise ValueError("Nenhum JSON válido encontrado na saída do LLM.")
except (json.JSONDecodeError, ValueError, IndexError) as e:
logging.error(f"Erro ao processar a saída do LLM: {e}\nSaída recebida: {resultado_raw}")
historico_chat.append({"role": "assistant", "content": "Desculpe, a resposta do assistente não veio no formato esperado. Tente novamente."})
estado_quiz = {} # Reseta o estado
return estado_quiz, historico_chat
def handle_resposta_quiz(mensagem_usuario: str, historico_chat: list, estado_quiz: dict) -> tuple[dict, str]:
"""Lida com a resposta do usuário a uma questão do quiz."""
indice_atual = estado_quiz["indice_atual"]
estado_quiz["respostas_aluno"].append({
"id_questao": estado_quiz["questoes"][indice_atual]["id"],
"resposta_aluno": mensagem_usuario.lower().strip()
})
novo_indice = indice_atual + 1
estado_quiz["indice_atual"] = novo_indice
if novo_indice < len(estado_quiz["questoes"]):
proxima_pergunta = formatar_pergunta_para_exibicao(estado_quiz["questoes"][novo_indice], novo_indice)
historico_chat.append({"role": "assistant", "content": proxima_pergunta})
else:
historico_chat.append({"role": "assistant", "content": "Quiz finalizado! Preparando seu feedback..."})
tarefa_avaliar = Task(
description=f"Avalie o desempenho do aluno com base nos dados JSON a seguir:\nQuestões Originais com Gabarito: {json.dumps(estado_quiz['questoes'])}\nRespostas do Aluno: {json.dumps(estado_quiz['respostas_aluno'])}\nForneça um relatório de feedback em português, curto e motivador, formatado em markdown.",
expected_output="Um relatório de feedback em texto (markdown).", agent=avaliador_desempenho
)
crew_avaliacao = Crew(agents=[avaliador_desempenho], tasks=[tarefa_avaliar])
feedback_final_raw = _executar_crew(crew_avaliacao)
feedback = feedback_final_raw or "Não foi possível gerar o feedback final."
historico_chat.append({"role": "assistant", "content": feedback})
estado_quiz = {} # Fim do quiz, reseta o estado
return estado_quiz, historico_chat
# --- FUNÇÃO PRINCIPAL DE INTERAÇÃO (ROTEADOR) ---
def processar_interacao(mensagem_usuario: str, num_questoes: int, historico_chat: list, estado_quiz: dict) -> tuple[str, list, dict]:
"""
Função central que gerencia o fluxo do quiz, agindo como um roteador.
"""
estado_quiz = estado_quiz or {"fase": "inicio"}
historico_chat.append({"role": "user", "content": mensagem_usuario})
fase_atual = estado_quiz.get("fase", "inicio")
if fase_atual == "inicio":
estado_quiz, historico_chat = handle_inicio_quiz(mensagem_usuario, historico_chat, estado_quiz)
elif fase_atual == "gerando_questoes":
estado_quiz, historico_chat = handle_geracao_questoes(mensagem_usuario, num_questoes, historico_chat, estado_quiz)
elif fase_atual == "respondendo":
estado_quiz, historico_chat = handle_resposta_quiz(mensagem_usuario, historico_chat, estado_quiz)
return "", historico_chat, estado_quiz
# --- INTERFACE GRÁFICA COM GRADIO ---
with gr.Blocks(theme=gr.themes.Soft(primary_hue="green")) as demo:
gr.Markdown("# 🧠 Quiz Interativo com CrewAI & Gemini API")
gr.Markdown("Escolha o número de questões e digite um tópico para começar. O assistente irá ajudá-lo a refinar o assunto.")
estado_quiz = gr.State(value={})
chatbot = gr.Chatbot(
elem_id="chatbot",
height=600,
# CORREÇÃO: Usando o caminho do arquivo local em vez da URL
avatar_images=(None, "gemini_avatar.png"),
placeholder="Aguardando o tópico do quiz...",
type="messages"
)
with gr.Row():
num_questoes_slider = gr.Slider(minimum=2, maximum=10, value=3, step=1, label="Número de Questões", scale=1)
txt_msg = gr.Textbox(label="Sua Mensagem", placeholder="Digite o tópico aqui para começar...", scale=3)
txt_msg.submit(
fn=processar_interacao,
inputs=[txt_msg, num_questoes_slider, chatbot, estado_quiz],
outputs=[txt_msg, chatbot, estado_quiz]
)
if __name__ == "__main__":
demo.launch(debug=True)