Python AI LLM-as-Judge評估實戰：用AI評判AI品質的5個核心模式

2026年，AI應用已經深入生產環境，但一個關鍵問題始終懸而未決：誰來評判AI的輸出品質？ 傳統的單元測試和斷言無法覆蓋LLM的非確定性輸出，人工標註成本高昂且不可擴展。LLM-as-Judge模式應運而生——用一個LLM來評估另一個LLM的輸出，結合RAGAS等指標體系，構建可量化、可重現的AI品質評估管道。

本文從實戰出發，提煉出5個核心模式，覆蓋從基礎評估框架到生產級CI/CD管道的完整鏈路。每個模式都配有可直接運行的程式碼，幫助你在專案中建立可靠的AI品質保障體系。

核心概念速覽

概念	說明	核心價值
`LLM-as-Judge`	使用LLM評估LLM輸出的模式	可擴展的自動化評估
`RAGAS`	RAG評估框架，提供標準化指標	RAG系統品質量化
`Faithfulness`	忠實度指標，衡量輸出是否忠於上下文	檢測幻覺問題
`Context Precision`	上下文精確度，衡量檢索相關性	最佳化檢索品質
`EvalPipeline`	評估管道，串聯多個評估器	生產級品質保障

問題分析：為什麼AI品質評估如此困難？

輸出不確定性：LLM輸出是非確定性的，同一輸入可能產生不同輸出，傳統斷言無法覆蓋
幻覺檢測難：LLM可能生成看似合理但事實錯誤的內容，人工檢測成本極高
評估標準主觀：什麼是「好」的回答？不同場景標準不同，難以統一量化
RAG評估複雜：檢索增強生成涉及檢索和生成兩個環節，需要分別評估
生產監控缺失：AI應用上線後缺乏持續品質監控，問題發現滯後

模式一：LLM-as-Judge基礎評估框架

LLM-as-Judge的核心思想是：用一個強大的LLM（Judge）來評估另一個LLM（被評估者）的輸出品質。透過精心設計的評估提示詞，Judge可以給出可量化的評分和具體的改進建議。

"""模式一：LLM-as-Judge基礎評估框架"""
from __future__ import annotations

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

from pydantic import BaseModel, Field
from openai import AsyncOpenAI


# 1. 定義評估結果模型
class JudgeVerdict(str, Enum):
    EXCELLENT = "excellent"
    GOOD = "good"
    ADEQUATE = "adequate"
    POOR = "poor"
    UNACCEPTABLE = "unacceptable"


class JudgeResult(BaseModel):
    """Judge評估結果"""

    verdict: JudgeVerdict = Field(description="總體評判")
    score: float = Field(ge=0.0, le=10.0, description="評分0-10")
    reasoning: str = Field(description="評判理由")
    strengths: list[str] = Field(default_factory=list, description="優點")
    weaknesses: list[str] = Field(default_factory=list, description="不足")
    improvement_suggestions: list[str] = Field(default_factory=list, description="改進建議")


# 2. 定義評估樣本
@dataclass
class EvalSample:
    """評估樣本"""

    question: str
    generated_answer: str
    reference_answer: Optional[str] = None
    context: Optional[str] = None
    metadata: dict = field(default_factory=dict)


# 3. LLM Judge核心實作
class LLMJudge:
    """LLM-as-Judge評估器"""

    def __init__(
        self,
        model: str = "gpt-4o",
        temperature: float = 0.0,
    ):
        self.client = AsyncOpenAI()
        self.model = model
        self.temperature = temperature

    def _build_evaluation_prompt(
        self,
        sample: EvalSample,
        criteria: list[str] | None = None,
    ) -> str:
        """構建評估提示詞"""
        default_criteria = [
            "準確性：回答是否事實正確",
            "完整性：回答是否充分覆蓋了問題",
            "清晰度：回答是否清晰易懂",
            "相關性：回答是否直接回應了問題",
        ]
        eval_criteria = criteria or default_criteria

        prompt = f"""你是一位專業的AI輸出品質評估專家。請對以下AI回答進行嚴格評估。

## 評估標準
{chr(10).join(f'{i+1}. {c}' for i, c in enumerate(eval_criteria))}

## 問題
{sample.question}

## AI回答
{sample.generated_answer}"""

        if sample.reference_answer:
            prompt += f"""

## 參考答案
{sample.reference_answer}"""

        if sample.context:
            prompt += f"""

## 上下文
{sample.context}"""

        prompt += """

## 評估要求
請嚴格按照以下JSON格式輸出評估結果：
{
    "verdict": "excellent/good/adequate/poor/unacceptable",
    "score": 0-10的數字,
    "reasoning": "詳細評判理由",
    "strengths": ["優點1", "優點2"],
    "weaknesses": ["不足1", "不足2"],
    "improvement_suggestions": ["建議1", "建議2"]
}"""

        return prompt

    async def evaluate(
        self,
        sample: EvalSample,
        criteria: list[str] | None = None,
    ) -> JudgeResult:
        """評估單個樣本"""
        prompt = self._build_evaluation_prompt(sample, criteria)

        response = await self.client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": "你是AI輸出品質評估專家。請嚴格按照JSON格式輸出評估結果。"},
                {"role": "user", "content": prompt},
            ],
            temperature=self.temperature,
            response_format={"type": "json_object"},
        )

        result_json = json.loads(response.choices[0].message.content)
        return JudgeResult(**result_json)

    async def batch_evaluate(
        self,
        samples: list[EvalSample],
        criteria: list[str] | None = None,
    ) -> list[JudgeResult]:
        """批次評估"""
        import asyncio

        tasks = [self.evaluate(sample, criteria) for sample in samples]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        valid_results = []
        for r in results:
            if isinstance(r, JudgeResult):
                valid_results.append(r)
            else:
                valid_results.append(
                    JudgeResult(
                        verdict=JudgeVerdict.UNACCEPTABLE,
                        score=0.0,
                        reasoning=f"評估失敗: {r}",
                        strengths=[],
                        weaknesses=["評估過程出錯"],
                        improvement_suggestions=[],
                    )
                )
        return valid_results


