Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-memory/src/lexigram/ai/memory/semantic/fact_store.py: 63%

1"""In-process fact store — stores subject/predicate/object triples.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass, field 

6from uuid import uuid4 

7 

8from lexigram.logging import ( 

9 get_logger, 

10) 

11 

12logger = get_logger(__name__) 

13 

14 

15@dataclass 

16class StoredFact: 

17 """A single structured knowledge triple.""" 

18 

19 id: str 

20 subject: str 

21 predicate: str 

22 object_: str 

23 confidence: float = 1.0 

24 metadata: dict = field(default_factory=dict) 

25 

26 

27class FactStore: 

28 """In-memory graph of subject/predicate/object facts. 

29 

30 Used by SemanticMemoryStore to persist extracted knowledge. 

31 """ 

32 

33 def __init__(self) -> None: 

34 """Initialise an empty fact store.""" 

35 self._facts: dict[str, StoredFact] = {} 

36 

37 def add( 

38 self, 

39 subject: str, 

40 predicate: str, 

41 object_: str, 

42 confidence: float = 1.0, 

43 metadata: dict | None = None, 

44 ) -> str: 

45 """Add a new fact triple. 

46 

47 Args: 

48 subject: Subject entity. 

49 predicate: Relationship type. 

50 object_: Object value. 

51 confidence: Confidence score in [0.0, 1.0]. 

52 metadata: Optional additional metadata. 

53 

54 Returns: 

55 Unique ID assigned to the stored fact. 

56 """ 

57 fact_id = str(uuid4()) 

58 self._facts[fact_id] = StoredFact( 

59 id=fact_id, 

60 subject=subject, 

61 predicate=predicate, 

62 object_=object_, 

63 confidence=confidence, 

64 metadata=metadata or {}, 

65 ) 

66 logger.debug( 

67 "fact_added", fact_id=fact_id, subject=subject, predicate=predicate 

68 ) 

69 return fact_id 

70 

71 def query_by_subject(self, subject: str) -> list[dict]: 

72 """Return all facts where subject matches (case-insensitive prefix). 

73 

74 Args: 

75 subject: Subject to filter by. 

76 

77 Returns: 

78 List of fact dicts (id, subject, predicate, object_, confidence). 

79 """ 

80 lower = subject.lower() 

81 return [ 

82 { 

83 "id": f.id, 

84 "subject": f.subject, 

85 "predicate": f.predicate, 

86 "object_": f.object_, 

87 "confidence": f.confidence, 

88 **f.metadata, 

89 } 

90 for f in self._facts.values() 

91 if f.subject.lower().startswith(lower) 

92 ] 

93 

94 def get_entity_facts(self, entity: str) -> list[dict]: 

95 """Return all facts mentioning *entity* as subject or object. 

96 

97 Args: 

98 entity: Entity name to search. 

99 

100 Returns: 

101 List of matching fact dicts. 

102 """ 

103 lower = entity.lower() 

104 return [ 

105 { 

106 "id": f.id, 

107 "subject": f.subject, 

108 "predicate": f.predicate, 

109 "object_": f.object_, 

110 "confidence": f.confidence, 

111 **f.metadata, 

112 } 

113 for f in self._facts.values() 

114 if lower in f.subject.lower() or lower in f.object_.lower() 

115 ] 

116 

117 def update_confidence(self, fact_id: str, confidence: float) -> None: 

118 """Update the confidence of an existing fact. 

119 

120 Args: 

121 fact_id: ID of the fact to update. 

122 confidence: New confidence value in [0.0, 1.0]. 

123 """ 

124 if fact_id in self._facts: 

125 fact = self._facts[fact_id] 

126 self._facts[fact_id] = StoredFact( 

127 id=fact.id, 

128 subject=fact.subject, 

129 predicate=fact.predicate, 

130 object_=fact.object_, 

131 confidence=confidence, 

132 metadata=fact.metadata, 

133 ) 

134 

135 def delete(self, fact_id: str) -> None: 

136 """Remove a fact. 

137 

138 Args: 

139 fact_id: ID of the fact to remove. 

140 """ 

141 self._facts.pop(fact_id, None) 

142 

143 def clear(self) -> None: 

144 """Remove all facts.""" 

145 self._facts.clear() 

146 

147 def __len__(self) -> int: 

148 return len(self._facts) 

149 

150 

151__all__ = ["FactStore", "StoredFact"]