app.py

import os
import gradio as gr
from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import FAISS
from langchain.chains import RetrievalQA
from langchain_community.llms import HuggingFaceHub
from typing import Optional
import tempfile

# Configuração do token da Hugging Face (deve ser definido como variável de ambiente)
HUGGINGFACE_API_TOKEN = os.getenv("HUGGINGFACE_API_TOKEN")
if not HUGGINGFACE_API_TOKEN:
    raise ValueError("Por favor, defina a variável de ambiente HUGGINGFACE_API_TOKEN")

# Configurações
EMBEDDING_MODEL = "sentence-transformers/all-mpnet-base-v2"
LLM_REPO_ID = "google/flan-t5-large"

def create_temporary_file(file_obj) -> str:
    """Cria um arquivo temporário a partir do objeto de arquivo do Gradio."""
    temp_dir = tempfile.mkdtemp()
    temp_path = os.path.join(temp_dir, "temp.pdf")
    
    with open(temp_path, 'wb') as f:
        f.write(file_obj.read())
    
    return temp_path

def load_and_process_pdf(pdf_path: str) -> Optional[FAISS]:
    """
    Carrega e processa o PDF, com tratamento de erros adequado.
    """
    try:
        # Carrega o PDF
        loader = PyPDFLoader(pdf_path)
        documents = loader.load()
        
        if not documents:
            return None
            
        # Divide o texto em chunks
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200,
            length_function=len
        )
        texts = text_splitter.split_documents(documents)
        
        # Cria embeddings e armazena no vetor store
        embeddings = HuggingFaceEmbeddings(
            model_name=EMBEDDING_MODEL,
            model_kwargs={'device': 'cpu'}
        )
        db = FAISS.from_documents(texts, embeddings)
        
        return db
        
    except Exception as e:
        print(f"Erro ao processar o PDF: {str(e)}")
        return None

def generate_response(pdf_file: Optional[tempfile._TemporaryFileWrapper], query: str) -> str:
    """
    Gera resposta para a consulta usando RAG, com tratamento de erros.
    """
    if pdf_file is None:
        return "Erro: Nenhum arquivo PDF foi carregado."
    
    if not query.strip():
        return "Erro: Por favor, insira uma pergunta."
    
    try:
        # Cria arquivo temporário e processa o PDF
        temp_path = create_temporary_file(pdf_file)
        db = load_and_process_pdf(temp_path)
        
        if db is None:
            return "Erro: Não foi possível processar o PDF."
        
        # Configura o modelo de linguagem
        llm = HuggingFaceHub(
            repo_id=LLM_REPO_ID,
            huggingfacehub_api_token=HUGGINGFACE_API_TOKEN,
            model_kwargs={
                "temperature": 0.7,
                "max_length": 512,
                "top_p": 0.95
            }
        )
        
        # Cria a cadeia de RAG
        qa_chain = RetrievalQA.from_chain_type(
            llm=llm,
            chain_type="stuff",
            retriever=db.as_retriever(search_kwargs={"k": 3}),
            return_source_documents=True,
            verbose=True
        )
        
        # Executa a consulta
        result = qa_chain({"query": query})
        
        # Limpa arquivos temporários
        os.remove(temp_path)
        
        return result["result"]
        
    except Exception as e:
        return f"Erro ao gerar resposta: {str(e)}"

# Interface Gradio
def create_interface():
    """Cria e configura a interface Gradio."""
    return gr.Interface(
        fn=generate_response,
        inputs=[
            gr.File(
                label="Upload PDF",
                type="binary",  # Alterado para binary
                file_types=[".pdf"]
            ),
            gr.Textbox(
                label="Sua Pergunta",
                placeholder="Digite sua pergunta aqui..."
            )
        ],
        outputs=gr.Textbox(label="Resposta Gerada"),
        title="Sistema de RAG com LangChain",
        description="Faça upload de um PDF e faça perguntas sobre o conteúdo.",
        examples=[
            [None, "Qual é o principal tema deste documento?"],
            [None, "Pode resumir os pontos principais?"]
        ]
    )

if __name__ == "__main__":
    iface = create_interface()
    iface.launch(share=True)