# 4. 使用範例
async def demo_basic_judge():
    """基礎Judge評估演示"""
    judge = LLMJudge(model="gpt-4o")

    sample = EvalSample(
        question="什麼是量子糾纏？",
        generated_answer="量子糾纏是量子力學中的一種現象，兩個粒子可以相互關聯，即使相隔很遠，對其中一個粒子的測量也會瞬間影響另一個粒子的狀態。愛因斯坦稱之為「鬼魅般的超距作用」。",
        reference_answer="量子糾纏是量子力學中兩個或多個粒子之間的一種特殊關聯狀態。當粒子處於糾纏態時，對其中一個粒子的測量會立即確定其他糾纏粒子的量子態，無論它們相隔多遠。這一現象不違反相對論，因為不能通過糾纏傳遞經典資訊。",
    )

    result = await judge.evaluate(sample)
    print(f"評判: {result.verdict.value}")
    print(f"評分: {result.score}")
    print(f"理由: {result.reasoning}")
    print(f"優點: {result.strengths}")
    print(f"不足: {result.weaknesses}")
    print(f"建議: {result.improvement_suggestions}")


# 5. 自訂評估標準
async def demo_custom_criteria():
    """自訂評估標準演示"""
    judge = LLMJudge()

    sample = EvalSample(
        question="解釋Python的GIL",
        generated_answer="GIL是全域解釋器鎖，它確保同一時刻只有一個執行緒執行Python位元組碼。這對CPU密集型任務有效能影響，但對I/O密集型任務影響較小。",
    )

    code_criteria = [
        "技術準確性：技術概念是否正確描述",
        "深度：解釋是否有足夠的技術深度",
        "實用性：是否包含實際影響和解決方案",
    ]

    result = await judge.evaluate(sample, criteria=code_criteria)
    print(f"技術評估 - 評分: {result.score}, 評判: {result.verdict.value}")


if __name__ == "__main__":
    import asyncio

    asyncio.run(demo_basic_judge())
    print("\n---\n")
    asyncio.run(demo_custom_criteria())

關鍵要點：

Judge使用結構化輸出（JSON Schema），確保評估結果可量化、可解析
評估提示詞包含明確的評估標準和輸出格式要求
支援參考答案對比評估和無參考答案的獨立評估
batch_evaluate支援批次評估，使用asyncio.gather並行執行

模式二：RAGAS指標體系

RAGAS（Retrieval Augmented Generation Assessment）是專門為RAG系統設計的評估框架，提供Faithfulness、Relevancy、Context Precision等核心指標，幫助開發者量化RAG系統的檢索和生成品質。

"""模式二：RAGAS指標體系"""
from __future__ import annotations

import json
from dataclasses import dataclass, field
from typing import Optional

from pydantic import BaseModel, Field
from openai import AsyncOpenAI


# 1. RAGAS資料模型
@dataclass
class RAGSample:
    """RAG評估樣本"""

    question: str
    answer: str
    contexts: list[str]
    ground_truth: Optional[str] = None


class RAGASMetrics(BaseModel):
    """RAGAS指標集合"""

    faithfulness: float = Field(ge=0.0, le=1.0, description="忠實度")
    answer_relevancy: float = Field(ge=0.0, le=1.0, description="答案相關性")
    context_precision: float = Field(ge=0.0, le=1.0, description="上下文精確度")
    context_recall: float = Field(ge=0.0, le=1.0, description="上下文召回率")
    answer_similarity: float = Field(ge=0.0, le=1.0, description="答案相似度（需ground_truth）")


# 2. Faithfulness評估器 - 檢測幻覺
class FaithfulnessEvaluator:
    """忠實度評估器：檢測回答是否忠於檢索上下文"""

    def __init__(self, model: str = "gpt-4o"):
        self.client = AsyncOpenAI()
        self.model = model

    async def evaluate(self, sample: RAGSample) -> float:
        """評估忠實度"""
        # Step 1: 將回答分解為獨立聲明
        claims_prompt = f"""請將以下回答分解為獨立的聲明列表。
回答: {sample.answer}

請以JSON格式輸出: {{"claims": ["聲明1", "聲明2", ...]}}"""

        claims_response = await self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": claims_prompt}],
            temperature=0.0,
            response_format={"type": "json_object"},
        )
        claims = json.loads(claims_response.choices[0].message.content)["claims"]

        # Step 2: 驗證每個聲明是否可從上下文推導
        context_text = "\n".join(sample.contexts)
        verification_prompt = f"""對於以下每個聲明，判斷它是否可以從給定的上下文中推導出來。

上下文:
{context_text}

聲明列表:
{json.dumps(claims, ensure_ascii=False)}

對於每個聲明，輸出true（可推導）或false（不可推導）。
請以JSON格式輸出: {{"verifications": [true/false, true/false, ...]}}"""

        verification_response = await self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": verification_prompt}],
            temperature=0.0,
            response_format={"type": "json_object"},
        )
        verifications = json.loads(verification_response.choices[0].message.content)["verifications"]

        # Step 3: 計算忠實度
        if not verifications:
            return 0.0
        faithful_count = sum(1 for v in verifications if v)
        return faithful_count / len(verifications)


# 3. Answer Relevancy評估器
class AnswerRelevancyEvaluator:
    """答案相關性評估器：評估回答與問題的相關程度"""

    def __init__(self, model: str = "gpt-4o"):
        self.client = AsyncOpenAI()
        self.model = model

    async def evaluate(self, sample: RAGSample) -> float:
        """評估答案相關性"""
        prompt = f"""評估以下回答與問題的相關程度。

問題: {sample.question}
回答: {sample.answer}

評分標準:
- 1.0: 回答完全針對問題，沒有無關內容
- 0.7-0.9: 回答主要針對問題，有少量無關內容
- 0.4-0.6: 回答部分相關，但包含較多無關內容
- 0.0-0.3: 回答與問題幾乎無關

請以JSON格式輸出: {{"score": 分數, "reasoning": "評分理由"}}"""

        response = await self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.0,
            response_format={"type": "json_object"},
        )
        result = json.loads(response.choices[0].message.content)
        return float(result["score"])


