Merge branch 'release/v0.2.3' into develop

# Conflicts:
#	api/app/core/agent/langchain_agent.py
#	api/app/core/memory/agent/langgraph_graph/write_graph.py
#	api/app/repositories/neo4j/graph_saver.py
#	api/app/services/draft_run_service.py

api/app/core/memory/agent/langgraph_graph/routing/write_router.py

@@ -1,8 +1,9 @@
+import json
 import os
 
 from app.core.logging_config import get_agent_logger
-from app.core.memory.agent.langgraph_graph.tools.write_tool import chat_data_format, format_parsing
-from app.core.memory.agent.langgraph_graph.write_graph import make_write_graph
+from app.core.memory.agent.langgraph_graph.tools.write_tool import  format_parsing,  messages_parse
+from app.core.memory.agent.langgraph_graph.write_graph import make_write_graph, long_term_storage
 
 from app.core.memory.agent.models.write_aggregate_model import WriteAggregateModel
 from app.core.memory.agent.utils.llm_tools import PROJECT_ROOT_
@@ -10,46 +11,108 @@ from app.core.memory.agent.utils.redis_tool import write_store
 from app.core.memory.agent.utils.redis_tool import count_store
 from app.core.memory.agent.utils.template_tools import TemplateService
 from app.core.memory.utils.llm.llm_utils import MemoryClientFactory
-from app.db import get_db_context
+from app.db import get_db_context, get_db
+from app.repositories.memory_short_repository import LongTermMemoryRepository
+from app.schemas.memory_agent_schema import AgentMemory_Long_Term
+from app.services.memory_konwledges_server import write_rag
+from app.services.task_service import get_task_memory_write_result
+from app.tasks import write_message_task
+from app.utils.config_utils import resolve_config_id
 logger = get_agent_logger(__name__)
 template_root = os.path.join(PROJECT_ROOT_, 'memory', 'agent', 'utils', 'prompt')
 
+async def write_rag_agent(end_user_id, user_message, ai_message, user_rag_memory_id):
+    # RAG 模式：组合消息为字符串格式（保持原有逻辑）
+    combined_message = f"user: {user_message}\nassistant: {ai_message}"
+    await write_rag(end_user_id, combined_message, user_rag_memory_id)
+    logger.info(f'RAG_Agent:{end_user_id};{user_rag_memory_id}')
+async def write(storage_type, end_user_id, user_message, ai_message, user_rag_memory_id, actual_end_user_id,
+                actual_config_id, long_term_messages=[]):
+    """
+    写入记忆（支持结构化消息）
 
-async def write_messages(end_user_id,langchain_messages,memory_config):
-    '''
-    写入数据到neo4j：
-     Args:
+    Args:
+        storage_type: 存储类型 (neo4j/rag)
         end_user_id: 终端用户ID
-        memory_config: 内存配置对象
-        langchain_messages：原始数据LIST
-    '''
+        user_message: 用户消息内容
+        ai_message: AI 回复内容
+        user_rag_memory_id: RAG 记忆ID
+        actual_end_user_id: 实际用户ID
+        actual_config_id: 配置ID
+
+    逻辑说明：
+    - RAG 模式：组合 user_message 和 ai_message 为字符串格式，保持原有逻辑不变
+    - Neo4j 模式：使用结构化消息列表
+      1. 如果 user_message 和 ai_message 都不为空：创建配对消息 [user, assistant]
+      2. 如果只有 user_message：创建单条用户消息 [user]（用于历史记忆场景）
+      3. 每条消息会被转换为独立的 Chunk，保留 speaker 字段
+    """
+
+    db = next(get_db())
     try:
