trio.SamplingClient

class SamplingClient:
    def __init__(
        self,
        task_id: str,
        base_model: str
    ):

SamplingClient 是用于文本生成与推理的客户端，通过 ServiceClient.create_sampling_client() 创建。

import pytrio as trio

client = trio.ServiceClient()
sampling_client = client.create_sampling_client(base_model="Qwen/Qwen3.5-4B")

tokenizer = sampling_client.get_tokenizer()
prompt_ids = tokenizer.encode("Hello, world!")

future = sampling_client.sample(
    prompt=trio.ModelInput.from_ints(prompt_ids),
    num_samples=4,
    sampling_params=trio.SamplingParams(temperature=1.0, max_tokens=128),
)
response = future.result()

属性

属性	类型	说明
`task_id`	`str`	当前采样任务 ID
`base_model`	`str`	使用的基础模型名称

方法

sample

def sample(
    self,
    prompt: ModelInput,
    num_samples: int = 1,
    sampling_params: SamplingParams = SamplingParams(),
    include_prompt_logprobs: bool = False,
    topk_prompt_logprobs: int = 0,
    return_text: bool = True,
) -> APIFuture[SampleResponse]

根据输入 prompt 生成文本补全。

参数

参数	类型	默认值	说明
`prompt`	`ModelInput`	—	输入的 token id 列表
`num_samples`	`int`	`1`	生成的样本数量
`sampling_params`	`SamplingParams`	`SamplingParams()`	采样参数，见 `SamplingParams`
`include_prompt_logprobs`	`bool`	`False`	是否在返回结果中包含 prompt 部分的对数概率
`topk_prompt_logprobs`	`int`	`0`	返回 prompt 部分 top-k 对数概率的数量，`0` 表示不返回
`return_text`	`bool`	`True`	是否在返回结果中包含生成的文本

返回值

APIFuture[SampleResponse] — 调用 .result() 获取生成结果。其中包含以下字段：

sequences：生成的序列，其中包含 stop_reason、tokens、tokens、logprobs 等信息。
prompt_logprobs：每个 prompt token 的对数概率。
topk_prompt_logprobs：每个 prompt token 的 top-k 对数概率。
output_tokens：生成的 token 数量。

示例

future = sampling_client.sample(
    prompt=prompt,
    num_samples=4,
    sampling_params=trio.SamplingParams(temperature=1.0, max_tokens=128),
)
response = future.result()
print(response.sequences)

compute_logprobs

def compute_logprobs(self, prompt: ModelInput) -> APIFuture[list[float | None]]

计算 prompt 中每个 token 的对数概率。

参数

参数	类型	说明
`prompt`	`ModelInput`	输入的 token id 列表

返回值

APIFuture[list[float | None]] — 调用 .result() 获取对数概率列表。

示例

logprobs = sampling_client.compute_logprobs(prompt=prompt).result()

get_tokenizer

def get_tokenizer(self)

获取与当前基础模型匹配的 tokenizer，基于 transformers / modelscope 提供的 AutoTokenizer。

示例

tokenizer = sampling_client.get_tokenizer()
prompt = tokenizer.encode("The meaning of life is")

异步方法

sample_async

async def sample_async(
    self,
    prompt: ModelInput,
    num_samples: int = 1,
    sampling_params: SamplingParams = SamplingParams(),
    include_prompt_logprobs: bool = False,
    topk_prompt_logprobs: int = 0,
    return_text: bool = True,
) -> SampleResponse

sample 的异步版本，参数相同。和同步 sample() 不同，sample_async() 在 await 后直接返回 SampleResponse，不需要再调用 .result()。

response = await sampling_client.sample_async(
    prompt=prompt,
    num_samples=4,
    sampling_params=trio.SamplingParams(temperature=1.0, max_tokens=128),
)
print(response.sequences)

compute_logprobs_async

async def compute_logprobs_async(self, prompt: ModelInput) -> APIFuture[list[float | None]]

compute_logprobs 的异步版本，参数相同。

logprobs = (await sampling_client.compute_logprobs_async(prompt=prompt)).result()

本页目录