跳到主要内容

批量推断

准备并上传您的批次​

批处理由API请求列表组成。个人请求的结构包括：

  • 独一无二 custom_id 用于识别每个请求并在完成后参考结果
  • A 身体 带有消息信息的对象

下面是一个如何构造批处理请求的示例：

{“custom_id”: "0", “身体”: {“max_tokens”: 100, “消息”: [{“角色”: “用户”, “内容”: “最好的法国奶酪是什么？”}]}}
{“custom_id”: 1., “身体”: {“max_tokens”: 100, “消息”: [{“角色”: “用户”, “内容”: “法国最好的葡萄酒是什么？”}]}}

将批处理保存到.json文件中。保存后，您可以上传批处理输入文件，以确保在启动批处理过程时正确引用该文件：

从 米斯特拉伊 进口 
进口 os

api_key = os.环境[“MIUI_API_KEY”]

客户 = (api_key =api_key )

batch_data =客户 .文件夹.上传(
    文件={
        “文件名”: “test.json”, 
        “内容”: 打开(“test.json”,  “rb”)}, 
 目的 = “批次”
    
)

创建新的批处理作业​

创建一个新的批处理作业，它将排队等待处理。

  • 输入文件：批处理输入文件ID的列表。
  • 模型：您只能使用一个模型（例如。， 最新代码)每批。但是，如果要比较输出，您可以在具有不同模型的同一文件上运行多个批处理。
  • 端点：我们目前支持 /v1/嵌入, /v1/聊天/补全, /v1/fim/完井, /v1/调节, /v1/聊天/版主.
  • 元数据：批处理的可选自定义元数据。
created_job =客户 .批次.工作.创造(
输入文件=[batch_data .身份证件], 
模型=“MIUI小最新”, 
端点=“/v1/聊天/补全”, 
元数据={“job_type”: “测试”}
)

获取批处理作业详细信息​

已检索job =客户 .批次.工作.得到(job_id=created_job .身份证件)

获取批处理作业结果​

客户 .文件夹.下载(file_id=已检索job .输出文件)

列出批处理作业​

您可以查看批处理作业的列表，并根据各种条件对其进行筛选，包括：

  • 状态： 排队, 跑步, 成功, 失败, 超时_已超过, 取消_请求, 取消_请求, 取消
  • 元数据：批处理的自定义元数据键和值
list_job =客户 .批次.工作.列表(
 地位=“正在运行”,    
元数据={“job_type”: “测试”}
)

请求取消批处理作业​

canceled_job =客户 .批次.工作.取消(job_id=created_job .身份证件)

端到端示例​

例子
进口 argparse
进口 json
进口 os
进口 随机
进口 时间
从 io 进口 BytesIO

进口 httpx
从 米斯特拉伊 进口文件,  


def 创建客户端():
    """
 使用环境变量中的API键创建MIUI客户端。

 退货：
 ：客户的一个例子。
"""
    返回 (api_key = os.环境[“MIUI_API_KEY”])

def generate_random_string(开始,  结束):
    """
 生成可变长度的随机字符串。

 Args：
 start（int）：字符串的最小长度。
 end（int）：字符串的最大长度。

 退货：
 str：随机生成的字符串。
"""
 长度 = 随机.随机范围(开始,  结束)
    返回 ' '.参加( 随机.选择(“abcdefghijklmnopqrstuvwxyz”,  k= 长度 ))

