Building a Review Response Agent: AI-Powered Reputation Management

Why Review Responses Matter at Scale

Online reviews directly impact revenue. Businesses that respond to reviews earn 35 percent more trust than those that do not, and response speed matters — customers expect replies within 24 hours. But responding to every review across Google, Yelp, G2, Trustpilot, and app stores is a full-time job. An AI review response agent monitors all platforms, analyzes sentiment, generates appropriate responses, and routes critical reviews to humans — ensuring every customer feels heard without overwhelming your team.

Review Data Model

First, define a structured representation of reviews that works across platforms. Different sources provide different metadata, so the model needs to accommodate that variation.

from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
from enum import Enum


class Platform(Enum):
    GOOGLE = "google"
    YELP = "yelp"
    G2 = "g2"
    TRUSTPILOT = "trustpilot"
    APP_STORE = "app_store"


class Sentiment(Enum):
    VERY_NEGATIVE = "very_negative"
    NEGATIVE = "negative"
    NEUTRAL = "neutral"
    POSITIVE = "positive"
    VERY_POSITIVE = "very_positive"


@dataclass
class Review:
    id: str
    platform: Platform
    author: str
    rating: int  # 1-5
    text: str
    date: datetime
    sentiment: Optional[Sentiment] = None
    topics: list[str] = field(default_factory=list)
    response: Optional[str] = None
    response_date: Optional[datetime] = None
    escalated: bool = False
    escalation_reason: Optional[str] = None

Review Collection Across Platforms

Each platform has its own API or scraping pattern. We abstract this behind a common interface.

import httpx
from abc import ABC, abstractmethod


class ReviewCollector(ABC):
    @abstractmethod
    async def fetch_new_reviews(
        self, since: datetime
    ) -> list[Review]:
        pass


class GoogleReviewCollector(ReviewCollector):
    def __init__(self, place_id: str, api_key: str):
        self.place_id = place_id
        self.api_key = api_key

    async def fetch_new_reviews(
        self, since: datetime
    ) -> list[Review]:
        async with httpx.AsyncClient() as client:
            resp = await client.get(
                "https://maps.googleapis.com/maps/api/place/details/json",
                params={
                    "place_id": self.place_id,
                    "fields": "reviews",
                    "key": self.api_key,
                },
            )
            data = resp.json()

        reviews = []
        for r in data.get("result", {}).get("reviews", []):
            review_time = datetime.fromtimestamp(r["time"])
            if review_time > since:
                reviews.append(Review(
                    id=f"google-{r['time']}-{r['author_name'][:10]}",
                    platform=Platform.GOOGLE,
                    author=r["author_name"],
                    rating=r["rating"],
                    text=r.get("text", ""),
                    date=review_time,
                ))
        return reviews


class MultiPlatformCollector:
    def __init__(self, collectors: list[ReviewCollector]):
        self.collectors = collectors

    async def fetch_all(self, since: datetime) -> list[Review]:
        all_reviews = []
        for collector in self.collectors:
            try:
                reviews = await collector.fetch_new_reviews(since)
                all_reviews.extend(reviews)
            except Exception as e:
                print(f"Error collecting from {type(collector).__name__}: {e}")
        return sorted(all_reviews, key=lambda r: r.date, reverse=True)

Sentiment Analysis and Topic Extraction

Before generating a response, the agent analyzes the review to understand the sentiment intensity and specific topics mentioned. This drives both the response tone and the escalation decision.

from openai import AsyncOpenAI
import json

client = AsyncOpenAI()

ANALYSIS_PROMPT = """Analyze this customer review.

Platform: {platform}
Rating: {rating}/5
Review text: {text}

Return JSON with:
- "sentiment": one of "very_negative", "negative", "neutral", "positive", "very_positive"
- "topics": list of specific topics mentioned (e.g., "shipping speed", "product quality", "customer service")
- "key_complaint": the main issue if negative, or null if positive
- "urgency": "low", "medium", "high" — high if the reviewer mentions legal action, health/safety, or threatens to leave
- "requires_human": boolean — true if the review mentions a specific employee, legal threats, or a complex technical issue
"""


async def analyze_review(review: Review) -> Review:
    response = await client.chat.completions.create(
        model="gpt-4o",
        response_format={"type": "json_object"},
        messages=[
            {"role": "system", "content": "Return valid JSON only."},
            {
                "role": "user",
                "content": ANALYSIS_PROMPT.format(
                    platform=review.platform.value,
                    rating=review.rating,
                    text=review.text,
                ),
            },
        ],
    )
    result = json.loads(response.choices[0].message.content)
    review.sentiment = Sentiment(result["sentiment"])
    review.topics = result.get("topics", [])
    if result.get("requires_human"):
        review.escalated = True
        review.escalation_reason = result.get("key_complaint", "Requires human review")
    return review

Sentiment-Aware Response Generation

Different sentiments require different response strategies. A positive 5-star review gets a short thank-you. A negative review gets an empathetic acknowledgment with a concrete next step. The agent adapts its tone and length accordingly.

