LLM 创意评估方法：从主观评分到自动化指标

创意评估方法
├── 人工评估
│   ├── 整体评分（Holistic Scoring）
│   ├── 多维 Rubric 评分
│   └── 排序比较（Pairwise Ranking）
├── 自动化指标
│   ├── 多样性指标（Distinct-n, Self-BLEU）
│   ├── 新颖度指标（Semantic Novelty）
│   └── 连贯性指标（Coherence Score）
└── LLM-as-Judge
    ├── 单维度评分
    ├── 多维度 Rubric
    └── 对比较（Pairwise Comparison）

class HolisticScoring:
    """整体评分体系"""
    
    SCALE_5 = {
        5: "杰出 — 创意独特，语言精炼，情感打动人心",
        4: "优秀 — 有明确创意点，语言流畅，有一定感染力",
        3: "合格 — 表达清晰，但创意普通，缺乏亮点",
        2: "较差 — 表达生硬，创意陈旧，难以读完",
        1: "不合格 — 逻辑混乱，语言不通，无法理解"
    }

    def __init__(self, num_raters: int = 3):
        self.num_raters = num_raters

    def aggregate(self, scores: list[int]) -> dict:
        import numpy as np
        scores = sorted(scores)
        return {
            "mean": np.mean(scores),
            "median": np.median(scores),
            "std": np.std(scores),
            "min": min(scores),
            "max": max(scores),
            "agreement": self._inter_rater_agreement(scores),
        }

    def _inter_rater_agreement(self, scores: list[int]) -> float:
        """计算评分者间一致性"""
        if len(scores) < 2:
            return 1.0
        mean = sum(scores) / len(scores)
        variance = sum((s - mean) ** 2 for s in scores) / len(scores)
        max_variance = (max(scores) - min(scores)) ** 2 / 4
        return 1 - (variance / max_variance) if max_variance > 0 else 1.0

RUBRIC = {
    "creativity": {
        "weight": 0.30,
        "descriptors": {
            5: "创意独特，角度新颖，令人惊喜",
            4: "有明显创意点，但非全新角度",
            3: "有一定新意，但整体较常见",
            2: "创意薄弱，多为套路化表达",
            1: "完全老套，毫无新意",
        }
    },
    "language": {
        "weight": 0.20,
        "descriptors": {
            5: "语言精准优美，用词生动，节奏感强",
            4: "语言流畅，用词得当",
            3: "表达清晰，但语言平淡",
            2: "偶有语病，表达不够准确",
            1: "语病频出，难以理解",
        }
    },
    "structure": {
        "weight": 0.20,
        "descriptors": {
            5: "结构精巧，层次分明，起承转合自然",
            4: "结构合理，层次清晰",
            3: "有基本结构，但部分松散",
            2: "结构不够清晰，逻辑跳跃",
            1: "结构混乱，无法跟随",
        }
    },
    "emotion": {
        "weight": 0.15,
        "descriptors": {
            5: "情感真挚，强烈感染力，引发共鸣",
            4: "有情感表达，有一定感染力",
            3: "情感表达一般，偶有触动",
            2: "情感淡薄，难以引起共鸣",
            1: "毫无情感，或情感虚假",
        }
    },
    "relevance": {
        "weight": 0.15,
        "descriptors": {
            5: "完美切题，深度回应主题",
            4: "紧扣主题，有适度延伸",
            3: "基本相关，但略有偏移",
            2: "与主题关联较弱",
            1: "严重偏题",
        }
    }
}

class RubricScorer:
    def __init__(self, rubric: dict):
        self.rubric = rubric

    def compute_weighted_score(self, dimension_scores: dict[str, int]) -> dict:
        total = 0.0
        breakdown = {}
        for dim, score in dimension_scores.items():
            weight = self.rubric[dim]["weight"]
            weighted = score * weight
            total += weighted
            breakdown[dim] = {
                "raw_score": score,
                "weight": weight,
                "weighted_score": round(weighted, 2),
            }
        return {
            "total_score": round(total, 2),
            "breakdown": breakdown,
            "grade": self._score_to_grade(total),
        }

    def _score_to_grade(self, score: float) -> str:
        if score >= 4.5: return "S"
        elif score >= 4.0: return "A"
        elif score >= 3.5: return "B"
        elif score >= 3.0: return "C"
        elif score >= 2.0: return "D"
        else: return "F"

