针对 Xiaomi-Robotics-U0 批量生成操作序列的需求，日常处理不能靠手动逐条提交。建议写一个 Python 脚本，按固定行读取输入，逐条调用模型接口，把结果写回文件，碰到超时或限流先重试，最后输出失败清单。这个流程不依赖具体机器人型号，只依赖接口的入参和出参结构。

批量生成操作序列的通用做法是统一输入格式、逐条循环调用模型接口、设置超时重试并记录日志、最后统计失败列表。适用场景是允许顺序生成且样本量较大的任务；操作上先小批量试跑；验证方式是查看日志与失败原因；边界是接口限流和生成质量需结合实际环境确认。

准备输入数据并统一格式

先把每个操作序列的触发条件、机器人状态、目标动作整理成一行一条记录。推荐使用 JSONL，每行一个 JSON 对象，方便追加失败样本。

{"id": "task_001", "trigger": "餐具在台面", "constraints": "左手持海绵", "expected": "抓取碗"}

若上游提供 CSV，脚本里可以用 csv.DictReader 读取，再映射成请求体。字段映射通常包括：id 对应样本编号，trigger 拼接成系统提示，constraints 拼进用户内容，expected 只用于事后对比，不输入模型。

import csv, json
with open('input.csv', encoding='utf-8') as f:
    for row in csv.DictReader(f):
        item = {
            'id': row['id'],
            'prompt': f"场景：{row['scene']}，约束：{row['constraints']}",
            'expected': row['expected']
        }
        print(json.dumps(item, ensure_ascii=False))

编写批量调用模型接口的循环

循环里只处理一件事：发送请求、接收响应、保存结果。不要把日志和重试堆在同一个函数里，方便单独调试。

import json, time, requests

API_URL = 'http://your-api.example/v1/generate'

def call_model(item, timeout=30):
    resp = requests.post(API_URL, json={
        'model': 'xiaomi-u0',
        'prompt': item['prompt'],
        'max_tokens': 512
    }, timeout=timeout)
    resp.raise_for_status()
    return resp.json()['output']

主循环顺序读取输入文件，每成功一条就把结果写入 .jsonl，失败则记录错误信息。建议使用 tqdm 显示进度，但不要依赖进度条做判断。

for item in batch:
    try:
        result = call_model(item)
        outputs.append({'id': item['id'], 'result': result})
    except Exception as e:
        errors.append({'id': item['id'], 'error': str(e)})
    time.sleep(0.5)  # 简单限速，具体间隔需结合接口限制

加入超时、重试与日志

网络波动时直接失败会浪费整批时间。requests 的 timeout 必须设置，推荐 30 秒以上。重试可以用 tenacity 库，也可以用简单递减函数。

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def call_model_with_retry(item, timeout=30):
    # 同样的请求逻辑，略
    return call_model(item, timeout=timeout)

日志配置要带上时间、样本 id、错误类型。不要只 print，否则中断后很难定位。

import logging
logging.basicConfig(
    filename='run.log',
    level=logging.INFO,
    format='%(asctime)s %(levelname)s %(message)s'
)

在每个样本处理完成时记一条 info，异常时记 warning，重试超过上限就记 error。日志行建议包含样本 id，方便后期 grep。

统计生成结果并输出失败原因

批跑完成后，程序要给出三样东西：成功数、失败数、失败列表。失败列表保存成单独文件，下次可以从这里续跑。

print(f'成功 {len(outputs)} 条，失败 {len(errors)} 条')

with open('failed.jsonl', 'w', encoding='utf-8') as f:
    for err in errors:
        print(json.dumps(err, ensure_ascii=False), file=f)

如果失败原因集中在 timeout、connection error、429 这类状态码，说明需要调大重试间隔或降低并发；如果是 400/422，通常是输入格式问题，要回头检查字段映射。续跑时，先读取 failed.jsonl 作为输入，并保留原 id，避免产生重复数据。