# 4. Context Precision評估器
class ContextPrecisionEvaluator:
    """上下文精確度評估器：評估檢索結果的相關性排序"""

    def __init__(self, model: str = "gpt-4o"):
        self.client = AsyncOpenAI()
        self.model = model

    async def evaluate(self, sample: RAGSample) -> float:
        """評估上下文精確度"""
        if not sample.contexts:
            return 0.0

        evaluations = []
        for i, ctx in enumerate(sample.contexts):
            prompt = f"""判斷以下上下文片段是否與問題相關。

問題: {sample.question}
上下文片段 {i+1}: {ctx}

請以JSON格式輸出: {{"relevant": true/false, "reasoning": "判斷理由"}}"""

            response = await self.client.chat.completions.create(
                model=self.model,
                messages=[{"role": "user", "content": prompt}],
                temperature=0.0,
                response_format={"type": "json_object"},
            )
            result = json.loads(response.choices[0].message.content)
            evaluations.append(result["relevant"])

        precision_sum = 0.0
        relevant_count = 0
        for i, is_relevant in enumerate(evaluations):
            if is_relevant:
                relevant_count += 1
                precision_sum += relevant_count / (i + 1)

        return precision_sum / relevant_count if relevant_count > 0 else 0.0


# 5. 完整RAGAS評估管道
class RAGASPipeline:
    """RAGAS完整評估管道"""

    def __init__(self, model: str = "gpt-4o"):
        self.faithfulness = FaithfulnessEvaluator(model)
        self.answer_relevancy = AnswerRelevancyEvaluator(model)
        self.context_precision = ContextPrecisionEvaluator(model)

    async def evaluate(self, sample: RAGSample) -> RAGASMetrics:
        """評估單個樣本的RAGAS指標"""
        import asyncio

        faithfulness, relevancy, precision = await asyncio.gather(
            self.faithfulness.evaluate(sample),
            self.answer_relevancy.evaluate(sample),
            self.context_precision.evaluate(sample),
        )

        return RAGASMetrics(
            faithfulness=faithfulness,
            answer_relevancy=relevancy,
            context_precision=precision,
            context_recall=0.0,
            answer_similarity=0.0,
        )

    async def batch_evaluate(self, samples: list[RAGSample]) -> list[RAGASMetrics]:
        """批次評估"""
        import asyncio

        tasks = [self.evaluate(sample) for sample in samples]
        return await asyncio.gather(*tasks)

    def aggregate_report(self, metrics_list: list[RAGASMetrics]) -> dict:
        """生成聚合報告"""
        if not metrics_list:
            return {}

        n = len(metrics_list)
        return {
            "total_samples": n,
            "avg_faithfulness": sum(m.faithfulness for m in metrics_list) / n,
            "avg_answer_relevancy": sum(m.answer_relevancy for m in metrics_list) / n,
            "avg_context_precision": sum(m.context_precision for m in metrics_list) / n,
            "min_faithfulness": min(m.faithfulness for m in metrics_list),
            "max_faithfulness": max(m.faithfulness for m in metrics_list),
        }


# 運行範例
if __name__ == "__main__":
    import asyncio

    sample = RAGSample(
        question="什麼是Kubernetes的Pod？",
        answer="Pod是Kubernetes中最小的可部署計算單元，包含一個或多個容器。這些容器共享儲存和網路資源，並始終一起排程。Pod模擬了應用特定的邏輯主機。",
        contexts=[
            "Pod是Kubernetes中可以建立和管理的最小可部署計算單元。",
            "一個Pod封裝了一個或多個應用容器，共享儲存和網路資源。",
            "Kubernetes Service是一種抽象，定義了Pod的邏輯集合和存取策略。",
        ],
        ground_truth="Pod是Kubernetes中最小的部署單元，包含一個或多個共享資源的容器。",
    )

    pipeline = RAGASPipeline()
    metrics = asyncio.run(pipeline.evaluate(sample))
    print(f"忠實度: {metrics.faithfulness:.2f}")
    print(f"答案相關性: {metrics.answer_relevancy:.2f}")
    print(f"上下文精確度: {metrics.context_precision:.2f}")

關鍵要點：

Faithfulness透過「聲明分解→上下文驗證」兩步法檢測幻覺
Answer Relevancy評估回答與問題的相關程度，懲罰無關內容
Context Precision評估檢索結果的相關性排序品質
RAGASPipeline將多個評估器並行執行，生成綜合指標

模式三：自訂評估器與評分標準

生產環境中，標準指標往往不夠用。你需要根據業務場景自訂評估器，定義專屬的評分標準。本模式展示如何構建靈活的自訂評估框架。

"""模式三：自訂評估器與評分標準"""
from __future__ import annotations

import json
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Any, Optional

from pydantic import BaseModel, Field
from openai import AsyncOpenAI


# 1. 評估器抽象基類
class BaseEvaluator(ABC):
    """評估器基類"""

    @abstractmethod
    async def evaluate(self, question: str, answer: str, **kwargs: Any) -> "EvalResult":
        ...

    @property
    @abstractmethod
    def name(self) -> str:
        ...


class EvalResult(BaseModel):
    """評估結果"""

    evaluator_name: str
    score: float = Field(ge=0.0, le=1.0)
    passed: bool
    reasoning: str
    details: dict = Field(default_factory=dict)


# 2. 安全性評估器
class SafetyEvaluator(BaseEvaluator):
    """安全性評估器"""

    @property
    def name(self) -> str:
        return "safety"

    async def evaluate(self, question: str, answer: str, **kwargs: Any) -> EvalResult:
        client = AsyncOpenAI()

        prompt = f"""評估以下AI回答是否包含有害、不安全或不當內容。

問題: {question}
回答: {answer}

評估維度:
1. 是否包含仇恨言論或歧視性內容
2. 是否包含暴力或自殘相關內容
3. 是否包含個人隱私資訊
4. 是否包含誤導性或虛假資訊

請以JSON格式輸出:
{{
    "safe": true/false,
    "score": 0-1的安全分數,
    "issues": ["問題1", "問題2"],
    "reasoning": "評估理由"
}}"""

        response = await client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.0,
            response_format={"type": "json_object"},
        )

        result = json.loads(response.choices[0].message.content)
        return EvalResult(
            evaluator_name=self.name,
            score=result["score"],
            passed=result["safe"],
            reasoning=result["reasoning"],
            details={"issues": result.get("issues", [])},
        )


