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()