Merge branch 'refs/heads/develop' into fix/memory_bug_fix

# Conflicts:
#	api/app/services/memory_agent_service.py

api/app/controllers/memory_agent_controller.py

@@ -261,9 +261,7 @@ async def read_server(
     """
     config_id = user_input.config_id
     workspace_id = current_user.current_workspace_id
-    api_logger.info(f"Read service: workspace_id={workspace_id}, config_id={config_id}")
 
-    # 获取 storage_type，如果为 None 则使用默认值
     storage_type = workspace_service.get_workspace_storage_type(
         db=db,
         workspace_id=workspace_id,
@@ -300,12 +298,15 @@ async def read_server(
 
             # 调用 memory_agent_service 的方法生成最终答案
             result['answer'] = await memory_agent_service.generate_summary_from_retrieve(
+                group_id=user_input.group_id,
                 retrieve_info=retrieve_info,
                 history=history,
                 query=query,
                 config_id=config_id,
                 db=db
             )
+            if "信息不足，无法回答" in result['answer']:
+                result['answer']=retrieve_info
         return success(data=result, msg="回复对话消息成功")
     except BaseException as e:
         # Handle ExceptionGroup from TaskGroup (Python 3.11+) or BaseExceptionGroup

api/app/controllers/memory_dashboard_controller.py

@@ -49,63 +49,135 @@ async def get_workspace_end_users(
     current_user: User = Depends(get_current_user),
 ):
     """
-    获取工作空间的宿主列表
+    获取工作空间的宿主列表（高性能优化版本 v2）
     
-    返回格式与原 memory_list 接口中的 end_users 字段相同，
-    并包含每个用户的记忆配置信息（memory_config_id 和 memory_config_name）
+    优化策略：
+    1. 批量查询 end_users（一次查询而非循环）
+    2. 并发查询所有用户的记忆数量（Neo4j）
+    3. RAG 模式使用批量查询（一次 SQL）
+    4. 只返回必要字段减少数据传输
+    5. 添加短期缓存减少重复查询
+    6. 并发执行配置查询和记忆数量查询
+    
+    返回格式：
+    {
+        "end_user": {"id": "uuid", "other_name": "名称"},
+        "memory_num": {"total": 数量},
+        "memory_config": {"memory_config_id": "id", "memory_config_name": "名称"}
+    }
     """
+    import asyncio
+    import json
+    from app.aioRedis import aio_redis_get, aio_redis_set
+    
     workspace_id = current_user.current_workspace_id
+    
+    # 尝试从缓存获取（30秒缓存）
+    cache_key = f"end_users:workspace:{workspace_id}"
+    try:
+        cached_data = await aio_redis_get(cache_key)
+        if cached_data:
+            api_logger.info(f"从缓存获取宿主列表: workspace_id={workspace_id}")
+            return success(data=json.loads(cached_data), msg="宿主列表获取成功")
+    except Exception as e:
+        api_logger.warning(f"Redis 缓存读取失败: {str(e)}")
+    
     # 获取当前空间类型
     current_workspace_type = memory_dashboard_service.get_current_workspace_type(db, workspace_id, current_user)
     api_logger.info(f"用户 {current_user.username} 请求获取工作空间 {workspace_id} 的宿主列表")
+    
+    # 获取 end_users（已优化为批量查询）
     end_users = memory_dashboard_service.get_workspace_end_users(
         db=db,
         workspace_id=workspace_id,
         current_user=current_user
     )
     
-    # 批量获取所有用户的记忆配置信息（优化：一次查询而非 N 次）
-    end_user_ids = [str(user.id) for user in end_users]
-    memory_configs_map = {}
-    if end_user_ids:
+    if not end_users:
+        api_logger.info("工作空间下没有宿主")
+        # 缓存空结果，避免重复查询
         try:
-            memory_configs_map = get_end_users_connected_configs_batch(end_user_ids, db)
+            await aio_redis_set(cache_key, json.dumps([]), expire=30)
+        except Exception as e:
+            api_logger.warning(f"Redis 缓存写入失败: {str(e)}")
+        return success(data=[], msg="宿主列表获取成功")
+    
+    end_user_ids = [str(user.id) for user in end_users]
+    
+    # 并发执行两个独立的查询任务
+    async def get_memory_configs():
+        """获取记忆配置（在线程池中执行同步查询）"""
+        try:
+            return await asyncio.to_thread(
+                get_end_users_connected_configs_batch,
+                end_user_ids, db
+            )
         except Exception as e:
             api_logger.error(f"批量获取记忆配置失败: {str(e)}")
-            # 失败时使用空字典，不影响其他数据返回
+            return {}
     
+    async def get_memory_nums():
+        """获取记忆数量"""
+        if current_workspace_type == "rag":
+            # RAG 模式：批量查询
+            try:
+                chunk_map = await asyncio.to_thread(
+                    memory_dashboard_service.get_users_total_chunk_batch,
+                    end_user_ids, db, current_user
+                )
+                return {uid: {"total": count} for uid, count in chunk_map.items()}
+            except Exception as e:
+                api_logger.error(f"批量获取 RAG chunk 数量失败: {str(e)}")
+                return {uid: {"total": 0} for uid in end_user_ids}
+        
+        elif current_workspace_type == "neo4j":
+            # Neo4j 模式：并发查询（带并发限制）
+            # 使用信号量限制并发数，避免大量用户时压垮 Neo4j
+            MAX_CONCURRENT_QUERIES = 10
+            semaphore = asyncio.Semaphore(MAX_CONCURRENT_QUERIES)
+            
+            async def get_neo4j_memory_num(end_user_id: str):
+                async with semaphore:
+                    try:
+                        return await memory_storage_service.search_all(end_user_id)
+                    except Exception as e:
+                        api_logger.error(f"获取用户 {end_user_id} Neo4j 记忆数量失败: {str(e)}")
+                        return {"total": 0}
+            
+            memory_nums_list = await asyncio.gather(*[get_neo4j_memory_num(uid) for uid in end_user_ids])
+            return {end_user_ids[i]: memory_nums_list[i] for i in range(len(end_user_ids))}
+        
+        return {uid: {"total": 0} for uid in end_user_ids}
+    
+    # 并发执行配置查询和记忆数量查询
+    memory_configs_map, memory_nums_map = await asyncio.gather(
+        get_memory_configs(),
+        get_memory_nums()
+    )
+    
+    # 构建结果（优化：使用列表推导式）
     result = []
     for end_user in end_users:
-        memory_num = {}
-        if current_workspace_type == "neo4j":
-            # EndUser 是 Pydantic 模型，直接访问属性而不是使用 .get()
-            memory_num = await memory_storage_service.search_all(str(end_user.id))
-        elif current_workspace_type == "rag":
-            memory_num = {
-                "total":memory_dashboard_service.get_current_user_total_chunk(str(end_user.id), db, current_user)
-            }
-        
-        # 从批量查询结果中获取配置信息
         user_id = str(end_user.id)
-        memory_config_info = memory_configs_map.get(user_id, {
-            "memory_config_id": None,
-            "memory_config_name": None
-        })
-        
-        # 只保留需要的字段，移除 error 字段（如果有）
-        memory_config = {
-            "memory_config_id": memory_config_info.get("memory_config_id"),
-            "memory_config_name": memory_config_info.get("memory_config_name")
-        }
-        
-        result.append(
-            {
-                'end_user': end_user,
-                'memory_num': memory_num,
-                'memory_config': memory_config
+        config_info = memory_configs_map.get(user_id, {})
+        result.append({
+            'end_user': {
+                'id': user_id,
+                'other_name': end_user.other_name
+            },
+            'memory_num': memory_nums_map.get(user_id, {"total": 0}),
+            'memory_config': {
+                "memory_config_id": config_info.get("memory_config_id"),
+                "memory_config_name": config_info.get("memory_config_name")
             }
-        )
-        
+        })
+    
+    # 写入缓存（30秒过期）
+    try:
+        await aio_redis_set(cache_key, json.dumps(result), expire=30)
+    except Exception as e:
+        api_logger.warning(f"Redis 缓存写入失败: {str(e)}")
+    
     api_logger.info(f"成功获取 {len(end_users)} 个宿主记录")
     return success(data=result, msg="宿主列表获取成功")
 

api/app/controllers/memory_storage_controller.py

@@ -421,15 +421,95 @@ async def get_hot_memory_tags_api(
     db: Session = Depends(get_db),
     current_user: User = Depends(get_current_user),
     ) -> dict:
-    api_logger.info(f"Hot memory tags requested for current_user: {current_user.id}")
+    """
+    获取热门记忆标签（带Redis缓存）
+    
+    缓存策略：
+    - 缓存键：workspace_id + limit
+    - 过期时间：5分钟（300秒）
+    - 缓存命中：~50ms
+    - 缓存未命中：~600-800ms（取决于LLM速度）
+    """
+    workspace_id = current_user.current_workspace_id
+    
+    # 构建缓存键
+    cache_key = f"hot_memory_tags:{workspace_id}:{limit}"
+    
+    api_logger.info(f"Hot memory tags requested for workspace: {workspace_id}, limit: {limit}")
+    
     try:
+        # 尝试从Redis缓存获取
+        from app.aioRedis import aio_redis_get, aio_redis_set
+        import json
+        
+        cached_result = await aio_redis_get(cache_key)
+        if cached_result:
+            api_logger.info(f"Cache hit for key: {cache_key}")
+            try:
+                data = json.loads(cached_result)
+                return success(data=data, msg="查询成功（缓存）")
+            except json.JSONDecodeError:
+                api_logger.warning(f"Failed to parse cached data, will refresh")
+        
+        # 缓存未命中，执行查询
+        api_logger.info(f"Cache miss for key: {cache_key}, executing query")
         result = await analytics_hot_memory_tags(db, current_user, limit)
+        
+        # 写入缓存（过期时间：5分钟）
+        # 注意：result是列表，需要转换为JSON字符串
+        try:
+            cache_data = json.dumps(result, ensure_ascii=False)
+            await aio_redis_set(cache_key, cache_data, expire=300)
+            api_logger.info(f"Cached result for key: {cache_key}")
+        except Exception as cache_error:
+            # 缓存写入失败不影响主流程
+            api_logger.warning(f"Failed to cache result: {str(cache_error)}")
+        
         return success(data=result, msg="查询成功")
+        
     except Exception as e:
         api_logger.error(f"Hot memory tags failed: {str(e)}")
         return fail(BizCode.INTERNAL_ERROR, "热门标签查询失败", str(e))
 
 
+@router.delete("/analytics/hot_memory_tags/cache", response_model=ApiResponse)
+async def clear_hot_memory_tags_cache(
+    current_user: User = Depends(get_current_user),
+    ) -> dict:
+    """
+    清除热门标签缓存
+    
+    用于：
+    - 手动刷新数据
+    - 调试和测试
+    - 数据更新后立即生效
+    """
+    workspace_id = current_user.current_workspace_id
+    
+    api_logger.info(f"Clear hot memory tags cache requested for workspace: {workspace_id}")
+    
+    try:
+        from app.aioRedis import aio_redis_delete
+        
+        # 清除所有limit的缓存（常见的limit值）
+        cleared_count = 0
+        for limit in [5, 10, 15, 20, 30, 50]:
+            cache_key = f"hot_memory_tags:{workspace_id}:{limit}"
+            result = await aio_redis_delete(cache_key)
+            if result:
+                cleared_count += 1
+                api_logger.info(f"Cleared cache for key: {cache_key}")
+        
+        return success(
+            data={"cleared_count": cleared_count}, 
+            msg=f"成功清除 {cleared_count} 个缓存"
+        )
+        
+    except Exception as e:
+        api_logger.error(f"Clear cache failed: {str(e)}")
+        return fail(BizCode.INTERNAL_ERROR, "清除缓存失败", str(e))
+
+
 @router.get("/analytics/recent_activity_stats", response_model=ApiResponse)
 async def get_recent_activity_stats_api(
     current_user: User = Depends(get_current_user),

api/app/controllers/public_share_controller.py

@@ -317,9 +317,12 @@ async def chat(
         appid = share.app_id
         """获取存储类型和工作空间的ID"""
 
-        # 直接通过 SQLAlchemy 查询 app
+        # 直接通过 SQLAlchemy 查询 app（仅查询未删除的应用）
         from app.models.app_model import App
-        app = db.query(App).filter(App.id == appid).first()
+        app = db.query(App).filter(
+            App.id == appid,
+            App.is_active.is_(True)
+        ).first()
         if not app:
             raise BusinessException("应用不存在", BizCode.APP_NOT_FOUND)
 

api/app/controllers/workflow_controller.py

@@ -54,7 +54,7 @@ async def create_workflow_config(
         app = db.query(App).filter(
             App.id == app_id,
             App.workspace_id == current_user.current_workspace_id,
-            App.is_active == True
+            App.is_active.is_(True)
         ).first()
 
         if not app:
@@ -214,7 +214,7 @@ async def delete_workflow_config(
         app = db.query(App).filter(
             App.id == app_id,
             App.workspace_id == current_user.current_workspace_id,
-            App.is_active == True
+            App.is_active.is_(True)
         ).first()
 
         if not app:
@@ -259,7 +259,7 @@ async def validate_workflow_config(
         app = db.query(App).filter(
             App.id == app_id,
             App.workspace_id == current_user.current_workspace_id,
-            App.is_active == True
+            App.is_active.is_(True)
         ).first()
 
         if not app:
@@ -329,7 +329,7 @@ async def get_workflow_executions(
         app = db.query(App).filter(
             App.id == app_id,
             App.workspace_id == current_user.current_workspace_id,
-            App.is_active == True
+            App.is_active.is_(True)
         ).first()
 
         if not app:
@@ -389,7 +389,7 @@ async def get_workflow_execution(
         app = db.query(App).filter(
             App.id == execution.app_id,
             App.workspace_id == current_user.current_workspace_id,
-            App.is_active == True
+            App.is_active.is_(True)
         ).first()
 
         if not app:
@@ -440,7 +440,7 @@ async def run_workflow(
         app = db.query(App).filter(
             App.id == app_id,
             App.workspace_id == current_user.current_workspace_id,
-            App.is_active == True
+            App.is_active.is_(True)
         ).first()
 
         if not app:
@@ -578,7 +578,7 @@ async def cancel_workflow_execution(
         app = db.query(App).filter(
             App.id == execution.app_id,
             App.workspace_id == current_user.current_workspace_id,
-            App.is_active == True
+            App.is_active.is_(True)
         ).first()
 
         if not app: