Push a message to the memory.
(self, role: str, content: str)
| 156 | self.memory = memory |
| 157 | |
| 158 | def push(self, role: str, content: str) -> int: |
| 159 | """Push a message to the memory.""" |
| 160 | ideal_ctx = self.get_ideal_ctx(self.model_provider) |
| 161 | if ideal_ctx is not None: |
| 162 | if self.memory_compression and len(content) > ideal_ctx * 1.5: |
| 163 | self.logger.info(f"Compressing memory: Content {len(content)} > {ideal_ctx} model context.") |
| 164 | self.compress() |
| 165 | curr_idx = len(self.memory) |
| 166 | if self.memory[curr_idx-1]['content'] == content: |
| 167 | pretty_print("Warning: same message have been pushed twice to memory", color="error") |
| 168 | time_str = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") |
| 169 | if config["MAIN"]["provider_name"] == "openrouter": |
| 170 | self.memory.append({'role': role, 'content': content}) |
| 171 | else: |
| 172 | self.memory.append({'role': role, 'content': content, 'time': time_str, 'model_used': self.model_provider}) |
| 173 | return curr_idx-1 |
| 174 | |
| 175 | def clear(self) -> None: |
| 176 | """Clear all memory except system prompt""" |