人工智能评估工程：从零开始构建一款可用于生产环境的大型语言模型评估平台【完整使用手册】

第6部分：当系统具备相应工具和资源时的智能体评估

6.1 为什么智能体评估具有根本性的不同之处

一个简单的RAG处理流程只包含一个交互环节：接收查询请求并生成答案。人们评估的只是最终生成的答案而已。然而，一个真正的智能体系统会经历一系列推理过程、调用各种工具，并产生多个中间结果，最终才形成最终的响应。如果只评估最终的响应结果，就会忽略其中可能出现的许多问题。

在实际应用中，对AI智能体的评估意味着要系统地检验该智能体是否能够正确、安全且高效地完成实际任务，而不仅仅是要验证其底层的大语言模型能否生成合理的文本。这两者之间的区别在于：知道一个智能体“听起来很聪明”与知道它“确实能够发挥作用”是截然不同的。

一个智能体可能会通过错误的推理路径得出正确的最终答案。虽然答案本身是正确的，但推理过程存在错误，而稍微改变输入数据就会暴露这一缺陷。此外，一个智能体也可能使用了正确的推理路径，但在使用某个特定工具时出现了问题；或者它虽然完成了任务，但却需要调用14个工具才能完成工作，而实际上只需要3个工具就足够了。所有这些情况都属于评估范围之内，因为在仅关注最终答案的评估中，这些缺陷都是无法被发现的。

对智能体的评估必须涵盖其整个推理过程，而不仅仅是最终的输出结果。

以下代码实现了三种针对特定智能体的评估指标，这些指标分别用于检测智能体在处理问题过程中出现的不同类型错误。

# evals/agent_metrics.py
# 用于评估使用工具和多步骤推理机制的智能体系统的指标

import json
from dataclasses import dataclass
from typing import Any

from openai import AsyncOpenAI

client = AsyncOpenAI()

@dataclass
class AgentTrace:
    """完整的智能体执行轨迹信息。"""
    query: str
    steps: list[dict]    # 每一步包含：{类型："reasoning|tool_call|tool_result", 内容: ...}
    final_answer: str
    total_tokens: int
    total_latency_ms: float


class TaskCompletionMetric:
    """
    评估指标：智能体是否真正完成了任务？

    这是衡量智能体成功与否的主要指标。该指标会将任务分解为多个子目标，并检查这些子目标是否都被完成。

    目标阈值：0.85。
    """

    name = "task_completion"
    threshold = 0.85

    async def score(
        self, case: Any, trace: AgentTrace
    ) -> tuple[float, str, float]:
        prompt = f"""
您正在评估一个AI智能体是否成功完成了任务。

原始任务：{trace.query}

智能体的最终答案：{trace.final_answer}

智能体的操作步骤摘要：
{self._summarize_steps(trace.steps)}

请将原始任务分解为所需的子目标，并检查每个子目标是否都被完成。

返回的JSON格式如下：
{{
  "sub_goals": [
    {{
      "goal": "<子目标描述>",
      "completed": true/false,
      "evidence": "<说明如何判断该子目标是否完成>"
    }}
  ],
  "overall_assessment": "<总体评估结果>"
}}
        """.strip()

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

        try:
            data = json.loads(r.choices[0].message.content)
            sub_goals = data.get("sub_goals", [])
        except (json.JSONDecodeError, KeyError):
            return 0.0, "无法解析任务完成情况评估结果", 0.003

        completed = sum(1 for g in subgoals if g.get("completed"))
        total = len(sub_goals)
        score = completed / total if total > 0 else 0.0

        missing = [g["goal"] for g in subGoals if not g.get("completed")]
        reason = (
            f"任务完成情况：{score:.2f}（共{completed}/{total}个子目标完成）"
            + (f"\n未完成的子目标有：{'; '.join(missing)}" if missing else "")
        )

        cost = r_usage.total_tokens * 0.000005
        return round(score, 4), reason, round(cost, 6)

    def _summarize_steps(self, steps: list[dict]) -> str:
        lines = []
        for i, step in enumerate(steps[:20]):  # 为了保证提示信息的长度，步骤数量限制为20个
            step_type = step.get("type", "unknown")
            content = str(step.get("content", ""))[:200]
            lines.append(f"步骤 {i+1}：{step_type}：{content}")
        return "\n".join(lines)


