chat_with_the_doc_demo.ipynb

{
  "cells": [
    {
      "attachments": {},
      "cell_type": "markdown",
      "metadata": {
        "id": "etSxOSvHofmf"
      },
      "source": [
        "To run this demo, you need the following:\n",
        "* OpenAI API key\n",
        "* pinecone API key"
      ]
    },
    {
      "attachments": {},
      "cell_type": "markdown",
      "metadata": {
        "id": "_i7m-mBooKLT"
      },
      "source": [
        "0. Read the document with the software of your choice. At this point, we need a contiguous string to represent the content of the document."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "doc_text = ''"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "dbSvXQG2o72W"
      },
      "source": [
        "## Algorithm:\n",
        "*   Chunk text\n",
        "*   Embed each text chunk\n",
        "*   Embed question/query\n",
        "*   Perform a similarity search to find the text chunk embeddings that is the most similar to the question/query (i.e. have highest cosine similarities with the question embedding). \n",
        "*   API call to the completions endpoint, with the query and the most relevant text chunks included in the prompt. \n",
        "*   The GPT model then gives the answer to the question found in the file chunks, if the answer can be found in the extracts.\n",
        "\n",
        "## Limitations:\n",
        "*   The app may sometimes generate answers that are not in the files, or hallucinate about the existence of files that are not uploaded.\n",
        "\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "2O6XLjQco-5O"
      },
      "source": [
        "1. Text Chunking"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "OCFcBCADo0KM"
      },
      "outputs": [],
      "source": [
        "%%capture \n",
        "!pip install tiktoken"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "eZSFWNj9pBJa"
      },
      "outputs": [],
      "source": [
        "from typing import Dict, List, Optional, Tuple\n",
        "import uuid\n",
        "import tiktoken\n",
        "\n",
        "tokenizer = tiktoken.get_encoding(\n",
        "    \"cl100k_base\"\n",
        ")  # The encoding scheme to use for tokenization\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "VYpdZBAdpDGn"
      },
      "outputs": [],
      "source": [
        "# Constants\n",
        "CHUNK_SIZE = 200  # The target size of each text chunk in tokens\n",
        "MIN_CHUNK_SIZE_CHARS = 350  # The minimum size of each text chunk in characters\n",
        "MIN_CHUNK_LENGTH_TO_EMBED = 5  # Discard chunks shorter than this\n",
        "EMBEDDINGS_BATCH_SIZE = 128  # The number of embeddings to request at a time\n",
        "MAX_NUM_CHUNKS = 10000  # The maximum number of chunks to generate from a text\n",
        "\n",
        "def get_text_chunks(text: str, chunk_token_size: Optional[int]) -> List[str]:\n",
        "    \"\"\"\n",
        "    Split a text into chunks of ~CHUNK_SIZE tokens, based on punctuation and newline boundaries.\n",
        "    Args:\n",
        "        text: The text to split into chunks.\n",
        "        chunk_token_size: The target size of each chunk in tokens, or None to use the default CHUNK_SIZE.\n",
        "    Returns:\n",
        "        A list of text chunks, each of which is a string of ~CHUNK_SIZE tokens.\n",
        "    \"\"\"\n",
        "    # Return an empty list if the text is empty or whitespace\n",
        "    if not text or text.isspace():\n",
        "        return []\n",
        "\n",
        "    # Tokenize the text\n",
        "    tokens = tokenizer.encode(text, disallowed_special=())\n",
        "\n",
        "    # Initialize an empty list of chunks\n",
        "    chunks = []\n",
        "\n",
        "    # Use the provided chunk token size or the default one\n",
        "    chunk_size = chunk_token_size or CHUNK_SIZE\n",
        "\n",
        "    # Initialize a counter for the number of chunks\n",
        "    num_chunks = 0\n",
        "\n",
        "    # Loop until all tokens are consumed\n",
        "    while tokens and num_chunks < MAX_NUM_CHUNKS:\n",
        "        # Take the first chunk_size tokens as a chunk\n",
        "        chunk = tokens[:chunk_size]\n",
        "\n",
        "        # Decode the chunk into text\n",
        "        chunk_text = tokenizer.decode(chunk)\n",
        "\n",
        "        # Skip the chunk if it is empty or whitespace\n",
        "        if not chunk_text or chunk_text.isspace():\n",
        "            # Remove the tokens corresponding to the chunk text from the remaining tokens\n",
        "            tokens = tokens[len(chunk) :]\n",
        "            # Continue to the next iteration of the loop\n",
        "            continue\n",
        "\n",
        "        # Find the last period or punctuation mark in the chunk\n",
        "        last_punctuation = max(\n",
        "            chunk_text.rfind(\".\"),\n",
        "            chunk_text.rfind(\"?\"),\n",
        "            chunk_text.rfind(\"!\"),\n",
        "            chunk_text.rfind(\"\\n\"),\n",
        "        )\n",
        "\n",
        "        # If there is a punctuation mark, and the last punctuation index is before MIN_CHUNK_SIZE_CHARS\n",
        "        if last_punctuation != -1 and last_punctuation > MIN_CHUNK_SIZE_CHARS:\n",
        "            # Truncate the chunk text at the punctuation mark\n",
        "            chunk_text = chunk_text[: last_punctuation + 1]\n",
        "\n",
        "        # Remove any newline characters and strip any leading or trailing whitespace\n",
        "        chunk_text_to_append = chunk_text.replace(\"\\n\", \" \").strip()\n",
        "\n",
        "        if len(chunk_text_to_append) > MIN_CHUNK_LENGTH_TO_EMBED:\n",
        "            # Append the chunk text to the list of chunks\n",
        "            chunks.append(chunk_text_to_append)\n",
        "\n",
        "        # Remove the tokens corresponding to the chunk text from the remaining tokens\n",
        "        tokens = tokens[len(tokenizer.encode(chunk_text, disallowed_special=())) :]\n",
        "\n",
        "        # Increment the number of chunks\n",
        "        num_chunks += 1\n",
        "\n",
        "    # Handle the remaining tokens\n",
        "    if tokens:\n",
        "        remaining_text = tokenizer.decode(tokens).replace(\"\\n\", \" \").strip()\n",
        "        if len(remaining_text) > MIN_CHUNK_LENGTH_TO_EMBED:\n",
        "            chunks.append(remaining_text)\n",
        "\n",
        "    return chunks"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "lCeEcv5ppFIw"
      },
      "outputs": [],
      "source": [
        "chunks = get_text_chunks(doc_text, 200)\n",
        "assert type(chunks[0]) == str"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "4Wt72LSApOnl"
      },
      "source": [
        "2. Embed each text chunk"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "iVzQ9E-NpKpi"
      },
      "outputs": [],
      "source": [
        "%%capture\n",
        "!pip install -qU openai\n",
        "!pip install tenacity"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "EIpo1Wj_pRGj"
      },
      "outputs": [],
      "source": [
        "import openai\n",
        "from tenacity import retry, wait_random_exponential, stop_after_attempt"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "MSGj0toQpSlz"
      },
      "outputs": [],
      "source": [
        "OPENAI_API_KEY = None\n",
        "# Note that for most operations, you need a paid account\n",
        "if not OPENAI_API_KEY:\n",
        "  OPENAI_API_KEY = input('Your OpenAI API key:').strip()\n",
        "os.environ[\"OPENAI_API_KEY\"] = OPENAI_API_KEY"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "-gBbtALEpUFk"
      },
      "outputs": [],
      "source": [
        "embed_model = \"text-embedding-ada-002\"\n",
        "\n",
        "@retry(wait=wait_random_exponential(min=1, max=20), stop=stop_after_attempt(3))\n",
        "def get_embeddings(texts: List[str]) -> List[List[float]]:\n",
        "    \"\"\"\n",
        "    Embed texts using OpenAI's ada model.\n",
        "    Args:\n",
        "        texts: The list of texts to embed.\n",
        "    Returns:\n",
        "        A list of embeddings, each of which is a list of floats.\n",
        "    Raises:\n",
        "        Exception: If the OpenAI API call fails.\n",
        "    \"\"\"\n",
        "    # Call the OpenAI API to get the embeddings\n",
        "    response = openai.Embedding.create(input=texts, model=embed_model)\n",
        "\n",
        "    # Extract the embedding data from the response\n",
        "    data = response[\"data\"]  # type: ignore\n",
        "\n",
        "    # Return the embeddings as a list of lists of floats\n",
        "    return [result[\"embedding\"] for result in data]"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "-qDWRgbipWBV"
      },
      "outputs": [],
      "source": [
        "# this would be the embeddings for the whole document: list of embeddings for each document chunk; \n",
        "# embedding is done in batches of size EMBEDDINGS_BATCH_SIZE\n",
        "\n",
        "embeddings: List[List[float]] = []\n",
        "for i in range(0, len(chunks), EMBEDDINGS_BATCH_SIZE):\n",
        "  # Get the text of the chunks in the current batch\n",
        "  batch_texts = [\n",
        "      chunk for chunk in chunks[i : i + EMBEDDINGS_BATCH_SIZE]\n",
        "  ]\n",
        "\n",
        "  # Get the embeddings for the batch texts\n",
        "  batch_embeddings = get_embeddings(batch_texts)\n",
        "\n",
        "  # Append the batch embeddings to the embeddings list\n",
        "  embeddings.extend(batch_embeddings)\n",
        "# this number may differ dependeing on the embedding model you use. For text-embedding-ada-002, it's this number.\n",
        "assert len(embeddings[0]) == 1536"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "bDoWeC5ypeMt"
      },
      "source": [
        "3. Index the obtained embeddings\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "gI90Cw4hpZ1X"
      },
      "outputs": [],
      "source": [
        "!pip install -qU pinecone-client"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "STvIgw_SphG1"
      },
      "outputs": [],
      "source": [
        "import pinecone\n",
        "\n",
        "index_name = 'openai-honda-manuals'\n",
        "PINECONE_API_KEY = None\n",
        "if not PINECONE_API_KEY:\n",
        "  PINECONE_API_KEY = input(\"Your pinecone API key:\").strip()\n",
        "# initialize connection to pinecone (get API key at app.pinecone.io)\n",
        "pinecone.init(\n",
        "    api_key=PINECONE_API_KEY,\n",
        "    environment=\"us-west4-gcp\"  # may be different, check at app.pinecone.io under API Keys\n",
        ")\n",
        "\n",
        "# check if index already exists (it shouldn't if this is first time)\n",
        "if index_name not in pinecone.list_indexes():\n",
        "    # if does not exist, create index\n",
        "    pinecone.create_index(\n",
        "        index_name,\n",
        "        dimension=len(embeddings[0]),\n",
        "        metric='cosine',\n",
        "        metadata_config={'indexed': ['channel_id', 'published']}\n",
        "    )\n",
        "# connect to index\n",
        "index = pinecone.Index(index_name)\n",
        "# view index stats\n",
        "index.describe_index_stats()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "N2bMeIPJpjZ1"
      },
      "outputs": [],
      "source": [
        "from tqdm.auto import tqdm\n",
        "import datetime\n",
        "from time import sleep\n",
        "\n",
        "for i in tqdm(range(0, len(chunks), EMBEDDINGS_BATCH_SIZE)):\n",
        "    # find end of batch\n",
        "    i_end = min(len(chunks), i+EMBEDDINGS_BATCH_SIZE)\n",
        "    meta_batch = chunks[i:i_end]\n",
        "    # # get ids\n",
        "    ids_batch = ['id'+str(i+100) for i in range(len(meta_batch))]\n",
        "    # get texts to encode\n",
        "    texts = [x for x in meta_batch]\n",
        "    # create embeddings (try-except added to avoid RateLimitError)\n",
        "    try:\n",
        "        res = openai.Embedding.create(input=texts, engine=embed_model)\n",
        "    except:\n",
        "        done = False\n",
        "        while not done:\n",
        "            sleep(5)\n",
        "            try:\n",
        "                res = openai.Embedding.create(input=texts, engine=embed_model)\n",
        "                done = True\n",
        "            except:\n",
        "                pass\n",
        "    embeds = [record['embedding'] for record in res['data']]\n",
        "    # cleanup metadata\n",
        "    meta_batch = [{'text': x} for x in meta_batch]\n",
        "    to_upsert = list(zip(ids_batch, embeds, meta_batch))\n",
        "    # upsert to Pinecone\n",
        "    index.upsert(vectors=to_upsert)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "snuMqTVopoF5"
      },
      "source": [
        "4. Embedding Question/Query"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "lHQXRE0apmXf"
      },
      "outputs": [],
      "source": [
        "# # This is the only line that you should change based on what you want to search\n",
        "query = 'how to change oil'\n",
        "\n",
        "res = openai.Embedding.create(\n",
        "    input=[query],\n",
        "    engine=embed_model\n",
        ")\n",
        "\n",
        "# query embedding\n",
        "xq = res['data'][0]['embedding']\n",
        "\n",
        "# get relevant contexts (including the questions)\n",
        "res = index.query(xq, top_k=10, include_metadata=True)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "kIndoDbnptLv"
      },
      "source": [
        "5. Use ChatGPT to pose a question to the document and find an answer for it."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "3WdIWCIypqn4"
      },
      "outputs": [],
      "source": [
        "!pip install -qU langchain\n",
        "# !pip install -qU transformers\n",
        "# !pip install -qU sentence_transformers\n",
        "# !pip install -qU chromadb\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "nqnpY4qWpv4f"
      },
      "outputs": [],
      "source": [
        "from langchain.chat_models import ChatOpenAI\n",
        "from langchain import PromptTemplate, LLMChain\n",
        "from langchain.prompts.chat import (\n",
        "    ChatPromptTemplate,\n",
        "    SystemMessagePromptTemplate,\n",
        "    AIMessagePromptTemplate,\n",
        "    HumanMessagePromptTemplate,\n",
        ")\n",
        "from langchain.schema import (\n",
        "    AIMessage,\n",
        "    HumanMessage,\n",
        "    SystemMessage\n",
        ")"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "iIauKXl6pxyv"
      },
      "outputs": [],
      "source": [
        "chat = ChatOpenAI(temperature=0)"
      ]
    },
    {
      "attachments": {},
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "I asked the following question because the document I read was a Honda engine instruction manual. You can replace the prompt according to the content of your document. For more prompt engineering hints, see LangChain documentation."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "q_6fykLfpzUN"
      },
      "outputs": [],
      "source": [
        "QA_prompt = \"\"\"Given the following text, search for the best answer for the following question.\n",
        "\n",
        "Text: {document_text}.\n",
        "Question: {query}.\n",
        "Answer:\"\"\"\n",
        "\n",
        "# This is the only line that you should change based on what question you want to ask\n",
        "myquestion = \"How do I change engine oil?\"\n",
        "\n",
        "my_context = ' '.join([x['metadata']['text'] for x in res['matches']])\n",
        "messages = [\n",
        "    SystemMessage(content=\"You are a helpful assistant that can search a text and find the answer to a question.\"),\n",
        "    HumanMessage(content=QA_prompt.format(document_text=my_context, query=myquestion)),\n",
        "]"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "GfxGxno2qD_f"
      },
      "outputs": [],
      "source": [
        "print(chat(messages).content)"
      ]
    }
  ],
  "metadata": {
    "colab": {
      "provenance": []
    },
    "kernelspec": {
      "display_name": "Python 3",
      "name": "python3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}