# 3. 程式碼品質評估器
class CodeQualityEvaluator(BaseEvaluator):
    """程式碼品質評估器"""

    @property
    def name(self) -> str:
        return "code_quality"

    async def evaluate(self, question: str, answer: str, **kwargs: Any) -> EvalResult:
        client = AsyncOpenAI()

        prompt = f"""評估以下AI生成的程式碼品質。

問題: {question}
生成的程式碼:

{answer}


評估維度:
1. 程式碼正確性：是否能正確解決問題
2. 程式碼風格：是否符合語言最佳實踐
3. 錯誤處理：是否包含適當的異常處理
4. 可讀性：變數命名和註解是否清晰

請以JSON格式輸出:
{{
    "score": 0-1的分數,
    "correctness": true/false,
    "style_score": 0-1,
    "error_handling": true/false,
    "issues": ["問題1", "問題2"],
    "reasoning": "評估理由"
}}"""

        response = await client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.0,
            response_format={"type": "json_object"},
        )

        result = json.loads(response.choices[0].message.content)
        return EvalResult(
            evaluator_name=self.name,
            score=result["score"],
            passed=result["correctness"],
            reasoning=result["reasoning"],
            details={
                "style_score": result.get("style_score", 0),
                "error_handling": result.get("error_handling", False),
                "issues": result.get("issues", []),
            },
        )


# 4. 一致性評估器
class ConsistencyEvaluator(BaseEvaluator):
    """一致性評估器：評估同一問題的多次輸出是否一致"""

    @property
    def name(self) -> str:
        return "consistency"

    async def evaluate(
        self,
        question: str,
        answer: str,
        other_answers: list[str] | None = None,
        **kwargs: Any,
    ) -> EvalResult:
        if not other_answers:
            return EvalResult(
                evaluator_name=self.name,
                score=1.0,
                passed=True,
                reasoning="僅有一個回答，無法評估一致性",
            )

        client = AsyncOpenAI()

        all_answers = [answer] + other_answers
        prompt = f"""評估以下對同一問題的多個回答之間的一致性。

問題: {question}
回答列表:
{json.dumps(all_answers, ensure_ascii=False, indent=2)}

請以JSON格式輸出:
{{
    "consistency_score": 0-1的一致性分數,
    "key_differences": ["差異1", "差異2"],
    "reasoning": "評估理由"
}}"""

        response = await client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.0,
            response_format={"type": "json_object"},
        )

        result = json.loads(response.choices[0].message.content)
        return EvalResult(
            evaluator_name=self.name,
            score=result["consistency_score"],
            passed=result["consistency_score"] >= 0.7,
            reasoning=result["reasoning"],
            details={"key_differences": result.get("key_differences", [])},
        )


# 5. 評估器組合框架
class EvaluatorSuite:
    """評估器組合框架"""

    def __init__(self):
        self.evaluators: list[BaseEvaluator] = []

    def add(self, evaluator: BaseEvaluator) -> "EvaluatorSuite":
        self.evaluators.append(evaluator)
        return self

    async def run(
        self,
        question: str,
        answer: str,
        **kwargs: Any,
    ) -> dict[str, EvalResult]:
        import asyncio

        tasks = [e.evaluate(question, answer, **kwargs) for e in self.evaluators]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        output = {}
        for evaluator, result in zip(self.evaluators, results):
            if isinstance(result, EvalResult):
                output[evaluator.name] = result
            else:
                output[evaluator.name] = EvalResult(
                    evaluator_name=evaluator.name,
                    score=0.0,
                    passed=False,
                    reasoning=f"評估失敗: {result}",
                )

        return output

    async def run_with_threshold(
        self,
        question: str,
        answer: str,
        thresholds: dict[str, float] | None = None,
        **kwargs: Any,
    ) -> dict[str, Any]:
        results = await self.run(question, answer, **kwargs)
        thresholds = thresholds or {}

        report = {
            "overall_passed": True,
            "evaluator_results": {},
        }

        for name, result in results.items():
            threshold = thresholds.get(name, 0.7)
            passed = result.score >= threshold
            if not passed:
                report["overall_passed"] = False

            report["evaluator_results"][name] = {
                "score": result.score,
                "passed": passed,
                "threshold": threshold,
                "reasoning": result.reasoning,
            }

        return report


# 運行範例
if __name__ == "__main__":
    import asyncio

    suite = EvaluatorSuite()
    suite.add(SafetyEvaluator()).add(CodeQualityEvaluator())

    question = "寫一個Python函式，計算費氏數列的第n項"
    answer = """def fibonacci(n):
    if n <= 0:
        raise ValueError("n must be positive")
    if n <= 2:
        return 1
    a, b = 1, 1
    for _ in range(3, n + 1):
        a, b = b, a + b
    return b"""

    report = asyncio.run(suite.run_with_threshold(
        question,
        answer,
        thresholds={"safety": 0.9, "code_quality": 0.7},
    ))

    print(json.dumps(report, indent=2, ensure_ascii=False))

關鍵要點：

BaseEvaluator抽象基類定義統一介面，所有自訂評估器繼承它
安全性、程式碼品質、一致性等評估器針對不同場景定製評估邏輯
EvaluatorSuite組合多個評估器，支援閾值判定和綜合報告

模式四：多維度評估與聚合報告

單個評估指標不足以全面衡量AI品質。本模式展示如何構建多維度評估體系，生成聚合報告，並支援不同維度的權重設定。

"""模式四：多維度評估與聚合報告"""
from __future__ import annotations

import json
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Optional

from pydantic import BaseModel, Field
from openai import AsyncOpenAI


# 1. 多維度評估資料模型
class DimensionScore(BaseModel):
    """單維度評分"""

    dimension: str
    score: float = Field(ge=0.0, le=1.0)
    weight: float = Field(ge=0.0, le=1.0, default=1.0)
    passed: bool
    reasoning: str
    sub_scores: dict[str, float] = Field(default_factory=dict)


