Real-time price recommendation

A practical example of creating a pipeline for real-time recommendation system using AI.

You will build a real-time price recommendation data pipeline to process car price data. The pipeline enriches data with predicted future prices using AI for the new vehicle registered in the primary PostgreSQL database.

Real-time car price recommendation pipeline with GlassFlow

Link to the GitHub project repository

Setting Up the Pipeline with GlassFlow

You will use the GlassFlow WebApp to create a data processing pipeline.

Prerequisites

To start with this setup, you need a free GlassFlow account.

Sign up for a free account

Step 1. Log in to GlassFlow WebApp

Navigate to the GlassFlow WebApp and log in with your credentials.

Step 2. Create a New Pipeline

Click on "Create New Pipeline" and provide a name. You can name it "Price recommendation".

Step 3. Configure a Data Source

Select "SDK" to configure the pipeline to use Python SDK to ingest price change event data from a source like PostgreSQL.

Step 4. Define the Transformer

AI-powered transformation function in Python predicts future car prices using LLMs(Large Language Models) like GPT-3.5-turbo from OpenAI. Create an API key and set the API key

openai.api_key="{REPLACE_WITH_YOUR_OPENAI_API_KEY}"

in the transformation code below. Paste the updated transformation function code into the transformer's built-in editor.

https://github.com/glassflow/glassflow-examples/blob/main/use-cases/predict-car-price/transform.py

import openai
import json

openai.api_key = "{REPLACE_WITH_YOUR_OPENAI_API_KEY}"


def handler(data, log):
    """
    Generate AI insights to predict car price data.
    """
    # log.info("Event:" + json.dumps(data), data=data)

    insights = predict_price(data)
    return json.loads(insights)


def predict_price(data):
    # Generate insights using OpenAI's chat completion endpoint
    response = openai.chat.completions.create(
        model="gpt-3.5-turbo",
        response_format={"type": "json_object"},
        messages=[
            {
                "role": "system",
                "content": "You are a great data analyst to predict car data prices.",
            },
            {
                "role": "user",
                "content": f"Analyze the following vehicle data: {data}, predict its price and return only the orginal full JSON with additional suggested_price attribute and value added",
            },
        ],
        max_tokens=500,
        temperature=0.5,
    )

    insights = response.choices[0].message.content
    return insights

Note that the handler function is mandatory to implement in your code. Without it, the transformation function will not be successful.

Step 5. Choose a transformer dependency

The transformation function uses openai external library in the code, so we need to choose it from the Dependencies dropdown menu. GlassFlow includes the library in the function deployment and runtime. Read more about Python dependencies for transformation.

Step 6. Configure a Data Sink

Select "SDK" to configure the pipeline to use Python SDK for sending data to the output file. In a real-world project, you send data to dashboards and real-time websites.

Step 7. Confirm the Pipeline

Confirm the pipeline settings in the final step and click "Create Pipeline".

Step 8. Copy the Pipeline Credentials

Once the pipeline is created, copy its credentials such as Pipeline ID and Access Token.

Send and consume data from the pipeline

Prerequisites

To complete this part you'll need the following:

• Python is installed on your machine.

• Download and Install Pip to manage project packages.

• You have an OpenAI API account.

Installation

1. Clone the glassflow-examples repository to your local machine:

   git clone https://github.com/glassflow/glassflow-examples.git

2. Navigate to the project directory:

   cd use-cases/predict-car-price

3. Create a new virtual environment:

   python -m venv .venv && source .venv/bin/activate

4. Install the required dependencies:

   pip install -r requirements.txt

Create an environment configuration file

Add a .env file in the project directory and add the following configuration variables:

PIPELINE_ID=your_pipeline_id
PIPELINE_ACCESS_TOKEN=your_pipeline_access_token

Replace your_pipeline_id and your_pipeline_access_token with appropriate values obtained from your GlassFlow account.

Run data producer

Run the producer.py script to start publishing data:

python producer.py

Run data consumer

Run the consumer.py to retrieve transformed data from the pipeline:

python consumer.py