Coverage for src/lexigram/ai/evaluation/evaluators/trajectory.py: 30%

1"""Agent trajectory evaluator.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7from lexigram.ai.evaluation.evaluators.base import BaseEvaluator 

8from lexigram.contracts.ai.evaluation import ( 

9 EvaluationResult, 

10 EvaluationScoreType, 

11 EvaluatorProtocol, 

12) 

13from lexigram.contracts.ai.exceptions import EvaluationError 

14from lexigram.logging import get_logger 

15from lexigram.result import Ok, Result 

16 

17logger = get_logger(__name__) 

18 

19 

20class TrajectoryEvaluator(BaseEvaluator, EvaluatorProtocol): 

21 """Agent trajectory fidelity evaluation. 

22 

23 Evaluates whether an agent's execution trajectory follows the 

24 expected path and reaches the expected final state. 

25 """ 

26 

27 def __init__(self) -> None: 

28 super().__init__(EvaluationScoreType.TRAJECTORY_FIDELITY) 

29 

30 @property 

31 def name(self) -> str: 

32 return "trajectory" 

33 

34 async def evaluate( 

35 self, 

36 input: str, 

37 output: str, 

38 reference: str, 

39 ) -> Result[EvaluationResult, Exception]: 

40 details: dict[str, Any] = {} 

41 

42 try: 

43 from lexigram.serialization import loads as json_loads 

44 

45 output_trajectory = json_loads(output) 

46 reference_trajectory = json_loads(reference) 

47 except (ValueError, EvaluationError) as e: 

48 return Ok( 

49 self._create_result( 

50 0.0, 

51 "Invalid trajectory format: expected JSON", 

52 {"error": "invalid_json"}, 

53 ) 

54 ) 

55 

56 steps_score = self._evaluate_steps( 

57 output_trajectory.get("steps", []), 

58 reference_trajectory.get("steps", []), 

59 ) 

60 final_state_score = self._evaluate_final_state( 

61 output_trajectory.get("final_state", {}), 

62 reference_trajectory.get("final_state", {}), 

63 ) 

64 

65 score = (steps_score + final_state_score) / 2.0 

66 

67 details = { 

68 "steps_score": steps_score, 

69 "final_state_score": final_state_score, 

70 "output_steps": len(output_trajectory.get("steps", [])), 

71 "reference_steps": len(reference_trajectory.get("steps", [])), 

72 } 

73 

74 feedback = f"Trajectory fidelity: {score:.2f}" 

75 

76 return Ok(self._create_result(score, feedback, details)) 

77 

78 def _evaluate_steps( 

79 self, 

80 output_steps: list[dict[str, Any]], 

81 reference_steps: list[dict[str, Any]], 

82 ) -> float: 

83 if not reference_steps: 

84 return 1.0 if output_steps else 0.0 

85 

86 correct = 0 

87 for i, ref_step in enumerate(reference_steps): 

88 if i < len(output_steps): 

89 out_step = output_steps[i] 

90 if self._step_matches(out_step, ref_step): 

91 correct += 1 

92 

93 return correct / len(reference_steps) 

94 

95 def _step_matches( 

96 self, 

97 output_step: dict[str, Any], 

98 reference_step: dict[str, Any], 

99 ) -> bool: 

100 action_match = output_step.get("action") == reference_step.get("action") 

101 if not action_match: 

102 return False 

103 

104 if "tool" in reference_step: 

105 return output_step.get("tool") == reference_step.get("tool") 

106 

107 return True 

108 

109 def _evaluate_final_state( 

110 self, 

111 output_state: dict[str, Any], 

112 reference_state: dict[str, Any], 

113 ) -> float: 

114 if not reference_state: 

115 return 1.0 

116 

117 matches = 0 

118 total = len(reference_state) 

119 

120 for key, expected_value in reference_state.items(): 

121 if key in output_state: 

122 if output_state[key] == expected_value: 

123 matches += 1 

124 

125 return matches / total if total > 0 else 0.0 

126 

127 

128__all__ = ["TrajectoryEvaluator"]