Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-rag/src/lexigram/ai/rag/multimodal/types/multimodal.py: 63%

1"""Combined multimodal document types.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass 

6from datetime import UTC, datetime 

7from typing import Any 

8 

9from lexigram.ai.rag.multimodal.types.audio import AudioDocument 

10from lexigram.ai.rag.multimodal.types.enums import Modality 

11from lexigram.ai.rag.multimodal.types.image import ImageDocument 

12from lexigram.ai.rag.multimodal.types.video import VideoDocument 

13from lexigram.domain import DomainModel 

14from lexigram.validation import Field 

15 

16 

17@dataclass(init=False) 

18class MultiModalEmbedding(DomainModel): 

19 """Combined embeddings for multi-modal documents.""" 

20 

21 text: list[float] | None = None 

22 image: list[float] | None = None 

23 audio: list[float] | None = None 

24 video: list[float] | None = None 

25 fused: list[float] | None = None 

26 fusion_method: str = "concat" 

27 

28 @property 

29 def available_modalities(self) -> list[Modality]: 

30 modalities = [] 

31 if self.text: 

32 modalities.append(Modality.TEXT) 

33 if self.image: 

34 modalities.append(Modality.IMAGE) 

35 if self.audio: 

36 modalities.append(Modality.AUDIO) 

37 if self.video: 

38 modalities.append(Modality.VIDEO) 

39 return modalities 

40 

41 @property 

42 def has_fused(self) -> bool: 

43 return self.fused is not None and len(self.fused) > 0 

44 

45 model_config = { 

46 "json_schema_extra": { 

47 "example": { 

48 "text": [0.1, 0.2, 0.3], 

49 "image": [0.4, 0.5, 0.6], 

50 "fused": [0.1, 0.2, 0.3, 0.4, 0.5, 0.6], 

51 "fusion_method": "concat", 

52 }, 

53 }, 

54 } 

55 

56 

57@dataclass(init=False) 

58class MultiModalDocument(DomainModel): 

59 """Document combining multiple modalities.""" 

60 

61 text_content: str | None = None 

62 images: list[ImageDocument] = Field(default_factory=list) 

63 audio: list[AudioDocument] = Field(default_factory=list) 

64 videos: list[VideoDocument] = Field(default_factory=list) 

65 metadata: dict[str, Any] = Field(default_factory=dict) 

66 embeddings: MultiModalEmbedding | None = None 

67 created_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) 

68 

69 @property 

70 def modalities(self) -> list[Modality]: 

71 modalities = [] 

72 if self.text_content: 

73 modalities.append(Modality.TEXT) 

74 if self.images: 

75 modalities.append(Modality.IMAGE) 

76 if self.audio: 

77 modalities.append(Modality.AUDIO) 

78 if self.videos: 

79 modalities.append(Modality.VIDEO) 

80 return modalities 

81 

82 @property 

83 def is_multimodal(self) -> bool: 

84 return len(self.modalities) > 1 

85 

86 @property 

87 def has_embeddings(self) -> bool: 

88 return self.embeddings is not None 

89 

90 @property 

91 def total_images(self) -> int: 

92 return len(self.images) 

93 

94 @property 

95 def total_audio(self) -> int: 

96 return len(self.audio) 

97 

98 @property 

99 def total_videos(self) -> int: 

100 return len(self.videos) 

101 

102 model_config = { 

103 "json_schema_extra": { 

104 "example": { 

105 "text_content": "Product review", 

106 "images": [], 

107 "metadata": {"source": "blog", "category": "review"}, 

108 }, 

109 }, 

110 }