+        actual_config_id = resolve_config_id(actual_config_id, db)
+        # Neo4j 模式：使用结构化消息列表
+        structured_messages = []
+
+        # 始终添加用户消息（如果不为空）
+        if isinstance(user_message, str) and user_message.strip() != "":
+            structured_messages.append({"role": "user", "content": user_message})
+
+        # 只有当 AI 回复不为空时才添加 assistant 消息
+        if isinstance(ai_message, str) and ai_message.strip() != "":
+            structured_messages.append({"role": "assistant", "content": ai_message})
+
+        # 如果提供了 long_term_messages，使用它替代 structured_messages
+        if long_term_messages and isinstance(long_term_messages, list):
+            structured_messages = long_term_messages
+        elif long_term_messages and isinstance(long_term_messages, str):
+            # 如果是 JSON 字符串，先解析
+            try:
+                structured_messages = json.loads(long_term_messages)
+            except json.JSONDecodeError:
+                logger.error(f"Failed to parse long_term_messages as JSON: {long_term_messages}")
+
+        # 如果没有消息，直接返回
+        if not structured_messages:
+            logger.warning(f"No messages to write for user {actual_end_user_id}")
+            return
+
+        logger.info(
+            f"[WRITE] Submitting Celery task - user={actual_end_user_id}, messages={len(structured_messages)}, config={actual_config_id}")
+        write_id = write_message_task.delay(
+            actual_end_user_id,  # end_user_id: 用户ID
+            structured_messages,  # message: JSON 字符串格式的消息列表
+            str(actual_config_id),  # config_id: 配置ID字符串
+            storage_type,  # storage_type: "neo4j"
+            user_rag_memory_id or ""  # user_rag_memory_id: RAG记忆ID（Neo4j模式下不使用）
+        )
+        logger.info(f"[WRITE] Celery task submitted - task_id={write_id}")
+        write_status = get_task_memory_write_result(str(write_id))
+        logger.info(f'[WRITE] Task result - user={actual_end_user_id}, status={write_status}')
+    finally:
+        db.close()
+
+async def term_memory_save(long_term_messages,actual_config_id,end_user_id,type,scope):
+    with get_db_context() as db_session:
+        repo = LongTermMemoryRepository(db_session)
+
+
+        from app.core.memory.agent.utils.redis_tool import write_store
+        result = write_store.get_session_by_userid(end_user_id)
+        if type==AgentMemory_Long_Term.STRATEGY_CHUNK or AgentMemory_Long_Term.STRATEGY_AGGREGATE:
+            data = await format_parsing(result, "dict")
+            chunk_data = data[:scope]
+            if len(chunk_data)==scope:
+                repo.upsert(end_user_id, chunk_data)
+                logger.info(f'---------写入短长期-----------')
+        else:
+            long_time_data = write_store.find_user_recent_sessions(end_user_id, 5)
+            long_messages = await messages_parse(long_time_data)
+            repo.upsert(end_user_id, long_messages)
+            logger.info(f'写入短长期：')
+
 
-        async with make_write_graph() as graph:
-            config = {"configurable": {"thread_id": end_user_id}}
-            # 初始状态 - 包含所有必要字段
-            initial_state = {
-                "messages": langchain_messages,
-                "end_user_id": end_user_id,
-                "memory_config": memory_config
-            }
 
-            # 获取节点更新信息
-            async for update_event in graph.astream(
-                    initial_state,
-                    stream_mode="updates",
-                    config=config
-            ):
-                for node_name, node_data in update_event.items():
-                    if 'save_neo4j' == node_name:
-                        massages = node_data
-            # TODO：删除
-            massagesstatus = massages.get('write_result')['status']
-            contents = massages.get('write_result')
-            print(contents)
-    except Exception as e:
-        import traceback
-        traceback.print_exc()
 '''根据窗口'''
 async def window_dialogue(end_user_id,langchain_messages,memory_config,scope):
     '''
@@ -61,25 +124,26 @@ async def window_dialogue(end_user_id,langchain_messages,memory_config,scope):
         scope：窗口大小
     '''
     scope=scope
-    redis_messages = []
     is_end_user_id = count_store.get_sessions_count(end_user_id)
     if is_end_user_id is not False:
         is_end_user_id = count_store.get_sessions_count(end_user_id)[0]
         redis_messages = count_store.get_sessions_count(end_user_id)[1]
     if is_end_user_id and int(is_end_user_id) != int(scope):
-        print(is_end_user_id)
         is_end_user_id += 1
         langchain_messages += redis_messages
         count_store.update_sessions_count(end_user_id, is_end_user_id, langchain_messages)
     elif int(is_end_user_id) == int(scope):
-        print('写入长期记忆，并且设置为0')
-        print(is_end_user_id)
-        formatted_messages = await chat_data_format(redis_messages)
-        print(100*'-')
-        print(formatted_messages)
-        print(100*'-')
-        await write_messages(end_user_id, formatted_messages, memory_config)
-        count_store.update_sessions_count(end_user_id, 0, '')
+        logger.info('写入长期记忆NEO4J')
+        formatted_messages = (redis_messages)
+        # 获取 config_id（如果 memory_config 是对象，提取 config_id；否则直接使用）
+        if hasattr(memory_config, 'config_id'):
+            config_id = memory_config.config_id
+        else:
+            config_id = memory_config
+        
+        await write(AgentMemory_Long_Term.STORAGE_NEO4J, end_user_id, "", "", None, end_user_id,
+                config_id, formatted_messages)
+        count_store.update_sessions_count(end_user_id, 1, langchain_messages)
     else:
         count_store.save_sessions_count(end_user_id, 1, langchain_messages)
 