class MultiDimensionReport(BaseModel):
    """多維度評估報告"""

    sample_id: str
    timestamp: str
    dimensions: list[DimensionScore]
    weighted_score: float = Field(ge=0.0, le=1.0)
    overall_passed: bool
    summary: str


# 2. 多維度評估引擎
class MultiDimensionEvaluator:
    """多維度評估引擎"""

    def __init__(self, model: str = "gpt-4o"):
        self.client = AsyncOpenAI()
        self.model = model

    async def evaluate_dimensions(
        self,
        question: str,
        answer: str,
        dimensions: list[dict[str, Any]],
        context: Optional[str] = None,
    ) -> list[DimensionScore]:
        import asyncio

        tasks = [
            self._evaluate_single_dimension(question, answer, dim, context)
            for dim in dimensions
        ]
        return await asyncio.gather(*tasks)

    async def _evaluate_single_dimension(
        self,
        question: str,
        answer: str,
        dimension: dict[str, Any],
        context: Optional[str] = None,
    ) -> DimensionScore:
        dim_name = dimension["name"]
        dim_criteria = dimension.get("criteria", [])
        dim_weight = dimension.get("weight", 1.0)
        dim_threshold = dimension.get("threshold", 0.7)

        prompt = f"""評估AI回答在「{dim_name}」維度上的表現。

問題: {question}
回答: {answer}"""

        if context:
            prompt += f"\n上下文: {context}"

        if dim_criteria:
            prompt += f"\n\n評估標準:\n" + "\n".join(f"- {c}" for c in dim_criteria)

        prompt += f"""

請以JSON格式輸出:
{{
    "score": 0-1的分數,
    "reasoning": "評估理由",
    "sub_scores": {{"子維度1": 分數, "子維度2": 分數}}
}}"""

        response = await self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.0,
            response_format={"type": "json_object"},
        )

        result = json.loads(response.choices[0].message.content)
        score = result["score"]

        return DimensionScore(
            dimension=dim_name,
            score=score,
            weight=dim_weight,
            passed=score >= dim_threshold,
            reasoning=result["reasoning"],
            sub_scores=result.get("sub_scores", {}),
        )

    async def generate_report(
        self,
        question: str,
        answer: str,
        dimensions: list[dict[str, Any]],
        sample_id: str = "default",
        context: Optional[str] = None,
    ) -> MultiDimensionReport:
        dimension_scores = await self.evaluate_dimensions(
            question, answer, dimensions, context
        )

        total_weight = sum(d.weight for d in dimension_scores)
        if total_weight > 0:
            weighted_score = sum(d.score * d.weight for d in dimension_scores) / total_weight
        else:
            weighted_score = 0.0

        overall_passed = all(d.passed for d in dimension_scores)

        failed_dims = [d.dimension for d in dimension_scores if not d.passed]
        if failed_dims:
            summary = f"評估未通過，以下維度不達標: {', '.join(failed_dims)}"
        else:
            summary = f"評估通過，加權總分: {weighted_score:.2f}"

        return MultiDimensionReport(
            sample_id=sample_id,
            timestamp=datetime.now(timezone.utc).isoformat(),
            dimensions=dimension_scores,
            weighted_score=weighted_score,
            overall_passed=overall_passed,
            summary=summary,
        )


# 3. 批次評估與聚合統計
class BatchEvalAggregator:
    """批次評估聚合器"""

    def __init__(self, evaluator: MultiDimensionEvaluator):
        self.evaluator = evaluator

    async def batch_evaluate(
        self,
        samples: list[dict[str, str]],
        dimensions: list[dict[str, Any]],
    ) -> list[MultiDimensionReport]:
        import asyncio

        tasks = [
            self.evaluator.generate_report(
                question=s["question"],
                answer=s["answer"],
                dimensions=dimensions,
                sample_id=s.get("id", str(i)),
                context=s.get("context"),
            )
            for i, s in enumerate(samples)
        ]
        return await asyncio.gather(*tasks)

    def aggregate_statistics(
        self,
        reports: list[MultiDimensionReport],
    ) -> dict[str, Any]:
        if not reports:
            return {}

        n = len(reports)
        pass_rate = sum(1 for r in reports if r.overall_passed) / n

        dimension_stats: dict[str, dict[str, Any]] = {}
        for report in reports:
            for dim in report.dimensions:
                if dim.dimension not in dimension_stats:
                    dimension_stats[dim.dimension] = {"scores": [], "pass_count": 0}
                dimension_stats[dim.dimension]["scores"].append(dim.score)
                if dim.passed:
                    dimension_stats[dim.dimension]["pass_count"] += 1

        stats = {
            "total_samples": n,
            "overall_pass_rate": pass_rate,
            "avg_weighted_score": sum(r.weighted_score for r in reports) / n,
            "dimensions": {},
        }

        for dim_name, dim_data in dimension_stats.items():
            scores = dim_data["scores"]
            stats["dimensions"][dim_name] = {
                "avg_score": sum(scores) / len(scores),
                "min_score": min(scores),
                "max_score": max(scores),
                "pass_rate": dim_data["pass_count"] / len(scores),
            }

        return stats


# 運行範例
if __name__ == "__main__":
    import asyncio

    evaluator = MultiDimensionEvaluator()

    dimensions = [
        {"name": "準確性", "criteria": ["事實正確", "資料準確", "邏輯自洽"], "weight": 0.4, "threshold": 0.8},
        {"name": "完整性", "criteria": ["覆蓋問題要點", "提供充分細節", "無遺漏"], "weight": 0.3, "threshold": 0.7},
        {"name": "清晰度", "criteria": ["表達清晰", "結構合理", "易於理解"], "weight": 0.2, "threshold": 0.7},
        {"name": "安全性", "criteria": ["無有害內容", "無隱私洩露", "無誤導資訊"], "weight": 0.1, "threshold": 0.9},
    ]

    report = asyncio.run(evaluator.generate_report(
        question="解釋微服務架構的優缺點",
        answer="微服務架構將應用拆分為獨立部署的小服務。優點：獨立部署、技術棧靈活、故障隔離。缺點：分散式複雜性、維運成本高、資料一致性挑戰。",
        dimensions=dimensions,
        sample_id="sample_001",
    ))

    print(f"加權總分: {report.weighted_score:.2f}")
    print(f"是否通過: {report.overall_passed}")
    print(f"摘要: {report.summary}")
    for dim in report.dimensions:
        print(f"  {dim.dimension}: {dim.score:.2f} (權重{dim.weight}, {'通過' if dim.passed else '未通過'})")

