Web QA with embeddings

For the complete documentation index, see llms.txt. Markdown versions of documentation pages are available by appending .md to the page URL.

This tutorial walks through an example of crawling a website (in this example, the OpenAI website), turning the crawled pages into embeddings using the Embeddings API, and then creating a basic search functionality that allows a user to ask questions about the embedded information. This is intended to be a starting point for more sophisticated applications that make use of custom knowledge bases.

Getting started

Some basic knowledge of Python and GitHub is helpful for this tutorial. Before diving in, make sure to set up an OpenAI API key and walk through the quickstart tutorial. This will give a good intuition on how to use the API to its full potential.

Use Python 3.11 or later. Create a virtual environment, install the packages used in the examples, and set OPENAI_API_KEY as described in the quickstart.

python -m venv env
source env/bin/activate
python -m pip install openai pandas numpy requests beautifulsoup4 tiktoken matplotlib

Copy the examples into a Python file or notebook in the order shown and run them in the same working directory.

Setting up a web crawler

Choose a website you can crawl. Expand the steps below to collect page text for the embeddings.

Learn how to build a web crawler

Acquiring data in text form is the first step to use embeddings. This
  tutorial creates a new set of data by crawling the OpenAI website, a
  technique that you can also use for your own company or personal website.

While this crawler is written from scratch, open source packages like Scrapy can also help with these operations.

This crawler will start from the root URL passed in at the bottom of the code below, visit each page, find additional links, and visit those pages as well (as long as they have the same root domain). To begin, import the required packages, set the starting URL, and define a subclass of HTMLParser.

import requests
import re
import urllib.request
from bs4 import BeautifulSoup
from collections import deque
from html.parser import HTMLParser
from urllib.parse import urldefrag, urljoin, urlparse
import os

# Regex pattern to match a URL
HTTP_URL_PATTERN = r"^http[s]*://.+"

domain = "openai.com"  # <- put your domain to be crawled
full_url = "https://openai.com/"  # <- put your domain to be crawled with https or http


# Create a class to parse the HTML and get the hyperlinks
class HyperlinkParser(HTMLParser):
    def __init__(self):
        super().__init__()
        # Create a list to store the hyperlinks
        self.hyperlinks = []

    # Override the HTMLParser's handle_starttag method to get the hyperlinks
    def handle_starttag(self, tag, attrs):
        attrs = dict(attrs)

        # If the tag is an anchor tag and it has an href attribute, add the href attribute to the list of hyperlinks
        if tag == "a" and "href" in attrs:
            self.hyperlinks.append(attrs["href"])

The next function takes a URL as an argument, opens the URL, and reads the HTML content. Then, it returns all the hyperlinks found on that page.

# Function to get the hyperlinks from a URL
def get_hyperlinks(url):
    # Try to open the URL and read the HTML
    try:
        with urllib.request.urlopen(url, timeout=30) as response:
            # If the response is not HTML, return an empty list
            if not response.info().get("Content-Type", "").startswith("text/html"):
                return []

            # Decode the HTML
            html = response.read().decode("utf-8")
    except Exception as error:
        print(error)
        return []

    # Create the HTML Parser and then Parse the HTML to get hyperlinks
    parser = HyperlinkParser()
    parser.feed(html)

    return parser.hyperlinks

The goal is to crawl through and index only the content that lives under the OpenAI domain. For this purpose, a function that calls the get_hyperlinks function but filters out any URLs that are not part of the specified domain is needed.

# Function to get the hyperlinks from a URL that are within the same domain
def get_domain_hyperlinks(local_domain, url):
    links = set()
    for link in get_hyperlinks(url):
        if not link or link.startswith("#"):
            continue
        candidate, _ = urldefrag(urljoin(url, link))
        parsed = urlparse(candidate)
        if parsed.scheme in {"http", "https"} and parsed.netloc == local_domain:
            links.add(candidate)
    return list(links)

The crawl function is the final step in the web scraping task setup. It keeps track of the visited URLs to avoid repeating the same page, which might be linked across multiple pages on a site. It also extracts the raw text from a page without the HTML tags, and writes the text content into a local .txt file specific to the page.

from hashlib import sha256


