From a550fe0ba205b2dc4150e50ce0dd43be1c3ec7a8 Mon Sep 17 00:00:00 2001 From: Lilili-04 <782697268@qq.com> Date: Thu, 9 Jul 2026 16:08:50 +0800 Subject: [PATCH] docs: switch scheduler quick-start samples to HTTP requests Update CN/EN scheduler wait and get_status docs to use direct HTTP API examples instead of SDK snippets, including blocking and SSE usage patterns. Co-authored-by: Cursor --- .../open_source_api/scheduler/ wait.md | 56 ++++++++++----- .../open_source_api/scheduler/get_status.md | 42 +++++++---- .../open_source_api/scheduler/get_status.md | 52 ++++++++------ .../open_source_api/scheduler/wait.md | 70 ++++++++++++------- 4 files changed, 143 insertions(+), 77 deletions(-) diff --git a/docs/cn/open_source/open_source_api/scheduler/ wait.md b/docs/cn/open_source/open_source_api/scheduler/ wait.md index 9849ffe68..00bec57a9 100644 --- a/docs/cn/open_source/open_source_api/scheduler/ wait.md +++ b/docs/cn/open_source/open_source_api/scheduler/ wait.md @@ -42,36 +42,56 @@ desc: 提供阻塞等待与流式进度观测能力，确保在执行后续操 ## 4. 快速上手示例 -使用开源版 SDK 进行阻塞式等待： +以下示例通过 HTTP 直接调用接口： ```python -from memos.api.client import MemOSClient +import json -client = MemOSClient(api_key="...", base_url="...") +import requests + +base_url = "http://localhost:8001" # 替换为您的 MemOS 服务地址 user_name = "dev_user_01" # --- 场景 A：同步阻塞等待 (常用于 Python 自动化脚本) --- print(f"正在等待用户 {user_name} 的任务队列清空...") -res = client.wait_until_idle( - user_name=user_name, - timeout_seconds=300, - poll_interval=2 +resp = requests.post( + f"{base_url}/product/scheduler/wait", + params={ + "user_name": user_name, + "timeout_seconds": 300, + "poll_interval": 2, + }, + timeout=310, # 客户端超时应略大于 timeout_seconds ) -if res and res.code == 200: - print("✅ 任务已全部完成。") +resp.raise_for_status() +result = resp.json() +if result["message"] == "idle": + print(f"✅ 任务已全部完成，耗时 {result['data']['waited_seconds']} 秒。") +else: + print(f"⚠️ 等待超时，仍有 {result['data']['running_tasks']} 个任务在执行。") # --- 场景 B：流式进度观测 (常用于前端进度条渲染) --- print("开始监听任务实时进度流...") -# 注意：SSE 接口在 SDK 中通常返回一个生成器 (Generator) -progress_stream = client.stream_scheduler_progress( - user_name=user_name, - timeout_seconds=300 +resp = requests.get( + f"{base_url}/product/scheduler/wait/stream", + params={"user_name": user_name, "timeout_seconds": 300}, + stream=True, + timeout=310, ) +resp.raise_for_status() + +for line in resp.iter_lines(decode_unicode=True): + # SSE 帧格式为 "data: {...}" + if not line or not line.startswith("data: "): + continue + event = json.loads(line[len("data: "):]) -for event in progress_stream: - # 实时打印剩余任务数 - print(f"当前排队任务数: {event['remaining_tasks_count']}") - if event['status'] == 'idle': + # 实时打印当前活跃任务数 + print(f"当前活跃任务数: {event['active_tasks']}") + if event["status"] == "idle": print("🎉 调度器已空闲") break -``` + if event["status"] == "timeout": + print("⚠️ 监听超时，调度器仍有任务在执行") + break +``` \ No newline at end of file diff --git a/docs/cn/open_source/open_source_api/scheduler/get_status.md b/docs/cn/open_source/open_source_api/scheduler/get_status.md index 87e1a4a5a..118d1e44b 100644 --- a/docs/cn/open_source/open_source_api/scheduler/get_status.md +++ b/docs/cn/open_source/open_source_api/scheduler/get_status.md @@ -64,34 +64,46 @@ desc: 监控 MemOS 异步任务的生命周期，提供包括任务进度、队 ## 4. 快速上手示例 -使用 SDK 轮询任务状态直至完成： +以下示例通过 HTTP 直接调用接口，轮询任务状态直至完成： ```python -from memos.api.client import MemOSClient import time -client = MemOSClient(api_key="...", base_url="...") +import requests + +base_url = "http://localhost:8001" # 替换为您的 MemOS 服务地址 # 1. 系统级概览：查看整个 MemOS 系统的运行健康度 -global_res = client.get_all_scheduler_status() -if global_res: - print(f"系统运行概况: {global_res.data['scheduler_summary']}") +resp = requests.get(f"{base_url}/product/scheduler/allstatus", timeout=10) +resp.raise_for_status() +print(f"系统运行概况: {resp.json()['data']['scheduler_summary']}") # 2. 队列指标监控：检查特定用户的任务积压情况 -queue_res = client.get_task_queue_status(user_id="dev_user_01") -if queue_res: - print(f"待处理任务数: {queue_res.data['remaining_tasks_count']}") - print(f"已下发未完成任务数: {queue_res.data['pending_tasks_count']}") +resp = requests.get( + f"{base_url}/product/scheduler/task_queue_status", + params={"user_id": "dev_user_01"}, + timeout=10, +) +resp.raise_for_status() +queue_data = resp.json()["data"] +print(f"排队中任务数: {queue_data['remaining_tasks_count']}") +print(f"已下发未确认任务数: {queue_data['pending_tasks_count']}") # 3. 任务进度追踪：轮询特定任务直至结束 task_id = "task_888999" while True: - res = client.get_task_status(user_id="dev_user_01", task_id=task_id) - if res and res.code == 200: - current_status = res.data[0]['status'] # data 为状态列表 + resp = requests.get( + f"{base_url}/product/scheduler/status", + params={"user_id": "dev_user_01", "task_id": task_id}, + timeout=10, + ) + resp.raise_for_status() + items = resp.json()["data"] # data 为状态列表 + if items: + current_status = items[0]["status"] print(f"任务 {task_id} 当前状态: {current_status}") - if current_status in ['completed', 'failed', 'cancelled']: + if current_status in ["completed", "failed", "cancelled"]: break time.sleep(2) -``` +``` \ No newline at end of file diff --git a/docs/en/open_source/open_source_api/scheduler/get_status.md b/docs/en/open_source/open_source_api/scheduler/get_status.md index f2014d9e5..dad5f31c8 100644 --- a/docs/en/open_source/open_source_api/scheduler/get_status.md +++ b/docs/en/open_source/open_source_api/scheduler/get_status.md @@ -66,34 +66,46 @@ When you send a status request, **SchedulerHandler** performs the following oper ## 4. Quick Start -Poll task status with the SDK until completion: +The following example calls the endpoints directly over HTTP and polls a task until it finishes: ```python -from memos.api.client import MemOSClient import time -client = MemOSClient(api_key="...", base_url="...") +import requests -# 1. System overview: inspect overall MemOS health. -global_res = client.get_all_scheduler_status() -if global_res: - print(f"System summary: {global_res.data['scheduler_summary']}") +base_url = "http://localhost:8001" # Replace with your MemOS server address -# 2. Queue metrics: inspect backlog for a specific user. -queue_res = client.get_task_queue_status(user_id="dev_user_01") -if queue_res: - print(f"Remaining tasks: {queue_res.data['remaining_tasks_count']}") - print(f"Pending tasks: {queue_res.data['pending_tasks_count']}") +# 1. System-level overview: check the overall health of the MemOS system +resp = requests.get(f"{base_url}/product/scheduler/allstatus", timeout=10) +resp.raise_for_status() +print(f"System summary: {resp.json()['data']['scheduler_summary']}") -# 3. Task progress: poll a specific task until it finishes. +# 2. Queue metrics: check the task backlog of a specific user +resp = requests.get( + f"{base_url}/product/scheduler/task_queue_status", + params={"user_id": "dev_user_01"}, + timeout=10, +) +resp.raise_for_status() +queue_data = resp.json()["data"] +print(f"Enqueued tasks: {queue_data['remaining_tasks_count']}") +print(f"Delivered but unacknowledged tasks: {queue_data['pending_tasks_count']}") + +# 3. Task progress tracking: poll a specific task until it reaches a terminal state task_id = "task_888999" while True: - res = client.get_task_status(user_id="dev_user_01", task_id=task_id) - if res and res.code == 200: - current_status = res.data[0]['status'] # data is a status list - print(f"Task {task_id} status: {current_status}") - - if current_status in ['completed', 'failed', 'cancelled']: + resp = requests.get( + f"{base_url}/product/scheduler/status", + params={"user_id": "dev_user_01", "task_id": task_id}, + timeout=10, + ) + resp.raise_for_status() + items = resp.json()["data"] # data is a list of status items + if items: + current_status = items[0]["status"] + print(f"Task {task_id} current status: {current_status}") + + if current_status in ["completed", "failed", "cancelled"]: break time.sleep(2) -``` +``` \ No newline at end of file diff --git a/docs/en/open_source/open_source_api/scheduler/wait.md b/docs/en/open_source/open_source_api/scheduler/wait.md index 9de0ff4be..3e3a9666b 100644 --- a/docs/en/open_source/open_source_api/scheduler/wait.md +++ b/docs/en/open_source/open_source_api/scheduler/wait.md @@ -41,36 +41,58 @@ Both endpoints share the following query parameters: ## 4. Quick Start -Use the open-source SDK for a blocking wait: +## 4. Quick Start + +The following example calls the endpoints directly over HTTP: ```python -from memos.api.client import MemOSClient +import json + +import requests -client = MemOSClient(api_key="...", base_url="...") +base_url = "http://localhost:8001" # Replace with your MemOS server address user_name = "dev_user_01" -# Scenario A: blocking wait, commonly used in Python automation scripts. +# --- Scenario A: synchronous blocking wait (common in Python automation scripts) --- print(f"Waiting for user {user_name}'s task queue to drain...") -res = client.wait_until_idle( - user_name=user_name, - timeout_seconds=300, - poll_interval=2 +resp = requests.post( + f"{base_url}/product/scheduler/wait", + params={ + "user_name": user_name, + "timeout_seconds": 300, + "poll_interval": 2, + }, + timeout=310, # client timeout should be slightly larger than timeout_seconds ) -if res and res.code == 200: - print("All tasks have completed.") - -# Scenario B: streaming progress, commonly used by frontend progress bars. -print("Listening to the live task progress stream...") -# The SSE endpoint usually returns a generator from the SDK. -progress_stream = client.stream_scheduler_progress( - user_name=user_name, - timeout_seconds=300 +resp.raise_for_status() +result = resp.json() +if result["message"] == "idle": + print(f"✅ All tasks completed in {result['data']['waited_seconds']} seconds.") +else: + print(f"⚠️ Wait timed out; {result['data']['running_tasks']} tasks still active.") + +# --- Scenario B: streaming progress observation (common for frontend progress bars) --- +print("Listening to the real-time task progress stream...") +resp = requests.get( + f"{base_url}/product/scheduler/wait/stream", + params={"user_name": user_name, "timeout_seconds": 300}, + stream=True, + timeout=310, ) - -for event in progress_stream: - # Print the remaining queued tasks in real time. - print(f"Remaining queued tasks: {event['remaining_tasks_count']}") - if event['status'] == 'idle': - print("Scheduler is idle") +resp.raise_for_status() + +for line in resp.iter_lines(decode_unicode=True): + # SSE frames are formatted as "data: {...}" + if not line or not line.startswith("data: "): + continue + event = json.loads(line[len("data: "):]) + + # Print the current number of active tasks in real time + print(f"Active tasks: {event['active_tasks']}") + if event["status"] == "idle": + print("🎉 Scheduler is idle") + break + if event["status"] == "timeout": + print("⚠️ Stream timed out; scheduler still has active tasks") break -``` +``` \ No newline at end of file