Mervin Praison

Not Working

import gradio as gr
import os
import time
import arxiv
from langchain_community.vectorstores import Qdrant
from langchain_community.document_loaders import PyPDFLoader, DirectoryLoader
from langchain_community.chat_models import ChatOllama
from langchain.prompts import ChatPromptTemplate
from langchain.pydantic_v1 import BaseModel
from langchain.schema.output_parser import StrOutputParser
from langchain.schema.runnable import RunnableParallel, RunnablePassthrough
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.embeddings import GPT4AllEmbeddings
from langchain_community.embeddings.ollama import OllamaEmbeddings

# Assuming Qdrant and embeddings setup done globally for simplicity
def init_or_load_qdrant(path="./tmp/local_qdrant", collection_name="arxiv_papers"):
    try:
        qdrant_instance = Qdrant.load(
            path=path,
            collection_name=collection_name,
        )
        print("Existing embeddings loaded.")
    except Exception as e:
        print("No existing embeddings found, starting new Qdrant instance.", e)
        qdrant_instance = None

    return qdrant_instance, qdrant_instance.as_retriever() if qdrant_instance else None

# Attempt to load existing Qdrant instance and retriever
qdrant, retriever = init_or_load_qdrant()

def process_papers(query):
    dirpath = "arxiv_papers"
    if not os.path.exists(dirpath):
        os.makedirs(dirpath)
    
    client = arxiv.Client()
    search = arxiv.Search(
        query=query,
        max_results=5,
        sort_order=arxiv.SortOrder.Descending
    )
    
    for result in client.results(search):
        while True:
            try:
                result.download_pdf(dirpath=dirpath)
                print(result)
                print(f"-> Paper id {result.get_short_id()} with title '{result.title}' is downloaded.")
                break
            except (FileNotFoundError, ConnectionResetError) as e:
                print("Error occurred:", e)
                time.sleep(5)
    
    papers = []
    loader = DirectoryLoader(dirpath, glob="./*.pdf", loader_cls=PyPDFLoader)
    try:
        papers = loader.load()
    except Exception as e:
        print(f"Error loading file: {e}")
    full_text = ''
    for paper in papers:
        full_text += paper.page_content
    
    full_text = " ".join(line for line in full_text.splitlines() if line)
    text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
    paper_chunks = text_splitter.create_documents([full_text])
    
    global qdrant, retriever
    qdrant = Qdrant.from_documents(
        documents=paper_chunks,
        embedding=OllamaEmbeddings(model='nomic-embed-text'),
        path="./tmp/local_qdrant",
        collection_name="arxiv_papers",
    )
    retriever = qdrant.as_retriever()
    
    return "Papers processed and saved to embeddings database."

def perform_query(question_text):
    global retriever
    if not retriever:
        return "Error: No papers processed. Please process papers first."
    
    template = """Answer the question based only on the following context:
{context}

Question: {question}
"""
    prompt = ChatPromptTemplate.from_template(template)
    
    ollama_llm = "llama2:7b-chat"
    model = ChatOllama(model=ollama_llm)
    
    chain = (
        RunnableParallel({"context": retriever, "question": RunnablePassthrough()})
        | prompt
        | model
        | StrOutputParser()
    )
    
    class Question(BaseModel):
        __root__: str
    
    chain = chain.with_types(input_type=Question)
    result = chain.invoke(Question(__root__=question_text))
    return result

with gr.Blocks() as demo:
    with gr.Tab("Process Papers"):
        query_input = gr.Text(label="Search Query")
        process_button = gr.Button("Process Papers")
        process_output = gr.Text(label="Process Output")
        process_button.click(process_papers, inputs=query_input, outputs=process_output)

    with gr.Tab("Query Retriever"):
        question_input = gr.Text(label="Question")
        query_button = gr.Button("Query")
        query_output = gr.Text(label="Query Output")
        query_button.click(perform_query, inputs=question_input, outputs=query_output)

demo.launch()

Error message

UnicodeEncodeError: 'utf-8' codec can't encode characters in position 100-103: surrogates not allowed

langchain_community/embeddings/gpt4all.py

from typing import Any, Dict, List

from langchain_core.embeddings import Embeddings
from langchain_core.pydantic_v1 import BaseModel, root_validator


class GPT4AllEmbeddings(BaseModel, Embeddings):
    """GPT4All embedding models.

    To use, you should have the gpt4all python package installed

    Example:
        .. code-block:: python

            from langchain_community.embeddings import GPT4AllEmbeddings

            embeddings = GPT4AllEmbeddings()
    """

    client: Any  #: :meta private:

    @root_validator()
    def validate_environment(cls, values: Dict) -> Dict:
        """Validate that GPT4All library is installed."""

        try:
            from gpt4all import Embed4All

            values["client"] = Embed4All()
        except ImportError:
            raise ImportError(
                "Could not import gpt4all library. "
                "Please install the gpt4all library to "
                "use this embedding model: pip install gpt4all"
            )
        return values
    def sanitize_text(self, text):
        return text.encode('utf-8', 'ignore').decode('utf-8')

    def embed_documents(self, texts: List[str]) -> List[List[float]]:
        """Embed a list of documents using GPT4All.

        Args:
            texts: The list of texts to embed.

        Returns:
            List of embeddings, one for each text.
        """
        texts_sanitized = [self.sanitize_text(text) for text in texts]
        embeddings = [self.client.embed(text) for text in texts_sanitized]
        return [list(map(float, e)) for e in embeddings]

    def embed_query(self, text: str) -> List[float]:
        """Embed a query using GPT4All.

        Args:
            text: The text to embed.

        Returns:
            Embeddings for the text.
        """
        return self.embed_documents([text])[0]