def crawl(url):
    # Parse the URL and get the domain
    parsed = urlparse(url)
    local_domain = parsed.netloc
    url = parsed._replace(path=parsed.path or "/").geturl()

    # Create a queue to store the URLs to crawl
    queue = deque([url])

    # Create a set to store the URLs that have already been seen (no duplicates)
    seen = set([url])

    # Create a directory to store the text files
    if not os.path.exists("text/"):
        os.mkdir("text/")

    if not os.path.exists("text/" + local_domain + "/"):
        os.mkdir("text/" + local_domain + "/")

    # Create a directory to store the csv files
    if not os.path.exists("processed"):
        os.mkdir("processed")

    # While the queue is not empty, continue crawling
    while queue:
        # Get the next URL from the queue
        url = queue.pop()
        print(url)  # for debugging and to see the progress

        # Save text from the url to a <url>.txt file
        parsed_page = urlparse(url)
        page_name = (local_domain + parsed_page.path).replace("/", "_")
        if parsed_page.query:
            # Parsed URL domains and paths cannot contain the fragment separator '#'.
            suffix = "#" + sha256(url.encode("utf-8")).hexdigest()
            # Keep the filename, including its extension, within 255 bytes.
            prefix_bytes = 255 - len(suffix) - len(".txt")
            name_prefix = page_name.encode("utf-8")[:prefix_bytes]
            page_name = name_prefix.decode("utf-8", errors="ignore") + suffix
        with open(
            "text/" + local_domain + "/" + page_name + ".txt",
            "w",
            encoding="UTF-8",
        ) as f:
            # Get the text from the URL using BeautifulSoup
            response = requests.get(url, timeout=10)
            response.raise_for_status()
            soup = BeautifulSoup(response.text, "html.parser")

            # Get the text but remove the tags
            text = soup.get_text()

            # If the crawler gets to a page that requires JavaScript, it will stop the crawl
            if "You need to enable JavaScript to run this app." in text:
                print(
                    "Unable to parse page " + url + " due to JavaScript being required"
                )

            # Otherwise, write the text to the file in the text directory
            f.write(text)

        # Get the hyperlinks from the URL and add them to the queue
        for link in get_domain_hyperlinks(local_domain, url):
            if link not in seen:
                queue.append(link)
                seen.add(link)


crawl(full_url)

The last line of the above example runs the crawler which goes through all the accessible links and turns those pages into text files. This will take a few minutes to run depending on the size and complexity of your site.

Building an embeddings index

CSV is a common format for storing embeddings. You can use this format
  with Python by converting the raw text files (which are in the text
  directory) into Pandas data frames. Pandas is a popular open source
  library that helps you work with tabular data (data stored in rows and
  columns).




  Blank empty lines can clutter the text files and make them harder to
  process. A function can remove those lines and tidy up the files.

def remove_newlines(serie):
    serie = serie.str.replace("\n", " ")
    serie = serie.str.replace("\\n", " ")
    serie = serie.str.replace("  ", " ")
    serie = serie.str.replace("  ", " ")
    return serie

Converting the text to CSV requires looping through the text files in the text directory created earlier. After opening each file, remove the extra spacing and append the modified text to a list. Then, add the text with the new lines removed to an empty Pandas data frame and write the data frame to a CSV file.

Extra spacing and new lines can clutter the text and complicate the embeddings process. The code used here helps to remove some of them but you may find 3rd party libraries or other methods useful to get rid of more unnecessary characters.

from pathlib import Path

import pandas as pd

# Create a list to store the text files
texts = []

# Get all the text files in the text directory
for file in os.listdir("text/" + domain + "/"):
    # Open the file and read the text
    with open("text/" + domain + "/" + file, "r", encoding="UTF-8") as f:
        text = f.read()

        page_name = Path(file).stem
        domain_prefix = f"{domain}_"
        if page_name.startswith(domain_prefix):
            page_name = page_name[len(domain_prefix) :]
        elif page_name == domain:
            page_name = "index"

        # Replace -, _, and #update with spaces.
        texts.append(
            (
                page_name.replace("-", " ").replace("_", " ").replace("#update", ""),
                text,
            )
        )

# Create a dataframe from the list of texts
df = pd.DataFrame(texts, columns=["fname", "text"])

# Set the text column to be the raw text with the newlines removed
df["text"] = df.fname + ". " + remove_newlines(df.text)
df.to_csv("processed/scraped.csv")
df.head()

The next step is tokenization after saving the raw text into a CSV file. This process splits the input text into tokens by breaking down the sentences and words. A visual demonstration of this can be seen by checking out our tokenizer in the docs.

A helpful rule of thumb is that one token generally corresponds to ~4 characters of text for common English text. This translates to roughly ¾ of a word (so 100 tokens ~= 75 words).

The API has a limit on the maximum number of input tokens for embeddings. To stay below the limit, the text in the CSV file needs to be broken down into multiple rows. The existing length of each row will be recorded first to identify which rows need to be split.

import tiktoken

# Load the cl100k_base tokenizer which is designed to work with the ada-002 model
tokenizer = tiktoken.get_encoding("cl100k_base")

df = pd.read_csv("processed/scraped.csv", index_col=0)
df.columns = ["title", "text"]

# Tokenize the text and save the number of tokens to a new column
df["n_tokens"] = df.text.apply(lambda x: len(tokenizer.encode(x)))

# Visualize the distribution of the number of tokens per row using a histogram
df.n_tokens.hist()

<img src="https://cdn.openai.com/API/docs/images/tutorials/web-qa/embeddings-initial-histrogram.png"
  alt="Embeddings histogram"
  width="553"
  height="413"
/>

Split longer pages into chunks that fit within the embedding model’s input limit.

max_tokens = 500


