llm-eval/backend/llm_eval/eval/evaluations/tasks/evaluate_test_case_task.py at e0c414683bbf2a01ef29ef5ed91aa8c00ba5a0c9 · codecentric/llm-eval · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
from datetime import datetime
from uuid import uuid4

from deepeval import evaluate
from deepeval.evaluate import AsyncConfig, DisplayConfig, CacheConfig
from deepeval.evaluate.types import TestResult as EvalTestResult
from deepeval.test_case import LLMTestCase
from loguru import logger
from sqlalchemy.ext.asyncio import AsyncSession

from llm_eval.database.model import (
    TestCase,
    TestCaseEvaluationResult,
    TestCaseStatus,
)
from llm_eval.eval.evaluate_results.db.find_test_case import find_test_case
from llm_eval.metrics.db.find_metric import find_metrics_by_ids
from llm_eval.metrics.plugins.factory import get_metric_plugin
from llm_eval.metrics.plugins.impl.metric_wrapper import MetricWrapper
from llm_eval.settings import SETTINGS
from llm_eval.tasks import app
from llm_eval.utils.task import async_task, with_session


@app.task(
    autoretry_for=(Exception,),
    retry_backoff=True,
    retry_backoff_max=600,
    max_retries=10,
)
@async_task
@with_session
async def evaluate_test_case_task(
    session: AsyncSession, test_case_id: str, metric_ids: list[str]
) -> None:
    logger.info(f"Evaluating test case '{test_case_id}' with metrics {str(metric_ids)}")

    test_case = await find_test_case(session, test_case_id)

    if test_case is None:
        logger.error(f"Test case '{test_case_id}' not found.")
        return

    if test_case.status != TestCaseStatus.EVALUATING:
        logger.info(f"Test case '{test_case_id}' not in evaluating state. Ignoring...")
        return

    metrics = await _build_metrics(session, metric_ids)

    if len(metric_ids) > 0:
        llm_test_case = LLMTestCase(
            input=test_case.input,
            actual_output=test_case.actual_output,
            expected_output=test_case.expected_output,
            context=test_case.context or [],
            retrieval_context=test_case.retrieval_context or [],
            additional_metadata=test_case.meta_data,
        )
        evaluate_result = evaluate(
            test_cases=[llm_test_case],
            metrics=metrics,
            async_config=AsyncConfig(run_async=False, max_concurrent=1),
            display_config=DisplayConfig(
                print_results=False, show_indicator=SETTINGS.evaluation.show_indicator
            ),
            cache_config=CacheConfig(write_cache=False, use_cache=False),
        )

        evaluation_results = _build_evaluation_results(
            test_case, metrics, evaluate_result.test_results[0]
        )

        test_case.evaluation_results = evaluation_results

    test_case.status = TestCaseStatus.SUCCESS


async def _build_metrics(
    session: AsyncSession, metric_ids: list[str]
) -> list[MetricWrapper]:
    evaluation_metrics = await find_metrics_by_ids(session, metric_ids)

    metrics: list[MetricWrapper] = []

    for metric in evaluation_metrics:
        metric_plugin = get_metric_plugin(metric)

        configuration = metric_plugin.configuration_from_db_json(metric.metric_config)
        metrics.append(
            MetricWrapper(
                evaluation_metric_id=metric.id,
                name=configuration.name,
                metric=await metric_plugin.create_deepeval_metric(
                    session, configuration
                ),
            )
        )

    return metrics


def _build_evaluation_results(
    test_case: TestCase, metrics: list[MetricWrapper], eval_test_result: EvalTestResult
) -> list[TestCaseEvaluationResult]:
    now = datetime.now()
    test_case_evaluation_results: list[TestCaseEvaluationResult] = []

    for metric_result in eval_test_result.metrics_data:
        metric_wrapper = next((x for x in metrics if x.matches(metric_result)), None)
        test_case_evaluation_results.append(
            TestCaseEvaluationResult(
                id=str(uuid4()),
                created_at=now,
                name=metric_wrapper.__name__,
                threshold=metric_result.threshold,
                success=metric_result.success,
                score=metric_result.score,
                reason=metric_result.reason,
                strict_mode=metric_result.strict_mode,
                evaluation_model=metric_result.evaluation_model,
                evaluation_cost=metric_result.evaluation_cost,
                verbose_logs=metric_result.verbose_logs,
                test_case_id=test_case.id,
                evaluation_metric_id=metric_wrapper.evaluation_metric_id,
            )
        )

    return test_case_evaluation_results