trio.RestClient

class RestClient:
    def __init__(self)

RestClient 是用于 REST API 操作的客户端，通过 ServiceClient.create_rest_client() 创建。

import pytrio as trio

client = trio.ServiceClient()
rest_client = client.create_rest_client()

# 列出权重
weights = rest_client.list_user_checkpoints().result()

# 列出训练运行
runs = rest_client.list_training_runs().result()

方法

list_user_checkpoints

def list_user_checkpoints(
    self,
    limit: int = 100,
    offset: int = 0,
) -> ConcurrentFuture[CheckpointsListResponse]

分页列出当前用户的模型权重。

参数

参数	类型	默认值	说明
`limit`	`int`	`100`	返回条数
`offset`	`int`	`0`	偏移量

返回值

ConcurrentFuture[CheckpointsListResponse] — 调用 .result() 获取权重列表及分页信息。

示例

checkpoints = rest_client.list_user_checkpoints(limit=100, offset=0).result()

get_checkpoint_archive_url

def get_checkpoint_archive_url(
    self,
    checkpoint_id: str,
) -> ConcurrentFuture[CheckpointArchiveUrlResponse]

获取指定模型权重的临时下载链接。

参数

参数	类型	说明
`checkpoint_id`	`str`	模型权重 ID，通过 `list_user_checkpoints` 获取

返回值

ConcurrentFuture[CheckpointArchiveUrlResponse] — 调用 .result() 获取包含临时下载链接的响应。

示例

url_info = rest_client.get_checkpoint_archive_url(checkpoint_id="abc123").result()

get_training_run

def get_training_run(self, training_run_id: str) -> ConcurrentFuture[TrainingRun]

获取指定训练运行的详细信息。

参数

参数	类型	说明
`training_run_id`	`str`	训练运行 ID

返回值

ConcurrentFuture[TrainingRun] — 调用 .result() 获取训练运行详情。

示例

run = rest_client.get_training_run(training_run_id="run-001").result()

list_training_runs

def list_training_runs(self, limit: int = 20, offset: int = 0) -> ConcurrentFuture[TrainingRunsResponse]

分页列出当前用户的训练运行。

参数

参数	类型	默认值	说明
`limit`	`int`	`20`	返回条数
`offset`	`int`	`0`	偏移量

返回值

ConcurrentFuture[TrainingRunsResponse] — 调用 .result() 获取训练运行列表及分页信息。

示例

runs = rest_client.list_training_runs(limit=10, offset=0).result()

list_checkpoints

def list_checkpoints(self, training_run_id: str) -> ConcurrentFuture[CheckpointsListResponse]

列出指定训练运行下的所有检查点。

参数

参数	类型	说明
`training_run_id`	`str`	训练运行 ID

返回值

ConcurrentFuture[CheckpointsListResponse] — 调用 .result() 获取检查点列表及分页信息。

示例

checkpoints = rest_client.list_checkpoints(training_run_id="run-001").result()

delete_checkpoint

    def delete_checkpoint(
        self,
        training_run_id: str,
        checkpoint_id: str,
    ) -> ConcurrentFuture[None]

删除指定训练运行下的检查点。

参数

参数	类型	说明
`training_run_id`	`str`	训练运行 ID
`checkpoint_id`	`str`	检查点 ID

示例

rest_client.delete_checkpoint(training_run_id="run-001", checkpoint_id="ckpt-step100").result()

get_session

def get_session(self, session_id: str) -> ConcurrentFuture[GetSessionResponse]

获取指定会话的信息。

参数

参数	类型	说明
`session_id`	`str`	会话 ID

返回值

Dict — 会话详情。

示例

session = rest_client.get_session(session_id="sess-abc").result()

list_sessions

def list_sessions(self, limit: int = 20, offset: int = 0) -> ConcurrentFuture[ListSessionsResponse]

分页列出当前用户的所有会话。

参数

参数	类型	默认值	说明
`limit`	`int`	`20`	返回条数
`offset`	`int`	`0`	偏移量

返回值

ConcurrentFuture[ListSessionsResponse] — 调用 .result() 获取会话列表及分页信息。

示例

sessions = rest_client.list_sessions().result()

get_sampler

def get_sampler(self, sampler_id: str) -> Dict

获取指定采样器的信息。

参数

参数	类型	说明
`sampler_id`	`str`	采样器 ID

返回值

Dict — 采样器详情。

示例

sampler = rest_client.get_sampler(sampler_id="sampler-xyz").result()

本页目录