Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-memory/src/lexigram/ai/memory/episodic/compressor.py: 44%

1"""Episodic memory compressor — summarises old entries to reclaim context space.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING 

6 

7from lexigram.ai.memory.exceptions import ConsolidationError 

8from lexigram.contracts.ai.memory import MemoryEntry 

9from lexigram.logging import ( 

10 get_logger, 

11) 

12 

13if TYPE_CHECKING: 

14 from collections.abc import Awaitable, Callable 

15 

16logger = get_logger(__name__) 

17 

18 

19class EpisodicCompressor: 

20 """Compresses episodic memory entries by LLM-assisted summarisation. 

21 

22 When the episodic store grows beyond a threshold, older entries are 

23 grouped by session and collapsed into a single summary entry. 

24 """ 

25 

26 def __init__( 

27 self, 

28 summarise_fn: Callable[[list[MemoryEntry]], Awaitable[MemoryEntry]] 

29 | None = None, 

30 ) -> None: 

31 """Initialise the compressor. 

32 

33 Args: 

34 summarise_fn: Async callable that accepts a list of entries and 

35 returns a single summary entry. When *None*, a simple 

36 concatenation fallback is used. 

37 """ 

38 self._summarise_fn = summarise_fn 

39 

40 async def compress( 

41 self, 

42 entries: list[MemoryEntry], 

43 *, 

44 max_tokens: int = 200, 

45 ) -> MemoryEntry: 

46 """Compress *entries* into a single condensed memory entry. 

47 

48 Args: 

49 entries: Chronologically ordered entries to compress. 

50 max_tokens: Hint to the summariser for output length. 

51 

52 Returns: 

53 A new MemoryEntry representing the compressed form. 

54 

55 Raises: 

56 ConsolidationError: If the summarisation callable raises. 

57 """ 

58 if not entries: 

59 raise ConsolidationError("Cannot compress an empty list of entries") 

60 

61 if self._summarise_fn: 

62 try: 

63 return await self._summarise_fn(entries) 

64 except (RuntimeError, ValueError, TypeError) as exc: 

65 raise ConsolidationError( 

66 "Summarisation failed during compression" 

67 ) from exc 

68 

69 return self._concatenate_fallback(entries) 

70 

71 def _concatenate_fallback(self, entries: list[MemoryEntry]) -> MemoryEntry: 

72 """Produce a naive concatenation summary when no LLM is configured.""" 

73 from datetime import UTC, datetime 

74 from uuid import uuid4 

75 

76 combined = " | ".join(e.content[:100] for e in entries) 

77 avg_importance = sum(e.importance for e in entries) / len(entries) 

78 return MemoryEntry( 

79 id=str(uuid4()), 

80 owner_id=entries[0].owner_id, 

81 content=f"[summary of {len(entries)} turns] {combined}", 

82 role="system", 

83 timestamp=datetime.now(UTC), 

84 importance=avg_importance, 

85 metadata={"compressed_ids": [e.id for e in entries], "type": "summary"}, 

86 ) 

87 

88 

89# Avoid circular import for type hints 

90 

91__all__ = ["EpisodicCompressor"]