การสร้างชุดเครื่องมือประเมินอัตโนมัติ

โครงทดสอบการประเมินคืออะไร

สถาปัตยกรรมของโครงทดสอบ

class RAGEvaluationHarness:
    def __init__(self, retriever, llm_client, config):
        self.retriever = retriever
        self.llm_client = llm_client
        self.config = config  # chunk_size, top_k, model, threshold, etc.
        self.results = []

    def run(self, golden_dataset):
        for item in golden_dataset:
            result = self._evaluate_single(item)
            self.results.append(result)
        metrics = self._compute_metrics()
        self._save_report(metrics)
        return metrics

เรียกใช้กรณีทดสอบแต่ละรายการ

import time

def _evaluate_single(self, item):
    start = time.perf_counter()
    query_vector = embed_query(item['question'])
    chunks = self.retriever.retrieve(query_vector, top_k=self.config['top_k'])
    filtered_chunks = filter_by_score(chunks, self.config['threshold'])
    context = format_context(filtered_chunks)
    answer_result = generate_answer(item['question'], context, self.llm_client)
    latency_ms = (time.perf_counter() - start) * 1000

    return {
        'question': item['question'],
        'expected_answer': item['answer'],
        'generated_answer': answer_result['answer'],
        'retrieved_chunk_ids': [c['id'] for c in filtered_chunks],
        'retrieved_scores': [c['score'] for c in filtered_chunks],
        'relevant_chunk_ids': item['relevant_chunk_ids'],
        'context_texts': [c['text'] for c in filtered_chunks],
        'tokens_used': answer_result['tokens_used'],
        'latency_ms': round(latency_ms)
    }

คำนวณตัวชี้วัดทั้งหมดในการประมวลผลครั้งเดียว

def _compute_metrics(self):
    # Retrieval metrics (no LLM calls needed)
    hit_rates = []
    mrr_scores = []
    for r in self.results:
        retrieved = r['retrieved_chunk_ids']
        relevant = set(r['relevant_chunk_ids'])
        hit = any(rid in relevant for rid in retrieved)
        hit_rates.append(1.0 if hit else 0.0)
        for rank, rid in enumerate(retrieved, 1):
            if rid in relevant:
                mrr_scores.append(1.0 / rank)
                break
        else:
            mrr_scores.append(0.0)

    metrics = {
        'hit_rate_at_5': sum(hit_rates) / len(hit_rates),
        'mrr': sum(mrr_scores) / len(mrr_scores),
        'mean_latency_ms': sum(r['latency_ms'] for r in self.results) / len(self.results),
        'mean_tokens': sum(r['tokens_used'] for r in self.results) / len(self.results)
    }
    return metrics

บันทึกผลลัพธ์พร้อมข้อมูลรุ่น

import json
import subprocess
from datetime import datetime

def _save_report(self, metrics):
    git_hash = subprocess.check_output(
        ['git', 'rev-parse', '--short', 'HEAD']
    ).decode().strip()

    report = {
        'run_id': datetime.utcnow().strftime('%Y%m%d_%H%M%S'),
        'git_commit': git_hash,
        'config': self.config,
        'metrics': metrics,
        'n_test_cases': len(self.results),
        'timestamp': datetime.utcnow().isoformat()
    }

    with open('eval_history.jsonl', 'a') as f:
        f.write(json.dumps(report) + '\n')
    print(f'Saved evaluation run: {report["run_id"]}')
    print(json.dumps(metrics, indent=2))

เปรียบเทียบกับค่าพื้นฐาน

def compare_to_baseline(current_metrics, baseline_file='best_eval.json'):
    import json
    from pathlib import Path
    if not Path(baseline_file).exists():
        print('No baseline yet. Saving current as baseline.')
        Path(baseline_file).write_text(json.dumps(current_metrics, indent=2))
        return True

    baseline = json.loads(Path(baseline_file).read_text())
    regressions = []
    print('\nMetric comparison (current vs baseline):')
    for metric, current_val in current_metrics.items():
        baseline_val = baseline.get(metric, 0)
        delta = current_val - baseline_val
        status = 'OK' if delta >= -0.02 else 'REGRESSION'
        print(f'  {metric}: {current_val:.3f} vs {baseline_val:.3f} ({delta:+.3f}) {status}')
        if status == 'REGRESSION':
            regressions.append(metric)
    return len(regressions) == 0

ผสานรวมเข้ากับ CI/CD

