Importante

Você está vendo a versão anterior da nova experiência da Alura que estamos preparando para você. Em breve, ela ganha uma identidade visual novinha totalmente pensada em potencializar seus estudos!

2
respostas

[Projeto] Faça como eu fiz: arquivos e pandas

import os
from google.colab import userdata
import google.generativeai as genai
from google.api_core.exceptions import TooManyRequests
import time
import pandas as pd

os.environ['GOOGLE_API_KEY'] = userdata.get('GEMINI_API_KEY')

model = genai.GenerativeModel('gemini-2.5-flash')
chat = model.start_chat(history=[])

# Lista para armazenar os resumos dos e-mails
resumo_emails = []

#agente que resume mensagens de e-mail com exponential backoff
def resume_mensagens_emails(mensagens):
  print("Resumo das mensagens recebidas:")
  max_retries = 5
  initial_delay = 5 # seconds

  for i, mensagem in enumerate(mensagens):
    retries = 0
    while retries < max_retries:
      try:
        resposta = chat.send_message(f"resuma essa mensagem mencionando em poucas palavras o que ela quis dizer: {mensagem}")
        print(f"Mensagem {i+1}: {resposta.text}")
        resumo_emails.append(resposta.text)
        break # Sai do loop de retry se a chamada for bem-sucedida
      except TooManyRequests as e:
        retries += 1
        wait_time = initial_delay * (2 ** (retries - 1)) # Exponential backoff
        print(f"\nERRO: Limite de requisições excedido. Tentando novamente em {wait_time} segundos. (Mensagem {i+1} de {len(mensagens)})\n")
        print(f"Detalhes do erro: {e}")
        time.sleep(wait_time)
      except Exception as e:
        print(f"\nERRO inesperado ao processar a mensagem '{mensagem}'. Pulando para a próxima. Detalhes: {e}\n")
        break # Sai do loop de retry para outros erros
    else:
      print(f"Não foi possível processar a mensagem '{mensagem}' após {max_retries} tentativas.\n")

mensagens_para_resumir = [
    "Olá, gostaria de confirmar o recebimento da documentação enviada ontem.",
    "Bom dia, segue em anexo o relatório solicitado para análise.",
    "A reunião foi reagendada para sexta-feira às 14h.",
    "Obrigado pelo contato. Retornaremos com mais informações em breve.",
    "Informamos que seu cadastro foi atualizado com sucesso.",
    "Estamos entrando em contato para lembrar do vencimento da sua fatura.",
    "Seu pedido foi processado e será enviado nos próximos dias.",
    "Parabéns! Sua inscrição foi aprovada com sucesso.",
    "Gostaríamos de agendar uma demonstração do produto para sua equipe.",
    "Segue a proposta comercial conforme solicitado."
]

resumo = resume_mensagens_emails(mensagens_para_resumir)

# Implemente o with open com modo 'W' para criar e sobrescrever um arquivo .txt com os resumos.
# Use write ou write lines com '\n' para separar cada entrada no arquivo.
with open("resumos.txt", "w", encoding="utf-8") as arquivo:
  for resumo in resumo_emails:
    arquivo.write(resumo + '\n')
print("Resumos dos emails salvos em 'resumos.txt'.")

# Abra o arquivo em modo 'R' e leia suas linhas, salvando em uma nova lista com strip para remover quebras de linha.
# Explore os métodos read, readline e readlines para diferentes formas de leitura em Python.
resumos_lidos = []
with open("resumos.txt", "r", encoding="utf-8") as arquivo:
  for linha in arquivo.readlines():
    resumos_lidos.append(linha.strip())

print("\nConteúdo de 'resumos.txt' lido:")
for index, resumo in enumerate(resumos_lidos):
  print(f"Resumo {index+1}: {resumo}")

# Importe o Pandas utilizando o alias pd para manipulação de dados.
# (Já importado no início do bloco)

# Construa um arquivo .csv manualmente, escrevendo o cabeçalho e as linhas com as chaves de cada dicionário.
# Este exemplo demonstra a criação manual para um único registro.
dados_manuais = {
    'id': '001',
    'nome': 'Alice',
    'idade': 30,
    'cidade': 'São Paulo'
}
nome_arquivo_manual = 'dados_manuais_criados.csv'

with open(nome_arquivo_manual, 'w', encoding='utf-8') as arquivo:
    header = ';'.join(dados_manuais.keys())
    arquivo.write(header + '\n')
    values = ';'.join(str(value) for value in dados_manuais.values())
    arquivo.write(values + '\n')
