WebGPT is a library based on the OpenAI API and LangChain. It allows you to connect ChatGPT to the Internet without a seam to make requests and get up-to-date information.
from setuptools import setup, find_packages


def readme():
    with open('README.md', 'r') as f:
        return f.read()


setup(
    name='web_gpt',
    version='1.0.0',
    author='ZeroN',
    author_email='dotdot20072607@gmail.com',
    description='WebGPT is a library based on the Openal API and Long Chain. It allows you to connect ChatGPT to the Internet without a seam to make requests and receive responses with up-to-date information.',
    long_description=readme(),
    url='https://github.com/OnaZeroN/WebGPT',
    packages=find_packages(),
    install_requires=['requests>=2.25.1'],
    classifiers=[
        'Programming Language :: Python :: 3.11',
        'License :: OSI Approved :: MIT License',
        'Operating System :: OS Independent'
    ],
    keywords='example python',
    project_urls={
        'Documentation': 'link'
    },
    python_requires='>=3.9'
)

setup.py
OnaZeroN/WebGPT
from web_gpt.web_gpt import WebGPT

web_gpt/__init__.py
OnaZeroN/WebGPT
__title__ = "web_gpt"
__version__ = "0.0.1"

web_gpt/_version.py
OnaZeroN/WebGPT
import asyncio

from duckduckgo_search import DDGS

from langchain.chains import RetrievalQA
from langchain.chat_models import ChatOpenAI
from langchain.document_loaders import AsyncChromiumLoader
from langchain.document_transformers import Html2TextTransformer
from langchain.embeddings import OpenAIEmbeddings
from langchain.prompts import PromptTemplate
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.vectorstores import Chroma


class LangChainHelper:
    def __init__(self, vector_store_model: str, search_region: str):
        self.vector_store_model = vector_store_model
        self.llm = ChatOpenAI(model_name=self.vector_store_model, temperature=0, max_tokens=10000)
        self.messages = None
        self.loop = asyncio.get_event_loop()
        self.search_region = search_region

    async def llm_asq(self, template: str, vectorstore, last_query: str):
        qa_chain = RetrievalQA.from_chain_type(self.llm, retriever=vectorstore.as_retriever(), chain_type_kwargs={"prompt": PromptTemplate.from_template(template)})
        return await self.loop.run_in_executor(None, qa_chain, {"query": last_query + "\n\nОтвечай на русском"})

    async def create_index(self, query=None, all_splits=None):
        if not all_splits:
            if query["url"] is None:
                url = await self.get_urls(query["text"])
            else:
                url = query["url"]
            data = Html2TextTransformer().transform_documents(
                await self.loop.run_in_executor(None, AsyncChromiumLoader([url]).load))
            all_splits = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=0).split_documents(data)
        vectorstore = await self.loop.run_in_executor(None, Chroma.from_documents, all_splits, OpenAIEmbeddings())
        return vectorstore, all_splits

    async def get_urls(self, query):
        with DDGS() as search:
            for r in search.text(query, region=self.search_region, safesearch='off', timelimit='y'):
                return r['href']

web_gpt/langchain_helper.py
OnaZeroN/WebGPT
import os
import openai


class OpenAiHelper:
    def __init__(self, model: str):
        self.model = model
        self.functions = [
                {
                    "name": "internet_search",
                    "description": "Функция для нахождения информации в интернете, ее надо вызывать когда для ответа юзера требуется актуальная информация из интернета",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "query": {
                                "type": "string",
                                "description": "Запрос в интернет, если в запросе есть ссылка, то передавай сообщение юзера БЕЗ изменений"
                            }
                        },
                        "required": ["properties"],
                    },
                }
            ]

    async def chat_completion(self, messages):
        openai.api_key = self.get_open_ai_key()
        return await openai.ChatCompletion.acreate(
            model=self.model,
            messages=messages,
            functions=self.functions
        )

    @staticmethod
    def get_open_ai_key():
        open_ai_key = os.getenv("OPENAI_API_KEY")
        if open_ai_key is not None:
            return open_ai_key
        else:
            raise Exception("""
    The environment variable Open AI API key is not set
    Information on how to get the API key can be found here
    https://platform.openai.com/docs/api-reference/
    For easy installation of the Open AI API key in environment variable:
    os.environ['OPENAI_API_KEY'] = "api_key"
                """)

web_gpt/open_ai_helper.py
OnaZeroN/WebGPT
# -*- coding: utf-8 -*-
import asyncio
import json
import re

from web_gpt.open_ai_helper import OpenAiHelper
from web_gpt.langchain_helper import LangChainHelper


