Internals of building a Customer Sentiment Intelligence with SerpAPI + System One

Source Code

Here's the link to the opensource code - serpapi-customer-sentiment

Introduction

Manual review triage is a critical bottleneck. As review volumes grow daily, high-priority signals like churn risks and serious complaints routinely get buried under a mountain of noise. Without automation, feedback remains fragmented across multiple platforms, and escalations surface only after customers are already lost.

Customer Sentiment Intelligence (CSI) solves this by automating the entire "listen-score-act" loop. Built as a lightweight Python reference implementation, CSI aggregates public data like Yelp reviews and Google News into a single command. It instantly normalizes, deduplicates, and scores each item across seven dimensions, using an explainable rule engine to route data into four clear business actions: DISCARD, QUEUE, ANALYZE, or ESCALATE.

Stack: Python 3.11+ · FastAPI · SerpAPI (Yelp Search, Yelp Reviews, Google News) · TypeSafe Jev / System One (or Laya gateway) · optional OpenAI-compatible LLM · Pydantic validation · pytest suite.

In this post I'll walk through the architecture, the scoring contract, the decision engine, and the exact code - so you can run it, extend it, or steal the pattern for your own triage problem.

1. The problem

Customer Experience (CX) teams doing manual review hit four walls:

1. Noisy data - The vast majority of reviews and articles are low-relevance or redundant, yet every item still has to be read and judged by hand.
2. Inconsistent triage - Relevance, sentiment, severity, and churn risk are assessed subjectively. Priorities differ per person, per day.
3. Late escalation - Serious complaints surface after customers are already lost, or land on the wrong team without shared context.
4. No structured record - Feedback lives in many formats and places, so sentiment trends never feed product, support, or leadership decisions.

The fix isn't a dashboard. It's a pipeline with four properties: collect without scrapers, normalize into one schema, score deterministically, route explainably.

2. Architecture: five stages, one payload

The pipeline (backend/app/pipeline.py) runs four steps per invocation:

class CustomerSentimentPipeline:
    def run(self, company, source="yelp", location=None, search=None,
            reviews=None, include_news=True, analysis=None,
            gl="us", hl="en"):
        evidence = self.discover(
            company, source, location, search, reviews,
            include_news, gl, hl
        )
        results = []

        for item in evidence:
            try:
                system_one_result = self.system_one.evaluate(
                    item, analysis=analysis
                )
                decision = make_decision(system_one_result)

                record = {
                    "evidence": item,
                    "jev": system_one_result,   # stable output key,
                    "decision": decision,       # whichever backend scored it
                }

                if decision["decision"] in {"ANALYZE", "ESCALATE"}:
                    record["deep_analysis"] = analyze(
                        item, system_one_result, self.settings
                    )

                results.append(record)

            except Exception as exc:
                results.append({
                    "evidence": item,
                    "error": str(exc),
                    "decision": {
                        "decision": "QUEUE",
                        "reason": "Processing error; manual review required",
                    },
                })

        return {
            "company": company,
            "evidence_count": len(evidence),
            "summary": self._summary(results),
            "results": results,
        }

Three design decisions worth noting:

• Per-item failure isolation. Each item is evaluated inside its own try/except. One bad item can't kill the run - it fails safe into QUEUE with a reason. Nothing silently disappears.
• Backend-agnostic scoring. SYSTEM_ONE_CLIENTS = {"jev": JevClient, "laya": LayaClient} - both expose the same evaluate(evidence, analysis=...) contract, and the output key is always jev regardless of which engine scored it. Swapping backends is a one-line .env change (SYSTEM_ONE_PROVIDER=laya), not a code change.
• LLM spend is gated. analyze() only runs for ANALYZE/ESCALATE items. Without OPENAI_ADMIN_KEY it returns prompt_only - fully usable, zero cost.

3. Stage 1 - Collect: SerpAPI, no scrapers