print(f"\nArquivo '{nome_arquivo_manual}' criado com sucesso!")

# Salvando com o pandas (para múltiplos registros)
# Crie um DataFrame com a lista de dicionários e utilize o método to_csv para salvar o CSV com encoding UTF-8.
lista_de_dicionarios = [
    {'nome': 'Alice', 'idade': 25, 'cidade': 'São Paulo'},
    {'nome': 'Bob', 'idade': 30, 'cidade': 'Nova York'},
    {'nome': 'Charlie', 'idade': 35, 'cidade': 'Londres'}
]
df_pandas = pd.DataFrame(lista_de_dicionarios)
df_pandas.to_csv('dados_pandas.csv', index=False, sep=';', encoding='utf-8')
print(f"Arquivo 'dados_pandas.csv' criado com sucesso usando pandas!")

continua no comentário...

2 respostas

continuando...

# Desenvolva um desafio para gerar um arquivo .txt a partir de uma lista de perguntas e depois lê-lo para criar uma lista.
perguntas_portugues_original = [
    'De que é feito o Sol?',
    'De que é feito o planeta Saturno?',
    'Qual é a galáxia mais antiga já encontrada?',
    'Qual é a maior estrela já encontrada?',
    'Qual é a estrela mais próxima do Sol?'
]

arquivo_txt_perguntas = "perguntas_para_llm.txt"
with open(arquivo_txt_perguntas, "w", encoding="utf-8") as arquivo:
  for pergunta in perguntas_portugues_original:
    arquivo.write(pergunta + '\n')
print(f"\nArquivo '{arquivo_txt_perguntas}' criado com sucesso!")

# Ler as perguntas desse arquivo
perguntas_lidas = []
with open(arquivo_txt_perguntas, "r", encoding="utf-8") as arquivo:
  for linha in arquivo.readlines():
    perguntas_lidas.append(linha.strip())
print("Perguntas lidas do arquivo:")
for index, pergunta in enumerate(perguntas_lidas):
  print(f"Pergunta {index+1}: {pergunta}")


# Integre uma chamada ao LLM para obter respostas sucintas, associando cada pergunta com sua resposta e salvando o resultado final em um novo arquivo .csv.

# Cliente Groq já configurado no início do bloco

import os
from google.colab import userdata
os.environ['GROQ_API_KEY'] = userdata.get('GROQ_API_KEY')

from groq import Groq
client = Groq()

#função para responder às perguntas usando a LLM gpt-oss-20B com exponential backoff
def responder_suscintamente_lista_de_perguntas(perguntas):
  print("\nGerando respostas para as perguntas:")
  respostas_com_perguntas = []
  max_retries = 5
  initial_delay = 5 # seconds

  for i, pergunta in enumerate(perguntas):
    retries = 0
    while retries < max_retries:
      try:
        completion = client.chat.completions.create(
          model="openai/gpt-oss-20b",
          messages=[
            {
              "role": "user",
              "content": "Responda de modo suscinto a seguinte pergunta: " + pergunta
            }
          ],
          temperature=0.5,
          max_completion_tokens=200, # Ajustado para respostas sucintas, com um pouco mais de margem
          top_p=1,
          reasoning_effort="medium",
          stream=False, # Para capturar a resposta completa de uma vez
          stop=None
        )
        resposta_llm = completion.choices[0].message.content.strip()
        print(f"Pergunta: {pergunta}\nResposta: {resposta_llm}\n")
        respostas_com_perguntas.append({'Pergunta': pergunta, 'Resposta': resposta_llm})
        break # Sai do loop de retry se a chamada for bem-sucedida
      except TooManyRequests as e:
        retries += 1
        wait_time = initial_delay * (2 ** (retries - 1)) # Exponential backoff
        print(f"\nERRO: Limite de requisições excedido. Tentando novamente em {wait_time} segundos. (Pergunta {i+1} de {len(perguntas)})\nDetalhes do erro: {e}\n")
        time.sleep(wait_time)
      except Exception as e:
        print(f"\nERRO inesperado ao processar a pergunta '{pergunta}'. Pulando para a próxima. Detalhes: {e}\n")
        break # Sai do loop de retry para outros erros
    else:
      print(f"Não foi possível processar a pergunta '{pergunta}' após {max_retries} tentativas.\n")
  return respostas_com_perguntas

