Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-evaluation/src/lexigram/ai/evaluation/evaluators/embedding_distance.py: 44%

1"""Embedding-based distance evaluator.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING 

6 

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

8from lexigram.contracts.ai.evaluation import ( 

9 EvaluationResult, 

10 EvaluationScoreType, 

11 EvaluatorProtocol, 

12) 

13from lexigram.logging import get_logger 

14from lexigram.result import Ok, Result 

15 

16if TYPE_CHECKING: 

17 from lexigram.contracts.ai.llm import EmbeddingClientProtocol 

18 

19logger = get_logger(__name__) 

20 

21 

22class EmbeddingDistanceEvaluator(BaseEvaluator, EvaluatorProtocol): 

23 """Embedding-based similarity evaluation. 

24 

25 Evaluates output against reference using embedding similarity. 

26 Requires an EmbeddingClientProtocol to be available in the container. 

27 """ 

28 

29 def __init__( 

30 self, 

31 embedding_client: EmbeddingClientProtocol | None = None, 

32 ) -> None: 

33 super().__init__(EvaluationScoreType.SEMANTIC_SIMILARITY) 

34 self._embedding_client = embedding_client 

35 

36 @property 

37 def name(self) -> str: 

38 return "embedding_distance" 

39 

40 def set_embedding_client( 

41 self, 

42 client: EmbeddingClientProtocol, 

43 ) -> None: 

44 self._embedding_client = client 

45 

46 async def evaluate( 

47 self, 

48 input: str, 

49 output: str, 

50 reference: str, 

51 ) -> Result[EvaluationResult, Exception]: 

52 if self._embedding_client is None: 

53 return Ok( 

54 self._create_result( 

55 0.0, 

56 "No embedding client configured", 

57 {"error": "embedding_client_not_available"}, 

58 ) 

59 ) 

60 

61 try: 

62 emb_output = await self._embedding_client.embed([output]) 

63 emb_ref = await self._embedding_client.embed([reference]) 

64 

65 emb_out_list: list[list[float]] = ( 

66 emb_output 

67 if isinstance(emb_output, list) 

68 else getattr(emb_output, "embeddings", emb_output) 

69 ) 

70 emb_ref_list: list[list[float]] = ( 

71 emb_ref 

72 if isinstance(emb_ref, list) 

73 else getattr(emb_ref, "embeddings", emb_ref) 

74 ) 

75 

76 if not emb_out_list or not emb_ref_list: 

77 return Ok( 

78 self._create_result( 

79 0.0, 

80 "Failed to compute embeddings", 

81 {"error": "embedding_computation_failed"}, 

82 ) 

83 ) 

84 

85 similarity = self._cosine_similarity( 

86 emb_out_list[0], 

87 emb_ref_list[0], 

88 ) 

89 

90 return Ok( 

91 self._create_result( 

92 similarity, 

93 f"Semantic similarity: {similarity:.2f}", 

94 {"similarity": similarity}, 

95 ) 

96 ) 

97 except Exception as e: 

98 logger.error("embedding_evaluation_failed", error=str(e)) 

99 return Ok( 

100 self._create_result( 

101 0.0, 

102 f"Evaluation failed: {e}", 

103 {"error": str(e)}, 

104 ) 

105 ) 

106 

107 def _cosine_similarity(self, a: list[float], b: list[float]) -> float: 

108 dot_product: float = sum(x * y for x, y in zip(a, b, strict=True)) 

109 norm_a: float = sum(x * x for x in a) ** 0.5 

110 norm_b: float = sum(x * x for x in b) ** 0.5 

111 if norm_a == 0 or norm_b == 0: 

112 return 0.0 

113 return dot_product / (norm_a * norm_b) 

114 

115 

116__all__ = ["EmbeddingDistanceEvaluator"]