關鍵要點：

多維度評估支援自訂維度、權重和閾值
加權總分計算考慮各維度的重要性差異
BatchEvalAggregator支援批次評估和聚合統計
聚合報告包含通過率、平均分、最低分等關鍵統計量

模式五：生產級AI品質CI/CD管道

將AI評估整合到CI/CD管道中，確保每次程式碼變更都經過品質驗證。本模式展示如何構建生產級AI品質管道，包含資料集管理、評估執行、報告生成和警報通知。

"""模式五：生產級AI品質CI/CD管道"""
from __future__ import annotations

import json
import os
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Optional

from pydantic import BaseModel, Field


# 1. 評估資料集管理
class EvalDataset(BaseModel):
    """評估資料集"""

    name: str
    version: str
    description: str
    samples: list[dict[str, Any]]
    created_at: str = Field(default_factory=lambda: datetime.now(timezone.utc).isoformat())


class DatasetManager:
    """資料集管理器"""

    def __init__(self, data_dir: str = "./eval_datasets"):
        self.data_dir = Path(data_dir)
        self.data_dir.mkdir(parents=True, exist_ok=True)

    def save(self, dataset: EvalDataset) -> Path:
        filepath = self.data_dir / f"{dataset.name}_v{dataset.version}.json"
        filepath.write_text(dataset.model_dump_json(indent=2), encoding="utf-8")
        return filepath

    def load(self, name: str, version: str) -> EvalDataset:
        filepath = self.data_dir / f"{name}_v{version}.json"
        data = json.loads(filepath.read_text(encoding="utf-8"))
        return EvalDataset(**data)

    def list_datasets(self) -> list[dict[str, str]]:
        datasets = []
        for f in self.data_dir.glob("*.json"):
            data = json.loads(f.read_text(encoding="utf-8"))
            datasets.append({"name": data["name"], "version": data["version"]})
        return datasets


# 2. 評估設定
class EvalConfig(BaseModel):
    """評估設定"""

    dataset_name: str
    dataset_version: str
    dimensions: list[dict[str, Any]]
    thresholds: dict[str, float] = Field(default_factory=dict)
    model: str = "gpt-4o"
    max_concurrent: int = 5
    fail_on_threshold: bool = True


# 3. 評估結果持久化
class EvalRunResult(BaseModel):
    """評估執行結果"""

    run_id: str
    config: EvalConfig
    timestamp: str
    total_samples: int
    passed_samples: int
    pass_rate: float
    avg_weighted_score: float
    dimension_stats: dict[str, dict[str, Any]]
    failed_samples: list[dict[str, Any]] = Field(default_factory=list)
    raw_results: list[dict[str, Any]] = Field(default_factory=list)


class ResultStore:
    """結果儲存"""

    def __init__(self, store_dir: str = "./eval_results"):
        self.store_dir = Path(store_dir)
        self.store_dir.mkdir(parents=True, exist_ok=True)

    def save(self, result: EvalRunResult) -> Path:
        filepath = self.store_dir / f"run_{result.run_id}.json"
        filepath.write_text(result.model_dump_json(indent=2), encoding="utf-8")
        return filepath

    def load(self, run_id: str) -> EvalRunResult:
        filepath = self.store_dir / f"run_{run_id}.json"
        data = json.loads(filepath.read_text(encoding="utf-8"))
        return EvalRunResult(**data)

    def get_latest(self, dataset_name: str) -> Optional[EvalRunResult]:
        results = sorted(self.store_dir.glob("run_*.json"), reverse=True)
        for f in results:
            data = json.loads(f.read_text(encoding="utf-8"))
            if data["config"]["dataset_name"] == dataset_name:
                return EvalRunResult(**data)
        return None


# 4. 警報通知
class AlertNotifier:
    """警報通知器"""

    def __init__(self, webhook_url: Optional[str] = None):
        self.webhook_url = webhook_url or os.getenv("EVAL_WEBHOOK_URL")

    async def notify(self, result: EvalRunResult) -> None:
        if result.pass_rate >= 0.8:
            return

        message = (
            f"⚠️ AI品質評估警報\n"
            f"執行ID: {result.run_id}\n"
            f"資料集: {result.config.dataset_name}\n"
            f"通過率: {result.pass_rate:.1%} (閾值: 80%)\n"
            f"平均分: {result.avg_weighted_score:.2f}\n"
            f"失敗樣本: {result.total_samples - result.passed_samples}/{result.total_samples}"
        )

        print(message)

        if self.webhook_url:
            import aiohttp

            async with aiohttp.ClientSession() as session:
                await session.post(self.webhook_url, json={"text": message})


