	In Browser
	StumbleUpon
	del.icio.us
	Google
	Google Buzz
	reddit
	LinkedIn

	Facebook
	Twitter
	Linkedin
	E-Mail

Generative AI > Large Language Models > AI Sentiment Analysis Pipeline for Product Reviews

AI Sentiment Analysis Pipeline for Product Reviews

Author: Venkata Sudhakar

Reading every customer review manually does not scale. When your e-commerce store receives 500 reviews per day across 200 products, your team cannot read them all - let alone spot trends. An AI sentiment analysis pipeline processes every review automatically, classifying sentiment, extracting the key topics customers mentioned, and flagging urgent issues like safety concerns or repeated defects. The output feeds directly into a business dashboard that product managers and customer service teams check each morning - no review goes unread, no urgent complaint gets missed.

The pipeline pattern is straightforward: batch the reviews, send each to the LLM with a structured extraction prompt, collect the results, and aggregate into summary statistics. Using structured output (either Pydantic or function calling) ensures each review produces a clean Python object with consistent fields - sentiment score, topic tags, urgency flag, and a one-line summary. Aggregating these objects into counts and averages takes a few lines of standard Python, giving you a product-level summary that would take a human analyst hours to produce.

The below example processes a batch of washing machine reviews, extracts structured sentiment data from each, and produces a product performance summary that could feed straight into a business intelligence dashboard.

from openai import OpenAI
from pydantic import BaseModel
from typing import Literal
import json

client = OpenAI(api_key="your-api-key")

class ReviewAnalysis(BaseModel):
    sentiment: Literal["POSITIVE", "NEUTRAL", "NEGATIVE"]
    score: int                  # 1-5 stars equivalent
    topics: list               # e.g. ["noise", "delivery", "build quality"]
    urgent: bool               # True if safety issue or serious defect
    one_line: str              # one sentence summary for dashboard

EXTRACT_SCHEMA = {
    "name": "analyse_review",
    "description": "Extract structured sentiment data from a product review",
    "parameters": {
        "type": "object",
        "properties": {
            "sentiment": {"type": "string", "enum": ["POSITIVE", "NEUTRAL", "NEGATIVE"]},
            "score":    {"type": "integer", "description": "1-5 star rating implied by the review"},
            "topics":   {"type": "array", "items": {"type": "string"},
                         "description": "Up to 4 topics the customer mentioned"},
            "urgent":   {"type": "boolean",
                         "description": "True if safety issue, serious defect or legal risk"},
            "one_line": {"type": "string", "description": "One sentence dashboard summary"}
        },
        "required": ["sentiment", "score", "topics", "urgent", "one_line"]
    }
}

def analyse_review(review_text: str) -> dict:
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "You analyse product reviews for an e-commerce platform. Extract structured data accurately."},
            {"role": "user",   "content": review_text}
        ],
        tools=[{"type": "function", "function": EXTRACT_SCHEMA}],
        tool_choice={"type": "function", "function": {"name": "analyse_review"}},
        temperature=0
    )
    return json.loads(resp.choices[0].message.tool_calls[0].function.arguments)

Processing a batch of washing machine reviews and generating a product summary,

reviews = [
    "Absolutely love this machine! Clothes come out perfectly clean and it is whisper quiet. Delivery was also faster than expected. 5 stars.",
    "The machine started making a grinding noise after just 3 washes. Quite worried about it. Called support but they put me on hold for 45 minutes.",
    "Decent machine for the price. Nothing special but does the job. Took 10 days to deliver which was a bit slow.",
    "WARNING: The door seal is leaking water onto the floor. This is a safety hazard. Returning immediately and reporting to consumer court.",
    "Second purchase from this brand. As reliable as ever. Energy efficient and the quick wash cycle is brilliant for school uniforms.",
]

results = [analyse_review(r) for r in reviews]

# Aggregate into product summary
total = len(results)
sentiment_counts = {"POSITIVE": 0, "NEUTRAL": 0, "NEGATIVE": 0}
all_topics = []
urgent_flags = []
avg_score = 0

for i, r in enumerate(results):
    sentiment_counts[r["sentiment"]] += 1
    all_topics.extend(r["topics"])
    avg_score += r["score"]
    if r["urgent"]:
        urgent_flags.append({"review_index": i+1, "summary": r["one_line"]})

avg_score = round(avg_score / total, 1)
top_topics = sorted(set(all_topics), key=lambda t: all_topics.count(t), reverse=True)[:5]

print("=== DAILY PRODUCT DASHBOARD: WashPro 8000 ===")
print("Reviews processed today:", total)
print("Average score:          ", avg_score, "/ 5")
print("Positive / Neutral / Negative:", sentiment_counts["POSITIVE"],
      "/", sentiment_counts["NEUTRAL"], "/", sentiment_counts["NEGATIVE"])
print("Top topics:", ", ".join(top_topics))
if urgent_flags:
    print("\nURGENT FLAGS - Immediate action required:")
    for flag in urgent_flags:
        print("  Review", flag["review_index"], "-", flag["summary"])
else:
    print("\nNo urgent flags today.")

It gives the following output,

=== DAILY PRODUCT DASHBOARD: WashPro 8000 ===
Reviews processed today: 5
Average score:           3.2 / 5
Positive / Neutral / Negative: 2 / 1 / 2
Top topics: noise, delivery, safety, build quality, energy efficiency

URGENT FLAGS - Immediate action required:
  Review 2 - Customer reports grinding noise after 3 washes, concerns about defect.
  Review 4 - Door seal leaking water; customer citing safety hazard and legal action.

# At 500 reviews per day this runs in under 2 minutes
# Product team sees the dashboard at 9am without reading a single review manually
# Urgent flags trigger an immediate alert to the quality team

In production, run this pipeline as a nightly scheduled job that pulls the previous day reviews from your database, processes them in parallel batches of 20 using asyncio and the async OpenAI client, and writes the results to your data warehouse. The dashboard then queries the warehouse for the aggregated metrics. Urgent flag emails go out in real time as each batch completes, not just at end of day. At 500 reviews and gpt-4o-mini pricing, the daily processing cost is under Rs 5 - far cheaper than one hour of a human analyst.

Send your comments, suggestions or queries regarding this site to [email protected].