Skip to content

docs: switch scheduler quick-start samples to HTTP requests #2086

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Lilili-04 wants to merge 1 commit into
base: main
Choose a base branch
from
+143 −77
Open

docs: switch scheduler quick-start samples to HTTP requests #2086

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
56 changes: 38 additions & 18 deletions docs/cn/open_source/open_source_api/scheduler/ wait.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
```
42 changes: 27 additions & 15 deletions docs/cn/open_source/open_source_api/scheduler/get_status.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
```
```
52 changes: 32 additions & 20 deletions docs/en/open_source/open_source_api/scheduler/get_status.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
```
```
70 changes: 46 additions & 24 deletions docs/en/open_source/open_source_api/scheduler/wait.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
```
```
Loading