تقييم واختبار RAG
خطوط أنابيب الاختبار الآلية
3 دقيقة للقراءة
التقييم اليدوي لا يتوسع. هذا الدرس يغطي كيفية إعداد خطوط أنابيب اختبار آلية تعمل مع كل تغيير وتلتقط التراجعات قبل وصولها للإنتاج.
بنية خط الأنابيب
┌─────────────────────────────────────────────────────────────────┐
│ خط أنابيب اختبار RAG │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ المحفز │───▶│ التشغيل │───▶│ التقييم │───▶│ التقرير │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
│ │ │ │ │ │
│ ▼ ▼ ▼ ▼ │
│ • دفع Git • تحميل مجموعة • مقاييس RAGAS • لوحة معلومات│
│ • إنشاء PR الاختبار • العتبات • تنبيهات │
│ • جدولة • تشغيل RAG • مقارنة • المخرجات │
│ • يدوي • جمع المخرج • مع الأساس • Slack/Email │
│ │
└─────────────────────────────────────────────────────────────────┘
إعداد إطار الاختبار
import json
import time
from pathlib import Path
from dataclasses import dataclass
from typing import Callable
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision
@dataclass
class TestResult:
"""نتائج من تشغيل اختبار واحد."""
question: str
answer: str
contexts: list[str]
latency_ms: float
metrics: dict
@dataclass
class PipelineResult:
"""نتائج من تشغيل خط أنابيب كامل."""
timestamp: str
total_questions: int
avg_latency_ms: float
metrics: dict
failures: list[dict]
passed: bool
class RAGTestPipeline:
"""خط أنابيب اختبار RAG الآلي."""
def __init__(
self,
rag_function: Callable,
test_set_path: str,
thresholds: dict = None,
):
self.rag_function = rag_function
self.test_set = self._load_test_set(test_set_path)
self.thresholds = thresholds or {
"faithfulness": 0.8,
"answer_relevancy": 0.75,
"context_precision": 0.7,
"avg_latency_ms": 2000,
}
def _load_test_set(self, path: str) -> list[dict]:
return json.loads(Path(path).read_text())
def run(self) -> PipelineResult:
"""تنفيذ خط أنابيب الاختبار."""
results = []
latencies = []
# تشغيل RAG على كل سؤال اختبار
for test_case in self.test_set:
start = time.time()
output = self.rag_function(test_case["question"])
latency = (time.time() - start) * 1000
latencies.append(latency)
results.append(TestResult(
question=test_case["question"],
answer=output["answer"],
contexts=output["contexts"],
latency_ms=latency,
metrics={},
))
# التقييم بـ RAGAS
from datasets import Dataset
eval_dataset = Dataset.from_dict({
"question": [r.question for r in results],
"answer": [r.answer for r in results],
"contexts": [r.contexts for r in results],
})
scores = evaluate(
eval_dataset,
metrics=[faithfulness, answer_relevancy, context_precision],
)
# تحقق من العتبات
failures = []
passed = True
for metric, threshold in self.thresholds.items():
if metric == "avg_latency_ms":
actual = sum(latencies) / len(latencies)
else:
actual = scores.get(metric, 0)
if actual < threshold and metric != "avg_latency_ms":
failures.append({
"metric": metric,
"expected": f">= {threshold}",
"actual": actual,
})
passed = False
elif actual > threshold and metric == "avg_latency_ms":
failures.append({
"metric": metric,
"expected": f"<= {threshold}",
"actual": actual,
})
passed = False
return PipelineResult(
timestamp=datetime.now().isoformat(),
total_questions=len(results),
avg_latency_ms=sum(latencies) / len(latencies),
metrics=dict(scores),
failures=failures,
passed=passed,
)
تكامل GitHub Actions
# .github/workflows/rag-tests.yml
name: تقييم RAG
on:
push:
branches: [main]
paths:
- 'src/rag/**'
- 'prompts/**'
pull_request:
branches: [main]
schedule:
- cron: '0 0 * * *' # يومياً عند منتصف الليل
jobs:
evaluate:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: إعداد Python
uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: تثبيت التبعيات
run: |
pip install -r requirements.txt
pip install ragas pytest
- name: تشغيل تقييم RAG
env:
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
run: |
python -m pytest tests/test_rag_quality.py \
--json-report \
--json-report-file=results.json
- name: تحقق من العتبات
run: python scripts/check_thresholds.py results.json
- name: رفع النتائج
uses: actions/upload-artifact@v4
with:
name: rag-evaluation-results
path: results.json
- name: تعليق على PR
if: github.event_name == 'pull_request'
uses: actions/github-script@v7
with:
script: |
const fs = require('fs');
const results = JSON.parse(fs.readFileSync('results.json'));
const body = `## نتائج تقييم RAG
| المقياس | الدرجة | العتبة | الحالة |
|---------|-------|-------|--------|
| الأمانة | ${results.faithfulness.toFixed(2)} | 0.80 | ${results.faithfulness >= 0.8 ? '✅' : '❌'} |
| الصلة | ${results.answer_relevancy.toFixed(2)} | 0.75 | ${results.answer_relevancy >= 0.75 ? '✅' : '❌'} |
| الدقة | ${results.context_precision.toFixed(2)} | 0.70 | ${results.context_precision >= 0.7 ? '✅' : '❌'} |
`;
github.rest.issues.createComment({
issue_number: context.issue.number,
owner: context.repo.owner,
repo: context.repo.repo,
body: body
});
تكامل pytest
# tests/test_rag_quality.py
import pytest
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision
from datasets import Dataset
@pytest.fixture(scope="module")
def rag_results():
"""تشغيل خط أنابيب RAG على مجموعة الاختبار مرة لكل وحدة."""
from src.rag import RAGPipeline
pipeline = RAGPipeline()
test_questions = load_test_questions("tests/data/test_set.json")
results = []
for q in test_questions:
output = pipeline.query(q["question"])
results.append({
"question": q["question"],
"answer": output["answer"],
"contexts": output["contexts"],
"ground_truth": q.get("ground_truth"),
})
return results
@pytest.fixture(scope="module")
def ragas_scores(rag_results):
"""تقييم نتائج RAG بـ RAGAS."""
dataset = Dataset.from_dict({
"question": [r["question"] for r in rag_results],
"answer": [r["answer"] for r in rag_results],
"contexts": [r["contexts"] for r in rag_results],
})
return evaluate(
dataset,
metrics=[faithfulness, answer_relevancy, context_precision],
)
class TestRAGQuality:
"""اختبارات جودة RAG مع العتبات."""
def test_faithfulness_threshold(self, ragas_scores):
"""الإجابات يجب أن تكون مؤسسة على السياق."""
assert ragas_scores["faithfulness"] >= 0.8, \
f"الأمانة {ragas_scores['faithfulness']:.2f} تحت العتبة 0.8"
def test_answer_relevancy_threshold(self, ragas_scores):
"""الإجابات يجب أن تعالج الأسئلة."""
assert ragas_scores["answer_relevancy"] >= 0.75, \
f"الصلة {ragas_scores['answer_relevancy']:.2f} تحت العتبة 0.75"
def test_context_precision_threshold(self, ragas_scores):
"""السياق المسترجع يجب أن يكون ذا صلة."""
assert ragas_scores["context_precision"] >= 0.7, \
f"الدقة {ragas_scores['context_precision']:.2f} تحت العتبة 0.7"
class TestRAGPerformance:
"""اختبارات أداء RAG."""
def test_latency_p95(self, rag_results):
"""النسبة المئوية 95 لزمن الاستجابة يجب أن تكون تحت 3 ثواني."""
latencies = [r.get("latency_ms", 0) for r in rag_results]
p95 = sorted(latencies)[int(len(latencies) * 0.95)]
assert p95 < 3000, f"P95 زمن الاستجابة {p95}ms يتجاوز عتبة 3000ms"
كشف التراجع
class RegressionDetector:
"""كشف تراجعات الجودة بين التشغيلات."""
def __init__(self, history_path: str):
self.history_path = Path(history_path)
self.history = self._load_history()
def _load_history(self) -> list[dict]:
if self.history_path.exists():
return json.loads(self.history_path.read_text())
return []
def record_run(self, result: PipelineResult):
"""تسجيل تشغيل اختبار جديد."""
self.history.append({
"timestamp": result.timestamp,
"metrics": result.metrics,
"avg_latency_ms": result.avg_latency_ms,
})
self.history_path.write_text(json.dumps(self.history, indent=2))
def check_regression(
self,
current: dict,
threshold: float = 0.05
) -> list[dict]:
"""تحقق إذا كانت المقاييس الحالية تراجعت من الأساس."""
if len(self.history) < 5:
return [] # تاريخ غير كافٍ
# استخدم آخر 5 تشغيلات كأساس
baseline = {}
for metric in current.keys():
values = [h["metrics"].get(metric, 0) for h in self.history[-5:]]
baseline[metric] = sum(values) / len(values)
regressions = []
for metric, current_value in current.items():
baseline_value = baseline.get(metric, 0)
if baseline_value > 0:
change = (current_value - baseline_value) / baseline_value
if change < -threshold: # أكثر من 5% تراجع
regressions.append({
"metric": metric,
"baseline": baseline_value,
"current": current_value,
"change_pct": change * 100,
})
return regressions
# الاستخدام في خط الأنابيب
detector = RegressionDetector("test_results/history.json")
result = pipeline.run()
detector.record_run(result)
regressions = detector.check_regression(result.metrics)
if regressions:
print("تم اكتشاف تراجع!")
for r in regressions:
print(f" {r['metric']}: {r['baseline']:.2f} -> {r['current']:.2f} ({r['change_pct']:.1f}%)")
التنبيه والمراقبة
import requests
class AlertManager:
"""إرسال تنبيهات لفشل الاختبارات."""
def __init__(self, slack_webhook: str = None, email_config: dict = None):
self.slack_webhook = slack_webhook
self.email_config = email_config
def alert_failure(self, result: PipelineResult):
"""إرسال تنبيهات للاختبارات الفاشلة."""
if result.passed:
return
message = self._format_message(result)
if self.slack_webhook:
self._send_slack(message)
if self.email_config:
self._send_email(message)
def _format_message(self, result: PipelineResult) -> str:
failures = "\n".join([
f"• {f['metric']}: {f['actual']:.2f} (المتوقع {f['expected']})"
for f in result.failures
])
return f"""تنبيه جودة RAG
فشل تشغيل الاختبار في {result.timestamp}
الفشل:
{failures}
المقاييس:
• الأمانة: {result.metrics.get('faithfulness', 0):.2f}
• الصلة: {result.metrics.get('answer_relevancy', 0):.2f}
• الدقة: {result.metrics.get('context_precision', 0):.2f}
• متوسط زمن الاستجابة: {result.avg_latency_ms:.0f}ms
"""
def _send_slack(self, message: str):
requests.post(self.slack_webhook, json={"text": message})
# التكامل
alert_manager = AlertManager(
slack_webhook=os.environ.get("SLACK_WEBHOOK"),
)
result = pipeline.run()
alert_manager.alert_failure(result)
أفضل الممارسات
| الجانب | التوصية |
|---|---|
| التكرار | تشغيل على كل PR + جدولة يومية |
| حجم مجموعة الاختبار | 50-100 سؤال (توازن التكلفة مع التغطية) |
| العتبات | ابدأ محافظاً، شدد مع الوقت |
| التاريخ | احتفظ بـ 30+ يوماً لتحليل الاتجاهات |
| التنبيهات | تنبيه على الفشل والتراجعات |
| التكلفة | استخدم نماذج أرخص للتشغيلات المتكررة |
رؤية رئيسية: الاختبار الآلي يلتقط المشاكل مبكراً، لكن راجع الفشل يدوياً. الدرجات المنخفضة قد تشير لمشاكل في مجموعة الاختبار، ليس مشاكل RAG. حدّث مجموعة الاختبار عندما تجد إيجابيات كاذبة.
الوحدة التالية: أنظمة RAG الإنتاجية - التحسين والموثوقية والمراقبة. :::