Update app.py

app.py

@@ -1,3 +1,4 @@
+import os
 import gradio as gr
 from langchain_community.document_loaders import PyPDFLoader
 from langchain.text_splitter import RecursiveCharacterTextSplitter
@@ -5,60 +6,135 @@ 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"  # Modelo de linguagem da Hugging Face
-
-# Função para carregar e processar PDFs
-def load_and_process_pdf(pdf_path):
-    # Carrega o PDF
-    loader = PyPDFLoader(pdf_path)
-    documents = loader.load()
-
-    # Divide o texto em chunks
-    text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
-    texts = text_splitter.split_documents(documents)
+LLM_REPO_ID = "google/flan-t5-large"
 
-    # Cria embeddings e armazena no vetor store
-    embeddings = HuggingFaceEmbeddings(model_name=EMBEDDING_MODEL)
-    db = FAISS.from_documents(texts, embeddings)
+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
 
-    return db
+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
 
-# Função para gerar respostas usando RAG
-def generate_response(pdf_file, query):
+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."
-
-    # Carrega e processa o PDF
-    db = load_and_process_pdf(pdf_file)
-
-    # Configura o modelo de linguagem
-    llm = HuggingFaceHub(repo_id=LLM_REPO_ID, model_kwargs={"temperature": 0.7, "max_length": 512})
-
-    # 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
-    )
-
-    # Executa a consulta
-    result = qa_chain({"query": query})
-    return result["result"]
+    
+    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
-iface = gr.Interface(
-    fn=generate_response,
-    inputs=[
-        gr.File(label="Upload PDF", type="filepath"),  # Corrigido para "filepath"
-        gr.Textbox(label="Sua Pergunta")
-    ],
-    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."
-)
+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?"]
+        ]
+    )
 
-iface.launch(share=True)
+if __name__ == "__main__":
+    iface = create_interface()
+    iface.launch(share=True)