# GitHub Actions workflow (eval.yml)
# on:
#   pull_request:
#     paths:
#       - 'rag/**'
#       - 'prompts/**'
#       - 'config/**'
# jobs:
#   evaluate:
#     runs-on: ubuntu-latest
#     steps:
#       - uses: actions/checkout@v3
#       - name: Install dependencies
#         run: pip install -r requirements.txt
#       - name: Run evaluation harness
#         run: |
#           python eval/run_harness.py \
#             --test-set eval/golden_dataset.json \
#             --config config/rag_config.yaml \
#             --fail-on-regression
#         env:
#           OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}

สร้างรายงานที่มนุษย์อ่านเข้าใจ

def generate_markdown_report(metrics, failed_cases, run_id):
    lines = [
        f'# RAG Evaluation Report — {run_id}\n',
        '## Summary Metrics',
        '| Metric | Score | Target |',
        '|--------|-------|--------|',
        f'| Hit Rate@5 | {metrics["hit_rate_at_5"]:.1%} | > 80% |',
        f'| MRR | {metrics["mrr"]:.3f} | > 0.70 |',
        f'| Mean Latency | {metrics["mean_latency_ms"]:.0f}ms | < 500ms |',
        '',
        f'## Failed Cases ({len(failed_cases)} failures)'
    ]
    for case in failed_cases[:10]:  # show first 10
        lines += [
            f'**Q:** {case["question"]}',
            f'**Expected:** {case["expected_answer"]}',
            f'**Generated:** {case["generated_answer"]}\n'
        ]
    return '\n'.join(lines)

ติดตามค่าใช้จ่ายต่อการประเมินหนึ่งครั้ง

def estimate_run_cost(results, config):
    # Embedding cost
    embed_tokens = sum(len(r['question'].split()) * 1.3 for r in results)
    embed_cost = (embed_tokens / 1_000_000) * 0.02  # $0.02/1M tokens

    # Generation cost
    total_gen_tokens = sum(r['tokens_used'] for r in results)
    gen_cost = (total_gen_tokens / 1_000_000) * 5.0  # gpt-4o approx

    # Judge cost (faithfulness evals)
    judge_cost = len(results) * 0.001  # ~$0.001 per eval with gpt-4o-mini

    total = embed_cost + gen_cost + judge_cost
    print(f'Evaluation cost estimate: ${total:.2f}')
    print(f'  Embedding: ${embed_cost:.3f}')
    print(f'  Generation: ${gen_cost:.3f}')
    print(f'  Judgment: ${judge_cost:.3f}')
    return total

กำหนดเวลาการประเมินเพื่อเฝ้าติดตามระบบใช้งานจริง

# Example scheduled evaluation (cron job or scheduled cloud function)
import random

def sample_production_queries(query_log_file, n=50):
    with open(query_log_file) as f:
        all_queries = [json.loads(line) for line in f]
    sample = random.sample(all_queries, min(n, len(all_queries)))
    # Convert to golden dataset format (without expected answers — use LLM judge)
    return [
        {'question': q['user_question'], 'relevant_chunk_ids': []}
        for q in sample
    ]

# Run weekly evaluation against production queries
if __name__ == '__main__':
    prod_queries = sample_production_queries('/var/log/rag_queries.jsonl')
    harness = RAGEvaluationHarness(retriever, llm_client, config)
    metrics = harness.run(prod_queries)
    send_slack_digest(metrics)

แสดงแนวโน้มของตัวชี้วัดตามเวลา

import json
import matplotlib.pyplot as plt
from pathlib import Path

def plot_metric_trends(history_file='eval_history.jsonl', metric='hit_rate_at_5'):
    records = [
        json.loads(line)
        for line in Path(history_file).read_text().strip().split('\n')
    ]
    timestamps = [r['timestamp'][:10] for r in records[-20:]]
    scores = [r['metrics'].get(metric, 0) for r in records[-20:]]
    plt.figure(figsize=(10, 4))
    plt.plot(timestamps, scores, marker='o', label=metric)
    plt.axhline(y=0.80, color='r', linestyle='--', label='Min threshold')
    plt.title(f'{metric} over last 20 evaluations')
    plt.xticks(rotation=45)
    plt.tight_layout()
    plt.savefig(f'eval_trend_{metric}.png')
    print(f'Saved trend chart for {metric}')

ตรวจสอบความเข้าใจอย่างรวดเร็ว

สรุปบทเรียน