(self, messages, user_id)
| 225 | self.client = Supermemory(api_key=os.getenv("SUPERMEMORY_API_KEY")) |
| 226 | |
| 227 | def add(self, messages, user_id): |
| 228 | content = "\n".join([f"{msg['role']}: {msg['content']}" for msg in messages]) |
| 229 | max_retries = 5 |
| 230 | for attempt in range(max_retries): |
| 231 | try: |
| 232 | self.client.memories.add(content=content, container_tag=user_id) |
| 233 | break |
| 234 | except Exception as e: |
| 235 | if attempt < max_retries - 1: |
| 236 | time.sleep(2**attempt) |
| 237 | else: |
| 238 | raise e |
| 239 | |
| 240 | def search(self, query, user_id, top_k): |
| 241 | max_retries = 10 |
no test coverage detected