Create app.py

app.py

@@ -0,0 +1,208 @@
+from llama_index.llms.groq import Groq
+from llama_index.core import PromptTemplate
+from llama_index.experimental.query_engine.pandas import PandasInstructionParser
+from llama_index.core.query_pipeline import (QueryPipeline as QP, Link, InputComponent)
+import gradio as gr
+import pandas as pd
+from fpdf import FPDF
+from datetime import datetime
+import os
+
+# API KEY do GROQ
+api_key = os.getenv("secret_key")
+
+# Configuração inicial do QP
+llm = Groq(model="llama3-70b-8192", api_key=api_key)
+
+# Pipeline de consulta
+def descrição_colunas(df):
+    descrição = '\n'.join([f"`{col}`: {str(df[col].dtype)}" for col in df.columns])
+    return "Aqui estão os detalhes das colunas do dataframe:\n" + descrição
+
+def pipeline_consulta(df):
+    instruction_str = (
+        "1. Converta a consulta para código Python executável usando Pandas.\n"
+        "2. A linha final do código deve ser uma expressão Python que possa ser chamada com a função `eval()`.\n"
+        "3. O código deve representar uma solução para a consulta.\n"
+        "4. IMPRIMA APENAS A EXPRESSÃO.\n"
+        "5. Não coloque a expressão entre aspas.\n")
+
+    pandas_prompt_str = (
+        "Você está trabalhando com um dataframe do pandas em Python chamado `df`.\n"
+        "{colunas_detalhes}\n\n"
+        "Este é o resultado de `print(df.head())`:\n"
+        "{df_str}\n\n"
+        "Siga estas instruções:\n"
+        "{instruction_str}\n"
+        "Consulta: {query_str}\n\n"
+        "Expressão:"
+)
+
+    response_synthesis_prompt_str = (
+       "Dada uma pergunta de entrada, atue como analista de dados e elabore uma resposta a partir dos resultados da consulta.\n"
+       "Responda de forma natural, sem introduções como 'A resposta é:' ou algo semelhante.\n"
+       "Consulta: {query_str}\n\n"
+       "Instruções do Pandas (opcional):\n{pandas_instructions}\n\n"
+       "Saída do Pandas: {pandas_output}\n\n"
+       "Resposta: \n\n"
+       "Ao final, exibir o código usado em para gerar a resposta, no formato: O código utilizado foi `{pandas_instructions}`"
+    )
+
+    pandas_prompt = PromptTemplate(pandas_prompt_str).partial_format(
+    instruction_str=instruction_str,
+    df_str=df.head(5),
+    colunas_detalhes=descrição_colunas(df)
+)
+
+    pandas_output_parser = PandasInstructionParser(df)
+    response_synthesis_prompt = PromptTemplate(response_synthesis_prompt_str)
+
+    # Criação do QueryPipeline
+
+    qp = QP(
+        modules={
+            "input": InputComponent(),
+            "pandas_prompt": pandas_prompt,
+            "llm1": llm,
+            "pandas_output_parser": pandas_output_parser,
+            "response_synthesis_prompt": response_synthesis_prompt,
+            "llm2": llm,
+        },
+        verbose=True,
+    )
+    qp.add_chain(["input", "pandas_prompt", "llm1", "pandas_output_parser"])
+    qp.add_links(
+        [
+            Link("input", "response_synthesis_prompt", dest_key="query_str"),
+            Link("llm1", "response_synthesis_prompt", dest_key="pandas_instructions"),
+            Link("pandas_output_parser", "response_synthesis_prompt", dest_key="pandas_output"),
+        ]
+    )
+    qp.add_link("response_synthesis_prompt", "llm2")
+    return qp
+
+# Função para carregar os dados
+def carregar_dados(caminho_arquivo, df_estado):
+    if caminho_arquivo is None or caminho_arquivo == "":
+        return "Por favor, faça o upload de um arquivo CSV para analisar.", pd.DataFrame(), df_estado
+    try:
+        df = pd.read_csv(caminho_arquivo)
+        return "Arquivo carregado com sucesso!", df.head(), df
+    except Exception as e:
+        return f"Erro ao carregar arquivo: {str(e)}", pd.DataFrame(), df_estado
+
+# Função para processar a pergunta
+def processar_pergunta(pergunta, df_estado):
+    if df_estado is not None and pergunta:
+        qp = pipeline_consulta(df_estado)
+        resposta = qp.run(query_str=pergunta)
+        return resposta.message.content
+    return ""
+
+# Função para adicionar a pergunta e a resposta ao histórico
+def add_historico(pergunta, resposta, historico_estado):
+    if pergunta and resposta:
+        historico_estado.append((pergunta, resposta))
+        gr.Info("Adicionado ao PDF!", duration=2)
+        return historico_estado
+
+# Função para gerar o PDF
+def gerar_pdf(historico_estado):
+    if not historico_estado:
+        return "Nenhum dado para adicionar ao PDF.", None
+
+    # Gerar nome de arquivo com timestamp
+    timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
+    caminho_pdf = f"relatorio_perguntas_respostas_{timestamp}.pdf"
+
+    pdf = FPDF()
+    pdf.add_page()
+    pdf.set_auto_page_break(auto=True, margin=15)
+
+    for pergunta, resposta in historico_estado:
+        pdf.set_font("Arial", 'B', 14)
+        pdf.multi_cell(0, 8, txt=pergunta)
+        pdf.ln(2)
+        pdf.set_font("Arial", '', 12)
+        pdf.multi_cell(0, 8, txt=resposta)
+        pdf.ln(6)
+
+    pdf.output(caminho_pdf)
+    return caminho_pdf
+
+# Função para limpar a pergunta e a resposta
+def limpar_pergunta_resposta():
+    return "", ""
+
+# Função para resetar a aplicação
+def resetar_aplicação():
+    return None, "A aplicação foi resetada. Por favor, faça upload de um novo arquivo CSV.", pd.DataFrame(), "", None, [], ""
+
+# Criação da interface gradio
+with gr.Blocks(theme='Soft') as app:
+
+    # Título da app
+    gr.Markdown("# Analisando os dados🔎🎲")
+
+    # Descrição
+    gr.Markdown('''
+    Carregue um arquivo CSV e faça perguntas sobre os dados. A cada pergunta, você poderá
+    visualizar a resposta e, se desejar, adicionar essa interação ao PDF final, basta clicar
+    em "Adicionar ao histórico do PDF". Para fazer uma nova pergunta, clique em "Limpar pergunta e resultado".
+    Após definir as perguntas e respostas no histórico, clique em "Gerar PDF". Assim, será possível
+    baixar um PDF com o registro completo das suas interações. Se você quiser analisar um novo dataset,
+    basta clicar em "Quero analisar outro dataset" ao final da página.
+    ''')
+
+    # Campo de entrada de arquivos
+    input_arquivo = gr.File(file_count="single", type="filepath", label="Upload CSV")
+
+    # Status de upload
+    upload_status = gr.Textbox(label="Status do Upload:")
+
+    # Tabela de dados
+    tabela_dados = gr.DataFrame()
+
+    # Exemplos de perguntas
+    gr.Markdown("""
+    Exemplos de perguntas:
+    1. Qual é o número de registros no arquivo?
+    2. Quais são os tipos de dados das colunas?
+    3. Quais são as estatísticas descritivas das colunas numéricas?
+    """)
+
+    # Campo de entrada de texto
+    input_pergunta = gr.Textbox(label="Digite sua pergunta sobre os dados")
+
+    # Botão de envio posicionado após a pergunta
+    botao_submeter = gr.Button("Enviar")
+
+    # Componente de resposta
+    output_resposta = gr.Textbox(label="Resposta")
+
+    # Botões para limpar a pergunta e a resposta, adicionar ao historico e gerar o PDF
+    with gr.Row():
+        botao_limpeza = gr.Button("Limpar pergunta e resultado")
+        botao_add_pdf = gr.Button("Adicionar ao histórico do PDF")
+        botao_gerar_pdf = gr.Button("Gerar PDF")
+
+    # Componente de download
+    arquivo_pdf = gr.File(label="Download do PDF")
+
+    # Botão para resetar a aplicação
+    botao_resetar = gr.Button("Quero analisar outro dataset!")
+
+    # Gerenciamento de estados
+    df_estado = gr.State(value=None)
+    historico_estado = gr.State(value=[])
+
+    # Conectando funções aos componentes
+    input_arquivo.change(fn=carregar_dados, inputs=[input_arquivo, df_estado], outputs=[upload_status, tabela_dados, df_estado])
+    botao_submeter.click(fn=processar_pergunta, inputs=[input_pergunta, df_estado], outputs=output_resposta)
+    botao_limpeza.click(fn=limpar_pergunta_resposta, inputs=[], outputs=[input_pergunta, output_resposta])
+    botao_add_pdf.click(fn=add_historico, inputs=[input_pergunta, output_resposta, historico_estado], outputs=historico_estado)
+    botao_gerar_pdf.click(fn=gerar_pdf, inputs=[historico_estado], outputs=arquivo_pdf)
+    botao_resetar.click(fn=resetar_aplicação, inputs=[], outputs=[input_arquivo, upload_status, tabela_dados, output_resposta, arquivo_pdf, historico_estado, input_pergunta])
+
+if __name__ == "__main__":
+    app.launch()