class WebGPT(OpenAiHelper, LangChainHelper):
    def __init__(
            self,
            model: str = "gpt-3.5-turbo",
            vector_store_model: str = "gpt-3.5-turbo-16k",
            prompt: str = """
            Use the following context snippets to answer the question at the end.
            Instructions:
            - Carefully read the document and analyze it fully understanding its meaning.
            - Conduct a hermeneutic analysis, exploring in depth the textual and contextual layers of the document.
            - Answer the question in as much detail as possible to the user's question, as much as the context allows.
            - If you don't know the answer, just say you don't know, don't try to come up with an answer.
            - Give fragments of the context from where you got the information.
            - Always answer in the form of a marking list.
            - Study the document, get into its essence, reveal hidden meanings and subtext.
            {context}
            """,
            urls_count: int = 1,
            search_region: str = "ru-ru"
    ):
        """
        Class based on Open air API and Langchain, seamlessly connecting ChatGPT to the Internet
        :param model: chat model
        :param vector_store_model: the model that will search the site, "gpt-3.5-turbo-16k" is recommended
        :param prompt: instructions for the neural network on how to work with the document, you must add {context} to the end, it is recommended to leave the default
        :param urls_count: the number of links from which the neural network will take information, it is recommended no more than 3
        :param search_region: region for internet search, for example "ru-ru"
        """
        self.urls_count = urls_count
        self.messages = None
        self.open_ai_key = self.get_open_ai_key()
        self.loop = asyncio.get_event_loop()
        self.prompt = prompt
        OpenAiHelper.__init__(self, model)
        LangChainHelper.__init__(self, vector_store_model, search_region)
        self.open_ai_key = self.get_open_ai_key()

    async def ask(self, messages: list) -> dict:
        """
        A method of the Class based on a request to Web GT, if the user's response requires information from the Internet, it will launch LangChain methods, otherwise
        WebGPT will give a normal response

        The response is returned as a dict:
        If the response did not require the Internet: {"type": "def", "content": "Response"}
        If the response required the Internet: {"type": "web", "content": "Response", "vectorstore": "formatted information source"}

        :param messages: list of messages, more details https://platform.openai.com/docs/api-reference/chat
        """
        self.messages = messages
        chat_completion = await self.chat_completion(messages=self.messages)
        try:
            if chat_completion["choices"][0]["message"]["content"] is not None:
                return {"type": "def", "content": chat_completion["choices"][0]["message"]["content"]}
            llm_answer = await self.vector_store_asq(json.loads(chat_completion["choices"][0]["message"]["function_call"]["arguments"])["query"])
            return {"type": "web", "content": llm_answer["content"], "vectorstore": llm_answer["vectorstore"]}
        except Exception as e:
            raise Exception(e)

    async def vector_store_asq(self, query=None, old_vectorstore=None, messages: list = None) -> dict:
        """
        A method of the Class based on a request to an already existing vectorstore
        WebGPT will give a normal response

        The response is returned as a dict:
        {"content": "Response", "vectorstore": vectorstore}

        :param old_vectorstore: formatted information source that is returned in the ask method
        :param messages: list of messages, more details https://platform.openai.com/docs/api-reference/chat
        """
        template = self.prompt
        if old_vectorstore is None:
            query = await self.find_links(query)
            if len(query["text"]) < 2:
                query["text"] = "Очень подробно суммаризируй эту статью"
            vector_data = await self.create_index(query)
            vectorstore = vector_data[0]
            old_vectorstore = vector_data[1]
            last_query = query["text"]
            template += "\n\nВопрос пользователя: {question}\nОтвет полезного помощника, который следует инструкциям:"
        else:
            vectorstore = (await self.create_index(all_splits=old_vectorstore))[0]
            self.messages = messages
            if len(self.messages) > 1:
                for message in self.messages[:-1]:
                    if message["role"] == "user":
                        template += f"Вопрос пользователя: {message['content']}"
                    else:
                        template += f"Ответ полезного помощника, который следует инструкциям: {message['content']}"
            template += "Вопрос пользователя: {question}\nОтвет полезного помощника, который следует инструкциям:"
            last_query = self.messages[-1]["content"]

        result = await self.llm_asq(template=template, vectorstore=vectorstore, last_query=last_query)

        return {"content": result["result"], "vectorstore": old_vectorstore}

    @staticmethod
    async def find_links(text):
        pattern = r'http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\(\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+'
        matches = re.findall(pattern, text)
        if len(matches) != 0:
            return {"text": text.replace(matches[0], ""), "url": matches[0]}
        return {"text": text, "url": None}

web_gpt/web_gpt.py
OnaZeroN/WebGPT