First commit

Dockerfile

@@ -0,0 +1,28 @@
+##############
+# Base Image #
+##############
+FROM ghcr.io/astral-sh/uv:python3.10-bookworm-slim AS base
+
+WORKDIR /app
+
+RUN --mount=type=cache,target=/root/.cache/uv \
+    --mount=type=bind,source=uv.lock,target=uv.lock \
+    --mount=type=bind,source=pyproject.toml,target=pyproject.toml \
+    uv sync --frozen --no-install-project
+
+#####################
+# Development Image #
+#####################
+FROM base AS development
+
+COPY --from=base /app/.venv /app/.venv
+
+COPY . /app
+
+RUN uv sync --frozen
+
+RUN uv run python src/scrape.py
+
+ENV GRADIO_SERVER_NAME="0.0.0.0"
+
+CMD ["uv", "run", "python", "-m", "src.app"]

README.md

@@ -1,13 +1,11 @@
 ---
 title: Bookstore Chatbot
-emoji: 📉
-colorFrom: indigo
-colorTo: red
-sdk: gradio
-sdk_version: 5.15.0
-app_file: app.py
+emoji: 👁
+colorFrom: red
+colorTo: indigo
+sdk: docker
 pinned: false
-short_description: An online bookstore chatbot
+short_description: A chabot for an online bookstore
 ---
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# store-data-chatbot
+A chatbot to ask questions about products in online stores.

pyproject.toml

@@ -0,0 +1,12 @@
+[project]
+name = "store-data-chatbot"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.10"
+dependencies = [
+    "beautifulsoup4>=4.13.3",
+    "gradio>=5.15.0",
+    "openai>=1.61.1",
+    "requests>=2.32.3",
+]

src/app.py

@@ -0,0 +1,14 @@
+import gradio as gr
+
+from src.llm import process_user_question
+
+app = gr.Interface(
+    fn=process_user_question,
+    inputs="text",
+    outputs="text",
+    title="Bookstore Chatbot",
+    description="A simple chatbot interface. Ask me anything about the books!",
+)
+
+# Launch the app
+app.launch()

src/llm.py

@@ -0,0 +1,93 @@
+import os
+
+import openai
+
+from src.prompts import (
+    ORDER_SYSTEM_MESSAGE,
+)
+from src.tools import (
+    books_dataset_tools,
+    llm_tools_map,
+)
+from src.utils import create_logger, handle_function_calls, log_execution_time
+
+TEMPERATURE = 0
+MAX_COMPLETION_TOKENS = 2048
+
+logger = create_logger(logger_name="llm", log_file="api.log", log_level="info")
+
+model = os.environ.get("CHAT_MODEL")
+if not model:
+    logger.error("CHAT_MODEL environment variable is not set.")
+    raise ValueError("CHAT_MODEL environment variable is not set.")
+
+client = openai.OpenAI()
+
+
+@log_execution_time(logger=logger)
+def create_completion(
+    thread: list,
+    system_message: str = ORDER_SYSTEM_MESSAGE,
+    tools: list = books_dataset_tools,
+):
+    """
+    Creates a completion response from the language model based on the system and user messages.
+
+    Args:
+        user_message (str): The conversation thread containing all messages.
+        system_message (str): The system message to be included in the prompt.
+        tools (list): A list of tools available for LLM to use.
+
+    Returns:
+        str: The response message generated by the LLM.
+
+    Raises:
+        Exception: If the LLM fails to generate a response.
+    """
+    try:
+        logger.debug(f" create_completion |{system_message = }\n\n{thread = }")
+        completion = client.chat.completions.create(
+            messages=[
+                {"role": "system", "content": system_message},
+                *thread,
+            ],
+            model=model,
+            n=1,
+            temperature=TEMPERATURE,
+            max_tokens=MAX_COMPLETION_TOKENS,
+            tools=tools,
+            tool_choice="auto"
+        )
+
+        logger.debug(f" create_completion | {completion = }")
+        response = completion.choices[0].message
+        return response
+    except Exception as e:
+        logger.error(
+            f" create_completion | Error generating responses for chat thread '{thread}': {e}"
+        )
+        raise
+
+
+@log_execution_time(logger=logger)
+def process_user_question(user_question: str) -> str:
+    """Handles user questions using an LLM"""
+    thread = [{"role": "user", "content": user_question}]
+    try:
+        while True:
+            logger.info(f"Processing user input: {thread}")
+            response = create_completion(thread=thread, tools=books_dataset_tools)
+            if not response.tool_calls:
+                break
+            thread.append(response)
+            thread = handle_function_calls(
+                function_map=llm_tools_map,
+                response_message=response,
+                thread=thread,
+            )
+        return response.content
+    except Exception as e:
+        logger.error(
+            f"Error generating responses for user input:\n'{user_question}'\n\nError:\n{e}"
+        )
+        raise

