Spaces:
Sleeping
Sleeping
File size: 4,457 Bytes
6852d71 7f29224 31fe207 4fa0927 31fe207 4fa0927 31fe207 6852d71 4fa0927 6852d71 4fa0927 6852d71 4fa0927 6852d71 4fa0927 6852d71 4fa0927 6852d71 7f29224 6852d71 7f29224 6852d71 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 |
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) |