class ToolUsageEfficiencyMetric:
    """
    评估指标：智能体是否高效且正确地使用了工具？

    该指标能够检测出以下问题：错误使用工具（为某项任务选择了错误的工具）、重复获取信息（多次调用同一工具来获取已经获得的数据），以及工具调用顺序错误。

    目标阈值：0.75。
    """

    name = "tool_usage_efficiency"
    threshold = 0.75

    async def score(
        self, case: Any, trace: AgentTrace
    ) -> tuple[float, str, float]:
        tool_calls = [
            s for s in trace.steps if s.get("type") == "tool_call"
        ]
        tool_results = [
            s for s in trace_steps if s.get("type") == "tool_result"
        ]

        if not tool_calls:
            # 如果没有使用任何工具，那么评分依据就是是否需要使用这些工具
            return 1.0, "此轨迹中未使用任何工具", 0.0

        prompt = f"""
您正在评估一个AI智能体使用工具的效率。

任务：{trace.query}

使用的工具调用记录：
{json.dumps([tc.get("content", "") for tc in tool_calls], indent=2)}

收到的工具结果：
{json.dumps([tr.get("content", "")[:300] for tr in tool_results], indent=2)[:3000]}

请从以下方面评估工具使用效率：
1. 必要性：所有工具调用是否都是完成任务所必需的？
2. 无冗余性：是否存在重复调用相同信息的情况？
3. 工具选择正确性：每个子任务是否都使用了正确的工具？
4. 调用顺序合理性：工具调用的顺序是否合理？

返回的JSON格式如下：
{{
  "total_calls": {len/tool_calls)},
  "unnecessary_calls": ["<描述不必要的调用>"],
  "redundant_calls": ["<描述重复的调用>"],
  "wrong_tool_calls": ["<描述使用了错误的工具>"],
  "ordering_issues": ["<描述调用顺序不合理的地方>"],
  "efficiency_score": <0-10>
}}
        """.strip()

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

        try:
            data = json.loads(r.choices[0].message.content)
            score = min(max(data.get("efficiency_score", 0) / 10.0, 0.0), 1.0)
        except (json.JSONDecodeError, KeyError, TypeError):
            return 0.5, "无法解析工具使用效率评估结果", 0.001

        issues = (
            data.get("unnecessary_calls", [])
            + data.get("redundant_calls", []
            + data.get("wrong_tool_calls", []
        )
        reason = (
            f"工具使用效率：{score:.2f}（共{len/toolcalls)}次调用，"
            f"{len(issues)}个问题")
            + (f"\n存在的问题有：{'; '.join(issues[:3])}" if issues else "")
        )

        cost = r_usage.total_tokens * 0.00000015
        return round(score, 4), reason, round(cost, 6)


class ReasoningCoherenceMetric:
    """
    评估指标：智能体的推理过程是否逻辑连贯？

    该指标能够发现那些通过错误推理得出正确答案的情况——这种推理方式很脆弱，遇到边界情况就会失效。

    目标阈值：0.80。
    """

    name = "reasoning_coherence"
    threshold = 0.80

    async def score(
        self, case: Any, trace: AgentTrace
    ) -> tuple[float, str, float]:
        reasoning_steps = [
            s.get("content", "")
            for s in trace.steps
            if s.get("type") == "reasoning"
        ]

        if not reasoning_steps:
            return 0.5, "轨迹中未记录任何推理步骤", 0.0

        reasoning_text = "\n\n".join(
            f"步骤 {i+1}：{step}"
            for i, step in enumerate(reasoning_steps)
        )

        prompt = f"""
请评估这个AI智能体的推理过程是否逻辑连贯。

任务：{trace.query}
最终答案：{trace.final_answer}

推理过程：
{reasoning_text[:3000]}

需要检查的内容包括：
- 推理过程中是否存在逻辑漏洞或跳跃
- 结论是否是根据前提正确得出的
- 各个推理步骤之间是否存在矛盾
- 是否通过错误的推理途径得出了正确的答案
- 是否存在不必要的或循环的推理

返回的JSON格式如下：
{{
  "coherence_score": <0-10>,
  "logical_gaps": ["<描述逻辑漏洞>"],
  "contradictions": ["<描述矛盾之处>"],
  "correct_answer_wrong_reasoning": true/false,
  "overall_assessment": "<总体评估结果>"
}}
        """.strip()

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

        try:
            data = json.loads(r.choices[0].message.content)
            score = min(max(data.get("coherence_score", 0) / 10.0, 0.0), 1.0)
        except (json.JSONDecodeError, KeyError, TypeError):
            return 0.5, "无法解析推理连贯性评估结果", 0.003

        issues = data.get("logical_gaps", []) + data.get("contradictions", [])
        if data.get("correct_answer_wrong_reasoning"):
            issues.append("通过错误的推理途径得出了正确答案")

        reason = (
            data.get("overall_assessment", "")
            + (f"\n存在的问题有：{'; '.join(issues[:3])}" if issues else "")
        )

        cost = r_usage.total_tokens * 0.000005
        return round(score, 4), reason, round(cost, 6)

`AgentTrace`数据类就是所需的输入格式。它能够记录单个智能体运行过程中的全部执行细节：原始查询语句、所有按类型标记的中间步骤（包括推理过程、工具调用或工具结果）、最终答案，以及总的令牌使用量和延迟成本。你的智能体框架必须能够生成这种格式的追踪数据。配套的代码库中包含了针对LangChain、LlamaIndex以及OpenAI提供的原始函数调用智能体的适配器。

TaskCompletionMetric是衡量任务完成情况的主要指标。该指标会利用评估提示将原始任务分解为多个子目标，然后逐一核对这些子目标与智能体的最终答案是否一致。

ToolUsageEfficiencyMetric用于评估智能体使用工具的效率。该指标会检查四种具体问题：不必要的工具调用（在答案已经可得的情况下仍进行工具调用）、重复性的操作（同一信息被多次获取）、错误的工具选择（本应使用数据库查询却选择了网络搜索工具），以及调用顺序错误（导致后续的操作变得多余）。

ReasoningCoherenceMetric是这三个评估指标中最具诊断意义的，它专门用于识别那些通过错误推理路径得出正确答案的智能体。该指标会检查每一步推理过程是否逻辑连贯、智能体在不同步骤之间是否存在自相矛盾的地方，以及最终答案是否确实是整个推理过程的必然结果，还是一个偶然正确的结论。

第7部分：CI/CD集成——用于阻止错误部署的评估机制

7.1 评估机制的原理

CI/CD集成中的评估机制会在每次收到拉取请求时自动运行评估流程，如果有任何指标低于预设阈值，就会阻止合并操作的发生。这项措施是你在评估体系中所能做出的最具成效的投资。

7.2 GitHub Actions集成

下面的GitHub Actions工作流程将第7.1节中介绍的评估机制应用到了你的拉取请求处理流程中。在阅读YAML配置文件之前，了解其中的关键设计决策是非常有必要的，因为这些决策会直接影响该评估机制的实际运行效果。

首先，《on: pull_request`配置项下的`paths`过滤器至关重要。只有当`app/`、`prompts/`或`config/`目录中的文件发生变化时，这个工作流程才会被触发。这意味着，仅用于提交文档的拉取请求不会触发评估过程；但需要注意的是，只要对提示文件进行了任何修改，系统就会执行完整的评估操作。

这种设计是非常合理的：在大型语言模型应用中，提示文件的变化往往是导致质量下降的最常见原因，而且工程师们也常常会在没有进行系统性测试的情况下就提交这类更改。

`concurrency`配置块中的`cancel-in-progress: true`选项意味着，如果开发人员连续提交了两次修改请求，第一次评估操作会被取消，只有第二次提交才会被执行。这样就可以避免在开发过程中队列任务堆积，同时也能确保能够获取到分支的最新状态。

每次评估开始时，系统会下载基准分数数据；如果评估通过，这些数据会在评估结束时被上传。这种机制使得跨多个拉取请求进行质量回归检测成为可能：当新的拉取请求被提交时，系统会使用上次在主分支上成功完成的评估结果作为基准，然后将当前请求的评估结果与这个基准进行比较。如果在第一次执行评估时还没有基准数据，`continue-on-error: true`选项可以确保工作流程至少运行一次后再判断是否失败。

最后，系统会在拉取请求中直接生成一条格式化的评论，其中会包含各项评估指标的分数、评估结果以及是否允许合并的信息。这样一来，开发人员就无需查看复杂的日志信息，就能立即了解评估的结果。

# .github/workflows/eval-gate.yml
# 适用于所有涉及AI系统的拉取请求

name: AI评估机制

on:
  pull_request:
    paths:
      - 'app/**'           # 应用代码目录
      - 'prompts/**'       # 提示文件目录——任何修改都会触发评估
      - 'config/**'        # 配置文件目录，包括模型选择相关设置

concurrency:
  group: eval-gate-${{ github.ref }}
  cancel-in-progress: true

jobs:
  eval-gate:
    runs-on: ubuntu-latest
    timeout-minutes: 30

    steps:
      - uses: actions/checkout@v4

      - name: 安装Python环境
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'
          cache: pip

      - name: 安装依赖库
        run: pip install -r requirements.txt

      - name: 下载基准分数数据
        uses: actions/download-artifact@v4
        with:
          name: eval-baseline-scores
          path: eval-results/
        continue-on-error: true   # 第一次执行时没有基准数据，这是正常的

      - name: 执行评估操作
        env:
          OPENAI_API_KEY:  ${{ secrets.OPENAI_API_KEY }}
          EVAL_suite:      rag-production
          EVAL_DATASET:    datasets/golden.jsonl
        run: python -m cicd.eval_gate

      - name: 上传基准分数数据
        if: success()
        uses: actions/upload-artifact@v4
        with:
          name: eval-baseline-scores
          path: eval-results/baseline_scores.json

      - name: 上传完整评估结果
        uses: actions/upload-artifact@v4
        with:
          name: eval-results-${{ github.sha }}
          path: eval-results/

      - name: 在拉取请求中添加评论
        if: always()
        uses: actions/github-script@v7
        with:
          script: |
            const fs = require('fs');
            const results = fs.readdirSync('eval-results/')
              .filter(f => f.endsWith('.json') && !f.includes('baseline'))
              .map(f => JSON.parse(fs.readFileSync(`eval-results/${f}`)))
              .sort((a, b) => b.timestamp.localeCompare(a.timestamp))[0];

            if (!results) return;

            const emoji   = results.passed ? '✅' : '❌';
            const status  = results-passed ? 'GATE PASSED' : 'GATE FAILED — merge blocked';
            const scores  = Object.entries(results(metric_scores)
              .map(([k, v]) => `| ${k} | ${v.toFixed(3)} |`)
              .join('\n');

            const body = `## ${emoji} 评估结果: ${status}

**测试套件:** ${results.suite_name}
**通过案例数:** ${results.passed_cases}/${results.total_cases}
**总成本:** $${results.total_cost_usd.toFixed(4)}

| 指标        | 分数       |
|-------------|------------|
${scores}

${!results.passed ? '⚠️ **此拉取请求无法合并。请先修复未通过的指标，然后再申请审核。**' : ''}`;

            github.rest.issues.createComment({
              owner: context.repo.owner,
              repo:  context.repo.repo,
              issue_number: context.issue.number,
              body,
            });

第8部分：生产监控——永不停歇的评估循环

8.1 为什么生产监控与离线评估不同

你的黄金数据集仅涵盖了你所了解的故障模式。但在实际生产环境中，用户会生成一些你从未预料到的输入数据。如果不对这些输入进行实时监控，那么当现实世界中的输入数据开始偏离黄金数据集所涵盖的范围时，这种变化是无法被发现的。

实时监控功能使平台能够实时监测生产环境中的检索延迟、生成质量以及错误出现频率。通过根本原因分析工具，可以及时发现检索、上下文处理和生成环节中存在的问题，从而迅速采取应对措施。

生产监控具备离线评估所无法实现的三项功能：

1. 能够检测数据分布的变化：当用户输入的数据开始出现新的特征（比如出现新的主题、表达方式或故障模式）时，生产监控系统能够及时发现这些变化，从而避免因此产生大量的支持请求。

2. 能收集新的评估案例：每一次生产中的故障其实都是一个可以用于训练数据集的优质样本。监控系统会自动识别出质量较低的异常记录，并将它们列入待人工审核的列表中。

3. 可验证模型更新的效果：当你对基础模型进行更新时，虽然黄金数据集的评分可能不会发生变化，但在那些黄金数据集未涵盖的新输入数据上，模型生成的质量很可能会下降。生产监控系统能在几小时内就发现这种变化，而不会等到数周之后才发现问题。

# monitors/production_monitor.py
# 实时监控生产环境中的质量状况，并自动触发警报

import asyncio
import json
import random
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any

import boto3
import structlog
from prometheus_client import Counter, Gauge, Histogram, start_http_server

from evals.rag_metrics import FaithfulnessMetric, HallucinationMetric

log = structlog.get_logger()

# 用于Grafana展示的Prometheus指标
EVAL SCORE = Gauge(
    "ai_eval_score",
    "当前各项指标的评估分数",
    labelnames=["metric", "system", "environment"],
)
EVAL_LATENCY = Histogram(
    "ai_eval_latency_ms",
    "评估延迟时间（以毫秒为单位）",
    labelnames=["metric"],
    buckets=[100, 500, 1000, 3000, 5000, 10000],
)
QUALITY_ALERTS = Counter(
    "ai_quality_alerts_total",
    "触发的质量警报总数",
    labelnames=["metric", "severity"],
)
TRACESEvalUATED = Counter(
    "ai_traces_evaluated_total",
    "被评估的生产记录总数",
    labelnames["outcome"],
)


@dataclass
class MonitorConfig:
    system_name: str
    environment: str
    # 评估样本率（1.0表示对所有记录进行评估，0.1表示仅评估10%的记录）
    sample_rate: float = 0.10
    # 警报阈值——当指标值低于这些阈值时触发警报
    alert_thresholds: dict[str, float] = None
    # 用于发送Slack警报的Webhook地址
    slack_webhook: str | None = None
    # 用于存储被评估记录的S3存储桶
    trace_bucket: str | None = None

    def __post_init__(self):
        if self.alert_thresholds is None:
            self_alert_thresholds = {
                "faithfulness": 0.75,
                "hallucination": 0.85,
            }


class ProductionMonitor:
    """
    实时监控生产环境中的AI系统质量。

    工作原理：
    1. 通过track()方法接收生产产生的记录。
    2. 按照配置的样本率对记录进行抽样评估（通常为10%–50%，以平衡成本和效率）。
    3. 对抽样的记录计算相关指标值。
    4. 将结果发送到Prometheus服务器。
    5. 将质量较低的记录转发至专门的数据收集流程中，用于更新黄金数据集。
    6. 当指标的滚动平均值低于预设阈值时，通过Slack发送警报。

    """

    def __init__(self, config: MonitorConfig):
        self.config = config
        self.metrics = [FaithfulnessMetric(), HallucinationMetric()]
        self.s3 = boto3.client('s3') if config.trace_bucket else None
        self._rolling_scores = {
            "faithfulness": [],
            "hallucination": []
        }
        self._window_size = 100  # 用于计算警报阈值的滚动窗口大小

    async def track(self, trace: dict) -> None:
        """
        对单条生产记录进行评估。
        在每次调用大语言模型后，应在API响应处理函数中调用此方法。
        """
        # 仅对部分记录进行抽样评估，以控制成本
        if random.random() > self.config.sample_rate:
            TRACES_evalUATED.labels(outcome="sampled_out").inc()
            return

        TRACESEvalUATED.labels(outcome="evaluated").inc()

        # 将记录存储起来，以便后续审计或数据收集
        if self.s3 and self.config.trace_bucket:
            await self._store_trace(trace)

        # 对记录计算相关指标值
        case = type('Case', (), {
            'query': trace.get('query', ""),
            'expected_context': [],
            'ideal_answer': "",
        })()

        for metric in self.metrics:
            t0 = time.monotonic()
            try:
                score, reason, cost = await metric.score(case, trace)
                latency_ms = (time.monotonic() - t0) * 1000

                # 更新Prometheus指标值
                EVAL SCORE.labels(
                    metric=metric.name,
                    system=self.config.system_name,
                    environment=self.config.environment,
                ).set(score)

                EVAL_LATENCY.labels(metric=metric.name).observe(latency_ms)

                # 更新滚动窗口中的数据
                window = self._rolling_scores[metric.name]
                window.append(score)
                if len(window) > self._window_size:
                    window.pop(0)

                # 检查指标的滚动平均值是否低于阈值
                if len(window) >= 10:  # 需要至少10个样本才能计算平均值
                    rolling_avg = sum(window) / len(window)
                    threshold = self.config.alert_thresholds.get(metric.name)

                    if threshold and rolling_avg < threshold:
                        severity = (
                            "critical"
                            if rolling_avg < threshold * 0.85
                            else "warning"
                        )
                        QUALITY_ALERTS.labels(
                            metric=metric.name, severity=severity
                        ).inc()

                        await self._send_alert(
                            metric_name=metric.name,
                            rolling_avg=rolling_avg,
                            threshold=threshold,
                            severity=severity,
                            trace=trace,
                            reason=reason,
                        )

                # 将质量较低的记录转发至数据收集流程
                if score < metric.threshold * 0.9:
                    await self._route_to_harvest(
                        trace=trace,
                        metric_name=metric.name,
                        score=score,
                        reason=reason,
                    )

                log.debug(
                    "trace_evaluated",
                    metric=metric.name,
                    score=score,
                    system=self.config.system_name,
                )

            except Exception as e:
                log.error("metric_evaluation_failed", metric=metric.name, error=str(e))

    async def _store_trace(self, trace: dict) -> None:
        """将记录存储到S3存储桶中，以便后续审计或数据收集。"""
        trace_id = trace.get("trace_id", datetime.now(timezone.utc).isoformat())
        date_str = datetime.now(timezone.utc).strftime("%Y/%m/%d")
        key = f"traces/{date_str}/{trace_id}.json"

        self.s3.put_object(
            Bucket=self.config.trace_bucket,
            Key=key,
            Body=json.dumps({
                **trace,
                "stored_at": datetime.now(timezone.utc).isoformat(),
                "system": self.config.system_name,
                "environment": self.config.environment,
            }),
            ContentType="application/json",
        )

    async def _send_alert(self, metric_name: str, rolling_avg: float, threshold: float, severity: str, trace: dict, reason: str) -> None:
        """通过Slack发送质量下降警报。"""
        if not self.config.slack_webhook:
            return

        emoji = "🚨" if severity == "critical" else "⚠️"
        message = {
            "text": (
                f"{emoji} *质量警报 — {self.config.system_name}*\n"
                f"指标名称：`{metric_name}`\n"
                f"滚动平均值：`{rolling_avg:.3f}` "
                f"(阈值：`{threshold:.3f}`)\n"
                f"严重程度：`{severity}`\n"
                f"样本原因：_{reason[:300]}_\n"
                f"环境：`{self.config.environment}`"
            )
        }

        req = urllib.request.Request(
            self.config.slack_webhook,
            data=json.dumps(message).encode(),
            headers={"Content-Type": "application/json"},
        )
        urllib.request.urlopen(req)

    async def _route_to_harvest(self, trace: dict, metric_name: str, score: float, reason: str) -> None:
        """将质量较低的记录转发至数据收集流程中，以便进一步处理。"""
        if not self.s3 or not self.config.trace_bucket:
            return

        date_str = datetime.now(timezone.utc).strftime("%Y/%m/%d")
        trace_id = trace.get("trace_id", datetime.now(timezone.utc).isoformat())
        key = f"harvest-candidates/{date_str}/{metric_name}/{trace_id}.json"

        self.s3.put_object(
            Bucket=self.config.trace_bucket,
            Key=key,
            Body=json.dumps({
                **trace,
                "harvest_reason": f"`{metric_name}`指标的得分`{score:.3f}`低于阈值",
                "failing_metric": metric_name,
                "metric_score": score,
                "judge_reason": reason,
                "review_status": "pending",
                "harvested_at": datetime.now(timezone.utc).isoformat(),
            }),
            ContentType="application/json",
        )

        log.info(
            "trace_routed_to_harvest",
            metric=metric_name,
            score=score,
            trace_id=trace_id,
        )

第9部分：构建完整的评估平台

9.1 将所有组件整合成可运行的系统

这个完整的评估平台将之前所有的组件连接成一个端到端的系统：包括用于接收评估结果的REST API、用于查看结果的仪表板，以及用于在本地或持续集成环境中运行测试套件的命令行工具。

# app/eval_platform.py
# 完整的评估平台 — REST API + 仪表板 + 命令行工具

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import asyncio
import json
from pathlib import Path
from typing import Any, Optional

from evalsrunner import EvalRunner
from evals.rag_metrics import (
    FaithfulnessMetric, ContextRecallMetric,
    ContextPrecisionMetric, AnswerRelevancyMetric,
    HallucinationMetric, Groundedness Metric,
)
from evals.agent_metrics import (
    TaskCompletionMetric, ToolUsageEfficiencyMetric, ReasoningCoherenceMetric,
)
from evals.judge import RAG_QUALITY_JUDGE, SAFETY_JUDGE
from monitors.production_monitor import ProductionMonitor, MonitorConfig

app = FastAPI(
    title="AI评估平台",
    description="适用于大语言模型应用的生产级评估工具",
    version="1.0.0",
)


# —————————————————————————————————————————
# API模型
# —————————————————————————————————————————

class EvaluateRequest(BaseModel):
    query: str
    answer: str
    retrieved_contexts: list[str] = []
    ideal_answer: str = ""
    expected_context: list[str] = []
    metrics: list[str] = ["faithfulness", "hallucination", "answer_relevancy"]


class EvalResponse(BaseModel):
    passed: bool
    scores: dict[str, float]
    reasons: dict[str, str]
    cost_usd: float
    recommendations: list[str]


class RunSuiteRequest(BaseModel):
    suite_name: str
    dataset_path: str
    system_endpoint: str      # 需要评估的系统的URL
    metrics: list[str] = ["faithfulness", "context_recall", "hallucination"]


# —————————————————————————————————————————
# 度量指标注册表
# —————————————————————————————————————————

METRIC_REGISTRY = {
    "faithfulness":        FaithfulnessMetric(),
    "contextrecall":      ContextRecall Metric(),
    "context_precision":   ContextPrecisionMetric(),
    "answer_relevancy":    AnswerRelevancyMetric(),
    "hallucination":       HallucinationMetric(),
    "groundedness":        GroundednessMetric(),
    "taskcompletion":     TaskCompletionMetric(),
    "tool_efficiency":     ToolUsageEfficiencyMetric(),
    "reasoning_coherence": ReasoningCoherenceMetric(),
}


# —————————————————————————————————————————
# API端点
# —————————————————————————————————————————

@app.post("/evaluate", response_model=EvalResponse)
async def evaluate_single(request: EvaluateRequest):
    """根据指定的指标评估单个大语言模型的回答。"""

    selected_metrics = []
    for name in request.metrics:
        if name not in METRIC_REGISTRY:
            raise HTTPException(400, f"未知的指标：{name}")
        selected_metrics.append(METRIC_REGISTRY[name])

    # 根据请求信息创建一个评估案例
    case = type("Case", (), {
        "query":            request.query,
        "expected_context": request.expected_context,
        "ideal_answer":     request.ideal_answer,
    })()

    output = {
        "answer":             request.answer,
        "retrieved_contexts": request.retrieved_contexts,
    }

    scores  = {}
    reasons = {}
    total_cost = 0.0

    for metric in selected_metrics:
        score, reason, cost = await metric.score(case, output)
        scores[metric.name]  = score
        reasons[metric.name] = reason
        total_cost += cost

    passed = all(
        scores[m.name] >= m.threshold
        for m in selected_metrics
    )

    # 为未通过评估的指标生成建议
    recommendations = []
    for metric in selected_metrics:
        if scores[metric.name] < metric_threshold:
            recommendations.append(
                _get_recommendation(metric_name, scores[metric.name])
            )

    return EvalResponse(
        passed=passed,
        scores=scores,
        reasons=reasons,
        cost_usd=round(total_cost, 6),
        recommendations=recommendations,
    )


@app.get("/results")
async def list_results():
    """列出所有存储的评估结果。"""

    results_dir = Path("eval-results")
    if not results_dir.exists():
        return {"results": []}

    results = []
    for f in sorted(results_dir.glob("*.json")):
        try:
            data = json.loads(f.read_text())
            results.append({
                "file":       f.name,
                "suite_name": data.get("suite_name"),
                "timestamp":  data.get("timestamp"),
                "passed":     data.get("passed"),
                "pass_rate":  f"{data.get('passed_cases')}/{data.get('total_cases')}",
                "scores":     data.get("metric_scores"),
                "cost_usd":   data.get("total_cost_usd"),
            })
        except (json.JSONDecodeError, KeyError):
            continue

    return {"results": sorted(results, key=lambda x: x["timestamp"], reverse=True)}


@app.get("/metrics")
async def list_metrics():
    """列出所有可用的评估指标及其阈值。"""

    return {
        "metrics": {
            name: {
                "threshold": metric.threshold,
                "description": metric.__class__.__doc__[:200].strip()
                if metric.__class__.__doc__ else "",
            }
            for name, metric in METRIC_REGISTRY.items()
        }
    }


def _get_recommendation(metric_name: str, score: float) -> str:
    recommendations = {
        "faithfulness": (
            "忠实度低于阈值。请检查：模型是否添加了不在检索到的上下文中的信息？可以考虑在系统提示中加入‘只能使用提供的上下文’这样的指示。”
        ),
        "context_recall": (
            "上下文召回率低于阈值。请检查：检索器是否返回了所有相关的文档？可以增加检索的文档数量或改进分割策略。”
        ),
        "context_precision": (
            "上下文精确度低于阈值。检索器返回了不相关的文档。需要优化嵌入模型或改进检索评分机制。”
        ),
        "answer_relevancy": (
            "答案相关性低于阈值。模型回答的问题与实际提出的问题不同。请检查系统提示，它可能会误导模型。”
        ),
        "hallucination": (
            "幻觉现象出现的频率高于可接受的范围。需要在系统提示中加入‘不要进行无根据的推测’这样的指示。可以考虑更换一个遵循指令能力更强的模型。”
        ),
        "groundedness": (
            “现实性低于阈值。模型的推理超出了提供的上下文范围。需要在回答格式中添加引用上下文的要求。”
        ),
    }
    return recommendations.get(
        metric_name,
        f"{metric_name}的得分{score:.3f}低于阈值——请检查系统行为是否正常。”
    )

9.2 运行平台

当平台搭建完成后，根据具体使用场景，有三种方式可以与它进行交互：通过REST API将评估功能集成到其他服务中或执行一次性检查；通过命令行界面在本地或持续集成环境中运行完整的数据集测试套件；或者利用Prometheus指标服务器将数据传输到Grafana仪表板中进行实时监控。

第一个bash脚本块用于启动FastAPI服务器和Prometheus数据导出器。FastAPI服务器提供了三个接口：`POST /evaluate`用于单次评估（在开发过程中有助于调试特定输出结果），`GET /results`用于查看历史测试结果，`GET /metrics`用于查询可用的指标名称及阈值。

Prometheus服务器运行在9090端口上，会导出`ai_eval_score`、`ai_eval_latency_ms`和`ai_quality_alerts_total`这些在生产环境中被使用的指标数据。

你可以将Grafana连接到`localhost:9090`，并从配套仓库中导入预先构建好的仪表板，从而实时查看你的产品质量评估结果。

第二个脚本示例展示了如何通过API进行单次评估。当你想要快速判断某个大型语言模型的输出是否满足质量标准时，就可以使用这个命令；请求体中的`metrics`数组用于指定需要查询的指标。只有那些与当前问题相关的指标才需要被计算出来。

第三个脚本则通过命令行界面运行完整的测试套件。在持续集成环境中，使用`--regression-tolerance 0.05`这个参数可以允许指标值比基准值下降最多5%才会触发警报机制，这样既能有效避免误报，又能及时发现真正的性能变化。

# 启动评估平台
uvicorn app.eval_platform:app --host 0.0.0.0 --port 8080 --reload

# 运行Prometheus指标服务器（用于Grafana仪表板）
python -c "from prometheus_client import start_http_server; start_http_server(9090)"

# 通过API进行单次评估
curl -X POST http://localhost:8080/evaluate \
  -H "Content-Type: application/json" \
  -d '{
    "query": "GDPR第33条规定的数据泄露通知期限是多久？",
    "answer": "根据GDPR第33条规定，企业在发现个人数据泄露后必须在72小时内向监管机构报告。",
    "retrieved_contexts": [
      "Article 33 GDPR: 在发生个人数据泄露的情况下，企业应当立即、且尽可能在72小时以内通知监管机构……"
    ],
    "metrics": ["faithfulness", "answer_relevancy", "hallucination"]
  }'

# 运行完整的测试套件
python -m evals-runner \
  --suite-name legal-rag-production \
  --dataset datasets/legal-rag-golden.jsonl \
  --metrics faithfulness context_recall hallucination answer_relevancy

# 在持续集成环境中运行
python -m cicd.eval_gate \
  --suite rag-production \
  --dataset datasets/golden.jsonl \
  --regression-tolerance 0.05

位于github.com/aayostem/ai-evals-platform的配套代码库中包含了这个完整可运行的评估平台，其中包含以下内容：

• 所有带有测试覆盖率的评估指标

• 用于RAG系统及智能体系统的示例黄金数据集

• 适用于本地开发的Docker Compose配置文件

• 用于生产环境监控的预构建Grafana仪表板

• 样本校准数据及校准脚本

• GitHub Actions工作流模板

• 用于对比评估的示例RAG应用程序

结论

AI评估工程是一门独立的学科，而不仅仅是一项功能。它决定了你发布的AI系统是能够被有效验证其正确性的，还是只能寄希望于它们在大规模应用时能够正常运行。

本指南开篇提到的那个法律研究系统，在团队进行的各项测试中都通过了，但在实际生产环境中仍然会给出错误的答案。原因在于他们的评估体系中没有包含“上下文召回率”这一指标——而正是这个指标本可以及时发现检索过程中的错误。

这种缺陷导致了长达数周的事故调查，并严重损害了用户对这个原本设计良好的系统的信任。如果有一个功能完善的评估平台，就能在代码集成阶段就发现这些问题，从而避免它们进入生产环境。

从本指南所涵盖的所有内容中，我们可以得出以下关键结论：

评估是需要成本的，因此必须对其进行跟踪记录。 如果使用GPT-4o对每一个生产环节进行评估，大规模的LLM评估工作每月所需的成本可能高达数百美元。通过采用合适的架构设计——在生产环境中仅抽取10%的数据进行测试，对于大多数评估指标使用gpt-4o-mini模型，而专门用gpt-4o模型来检测虚假生成的内容——就可以将成本控制在任何工程团队都能承受的范围之内，同时依然能够保证评估的准确性。

根据本指南构建的完整评估平台包括：评估运行工具、黄金数据集结构、六种RAG评估指标、经过校准的LLM评估模型、用于评估代理行为的指标系统、持续集成/持续部署流程控制机制以及生产环境监控工具。你可以立即将这个平台应用于任何基于LLM的应用程序中。只需克隆github.com/aayostem/ai-evals-platform中的代码库，将评估运行工具配置到你的系统中，一小时内你就能获得第一份评估结果。

正是这些评估结果，构成了后续工作的基础。

最佳实践总结

✅ 建议：在制定评估指标之前，先构建你的黄金数据集。数据集决定了评估的范围；如果没有高质量的数据集，再好的评估指标也会导致错误的评估结果。

✅ 建议：将检索层与生成层分开进行评估。仅仅关注“准确性”是不够的，还需要考虑“上下文召回率”，以便能够发现那些看似生成成功但实际上存在问题的情况。

✅ 建议：在将LLM评估模型用于持续集成流程控制之前，先使用人类标注数据对其进行校准。未经校准的模型会错误地拒绝优秀的修改请求，却允许劣质的修改通过。

✅ 建议：以5%到10%的抽样比例进行生产环境监控。对每一个生产环节都进行评估既耗时又没有必要；抽取10%的数据并进行全面评估，其效果要比挑选出1%的样本进行测试要好得多。

✅ 建议：系统地收集生产过程中出现的问题，并将其纳入黄金数据集。最好的评估案例往往来源于实际发生的问题，而不是通过预测可能出现的故障模式得出的结果。

✅ 建议：详细记录每次评估所消耗的成本。如果每次测试的成本在0.001美元到0.003美元之间，那么每周对数千个案例进行评估也是完全可行的。了解自己的成本支出情况，并据此制定相应的预算。

❌ 不要：将BLEU或ROUGE作为评估LLM输出质量的主要指标。这些指标仅能反映文本表面上的相似性，与内容的准确性、逻辑性或相关性几乎没有关联。它们属于自然语言处理早期阶段使用的评估方法。

❌ 不要：仅仅依赖一个指标来评价系统的性能。如果一个模型在“准确性”方面得分很高，但在“上下文召回率”方面表现不佳，那么这个模型肯定存在问题。必须同时考虑这四项RAGA评估指标才能得出准确的结论。

❌ 不要：将评估工作视为产品发布前的一次性活动。模型的行为会随着提示语的变化、模型版本的更新、数据分布的变动以及系统配置的改变而发生变化，因此评估工作必须持续进行下去。

❌ 不要：使用同一个LLM模型既作为被测试的系统，又作为评估工具。这种做法会导致评估结果出现系统性偏差：评估模型会不加区分地偏爱自己生成的输出结果，而不管这些结果是否正确。因此应该使用更强或不同的模型来进行评估。

资源

• RAGAS文档：官方的RAG评估框架。本指南中提到的各项指标都是基于RAGAS概念框架进行实现的。

• DeepEval：一款开源评估框架，支持与Pytest集成，具备CI/CD功能，并内置了50多种评估指标。对于工程团队而言，这是最为实用的通用工具。

• MLflow评估指南：MLflow于2026年发布的关于如何将评估流程整合到AI开发工作流中的指导文档。

• FinOps Foundation – AI领域的财务运营管理框架：该框架用于同时管理评估基础设施的成本以及模型推理所产生的费用。

• OpenTelemetry与大型语言模型的追踪功能：一种用于收集生产环境监控所需数据的标准工具。

• 欧盟AI法案相关技术标准：针对高风险AI系统的评估工作，这些法规为相关实践提供了明确的规范。如今，进行评估已经不再仅仅是工程上的最佳实践，而成为一项必须遵守的要求。

• 配套代码库：本指南中提到的所有内容都已有完整的实现方案：包括各种评估指标、黄金数据集的管理机制、CI/CD流程、生产环境监控工具，以及Grafana数据面板等。