import os import yaml import gradio as gr # Importe o Gradio from langchain_huggingface import ChatHuggingFace from langchain_huggingface.llms.huggingface_endpoint import HuggingFaceEndpoint from langchain_community.vectorstores import FAISS from langchain.chains import RetrievalQA from langchain_huggingface.embeddings import HuggingFaceEmbeddings from langchain.prompts import PromptTemplate from langchain_community.document_loaders import WebBaseLoader from langchain_text_splitters import CharacterTextSplitter from langchain_community.vectorstores import FAISS from langchain_huggingface import HuggingFaceEmbeddings from dotenv import load_dotenv import os import logging logging.getLogger("langchain.text_splitter").setLevel(logging.ERROR) import warnings warnings.filterwarnings("ignore") from langchain_community.document_loaders import RecursiveUrlLoader import yaml # #----------- local paths ----------- CACHE_FOLDER = "./model/" LOCAL_FOLDER = "./model/" VS_BASE = "./vector_store/vs_base" os.makedirs(CACHE_FOLDER, exist_ok=True) os.makedirs(LOCAL_FOLDER, exist_ok=True) os.makedirs(VS_BASE, exist_ok=True) # --- CONFIGURAÇÕES DE MODELOS --- # LLM_MODEL = 'google/gemma-3-4b-it' LLM_MODEL = 'google/gemma-3-12b-it' EMBEDDING_MODEL = "sentence-transformers/all-MiniLM-L6-v2" # ------------ criando vs ----------------- ## knowledge base offline url_list = [ "https://www.infinitepay.io", "https://www.infinitepay.io/maquininha", "https://www.infinitepay.io/maquininha-celular", "https://www.infinitepay.io/tap-to-pay", "https://www.infinitepay.io/pdv", "https://www.infinitepay.io/receba-na-hora", "https://www.infinitepay.io/gestao-de-cobranca", "https://www.infinitepay.io/gestao-de-cobranca-2", "https://www.infinitepay.io/link-de-pagamento", "https://www.infinitepay.io/loja-online", "https://www.infinitepay.io/boleto", "https://www.infinitepay.io/conta-digital", "https://www.infinitepay.io/conta-pj", "https://www.infinitepay.io/pix", "https://www.infinitepay.io/pix-parcelado", "https://www.infinitepay.io/emprestimo", "https://www.infinitepay.io/cartao", "https://www.infinitepay.io/rendimento", 'https://www.infinitepay.io/taxas', 'https://www.cloudwalk.io/', 'https://www.cloudwalk.io/#our-mission', 'https://www.cloudwalk.io/#our-pillars', 'https://www.cloudwalk.io/#our-products', ] # Carregue o conteúdo da página web como documentos LangChain loader = WebBaseLoader(web_paths=url_list) docs = loader.load() #print(f"Total de páginas carregadas: {len(docs)}") text_splitter = CharacterTextSplitter(chunk_size=1500, chunk_overlap=100) split_docs = text_splitter.split_documents(docs) embeddings = HuggingFaceEmbeddings(model_name=EMBEDDING_MODEL) vector_store = FAISS.from_documents(split_docs, embeddings) os.makedirs(VS_BASE, exist_ok=True) vector_store.save_local(VS_BASE) print(f"vs_base salva em {VS_BASE}") # --- CONFIGURAÇÃO HF --- HF_TOKEN = os.getenv("HF_TOKEN") # Use os.getenv para ler a variável de ambiente if not HF_TOKEN: raise ValueError("HF_TOKEN não encontrado. Por favor, defina-o nos segredos do Hugging Face Space.") os.environ["HF_HUB_USER_AGENT"] = "CloudWalk_Chatbot" # --- 1. Inicializa o LLM Hugging Face --- llm = HuggingFaceEndpoint( repo_id=LLM_MODEL, task="text-generation", max_new_tokens=1024, do_sample=False, repetition_penalty=1.03, huggingfacehub_api_token=HF_TOKEN, ) chat_model = ChatHuggingFace(llm=llm) # --- 2. Inicializa os embeddings --- embeddings = HuggingFaceEmbeddings(model_name=EMBEDDING_MODEL) # --- 3. Carregando a vector store FAISS salva localmente --- vector_store_path = './vector_store/vs_base/' try: faiss_store = FAISS.load_local(vector_store_path, embeddings, allow_dangerous_deserialization=True) except Exception as e: print(f"Erro ao carregar a vector store FAISS: {e}") print("Verifique se o caminho está correto e se o arquivo não está corrompido.") exit() # --- 4. Retriever a partir da vector store (usando similarity como exemplo) --- retriever = faiss_store.as_retriever(search_type="similarity", search_kwargs={"k": 10}) # retriever = faiss_store.as_retriever( # search_type="mmr", # search_kwargs={"k": 5, "fetch_k": 20, "lambda_mult": 0.7} # ) # --- 5. PromptTemplate personalizado --- custom_prompt_template = """ You are a useful chatbot for customer service. If you don't know the answer, just say that you don't know, don't try to make up an answer. ALWAYS RESPONDE IN THE SAME LANGUAGE AS THE INPUT. Use the following pieces of context to answer the user's question. {context} Question: {question} Helpful Answer:""" # Crie um objeto PromptTemplate a partir da string QA_CHAIN_PROMPT = PromptTemplate.from_template(custom_prompt_template) # --- 6. Cria uma cadeia de QA que usa o retriever e o modelo --- qa_chain = RetrievalQA.from_chain_type( llm=chat_model, chain_type="stuff", retriever=retriever, return_source_documents=True, chain_type_kwargs={"prompt": QA_CHAIN_PROMPT} ) # --- 7. Função principal para a interface do Gradio --- def chat_response(question: str, history: list): # `history` é um parâmetro obrigatório para gr.ChatInterface """ Gera a resposta do chatbot usando o modelo de QA e formata para exibição no Gradio. Args: question (str): A pergunta do usuário. history (list): Histórico de conversas (não usado diretamente aqui, mas necessário para a interface). Returns: str: A resposta formatada do chatbot, incluindo as fontes. """ print(f"Recebida pergunta: '{question}'") # Para debug no console try: result = qa_chain.invoke({"query": question}) answer = result["result"] sources = result.get("source_documents", []) response_text = f"**Resposta:** {answer}\n\n" if sources: response_text += "**Saiba mais em:**\n" unique_sources = set() source_list_for_printing = [] for doc in sources: source_name = doc.metadata.get('source', 'Fonte desconhecida') if source_name not in unique_sources: unique_sources.add(source_name) source_list_for_printing.append(source_name) for i, source_name in enumerate(source_list_for_printing): response_text += f"- {i+1}. '{source_name}'\n" else: response_text += "Nenhuma fonte específica foi utilizada para esta resposta.\n" return response_text except Exception as e: error_message = f"Ocorreu um erro ao processar sua pergunta: {e}. Por favor, tente novamente." print(error_message) # Para debug no console return error_message # --- 8. Criação da Interface Gradio --- if __name__ == "__main__": print("Iniciando a interface Gradio...") demo = gr.ChatInterface( type="messages", fn=chat_response, # A função que processa a pergunta e retorna a resposta title="CloudWalk Chatbot", description="Olá! Estou aqui para responder suas dúvidas.", submit_btn="Enviar Pergunta", examples=[ #["What product and services InfinitePay offer ?"], #['What are the fees for the card machine?'], #['How can I order one card machine?'], ["Quais serviços a infinite pay oferece?"], ["Quais as taxas da maquininha?"], ["Como pedir uma maquininha?"], ], chatbot=gr.Chatbot(type="messages") ) demo.launch()