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

1"""Semantic memory store — structured subject/predicate/object knowledge.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Any 

6 

7from lexigram.ai.memory.semantic.fact_store import FactStore 

8from lexigram.contracts.core import HealthCheckResult, HealthStatus 

9from lexigram.logging import ( 

10 get_logger, 

11) 

12 

13if TYPE_CHECKING: 

14 from lexigram.ai.memory.semantic.entity_extractor import EntityExtractor 

15 from lexigram.contracts.ai.memory import MemoryEntry 

16 

17logger = get_logger(__name__) 

18 

19 

20class SemanticMemoryStore: 

21 """Semantic memory backed by an in-process FactStore.""" 

22 

23 def __init__( 

24 self, 

25 fact_store: FactStore | None = None, 

26 extractor: EntityExtractor | None = None, 

27 min_confidence: float = 0.5, 

28 max_facts_per_entity: int = 100, 

29 ) -> None: 

30 self._facts = fact_store or FactStore() 

31 self._extractor = extractor 

32 self._min_confidence = min_confidence 

33 self._max_facts = max_facts_per_entity 

34 

35 async def store_fact( 

36 self, 

37 subject: str, 

38 predicate: str, 

39 object_: str, 

40 confidence: float = 1.0, 

41 ) -> str: 

42 existing = self._facts.query_by_subject(subject) 

43 if len(existing) >= self._max_facts: 

44 logger.warning( 

45 "semantic_max_facts_reached", 

46 subject=subject, 

47 cap=self._max_facts, 

48 ) 

49 lowest = min(existing, key=lambda fact: fact["confidence"]) 

50 self._facts.delete(lowest["id"]) 

51 

52 return self._facts.add(subject, predicate, object_, confidence) 

53 

54 async def query_facts(self, subject: str) -> list[dict[str, Any]]: 

55 return [ 

56 fact 

57 for fact in self._facts.query_by_subject(subject) 

58 if fact["confidence"] >= self._min_confidence 

59 ] 

60 

61 async def get_entity_facts(self, entity: str) -> list[dict[str, Any]]: 

62 return [ 

63 fact 

64 for fact in self._facts.get_entity_facts(entity) 

65 if fact["confidence"] >= self._min_confidence 

66 ] 

67 

68 async def update_fact(self, fact_id: str, confidence: float) -> None: 

69 self._facts.update_confidence(fact_id, confidence) 

70 

71 async def ingest(self, entry: MemoryEntry) -> int: 

72 if not self._extractor: 

73 return 0 

74 triples = await self._extractor.extract(entry) 

75 count = 0 

76 for subject, predicate, object_ in triples: 

77 await self.store_fact(subject, predicate, object_, confidence=0.8) 

78 count += 1 

79 return count 

80 

81 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult: 

82 return HealthCheckResult( 

83 component="memory.semantic", 

84 status=HealthStatus.HEALTHY, 

85 details={"facts": len(self._facts), "timeout": timeout}, 

86 ) 

87 

88 

89__all__ = ["SemanticMemoryStore"]