richness

reviews.py

import json
from textwrap import dedent
from typing import Literal

from openai import AsyncAzureOpenAI, DefaultAioHttpClient
from openai.lib._pydantic import to_strict_json_schema
from pydantic import BaseModel, Field

from config import settings


def get_client(http_client) -> AsyncAzureOpenAI:
    return AsyncAzureOpenAI(
        api_key=settings.AZURE_OPENAI_KEY.get_secret_value(),
        azure_endpoint=settings.AZURE_OPENAI_ENDPOINT,
        api_version=settings.AZURE_OPENAI_API_VERSION,
        http_client=http_client,
    )

TopicCategory = Literal[
    "customer_service",  # general service quality
    "response_time",     # speed of response/resolution
    "staff_attitude",    # helpfulness, politeness
    "issue_resolution",  # whether problem was solved
    "communication",     # clarity of communication
    "product_quality",   # product-related issues
    "delivery",          # shipping/logistics
    "pricing",           # cost/value concerns
    "other",
]

UrgencyTrigger = Literal[
    "refund_demand",
    "legal_threat",
    "social_media_threat",
    "repeat_issue",
    "extreme_frustration",
    "regulatory_complaint",
]


class Sentiment(BaseModel):
    """Detailed sentiment analysis with valence, intensity, and emotion."""

    valence: Literal["positive", "negative", "neutral"] = Field(
        description="Overall direction of the sentiment"
    )
    intensity: Literal["mild", "moderate", "strong"] = Field(
        description="How strongly the sentiment is expressed"
    )
    emotion: Literal[
        "anger", "frustration", "disappointment", "satisfaction",
        "delight", "confusion", "anxiety", "relief", "neutral"
    ] = Field(
        description="The primary emotion expressed"
    )


class TopicReasoning(BaseModel):
    """Reasoning and evidence for a single topic classification."""

    reasoning: str = Field(
        description="Step-by-step reasoning for why this topic was identified in the review"
    )
    topic: TopicCategory = Field(
        description="The identified topic category"
    )
    supporting_quotes: list[str] = Field(
        description="Direct quotes from the review that support this topic classification"
    )
    sentiment: Sentiment = Field(
        description="Detailed sentiment analysis for this topic"
    )


class UrgencyAssessment(BaseModel):
    """Urgency/escalation assessment for the review."""

    reasoning: str = Field(
        description="Explanation for the urgency level assignment"
    )
    level: Literal["low", "medium", "high", "critical"] = Field(
        description="Urgency level for response prioritization"
    )
    triggers: list[UrgencyTrigger] = Field(
        default_factory=list,
        description="Specific triggers that contributed to the urgency assessment"
    )


class ReviewAnalysis(BaseModel):
    """Complete analysis of a Trustpilot review."""

    language: Literal["en", "da"] = Field(
        description="Detected language of the review"
    )
    star_rating: int = Field(
        ge=1, le=5,
        description="Star rating from 1-5"
    )
    topics: list[TopicReasoning] = Field(
        description="List of identified topics with reasoning and evidence"
    )
    urgency: UrgencyAssessment = Field(
        description="Urgency assessment for response prioritization"
    )
    overall_sentiment: Sentiment = Field(
        description="Overall sentiment of the review"
    )
    summary: str = Field(
        description="Brief summary of the review in English"
    )


class Prompt:
    prompt: str = dedent(text="""You are an expert customer service analyst for a brand monitoring their Trustpilot reviews.
Your goal is to analyze reviews to help improve customer service.

Analyze the following review and extract structured insights.

IMPORTANT: For each topic you identify, you MUST:
1. First explain your reasoning for why this topic applies
2. Then classify the topic
3. Provide direct quotes from the review as evidence

Languages: The review may be in English or Danish. Detect the language and analyze accordingly.
Always provide the summary in English, regardless of the review language.

<review>
Star Rating: {star_rating}/5
Review Text: {review_text}
</review>

Analyze this review thoroughly, identifying all relevant customer service topics."""
)
    def fill(self, review_text: str, star_rating: int = 3) -> str:
        return self.prompt.format(review_text=review_text, star_rating=star_rating)


async def analyze_review(
    review_text: str,
    star_rating: int,
    prompt: Prompt | None = None,
) -> ReviewAnalysis:
    """
    Analyzes a Trustpilot review and extracts structured insights.

    Args:
        review_text: The review text to analyze
        star_rating: The star rating (1-5)
        prompt: Optional custom prompt, defaults to standard Prompt

    Returns:
        ReviewAnalysis with topics, urgency, sentiment, and summary
    """
    if prompt is None:
        prompt = Prompt()

    content: str = prompt.fill(review_text=review_text, star_rating=star_rating)

    async with DefaultAioHttpClient() as http_client:
        client = get_client(http_client)
        response = await client.responses.create(
            model=settings.AZURE_OPENAI_MODEL,
            input=content,
            text={
                "format": {
                    "type": "json_schema",
                    "name": "ReviewAnalysis",
                    "strict": True,
                    "schema": to_strict_json_schema(ReviewAnalysis),
                }
            },
        )

    result = response.output_text
    if result is None:
        raise ValueError("No response content received from the model")

    return ReviewAnalysis.model_validate(json.loads(result))


if __name__ == "__main__":
    import asyncio

    SAMPLE_REVIEW = """
    Terrible experience! I ordered a laptop 3 weeks ago and still haven't received it. 
    I've called customer service 5 times and each time they tell me something different.
    The last agent was incredibly rude and just hung up on me. 
    I want a full refund immediately or I will be contacting my lawyer and posting about 
    this on social media. This is the worst company I've ever dealt with.
    """

    async def main():
        print("Analyzing sample review...")
        print("-" * 50)

        result = await analyze_review(
            review_text=SAMPLE_REVIEW,
            star_rating=1,
        )
        print(result.model_dump_json(indent=2))

    asyncio.run(main())