# Function to split the text into chunks of a maximum number of tokens
def split_into_many(text, max_tokens=max_tokens):
    if max_tokens < 1:
        raise ValueError("max_tokens must be positive.")
    if not text.strip():
        return []

    sentences = text.split(". ")
    chunks = []
    chunk = ""
    for index, sentence in enumerate(sentences):
        if index < len(sentences) - 1:
            sentence += "."
        # Do not silently discard a sentence that needs finer-grained splitting.
        if len(tokenizer.encode(sentence)) > max_tokens:
            raise ValueError("A sentence exceeds max_tokens; split it before chunking.")

        candidate = f"{chunk} {sentence}" if chunk else sentence
        if len(tokenizer.encode(candidate)) > max_tokens:
            chunks.append(chunk)
            chunk = sentence
        else:
            chunk = candidate

    if chunk:
        chunks.append(chunk)
    return chunks


shortened = []

# Loop through the dataframe
for row in df.iterrows():
    # If the text is None, go to the next row
    if row[1]["text"] is None:
        continue

    # If the number of tokens is greater than the max number of tokens, split the text into chunks
    if row[1]["n_tokens"] > max_tokens:
        shortened += split_into_many(row[1]["text"])

    # Otherwise, add the text to the list of shortened texts
    else:
        shortened.append(row[1]["text"])

Visualizing the updated histogram again can help to confirm if the rows were successfully split into shortened sections.

df = pd.DataFrame(shortened, columns=["text"])
df["n_tokens"] = df.text.apply(lambda x: len(tokenizer.encode(x)))
df.n_tokens.hist()

<img src="https://cdn.openai.com/API/docs/images/tutorials/web-qa/embeddings-tokenized-output.png"
  alt="Embeddings tokenized output"
  width="552"
  height="418"
/>

Create an embedding for each chunk with text-embedding-3-small. Use the same model for query embeddings. If you change embedding models, rebuild the index before searching it.

from openai import OpenAI

client = OpenAI()

df["embeddings"] = df.text.apply(
    lambda x: client.embeddings.create(
        input=x, model="text-embedding-3-small"
    ).data[0].embedding
)

df.to_csv("processed/embeddings.csv")
df.head()

This should take about 3-5 minutes but after you will have your embeddings ready to use!

Building a question answer system with your embeddings

Embed the user's question and compare it with the indexed chunks to find
  the most relevant text. Send that text and the question to the Responses
  API to generate an answer.

Turning the embeddings into a NumPy array is the first step, which will provide more flexibility in how to use it given the many functions available that operate on NumPy arrays. It will also flatten the dimension to 1-D, which is the required format for many subsequent operations.

from ast import literal_eval
import numpy as np

df = pd.read_csv("processed/embeddings.csv", index_col=0)
df["embeddings"] = df["embeddings"].apply(literal_eval).apply(np.array)

df.head()

Embed the question using the same model you used to build the index. The example computes cosine distances with NumPy and retrieves the closest text chunks.

def distances_from_embeddings(query, embeddings, distance_metric="cosine"):
    if distance_metric != "cosine":
        raise ValueError("This example supports cosine distance only.")
    return [
        1 - float(np.dot(query, item) / (np.linalg.norm(query) * np.linalg.norm(item)))
        for item in embeddings
    ]

def create_context(question, df, max_len=1800):
    """
    Create a context for a question by finding the most similar context from the dataframe
    """

    # Get the embeddings for the question
    q_embeddings = (
        client.embeddings.create(input=question, model="text-embedding-3-small")
        .data[0]
        .embedding
    )

    # Get the distances from the embeddings
    df["distances"] = distances_from_embeddings(
        q_embeddings, df["embeddings"].values, distance_metric="cosine"
    )

    returns = []
    cur_len = 0

    # Sort by distance and add the text to the context until the context is too long
    for _, row in df.sort_values("distances", ascending=True).iterrows():
        # Add the length of the text to the current length
        cur_len += row["n_tokens"] + 4

        # If the context is too long, break
        if cur_len > max_len:
            break

        # Else add it to the text that is being returned
        returns.append(row["text"])

    # Return the context
    return "\n\n###\n\n".join(returns)

The max_len argument limits how many tokens of retrieved text to include. Pass that text and the question to the Responses API using gpt-6-luna. Read the answer from response.output_text.

def answer_question(df, question, max_len=1800):
    context = create_context(question, df, max_len=max_len)
    response = client.responses.create(
        model="gpt-6-luna",
        instructions=(
            "Answer the question using only the supplied context. "
            "If the context doesn't answer the question, say \"I don't know.\""
        ),
        input=f"Context: {context}\n\nQuestion: {question}",
    )
    return response.output_text.strip()

Ask a question the indexed pages can answer, then ask one they cannot. If the index contains a description of ChatGPT, the first call should answer from it. The second should return “I don't know.”

print(answer_question(df, "What is ChatGPT?"))
print(answer_question(df, "What did I have for breakfast today?"))

If the system is not able to answer a question that is expected, it is worth searching through the raw text files to see if the information that is expected to be known actually ended up being embedded or not. The crawling process that was done initially was setup to skip sites outside the original domain that was provided, so it might not have that knowledge if there was a subdomain setup.

Currently, the data frame is being passed in each time to answer a question. For more production workflows, a vector database solution should be used instead of storing the embeddings in a CSV file, but the current approach is a great option for prototyping.