RESPONSE_STRATEGIES = {
    Sentiment.VERY_POSITIVE: {
        "tone": "warm and grateful",
        "max_words": 60,
        "template_hint": "Thank them specifically for what they praised",
    },
    Sentiment.POSITIVE: {
        "tone": "appreciative and encouraging",
        "max_words": 80,
        "template_hint": "Acknowledge their positive experience, mention a feature they might like",
    },
    Sentiment.NEUTRAL: {
        "tone": "professional and helpful",
        "max_words": 100,
        "template_hint": "Acknowledge their feedback, offer to improve their experience",
    },
    Sentiment.NEGATIVE: {
        "tone": "empathetic and solution-oriented",
        "max_words": 120,
        "template_hint": "Apologize sincerely, address the specific issue, provide a resolution path",
    },
    Sentiment.VERY_NEGATIVE: {
        "tone": "deeply empathetic, urgent, and accountable",
        "max_words": 150,
        "template_hint": "Sincere apology, take responsibility, offer direct contact for resolution",
    },
}

RESPONSE_PROMPT = """Write a response to this customer review.

Review by {author} on {platform} ({rating}/5):
"{text}"

Sentiment: {sentiment}
Topics: {topics}

Response guidelines:
- Tone: {tone}
- Maximum length: {max_words} words
- Strategy: {template_hint}
- Use the reviewer's first name if available
- Never be defensive or dismissive
- Do not offer specific discounts or compensation (route to DM/email)
- Sign off as the company name, not a specific person

Company name: {company_name}
"""


async def generate_response(
    review: Review, company_name: str
) -> str:
    if review.escalated:
        return None  # Do not auto-respond to escalated reviews

    strategy = RESPONSE_STRATEGIES.get(
        review.sentiment, RESPONSE_STRATEGIES[Sentiment.NEUTRAL]
    )

    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": RESPONSE_PROMPT.format(
                author=review.author,
                platform=review.platform.value,
                rating=review.rating,
                text=review.text,
                sentiment=review.sentiment.value,
                topics=", ".join(review.topics),
                tone=strategy["tone"],
                max_words=strategy["max_words"],
                template_hint=strategy["template_hint"],
                company_name=company_name,
            ),
        }],
    )
    return response.choices[0].message.content

Escalation and Routing

Critical reviews need human attention. The agent routes these to the right team member based on the topic and severity.

ESCALATION_ROUTES = {
    "billing": "billing-team@company.com",
    "technical": "support-lead@company.com",
    "safety": "legal@company.com",
    "employee": "hr@company.com",
    "default": "cx-manager@company.com",
}


async def route_escalation(review: Review, notifier) -> str:
    # Determine routing based on topics
    route_email = ESCALATION_ROUTES["default"]
    for topic in review.topics:
        topic_lower = topic.lower()
        for keyword, email in ESCALATION_ROUTES.items():
            if keyword in topic_lower:
                route_email = email
                break

    await notifier.send_email(
        to=route_email,
        subject=f"Escalated Review - {review.platform.value} - {review.rating}/5",
        body=(
            f"A review requires human attention.\n\n"
            f"Author: {review.author}\n"
            f"Platform: {review.platform.value}\n"
            f"Rating: {review.rating}/5\n"
            f"Sentiment: {review.sentiment.value}\n"
            f"Reason: {review.escalation_reason}\n\n"
            f"Review text:\n{review.text}"
        ),
    )
    return route_email


async def process_reviews(
    reviews: list[Review], company_name: str, notifier
):
    """Main processing pipeline for new reviews."""
    for review in reviews:
        review = await analyze_review(review)

        if review.escalated:
            route = await route_escalation(review, notifier)
            print(f"Escalated review {review.id} to {route}")
        else:
            response_text = await generate_response(review, company_name)
            if response_text:
                review.response = response_text
                review.response_date = datetime.utcnow()
                # Queue for posting (with optional human approval)
                await notifier.queue_response(review)

Running the Agent on a Schedule

The review response agent runs periodically, collecting new reviews and processing them through the analysis and response pipeline.

import asyncio
from datetime import datetime, timedelta


async def review_agent_loop(
    collector: MultiPlatformCollector,
    company_name: str,
    notifier,
    check_interval_minutes: int = 30,
):
    last_check = datetime.utcnow() - timedelta(hours=1)

    while True:
        try:
            new_reviews = await collector.fetch_all(since=last_check)
            if new_reviews:
                await process_reviews(new_reviews, company_name, notifier)
                print(f"Processed {len(new_reviews)} new reviews")
            last_check = datetime.utcnow()
        except Exception as e:
            print(f"Error in review agent loop: {e}")

        await asyncio.sleep(check_interval_minutes * 60)

FAQ

Should I auto-post AI-generated responses or require human approval?

Start with human approval for all responses until you have confidence in the quality. After a calibration period where you verify the agent consistently produces appropriate responses, switch to auto-posting for positive reviews (4-5 stars) while keeping human approval for negative reviews. Never auto-post responses to 1-star reviews or escalated cases.

How do I prevent the agent from generating responses that sound robotic or templated?

Include 3-5 examples of your best human-written responses in the system prompt as few-shot examples. Vary the response structure by including randomized constraints (e.g., sometimes start with the reviewer's name, sometimes with an acknowledgment of their experience). Run a post-generation diversity check that flags responses too similar to recent ones.

How do I handle fake or spam reviews?

Add a pre-analysis step that checks for spam indicators: extremely short reviews with extreme ratings, reviews from accounts with no history, or reviews containing competitor product names. Flag suspected spam reviews for manual verification rather than generating automatic responses. Responding to fake reviews can legitimize them.

---

#ReputationManagement #ReviewResponse #SentimentAnalysis #CustomerFeedback #Python #AgenticAI #LearnAI #AIEngineering