src/prompts.py

@@ -0,0 +1,17 @@
+from textwrap import dedent
+
+ORDER_SYSTEM_MESSAGE = dedent("""\
+    You are an online bookstore AI assistant that helps users with their queries.
+    Your responses should be eloquent, concise and succinct.
+    You can use the provided tools to get relevant information that help you in assisting the user.
+    One of these tools, `query_books_database`, gives you access to a SQL database. You can use this tool to run SQL queries against it.
+    If there are no entries for the given SQL query, the tool responds with "No rows found for this sql query. This usually means there are no entries with the specified conditions.".
+    This means there are no books for the given conditions. Your response to the user question should reflect this.
+    The final answer should directly answer the user's question as well as cite the sources used for generating the answer.
+    Citing a source could be, for example, listing the SQL query (or queries) used to get the results.
+    The final response should be in the following format:
+    # Answer:
+    <answer goes here>
+    # References:
+    <references go here>
+""")

src/scrape.py

@@ -0,0 +1,225 @@
+import re
+import sqlite3
+import time
+from typing import Dict, List, Union
+
+import requests
+from bs4 import BeautifulSoup
+
+
+def get_total_pages_and_products(base_url: str) -> tuple[int, int]:
+    """
+    Dynamically determine the total number of pages and products from the home page.
+
+    Args:
+        base_url (str): The URL of the home page.
+
+    Returns:
+        tuple[int, int]: A tuple containing the total number of pages and total number of products.
+    """
+    response = requests.get(base_url)
+    soup = BeautifulSoup(response.content, "html.parser")
+
+    # Find results summary text
+    form = soup.find("form", class_="form-horizontal")
+    results_text = form.get_text(strip=True) if form else ""
+
+    # Extract total products
+    match = re.search(r"(\d+)\s*results", results_text)
+    total_products = int(match.group(1)) if match else 0
+
+    # Find page summary text
+    page_text_elem = soup.find("li", class_="current")
+    page_text = page_text_elem.text.strip() if page_text_elem else ""
+
+    # Extract total pages
+    match = re.search(r"Page \d+ of (\d+)", page_text)
+    total_pages = int(match.group(1)) if match else 0
+
+    return total_pages, total_products
+
+
+def scrape_book_details(book_url: str) -> tuple[str, str, int]:
+    """
+    Scrape detailed information for a specific book.
+
+    Args:
+        book_url (str): The URL of the book's detail page.
+
+    Returns:
+        tuple[str, str, int]: A tuple containing the book's description, category, and stock quantity.
+    """
+    response = requests.get(book_url)
+    soup = BeautifulSoup(response.content, "html.parser")
+
+    # Extract description
+    description_elem = soup.find("div", id="product_description")
+    description = (
+        description_elem.find_next("p").text if description_elem else "No description"
+    )
+
+    # Extract category
+    breadcrumb = soup.find("ul", class_="breadcrumb")
+    category = (
+        breadcrumb.find_all("a")[2].text
+        if breadcrumb and len(breadcrumb.find_all("a")) > 2
+        else "Unknown"
+    )
+
+    # Extract stock quantity
+    availability_elem = soup.find("p", class_="instock availability")
+    stock_text = availability_elem.text.strip() if availability_elem else ""
+    match = re.search(r"In stock \((\d+) available\)", stock_text)
+    stock_quantity = int(match.group(1)) if match else 0
+
+    return description, category, stock_quantity
+
+
+def scrape_book_page(url: str) -> List[Dict[str, Union[str, float, int]]]:
+    """
+    Scrape details for books on a single page.
+
+    Args:
+        url (str): The URL of the page to scrape.
+
+    Returns:
+        list[dict[str, str | float | int]]: A list of dictionaries containing book details.
+    """
+    response = requests.get(url)
+    soup = BeautifulSoup(response.content, "html.parser")
+
+    books = []
+    book_elements = soup.find_all("article", class_="product_pod")
+
+    for book in book_elements:
+        # Basic book information
+        title = book.h3.a["title"]
+        price = book.find("p", class_="price_color").text[1:]  # Remove £ symbol
+
+        # Get star rating
+        star_class = book.find("p", class_="star-rating")["class"][1]
+        rating_map = {"One": 1, "Two": 2, "Three": 3, "Four": 4, "Five": 5}
+        star_rating = rating_map.get(star_class, 0)
+
+        # Get availability
+        availability = book.find("p", class_="instock availability").text.strip()
+
+        # Get book page URL to scrape more details
+        book_page_url = "https://books.toscrape.com/catalogue/" + book.h3.a[
+            "href"
+        ].replace("../", "")
+
+        # Scrape additional details
+        description, category, quantity = scrape_book_details(book_page_url)
+
+        books.append(
+            {
+                "title": title,
+                "price": float(price),
+                "star_rating": star_rating,
+                "availability": availability,
+                "description": description,
+                "category": category,
+                "quantity": quantity,
+            }
+        )
+
+    return books
+
+
+def scrape_all_books(
+    base_url: str, page_url_template: str
+) -> List[Dict[str, Union[str, float, int]]]:
+    """
+    Scrape books from all pages.
+
+    Args:
+        base_url (str): The base URL of the website.
+        page_url_template (str): The URL template for pagination.
+
+    Returns:
+        list[dict[str, str | float | int]]: A list of dictionaries containing all scraped book details.
+    """
+    total_pages, total_products = get_total_pages_and_products(base_url)
+
+    print(f"Total Products: {total_products}")
+    print(f"Total Pages: {total_pages}")
+
+    all_books = []
+
+    for page_num in range(1, total_pages + 1):
+        url = page_url_template.format(page_num)
+        print(f"Scraping page {page_num}")
+
+        try:
+            page_books = scrape_book_page(url)
+            all_books.extend(page_books)
+            time.sleep(0.1)  # Polite scraping: add a delay between requests
+        except Exception as e:
+            print(f"Error scraping page {page_num}: {e}")
+
+    return all_books
+
+
+def save_to_sqlite(data: List[Dict[str, Union[str, float, int]]], db_path: str) -> None:
+    """
+    Save the scraped data to an SQLite database.
+
+    Args:
+        data (list[dict[str, str | float | int]]): The data to save.
+        db_path (str): The path to the SQLite database file.
+    """
+    conn = sqlite3.connect(db_path)
+    cursor = conn.cursor()
+
+    # Create table
+    cursor.execute("""
+        CREATE TABLE IF NOT EXISTS books (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            title TEXT NOT NULL,
+            price REAL NOT NULL,
+            star_rating INTEGER NOT NULL,
+            availability TEXT NOT NULL,
+            description TEXT NOT NULL,
+            category TEXT NOT NULL,
+            quantity INTEGER NOT NULL
+        )
+    """)
+    # ```sql\nCREATE TABLE products (\n\tmain_category TEXT, \n\ttitle TEXT, \n\taverage_rating FLOAT, \n\trating_number BIGINT, \n\tfeatures TEXT, \n\tdescription TEXT, \n\tprice TEXT, \n\tstore TEXT, \n\tcategories TEXT, \n\tdetails TEXT, \n\tparent_asin TEXT\n)```
+
+    # Insert data
+    for book in data:
+        cursor.execute(
+            """
+            INSERT INTO books (title, price, star_rating, availability, description, category, quantity)
+            VALUES (?, ?, ?, ?, ?, ?, ?)
+        """,
+            (
+                book["title"],
+                book["price"],
+                book["star_rating"],
+                book["availability"],
+                book["description"],
+                book["category"],
+                book["quantity"],
+            ),
+        )
+
+    conn.commit()
+    conn.close()
+    print(f"Data saved to {db_path}")
+
+
+def main() -> None:
+    base_url = "https://books.toscrape.com/index.html"
+    page_url_template = "https://books.toscrape.com/catalogue/page-{0}.html"
+
+    books_data = scrape_all_books(base_url, page_url_template)
+
+    save_to_sqlite(books_data, "data/books_data.db")
+
+    print(f"Scraped {len(books_data)} books. Data saved to books.db")
+
+
+if __name__ == "__main__":
+    main()

