scheduler feat: implementation of redis queue and new api search functions of mixture and fine mode

examples/mem_scheduler/api_w_scheduler.py

@@ -0,0 +1,62 @@
+from memos.api.routers.server_router import mem_scheduler
+from memos.mem_scheduler.schemas.message_schemas import ScheduleMessageItem
+
+
+# Debug: Print scheduler configuration
+print("=== Scheduler Configuration Debug ===")
+print(f"Scheduler type: {type(mem_scheduler).__name__}")
+print(f"Config: {mem_scheduler.config}")
+print(f"use_redis_queue: {mem_scheduler.use_redis_queue}")
+print(f"Queue type: {type(mem_scheduler.memos_message_queue).__name__}")
+print(f"Queue maxsize: {getattr(mem_scheduler.memos_message_queue, 'maxsize', 'N/A')}")
+
+# Check if Redis queue is connected
+if hasattr(mem_scheduler.memos_message_queue, "_is_connected"):
+    print(f"Redis connected: {mem_scheduler.memos_message_queue._is_connected}")
+if hasattr(mem_scheduler.memos_message_queue, "_redis_conn"):
+    print(f"Redis connection: {mem_scheduler.memos_message_queue._redis_conn}")
+print("=====================================\n")
+
+queue = mem_scheduler.memos_message_queue
+queue.clear()
+
+
+# 1. Define a handler function
+def my_test_handler(messages: list[ScheduleMessageItem]):
+    print(f"My test handler received {len(messages)} messages:")
+    for msg in messages:
+        print(f" my_test_handler - {msg.item_id}: {msg.content}")
+        print(
+            f"{queue._redis_conn.xinfo_groups(queue.stream_name)} qsize: {queue.qsize()} messages:{messages}"
+        )
+
+
+# 2. Register the handler
+TEST_HANDLER_LABEL = "test_handler"
+mem_scheduler.register_handlers({TEST_HANDLER_LABEL: my_test_handler})
+
+# 3. Create messages
+messages_to_send = [
+    ScheduleMessageItem(
+        item_id=f"test_item_{i}",
+        user_id="test_user",
+        mem_cube_id="test_mem_cube",
+        label=TEST_HANDLER_LABEL,
+        content=f"This is test message {i}",
+    )
+    for i in range(5)
+]
+
+# 5. Submit messages
+for mes in messages_to_send:
+    print(f"Submitting message {mes.item_id} to the scheduler...")
+    mem_scheduler.submit_messages([mes])
+
+# 6. Wait for messages to be processed (limited to 100 checks)
+print("Waiting for messages to be consumed (max 100 checks)...")
+mem_scheduler.mem_scheduler_wait()
+
+
+# 7. Stop the scheduler
+print("Stopping the scheduler...")
+mem_scheduler.stop()

examples/mem_scheduler/memos_w_scheduler.py

@@ -70,13 +70,48 @@ def init_task():
     return conversations, questions
 
 
+def show_web_logs(mem_scheduler: GeneralScheduler):
+    """Display all web log entries from the scheduler's log queue.
+
+    Args:
+        mem_scheduler: The scheduler instance containing web logs to display
+    """
+    if mem_scheduler._web_log_message_queue.empty():
+        print("Web log queue is currently empty.")
+        return
+
+    print("\n" + "=" * 50 + " WEB LOGS " + "=" * 50)
+
+    # Create a temporary queue to preserve the original queue contents
+    temp_queue = Queue()
+    log_count = 0
+
+    while not mem_scheduler._web_log_message_queue.empty():
+        log_item: ScheduleLogForWebItem = mem_scheduler._web_log_message_queue.get()
+        temp_queue.put(log_item)
+        log_count += 1
+
+        # Print log entry details
+        print(f"\nLog Entry #{log_count}:")
+        print(f'- "{log_item.label}" log: {log_item}')
+
+        print("-" * 50)
+
+    # Restore items back to the original queue
+    while not temp_queue.empty():
+        mem_scheduler._web_log_message_queue.put(temp_queue.get())
+
+    print(f"\nTotal {log_count} web log entries displayed.")
+    print("=" * 110 + "\n")
+
+
 def run_with_scheduler_init():
     print("==== run_with_automatic_scheduler_init ====")
     conversations, questions = init_task()
 
     # set configs
     mos_config = MOSConfig.from_yaml_file(
-        f"{BASE_DIR}/examples/data/config/mem_scheduler/memos_config_w_scheduler.yaml"
+        f"{BASE_DIR}/examples/data/config/mem_scheduler/memos_config_w_optimized_scheduler.yaml"
     )
 
     mem_cube_config = GeneralMemCubeConfig.from_yaml_file(
@@ -118,6 +153,7 @@ def run_with_scheduler_init():
     )
 
     mos.add(conversations, user_id=user_id, mem_cube_id=mem_cube_id)
+    mos.mem_scheduler.current_mem_cube = mem_cube
 
     for item in questions:
         print("===== Chat Start =====")
@@ -131,40 +167,5 @@ def run_with_scheduler_init():
     mos.mem_scheduler.stop()
 
 
-def show_web_logs(mem_scheduler: GeneralScheduler):
-    """Display all web log entries from the scheduler's log queue.
-
-    Args:
-        mem_scheduler: The scheduler instance containing web logs to display
-    """
-    if mem_scheduler._web_log_message_queue.empty():
-        print("Web log queue is currently empty.")
-        return
-
-    print("\n" + "=" * 50 + " WEB LOGS " + "=" * 50)
-
-    # Create a temporary queue to preserve the original queue contents
-    temp_queue = Queue()
-    log_count = 0
-
-    while not mem_scheduler._web_log_message_queue.empty():
-        log_item: ScheduleLogForWebItem = mem_scheduler._web_log_message_queue.get()
-        temp_queue.put(log_item)
-        log_count += 1
-
-        # Print log entry details
-        print(f"\nLog Entry #{log_count}:")
-        print(f'- "{log_item.label}" log: {log_item}')
-
-        print("-" * 50)
-
-    # Restore items back to the original queue
-    while not temp_queue.empty():
-        mem_scheduler._web_log_message_queue.put(temp_queue.get())
-
-    print(f"\nTotal {log_count} web log entries displayed.")
-    print("=" * 110 + "\n")
-
-
 if __name__ == "__main__":
     run_with_scheduler_init()