بناء خط أنابيب للتقييم المستمر

لماذا يجب أن يكون التقييم مستمرًا

المكونات الأساسية لخط الأنابيب

# Pipeline architecture:
#
# test_dataset.json
#       |
#       v
# system_runner.py  --> calls your LLM pipeline
#       |
#       v
# judge.py          --> scores each (question, answer) pair
#       |
#       v
# results_db        --> stores timestamped metric history
#       |
#       v
# dashboard + alert --> Grafana / Slack notification

هيكلة مجموعة بيانات الاختبار

# eval/test_set_v3.json
# {
#   'version': '3.0',
#   'created': '2026-06-01',
#   'cases': [
#     {
#       'id': 'faq_001',
#       'category': 'factual',
#       'question': 'What is the cancellation policy?',
#       'reference': 'Cancellations must be made 24 hours in advance.',
#       'min_correctness': 4
#     },
#     ...
#   ]
# }

تشغيل مجموعة التقييم

import asyncio
import uuid
from datetime import datetime

async def run_eval_suite(system, test_set: list, commit_sha: str) -> dict:
    run_id = str(uuid.uuid4())
    results = []
    for case in test_set:
        response = await system.answer(case['question'])
        score = await judge(case['question'], response, case.get('reference'))
        results.append({
            'run_id': run_id,
            'commit_sha': commit_sha,
            'case_id': case['id'],
            'category': case['category'],
            'response': response,
            'score': score.model_dump(),
            'evaluated_at': datetime.utcnow().isoformat()
        })
    return {'run_id': run_id, 'results': results}

تخزين المقاييس التاريخية والاستعلام عنها

-- PostgreSQL schema
CREATE TABLE eval_runs (
    run_id UUID PRIMARY KEY,
    commit_sha TEXT NOT NULL,
    test_set_version TEXT NOT NULL,
    triggered_by TEXT,  -- 'ci', 'scheduled', 'manual'
    started_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE TABLE eval_results (
    id SERIAL PRIMARY KEY,
    run_id UUID REFERENCES eval_runs(run_id),
    case_id TEXT NOT NULL,
    category TEXT,
    correctness INT,
    overall INT,
    response TEXT
);

CREATE INDEX idx_run_id ON eval_results(run_id);

اكتشاف حالات التراجع تلقائيًا

def detect_regression(current: dict, baseline: dict, threshold_pct: float = 5.0) -> dict:
    regressions = []
    for metric in ['correctness', 'helpfulness', 'clarity']:
        delta_pct = (current[metric] - baseline[metric]) / baseline[metric] * 100
        if delta_pct < -threshold_pct:
            regressions.append({
                'metric': metric,
                'baseline': baseline[metric],
                'current': current[metric],
                'delta_pct': round(delta_pct, 1)
            })
    return {'has_regression': len(regressions) > 0, 'regressions': regressions}

التكامل مع CI/CD

# .github/workflows/eval.yml
# name: LLM Quality Evaluation
# on: [pull_request]
# jobs:
#   eval:
#     runs-on: ubuntu-latest
#     steps:
#       - uses: actions/checkout@v4
#       - name: Run eval suite
#         run: python eval/run_suite.py --commit $GITHUB_SHA --mode pr
#         env:
#           OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
#       - name: Check for regressions
#         run: python eval/check_regression.py --run-id $EVAL_RUN_ID

عمليات تشغيل التقييم الكامل المجدولة

# Separate eval modes:
EVAL_CONFIGS = {
    'pr_check': {
        'test_cases': 'eval/test_set_core_100.json',
        'target': 'staging',
        'max_runtime_min': 8
    },
    'nightly': {
        'test_cases': 'eval/test_set_full_350.json',
        'target': 'production',
        'max_runtime_min': 30
    },
    'weekly_deep': {
        'test_cases': 'eval/test_set_full_350.json',
        'target': 'production',
        'include_pairwise': True,
        'max_runtime_min': 90
    }
}

التنبيه عند تدهور الجودة

import httpx

def send_regression_alert(regression_report: dict, webhook_url: str):
    regressions = regression_report['regressions']
    blocks = [{
        'type': 'section',
        'text': {'type': 'mrkdwn', 'text': '*LLM Quality Regression Detected*'}
    }]
    for r in regressions:
        blocks.append({
            'type': 'section',
            'text': {'type': 'mrkdwn',
                     'text': f'*{r["metric"]}*: {r["baseline"]} -> {r["current"]} ({r["delta_pct"]}%)'}
        })
    httpx.post(webhook_url, json={'blocks': blocks})

إدارة توسيع مجموعة الاختبار

def add_to_test_set(question: str, reference_answer: str, category: str,
                    source: str, test_set_path: str):
    import json, uuid
    with open(test_set_path, 'r') as f:
        test_set = json.load(f)
    test_set['cases'].append({
        'id': f'user_report_{uuid.uuid4().hex[:8]}',
        'category': category,
        'question': question,
        'reference': reference_answer,
        'source': source,  # 'user_report', 'regression', 'manual'
        'added': '2026-06-21'
    })
    with open(test_set_path, 'w') as f:
        json.dump(test_set, f, indent=2)

تصوير اتجاهات الجودة بمرور الوقت

import matplotlib.pyplot as plt
import pandas as pd

def plot_quality_trend(eval_history: list):
    df = pd.DataFrame(eval_history)
    df['date'] = pd.to_datetime(df['evaluated_at'])
    df = df.sort_values('date')
    # 7-day rolling average
    df['score_ma7'] = df['mean_correctness'].rolling(window=7).mean()
    plt.figure(figsize=(12, 4))
    plt.plot(df['date'], df['mean_correctness'], alpha=0.3, label='Daily')
    plt.plot(df['date'], df['score_ma7'], label='7-day avg', linewidth=2)
    plt.axhline(y=4.0, color='r', linestyle='--', label='Min threshold')
    plt.legend()
    plt.title('LLM Answer Quality Over Time')
    plt.savefig('quality_trend.png')

تحقق سريع

مراجعة الدرس