Data Labeling w/ LLMs

This is part 1/4.

Overview

In this tutorial, you’ll learn how to:

Set up xorq and configure the necessary components
Fetch data from the HackerNews API
Use OpenAI gpt-3.5-turbo model to automatically label data with sentiment analysis
Create a labeled dataset ready for future machine learning tasks

Prerequisites

Make sure to set your OpenAI API key in your environment:

export OPENAI_API_KEY=your_api_key

Installation and Imports

First, install xorq and the required dependencies:

pip install xorq pandas

Then import the necessary modules:

import pandas as pd
import xorq as xo
import xorq.expr.datatypes as dt

from xorq.caching import ParquetStorage
from xorq.common.utils.import_utils import import_python

m = import_python(xo.options.pins.get_path("hackernews_lib"))
o = import_python(xo.options.pins.get_path("openai_lib"))

The imported modules m (hackernews_lib) and o (openai_lib) contain utility functions for:

Connecting to the HackerNews Firebase API
Fetching and processing HackerNews stories
Making calls to OpenAI’s API for sentiment analysis
Processing the response into structured data

You’ll need to ensure these files are accessible in your environment or create them based on the code snippets in the Appendix.

Defining the HackerNews Fetcher

We’ll define a User-Defined Exchanger Function (UDXF) that fetches HackerNews stories:

do_hackernews_fetcher_udxf = xo.expr.relations.flight_udxf(
    process_df=m.get_hackernews_stories_batch,
    maybe_schema_in=m.schema_in.to_pyarrow(),
    maybe_schema_out=m.schema_out.to_pyarrow(),
    name="HackerNewsFetcher",
)

Setting Up the Backend and Storage

Let’s initialize the xorq backend and storage:

The below code will attempt to download ~100k items from HackerNew API that can take a long time. If you want to just run the tutorial with a smaller data, change the variable name of the code below to "hn-fetcher-input-small"

name = "hn-fetcher-input-large" # or use hn-fercher-input-small to avoid downloading all data
con = xo.connect()
storage = ParquetStorage(source=con)

Building the Data Pipeline

Now, let’s set up our complete data pipeline:

# Start by reading the input for the show HN
raw_expr = (
    xo.deferred_read_parquet(
        con,
        xo.options.pins.get_path(name), # this fetches a DataFrame with two columns; maxitem and n 
        name,
    )
    # Pipe into the HackerNews fetcher to get the full stories
    .pipe(m.do_hackernews_fetcher_udxf)
)

# Build complete pipeline with filtering, labeling, and caching
t = (
    raw_expr
    # Filter stories with text
    .filter(xo._.text.notnull())
    # Apply model-assisted labeling with OpenAI
    .pipe(o.do_hackernews_sentiment_udxf, con=con)
    # Cache the labeled data to Parquet
    .cache(storage=ParquetStorage(con))
    # Filter out any labeling errors
    .filter(~xo._.sentiment.contains("ERROR"))
    # Convert sentiment strings to integer codes (useful for future ML tasks)
    .mutate(
        sentiment_int=xo._.sentiment.cases(
            {"POSITIVE": 2, "NEUTRAL": 1, "NEGATIVE": 0}.items()
        ).cast(int)
    )
)

Execute and Inspect the Labeled Data

Now let’s execute the pipeline to get our labeled DataFrame:

# Execute the pipeline and get the final DataFrame
labeled_df = t.execute()

# Inspect the results
print(labeled_df[["id", "title", "sentiment", "sentiment_int"]].head())

This will output something like:

         id                                              title sentiment  sentiment_int
43083439  Show HN: Xenoeye – high performance network tr...  POSITIVE              2
43083558                   Toronto 'Plane Crash Submissions  NEGATIVE              0
43083656  Show HN: Generic and variadic printing library...  POSITIVE              2
43083755  Show HN: WebMorph – Automate Your Website Tran...  POSITIVE              2
43083845                 Ask HN: Small Ideas vs. Big Ideas?  POSITIVE              2

Summary

Congratulations! You’ve now:

Set up xorq
Fetched data from the HackerNews API
Set up local caching with ParquetStorage
Used OpenAI GPT to automatically label the data with sentiment analysis
Created a labeled dataset ready for future machine learning tasks

Next Steps

With this labeled dataset, you can now proceed to:

Split the data into train/test sets for model development
Apply text preprocessing and feature extraction techniques
Train and evaluate various machine learning models
Perform data analysis to gain insights about sentiment patterns in HackerNews stories

Appendix

Helper Modules Structure

import functools
import json
import pathlib

import pandas as pd
import requests
import toolz

import xorq as xo


base_api_url = "https://hacker-news.firebaseio.com/v0"


@toolz.curry
def simple_disk_cache(f, cache_dir, serde):
    cache_dir.mkdir(parents=True, exist_ok=True)

    def wrapped(**kwargs):
        name = ",".join(f"{key}={value}" for key, value in kwargs.items())
        path = cache_dir.joinpath(name)
        if path.exists():
            value = serde.loads(path.read_text())
        else:
            value = f(**kwargs)
            path.write_text(serde.dumps(value))
        return value

    return wrapped


def get_json(url):
    resp = requests.get(url)
    resp.raise_for_status()
    return resp.json()


