fix: Fix the issue where scheduled tasks are not using Redis problem (#1267)

* fix: add filter of nolike && MOS_SCHEDULER_THREAD_POOL_MAX_WORKERS

* fix: create_config null

* fix: add use_redis_queue config

* fix: add related_id

Author: wustzdy

src/memos/api/config.py

@@ -854,7 +854,7 @@ def get_scheduler_config() -> dict[str, Any]:
                 ),
                 "context_window_size": int(os.getenv("MOS_SCHEDULER_CONTEXT_WINDOW_SIZE", "5")),
                 "thread_pool_max_workers": int(
-                    os.getenv("MOS_SCHEDULER_THREAD_POOL_MAX_WORKERS", "10000")
+                    os.getenv("MOS_SCHEDULER_THREAD_POOL_MAX_WORKERS", "200")
                 ),
                 "consume_interval_seconds": float(
                     os.getenv("MOS_SCHEDULER_CONSUME_INTERVAL_SECONDS", "0.01")
@@ -867,6 +867,8 @@ def get_scheduler_config() -> dict[str, Any]:
                     "MOS_SCHEDULER_ENABLE_ACTIVATION_MEMORY", "false"
                 ).lower()
                 == "true",
+                "use_redis_queue": os.getenv("MEMSCHEDULER_USE_REDIS_QUEUE", "False").lower()
+                == "true",
             },
         }
 

src/memos/api/handlers/component_init.py

@@ -255,7 +255,8 @@ def init_server() -> dict[str, Any]:
     # Initialize Scheduler
     scheduler_config_dict = APIConfig.get_scheduler_config()
     scheduler_config = SchedulerConfigFactory(
-        backend="optimized_scheduler", config=scheduler_config_dict
+        backend=scheduler_config_dict["backend"],
+        config=scheduler_config_dict["config"],
     )
     mem_scheduler: OptimizedScheduler = SchedulerFactory.from_config(scheduler_config)
     mem_scheduler.initialize_modules(

src/memos/configs/mem_scheduler.py

@@ -155,7 +155,10 @@ def validate_backend(cls, backend: str) -> str:
     @model_validator(mode="after")
     def create_config(self) -> "SchedulerConfigFactory":
         config_class = self.backend_to_class[self.backend]
-        self.config = config_class(**self.config)
+        raw = self.config
+        if isinstance(raw, dict) and "config" in raw and "use_redis_queue" not in raw:
+            raw = raw["config"]
+        self.config = config_class(**raw)
         return self
 
 

src/memos/graph_dbs/polardb.py

@@ -4667,6 +4667,36 @@ def build_filter_condition(condition_dict: dict) -> str:
                                         condition_parts.append(
                                             f"ag_catalog.agtype_access_operator(properties, '\"{key}\"'::agtype)::text LIKE '%{op_value}%'"
                                         )
+                            elif op == "nolike":
+                                if key.startswith("info."):
+                                    info_field = key[5:]
+                                    if isinstance(op_value, str):
+                                        escaped_value = (
+                                            escape_sql_string(op_value)
+                                            .replace("%", "\\%")
+                                            .replace("_", "\\_")
+                                        )
+                                        condition_parts.append(
+                                            f"ag_catalog.agtype_access_operator(VARIADIC ARRAY[properties, '\"info\"'::ag_catalog.agtype, '\"{info_field}\"'::ag_catalog.agtype])::text NOT LIKE '%{escaped_value}%'"
+                                        )
+                                    else:
+                                        condition_parts.append(
+                                            f"ag_catalog.agtype_access_operator(VARIADIC ARRAY[properties, '\"info\"'::ag_catalog.agtype, '\"{info_field}\"'::ag_catalog.agtype])::text NOT LIKE '%{op_value}%'"
+                                        )
+                                else:
+                                    if isinstance(op_value, str):
+                                        escaped_value = (
+                                            escape_sql_string(op_value)
+                                            .replace("%", "\\%")
+                                            .replace("_", "\\_")
+                                        )
+                                        condition_parts.append(
+                                            f"ag_catalog.agtype_access_operator(properties, '\"{key}\"'::agtype)::text NOT LIKE '%{escaped_value}%'"
+                                        )
+                                    else:
+                                        condition_parts.append(
+                                            f"ag_catalog.agtype_access_operator(properties, '\"{key}\"'::agtype)::text NOT LIKE '%{op_value}%'"
+                                        )
                     # Check if key starts with "info." prefix (for simple equality)
                     elif key.startswith("info."):
                         # Extract the field name after "info."
@@ -4756,6 +4786,7 @@ def parse_filter(
             "project_id",
             "manager_user_id",
             "delete_time",
+            "related_id",
         }
 
         def process_condition(condition):