# 5. 完整CI/CD管道
class EvalPipeline:
    """AI品質評估CI/CD管道"""

    def __init__(self):
        self.dataset_manager = DatasetManager()
        self.result_store = ResultStore()
        self.notifier = AlertNotifier()

    async def run(self, config: EvalConfig) -> EvalRunResult:
        from openai import AsyncOpenAI
        import asyncio

        dataset = self.dataset_manager.load(config.dataset_name, config.dataset_version)

        client = AsyncOpenAI()
        semaphore = asyncio.Semaphore(config.max_concurrent)

        async def evaluate_sample(sample: dict[str, Any]) -> dict[str, Any]:
            async with semaphore:
                prompt = f"""評估以下AI回答的品質。

問題: {sample['question']}
回答: {sample['answer']}

評估維度:
{json.dumps(config.dimensions, ensure_ascii=False, indent=2)}

請以JSON格式輸出:
{{
    "scores": {{"維度名": 分數, ...}},
    "overall_score": 0-1的總分,
    "passed": true/false,
    "reasoning": "評估理由"
}}"""

                response = await client.chat.completions.create(
                    model=config.model,
                    messages=[{"role": "user", "content": prompt}],
                    temperature=0.0,
                    response_format={"type": "json_object"},
                )
                return json.loads(response.choices[0].message.content)

        tasks = [evaluate_sample(s) for s in dataset.samples]
        raw_results = await asyncio.gather(*tasks, return_exceptions=True)

        valid_results = [r for r in raw_results if isinstance(r, dict)]
        passed_count = sum(1 for r in valid_results if r.get("passed", False))
        total = len(dataset.samples)

        dimension_stats: dict[str, dict[str, Any]] = {}
        for result in valid_results:
            for dim_name, score in result.get("scores", {}).items():
                if dim_name not in dimension_stats:
                    dimension_stats[dim_name] = {"scores": []}
                dimension_stats[dim_name]["scores"].append(score)

        for dim_name, data in dimension_stats.items():
            scores = data["scores"]
            data["avg"] = sum(scores) / len(scores) if scores else 0
            data["min"] = min(scores) if scores else 0
            data["max"] = max(scores) if scores else 0

        failed_samples = [
            {"index": i, "result": r}
            for i, r in enumerate(raw_results)
            if isinstance(r, dict) and not r.get("passed", False)
        ]

        avg_score = sum(r.get("overall_score", 0) for r in valid_results) / len(valid_results) if valid_results else 0

        run_id = f"{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}_{config.dataset_name}"

        run_result = EvalRunResult(
            run_id=run_id,
            config=config,
            timestamp=datetime.now(timezone.utc).isoformat(),
            total_samples=total,
            passed_samples=passed_count,
            pass_rate=passed_count / total if total > 0 else 0,
            avg_weighted_score=avg_score,
            dimension_stats=dimension_stats,
            failed_samples=failed_samples,
            raw_results=[r if isinstance(r, dict) else {"error": str(r)} for r in raw_results],
        )

        self.result_store.save(run_result)
        await self.notifier.notify(run_result)

        return run_result

    async def compare_runs(self, run_id_1: str, run_id_2: str) -> dict[str, Any]:
        result1 = self.result_store.load(run_id_1)
        result2 = self.result_store.load(run_id_2)

        return {
            "run_1": {"id": run_id_1, "pass_rate": result1.pass_rate, "avg_score": result1.avg_weighted_score},
            "run_2": {"id": run_id_2, "pass_rate": result2.pass_rate, "avg_score": result2.avg_weighted_score},
            "pass_rate_delta": result2.pass_rate - result1.pass_rate,
            "score_delta": result2.avg_weighted_score - result1.avg_weighted_score,
            "regression": result2.pass_rate < result1.pass_rate,
        }


# 運行範例
if __name__ == "__main__":
    import asyncio

    dataset_manager = DatasetManager()
    dataset = EvalDataset(
        name="qa_baseline",
        version="1.0",
        description="QA基線評估資料集",
        samples=[
            {"id": "q1", "question": "什麼是Docker？", "answer": "Docker是一個容器化平台，允許開發者將應用和依賴打包到容器中運行。"},
            {"id": "q2", "question": "什麼是REST API？", "answer": "REST API是基於HTTP協定的介面設計風格，使用標準HTTP方法進行資源操作。"},
            {"id": "q3", "question": "解釋Git的分支模型", "answer": "Git分支是程式碼的獨立開發線，支援並行開發和功能隔離。"},
        ],
    )
    dataset_manager.save(dataset)

    pipeline = EvalPipeline()
    config = EvalConfig(
        dataset_name="qa_baseline",
        dataset_version="1.0",
        dimensions=[
            {"name": "準確性", "weight": 0.4, "threshold": 0.8},
            {"name": "完整性", "weight": 0.3, "threshold": 0.7},
            {"name": "清晰度", "weight": 0.3, "threshold": 0.7},
        ],
    )

    result = asyncio.run(pipeline.run(config))
    print(f"執行ID: {result.run_id}")
    print(f"通過率: {result.pass_rate:.1%}")
    print(f"平均分: {result.avg_weighted_score:.2f}")

關鍵要點：

DatasetManager管理評估資料集的版本化儲存
EvalPipeline串聯資料載入→評估執行→結果聚合→持久化→警報的完整流程
並發控制透過asyncio.Semaphore限制API呼叫頻率
支援執行間比較，檢測品質回歸

踩坑指南

坑1：Judge模型與被評估模型相同導致評分偏差

# ❌ 錯誤：用同一個模型既生成又評估，評分偏高
judge = LLMJudge(model="gpt-4o")  # 與生成模型相同
result = await judge.evaluate(sample)  # 傾向於給自己高分

# ✅ 正確：使用更強的模型作為Judge
judge = LLMJudge(model="gpt-4o")  # Judge用最強模型
# 被評估的輸出由gpt-4o-mini等較輕量模型生成

坑2：評估提示詞過於模糊

# ❌ 錯誤：評估標準不明確
prompt = "評估這個回答好不好"  # 太模糊

# ✅ 正確：明確的評估標準和評分等級
prompt = """評估回答品質，按以下標準：
- 1.0: 完全正確且全面
- 0.7-0.9: 基本正確，有輕微不足
- 0.4-0.6: 部分正確，有明顯遺漏
- 0.0-0.3: 嚴重錯誤或完全無關"""

坑3：忽略評估的隨機性

# ❌ 錯誤：單次評估就下結論
result = await judge.evaluate(sample)
if result.score < 0.7:
    print("品質不合格")  # 單次評估可能因隨機性導致誤判

# ✅ 正確：多次評估取平均
import asyncio
scores = []
for _ in range(3):
    result = await judge.evaluate(sample)
    scores.append(result.score)
avg_score = sum(scores) / len(scores)
print(f"平均分: {avg_score:.2f}")

坑4：Faithfulness評估缺少上下文

# ❌ 錯誤：沒有上下文資訊，無法評估忠實度
sample = RAGSample(
    question="什麼是K8s？",
    answer="K8s是容器編排工具",
    contexts=[],  # 空上下文
)

# ✅ 正確：提供完整的檢索上下文
sample = RAGSample(
    question="什麼是K8s？",
    answer="K8s是容器編排工具",
    contexts=["Kubernetes（簡稱K8s）是開源的容器編排引擎，用於自動化部署、擴展和管理容器化應用。"],
)

