# SmolVLA实战教程：批量处理CSV指令集并导出动作轨迹的Python脚本 ## 1. 学习目标与前置知识 本教程将手把手教你如何使用Python脚本批量处理CSV格式的机器人指令，并通过SmolVLA模型生成对应的动作轨迹。无论你是机器人技术的新手还是有一定经验的开发者，都能快速上手。 **学习完成后，你将掌握：** - 如何准备和格式化CSV指令文件 - 如何使用SmolVLA进行批量推理 - 如何导出和处理生成的动作轨迹数据 - 如何自动化整个处理流程 **前置知识要求：** - 基本的Python编程经验（会写简单脚本即可） - 了解CSV文件的基本结构 - 不需要深入的机器人学知识 ## 2. 环境准备与快速部署 ### 2.1 确保SmolVLA环境正常运行 首先确认你的SmolVLA环境已经启动并运行在7860端口： ```bash # 检查服务状态 cd /root/smolvla_base python /root/smolvla_base/app.py ``` 如果看到类似"Running on local URL: http://127.0.0.1:7860"的输出，说明服务正常运行。 ### 2.2 安装必要的Python库 除了SmolVLA自带的依赖，我们还需要安装pandas库来处理CSV文件： ```bash pip install pandas requests ``` ## 3. CSV指令文件格式详解 ### 3.1 基础CSV结构 创建一个名为`robot_commands.csv`的文件，包含以下列： ```csv command_id,instruction,joint0,joint1,joint2,joint3,joint4,joint5 1,"Pick up the red cube",0.12,-0.45,0.78,-0.23,0.15,0.05 2,"Place it in the blue box",0.18,-0.38,0.82,-0.19,0.12,0.03 3,"Move to home position",0.0,0.0,0.0,0.0,0.0,0.0 ``` ### 3.2 各列说明 - **command_id**: 指令的唯一标识符 - **instruction**: 自然语言指令（必填） - **joint0-joint5**: 机器人的当前关节状态（可选，但建议提供） ## 4. 批量处理Python脚本详解 ### 4.1 完整的处理脚本 创建一个名为`batch_process_smolvla.py`的文件： ```python import pandas as pd import requests import json import time from typing import List, Dict class SmolVLAClient: def __init__(self, base_url: str = "http://localhost:7860"): self.base_url = base_url self.api_url = f"{base_url}/api/predict" def generate_action(self, instruction: str, joint_states: List[float] = None): """生成单个动作""" if joint_states is None: joint_states = [0.0] * 6 payload = { "instruction": instruction, "joint0": joint_states[0], "joint1": joint_states[1], "joint2": joint_states[2], "joint3": joint_states[3], "joint4": joint_states[4], "joint5": joint_states[5] } try: response = requests.post(self.api_url, json=payload, timeout=30) response.raise_for_status() return response.json() except Exception as e: print(f"Error generating action: {e}") return None def process_csv_batch(csv_file: str, output_file: str, batch_size: int = 10): """批量处理CSV文件""" # 读取CSV文件 df = pd.read_csv(csv_file) client = SmolVLAClient() results = [] for index, row in df.iterrows(): print(f"Processing command {index + 1}/{len(df)}: {row['instruction']}") # 准备关节状态 joint_states = [ row.get('joint0', 0.0), row.get('joint1', 0.0), row.get('joint2', 0.0), row.get('joint3', 0.0), row.get('joint4', 0.0), row.get('joint5', 0.0) ] # 生成动作 result = client.generate_action(row['instruction'], joint_states) if result: # 添加原始指令信息 result['command_id'] = row['command_id'] result['original_instruction'] = row['instruction'] result['original_joint_states'] = joint_states results.append(result) # 批量处理时添加延迟，避免服务器过载 time.sleep(1) # 保存结果 with open(output_file, 'w') as f: json.dump(results, f, indent=2) print(f"Processing complete. Results saved to {output_file}") return results if __name__ == "__main__": # 使用示例 process_csv_batch("robot_commands.csv", "action_results.json") ``` ### 4.2 脚本使用方法 运行脚本非常简单： ```bash python batch_process_smolvla.py ``` 脚本会自动读取`robot_commands.csv`文件，处理所有指令，并将结果保存到`action_results.json`。 ## 5. 结果文件解析与使用 ### 5.1 输出JSON结构 处理完成后，你会得到一个结构清晰的JSON文件： ```json [ { "command_id": 1, "original_instruction": "Pick up the red cube", "original_joint_states": [0.12, -0.45, 0.78, -0.23, 0.15, 0.05], "predicted_action": [0.15, -0.42, 0.81, -0.20, 0.18, 0.02], "input_state": [0.12, -0.45, 0.78, -0.23, 0.15, 0.05], "run_mode": "real" } ] ``` ### 5.2 结果可视化脚本 创建一个简单的可视化脚本： ```python import json import matplotlib.pyplot as plt import numpy as np def visualize_actions(results_file: str): """可视化动作轨迹""" with open(results_file, 'r') as f: data = json.load(f) # 提取数据 commands = [f"Cmd {item['command_id']}" for item in data] original_states = np.array([item['original_joint_states'] for item in data]) predicted_actions = np.array([item['predicted_action'] for item in data]) # 创建图表 fig, axes = plt.subplots(2, 3, figsize=(15, 10)) axes = axes.flatten() joint_names = ['Joint 0 (Base)', 'Joint 1 (Shoulder)', 'Joint 2 (Elbow)', 'Joint 3 (Wrist Bend)', 'Joint 4 (Wrist Rotate)', 'Joint 5 (Gripper)'] for i in range(6): axes[i].plot(original_states[:, i], 'bo-', label='Original State') axes[i].plot(predicted_actions[:, i], 'ro-', label='Predicted Action') axes[i].set_title(joint_names[i]) axes[i].set_ylabel('Position') axes[i].legend() axes[i].set_xticks(range(len(commands))) axes[i].set_xticklabels(commands, rotation=45) plt.tight_layout() plt.savefig('action_trajectory.png', dpi=300, bbox_inches='tight') plt.show() # 使用示例 visualize_actions('action_results.json') ``` ## 6. 实用技巧与进阶功能 ### 6.1 错误处理与重试机制 在实际使用中，网络波动或服务暂时不可用是常见问题。添加重试机制： ```python def generate_action_with_retry(self, instruction: str, joint_states: List[float], max_retries: int = 3): """带重试的动作生成""" for attempt in range(max_retries): try: result = self.generate_action(instruction, joint_states) if result: return result except Exception as e: print(f"Attempt {attempt + 1} failed: {e}") time.sleep(2 ** attempt) # 指数退避 return None ``` ### 6.2 批量处理优化 对于大量指令，可以使用并行处理： ```python from concurrent.futures import ThreadPoolExecutor, as_completed def process_batch_parallel(df, max_workers: int = 4): """并行处理批量指令""" client = SmolVLAClient() results = [] with ThreadPoolExecutor(max_workers=max_workers) as executor: future_to_index = { executor.submit( client.generate_action_with_retry, row['instruction'], [row.get(f'joint{i}', 0.0) for i in range(6)] ): index for index, row in df.iterrows() } for future in as_completed(future_to_index): index = future_to_index[future] try: result = future.result() if result: result['command_id'] = df.iloc[index]['command_id'] results.append(result) except Exception as e: print(f"Command {index} failed: {e}") return results ``` ## 7. 常见问题解答 ### 7.1 服务连接失败怎么办？ ```python # 添加服务健康检查 def check_service_health(base_url: str = "http://localhost:7860"): try: response = requests.get(f"{base_url}/", timeout=5) return response.status_code == 200 except: return False if not check_service_health(): print("SmolVLA服务未启动，请先启动服务") exit(1) ``` ### 7.2 处理速度太慢怎么办？ - 减少批量大小（batch_size） - 使用并行处理（但注意不要过度占用服务器资源） - 确保使用GPU加速 ### 7.3 结果不准确如何调试？ ```python # 添加调试信息输出 def debug_command(client, instruction, joint_states): print(f"Debug: Instruction='{instruction}'") print(f"Debug: Joint States={joint_states}") result = client.generate_action(instruction, joint_states) print(f"Debug: Result={result}") return result ``` ## 8. 总结回顾 通过本教程，你已经学会了： 1. **环境准备**：确保SmolVLA服务正常运行并安装必要依赖 2. **数据准备**：创建符合格式要求的CSV指令文件 3. **批量处理**：使用Python脚本自动处理大量指令 4. **结果分析**：解析和可视化生成的动作轨迹 5. **进阶技巧**：错误处理、并行处理和调试方法 这个批量处理脚本可以大大提升工作效率，特别适合需要处理大量机器人指令的场景。你可以根据实际需求进一步定制脚本，比如添加更多的数据验证、结果分析功能，或者集成到更大的自动化流程中。 **下一步建议：** - 尝试处理不同类型的指令，观察模型的表现 - 探索SmolVLA的其他功能和应用场景 - 考虑将脚本集成到你的机器人控制系统中 > **获取更多AI镜像** > > 想探索更多AI镜像和应用场景？访问 [CSDN星图镜像广场](https://ai.csdn.net/?utm_source=mirror_blog_end)，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。