messages = [{"role": "assistant", "content": data, "created_at": iso_date}]
(self, messages, user_id, batch_size=2)
| 101 | ) |
| 102 | |
| 103 | def add(self, messages, user_id, batch_size=2): |
| 104 | """ |
| 105 | messages = [{"role": "assistant", "content": data, "created_at": iso_date}] |
| 106 | """ |
| 107 | from memobase import ChatBlob |
| 108 | |
| 109 | real_uid = self.string_to_uuid(user_id) |
| 110 | user = self.client.get_or_create_user(real_uid) |
| 111 | for i in range(0, len(messages), batch_size): |
| 112 | batch_messages = messages[i : i + batch_size] |
| 113 | max_retries = 5 |
| 114 | for attempt in range(max_retries): |
| 115 | try: |
| 116 | _ = user.insert(ChatBlob(messages=batch_messages), sync=True) |
| 117 | except Exception as e: |
| 118 | if attempt < max_retries - 1: |
| 119 | time.sleep(2**attempt) |
| 120 | else: |
| 121 | raise e |
| 122 | |
| 123 | def search(self, query, user_id, top_k): |
| 124 | real_uid = self.string_to_uuid(user_id) |
nothing calls this directly
no test coverage detected