MOS Multi-Source Opinion Summary — LLM 多源评论整合摘要

from dataclasses import dataclass, field
from typing import List, Dict
from collections import Counter

@dataclass
class PlatformReviews:
    platform: str
    reviews: List[str]
    avg_rating: float
    rating_dist: Dict[int, int] = field(default_factory=dict)

def extract_platform_signals(pr: PlatformReviews) -> Dict:
    all_text = ' '.join(pr.reviews).lower()
    pos_keywords = ['好用', 'love', 'great', 'quiet', '安静', '舒适', 'comfortable',
                    '方便', 'convenient', '效果好', 'effective', '推荐', 'recommend']
    neg_keywords = ['噪音', 'loud', 'noise', '漏奶', 'leak', '痛', 'painful',
                    '难清洗', 'hard to clean', '贵', 'expensive', '断货', '售后']
    pos_count = sum(all_text.count(k) for k in pos_keywords)
    neg_count = sum(all_text.count(k) for k in neg_keywords)
    pos_ratio = pos_count / (pos_count + neg_count + 1)
    top_pos = [k for k in pos_keywords if all_text.count(k) > 0][:3]
    top_neg = [k for k in neg_keywords if all_text.count(k) > 0][:3]
    return {'platform': pr.platform, 'avg_rating': pr.avg_rating,
            'pos_ratio': round(pos_ratio, 2), 'top_positive': top_pos,
            'top_negative': top_neg, 'review_count': len(pr.reviews)}

def multi_source_summary(platforms: List[PlatformReviews]) -> Dict:
    platform_signals = [extract_platform_signals(p) for p in platforms]
    all_pos = Counter()
    all_neg = Counter()
    for p in platforms:
        text = ' '.join(p.reviews).lower()
        for k in ['好用','安静','舒适','方便','效果好','推荐','love','great','quiet']:
            if text.count(k) > 0:
                all_pos[k] += text.count(k)
        for k in ['噪音','漏奶','痛','难清洗','贵','expensive','loud','painful']:
            if text.count(k) > 0:
                all_neg[k] += text.count(k)
    total_reviews = sum(len(p.reviews) for p in platforms)
    weighted_rating = sum(p.avg_rating * len(p.reviews) for p in platforms) / total_reviews
    platform_diff = []
    for sig in platform_signals:
        if sig['avg_rating'] > weighted_rating + 0.3:
            platform_diff.append(f"{sig['platform']} 评价明显更正面（{sig['avg_rating']:.1f}星）")
        elif sig['avg_rating'] < weighted_rating - 0.3:
            platform_diff.append(f"{sig['platform']} 评价偏负面（{sig['avg_rating']:.1f}星）")
    return {
        'total_reviews': total_reviews,
        'weighted_avg_rating': round(weighted_rating, 2),
        'top5_positives': [k for k, _ in all_pos.most_common(5)],
        'top5_negatives': [k for k, _ in all_neg.most_common(5)],
        'platform_differences': platform_diff,
        'platform_details': platform_signals,
    }

platforms = [
    PlatformReviews('Amazon US', [
        'Super quiet pump, love it! Easy to clean, my baby likes it.',
        'A bit expensive but worth it. Very comfortable, no pain at all.',
        'Loud noise at max level, otherwise great. Would recommend.',