@@ -93,12 +157,15 @@ async def memory_long_term_storage(end_user_id,memory_config,time):
         memory_config: 内存配置对象
     '''
     long_time_data = write_store.find_user_recent_sessions(end_user_id, time)
-    # Handle case where no session exists in Redis (returns False or empty)
-    if not long_time_data or long_time_data is False:
-        return
-    format_messages = await chat_data_format(long_time_data)
+    format_messages = (long_time_data)
+    messages=[]
+    memory_config=memory_config.config_id
+    for i in format_messages:
+        message=json.loads(i['Query'])
+        messages+= message
     if format_messages!=[]:
-        await write_messages(end_user_id, format_messages, memory_config)
+        await write(AgentMemory_Long_Term.STORAGE_NEO4J, end_user_id, "", "", None, end_user_id,
+                    memory_config, messages)
 '''聚合判断'''
 async def aggregate_judgment(end_user_id: str, ori_messages: list, memory_config) -> dict:
     """
@@ -109,13 +176,12 @@ async def aggregate_judgment(end_user_id: str, ori_messages: list, memory_config
         ori_messages: 原始消息列表，格式如 [{"role": "user", "content": "..."}, {"role": "assistant", "content": "..."}]
         memory_config: 内存配置对象
     """
-    
+
     try:
         # 1. 获取历史会话数据（使用新方法）
         result = write_store.get_all_sessions_by_end_user_id(end_user_id)
-        
-        # Handle case where no session exists in Redis (returns False or empty)
-        if not result or result is False:
+        history = await format_parsing(result)
+        if not result:
             history = []
         else:
             history = await format_parsing(result)
@@ -154,7 +220,8 @@ async def aggregate_judgment(end_user_id: str, ori_messages: list, memory_config
         }
         if not structured.is_same_event:
             logger.info(result_dict)
-            await write_messages(end_user_id, output_value, memory_config)
+            await write("neo4j", end_user_id, "", "", None, end_user_id,
+                        memory_config.config_id, output_value)
         return result_dict
         
     except Exception as e:

api/app/core/memory/agent/langgraph_graph/tools/tool.py

@@ -186,10 +186,11 @@ def create_hybrid_retrieval_tool_async(memory_config, **search_params):
             清理后的数据
         """
         # 需要过滤的字段列表
+        # TODO: fact_summary 功能暂时禁用，待后续开发完善后启用
         fields_to_remove = {
             'invalid_at', 'valid_at', 'chunk_id_from_rel', 'entity_ids', 
             'expired_at', 'created_at', 'chunk_id', 'id', 'apply_id', 
-            'user_id', 'statement_ids', 'updated_at',"chunk_ids","fact_summary"
+            'user_id', 'statement_ids', 'updated_at',"chunk_ids"  ,"fact_summary"
         }
         
         if isinstance(data, dict):

api/app/core/memory/agent/langgraph_graph/tools/write_tool.py

@@ -1,8 +1,6 @@
 import json
 
 from langchain_core.messages import HumanMessage, AIMessage
-
-
 async def format_parsing(messages: list,type:str='string'):
     """
     格式化解析消息列表
@@ -26,13 +24,13 @@ async def format_parsing(messages: list,type:str='string'):
                 role = content['role']
                 content = content['content']
                 if type == "string":
-                    if role == 'human':
+                    if role == 'human' or role=="user":
                         content = '用户:' + content
                     else:
                         content = 'AI:' + content
                     result.append(content)
-                if type == "dict":
-                    if role == 'human':
+                if type == "dict" :
+                    if role == 'human'  or role=="user":
                         user.append( content)
                     else:
                         ai.append(content)
@@ -57,33 +55,7 @@ async def messages_parse(messages: list | dict):
     for key, values in zip(user, ai):
         database.append({key, values})
     return  database
-async def chat_data_format(messages: list | dict):
-    """
-    将消息格式化为 LangChain 消息格式
-    
-    Args:
-        messages: 消息列表或字典
-        
-    Returns:
-        LangChain 消息列表
-    """
-    langchain_messages = []
-    if isinstance(messages, list):
-        for msg in messages:
-            if 'role' in msg.keys():
-                if msg['role'] == 'user':
-                    langchain_messages.append(HumanMessage(content=msg['content']))
-                elif msg['role'] == 'assistant':
-                    langchain_messages.append(AIMessage(content=msg['content']))
-            if "Query" in msg.keys():
-                langchain_messages.append(HumanMessage(content=msg['Query']))
-                langchain_messages.append(AIMessage(content=msg['Answer']))
-    if isinstance(messages, dict):
-        if messages['type'] == 'human':
-            langchain_messages.append(HumanMessage(content=messages['content']))
-        elif messages['type'] == 'ai':
-            langchain_messages.append(AIMessage(content=messages['content']))
-    return langchain_messages
+
 
 async def agent_chat_messages(user_content,ai_content):
     messages = [

api/app/core/memory/agent/langgraph_graph/write_graph.py

@@ -1,13 +1,18 @@
 import asyncio
+import json
 import sys
 import warnings
 from contextlib import asynccontextmanager
 from langgraph.constants import END, START
 from langgraph.graph import StateGraph
 
+from app.db import get_db, get_db_context
 from app.core.logging_config import get_agent_logger
 from app.core.memory.agent.utils.llm_tools import WriteState
 from app.core.memory.agent.langgraph_graph.nodes.write_nodes import write_node
+from app.schemas.memory_agent_schema import AgentMemory_Long_Term
+from app.services.memory_config_service import MemoryConfigService
+
 
 warnings.filterwarnings("ignore", category=RuntimeWarning)
 logger = get_agent_logger(__name__)
@@ -37,76 +42,61 @@ async def make_write_graph():
 
     yield graph
 
-
-async def long_term_storage(long_term_type: str = "chunk", langchain_messages: list = [], memory_config: str = '',
-                            end_user_id: str = '', scope: int = 6):
-    """Dispatch long-term memory storage to Celery background tasks.
-
-    Args:
-        long_term_type: Storage strategy - 'chunk' (window), 'time', or 'aggregate'
-        langchain_messages: List of messages to store
-        memory_config: Memory configuration ID (string)
-        end_user_id: End user identifier
-        scope: Window size for 'chunk' strategy (default: 6)
-    """
-    from app.tasks import (
-        long_term_storage_window_task,
-        # TODO: Uncomment when implemented
-        # long_term_storage_time_task,
-        # long_term_storage_aggregate_task,
-    )
-    from app.core.logging_config import get_logger
-
-    logger = get_logger(__name__)
-
-    # Convert config to string if needed
-    config_id = str(memory_config) if memory_config else ''
-
-    if long_term_type == 'chunk':
-        # Strategy 1: Window-based batching (6 rounds of dialogue)
-        logger.info(f"[LONG_TERM] Dispatching window task - end_user_id={end_user_id}, scope={scope}")
-        long_term_storage_window_task.delay(
-            end_user_id=end_user_id,
-            langchain_messages=langchain_messages,
-            config_id=config_id,
-            scope=scope
+async def long_term_storage(long_term_type:str="chunk",langchain_messages:list=[],memory_config:str='',end_user_id:str='',scope:int=6):
+    from app.core.memory.agent.langgraph_graph.routing.write_router import memory_long_term_storage, window_dialogue,aggregate_judgment
+    from app.core.memory.agent.utils.redis_tool import write_store
+    write_store.save_session_write(end_user_id,  (langchain_messages))
+    # 获取数据库会话
+    with get_db_context() as db_session:
+        config_service = MemoryConfigService(db_session)
+        memory_config = config_service.load_memory_config(
+            config_id=memory_config,  # 改为整数
+            service_name="MemoryAgentService"
         )
-    # TODO: Uncomment when time-based strategy is fully implemented
-    # elif long_term_type == 'time':
-    #     # Strategy 2: Time-based retrieval
-    #     logger.info(f"[LONG_TERM] Dispatching time task - end_user_id={end_user_id}")
-    #     long_term_storage_time_task.delay(
-    #         end_user_id=end_user_id,
-    #         config_id=config_id,
-    #         time_window=5
-    #     )
-    # TODO: Uncomment when aggregate strategy is fully implemented
-    # elif long_term_type == 'aggregate':
-    #     # Strategy 3: Aggregate judgment (deduplication)
-    #     logger.info(f"[LONG_TERM] Dispatching aggregate task - end_user_id={end_user_id}")
-    #     long_term_storage_aggregate_task.delay(
-    #         end_user_id=end_user_id,
-    #         langchain_messages=langchain_messages,
-    #         config_id=config_id
-    #     )
+        if long_term_type=='chunk':
+            '''方案一:对话窗口6轮对话'''
+            await window_dialogue(end_user_id,langchain_messages,memory_config,scope)
+        if long_term_type=='time':
+            """时间"""
+            await memory_long_term_storage(end_user_id, memory_config,5)
+        if  long_term_type=='aggregate':
+            """方案三：聚合判断"""
+            await aggregate_judgment(end_user_id, langchain_messages, memory_config)
+
+
+
+async def write_long_term(storage_type,end_user_id,message_chat,aimessages,user_rag_memory_id,actual_config_id):
+    from app.core.memory.agent.langgraph_graph.routing.write_router import write_rag_agent
+    from app.core.memory.agent.langgraph_graph.routing.write_router import term_memory_save
+    from app.core.memory.agent.langgraph_graph.tools.write_tool import  agent_chat_messages
+    if storage_type == AgentMemory_Long_Term.STORAGE_RAG:
+        await write_rag_agent(end_user_id, message_chat, aimessages, user_rag_memory_id)
+    else:
+        # AI 回复写入（用户消息和 AI 回复配对，一次性写入完整对话）
+        CHUNK = AgentMemory_Long_Term.STRATEGY_CHUNK
+        SCOPE = AgentMemory_Long_Term.DEFAULT_SCOPE
+        long_term_messages = await agent_chat_messages(message_chat, aimessages)
+        await long_term_storage(long_term_type=CHUNK, langchain_messages=long_term_messages,
+                                memory_config=actual_config_id, end_user_id=end_user_id, scope=SCOPE)
+        await term_memory_save(long_term_messages, actual_config_id, end_user_id, CHUNK, scope=SCOPE)
 
 # async def main():
 #     """主函数 - 运行工作流"""
 #     langchain_messages = [
 #     {
 #       "role": "user",
-#       "content": "今天周五好开心啊"
+#       "content": "今天周五去爬山"
 #     },
 #     {
 #       "role": "assistant",
-#       "content": "你也这么觉得，我也是耶"
+#       "content": "好耶"
 #     }
 #
 #   ]
 #     end_user_id = '837fee1b-04a2-48ee-94d7-211488908940'  # 组ID
 #     memory_config="08ed205c-0f05-49c3-8e0c-a580d28f5fd4"
-#     # await long_term_storage(long_term_type="chunk",langchain_messages=langchain_messages,memory_config=memory_config,end_user_id=end_user_id,scope=2)
-#     result=await long_term_storage(long_term_type="chunk",langchain_messages=langchain_messages,memory_config=memory_config,end_user_id=end_user_id,scope=2)
+#     await long_term_storage(long_term_type="chunk",langchain_messages=langchain_messages,memory_config=memory_config,end_user_id=end_user_id,scope=2)
+#
 #
 #
 # if __name__ == "__main__":

api/app/core/memory/agent/utils/redis_tool.py

@@ -294,6 +294,7 @@ class RedisCountStore:
         """
         session_id = str(uuid.uuid4())
         key = generate_session_key(session_id, key_type="count")
+        index_key = f'session:count:index:{end_user_id}'  # 索引键
         
         pipe = self.r.pipeline()
         pipe.hset(key, mapping={
@@ -304,6 +305,10 @@ class RedisCountStore:
             "starttime": get_current_timestamp()
         })
         pipe.expire(key, 30 * 24 * 60 * 60)  # 30天过期
+        
+        # 创建索引：end_user_id -> session_id 映射
+        pipe.set(index_key, session_id, ex=30 * 24 * 60 * 60)
+        
         result = pipe.execute()
         
         print(f"[save_sessions_count] 保存结果: {result}, session_id: {session_id}")
@@ -320,31 +325,47 @@ class RedisCountStore:
             list 或 False: 如果找到返回 [count, messages]，否则返回 False
         """
         try:
-            search_pattern = 'session:count:*'
+            # 使用索引键快速查找
+            index_key = f'session:count:index:{end_user_id}'
             
-            for key in self.r.keys(search_pattern):
-                data = self.r.hgetall(key)
-                
-                if not data:
-                    continue
-                
-                if data.get('end_user_id') == end_user_id:
-                    count = data.get('count')
-                    messages_str = data.get('messages')
-                    
-                    if count is not None:
-                        messages = deserialize_messages(messages_str)
-                        return [int(count), messages]
+            # 检查索引键类型，避免 WRONGTYPE 错误
+            try:
+                key_type = self.r.type(index_key)
+                if key_type != 'string' and key_type != 'none':
+                    self.r.delete(index_key)
+                    return False
+            except Exception as type_error:
+                print(f"[get_sessions_count] 检查键类型失败: {type_error}")
+            
+            session_id = self.r.get(index_key)
+            
+            if not session_id:
+                return False
+            
+            # 直接获取数据
+            key = generate_session_key(session_id, key_type="count")
+            data = self.r.hgetall(key)
+            
+            if not data:
+                # 索引存在但数据不存在，清理索引
+                self.r.delete(index_key)
+                return False
+            
+            count = data.get('count')
+            messages_str = data.get('messages')
+            
+            if count is not None:
+                messages = deserialize_messages(messages_str)
+                return [int(count), messages]
             
             return False
         except Exception as e:
             print(f"[get_sessions_count] 查询失败: {e}")
             return False
-
     def update_sessions_count(self, end_user_id: str, new_count: int, 
                              messages: Any) -> bool:
         """
-        通过 end_user_id 修改访问次数统计
+        通过 end_user_id 修改访问次数统计（优化版：使用索引）
         
         Args:
             end_user_id: 终端用户ID
@@ -355,23 +376,39 @@ class RedisCountStore:
             bool: 更新成功返回 True，未找到记录返回 False
         """
         try:
+            # 使用索引键快速查找
+            index_key = f'session:count:index:{end_user_id}'
+            
+            # 检查索引键类型，避免 WRONGTYPE 错误
+            try:
+                key_type = self.r.type(index_key)
+                if key_type != 'string' and key_type != 'none':
+                    # 索引键类型错误，删除并返回 False
+                    print(f"[update_sessions_count] 索引键类型错误: {key_type}，删除索引")
+                    self.r.delete(index_key)
+                    print(f"[update_sessions_count] 未找到记录: end_user_id={end_user_id}")
+                    return False
+            except Exception as type_error:
+                print(f"[update_sessions_count] 检查键类型失败: {type_error}")
+            
+            session_id = self.r.get(index_key)
+            
+            if not session_id:
+                print(f"[update_sessions_count] 未找到记录: end_user_id={end_user_id}")
+                return False
+            
+            # 直接更新数据
+            key = generate_session_key(session_id, key_type="count")
             messages_str = serialize_messages(messages)
-            search_pattern = 'session:count:*'
             
-            for key in self.r.keys(search_pattern):
-                data = self.r.hgetall(key)
-                
-                if not data:
-                    continue
-                
-                if data.get('end_user_id') == end_user_id:
-                    self.r.hset(key, 'count', int(new_count))
-                    self.r.hset(key, 'messages', messages_str)
-                    print(f"[update_sessions_count] 更新成功: end_user_id={end_user_id}, new_count={new_count}, key={key}")
-                    return True
+            pipe = self.r.pipeline()
+            pipe.hset(key, 'count', int(new_count))
+            pipe.hset(key, 'messages', messages_str)
+            result = pipe.execute()
+            
+            print(f"[update_sessions_count] 更新成功: end_user_id={end_user_id}, new_count={new_count}, key={key}")
+            return True
             
-            print(f"[update_sessions_count] 未找到记录: end_user_id={end_user_id}")
-            return False
         except Exception as e:
             print(f"[update_sessions_count] 更新失败: {e}")
             return False

api/app/core/memory/agent/utils/write_tools.py

@@ -4,6 +4,7 @@ Write Tools for Memory Knowledge Extraction Pipeline
 This module provides the main write function for executing the knowledge extraction
 pipeline. Only MemoryConfig is needed - clients are constructed internally.
 """
+import asyncio
 import time
 from datetime import datetime
 
@@ -124,23 +125,48 @@ async def write(
     except Exception as e:
         logger.error(f"Error creating indexes: {e}", exc_info=True)
 
+    # 添加死锁重试机制
+    max_retries = 3
+    retry_delay = 1  # 秒
+
+    for attempt in range(max_retries):
+        try:
+            success = await save_dialog_and_statements_to_neo4j(
+                dialogue_nodes=all_dialogue_nodes,
+                chunk_nodes=all_chunk_nodes,
+                statement_nodes=all_statement_nodes,
+                entity_nodes=all_entity_nodes,
+                statement_chunk_edges=all_statement_chunk_edges,
+                statement_entity_edges=all_statement_entity_edges,
+                entity_edges=all_entity_entity_edges,
+                connector=neo4j_connector
+            )
+            if success:
+                logger.info("Successfully saved all data to Neo4j")
+                break
+            else:
+                logger.warning("Failed to save some data to Neo4j")
+                if attempt < max_retries - 1:
+                    logger.info(f"Retrying... (attempt {attempt + 2}/{max_retries})")
+                    await asyncio.sleep(retry_delay * (attempt + 1))  # 指数退避
+        except Exception as e:
+            error_msg = str(e)
+            # 检查是否是死锁错误
+            if "DeadlockDetected" in error_msg or "deadlock" in error_msg.lower():
+                if attempt < max_retries - 1:
+                    logger.warning(f"Deadlock detected, retrying... (attempt {attempt + 2}/{max_retries})")
+                    await asyncio.sleep(retry_delay * (attempt + 1))  # 指数退避
+                else:
+                    logger.error(f"Failed after {max_retries} attempts due to deadlock: {e}")
+                    raise
+            else:
+                # 非死锁错误，直接抛出
+                raise
+
     try:
-        success = await save_dialog_and_statements_to_neo4j(
-            dialogue_nodes=all_dialogue_nodes,
-            chunk_nodes=all_chunk_nodes,
-            statement_nodes=all_statement_nodes,
-            entity_nodes=all_entity_nodes,
-            statement_chunk_edges=all_statement_chunk_edges,
-            statement_entity_edges=all_statement_entity_edges,
-            entity_edges=all_entity_entity_edges,
-            connector=neo4j_connector
-        )
-        if success:
-            logger.info("Successfully saved all data to Neo4j")
-        else:
-            logger.warning("Failed to save some data to Neo4j")
-    finally:
         await neo4j_connector.close()
+    except Exception as e:
+        logger.error(f"Error closing Neo4j connector: {e}")
 
     log_time("Neo4j Database Save", time.time() - step_start, log_file)
 

api/app/core/memory/models/graph_models.py

@@ -413,7 +413,8 @@ class ExtractedEntityNode(Node):
         description="Entity aliases - alternative names for this entity"
     )
     name_embedding: Optional[List[float]] = Field(default_factory=list, description="Name embedding vector")
-    fact_summary: str = Field(default="", description="Summary of the fact about this entity")
+    # TODO: fact_summary 功能暂时禁用，待后续开发完善后启用
+    # fact_summary: str = Field(default="", description="Summary of the fact about this entity")
     connect_strength: str = Field(..., description="Strong VS Weak about this entity")
     config_id: Optional[int | str] = Field(None, description="Configuration ID used to process this entity (integer or string)")
     

api/app/core/memory/storage_services/extraction_engine/deduplication/deduped_and_disamb.py

@@ -134,42 +134,45 @@ def _merge_attribute(canonical: ExtractedEntityNode, ent: ExtractedEntityNode):
         if len(desc_b) > len(desc_a):
             canonical.description = desc_b
         # 合并事实摘要：统一保留一个“实体: name”行，来源行去重保序
-        fact_a = getattr(canonical, "fact_summary", "") or ""
-        fact_b = getattr(ent, "fact_summary", "") or ""
-        def _extract_sources(txt: str) -> List[str]:
-            sources: List[str] = []
-            if not txt:
-                return sources
-            for line in str(txt).splitlines():
-                ln = line.strip()
+        # TODO: fact_summary 功能暂时禁用，待后续开发完善后启用
+        # fact_a = getattr(canonical, "fact_summary", "") or ""
+        # fact_b = getattr(ent, "fact_summary", "") or ""
+        # def _extract_sources(txt: str) -> List[str]:
+            # sources: List[str] = []
+            # if not txt:
+                # return sources
+            # for line in str(txt).splitlines():
+                # ln = line.strip()
                 # 支持“来源:”或“来源：”前缀
-                m = re.match(r"^来源[:：]\s*(.+)$", ln)
-                if m:
-                    content = m.group(1).strip()
-                    if content:
-                        sources.append(content)
+                # m = re.match(r"^来源[:：]\s*(.+)$", ln)
+                # if m:
+                    # content = m.group(1).strip()
+                    # if content:
+                        # sources.append(content)
             # 如果不存在“来源”前缀，则将整体文本视为一个来源片段，避免信息丢失
-            if not sources and txt.strip():
-                sources.append(txt.strip())
-            return sources
+            # if not sources and txt.strip():
+                # sources.append(txt.strip())
+            # return sources
         try:
-            src_a = _extract_sources(fact_a)
-            src_b = _extract_sources(fact_b)
-            seen = set()
-            merged_sources: List[str] = []
-            for s in src_a + src_b:
-                if s and s not in seen:
-                    seen.add(s)
-                    merged_sources.append(s)
-            if merged_sources:
-                name_line = f"实体: {getattr(canonical, 'name', '')}".strip()
-                canonical.fact_summary = "\n".join([name_line] + [f"来源: {s}" for s in merged_sources])
-            elif fact_b and not fact_a:
-                canonical.fact_summary = fact_b
+            #     src_a = _extract_sources(fact_a)
+            #     src_b = _extract_sources(fact_b)
+            #     seen = set()
+            #     merged_sources: List[str] = []
+            #     for s in src_a + src_b:
+            #         if s and s not in seen:
+            #             seen.add(s)
+            #             merged_sources.append(s)
+            #     if merged_sources:
+            #         name_line = f"实体: {getattr(canonical, 'name', '')}".strip()
+            #         canonical.fact_summary = "\n".join([name_line] + [f"来源: {s}" for s in merged_sources])
+            #     elif fact_b and not fact_a:
+            #         canonical.fact_summary = fact_b
+            pass
         except Exception:
             # 兜底：若解析失败，保留较长文本
-            if len(fact_b) > len(fact_a):
-                canonical.fact_summary = fact_b
+            # if len(fact_b) > len(fact_a):
+            #     canonical.fact_summary = fact_b
+            pass
     except Exception:
         pass
 

api/app/core/memory/storage_services/extraction_engine/deduplication/entity_dedup_llm.py

@@ -145,10 +145,13 @@ def _choose_canonical(a: ExtractedEntityNode, b: ExtractedEntityNode) -> int: #
      # 2. 第二优先级：按“描述+事实摘要”的总长度排序（内容越长，信息越完整）
     desc_a = (getattr(a, "description", "") or "")
     desc_b = (getattr(b, "description", "") or "")
-    fact_a = (getattr(a, "fact_summary", "") or "")
-    fact_b = (getattr(b, "fact_summary", "") or "")
-    score_a = len(desc_a) + len(fact_a)
-    score_b = len(desc_b) + len(fact_b)
+    # TODO: fact_summary 功能暂时禁用，待后续开发完善后启用
+    # fact_a = (getattr(a, "fact_summary", "") or "")
+    # fact_b = (getattr(b, "fact_summary", "") or "")
+    # score_a = len(desc_a) + len(fact_a)
+    # score_b = len(desc_b) + len(fact_b)
+    score_a = len(desc_a)
+    score_b = len(desc_b)
     if score_a != score_b:
         return 0 if score_a >= score_b else 1
     return 0
@@ -189,7 +192,8 @@ async def _judge_pair(
         "entity_type": getattr(a, "entity_type", None),
         "description": getattr(a, "description", None),
         "aliases": getattr(a, "aliases", None) or [],
-        "fact_summary": getattr(a, "fact_summary", None),
+        # TODO: fact_summary 功能暂时禁用，待后续开发完善后启用
+        # "fact_summary": getattr(a, "fact_summary", None),
         "connect_strength": getattr(a, "connect_strength", None),
     }
     entity_b = {
@@ -197,7 +201,8 @@ async def _judge_pair(
         "entity_type": getattr(b, "entity_type", None),
         "description": getattr(b, "description", None),
         "aliases": getattr(b, "aliases", None) or [],
-        "fact_summary": getattr(b, "fact_summary", None),
+        # TODO: fact_summary 功能暂时禁用，待后续开发完善后启用
+        # "fact_summary": getattr(b, "fact_summary", None),
         "connect_strength": getattr(b, "connect_strength", None),
     }
  # 5. 渲染LLM提示词（用工具函数填充模板，包含实体信息、上下文、输出格式）
@@ -248,7 +253,8 @@ async def _judge_pair_disamb(
         "entity_type": getattr(a, "entity_type", None),
         "description": getattr(a, "description", None),
         "aliases": getattr(a, "aliases", None) or [],
-        "fact_summary": getattr(a, "fact_summary", None),
+        # TODO: fact_summary 功能暂时禁用，待后续开发完善后启用
+        # "fact_summary": getattr(a, "fact_summary", None),
         "connect_strength": getattr(a, "connect_strength", None),
     }
     entity_b = {
@@ -256,7 +262,8 @@ async def _judge_pair_disamb(
         "entity_type": getattr(b, "entity_type", None),
         "description": getattr(b, "description", None),
         "aliases": getattr(b, "aliases", None) or [],
-        "fact_summary": getattr(b, "fact_summary", None),
+        # TODO: fact_summary 功能暂时禁用，待后续开发完善后启用
+        # "fact_summary": getattr(b, "fact_summary", None),
         "connect_strength": getattr(b, "connect_strength", None),
     }
     prompt = render_entity_dedup_prompt(

api/app/core/memory/storage_services/extraction_engine/deduplication/second_layer_dedup.py

@@ -72,7 +72,8 @@ def _row_to_entity(row: Dict[str, Any]) -> ExtractedEntityNode:
         description=row.get("description") or "",
         aliases=row.get("aliases") or [],
         name_embedding=row.get("name_embedding") or [],
-        fact_summary=row.get("fact_summary") or "",
+        # TODO: fact_summary 功能暂时禁用，待后续开发完善后启用
+        # fact_summary=row.get("fact_summary") or "",
         connect_strength=row.get("connect_strength") or "",
     )
 

api/app/core/memory/storage_services/extraction_engine/extraction_orchestrator.py

@@ -1088,7 +1088,8 @@ class ExtractionOrchestrator:
                                     entity_type=getattr(entity, 'type', 'unknown'),  # 使用 type 而不是 entity_type
                                     description=getattr(entity, 'description', ''),  # 添加必需的 description 字段
                                     example=getattr(entity, 'example', ''),  # 新增：传递示例字段
-                                    fact_summary=getattr(entity, 'fact_summary', ''),  # 添加必需的 fact_summary 字段
+                                    # TODO: fact_summary 功能暂时禁用，待后续开发完善后启用
+                                    # fact_summary=getattr(entity, 'fact_summary', ''),  # 添加必需的 fact_summary 字段
                                     connect_strength=entity_connect_strength if entity_connect_strength is not None else 'Strong',  # 添加必需的 connect_strength 字段
                                     aliases=getattr(entity, 'aliases', []) or [],  # 传递从三元组提取阶段获取的aliases
                                     name_embedding=getattr(entity, 'name_embedding', None),

api/app/core/memory/utils/alias_utils.py

@@ -296,7 +296,9 @@ def resolve_alias_cycles(entities: List[Any], cycles: Dict[str, Set[str]]) -> Li
             key=lambda eid: (
                 _strength_rank(eid),
                 len(getattr(entity_by_id.get(eid), 'description', '') or ''),
-                len(getattr(entity_by_id.get(eid), 'fact_summary', '') or '')
+                # TODO: fact_summary 功能暂时禁用，待后续开发完善后启用
+                # len(getattr(entity_by_id.get(eid), 'fact_summary', '') or '')
+                0  # 临时占位
             ),
             reverse=True
         )

api/app/core/memory/utils/prompt/prompts/entity_dedup.jinja2

@@ -9,7 +9,8 @@
 - 类型: "{{ entity_a.entity_type | default('') }}"
 - 描述: "{{ entity_a.description | default('') }}"
 - 别名: {{ entity_a.aliases | default([]) }}
-- 摘要: "{{ entity_a.fact_summary | default('') }}"
+{# TODO: fact_summary 功能暂时禁用，待后续开发完善后启用 #}
+{# - 摘要: "{{ entity_a.fact_summary | default('') }}" #}
 - 连接强弱: "{{ entity_a.connect_strength | default('') }}"
 
 实体B:
@@ -17,7 +18,8 @@
 - 类型: "{{ entity_b.entity_type | default('') }}"
 - 描述: "{{ entity_b.description | default('') }}"
 - 别名: {{ entity_b.aliases | default([]) }}
-- 摘要: "{{ entity_b.fact_summary | default('') }}"
+{# TODO: fact_summary 功能暂时禁用，待后续开发完善后启用 #}
+{# - 摘要: "{{ entity_b.fact_summary | default('') }}" #}
 - 连接强弱: "{{ entity_b.connect_strength | default('') }}"
 
 上下文：