def 打印开始(batch_job):
    """
 打印批处理作业的统计数据。

 Args：
 batch_job：包含作业统计信息的批处理作业对象。
"""
    打印(f“请求总数： {batch_job.总计请求数}")
    打印(f“失败的请求： {batch_job.失败的请求}")
    打印(f“成功的请求： {batch_job.成功的请求}")
    打印(
        f“完成百分比： {圆((batch_job.成功的请求+batch_job.失败的请求) /batch_job.总计请求数,  4.) * 100}")


def 创建输入文件(客户 ,  num_samples):
    """
 为批处理作业创建输入文件。

 Args：
 client（MIUI）：MIUI客户端实例。
 num_samples（int）：要生成的样本数。

 退货：
 File：上传的输入文件对象。
"""
    缓冲器 = BytesIO()
    对于 idx 在里面 范围( num_samples):
 请求 = {
            “custom_id”: str( idx ), 
            “身体”: {
                “max_tokens”: 随机.兰丁(10,  1000), 
                “消息”: [{“角色”: “用户”,  “内容”:generate_random_string(100,  5000)}]
            }
        }
        缓冲器.写( json.垃圾场( 请求 ).编码(“utf-8”))
        缓冲器.写(n.编码(“utf-8”))
    返回客户 .文件夹.上传(文件=文件(文件名=“文件.jsonl”,  内容=缓冲器.getvalue()),  目的 =“批次”)


def run_batch_job(客户 ,  输入文件, 模型):
    """
 使用提供的输入文件和模型运行批处理作业。

 Args：
 client（MIUI）：MIUI客户端实例。
 input_file（file）：输入文件对象。
 model（str）：用于批处理作业的模型。

 退货：
 BatchJob：已完成的批处理作业对象。
"""
batch_job=客户 .批次.工作.创造(
输入文件=[ 输入文件.身份证件], 
模型=模型, 
端点=“/v1/聊天/补全”, 
元数据={“job_type”: “测试”}
    )

    虽然batch_job. 地位在里面 [“排队”,  “正在运行”]:
batch_job=客户 .批次.工作.得到(job_id=batch_job.身份证件)
打印开始(batch_job)
 时间.睡觉(1.)

    打印(f“批处理作业 {batch_job.身份证件} 已完成，状态： {batch_job. 地位}")
    返回batch_job


def 下载文件(客户 , file_id,  输出路径):
    """
 从MIUI服务器下载文件。

 Args：
 client（MIUI）：MIUI客户端实例。
 file_id（str）：要下载的文件的id。
 output_path（str）：文件保存的路径。
"""
    如果file_id是 不 没有:
        打印(f“正在将文件下载到 { 输出路径}")
输出文件=客户 .文件夹.下载(file_id=file_id)
        具有 打开( 输出路径,  w) 作为 f:
            对于 块 在里面输出文件.流动:
 f.写( 块 .解码(“utf-8”))
        打印(f“已将文件下载到 { 输出路径}")


def 主要的( num_samples,  成功路径,  错误路径, 模型):
    """
 运行批处理作业的主要功能。

 Args：
 num_samples（int）：要处理的样本数。
 success_path（str）：保存成功输出的路径。
 error_path（str）：保存错误输出的路径。
 model（str）：要使用的模型名称。
"""
客户 =创建客户端()
 输入文件=创建输入文件(客户 ,  num_samples)
    打印(f“已创建输入文件 { 输入文件}")

batch_job=run_batch_job(客户 ,  输入文件, 模型)
    打印(f“工作时间： {batch_job.已完成_ at -batch_job.created_at} 秒”)
下载文件(客户 , batch_job.错误文件,  错误路径)
下载文件(客户 , batch_job.输出文件,  成功路径)


如果 __姓名__ == “主__”:
 解析器 = argparse.ArgumentParser(描述=“运行MIUI AI批处理作业”)
 解析器 .添加论据(“--num_samples”,  类型=int,  违约=100,  帮助=“要处理的样本数量”)
 解析器 .添加论据(“--成功路径”,  类型=str,  违约=“output.json”,  帮助=“保存成功输出的路径”)
 解析器 .添加论据(“--error_path”,  类型=str,  违约=“error.jsonl”,  帮助=“保存错误输出的路径”)
 解析器 .添加论据(“--型号”,  类型=str,  违约=“最新代码”,  帮助=“要使用的型号名称”)

 args = 解析器 .parse_args()

主要的( args . num_samples,  args . 成功路径,  args . 错误路径,  args .模型)

常见问题​

批次API是否适用于所有型号？​

是的，批量API可用于所有模型，包括用户微调模型。

API批次是否影响定价？​

批API提供价格折扣。请查看我们的详细信息 定价页面.

API批次是否影响速率限制？​

不

批处理中的最大请求数是多少？​

目前，每个工作区最多可处理100万个请求。这意味着您不能提交超过100万个请求的作业。此外，您不能同时提交两个作业，每个作业有600000个请求。您需要等到第一个作业处理了至少200000个请求，将其待处理数量减少到400000个。到那时，有60万个请求的新工作将在限制范围内。

一个人可以创建的批处理作业的最大数量是多少？​

目前，没有最高限额。

API批处理需要多长时间？​

批量生成可能需要24小时，尽管它可能会提前完成。请注意，处理速度可能会根据当前需求和您的请求量进行调整。您的批处理结果只有在整个批处理完成后才能访问。

用户可以设置 超时_小时 创建作业时，指定作业应过期的小时数。默认为24小时，目前不能超过此限制。这意味着，如果处理未在24小时内完成，批次将过期。这在未来可能会发生变化。

我可以从我的工作区查看批处理结果吗？​

是的，批次是特定于工作区的。您可以查看在与API键关联的工作区中创建的所有批及其结果。

批量结果会过期吗？​

不，结果此时不会过期。

批量可以超过支出限额吗？​

是的，由于高吞吐量和并发处理，批处理可能会略微超过工作区配置的支出限制。