INTRODUCTION
Sentiment analysis has evolved dramatically with the advent of Large Language Models. Traditional approaches relied on lexicon-based methods or classical machine learning algorithms trained on labeled datasets. Today, LLMs offer a more nuanced understanding of context, sarcasm, and complex emotional states that were previously difficult to capture. This article explores how to build a production-ready sentiment analysis system using both open-source models and commercial APIs like OpenAI, providing you with the knowledge to implement sophisticated text analysis in your applications.
The power of LLM-based sentiment analysis lies in its ability to understand
context beyond simple keyword matching. Where traditional methods might
struggle with phrases like "not bad" or "could be worse," modern LLMs grasp the subtle implications and deliver accurate sentiment assessments. This guide will walk you through every component needed to build such a system, from basic setup to production deployment.
UNDERSTANDING THE ARCHITECTURE
Before diving into implementation, we need to understand the architectural
components of an LLM-powered sentiment analysis system. The architecture
consists of several key layers that work together to process text and extract
sentiment information.
The first layer is the input processing layer, which handles text
normalization, cleaning, and preparation. This layer ensures that the text
fed to the LLM is in the optimal format for analysis. While LLMs are robust
to various input formats, proper preprocessing can improve accuracy and reduce token usage, which directly impacts cost when using commercial APIs.
The second layer is the prompt engineering layer, where we craft instructions that guide the LLM to perform sentiment analysis according to our specific requirements. This is perhaps the most critical component because the quality of prompts directly determines the quality of results. A well-designed prompt specifies the task clearly, provides examples if needed, and defines the expected output format.
The third layer is the LLM interface layer, which manages communication with either local open-source models or remote API endpoints. This layer handles authentication, request formatting, error handling, and response parsing. It abstracts away the differences between various LLM providers, allowing the rest of the system to work with a consistent interface.
The fourth layer is the result processing layer, which takes the raw LLM
output and transforms it into structured data that applications can use. This
includes parsing sentiment labels, extracting confidence scores, and handling edge cases where the LLM might return unexpected formats.
Finally, the caching and optimization layer improves performance and reduces costs by storing results for previously analyzed texts and implementing batch processing strategies.
SETTING UP THE DEVELOPMENT ENVIRONMENT
To begin implementing our sentiment analysis system, we need to set up a
proper development environment with all necessary dependencies. The choice of libraries depends on whether we want to use open-source models, commercial APIs, or both.
For open-source models, we will use the Transformers library from Hugging
Face, which provides access to thousands of pre-trained models. For OpenAI integration, we will use their official Python client. Additionally, we need supporting libraries for text processing and data handling.
Here is the initial setup code showing the required imports and basic
configuration:
import os
import json
import time
from typing import Dict, List, Optional, Union, Any
from dataclasses import dataclass, asdict
from enum import Enum
import logging
# For open-source models
from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification
import torch
# For OpenAI API
from openai import OpenAI
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
This setup imports all essential components we will need throughout the
implementation. The logging configuration ensures we can track the system's behavior during development and production use. The type hints from the typing module enable better code documentation and IDE support.
DEFINING DATA STRUCTURES
Before implementing the core functionality, we need to define clear data
structures that represent sentiment analysis results. Well-defined data
structures make the code more maintainable and easier to test.
We will create an enumeration for sentiment labels and a dataclass to
represent analysis results:
class SentimentLabel(Enum):
POSITIVE = "positive"
NEGATIVE = "negative"
NEUTRAL = "neutral"
MIXED = "mixed"
@dataclass
class SentimentResult:
text: str
sentiment: SentimentLabel
confidence: float
raw_scores: Optional[Dict[str, float]] = None
model_used: Optional[str] = None
processing_time: Optional[float] = None
metadata: Optional[Dict[str, Any]] = None
def to_dict(self) -> Dict:
result = asdict(self)
result['sentiment'] = self.sentiment.value
return result
The SentimentLabel enumeration provides a type-safe way to represent different sentiment categories. Using an enum instead of plain strings prevents typos and makes the code more robust. The MIXED category is particularly useful for texts that express multiple sentiments simultaneously.
The SentimentResult dataclass encapsulates all information about a sentiment analysis operation. It includes not just the sentiment label but also confidence scores, timing information, and metadata. This comprehensive approach allows for detailed analysis of system performance and result quality. The to_dict method facilitates serialization for storage or API responses.
IMPLEMENTING THE TEXT PREPROCESSOR
Text preprocessing is crucial for optimal LLM performance. While modern LLMs are remarkably robust, proper preprocessing can improve accuracy and reduce token consumption. Our preprocessor will handle common text cleaning tasks while preserving important contextual information.
class TextPreprocessor:
def __init__(self, max_length: int = 512, preserve_case: bool = True):
self.max_length = max_length
self.preserve_case = preserve_case
logger.info(f"Initialized TextPreprocessor with max_length={max_length}")
def preprocess(self, text: str) -> str:
if not text or not isinstance(text, str):
raise ValueError("Input must be a non-empty string")
# Remove excessive whitespace while preserving single spaces
processed = ' '.join(text.split())
# Optionally normalize case
if not self.preserve_case:
processed = processed.lower()
# Truncate if necessary while trying to preserve complete sentences
if len(processed) > self.max_length:
processed = self._smart_truncate(processed, self.max_length)
return processed
def _smart_truncate(self, text: str, max_length: int) -> str:
if len(text) <= max_length:
return text
# Try to cut at sentence boundary
truncated = text[:max_length]
last_period = truncated.rfind('.')
last_exclamation = truncated.rfind('!')
last_question = truncated.rfind('?')
last_sentence_end = max(last_period, last_exclamation, last_question)
if last_sentence_end > max_length * 0.7:
return truncated[:last_sentence_end + 1]
return truncated
The TextPreprocessor class provides intelligent text cleaning that balances
thoroughness with preservation of meaning. The max_length parameter prevents token limit violations, which is especially important when working with APIs that charge per token. The preserve_case option allows flexibility depending on whether case information is relevant for the analysis.
The smart_truncate method demonstrates an important principle in text
processing: when truncation is necessary, doing it at sentence boundaries
preserves more context than arbitrary character-based cutting. This method looks for the last sentence-ending punctuation mark within the allowed length and cuts there, ensuring the LLM receives complete thoughts rather than fragments.
BUILDING THE PROMPT ENGINEERING SYSTEM
Prompt engineering is the art and science of instructing LLMs to perform
specific tasks effectively. For sentiment analysis, we need prompts that are
clear, consistent, and produce parseable outputs. The prompt system should be flexible enough to accommodate different use cases while maintaining quality.
class PromptTemplate:
def __init__(self, template_type: str = "standard"):
self.template_type = template_type
self.templates = self._initialize_templates()
def _initialize_templates(self) -> Dict[str, str]:
return {
"standard": """Analyze the sentiment of the following text and classify it as positive, negative, neutral, or mixed.
Text: {text}
Respond with a JSON object containing:
- sentiment: one of [positive, negative, neutral, mixed]
- confidence: a number between 0 and 1
- reasoning: brief explanation of your classification
Response:""",
"detailed": """You are a sentiment analysis expert. Analyze the following text carefully, considering context, tone, and implicit meanings.
Text: {text}
Provide a detailed sentiment analysis including:
- Overall sentiment classification (positive, negative, neutral, or mixed)
- Confidence level in your assessment
- Key phrases that influenced your decision
- Any detected sarcasm or irony
Format your response as JSON:""",
"simple": """Classify the sentiment of this text as positive, negative, neutral, or mixed: {text}
Respond in JSON format with sentiment and confidence fields:"""
}
def format_prompt(self, text: str, custom_instructions: Optional[str] = None) -> str:
base_prompt = self.templates.get(self.template_type, self.templates["standard"])
prompt = base_prompt.format(text=text)
if custom_instructions:
prompt = f"{prompt}\n\nAdditional instructions: {custom_instructions}"
return prompt
The PromptTemplate class encapsulates different prompting strategies for
various use cases. The standard template provides clear instructions and
specifies the exact output format, which is crucial for reliable parsing. The
detailed template is useful when you need more insight into the LLM's
reasoning process, while the simple template minimizes token usage for
high-volume applications.
The format_prompt method allows for runtime customization through the
custom_instructions parameter. This flexibility enables domain-specific
adaptations without modifying the core templates. For example, when analyzing product reviews, you might add instructions to pay special attention to specific product features.
IMPLEMENTING THE OPEN-SOURCE MODEL INTERFACE
Open-source models offer cost-effective sentiment analysis without per-request charges. The Hugging Face Transformers library provides access to numerous pre-trained models specifically designed for sentiment analysis. We will implement an interface that makes using these models straightforward.
class OpenSourceSentimentAnalyzer:
def __init__(self, model_name: str = "distilbert-base-uncased-finetuned-sst-2-english"):
self.model_name = model_name
self.device = "cuda" if torch.cuda.is_available() else "cpu"
logger.info(f"Initializing model {model_name} on {self.device}")
try:
self.pipeline = pipeline(
"sentiment-analysis",
model=model_name,
device=0 if self.device == "cuda" else -1
)
logger.info("Model loaded successfully")
except Exception as e:
logger.error(f"Failed to load model: {e}")
raise
def analyze(self, text: str) -> SentimentResult:
start_time = time.time()
try:
# Run inference
result = self.pipeline(text)[0]
# Map model output to our standard format
sentiment = self._map_sentiment(result['label'])
confidence = result['score']
processing_time = time.time() - start_time
return SentimentResult(
text=text,
sentiment=sentiment,
confidence=confidence,
raw_scores={result['label']: result['score']},
model_used=self.model_name,
processing_time=processing_time
)
except Exception as e:
logger.error(f"Analysis failed: {e}")
raise
def _map_sentiment(self, label: str) -> SentimentLabel:
label_lower = label.lower()
if 'pos' in label_lower:
return SentimentLabel.POSITIVE
elif 'neg' in label_lower:
return SentimentLabel.NEGATIVE
else:
return SentimentLabel.NEUTRAL
The OpenSourceSentimentAnalyzer class demonstrates how to integrate Hugging Face models into our system. The constructor automatically detects GPU availability and uses it when possible, significantly speeding up inference. The model_name parameter allows easy switching between different pre-trained models depending on accuracy requirements and performance constraints.
The analyze method wraps the model inference in proper error handling and
timing measurement. This information is valuable for monitoring system
performance and identifying bottlenecks. The _map_sentiment method handles the fact that different models may use different label formats, normalizing them to our standard SentimentLabel enumeration.
IMPLEMENTING THE OPENAI API INTERFACE
Commercial LLM APIs like OpenAI's GPT models offer superior understanding of context and nuance compared to smaller open-source models. The trade-off is cost and latency, but for many applications, the improved accuracy justifies these factors. Our OpenAI interface will handle API communication, error recovery, and response parsing.
class OpenAISentimentAnalyzer:
def __init__(self, api_key: Optional[str] = None, model: str = "gpt-4-turbo-preview"):
self.api_key = api_key or os.getenv("OPENAI_API_KEY")
if not self.api_key:
raise ValueError("OpenAI API key must be provided or set in OPENAI_API_KEY environment variable")
self.model = model
self.client = OpenAI(api_key=self.api_key)
self.prompt_template = PromptTemplate("standard")
logger.info(f"Initialized OpenAI analyzer with model {model}")
def analyze(self, text: str, max_retries: int = 3) -> SentimentResult:
start_time = time.time()
prompt = self.prompt_template.format_prompt(text)
for attempt in range(max_retries):
try:
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "You are a sentiment analysis expert."},
{"role": "user", "content": prompt}
],
temperature=0.3,
response_format={"type": "json_object"}
)
# Parse the response
content = response.choices[0].message.content
parsed = json.loads(content)
sentiment = self._parse_sentiment(parsed.get('sentiment', ''))
confidence = float(parsed.get('confidence', 0.0))
processing_time = time.time() - start_time
return SentimentResult(
text=text,
sentiment=sentiment,
confidence=confidence,
raw_scores=parsed,
model_used=self.model,
processing_time=processing_time,
metadata={'reasoning': parsed.get('reasoning', '')}
)
except json.JSONDecodeError as e:
logger.warning(f"Failed to parse JSON response (attempt {attempt + 1}): {e}")
if attempt == max_retries - 1:
raise
except Exception as e:
logger.error(f"API call failed (attempt {attempt + 1}): {e}")
if attempt == max_retries - 1:
raise
time.sleep(2 ** attempt)
raise RuntimeError("Failed to get valid response after all retries")
def _parse_sentiment(self, sentiment_str: str) -> SentimentLabel:
sentiment_lower = sentiment_str.lower()
if sentiment_lower == 'positive':
return SentimentLabel.POSITIVE
elif sentiment_lower == 'negative':
return SentimentLabel.NEGATIVE
elif sentiment_lower == 'mixed':
return SentimentLabel.MIXED
else:
return SentimentLabel.NEUTRAL
The OpenAISentimentAnalyzer class implements robust API communication with proper error handling and retry logic. The max_retries parameter with
exponential backoff ensures transient network issues do not cause immediate failures. This is essential for production systems where reliability is paramount.
The temperature parameter is set to 0.3, which provides a good balance between consistency and natural language understanding. Lower temperatures make the model more deterministic, which is generally desirable for classification tasks. The response_format parameter ensures the API returns valid JSON, simplifying parsing.
CREATING A UNIFIED INTERFACE
To make our system truly flexible, we need a unified interface that allows
seamless switching between different LLM backends. This abstraction enables applications to use the most appropriate model for each situation without changing code.
class SentimentAnalyzer:
def __init__(self, backend: str = "openai", **kwargs):
self.backend = backend
self.preprocessor = TextPreprocessor()
if backend == "openai":
self.analyzer = OpenAISentimentAnalyzer(**kwargs)
elif backend == "opensource":
self.analyzer = OpenSourceSentimentAnalyzer(**kwargs)
else:
raise ValueError(f"Unknown backend: {backend}")
logger.info(f"Initialized SentimentAnalyzer with {backend} backend")
def analyze(self, text: str, preprocess: bool = True) -> SentimentResult:
if preprocess:
text = self.preprocessor.preprocess(text)
return self.analyzer.analyze(text)
def analyze_batch(self, texts: List[str], preprocess: bool = True) -> List[SentimentResult]:
results = []
for text in texts:
try:
result = self.analyze(text, preprocess=preprocess)
results.append(result)
except Exception as e:
logger.error(f"Failed to analyze text: {e}")
results.append(None)
return results
The SentimentAnalyzer class provides a clean, unified interface regardless of the underlying model. The backend parameter determines which implementation to use, while kwargs allows passing backend-specific configuration. This design pattern makes it easy to add new backends in the future without changing existing code.
The analyze_batch method demonstrates how to process multiple texts
efficiently. While this simple implementation processes texts sequentially, it
provides a foundation for more sophisticated batch processing strategies like parallel execution or request batching for API efficiency.
IMPLEMENTING CACHING FOR PERFORMANCE
Caching is essential for production sentiment analysis systems. Many
applications analyze the same texts repeatedly, and caching eliminates
redundant processing while reducing costs. We will implement a simple but
effective caching layer.
import hashlib
from functools import lru_cache
class CachedSentimentAnalyzer:
def __init__(self, analyzer: SentimentAnalyzer, cache_size: int = 1000):
self.analyzer = analyzer
self.cache_size = cache_size
self.cache = {}
self.hits = 0
self.misses = 0
logger.info(f"Initialized cache with size {cache_size}")
def _get_cache_key(self, text: str) -> str:
return hashlib.md5(text.encode()).hexdigest()
def analyze(self, text: str, preprocess: bool = True) -> SentimentResult:
cache_key = self._get_cache_key(text)
if cache_key in self.cache:
self.hits += 1
logger.debug(f"Cache hit for text (hit rate: {self.get_hit_rate():.2%})")
return self.cache[cache_key]
self.misses += 1
result = self.analyzer.analyze(text, preprocess=preprocess)
# Implement simple LRU eviction
if len(self.cache) >= self.cache_size:
# Remove oldest entry
oldest_key = next(iter(self.cache))
del self.cache[oldest_key]
self.cache[cache_key] = result
return result
def get_hit_rate(self) -> float:
total = self.hits + self.misses
return self.hits / total if total > 0 else 0.0
def clear_cache(self):
self.cache.clear()
self.hits = 0
self.misses = 0
logger.info("Cache cleared")
The CachedSentimentAnalyzer wraps any SentimentAnalyzer instance and adds caching functionality. The cache key is generated using MD5 hashing of the input text, which provides a good balance between collision resistance and performance. The cache tracks hit and miss statistics, which are valuable for monitoring and optimization.
The simple LRU eviction strategy ensures the cache does not grow unbounded. When the cache reaches its size limit, the oldest entry is removed. For production systems with higher performance requirements, you might consider using Redis or Memcached instead of this in-memory implementation.
HANDLING ERRORS AND EDGE CASES
Robust error handling is crucial for production systems. Our sentiment
analysis system needs to gracefully handle various failure modes including
network errors, invalid inputs, and unexpected model outputs.
class RobustSentimentAnalyzer:
def __init__(self, primary_backend: str = "openai", fallback_backend: str = "opensource", **kwargs):
self.primary = SentimentAnalyzer(backend=primary_backend, **kwargs)
self.fallback = SentimentAnalyzer(backend=fallback_backend, **kwargs)
self.fallback_count = 0
logger.info("Initialized robust analyzer with fallback support")
def analyze(self, text: str, use_fallback: bool = True) -> SentimentResult:
if not text or not text.strip():
raise ValueError("Cannot analyze empty text")
try:
return self.primary.analyze(text)
except Exception as e:
logger.warning(f"Primary analyzer failed: {e}")
if use_fallback:
try:
self.fallback_count += 1
logger.info(f"Using fallback analyzer (fallback count: {self.fallback_count})")
return self.fallback.analyze(text)
except Exception as fallback_error:
logger.error(f"Fallback analyzer also failed: {fallback_error}")
raise
else:
raise
The RobustSentimentAnalyzer implements a fallback pattern where if the primary analyzer fails, the system automatically tries a secondary analyzer. This is particularly useful when using OpenAI as the primary backend with an open-source model as fallback, ensuring the system remains operational even during API outages.
MONITORING AND LOGGING
Production systems require comprehensive monitoring to track performance, identify issues, and optimize costs. We will implement a monitoring layer that collects metrics about system behavior.
class MonitoredSentimentAnalyzer:
def __init__(self, analyzer: SentimentAnalyzer):
self.analyzer = analyzer
self.metrics = {
'total_requests': 0,
'successful_requests': 0,
'failed_requests': 0,
'total_processing_time': 0.0,
'sentiment_distribution': {
'positive': 0,
'negative': 0,
'neutral': 0,
'mixed': 0
}
}
def analyze(self, text: str) -> SentimentResult:
self.metrics['total_requests'] += 1
try:
result = self.analyzer.analyze(text)
self.metrics['successful_requests'] += 1
self.metrics['total_processing_time'] += result.processing_time or 0
self.metrics['sentiment_distribution'][result.sentiment.value] += 1
return result
except Exception as e:
self.metrics['failed_requests'] += 1
raise
def get_metrics(self) -> Dict:
metrics = self.metrics.copy()
if metrics['total_requests'] > 0:
metrics['success_rate'] = metrics['successful_requests'] / metrics['total_requests']
metrics['average_processing_time'] = metrics['total_processing_time'] / metrics['successful_requests'] if metrics['successful_requests'] > 0 else 0
return metrics
The MonitoredSentimentAnalyzer tracks key metrics including request counts, success rates, processing times, and sentiment distributions. These metrics provide insights into system health and usage patterns. In a production environment, you would typically export these metrics to a monitoring system like Prometheus or CloudWatch.
COMPLETE RUNNING EXAMPLE
Below is a complete, production-ready implementation that integrates all the components discussed above. This code can be used directly in applications and supports both open-source and OpenAI backends with caching, monitoring, and robust error handling.
#!/usr/bin/env python3
import os
import json
import time
import hashlib
from typing import Dict, List, Optional, Union, Any
from dataclasses import dataclass, asdict
from enum import Enum
import logging
# For open-source models
from transformers import pipeline
import torch
# For OpenAI API
from openai import OpenAI
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class SentimentLabel(Enum):
POSITIVE = "positive"
NEGATIVE = "negative"
NEUTRAL = "neutral"
MIXED = "mixed"
@dataclass
class SentimentResult:
text: str
sentiment: SentimentLabel
confidence: float
raw_scores: Optional[Dict[str, float]] = None
model_used: Optional[str] = None
processing_time: Optional[float] = None
metadata: Optional[Dict[str, Any]] = None
def to_dict(self) -> Dict:
result = asdict(self)
result['sentiment'] = self.sentiment.value
return result
def __str__(self) -> str:
return f"Sentiment: {self.sentiment.value} (confidence: {self.confidence:.2f})"
class TextPreprocessor:
def __init__(self, max_length: int = 512, preserve_case: bool = True):
self.max_length = max_length
self.preserve_case = preserve_case
logger.info(f"Initialized TextPreprocessor with max_length={max_length}")
def preprocess(self, text: str) -> str:
if not text or not isinstance(text, str):
raise ValueError("Input must be a non-empty string")
processed = ' '.join(text.split())
if not self.preserve_case:
processed = processed.lower()
if len(processed) > self.max_length:
processed = self._smart_truncate(processed, self.max_length)
return processed
def _smart_truncate(self, text: str, max_length: int) -> str:
if len(text) <= max_length:
return text
truncated = text[:max_length]
last_period = truncated.rfind('.')
last_exclamation = truncated.rfind('!')
last_question = truncated.rfind('?')
last_sentence_end = max(last_period, last_exclamation, last_question)
if last_sentence_end > max_length * 0.7:
return truncated[:last_sentence_end + 1]
return truncated
class PromptTemplate:
def __init__(self, template_type: str = "standard"):
self.template_type = template_type
self.templates = self._initialize_templates()
def _initialize_templates(self) -> Dict[str, str]:
return {
"standard": """Analyze the sentiment of the following text and classify it as positive, negative, neutral, or mixed.
Text: {text}
Respond with a JSON object containing:
- sentiment: one of [positive, negative, neutral, mixed]
- confidence: a number between 0 and 1
- reasoning: brief explanation of your classification
Response:""",
"detailed": """You are a sentiment analysis expert. Analyze the following text carefully, considering context, tone, and implicit meanings.
Text: {text}
Provide a detailed sentiment analysis including:
- Overall sentiment classification (positive, negative, neutral, or mixed)
- Confidence level in your assessment
- Key phrases that influenced your decision
- Any detected sarcasm or irony
Format your response as JSON:""",
"simple": """Classify the sentiment of this text as positive, negative, neutral, or mixed: {text}
Respond in JSON format with sentiment and confidence fields:"""
}
def format_prompt(self, text: str, custom_instructions: Optional[str] = None) -> str:
base_prompt = self.templates.get(self.template_type, self.templates["standard"])
prompt = base_prompt.format(text=text)
if custom_instructions:
prompt = f"{prompt}\n\nAdditional instructions: {custom_instructions}"
return prompt
class OpenSourceSentimentAnalyzer:
def __init__(self, model_name: str = "distilbert-base-uncased-finetuned-sst-2-english"):
self.model_name = model_name
self.device = "cuda" if torch.cuda.is_available() else "cpu"
logger.info(f"Initializing model {model_name} on {self.device}")
try:
self.pipeline = pipeline(
"sentiment-analysis",
model=model_name,
device=0 if self.device == "cuda" else -1
)
logger.info("Model loaded successfully")
except Exception as e:
logger.error(f"Failed to load model: {e}")
raise
def analyze(self, text: str) -> SentimentResult:
start_time = time.time()
try:
result = self.pipeline(text)[0]
sentiment = self._map_sentiment(result['label'])
confidence = result['score']
processing_time = time.time() - start_time
return SentimentResult(
text=text,
sentiment=sentiment,
confidence=confidence,
raw_scores={result['label']: result['score']},
model_used=self.model_name,
processing_time=processing_time
)
except Exception as e:
logger.error(f"Analysis failed: {e}")
raise
def _map_sentiment(self, label: str) -> SentimentLabel:
label_lower = label.lower()
if 'pos' in label_lower:
return SentimentLabel.POSITIVE
elif 'neg' in label_lower:
return SentimentLabel.NEGATIVE
else:
return SentimentLabel.NEUTRAL
class OpenAISentimentAnalyzer:
def __init__(self, api_key: Optional[str] = None, model: str = "gpt-4-turbo-preview"):
self.api_key = api_key or os.getenv("OPENAI_API_KEY")
if not self.api_key:
raise ValueError("OpenAI API key must be provided or set in OPENAI_API_KEY environment variable")
self.model = model
self.client = OpenAI(api_key=self.api_key)
self.prompt_template = PromptTemplate("standard")
logger.info(f"Initialized OpenAI analyzer with model {model}")
def analyze(self, text: str, max_retries: int = 3) -> SentimentResult:
start_time = time.time()
prompt = self.prompt_template.format_prompt(text)
for attempt in range(max_retries):
try:
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "You are a sentiment analysis expert."},
{"role": "user", "content": prompt}
],
temperature=0.3,
response_format={"type": "json_object"}
)
content = response.choices[0].message.content
parsed = json.loads(content)
sentiment = self._parse_sentiment(parsed.get('sentiment', ''))
confidence = float(parsed.get('confidence', 0.0))
processing_time = time.time() - start_time
return SentimentResult(
text=text,
sentiment=sentiment,
confidence=confidence,
raw_scores=parsed,
model_used=self.model,
processing_time=processing_time,
metadata={'reasoning': parsed.get('reasoning', '')}
)
except json.JSONDecodeError as e:
logger.warning(f"Failed to parse JSON response (attempt {attempt + 1}): {e}")
if attempt == max_retries - 1:
raise
except Exception as e:
logger.error(f"API call failed (attempt {attempt + 1}): {e}")
if attempt == max_retries - 1:
raise
time.sleep(2 ** attempt)
raise RuntimeError("Failed to get valid response after all retries")
def _parse_sentiment(self, sentiment_str: str) -> SentimentLabel:
sentiment_lower = sentiment_str.lower()
if sentiment_lower == 'positive':
return SentimentLabel.POSITIVE
elif sentiment_lower == 'negative':
return SentimentLabel.NEGATIVE
elif sentiment_lower == 'mixed':
return SentimentLabel.MIXED
else:
return SentimentLabel.NEUTRAL
class SentimentAnalyzer:
def __init__(self, backend: str = "openai", **kwargs):
self.backend = backend
self.preprocessor = TextPreprocessor()
if backend == "openai":
self.analyzer = OpenAISentimentAnalyzer(**kwargs)
elif backend == "opensource":
self.analyzer = OpenSourceSentimentAnalyzer(**kwargs)
else:
raise ValueError(f"Unknown backend: {backend}")
logger.info(f"Initialized SentimentAnalyzer with {backend} backend")
def analyze(self, text: str, preprocess: bool = True) -> SentimentResult:
if preprocess:
text = self.preprocessor.preprocess(text)
return self.analyzer.analyze(text)
def analyze_batch(self, texts: List[str], preprocess: bool = True) -> List[SentimentResult]:
results = []
for text in texts:
try:
result = self.analyze(text, preprocess=preprocess)
results.append(result)
except Exception as e:
logger.error(f"Failed to analyze text: {e}")
results.append(None)
return results
class CachedSentimentAnalyzer:
def __init__(self, analyzer: SentimentAnalyzer, cache_size: int = 1000):
self.analyzer = analyzer
self.cache_size = cache_size
self.cache = {}
self.hits = 0
self.misses = 0
logger.info(f"Initialized cache with size {cache_size}")
def _get_cache_key(self, text: str) -> str:
return hashlib.md5(text.encode()).hexdigest()
def analyze(self, text: str, preprocess: bool = True) -> SentimentResult:
cache_key = self._get_cache_key(text)
if cache_key in self.cache:
self.hits += 1
logger.debug(f"Cache hit (hit rate: {self.get_hit_rate():.2%})")
return self.cache[cache_key]
self.misses += 1
result = self.analyzer.analyze(text, preprocess=preprocess)
if len(self.cache) >= self.cache_size:
oldest_key = next(iter(self.cache))
del self.cache[oldest_key]
self.cache[cache_key] = result
return result
def get_hit_rate(self) -> float:
total = self.hits + self.misses
return self.hits / total if total > 0 else 0.0
def clear_cache(self):
self.cache.clear()
self.hits = 0
self.misses = 0
logger.info("Cache cleared")
class RobustSentimentAnalyzer:
def __init__(self, primary_backend: str = "openai", fallback_backend: str = "opensource", **kwargs):
self.primary = SentimentAnalyzer(backend=primary_backend, **kwargs)
self.fallback = SentimentAnalyzer(backend=fallback_backend, **kwargs)
self.fallback_count = 0
logger.info("Initialized robust analyzer with fallback support")
def analyze(self, text: str, use_fallback: bool = True) -> SentimentResult:
if not text or not text.strip():
raise ValueError("Cannot analyze empty text")
try:
return self.primary.analyze(text)
except Exception as e:
logger.warning(f"Primary analyzer failed: {e}")
if use_fallback:
try:
self.fallback_count += 1
logger.info(f"Using fallback analyzer (fallback count: {self.fallback_count})")
return self.fallback.analyze(text)
except Exception as fallback_error:
logger.error(f"Fallback analyzer also failed: {fallback_error}")
raise
else:
raise
class MonitoredSentimentAnalyzer:
def __init__(self, analyzer: Union[SentimentAnalyzer, CachedSentimentAnalyzer]):
self.analyzer = analyzer
self.metrics = {
'total_requests': 0,
'successful_requests': 0,
'failed_requests': 0,
'total_processing_time': 0.0,
'sentiment_distribution': {
'positive': 0,
'negative': 0,
'neutral': 0,
'mixed': 0
}
}
def analyze(self, text: str) -> SentimentResult:
self.metrics['total_requests'] += 1
try:
result = self.analyzer.analyze(text)
self.metrics['successful_requests'] += 1
self.metrics['total_processing_time'] += result.processing_time or 0
self.metrics['sentiment_distribution'][result.sentiment.value] += 1
return result
except Exception as e:
self.metrics['failed_requests'] += 1
raise
def get_metrics(self) -> Dict:
metrics = self.metrics.copy()
if metrics['total_requests'] > 0:
metrics['success_rate'] = metrics['successful_requests'] / metrics['total_requests']
metrics['average_processing_time'] = metrics['total_processing_time'] / metrics['successful_requests'] if metrics['successful_requests'] > 0 else 0
return metrics
def print_metrics(self):
metrics = self.get_metrics()
print("\\n" + "="*60)
print("SENTIMENT ANALYSIS METRICS")
print("="*60)
print(f"Total Requests: {metrics['total_requests']}")
print(f"Successful: {metrics['successful_requests']}")
print(f"Failed: {metrics['failed_requests']}")
if 'success_rate' in metrics:
print(f"Success Rate: {metrics['success_rate']:.2%}")
print(f"Average Processing Time: {metrics['average_processing_time']:.3f}s")
print("\\nSentiment Distribution:")
for sentiment, count in metrics['sentiment_distribution'].items():
print(f" {sentiment.capitalize()}: {count}")
print("="*60 + "\\n")
def main():
print("\\n" + "="*60)
print("LLM-POWERED SENTIMENT ANALYSIS DEMONSTRATION")
print("="*60 + "\\n")
# Example texts for analysis
test_texts = [
"This product is absolutely amazing! I love it so much.",
"Terrible experience. Would not recommend to anyone.",
"It's okay, nothing special but does the job.",
"I love the design but hate the price. Mixed feelings overall.",
"The customer service was outstanding and resolved my issue quickly!",
"Worst purchase ever. Complete waste of money.",
"Average quality for the price point.",
"Not bad, could be worse I suppose."
]
print("Initializing sentiment analyzer with open-source backend...\\n")
# Create analyzer with caching and monitoring
base_analyzer = SentimentAnalyzer(backend="opensource")
cached_analyzer = CachedSentimentAnalyzer(base_analyzer, cache_size=100)
monitored_analyzer = MonitoredSentimentAnalyzer(cached_analyzer)
print("Analyzing sample texts...\\n")
for i, text in enumerate(test_texts, 1):
print(f"Text {i}: {text}")
try:
result = monitored_analyzer.analyze(text)
print(f"Result: {result}")
if result.metadata and 'reasoning' in result.metadata:
print(f"Reasoning: {result.metadata['reasoning']}")
print(f"Processing time: {result.processing_time:.3f}s\\n")
except Exception as e:
print(f"Error: {e}\\n")
# Test cache by analyzing same texts again
print("\\nTesting cache with repeated analysis...\\n")
for text in test_texts[:3]:
result = monitored_analyzer.analyze(text)
print(f"Cached result for: {text[:50]}...")
print(f"Sentiment: {result.sentiment.value}\\n")
# Print final metrics
monitored_analyzer.print_metrics()
# Show cache statistics
if isinstance(cached_analyzer, CachedSentimentAnalyzer):
print(f"Cache hit rate: {cached_analyzer.get_hit_rate():.2%}")
print(f"Cache hits: {cached_analyzer.hits}")
print(f"Cache misses: {cached_analyzer.misses}\\n")
print("Demonstration complete!\\n")
if __name__ == "__main__":
main()
CONCLUSION
Implementing LLM-powered sentiment analysis requires careful consideration of
multiple components working together harmoniously. From text preprocessing and
prompt engineering to model selection and caching strategies, each element
plays a crucial role in building a robust, production-ready system.
The choice between open-source and commercial LLM APIs depends on your
specific requirements. Open-source models offer cost-effectiveness and data
privacy, making them ideal for high-volume applications where sentiment
analysis is straightforward. Commercial APIs like OpenAI provide superior
accuracy and nuance detection, justifying their cost for applications where
precision is paramount.
The architecture presented in this article provides a solid foundation that
can be extended in numerous ways. You might add support for additional
languages, implement more sophisticated caching strategies using Redis, or
integrate with message queues for asynchronous processing. The modular design
ensures that such enhancements can be added without disrupting existing
functionality.
Remember that sentiment analysis is not a solved problem. Even the most
advanced LLMs can struggle with highly contextual or domain-specific language.
Always validate your system's performance on representative data from your
specific use case, and be prepared to fine-tune prompts or even models to
achieve optimal results.
The future of sentiment analysis lies in increasingly sophisticated language
understanding, multimodal analysis incorporating images and audio, and
real-time processing at scale. The foundation you have built here positions
you well to adopt these advances as they become available.
By following the principles and patterns outlined in this article, you can
build sentiment analysis systems that are not only accurate and efficient but
also maintainable and extensible. Whether you are analyzing customer feedback,
monitoring social media sentiment, or building conversational AI systems, the
techniques presented here will serve you well in creating production-quality
solutions.
No comments:
Post a Comment