from itertools import combinations
import numpy as np

class PairwiseRanker:
    """基于两两比较的排序系统"""
    
    def __init__(self):
        self.comparisons = []

    def add_comparison(self, winner_id: str, loser_id: str, confidence: float = 1.0):
        self.comparisons.append({
            "winner": winner_id,
            "loser": loser_id,
            "confidence": confidence,
        })

    def compute_rankings(self, item_ids: list[str]) -> list[dict]:
        """使用 Bradley-Terry 模型计算排名"""
        # 简化的 Elo 评分
        elo = {item_id: 1000 for item_id in item_ids}
        K = 32
        for comp in self.comparisons:
            w, l = comp["winner"], comp["loser"]
            expected_w = 1 / (1 + 10 ** ((elo[l] - elo[w]) / 400))
            elo[w] += K * (1 - expected_w) * comp["confidence"]
            elo[l] -= K * (1 - expected_w) * comp["confidence"]
        
        ranked = sorted(item_ids, key=lambda x: elo[x], reverse=True)
        return [
            {"rank": i + 1, "item_id": item, "elo": round(elo[item])}
            for i, item in enumerate(ranked)
        ]

from collections import Counter
from typing import List

class DiversityMetrics:
    """文本多样性指标集合"""
    
    @staticmethod
    def distinct_n(text: str, n: int = 2) -> float:
        """Distinct-N: 唯一 n-gram 占比"""
        tokens = text.split()
        if len(tokens) < n:
            return 0.0
        ngrams = [tuple(tokens[i:i+n]) for i in range(len(tokens) - n + 1)]
        unique = len(set(ngrams))
        total = len(ngrams)
        return unique / total if total > 0 else 0.0

    @staticmethod
    def self_bleu(texts: List[str], n: int = 4) -> float:
        """Self-BLEU: 越低表示多样性越高"""
        from nltk.translate.bleu_score import sentence_bleu, SmoothingFunction
        smoothie = SmoothingFunction().method1
        scores = []
        for i, ref in enumerate(texts):
            references = [t.split() for j, t in enumerate(texts) if j != i]
            candidate = ref.split()
            if len(candidate) < n or any(len(r) < n for r in references):
                continue
            score = sentence_bleu(references, candidate,
                                  weights=tuple(1/n for _ in range(n)),
                                  smoothing_function=smoothie)
            scores.append(score)
        return sum(scores) / len(scores) if scores else 0.0

    @staticmethod
    def entropy_n(text: str, n: int = 2) -> float:
        """n-gram 熵：越高表示越多样"""
        import math
        tokens = text.split()
        if len(tokens) < n:
            return 0.0
        ngrams = [tuple(tokens[i:i+n]) for i in range(len(tokens) - n + 1)]
        counts = Counter(ngrams)
        total = len(ngrams)
        entropy = 0.0
        for count in counts.values():
            p = count / total
            entropy -= p * math.log2(p)
        return entropy

    @staticmethod
    def semantic_diversity(texts: List[str], model_name: str = "all-MiniLM-L6-v2") -> float:
        """语义多样性：基于嵌入的余弦距离矩阵均值"""
        from sentence_transformers import SentenceTransformer
        import numpy as np
        model = SentenceTransformer(model_name)
        embeddings = model.encode(texts)
        # 计算两两余弦距离
        normalized = embeddings / np.linalg.norm(embeddings, axis=1, keepdims=True)
        sim_matrix = normalized @ normalized.T
        n = len(texts)
        # 取上三角（不含对角线）
        upper_tri = sim_matrix[np.triu_indices(n, k=1)]
        avg_similarity = np.mean(upper_tri)
        return 1 - avg_similarity  # 多样性 = 1 - 相似性