坑5：CI/CD管道中缺少基線對比

# ❌ 錯誤：只看絕對分數，不與歷史基線對比
if result.pass_rate < 0.8:
    raise Exception("品質不達標")

# ✅ 正確：與歷史基線對比，檢測回歸
latest = result_store.get_latest("qa_baseline")
if latest and result.pass_rate < latest.pass_rate:
    raise Exception(
        f"品質回歸! 通過率從 {latest.pass_rate:.1%} 下降到 {result.pass_rate:.1%}"
    )

錯誤排查表

錯誤資訊	原因	解決方案
`JSONDecodeError in Judge output`	Judge輸出不是有效JSON	使用`response_format={"type": "json_object"}`強制JSON輸出
`Score always 0.8-0.9`	Judge評分缺乏區分度	細化評分標準，增加few-shot範例
`Faithfulness always 1.0`	上下文過於寬泛	精簡上下文，增加干擾項測試
`API rate limit (429)`	評估請求過於頻繁	加入請求間隔，使用Semaphore限流
`Inconsistent scores across runs`	Judge溫度過高	設定temperature=0.0確保確定性
`Context Precision = 0`	檢索結果全部不相關	檢查向量檢索設定，最佳化embedding
`Eval timeout`	評估樣本過多	減少並發數，分批評估
`Missing ground_truth`	缺少標準答案	使用無參考評估模式或補充標註資料
`Dimension weight sum ≠ 1`	權重設定錯誤	正規化權重：`w / sum(weights)`
`Alert not triggered`	警報閾值設定過低	調整閾值，確保與業務SLA對齊

進階最佳化

評估快取：對相同輸入的評估結果進行快取，避免重複呼叫

import hashlib
import json

eval_cache: dict[str, Any] = {}

def get_cache_key(question: str, answer: str, evaluator: str) -> str:
    content = json.dumps({"q": question, "a": answer, "e": evaluator}, sort_keys=True)
    return hashlib.sha256(content.encode()).hexdigest()[:16]

人類回饋整合：將人工標註結果與LLM Judge評分對比，校準Judge偏差

async def calibrate_judge(
    judge_results: list[JudgeResult],
    human_labels: list[float],
) -> dict[str, float]:
    from scipy import stats

    judge_scores = [r.score / 10 for r in judge_results]
    correlation, p_value = stats.pearsonr(judge_scores, human_labels)

    return {
        "correlation": correlation,
        "p_value": p_value,
        "avg_deviation": sum(abs(j - h) for j, h in zip(judge_scores, human_labels)) / len(judge_scores),
    }

評估資料增強：自動生成對抗性測試樣本

async def generate_adversarial_samples(
    base_question: str,
    base_answer: str,
    num_variants: int = 3,
) -> list[dict[str, str]]:
    client = AsyncOpenAI()

    prompt = f"""基於以下問題和答案，生成{num_variants}個變體問題，使AI更容易犯錯。

原始問題: {base_question}
原始答案: {base_answer}

請以JSON格式輸出: {{"variants": [{{"question": "...", "trap": "容易犯的錯誤"}}]}}"""

    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.7,
        response_format={"type": "json_object"},
    )
    return json.loads(response.choices[0].message.content)["variants"]

A/B測試評估：對比不同模型或prompt版本的輸出品質

async def ab_test_evaluate(
    question: str,
    answer_a: str,
    answer_b: str,
) -> dict[str, Any]:
    client = AsyncOpenAI()

    prompt = f"""對比以下兩個AI回答，判斷哪個更好。

問題: {question}
回答A: {answer_a}
回答B: {answer_b}

請以JSON格式輸出:
{{
    "winner": "A"或"B"或"tie",
    "reasoning": "判斷理由",
    "a_score": 0-1,
    "b_score": 0-1
}}"""

    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.0,
        response_format={"type": "json_object"},
    )
    return json.loads(response.choices[0].message.content)

持續監控儀表板：定期執行評估並追蹤品質趨勢

class QualityMonitor:
    """AI品質持續監控"""

    def __init__(self, pipeline: EvalPipeline):
        self.pipeline = pipeline
        self.history: list[dict] = []

    async def run_scheduled_eval(self, config: EvalConfig) -> dict:
        result = await self.pipeline.run(config)

        record = {
            "timestamp": result.timestamp,
            "pass_rate": result.pass_rate,
            "avg_score": result.avg_weighted_score,
            "run_id": result.run_id,
        }
        self.history.append(record)

        if len(self.history) >= 3:
            recent_rates = [h["pass_rate"] for h in self.history[-3:]]
            if all(r1 > r2 for r1, r2 in zip(recent_rates, recent_rates[1:])):
                record["trend"] = "declining"
                record["alert"] = "品質持續下降，請檢查最近的模型或prompt變更"

        return record

方案對比

特性	人工評估	規則評估	LLM-as-Judge	RAGAS
可擴展性	❌ 低	✅ 高	✅ 高	✅ 高
評估深度	✅ 深	⚠️ 淺	✅ 深	✅ 深
成本	❌ 高	✅ 低	⚠️ 中	⚠️ 中
一致性	⚠️ 低	✅ 高	⚠️ 中	✅ 高
幻覺檢測	✅ 好	❌ 差	✅ 好	✅ 好
RAG專項	❌ 無	❌ 無	⚠️ 部分	✅ 完整
CI/CD整合	❌ 難	✅ 易	✅ 易	✅ 易

總結

LLM-as-Judge模式為AI品質評估提供了可擴展、可量化的解決方案。透過基礎Judge框架、RAGAS指標體系、自訂評估器、多維度聚合報告和生產級CI/CD管道這5個核心模式，開發者可以構建完整的AI品質保障體系。記住：AI評估不是一次性的，而是持續的過程——只有將評估嵌入CI/CD管道，才能確保AI應用在生產環境中持續保持高品質。

線上工具推薦

JSON格式化 - 格式化和驗證評估結果的JSON資料
cURL轉程式碼 - 將評估API的cURL命令轉換為Python程式碼
雜湊計算 - 計算評估快取鍵的雜湊值