Class MemuClient

evaluation/scripts/utils/client.py:319–354

class MemuClient:
    def __init__(self):
        from memu import MemuClient

        self.memu_client = MemuClient(
            base_url="https://api.memu.so", api_key=os.getenv("MEMU_API_KEY")
        )
        self.agent_id = "assistant_001"

    def add(self, messages, user_id, iso_date):
        try:
            response = self.memu_client.memorize_conversation(
                conversation=messages,
                user_id=user_id,
                user_name=user_id,
                agent_id=self.agent_id,
                agent_name=self.agent_id,
                session_date=iso_date,
            )
            self.wait_for_completion(response.item_id)
        except Exception as error:
            print("❌ Error saving conversation:", error)

    def search(self, query, user_id, top_k):
        user_memories = self.memu_client.retrieve_related_memory_items(
            user_id=user_id, agent_id=self.agent_id, query=query, top_k=top_k, min_similarity=0.1
        )
        res = [m.memory.content for m in user_memories.related_memories]
        return res

    def wait_for_completion(self, task_id):
        while True:
            status = self.memu_client.get_task_status(task_id)
            if status.status in ["SUCCESS", "FAILURE", "REVOKED"]:
                break
            time.sleep(2)


if __name__ == "__main__":