class NoveltyMetrics:
    """创意新颖度评估"""
    
    def __init__(self, reference_corpus: list[str]):
        """用参考语料库定义"已知创意空间" """
        self.reference_embeddings = None
        self.reference_corpus = reference_corpus

    def _ensure_embeddings(self, model_name="all-MiniLM-L6-v2"):
        if self.reference_embeddings is None:
            from sentence_transformers import SentenceTransformer
            import numpy as np
            model = SentenceTransformer(model_name)
            self.reference_embeddings = model.encode(self.reference_corpus)
            self.model = model

    def novelty_score(self, text: str) -> float:
        """
        新颖度 = 与参考语料库的最大相似度的补数
        1.0 = 完全新颖，0.0 = 与已有作品高度相似
        """
        import numpy as np
        self._ensure_embeddings()
        text_emb = self.model.encode([text])
        similarities = np.dot(self.reference_embeddings, text_emb.T).flatten()
        max_sim = np.max(similarities)
        return float(1 - max_sim)

    def novelty_percentile(self, text: str) -> float:
        """新颖度百分位：比多少比例的参考作品更新颖"""
        import numpy as np
        self._ensure_embeddings()
        text_emb = self.model.encode([text])
        ref_sims = np.dot(self.reference_embeddings, text_emb.T).flatten()
        text_novelty = 1 - np.max(ref_sims)
        ref_novelties = []
        for i in range(len(self.reference_embeddings)):
            others = np.delete(self.reference_embeddings, i, axis=0)
            sim = np.dot(others, self.reference_embeddings[i])
            ref_novelties.append(1 - np.max(sim))
        return float(np.mean(np.array(ref_novelties) < text_novelty))

class LLMCreativeJudge:
    """使用 LLM 进行创意评估"""
    
    def __init__(self, judge_model: str = "gpt-4o"):
        self.judge_model = judge_model

    def score_creativity(self, prompt: str, response: str) -> dict:
        judge_prompt = f"""你是一位创意写作评审专家。请对以下 AI 生成内容的【创意性】打分。

原始题目：{prompt}
生成内容：{response}

评分标准（0-10分）：
- 9-10: 创意极为独特，角度前所未见，令人惊喜
- 7-8: 创意明显，有新颖角度
- 5-6: 有一定新意，但整体较常见
- 3-4: 创意薄弱，多为套路
- 0-2: 完全老套，毫无创意

请输出 JSON 格式：
{{"score": <0-10>, "reasoning": "<50字以内理由>", "highlights": ["<亮点1>", "<亮点2>"]}}"""

        result = self._call_judge(judge_prompt)
        return result

    def score_multi_dimension(self, prompt: str, response: str) -> dict:
        judge_prompt = f"""你是一位创意写作评审专家。请对以下 AI 生成内容进行多维度评分。

原始题目：{prompt}
生成内容：{response}

请对以下 5 个维度各打 0-10 分：
1. creativity（创意性）：角度是否新颖独特
2. language（语言质量）：用词是否精准优美
3. structure（结构组织）：篇章布局是否精巧
4. emotion（情感共鸣）：是否有感染力
5. relevance（主题契合）：是否深度回应题目

输出 JSON：
{{
  "creativity": {{"score": <0-10>, "note": "<简评>"}},
  "language": {{"score": <0-10>, "note": "<简评>"}},
  "structure": {{"score": <0-10>, "note": "<简评>"}},
  "emotion": {{"score": <0-10>, "note": "<简评>"}},
  "relevance": {{"score": <0-10>, "note": "<简评>"}},
  "overall": {{"score": <0-10>, "summary": "<总体评价>"}}
}}"""

        return self._call_judge(judge_prompt)

    def pairwise_compare(self, prompt: str, response_a: str, response_b: str) -> dict:
        judge_prompt = f"""你是创意写作评审专家。请比较以下两个 AI 生成内容哪个更好。

题目：{prompt}

内容 A：{response_a}
内容 B：{response_b}

评判维度：创意性、语言、结构、情感、主题契合。

输出 JSON：
{{
  "winner": "A" | "B" | "tie",
  "confidence": <0-1>,
  "reasoning": "<100字以内理由>"
}}"""

        return self._call_judge(judge_prompt)

    def _call_judge(self, prompt: str) -> dict:
        # 实际实现调用 LLM API
        import json
        raw = call_llm(self.judge_model, prompt)
        try:
            return json.loads(raw)
        except json.JSONDecodeError:
            # 尝试提取 JSON
            import re
            match = re.search(r'\{.*\}', raw, re.DOTALL)
            if match:
                return json.loads(match.group())
            return {"error": "Failed to parse judge output", "raw": raw}

