Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-memory/src/lexigram/ai/memory/working/conversation_buffer.py: 45%

1"""ConversationBuffer — simple FIFO working memory strategy. 

2 

3Maintains the most recent N conversation turns in memory, evicting oldest 

4entries when limits are exceeded. Useful as a lightweight context window 

5strategy when full episodic/semantic recall is not needed. 

6""" 

7 

8from __future__ import annotations 

9 

10from collections import deque 

11from typing import TYPE_CHECKING 

12 

13from lexigram.logging import ( 

14 get_logger, 

15) 

16 

17if TYPE_CHECKING: 

18 from lexigram.contracts.ai.memory import MemoryEntry 

19 

20logger = get_logger(__name__) 

21 

22 

23class ConversationBuffer: 

24 """FIFO buffer that keeps the most recent conversation turns. 

25 

26 Provides a simple, bounded working-memory strategy that auto-evicts 

27 the oldest entries when ``max_turns`` or ``max_tokens`` limits are hit. 

28 

29 Example:: 

30 

31 buffer = ConversationBuffer(max_turns=20, max_tokens=4096) 

32 await buffer.add(entry) 

33 context = buffer.get_context() 

34 

35 Args: 

36 max_turns: Maximum number of turns to retain. 

37 max_tokens: Soft token cap — oldest entries are evicted until the 

38 total estimated token count is at or below this limit. 

39 Set to 0 to disable token-based eviction. 

40 """ 

41 

42 def __init__( 

43 self, 

44 max_turns: int = 20, 

45 max_tokens: int = 4096, 

46 ) -> None: 

47 """Initialise the conversation buffer. 

48 

49 Args: 

50 max_turns: Maximum number of turns to retain. 

51 max_tokens: Soft token cap (0 = no token limit). 

52 """ 

53 self._max_turns = max_turns 

54 self._max_tokens = max_tokens 

55 self._entries: deque[MemoryEntry] = deque(maxlen=max_turns) 

56 self._total_tokens: int = 0 

57 

58 # ------------------------------------------------------------------ 

59 # Public API 

60 # ------------------------------------------------------------------ 

61 

62 async def add(self, entry: MemoryEntry) -> None: 

63 """Add a memory entry to the buffer. 

64 

65 If adding the entry would exceed ``max_turns``, the oldest entry 

66 is automatically evicted. After insertion, token-based eviction 

67 is applied if ``max_tokens > 0``. 

68 

69 Args: 

70 entry: The memory entry to add. 

71 """ 

72 tokens = self._estimate_tokens(entry) 

73 

74 # If deque is at capacity, account for the evicted entry 

75 if len(self._entries) == self._max_turns: 

76 evicted = self._entries[0] 

77 self._total_tokens -= self._estimate_tokens(evicted) 

78 

79 self._entries.append(entry) 

80 self._total_tokens += tokens 

81 

82 # Token-based eviction 

83 if self._max_tokens > 0: 

84 while self._total_tokens > self._max_tokens and len(self._entries) > 1: 

85 evicted = self._entries.popleft() 

86 self._total_tokens -= self._estimate_tokens(evicted) 

87 

88 logger.debug( 

89 "buffer_add", 

90 entry_id=entry.id, 

91 buffer_size=len(self._entries), 

92 total_tokens=self._total_tokens, 

93 ) 

94 

95 def get_context(self) -> list[MemoryEntry]: 

96 """Return all entries currently in the buffer, oldest-first. 

97 

98 Returns: 

99 Ordered list of memory entries. 

100 """ 

101 return list(self._entries) 

102 

103 def clear(self) -> None: 

104 """Remove all entries from the buffer.""" 

105 self._entries.clear() 

106 self._total_tokens = 0 

107 

108 @property 

109 def size(self) -> int: 

110 """Number of entries currently in the buffer.""" 

111 return len(self._entries) 

112 

113 @property 

114 def total_tokens(self) -> int: 

115 """Estimated total token count across all buffered entries.""" 

116 return self._total_tokens 

117 

118 # ------------------------------------------------------------------ 

119 # Internal 

120 # ------------------------------------------------------------------ 

121 

122 @staticmethod 

123 def _estimate_tokens(entry: MemoryEntry) -> int: 

124 """Rough token estimate: ~4 chars per token. 

125 

126 Args: 

127 entry: Memory entry. 

128 

129 Returns: 

130 Estimated token count. 

131 """ 

132 return max(1, len(entry.content) // 4) 

133 

134 def __len__(self) -> int: 

135 """Number of entries in the buffer.""" 

136 return len(self._entries) 

137 

138 def __repr__(self) -> str: 

139 """String representation.""" 

140 return ( 

141 f"ConversationBuffer(size={len(self._entries)}, " 

142 f"max_turns={self._max_turns}, " 

143 f"tokens={self._total_tokens}/{self._max_tokens})" 

144 ) 

145 

146 

147__all__ = ["ConversationBuffer"]