[add] Create a Celery task for checking the existence of the "implicit_emotions" data

api/app/celery_app.py

@@ -113,6 +113,7 @@ celery_app.conf.update(
         'app.tasks.run_forgetting_cycle_task': {'queue': 'periodic_tasks'},
         'app.tasks.write_all_workspaces_memory_task': {'queue': 'periodic_tasks'},
         'app.tasks.update_implicit_emotions_storage': {'queue': 'periodic_tasks'},
+        'app.tasks.init_implicit_emotions_for_users': {'queue': 'ondemand_tasks'},
     },
 )
 

api/app/controllers/memory_dashboard_controller.py

@@ -149,6 +149,17 @@ async def get_workspace_end_users(
         
         return {uid: {"total": 0} for uid in end_user_ids}
     
+    # 触发按需初始化：为 implicit_emotions_storage 中没有记录的用户异步生成数据
+    try:
+        from app.celery_app import celery_app as _celery_app
+        _celery_app.send_task(
+            "app.tasks.init_implicit_emotions_for_users",
+            kwargs={"end_user_ids": end_user_ids},
+        )
+        api_logger.info(f"已触发隐性记忆按需初始化任务，候选用户数: {len(end_user_ids)}")
+    except Exception as e:
+        api_logger.warning(f"触发隐性记忆按需初始化任务失败（不影响主流程）: {e}")
+
     # 并发执行配置查询和记忆数量查询
     memory_configs_map, memory_nums_map = await asyncio.gather(
         get_memory_configs(),

api/app/services/workspace_service.py

@@ -130,6 +130,7 @@ def _create_workspace_only(
         business_logger.error(f"创建工作空间失败: {workspace.name} - {str(e)}")
         raise
 
+
 def create_workspace(
         db: Session, workspace: WorkspaceCreate, user: User, language: str = "zh"
 ) -> Workspace:
@@ -966,6 +967,125 @@ def update_workspace_models_configs(
         raise BusinessException(f"更新模型配置失败: {str(e)}", BizCode.INTERNAL_ERROR)
 
 
+def _fill_workspace_configs_model_defaults(
+    db: Session,
+    workspace: Workspace
+) -> None:
+    """Fill empty model fields for all memory configs in a workspace.
+    
+    Updates llm_id, embedding_id, rerank_id, reflection_model_id, and emotion_model_id
+    if they are None, using the corresponding workspace default models.
+    
+    Args:
+        db: Database session
+        workspace: The workspace containing default model settings
+    """
+    from app.models.memory_config_model import MemoryConfig
+    
+    # Get all configs for this workspace
+    configs = db.query(MemoryConfig).filter(
+        MemoryConfig.workspace_id == workspace.id
+    ).all()
+    
+    if not configs:
+        return
+    
+    # Map of memory_config field -> workspace field
+    model_field_mappings = [
+        ("llm_id", "llm"),
+        ("embedding_id", "embedding"),
+        ("rerank_id", "rerank"),
+        ("reflection_model_id", "llm"),  # reflection uses LLM
+        ("emotion_model_id", "llm"),     # emotion uses LLM
+    ]
+    
+    configs_updated = 0
+    
+    for memory_config in configs:
+        updated_fields = []
+        
+        for config_field, workspace_field in model_field_mappings:
+            config_value = getattr(memory_config, config_field, None)
+            workspace_value = getattr(workspace, workspace_field, None)
+            
+            if not config_value and workspace_value:
+                setattr(memory_config, config_field, workspace_value)
+                updated_fields.append(config_field)
+        
+        if updated_fields:
+            configs_updated += 1
+            business_logger.debug(
+                f"Updated memory config {memory_config.config_id} fields: {updated_fields}"
+            )
+    
+    if configs_updated > 0:
+        try:
+            db.commit()
+            business_logger.info(
+                f"Updated {configs_updated} memory configs in workspace {workspace.id} with default models"
+            )
+        except Exception as e:
+            db.rollback()
+            business_logger.error(
+                f"Failed to update memory configs in workspace {workspace.id}: {str(e)}"
+            )
+
+
+def _create_default_memory_config(
+    db: Session,
+    workspace_id: uuid.UUID,
+    workspace_name: str,
+    llm_id: Optional[uuid.UUID] = None,
+    embedding_id: Optional[uuid.UUID] = None,
+    rerank_id: Optional[uuid.UUID] = None,
+    scene_id: Optional[uuid.UUID] = None,
+    pruning_scene_name: Optional[str] = None,
+) -> None:
+    """Create a default memory config for a newly created workspace.
+    
+    Args:
+        db: Database session
+        workspace_id: The workspace ID
+        workspace_name: The workspace name (used for config naming)
+        llm_id: Optional LLM model ID
+        embedding_id: Optional embedding model ID
+        rerank_id: Optional rerank model ID
+        scene_id: Optional ontology scene ID (默认关联教育场景)
+        pruning_scene_name: Optional pruning scene name，取自 ontology_scene.scene_name
+    """
+    from app.models.memory_config_model import MemoryConfig
+    
+    config_id = uuid.uuid4()
+    
+    default_config = MemoryConfig(
+        config_id=config_id,
+        config_name=f"{workspace_name} 默认配置",
+        config_desc="工作空间创建时自动生成的默认记忆配置",
+        workspace_id=workspace_id,
+        llm_id=str(llm_id) if llm_id else None,
+        embedding_id=str(embedding_id) if embedding_id else None,
+        rerank_id=str(rerank_id) if rerank_id else None,
+        scene_id=scene_id,  # 关联本体场景ID（默认为"在线教育"场景）
+        pruning_scene=pruning_scene_name,  # 语义剪枝场景直接使用 scene_name
+        state=True,  # Active by default
+        is_default=True,  # Mark as workspace default
+    )
+    
+    db.add(default_config)
+    db.flush()  # 使用 flush 而不是 commit，让调用者统一提交
+    
+    business_logger.info(
+        "Created default memory config for workspace",
+        extra={
+            "workspace_id": str(workspace_id),
+            "config_id": str(config_id),
+            "config_name": default_config.config_name,
+            "scene_id": str(scene_id) if scene_id else None,
+        }
+    )
+
+# ==================== 检查配置相关服务 ====================
+
 def _ensure_default_memory_config(db: Session, workspace: Workspace) -> None:
     """Ensure a workspace has a default memory config, creating one if missing.
     
@@ -1045,70 +1165,6 @@ def _ensure_default_memory_config(db: Session, workspace: Workspace) -> None:
     _fill_workspace_configs_model_defaults(db, workspace)
 
 
-def _fill_workspace_configs_model_defaults(
-    db: Session,
-    workspace: Workspace
-) -> None:
-    """Fill empty model fields for all memory configs in a workspace.
-    
-    Updates llm_id, embedding_id, rerank_id, reflection_model_id, and emotion_model_id
-    if they are None, using the corresponding workspace default models.
-    
-    Args:
-        db: Database session
-        workspace: The workspace containing default model settings
-    """
-    from app.models.memory_config_model import MemoryConfig
-    
-    # Get all configs for this workspace
-    configs = db.query(MemoryConfig).filter(
-        MemoryConfig.workspace_id == workspace.id
-    ).all()
-    
-    if not configs:
-        return
-    
-    # Map of memory_config field -> workspace field
-    model_field_mappings = [
-        ("llm_id", "llm"),
-        ("embedding_id", "embedding"),
-        ("rerank_id", "rerank"),
-        ("reflection_model_id", "llm"),  # reflection uses LLM
-        ("emotion_model_id", "llm"),     # emotion uses LLM
-    ]
-    
-    configs_updated = 0
-    
-    for memory_config in configs:
-        updated_fields = []
-        
-        for config_field, workspace_field in model_field_mappings:
-            config_value = getattr(memory_config, config_field, None)
-            workspace_value = getattr(workspace, workspace_field, None)
-            
-            if not config_value and workspace_value:
-                setattr(memory_config, config_field, workspace_value)
-                updated_fields.append(config_field)
-        
-        if updated_fields:
-            configs_updated += 1
-            business_logger.debug(
-                f"Updated memory config {memory_config.config_id} fields: {updated_fields}"
-            )
-    
-    if configs_updated > 0:
-        try:
-            db.commit()
-            business_logger.info(
-                f"Updated {configs_updated} memory configs in workspace {workspace.id} with default models"
-            )
-        except Exception as e:
-            db.rollback()
-            business_logger.error(
-                f"Failed to update memory configs in workspace {workspace.id}: {str(e)}"
-            )
-
-
 def _ensure_default_ontology_scenes(db: Session, workspace: Workspace) -> None:
     """Ensure a workspace has default ontology scenes, creating them if missing.
 
@@ -1154,56 +1210,3 @@ def _ensure_default_ontology_scenes(db: Session, workspace: Workspace) -> None:
             f"为工作空间 {workspace.id} 补建默认本体场景异常: {str(e)}"
         )
 
-
-def _create_default_memory_config(
-    db: Session,
-    workspace_id: uuid.UUID,
-    workspace_name: str,
-    llm_id: Optional[uuid.UUID] = None,
-    embedding_id: Optional[uuid.UUID] = None,
-    rerank_id: Optional[uuid.UUID] = None,
-    scene_id: Optional[uuid.UUID] = None,
-    pruning_scene_name: Optional[str] = None,
-) -> None:
-    """Create a default memory config for a newly created workspace.
-    
-    Args:
-        db: Database session
-        workspace_id: The workspace ID
-        workspace_name: The workspace name (used for config naming)
-        llm_id: Optional LLM model ID
-        embedding_id: Optional embedding model ID
-        rerank_id: Optional rerank model ID
-        scene_id: Optional ontology scene ID (默认关联教育场景)
-        pruning_scene_name: Optional pruning scene name，取自 ontology_scene.scene_name
-    """
-    from app.models.memory_config_model import MemoryConfig
-    
-    config_id = uuid.uuid4()
-    
-    default_config = MemoryConfig(
-        config_id=config_id,
-        config_name=f"{workspace_name} 默认配置",
-        config_desc="工作空间创建时自动生成的默认记忆配置",
-        workspace_id=workspace_id,
-        llm_id=str(llm_id) if llm_id else None,
-        embedding_id=str(embedding_id) if embedding_id else None,
-        rerank_id=str(rerank_id) if rerank_id else None,
-        scene_id=scene_id,  # 关联本体场景ID（默认为"在线教育"场景）
-        pruning_scene=pruning_scene_name,  # 语义剪枝场景直接使用 scene_name
-        state=True,  # Active by default
-        is_default=True,  # Mark as workspace default
-    )
-    
-    db.add(default_config)
-    db.flush()  # 使用 flush 而不是 commit，让调用者统一提交
-    
-    business_logger.info(
-        "Created default memory config for workspace",
-        extra={
-            "workspace_id": str(workspace_id),
-            "config_id": str(config_id),
-            "config_name": default_config.config_name,
-            "scene_id": str(scene_id) if scene_id else None,
-        }
-    )

api/app/tasks.py

@@ -2416,3 +2416,129 @@ def update_implicit_emotions_storage(self) -> Dict[str, Any]:
             "elapsed_time": elapsed_time,
             "task_id": self.request.id
         }
+
+
+# =============================================================================
+
+@celery_app.task(
+    name="app.tasks.init_implicit_emotions_for_users",
+    bind=True,
+    ignore_result=True,
+    max_retries=0,
+    acks_late=False,
+    time_limit=3600,
+    soft_time_limit=3300,
+)
+def init_implicit_emotions_for_users(self, end_user_ids: List[str]) -> Dict[str, Any]:
+    """按需初始化：为指定用户列表中尚未生成隐性记忆/情绪数据的用户执行首次生成。
+
+    由 /dashboard/end_users 接口触发，仅处理 implicit_emotions_storage 表中不存在记录的用户。
+
+    Args:
+        end_user_ids: 需要检查并初始化的用户ID列表
+
+    Returns:
+        包含任务执行结果的字典
+    """
+    start_time = time.time()
+
+    async def _run() -> Dict[str, Any]:
+        from app.core.logging_config import get_logger
+        from app.repositories.implicit_emotions_storage_repository import ImplicitEmotionsStorageRepository
+        from app.services.implicit_memory_service import ImplicitMemoryService
+        from app.services.emotion_analytics_service import EmotionAnalyticsService
+
+        logger = get_logger(__name__)
+        logger.info(f"开始按需初始化隐性记忆/情绪数据，候选用户数: {len(end_user_ids)}")
+
+        initialized = 0
+        failed = 0
+        skipped = 0
+
+        with get_db_context() as db:
+            repo = ImplicitEmotionsStorageRepository(db)
+
+            for end_user_id in end_user_ids:
+                # 幂等检查：已有记录则跳过
+                existing = repo.get_by_end_user_id(end_user_id)
+                if existing is not None:
+                    skipped += 1
+                    continue
+
+                logger.info(f"用户 {end_user_id} 无隐性记忆数据，开始初始化")
+                implicit_ok = False
+                emotion_ok = False
+
+                try:
+                    try:
+                        implicit_service = ImplicitMemoryService(db=db, end_user_id=end_user_id)
+                        profile_data = await implicit_service.generate_complete_profile(user_id=end_user_id)
+                        await implicit_service.save_profile_cache(
+                            end_user_id=end_user_id,
+                            profile_data=profile_data,
+                            db=db
+                        )
+                        implicit_ok = True
+                    except Exception as e:
+                        logger.error(f"用户 {end_user_id} 隐性记忆初始化失败: {e}")
+
+                    try:
+                        emotion_service = EmotionAnalyticsService()
+                        suggestions_data = await emotion_service.generate_emotion_suggestions(
+                            end_user_id=end_user_id,
+                            db=db,
+                            language="zh"
+                        )
+                        await emotion_service.save_suggestions_cache(
+                            end_user_id=end_user_id,
+                            suggestions_data=suggestions_data,
+                            db=db
+                        )
+                        emotion_ok = True
+                    except Exception as e:
+                        logger.error(f"用户 {end_user_id} 情绪建议初始化失败: {e}")
+
+                    if implicit_ok or emotion_ok:
+                        initialized += 1
+                    else:
+                        failed += 1
+
+                except Exception as e:
+                    failed += 1
+                    logger.error(f"用户 {end_user_id} 初始化异常: {e}")
+
+        logger.info(f"按需初始化完成: 初始化={initialized}, 跳过={skipped}, 失败={failed}")
+        return {
+            "status": "SUCCESS",
+            "initialized": initialized,
+            "skipped": skipped,
+            "failed": failed,
+        }
+
+    try:
+        try:
+            import nest_asyncio
+            nest_asyncio.apply()
+        except ImportError:
+            pass
+
+        try:
+            loop = asyncio.get_event_loop()
+            if loop.is_closed():
+                loop = asyncio.new_event_loop()
+                asyncio.set_event_loop(loop)
+        except RuntimeError:
+            loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(loop)
+
+        result = loop.run_until_complete(_run())
+        result["elapsed_time"] = time.time() - start_time
+        result["task_id"] = self.request.id
+        return result
+    except Exception as e:
+        return {
+            "status": "FAILURE",
+            "error": str(e),
+            "elapsed_time": time.time() - start_time,
+            "task_id": self.request.id,
+        }

api/docker-compose.yml

@@ -63,6 +63,20 @@ services:
     networks:
       - celery
 
+  # On-demand worker - API-triggered tasks (e.g. implicit emotions init)
+  worker-ondemand:
+    image: redbear-mem-open:latest
+    container_name: worker-ondemand
+    env_file:
+      - .env
+    volumes:
+      - ./files:/files
+      - /etc/localtime:/etc/localtime:ro
+    command: celery -A app.celery_worker.celery_app worker -E --loglevel=info --pool=prefork --concurrency=4 --queues=ondemand_tasks --max-tasks-per-child=50 -n ondemand_worker@%h
+    restart: unless-stopped
+    networks:
+      - celery
+
   # Celery Beat - scheduler
   beat:
     image: redbear-mem-open:latest