@simple_disk_cache(cache_dir=pathlib.Path("./hackernews-items"), serde=json)
def get_hackernews_item(*, item_id):
    return get_json(f"{base_api_url}/item/{item_id}.json")


@functools.cache
def get_hackernews_maxitem():
    return get_json(f"{base_api_url}/maxitem.json")


def get_hackernews_stories(maxitem, n):
    gen = (
        toolz.excepts(requests.exceptions.SSLError, get_hackernews_item)(
            item_id=item_id
        )
        for item_id in range(maxitem - n, maxitem)
    )
    gen = filter(None, gen)
    df = pd.DataFrame(gen).reindex(columns=schema_out)
    return df


@toolz.curry
def get_hackernews_stories_batch(df, filter=slice(None)):
    series = df.apply(lambda row: get_hackernews_stories(**row), axis=1)
    return pd.concat(series.values, ignore_index=True).loc[filter].reset_index(drop=True)


schema_in = xo.schema({"maxitem": int, "n": int})
schema_out = xo.schema(
    {
        "by": "string",
        "id": "int64",
        "parent": "float64",
        "text": "string",
        "time": "int64",
        "type": "string",
        "kids": "array<int64>",
        "descendants": "float64",
        "score": "float64",
        "title": "string",
        "url": "string",
    }
)


do_hackernews_fetcher_udxf = xo.expr.relations.flight_udxf(
    process_df=get_hackernews_stories_batch(
        filter=lambda t: t.type.eq("story") & t.title.notnull()
    ),
    maybe_schema_in=schema_in.to_pyarrow(),
    maybe_schema_out=schema_out.to_pyarrow(),
    name="HackerNewsFetcher",
)

import functools
import operator
import os
from urllib.parse import unquote_plus

import pandas as pd
import toolz
from openai import OpenAI
from tenacity import (
    retry,
    stop_after_attempt,
    wait_random_exponential,
)

import xorq as xo
from xorq.flight.utils import (
    schema_concat,
    schema_contains,
)


@functools.cache
def get_client():
    client = OpenAI(
        api_key=os.environ["OPENAI_API_KEY"],
    )
    return client


request_timeout = 3


@functools.cache
def extract_sentiment(text):
    @retry(wait=wait_random_exponential(min=1, max=60), stop=stop_after_attempt(3))
    def completion_with_backoff(**kwargs):
        return get_client().chat.completions.create(**kwargs)

    if text == "":
        return "NEUTRAL"
    messages = [
        {
            "role": "system",
            "content": "You are an AI language model trained to analyze and detect the sentiment of hackernews forum comments.",
        },
        {
            "role": "user",
            "content": f"Analyze the following hackernews comment and determine if the sentiment is: positive, negative or neutral. "
            f"Return only a single word, either POSITIVE, NEGATIVE or NEUTRAL: {text}",
        },
    ]
    try:
        response = completion_with_backoff(
            model="gpt-3.5-turbo",
            messages=messages,
            max_tokens=30,
            temperature=0,
            timeout=request_timeout,
        )
        return response.choices[0].message.content
    except Exception as e:
        return f"ERROR: {e}"


@toolz.curry
def get_hackernews_sentiment_batch(df: pd.DataFrame, input_col, append_col):
    import concurrent.futures
    with concurrent.futures.ThreadPoolExecutor() as executor:
        values = tuple(executor.map(toolz.compose(extract_sentiment, unquote_plus), df[input_col]))
    return df.assign(**{append_col: values})


input_col = "text"
append_col = "sentiment"
schema_requirement = xo.schema({input_col: "str"})
schema_append = xo.schema({append_col: "str"})
maybe_schema_in = toolz.compose(schema_contains(schema_requirement), xo.schema)
maybe_schema_out = toolz.compose(
    operator.methodcaller("to_pyarrow"),
    schema_concat(to_concat=schema_append),
    xo.Schema.from_pyarrow,
)


do_hackernews_sentiment_udxf = xo.expr.relations.flight_udxf(
    process_df=get_hackernews_sentiment_batch(
        input_col=input_col, append_col=append_col
    ),
    maybe_schema_in=maybe_schema_in,
    maybe_schema_out=maybe_schema_out,
    name="HackerNewsSentimentAnalyzer",
)

User Guide

Core Concepts

Vignettes

Tutorials

Overview

Prerequisites

Installation and Imports

Defining the HackerNews Fetcher

Setting Up the Backend and Storage

Building the Data Pipeline

Execute and Inspect the Labeled Data

Summary

Next Steps

Further Reading

Appendix

Helper Modules Structure

Troubleshooting

User Guide

Core Concepts

Vignettes

Tutorials

​Overview

​Prerequisites

​Installation and Imports

​Defining the HackerNews Fetcher

​Setting Up the Backend and Storage

​Building the Data Pipeline

​Execute and Inspect the Labeled Data

​Summary

​Next Steps

​Further Reading

​Appendix

​Helper Modules Structure

​Troubleshooting

Overview

Prerequisites

Installation and Imports

Defining the HackerNews Fetcher

Setting Up the Backend and Storage

Building the Data Pipeline

Execute and Inspect the Labeled Data

Summary

Next Steps

Further Reading

Appendix

Helper Modules Structure

Troubleshooting