Fix/develop memory bug (#354)

* 遗漏的历史映射 * 遗漏的历史映射 * fix_timeline_memories * fix_timeline_memories * write_gragp/bug_fix * write_gragp/bug_fix * write_gragp/bug_fix * write_gragp/bug_fix * Multiple independent transactions - single transaction * memory_content ->memory_config_id * memory_content ->memory_config_id * memory_content ->memory_config_id * memory_content ->memory_config_id * memory_content ->memory_config_id * memory_content ->memory_config_id * memory_content ->memory_config_id * tasks/bug_fix/long * tasks_reflection/bug/fix * tasks_reflection/bug/fix * tasks_reflection/bug/fix * tasks_reflection/bug/fix * change/get_db_context/way * change/get_db_context/way
2026-02-06 18:45:47 +08:00
parent ac47ab3deb
commit 09afec17f9
4 changed files with 63 additions and 43 deletions
--- a/api/app/controllers/memory_reflection_controller.py
+++ b/api/app/controllers/memory_reflection_controller.py
@@ -104,14 +104,18 @@ async def start_workspace_reflection(
 ) -> dict:
    """启动工作空间中所有匹配应用的反思功能"""
    workspace_id = current_user.current_workspace_id
-    reflection_service = MemoryReflectionService(db)

    try:
        api_logger.info(f"用户 {current_user.username} 启动workspace反思，workspace_id: {workspace_id}")

-        service = WorkspaceAppService(db)
-        result = service.get_workspace_apps_detailed(workspace_id)
+        # 使用独立的数据库会话来获取工作空间应用详情，避免事务失败
+        from app.db import get_db_context
+        with get_db_context() as query_db:
+            service = WorkspaceAppService(query_db)
+            result = service.get_workspace_apps_detailed(workspace_id)
+        
        reflection_results = []
+        
        for data in result['apps_detailed_info']:
            # 跳过没有配置的应用
            if not data['memory_configs']:
@@ -133,33 +137,36 @@ async def start_workspace_reflection(
                    api_logger.debug(f"配置 {config_id_str} 没有匹配的release")
                    continue

-                # 为每个用户执行反思
+                # 为每个用户执行反思 - 使用独立的数据库会话
                for user in end_users:
                    api_logger.info(f"为用户 {user['id']} 启动反思，config_id: {config_id_str}")

-                    try:
-                        reflection_result = await reflection_service.start_text_reflection(
-                            config_data=config,
-                            end_user_id=user['id']
-                        )
+                    # 为每个用户创建独立的数据库会话，避免事务失败影响其他用户
+                    with get_db_context() as user_db:
+                        try:
+                            reflection_service = MemoryReflectionService(user_db)
+                            reflection_result = await reflection_service.start_text_reflection(
+                                config_data=config,
+                                end_user_id=user['id']
+                            )

-                        reflection_results.append({
-                            "app_id": data['id'],
-                            "config_id": config_id_str,
-                            "end_user_id": user['id'],
-                            "reflection_result": reflection_result
-                        })
-                    except Exception as e:
-                        api_logger.error(f"用户 {user['id']} 反思失败: {str(e)}")
-                        reflection_results.append({
-                            "app_id": data['id'],
-                            "config_id": config_id_str,
-                            "end_user_id": user['id'],
-                            "reflection_result": {
-                                "status": "错误",
-                                "message": f"反思失败: {str(e)}"
-                            }
-                        })
+                            reflection_results.append({
+                                "app_id": data['id'],
+                                "config_id": config_id_str,
+                                "end_user_id": user['id'],
+                                "reflection_result": reflection_result
+                            })
+                        except Exception as e:
+                            api_logger.error(f"用户 {user['id']} 反思失败: {str(e)}")
+                            reflection_results.append({
+                                "app_id": data['id'],
+                                "config_id": config_id_str,
+                                "end_user_id": user['id'],
+                                "reflection_result": {
+                                    "status": "错误",
+                                    "message": f"反思失败: {str(e)}"
+                                }
+                            })

        return success(data=reflection_results, msg="反思配置成功")

--- a/api/app/core/memory/storage_services/reflection_engine/self_reflexion.py
+++ b/api/app/core/memory/storage_services/reflection_engine/self_reflexion.py
@@ -462,8 +462,8 @@ class ReflectionEngine:
            List[Any]: 反思数据列表
        """

-
-
+        print("=== 获取反思数据 ===")
+        print(f"  主机ID: {host_id}")
        if self.config.reflexion_range == ReflectionRange.PARTIAL:
            neo4j_query = neo4j_query_part.format(host_id)
            neo4j_statement = neo4j_statement_part.format(host_id)
--- a/api/app/services/memory_agent_service.py
+++ b/api/app/services/memory_agent_service.py
@@ -313,13 +313,16 @@ class MemoryAgentService:
        start_time = time.time()

        # Load configuration from database with workspace fallback
+        # Use a separate database session to avoid transaction failures
        try:
-            config_service = MemoryConfigService(db)
-            memory_config = config_service.load_memory_config(
-                config_id=config_id,
-                workspace_id=workspace_id,
-                service_name="MemoryAgentService"
-            )
+            from app.db import get_db_context
+            with get_db_context() as config_db:
+                config_service = MemoryConfigService(config_db)
+                memory_config = config_service.load_memory_config(
+                    config_id=config_id,
+                    workspace_id=workspace_id,
+                    service_name="MemoryAgentService"
+                )
            logger.info(f"Configuration loaded successfully: {memory_config.config_name}")
        except ConfigurationError as e:
            error_msg = f"Failed to load configuration for config_id: {config_id}: {e}"
@@ -454,12 +457,15 @@ class MemoryAgentService:

        config_load_start = time.time()
        try:
-            config_service = MemoryConfigService(db)
-            memory_config = config_service.load_memory_config(
-                config_id=config_id,
-                workspace_id=workspace_id,
-                service_name="MemoryAgentService"
-            )
+            # Use a separate database session to avoid transaction failures
+            from app.db import get_db_context
+            with get_db_context() as config_db:
+                config_service = MemoryConfigService(config_db)
+                memory_config = config_service.load_memory_config(
+                    config_id=config_id,
+                    workspace_id=workspace_id,
+                    service_name="MemoryAgentService"
+                )
            config_load_time = time.time() - config_load_start
            logger.info(f"[PERF] Configuration loaded in {config_load_time:.4f}s: {memory_config.config_name}")
        except ConfigurationError as e:
--- a/api/app/services/memory_reflection_service.py
+++ b/api/app/services/memory_reflection_service.py
@@ -364,7 +364,8 @@ class MemoryReflectionService:
        reflexion_range_value = config_data.get("reflexion_range")
        if reflexion_range_value is None or reflexion_range_value == "":
            reflexion_range_value = "partial"
-            # Map legacy/invalid values to valid enum values
+        
+        # Map legacy/invalid values to valid enum values
        reflexion_range_mapping = {
            "retrieval": "partial",  # Map old 'retrieval' to 'partial'
            "partial": "partial",
@@ -378,13 +379,19 @@ class MemoryReflectionService:
            baseline_value = "TIME"
        baseline = ReflectionBaseline(baseline_value)

-        # iteration_period =
+        # iteration_period
        iteration_period = config_data.get("iteration_period", 24)
        if isinstance(iteration_period, str):
            try:
                iteration_period = int(iteration_period)
            except (ValueError, TypeError):
                iteration_period = 24  # 默认24小时
+        
+        # 获取 model_id 并转换为字符串（如果是 UUID 对象）
+        reflection_model_id = config_data.get("reflection_model_id", "")
+        if reflection_model_id:
+            reflection_model_id = str(reflection_model_id)
+        
        return ReflectionConfig(
            enabled=config_data.get("enable_self_reflexion", False),
            iteration_period=str(iteration_period),  # ReflectionConfig期望字符串
@@ -392,7 +399,7 @@ class MemoryReflectionService:
            baseline=baseline,
            memory_verify=config_data.get("memory_verify", False),
            quality_assessment=config_data.get("quality_assessment", False),
-            model_id=config_data.get("reflection_model_id", "")
+            model_id=reflection_model_id
        )
    
    async def _execute_reflection_engine(