# Chamar a função e salvar os resultados
respostas_do_llm = responder_suscintamente_lista_de_perguntas(perguntas_lidas)

# Converter a lista de dicionários para DataFrame e salvar em CSV
df_respostas_suscintas = pd.DataFrame(respostas_do_llm)
df_respostas_suscintas.to_csv('respostas_suscintas.csv', index=False, sep=';', encoding='utf-8')
print("Respostas salvas com sucesso em 'respostas_suscintas.csv'!")

# Lendo o csv para testar
respostas_suscintas_lidas = pd.read_csv("respostas_suscintas.csv", sep=';')
print("\nConteúdo de 'respostas_suscintas.csv' lido:")
print(respostas_suscintas_lidas.head())

Olá, Paulo. Como vai?

Uau, que implementação espetacular! O seu projeto foi muito além de uma manipulação de arquivos básica, integrando técnicas avançadas de engenharia de software e inteligência artificial que são extremamente valorizadas no mercado de dados.

A forma como você utilizou os blocos with open com os encodings corretos e gerenciou a leitura com readlines() e .strip() está impecável. Mas o verdadeiro destaque do seu código é a implementação do algoritmo de Exponential Backoff (recuo exponencial) nos loops de requisição. Lidar com limites de taxa (Rate Limits) e erros TooManyRequests através do cálculo initial_delay * (2 (retries - 1)) é uma excelente prática ao trabalhar com APIs de modelos de linguagem como o Gemini e o Groq, pois evita que o seu script quebre no meio de automações em lote.

Para agregar ainda mais valor ao seu código e trazer insights sobre manipulação de DataFrames e otimização, separei duas sugestões bem legais:

1. Cuidados com a cláusula else no laço while

No seu código, você utilizou uma estrutura muito interessante e pouco conhecida do Python: o while ... else.

while retries < max_retries:
    try:
        # Lógica da API
        break
    except TooManyRequests as e:
        # Tratamento
    except Exception as e:
        break
else:
    print(f"Não foi possível processar...")

Essa estrutura funciona perfeitamente no seu caso: o bloco else só é executado se o loop while chegar ao fim naturalmente sem atingir nenhum break. No entanto, note que se uma exceção genérica cair no except Exception as e, o seu código executará o break. Esse break vai fazer o programa sair do loop e não passará pelo else.

Se a sua intenção era avisar que a mensagem falhou também nos erros inesperados, uma boa prática é colocar o print de falha diretamente dentro do except Exception antes do break, ou usar uma variável de controle (uma flag booleana) para deixar o fluxo mais explícito e fácil de ler para outros desenvolvedores.


2. Otimizando a escrita manual de CSV com o módulo csv

No trecho onde você gerou o arquivo CSV manualmente a partir do dicionário dados_manuais, você utilizou o método ';'.join(). Funcionou perfeitamente porque os dados eram simples!

Contudo, se algum campo de texto futuramente contiver um ponto e vírgula por acidente (ex: 'cidade': 'São Paulo; Zona Sul'), o seu CSV manual vai quebrar, pois ele vai criar uma coluna extra inesperada.

Para escritas manuais onde você não queira carregar o Pandas inteiro, a biblioteca nativa csv do Python resolve isso automaticamente, adicionando aspas nos textos que possuem caracteres de divisão. Veja como usar o csv.DictWriter:

import csv

dados_manuais = {
    'id': '001',
    'nome': 'Alice',
    'idade': 30,
    'cidade': 'São Paulo; Zona Sul' # Exemplo com caractere especial
}

nome_arquivo_manual = 'dados_manuais_criados.csv'

with open(nome_arquivo_manual, 'w', encoding='utf-8', newline='') as arquivo:
    # O DictWriter mapeia as chaves do dicionário automaticamente
    escritor = csv.DictWriter(arquivo, fieldnames=dados_manuais.keys(), delimiter=';')
    
    escritor.writeheader() # Escreve o cabeçalho
    escritor.writerow(dados_manuais) # Escreve a linha de dados

Isso blinda o seu código contra falhas de formatação de strings na geração de arquivos.

O seu projeto mostra que você tem uma excelente base tanto em Python estruturado quanto no consumo de LLMs em ambiente de produção. Parabéns pelo empenho e por compartilhar um exemplo tão rico com a comunidade!

Espero que possa ter lhe ajudado!