src/tools.py

@@ -0,0 +1,25 @@
+from src import utils
+
+books_dataset_tools = [
+    {
+        "type": "function",
+        "function": {
+            "name": "query_books_database",
+            "description": "Quries the books database using SQL. Returns `None` if no there are no entries for the provided query. The database has the following table:```sql\nCREATE TABLE IF NOT EXISTS books (\n    id INTEGER PRIMARY KEY AUTOINCREMENT,\n    title TEXT NOT NULL,\n    price REAL NOT NULL,\n    star_rating INTEGER NOT NULL,\n    availability TEXT NOT NULL,\n    description TEXT NOT NULL,\n    category TEXT NOT NULL,\n    quantity INTEGER NOT NULL\n    )\n```",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "sql_query": {
+                        "type": "string",
+                        "description": "The sql query to run against the SQLite database. Must be in the SQLite format.",
+                    }
+                },
+                "required": ["sql_query"],
+            },
+        },
+    },
+]
+
+llm_tools_map = {
+    "query_books_database": utils.query_books_database,
+}

src/utils.py

@@ -0,0 +1,132 @@
+import json
+import logging
+import os
+import sqlite3
+import time
+from contextlib import closing
+from pathlib import Path
+
+
+def query_books_database(
+    sql_query: str, db_url: str = "data/books_data.db"
+) -> list[dict]:
+    """
+    Execute a read-only SQL query on the books database and return the results.
+
+    Args:
+        sql_query (str): SQL query string to execute.
+        db_url (str): URL to the database file, defaults to "data/books_data.db".
+
+    Returns:
+        list[dict[str, str | float | int]]: A list of rows as dictionaries
+    """
+    with closing(sqlite3.connect(f"file:{db_url}?mode=ro", uri=True)) as connection:
+        connection.row_factory = lambda cursor, row: {
+            col[0]: row[i] for i, col in enumerate(cursor.description)
+        }
+        with closing(connection.cursor()) as cursor:
+            rows = cursor.execute(sql_query).fetchall()
+    return rows
+
+
+def handle_function_calls(
+    function_map: dict, response_message, thread: list
+) -> list | None:
+    """
+    Handle function tool calls and map them to actual function executions.
+
+    Arguments:
+        function_map (dict): A dictionary mapping function names to function objects.
+        response_message: The message containing tool call information.
+        thread (list): List to append results of function calls.
+
+    Returns:
+        list: Updated list of messages.
+
+    Raises:
+        ValueError: If no tool calls are present in the response message.
+        KeyError: If a function mapping is not found.
+    """
+    if not response_message.tool_calls:
+        raise ValueError("No tool calls found in the response message.")
+
+    for tool_call in response_message.tool_calls:
+        function_name = tool_call.function.name
+        if function_name in function_map:
+            function_args = json.loads(tool_call.function.arguments)
+            print(f"Function arguments: {function_args}")
+
+            function_to_call = function_map[function_name]
+            try:
+                function_response = function_to_call(**function_args)
+            except Exception as e:
+                function_response = str(e)
+
+            thread.append(
+                {
+                    "tool_call_id": tool_call.id,
+                    "role": "tool",
+                    "name": function_name,
+                    "content": str(function_response),
+                }
+            )
+            return thread
+        else:
+            print(f"Function {function_name} not found.")
+            raise KeyError(f"Function {function_name} not found in function map.")
+
+
+def create_logger(logger_name: str, log_file: str, log_level: str) -> logging.Logger:
+    """
+    Create and configure a logger with specified name, log file, and log level.
+
+    Arguments:
+        logger_name (str): Name of the logger.
+        log_file (str): Path to the log file.
+        log_level (str): Logging level as a string (e.g., 'INFO', 'DEBUG').
+
+    Returns:
+        logging.Logger: Configured logger object.
+    """
+    LOG_FORMAT = "[%(asctime)s | %(name)s | %(levelname)s | %(funcName)s | %(message)s]"
+    log_level = getattr(logging, log_level.upper())
+
+    logger = logging.getLogger(logger_name)
+    logger.setLevel(logging.DEBUG)
+
+    file_handler = logging.FileHandler(log_file)
+    file_handler.setLevel(logging.DEBUG)
+    file_handler.setFormatter(logging.Formatter(LOG_FORMAT))
+    logger.addHandler(file_handler)
+
+    console_handler = logging.StreamHandler()
+    console_handler.setLevel(logging.INFO)
+    console_handler.setFormatter(logging.Formatter(LOG_FORMAT))
+    logger.addHandler(console_handler)
+
+    return logger
+
+
+def log_execution_time(logger: logging.Logger):
+    """
+    Decorator factory to log the execution time of a function using a specified logger.
+
+    Arguments:
+        logger (logging.Logger): Logger object used for logging.
+
+    Returns:
+        function: A decorator that logs the execution time of the wrapped function.
+    """
+
+    def decorator(func):
+        def wrapper(*args, **kwargs):
+            start_time = time.time()
+            result = func(*args, **kwargs)
+            end_time = time.time()
+            execution_time = end_time - start_time
+            logger.info(f"Executing {func.__name__} took {execution_time:.4f} seconds")
+            return result
+
+        return wrapper
+
+    return decorator