class BiasMitigatedJudge:
    """带偏差缓解的 LLM Judge"""
    
    def __init__(self, judge_models: list[str]):
        self.judges = judge_models  # 多个不同模型交叉评估

    def evaluate(self, prompt: str, responses: list[str]) -> dict:
        results = []
        for judge in self.judges:
            # 每个 judge 看到不同顺序的 responses
            import random
            ordered = list(enumerate(responses))
            random.shuffle(ordered)
            
            judge_result = self._evaluate_with_judge(judge, prompt, ordered)
            results.append(judge_result)
        
        # 聚合多个 judge 的结果
        return self._aggregate(results)

    def _aggregate(self, results: list[dict]) -> dict:
        """多 judge 结果聚合"""
        from collections import defaultdict
        scores = defaultdict(list)
        for r in results:
            for resp_id, score in r["scores"].items():
                scores[resp_id].append(score)
        
        final = {}
        for resp_id, score_list in scores.items():
            final[resp_id] = {
                "mean_score": sum(score_list) / len(score_list),
                "std": (sum((s - sum(score_list)/len(score_list))**2 for s in score_list) / len(score_list)) ** 0.5,
                "individual_scores": score_list,
            }
        return final

class CreativeEvalFramework:
    """端到端创意评估框架"""
    
    def __init__(self, config: dict):
        self.human_rubric = RubricScorer(RUBRIC)
        self.diversity = DiversityMetrics()
        self.novelty = NoveltyMetrics(config["reference_corpus"])
        self.llm_judge = LLMCreativeJudge(config["judge_model"])
        
    def evaluate(self, prompt: str, response: str, 
                 human_scores: dict = None) -> dict:
        results = {}
        
        # 1. 自动化指标
        results["diversity"] = {
            "distinct_1": self.diversity.distinct_n(response, 1),
            "distinct_2": self.diversity.distinct_n(response, 2),
            "entropy_2": self.diversity.entropy_n(response, 2),
        }
        results["novelty"] = {
            "score": self.novelty.novelty_score(response),
            "percentile": self.novelty.novelty_percentile(response),
        }
        
        # 2. LLM-as-Judge 多维评分
        results["llm_judge"] = self.llm_judge.score_multi_dimension(prompt, response)
        
        # 3. 人工评分（如有）
        if human_scores:
            results["human"] = self.human_rubric.compute_weighted_score(human_scores)
        
        # 4. 综合分数
        results["composite"] = self._compute_composite(results)
        return results

    def _compute_composite(self, results: dict) -> float:
        """计算综合创意分数"""
        weights = {
            "novelty": 0.25,
            "llm_creativity": 0.25,
            "llm_language": 0.15,
            "llm_structure": 0.10,
            "llm_emotion": 0.15,
            "llm_relevance": 0.10,
        }
        score = 0
        score += results["novelty"]["score"] * weights["novelty"] * 10  # 归一化到 0-10
        llm = results["llm_judge"]
        for dim, w in [("creativity", 0.25), ("language", 0.15), 
                       ("structure", 0.10), ("emotion", 0.15), ("relevance", 0.10)]:
            if dim in llm:
                score += llm[dim]["score"] * w
        return round(score, 2)