(self,
user_id: Optional[str] = None,
agent_id: Optional[str] = None,
run_id: Optional[str] = None,
memory_ids: Optional[List[str]] = None)
| 481 | return new_messages |
| 482 | |
| 483 | def delete(self, |
| 484 | user_id: Optional[str] = None, |
| 485 | agent_id: Optional[str] = None, |
| 486 | run_id: Optional[str] = None, |
| 487 | memory_ids: Optional[List[str]] = None) -> Tuple[bool, str]: |
| 488 | failed = {} |
| 489 | if memory_ids is None: |
| 490 | try: |
| 491 | self.memory.delete_all( |
| 492 | user_id=user_id, agent_id=agent_id, run_id=run_id) |
| 493 | return True, '' |
| 494 | except Exception as e: |
| 495 | return False, str(e) + '\n' + traceback.format_exc() |
| 496 | for memory_id in memory_ids: |
| 497 | try: |
| 498 | self.memory.delete(memory_id=memory_id) |
| 499 | except IndexError: |
| 500 | failed[ |
| 501 | memory_id] = 'This memory_id does not exist in the database.\n' + traceback.format_exc( |
| 502 | ) # noqa |
| 503 | except Exception as e: |
| 504 | failed[memory_id] = str(e) + '\n' + traceback.format_exc() |
| 505 | if failed: |
| 506 | return False, json.dumps(failed) |
| 507 | else: |
| 508 | return True, '' |
| 509 | |
| 510 | def get_all(self, |
| 511 | user_id: Optional[str] = None, |
no outgoing calls
no test coverage detected