Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-evaluation/src/lexigram/ai/evaluation/analysis.py: 40%

1"""Run analysis for the AI evaluation framework. 

2 

3The :class:`ErrorAnalysis` aggregates a tracked run's metric and error 

4records into an :class:`~lexigram.contracts.ai.experiment.AnalysisReport` 

5with error-kind counts, score statistics, and the most frequent errors — 

6the input for post-hoc error analysis of failed trials. 

7""" 

8 

9from __future__ import annotations 

10 

11from collections import Counter 

12 

13from lexigram.ai.evaluation.exceptions import AnalysisError 

14from lexigram.contracts.ai.experiment import ( 

15 AnalysisReport, 

16 ErrorRecord, 

17 ExperimentTrackerProtocol, 

18) 

19 

20_SCORE_METRIC = "score" 

21_TOP_ERRORS_LIMIT = 10 

22 

23 

24class ErrorAnalysis: 

25 """Aggregate a run's tracked records into an analysis report. 

26 

27 Args: 

28 tracker: Tracker holding the run's metric and error records. 

29 """ 

30 

31 def __init__(self, tracker: ExperimentTrackerProtocol) -> None: 

32 self._tracker = tracker 

33 

34 async def report(self, run_id: str) -> AnalysisReport: 

35 """Produce an analysis report for a run. 

36 

37 Args: 

38 run_id: Run identifier. 

39 

40 Returns: 

41 Aggregated error kinds, score statistics, and top errors. 

42 

43 Raises: 

44 AnalysisError: If the run is unknown to the tracker. 

45 """ 

46 run = await self._tracker.resume(run_id) 

47 if run is None: 

48 raise AnalysisError(f"unknown run {run_id!r}") 

49 metrics = await self._tracker.metrics(run_id) 

50 errors = await self._tracker.errors(run_id) 

51 

52 scores = [record.value for record in metrics if record.name == _SCORE_METRIC] 

53 score_stats = ( 

54 (sum(scores) / len(scores), min(scores), max(scores)) 

55 if scores 

56 else (None, None, None) 

57 ) 

58 

59 kinds = Counter(error.kind for error in errors) 

60 first_of_kind: dict[str, ErrorRecord] = {} 

61 for error in errors: 

62 if error.kind not in first_of_kind: 

63 first_of_kind[error.kind] = error 

64 top_errors = tuple( 

65 first_of_kind[kind] for kind, _count in kinds.most_common(_TOP_ERRORS_LIMIT) 

66 ) 

67 

68 return AnalysisReport( 

69 total_records=len(metrics), 

70 error_count=len(errors), 

71 error_kinds=dict(kinds), 

72 score_mean=score_stats[0], 

73 score_min=score_stats[1], 

74 score_max=score_stats[2], 

75 top_errors=top_errors, 

76 ) 

77 

78 

79__all__ = ["ErrorAnalysis"]