From 95c0f5f4f0d06280fb718e5a3d275f253d8c35a8 Mon Sep 17 00:00:00 2001 From: xcr_1 <530551426@qq.com> Date: Sat, 10 May 2025 10:00:23 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8DBUG=EF=BC=8C=E8=AE=B0?= =?UTF-8?q?=E5=BF=86=E5=8F=AF=E8=83=BD=E4=B8=A2=E5=A4=B1=E7=9A=84=E9=97=AE?= =?UTF-8?q?=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/plugins/memory_system/Hippocampus.py | 94 +++++++++++++----------- 1 file changed, 52 insertions(+), 42 deletions(-) diff --git a/src/plugins/memory_system/Hippocampus.py b/src/plugins/memory_system/Hippocampus.py index 24d320f7..aaa8b0b5 100644 --- a/src/plugins/memory_system/Hippocampus.py +++ b/src/plugins/memory_system/Hippocampus.py @@ -1045,53 +1045,63 @@ class EntorhinalCortex: start_time = time.time() logger.info("[数据库] 开始重新同步所有记忆数据...") - # 清空数据库 - clear_start = time.time() - db.graph_data.nodes.delete_many({}) - db.graph_data.edges.delete_many({}) - clear_end = time.time() - logger.info(f"[数据库] 清空数据库耗时: {clear_end - clear_start:.2f}秒") + with db.client.start_session() as session: + session.start_transaction() + try: + # 清空数据库 + clear_start = time.time() + # delete_many({})是危险操作需要在事务中执行 + db.graph_data.nodes.delete_many({}, session=session) + db.graph_data.edges.delete_many({}, session=session) + clear_end = time.time() + logger.info(f"[数据库] 清空数据库耗时: {clear_end - clear_start:.2f}秒") - # 获取所有节点和边 - memory_nodes = list(self.memory_graph.G.nodes(data=True)) - memory_edges = list(self.memory_graph.G.edges(data=True)) + # 获取所有节点和边 + memory_nodes = list(self.memory_graph.G.nodes(data=True)) + memory_edges = list(self.memory_graph.G.edges(data=True)) - # 重新写入节点 - node_start = time.time() - for concept, data in memory_nodes: - memory_items = data.get("memory_items", []) - if not isinstance(memory_items, list): - memory_items = [memory_items] if memory_items else [] + # 重新写入节点 + node_start = time.time() + for concept, data in memory_nodes: + memory_items = data.get("memory_items", []) + if not isinstance(memory_items, list): + memory_items = [memory_items] if memory_items else [] - node_data = { - "concept": concept, - "memory_items": memory_items, - "hash": self.hippocampus.calculate_node_hash(concept, memory_items), - "created_time": data.get("created_time", datetime.datetime.now().timestamp()), - "last_modified": data.get("last_modified", datetime.datetime.now().timestamp()), - } - db.graph_data.nodes.insert_one(node_data) - node_end = time.time() - logger.info(f"[数据库] 写入 {len(memory_nodes)} 个节点耗时: {node_end - node_start:.2f}秒") + node_data = { + "concept": concept, + "memory_items": memory_items, + "hash": self.hippocampus.calculate_node_hash(concept, memory_items), + "created_time": data.get("created_time", datetime.datetime.now().timestamp()), + "last_modified": data.get("last_modified", datetime.datetime.now().timestamp()), + } + db.graph_data.nodes.insert_one(node_data, session=session) + node_end = time.time() + logger.info(f"[数据库] 写入 {len(memory_nodes)} 个节点耗时: {node_end - node_start:.2f}秒") - # 重新写入边 - edge_start = time.time() - for source, target, data in memory_edges: - edge_data = { - "source": source, - "target": target, - "strength": data.get("strength", 1), - "hash": self.hippocampus.calculate_edge_hash(source, target), - "created_time": data.get("created_time", datetime.datetime.now().timestamp()), - "last_modified": data.get("last_modified", datetime.datetime.now().timestamp()), - } - db.graph_data.edges.insert_one(edge_data) - edge_end = time.time() - logger.info(f"[数据库] 写入 {len(memory_edges)} 条边耗时: {edge_end - edge_start:.2f}秒") + # 重新写入边 + edge_start = time.time() + for source, target, data in memory_edges: + edge_data = { + "source": source, + "target": target, + "strength": data.get("strength", 1), + "hash": self.hippocampus.calculate_edge_hash(source, target), + "created_time": data.get("created_time", datetime.datetime.now().timestamp()), + "last_modified": data.get("last_modified", datetime.datetime.now().timestamp()), + } + db.graph_data.edges.insert_one(edge_data, session=session) + edge_end = time.time() + logger.info(f"[数据库] 写入 {len(memory_edges)} 条边耗时: {edge_end - edge_start:.2f}秒") + + end_time = time.time() + logger.success(f"[数据库] 重新同步完成，总耗时: {end_time - start_time:.2f}秒") + logger.success(f"[数据库] 同步了 {len(memory_nodes)} 个节点和 {len(memory_edges)} 条边") + except Exception as e: + logger.exception(f"[数据库] 重新同步失败: {e}") + session.abort_transaction() + else: + session.commit_transaction() - end_time = time.time() - logger.success(f"[数据库] 重新同步完成，总耗时: {end_time - start_time:.2f}秒") - logger.success(f"[数据库] 同步了 {len(memory_nodes)} 个节点和 {len(memory_edges)} 条边") # 负责整合，遗忘，合并记忆