Discovery (discover()) does two SerpAPI calls. First, find_place() hits engine=yelp with the search term and a structured find_loc ("City, State, Country" built from the location object), scans organic_results, and takes the first place_ids[0]. Then reviews() hits engine=yelp_reviews with that place_id, paginating up to the requested limit (49 per page, sort_by mapped to Yelp's options). Optionally, news() hits engine=google_news for the latest coverage of the company.

# CLI - same pipeline, stdout or file
python -m app.main --company "The Plaza" --city "New York" --state NY --country US
python -m app.main --company "The Plaza" --city "New York" --state NY --country US \
  --review-limit 50 --sort-by newest --output output.json
python -m app.main --company "The Plaza" --city "New York" --state NY --country US --no-news

Key detail: Yelp Search requires a location, so when the structured location is empty the pipeline falls back to the gl country name (e.g. us - United States). No location, no place_id, no reviews - fail fast with a clear RuntimeError, not an empty payload.

4. Stage 2 - Normalize and deduplicate: the evidence layer

Every review and article becomes one record in a shared schema (backend/app/evidence.py):

import hashlib

def _id(text):
    return hashlib.sha256(text.encode("utf-8")).hexdigest()[:16]

def normalize_review(company, review):
    comment = review.get("comment") or {}
    text = comment.get("text") or review.get("text") or ""
    return {
        "evidence_id": _id(f"{company}|yelp_review|{text}|{review.get('date')}"),
        "company": company,
        "source": "yelp",
        "source_type": "customer_review",
        "published": review.get("date"),
        "rating": review.get("rating"),
        "author": (review.get("user") or {}).get("name"),
        "text": text,
        "url": review.get("link") or (review.get("user") or {}).get("link"),
    }

def deduplicate(evidence):
    seen, result = set(), []
    for item in evidence:
        if item["evidence_id"] not in seen and item.get("text"):
            seen.add(item["evidence_id"])
            result.append(item)
    return result

Deterministic sha256 IDs (truncated to 16 hex chars) make dedup cheap and repeatable across runs - the same review always maps to the same ID, so re-runs and overlapping news/review fetches never double-count. Empty-text items are rejected, so each real signal is analyzed exactly once. Nine canonical fields, no surprises downstream.

5. Stage 3 - Score: seven typed questions, one model call

This is the heart of the system. Each evidence item is sent to Jev / System One (POST {base}/v1/systemone, 45s timeout, Bearer key) with a state block (company, source, rating, text, URL) and seven fixed questions (JevClient.questions() in backend/app/jev_client.py):

#	Question	Type	What it returns
1	relevant	noul	0–1 confidence the item concerns customer experience
2	sentiment	choice	positive / neutral / negative + confidence + full probability distribution
3	intent	choice	praise, complaint, question, request, recommendation, warning, general_feedback
4	topic	choice	product, pricing, support, delivery, staff, quality, technology, billing, other
5	severity	score	0–4 graded scale with a legend (No issue - Critical issue)
6	requires_escalation	noul	0–1 confidence a human needs to look
7	churn_risk	score	0–4 graded scale (No churn signal - Very strong churn signal)

Three things I like about this contract:

• Typed answers, not free text. noul (yes/no + confidence), choice (pick + per-option probabilities), score (graded + legend). The downstream decision engine never parses textual content - it reads numbers. That's what makes the routing deterministic.
• relevant is always on. The decision engine requires it, so the analysis toggles can disable the other six per request (send {"sentiment": false, ...} to skip dimensions and save time/money) but never relevance.
• One call, seven answers. The full question set rides in a single {"model", "state", "questions"} POST. No N+1 fan-out, no stitching partial results.

6. Stage 4 - Decide: 12 lines that replace a triage meeting

The entire routing engine is make_decision() in backend/app/decision.py - ordered rules, first match wins:

def make_decision(jev_result):
    answers = jev_result.get("answers", jev_result)

    relevant = float(_answer(answers, "relevant").get("noul", 0.0))
    sentiment_conf = float(_answer(answers, "sentiment").get("confidence", 0.0))
    escalation = float(_answer(answers, "requires_escalation").get("noul", 0.0))

    churn = _answer(answers, "churn_risk").get("score")
    churn = float(churn) if churn is not None else 0.0

    if relevant < 0.60:
        return {"decision": "DISCARD", "reason": "Low customer relevance"}

    if sentiment_conf and sentiment_conf < 0.70:
        return {"decision": "QUEUE", "reason": "Low sentiment confidence"}

    if escalation >= 0.80 or churn >= 3.0:
        return {"decision": "ESCALATE", "reason": "Material customer risk signal"}

    return {"decision": "ANALYZE", "reason": "Relevant customer signal"}

Decision	Trigger	Meaning
DISCARD	relevance < 0.60	Not materially about CX - drop it
QUEUE	sentiment confidence < 0.70, or any processing error	Ambiguous - human review required
ANALYZE	relevant, confident, no strong risk signal	Feed to LLM deep analysis for insight
ESCALATE	escalation ≥ 0.80 or churn risk ≥ 3.0	Material risk - immediate attention

Why this ordering matters: relevance gates everything (don't score what isn't about you), confidence gates actionability (don't act on guesses), and risk short-circuits to escalation before the default ANALYZE path. Every branch returns a human-readable reason, so the routing is auditable - tune the four thresholds in one place to match your risk appetite.

7. Stage 5 - Deep analysis: spend the LLM only where it pays

analyze() in backend/app/deep_analysis.py builds a prompt from the evidence + Jev output and asks for seven structured fields: customer_issue, observed_sentiment, likely_root_cause, customer_impact, churn_implication, recommended_business_response, product_or_service_opportunity - with explicit instructions to separate observed facts from hypotheses and never invent unsupported facts. Temperature 0.1, 60s timeout, any OpenAI-compatible endpoint via OPENAI_BASE_URL.

The cost control is the point: in a typical 120-item run, ~98 items DISCARD, ~4 QUEUE, and only ~18 reach the LLM. You pay for insight on the 15% that matters, not tokens on the 85% that doesn't.

8. The interface: CLI, HTTP, and the payload contract

The FastAPI layer (backend/app/api.py) is thin by design - Pydantic validation, then straight into the pipeline:

@app.post("/customer-sentiment/analyze")
def analyze(request: SentimentRequest):
    try:
        settings = Settings.from_env()
        settings.validate()
        return CustomerSentimentPipeline(settings).run(
            company=request.company,
            source=request.source,
            location=request.location.model_dump() if request.location else None,
            search=request.search.model_dump(exclude_none=True),
            reviews=request.reviews.model_dump(exclude_none=True),
            include_news=request.include_news,
            analysis=request.analysis.model_dump(exclude_none=True),
            gl=request.gl,
            hl=request.hl,
        )
    except Exception as exc:
        raise HTTPException(status_code=500, detail=str(exc)) from exc

# run it
uvicorn app.api:app --reload   # from backend/
curl -X POST localhost:8000/customer-sentiment/analyze \
  -H 'Content-Type: application/json' \
  -d '{"company": "The Plaza", "location": {"city": "New York", "state": "NY", "country": "US"}}'

Config lives in backend/.env: SERPAPI_KEY (required), SYSTEM_ONE_PROVIDER (jev default / laya), then exactly one matching credential (TYPESAFE_API_KEY or LAYA_API_KEY) - Settings.validate() enforces this before any run. Optional: JEV_MODEL, JEV_BASE_URL, LAYA_*, OPENAI_ADMIN_KEY / OPENAI_MODEL / OPENAI_BASE_URL, CORS_ORIGINS.

Every run returns the same contract: company, evidence_count, summary (decision_counts + sentiment_counts), and results[] where each item holds evidence, jev (answers), decision (with reason), and deep_analysis for ANALYZE/ESCALATE - or an error record that still carries a QUEUE decision.

9. Extending it: four starting points

1. More sources. Add X/Twitter, Reddit, or app-store reviews in serpapi_client.py + evidence.py - the normalizer/dedup pattern already handles heterogeneous inputs.
2. Tune the thresholds. The four numbers in decision.py (0.60 / 0.70 / 0.80 / 3.0) encode a risk appetite. Calibrate against labeled data for your domain.
3. Point the LLM at a real endpoint. Set the OPENAI_* vars to swap prompt_only briefs for full structured responses.
4. Schedule it. Cron the CLI or poll the API on a timer and you've got a sentiment-tracking dashboard with trend history.

The test suite (backend/tests/: pipeline, decision, evidence, clients, CORS) runs under pytest, so extensions land with a safety net.

---

Key Insights

• Pipelines are simply ordered steps that transform messy input into usable output.
• APIs simplify data collection and reduce fragility compared to scraping.
• Normalizing data upfront makes downstream analysis smoother.
• AI can be framed as structured Q&A rather than opaque predictions.
• Smart categorization helps teams focus on urgent or actionable feedback.
• Cost‑effective design ensures scalability without overspending.