Pi0模型部署中的持续集成与交付实践
如果你正在把Pi0这样的机器人基础模型部署到实际应用中，可能会遇到这样的问题：每次模型更新都要手动重新训练、测试、部署，流程繁琐还容易出错。好不容易调好了一个版本，过两天数据更新了或者模型结构改了，又得从头再来一遍。
这其实就是缺少一套自动化流程的问题。在软件工程领域，我们管这叫持续集成与持续交付，也就是CI/CD。但把CI/CD套用到AI模型部署上，特别是像Pi0这样的机器人模型，会遇到不少新挑战——模型文件动辄几十GB、推理环境复杂、硬件依赖多，传统的CI/CD工具可能不太够用。
我在这篇文章里，想跟你聊聊怎么为Pi0模型搭建一套实用的CI/CD流程。这不是什么高深的理论，而是我们团队在实际项目中踩过坑、总结出来的经验。我会从最基础的自动化构建开始，讲到测试策略，再到部署上线，最后还会分享一些我们遇到的真实问题和解决方案。
1. 为什么Pi0模型需要CI/CD？
你可能觉得，机器人模型部署跟传统软件不太一样，有必要搞CI/CD吗？我先说说我们的经历。
去年我们接手了一个仓储机器人项目，用的就是Pi0模型。最开始的时候，每次模型更新都是手动操作：下载新权重、跑测试脚本、部署到测试机器人、观察效果、没问题再推到生产环境。听起来好像还行，但实际操作起来问题一大堆。
最头疼的是版本管理混乱。有一次，测试环境用的模型版本是v1.2.3，生产环境却误用了v1.2.2，结果机器人抓取成功率直接掉了15%。排查了半天才发现是版本问题。还有一次，新模型在开发人员的机器上跑得好好的，一到测试机器人上就各种报错，后来发现是CUDA版本不匹配。
这些问题其实都可以通过CI/CD来解决。但Pi0模型有几个特殊的地方，让传统的CI/CD方案需要调整：
模型体积巨大：Pi0的基础模型就有几十GB，加上微调后的权重，轻松超过100GB。这意味着你的CI/CD流水线得有足够的存储空间，而且数据传输会成为瓶颈。
硬件依赖复杂：Pi0需要GPU进行推理，而且对显存要求不低。你的测试环境得有真实的GPU硬件，或者至少能模拟GPU环境。
测试成本高：传统软件的单元测试跑起来很快，但Pi0模型的测试可能需要实际控制机器人执行任务，一次测试可能就要几分钟甚至几十分钟。
部署环境多样：有的机器人是固定工位的，有的是移动的；有的用RTX 4090，有的用A100。你的部署流程得能适应不同的硬件配置。
正是因为这些挑战，为Pi0设计CI/CD流程时，不能简单照搬传统软件的那一套。需要针对性地解决存储、测试、部署这些特殊问题。
2. 搭建基础CI/CD流水线
说了这么多，具体该怎么搭建呢？我从最简单的开始，一步步带你搭建一个能用的CI/CD流水线。
2.1 环境准备与工具选择
首先得选对工具。我们试过好几套方案，最后发现GitLab CI/CD + Docker + MinIO这个组合比较适合Pi0这类大模型。
为什么选这些工具？GitLab CI/CD的好处是跟代码仓库深度集成，配置起来相对简单。Docker能保证环境一致性，避免“在我机器上能跑”的问题。MinIO是用来存模型权重的，相当于自己搭建的S3，比直接存Git仓库要靠谱得多——你肯定不想把100GB的模型文件也提交到Git里吧。
这是我们的基础目录结构：
pi0-ci-cd/ ├── .gitlab-ci.yml # CI/CD配置文件 ├── Dockerfile # 构建镜像的Dockerfile ├── docker-compose.yml # 本地测试用的编排文件 ├── scripts/ │ ├── build.sh # 构建脚本 │ ├── test.sh # 测试脚本 │ └── deploy.sh # 部署脚本 ├── tests/ │ ├── unit/ # 单元测试 │ ├── integration/ # 集成测试 │ └── performance/ # 性能测试 └── config/ ├── model_config.yaml # 模型配置 └── deploy_config.yaml # 部署配置2.2 自动化构建流程
构建阶段的核心任务是把代码和模型权重打包成可部署的镜像。这里有个关键点：不要把模型权重直接打包进Docker镜像，否则镜像会变得巨大无比，构建和推送都很慢。
我们的做法是，在构建时只打包代码和依赖，模型权重通过环境变量或者配置文件在运行时动态加载。这样镜像大小能控制在几个GB以内，而不是几十个GB。
这是我们的Dockerfile示例：
# 使用NVIDIA基础镜像，确保CUDA兼容性 FROM nvidia/cuda:12.1-runtime-ubuntu22.04 # 安装系统依赖 RUN apt-get update && apt-get install -y \ python3.10 \ python3-pip \ git \ wget \ && rm -rf /var/lib/apt/lists/* # 设置工作目录 WORKDIR /app # 复制依赖文件 COPY requirements.txt . COPY pyproject.toml . # 安装Python依赖 # 注意：这里用uv替代pip，因为OpenPI官方推荐uv RUN curl -LsSf https://astral.sh/uv/install.sh | sh ENV PATH="/root/.local/bin:${PATH}" RUN uv pip install -r requirements.txt # 复制应用代码 COPY . . # 设置环境变量 ENV PYTHONPATH=/app ENV OPENPI_DATA_HOME=/data/models # 创建模型缓存目录 RUN mkdir -p /data/models # 启动命令 CMD ["python3", "scripts/start_server.py"]对应的构建脚本scripts/build.sh：
#!/bin/bash set -e # 读取版本号 VERSION=${CI_COMMIT_TAG:-${CI_COMMIT_SHORT_SHA:-"latest"}} # 构建Docker镜像 echo "Building Docker image..." docker build -t registry.example.com/pi0-model:${VERSION} . # 推送到镜像仓库 echo "Pushing to registry..." docker push registry.example.com/pi0-model:${VERSION} # 保存版本信息 echo "VERSION=${VERSION}" > .build-info2.3 GitLab CI/CD配置
有了Dockerfile和构建脚本，接下来配置GitLab CI/CD。我们在.gitlab-ci.yml里定义了三个阶段：构建、测试、部署。
stages: - build - test - deploy variables: DOCKER_REGISTRY: registry.example.com MODEL_STORAGE: s3://models-bucket # 构建阶段 build: stage: build image: docker:latest services: - docker:dind script: - apk add --no-cache bash - chmod +x scripts/build.sh - ./scripts/build.sh artifacts: paths: - .build-info expire_in: 1 week # 测试阶段 test-unit: stage: test image: ${DOCKER_REGISTRY}/pi0-model:${CI_COMMIT_SHORT_SHA} services: - name: nvidia/cuda:12.1-runtime alias: cuda script: - chmod +x scripts/test.sh - ./scripts/test.sh --unit dependencies: - build test-integration: stage: test image: ${DOCKER_REGISTRY}/pi0-model:${CI_COMMIT_SHORT_SHA} services: - name: nvidia/cuda:12.1-runtime alias: cuda script: - chmod +x scripts/test.sh - ./scripts/test.sh --integration dependencies: - build only: - main - tags # 部署阶段 deploy-staging: stage: deploy image: alpine:latest script: - chmod +x scripts/deploy.sh - ./scripts/deploy.sh --env staging dependencies: - build only: - main deploy-production: stage: deploy image: alpine:latest script: - chmod +x scripts/deploy.sh - ./scripts/deploy.sh --env production dependencies: - build only: - tags这个配置有几个关键点：
- 使用dind（Docker in Docker）：让GitLab Runner能在容器内构建Docker镜像
- 分阶段测试：单元测试每次提交都跑，集成测试只在主分支和标签上跑
- 条件部署：开发分支自动部署到测试环境，打标签才部署到生产环境
3. 模型测试策略
测试AI模型跟测试传统软件不太一样。你不能只测代码逻辑，还得测模型效果。我们设计了三级测试策略：单元测试、集成测试、性能测试。
3.1 单元测试：确保代码质量
单元测试主要测工具函数、数据处理逻辑这些纯代码部分。比如测试数据预处理是否正确、模型配置加载是否正常。
# tests/unit/test_data_processing.py import pytest import numpy as np from src.data_processing import normalize_actions, preprocess_images def test_normalize_actions(): """测试动作归一化""" # 模拟机器人动作数据 raw_actions = np.array([[1.0, 2.0, 3.0], [4.0, 5.0, 6.0]]) stats = { 'mean': np.array([2.5, 3.5, 4.5]), 'std': np.array([1.5, 1.5, 1.5]) } normalized = normalize_actions(raw_actions, stats) expected = np.array([[-1.0, -1.0, -1.0], [1.0, 1.0, 1.0]]) assert np.allclose(normalized, expected), "动作归一化结果不符合预期" def test_preprocess_images(): """测试图像预处理""" # 模拟输入图像 dummy_image = np.random.randint(0, 256, (480, 640, 3), dtype=np.uint8) processed = preprocess_images(dummy_image) # 检查输出形状 assert processed.shape == (3, 224, 224), "图像预处理后的形状不正确" # 检查数值范围 assert processed.min() >= -1.0 and processed.max() <= 1.0, "图像数值范围不正确"3.2 集成测试：验证模型功能
集成测试要验证模型能不能正常加载、推理流程是否通畅。这里我们不用真实机器人，而是用模拟的观测数据。
# tests/integration/test_model_inference.py import pytest import torch from src.model_loader import load_pi0_model @pytest.mark.integration class TestPi0Inference: @pytest.fixture(scope="class") def model(self): """加载Pi0模型""" model = load_pi0_model( checkpoint_path="gs://openpi-assets/checkpoints/pi0_base", device="cuda" if torch.cuda.is_available() else "cpu" ) return model def test_model_loading(self, model): """测试模型加载""" assert model is not None assert hasattr(model, 'infer') def test_inference_shape(self, model): """测试推理输出形状""" # 创建模拟输入 batch_size = 2 dummy_input = { "observation/exterior_image_1_left": torch.randn(batch_size, 3, 224, 224), "observation/wrist_image_left": torch.randn(batch_size, 3, 224, 224), "observation/state": torch.randn(batch_size, 14), "prompt": ["pick up the cup", "place it on the table"] } # 运行推理 with torch.no_grad(): output = model.infer(dummy_input) # 检查输出 assert "actions" in output actions = output["actions"] # Pi0输出的是动作序列，形状应该是 [batch_size, horizon, action_dim] assert actions.shape[0] == batch_size assert actions.shape[2] == 14 # 默认的ALOHA动作维度 def test_inference_time(self, model): """测试推理时间""" import time dummy_input = { "observation/exterior_image_1_left": torch.randn(1, 3, 224, 224), "observation/wrist_image_left": torch.randn(1, 3, 224, 224), "observation/state": torch.randn(1, 14), "prompt": ["test prompt"] } # 预热 for _ in range(3): _ = model.infer(dummy_input) # 正式测试 start_time = time.time() for _ in range(10): _ = model.infer(dummy_input) end_time = time.time() avg_time = (end_time - start_time) / 10 print(f"平均推理时间: {avg_time:.3f}秒") # 要求实时性：推理时间应小于0.1秒（10Hz） assert avg_time < 0.1, f"推理时间{avg_time:.3f}秒过长，不满足实时性要求"3.3 性能测试：确保满足实时要求
机器人控制对实时性要求很高，Pi0需要达到至少10Hz的控制频率。我们的性能测试主要关注推理速度和内存使用。
# tests/performance/test_realtime_performance.py import pytest import torch import psutil import time from src.model_loader import load_pi0_model @pytest.mark.performance class TestRealtimePerformance: def test_memory_usage(self): """测试内存使用""" process = psutil.Process() # 记录初始内存 initial_memory = process.memory_info().rss / 1024 / 1024 # MB # 加载模型 model = load_pi0_model( checkpoint_path="gs://openpi-assets/checkpoints/pi0_base", device="cuda" if torch.cuda.is_available() else "cpu" ) # 记录加载后内存 after_load_memory = process.memory_info().rss / 1024 / 1024 memory_increase = after_load_memory - initial_memory print(f"模型加载内存增加: {memory_increase:.1f} MB") # 要求：内存增加不超过8GB assert memory_increase < 8192, f"内存使用过多: {memory_increase:.1f} MB" def test_throughput(self): """测试吞吐量""" model = load_pi0_model( checkpoint_path="gs://openpi-assets/checkpoints/pi0_base", device="cuda" if torch.cuda.is_available() else "cpu" ) # 准备测试数据 batch_sizes = [1, 2, 4, 8] results = {} for batch_size in batch_sizes: dummy_input = { "observation/exterior_image_1_left": torch.randn(batch_size, 3, 224, 224), "observation/wrist_image_left": torch.randn(batch_size, 3, 224, 224), "observation/state": torch.randn(batch_size, 14), "prompt": ["test"] * batch_size } # 预热 for _ in range(3): _ = model.infer(dummy_input) # 测试吞吐量 num_iterations = 20 start_time = time.time() for _ in range(num_iterations): _ = model.infer(dummy_input) end_time = time.time() total_samples = batch_size * num_iterations throughput = total_samples / (end_time - start_time) results[batch_size] = throughput print(f"Batch size {batch_size}: {throughput:.1f} samples/sec") # 生成报告 print("\n吞吐量测试结果:") for batch_size, throughput in results.items(): print(f" Batch {batch_size}: {throughput:.1f} samples/sec")4. 部署与监控
测试通过后，就该部署了。Pi0模型的部署有几个特殊考虑：模型权重管理、硬件适配、服务发现。
4.1 模型权重管理
我们之前提到过，不要把模型权重打包进Docker镜像。那怎么管理呢？我们的方案是用模型注册表。
# scripts/model_registry.py import boto3 import json import hashlib from pathlib import Path from typing import Dict, Optional class ModelRegistry: def __init__(self, bucket_name: str, endpoint_url: Optional[str] = None): """初始化模型注册表 Args: bucket_name: S3兼容存储的桶名 endpoint_url: MinIO端点URL，如果是AWS S3则留空 """ self.bucket_name = bucket_name if endpoint_url: self.s3_client = boto3.client( 's3', endpoint_url=endpoint_url, aws_access_key_id='minioadmin', aws_secret_access_key='minioadmin' ) else: self.s3_client = boto3.client('s3') def register_model(self, model_path: str, metadata: Dict) -> str: """注册新模型 Args: model_path: 本地模型文件路径 metadata: 模型元数据 Returns: 模型版本ID """ # 计算模型哈希作为版本ID with open(model_path, 'rb') as f: model_hash = hashlib.sha256(f.read()).hexdigest()[:16] version_id = f"pi0_{metadata.get('type', 'base')}_{model_hash}" # 上传模型文件 s3_key = f"models/{version_id}/weights.safetensors" self.s3_client.upload_file(model_path, self.bucket_name, s3_key) # 保存元数据 metadata_key = f"models/{version_id}/metadata.json" metadata['version'] = version_id metadata['s3_key'] = s3_key self.s3_client.put_object( Bucket=self.bucket_name, Key=metadata_key, Body=json.dumps(metadata, indent=2) ) print(f"模型已注册: {version_id}") return version_id def get_model(self, version_id: str, download_path: str) -> Dict: """获取模型 Args: version_id: 模型版本ID download_path: 下载到的本地路径 Returns: 模型元数据 """ # 获取元数据 metadata_key = f"models/{version_id}/metadata.json" response = self.s3_client.get_object( Bucket=self.bucket_name, Key=metadata_key ) metadata = json.loads(response['Body'].read()) # 下载模型权重 s3_key = metadata['s3_key'] model_dir = Path(download_path) / version_id model_dir.mkdir(parents=True, exist_ok=True) model_path = model_dir / "weights.safetensors" self.s3_client.download_file( self.bucket_name, s3_key, str(model_path) ) metadata['local_path'] = str(model_path) return metadata def list_models(self, model_type: Optional[str] = None): """列出所有模型""" prefix = "models/" response = self.s3_client.list_objects_v2( Bucket=self.bucket_name, Prefix=prefix, Delimiter='/' ) models = [] for prefix_obj in response.get('CommonPrefixes', []): model_prefix = prefix_obj['Prefix'] version_id = model_prefix.split('/')[1] # 获取元数据 try: metadata = self.get_metadata(version_id) if model_type and metadata.get('type') != model_type: continue models.append(metadata) except: continue return models def get_metadata(self, version_id: str) -> Dict: """获取模型元数据""" metadata_key = f"models/{version_id}/metadata.json" response = self.s3_client.get_object( Bucket=self.bucket_name, Key=metadata_key ) return json.loads(response['Body'].read())4.2 部署脚本
部署脚本需要处理不同环境（测试、生产）的配置差异，还要能回滚到之前的版本。
#!/bin/bash # scripts/deploy.sh set -e # 解析参数 ENV="staging" ROLLBACK=false VERSION="" while [[ $# -gt 0 ]]; do case $1 in --env) ENV="$2" shift 2 ;; --rollback) ROLLBACK=true shift ;; --version) VERSION="$2" shift 2 ;; *) echo "未知参数: $1" exit 1 ;; esac done # 加载环境配置 source config/${ENV}.env # 如果是回滚，获取上一个版本 if [ "$ROLLBACK" = true ]; then if [ -z "$VERSION" ]; then echo "获取上一个部署版本..." VERSION=$(curl -s "${DEPLOY_API}/api/deployments/${ENV}/previous") fi echo "回滚到版本: $VERSION" else # 获取最新构建的版本 if [ -z "$VERSION" ]; then VERSION=$(cat .build-info | grep VERSION | cut -d'=' -f2) fi echo "部署版本: $VERSION" fi # 下载模型权重（如果不存在） MODEL_DIR="/data/models/pi0_${VERSION}" if [ ! -d "$MODEL_DIR" ]; then echo "下载模型权重..." mkdir -p "$MODEL_DIR" # 从模型注册表获取 python scripts/download_model.py \ --version "$VERSION" \ --output "$MODEL_DIR" fi # 更新Docker Compose配置 echo "更新部署配置..." cat > docker-compose.${ENV}.yml << EOF version: '3.8' services: pi0-inference: image: ${DOCKER_REGISTRY}/pi0-model:${VERSION} container_name: pi0-inference-${ENV} restart: unless-stopped runtime: nvidia environment: - OPENPI_DATA_HOME=/data/models - MODEL_VERSION=${VERSION} - ENVIRONMENT=${ENV} volumes: - ${MODEL_DIR}:/data/models - ./logs:/app/logs ports: - "${INFERENCE_PORT}:8000" command: ["python3", "scripts/start_server.py", "--model-dir", "/data/models/pi0_${VERSION}"] monitor: image: prom/prometheus:latest container_name: prometheus-${ENV} volumes: - ./config/prometheus.yml:/etc/prometheus/prometheus.yml - prometheus-data:/prometheus ports: - "${MONITOR_PORT}:9090" grafana: image: grafana/grafana:latest container_name: grafana-${ENV} environment: - GF_SECURITY_ADMIN_PASSWORD=${GRAFANA_PASSWORD} volumes: - grafana-data:/var/lib/grafana ports: - "${GRAFANA_PORT}:3000" volumes: prometheus-data: grafana-data: EOF # 停止旧服务，启动新服务 echo "重启服务..." docker-compose -f docker-compose.${ENV}.yml down docker-compose -f docker-compose.${ENV}.yml up -d # 健康检查 echo "等待服务启动..." sleep 10 MAX_RETRIES=30 RETRY_COUNT=0 while [ $RETRY_COUNT -lt $MAX_RETRIES ]; do if curl -s "http://localhost:${INFERENCE_PORT}/health" | grep -q "healthy"; then echo "服务启动成功！" # 记录部署历史 curl -X POST "${DEPLOY_API}/api/deployments" \ -H "Content-Type: application/json" \ -d "{\"environment\": \"${ENV}\", \"version\": \"${VERSION}\", \"status\": \"success\"}" exit 0 fi RETRY_COUNT=$((RETRY_COUNT + 1)) echo "健康检查失败，重试 ($RETRY_COUNT/$MAX_RETRIES)..." sleep 5 done echo "服务启动失败，执行回滚..." # 回滚逻辑 ./scripts/deploy.sh --env "$ENV" --rollback exit 14.3 监控与告警
部署上线后，监控就很重要了。我们需要知道模型服务是否健康、推理延迟是否正常、资源使用是否合理。
# config/prometheus.yml global: scrape_interval: 15s evaluation_interval: 15s rule_files: - "alerts.yml" scrape_configs: - job_name: 'pi0-inference' static_configs: - targets: ['pi0-inference:8000'] metrics_path: '/metrics' - job_name: 'node-exporter' static_configs: - targets: ['node-exporter:9100'] alerting: alertmanagers: - static_configs: - targets: ['alertmanager:9093']对应的告警规则：
# config/alerts.yml groups: - name: pi0_alerts rules: - alert: HighInferenceLatency expr: rate(pi0_inference_duration_seconds_sum[5m]) / rate(pi0_inference_duration_seconds_count[5m]) > 0.1 for: 2m labels: severity: warning annotations: summary: "Pi0推理延迟过高" description: "过去5分钟内平均推理延迟超过100ms，当前值 {{ $value }}s" - alert: ModelServiceDown expr: up{job="pi0-inference"} == 0 for: 1m labels: severity: critical annotations: summary: "Pi0推理服务下线" description: "Pi0推理服务已下线超过1分钟" - alert: HighGPUUsage expr: (1 - (avg by (instance) (rate(node_gpu_memory_free_bytes[5m])) / avg by (instance) (node_gpu_memory_total_bytes))) * 100 > 90 for: 5m labels: severity: warning annotations: summary: "GPU使用率过高" description: "GPU使用率超过90%，当前值 {{ $value }}%" - alert: LowSuccessRate expr: (1 - (rate(pi0_inference_success_total[10m]) / rate(pi0_inference_requests_total[10m]))) * 100 > 10 for: 5m labels: severity: warning annotations: summary: "推理成功率下降" description: "过去10分钟内推理失败率超过10%，当前值 {{ $value }}%"5. 实际应用中的挑战与解决方案
在实际项目中，我们遇到了不少挑战。这里分享几个典型问题和我们的解决方案。
5.1 挑战一：模型版本管理混乱
问题：早期我们直接用Git管理模型权重，结果仓库体积爆炸，clone一次要几个小时。而且分不清哪个代码版本对应哪个模型版本。
解决方案：我们引入了模型注册表，把模型权重存在对象存储里，用数据库记录版本信息。每个模型版本都有唯一的哈希ID，跟代码版本通过Git标签关联。
# scripts/version_manager.py import sqlite3 from datetime import datetime from typing import Dict, List class VersionManager: def __init__(self, db_path: str = "versions.db"): self.conn = sqlite3.connect(db_path) self._init_db() def _init_db(self): """初始化数据库""" cursor = self.conn.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS model_versions ( id INTEGER PRIMARY KEY AUTOINCREMENT, version_hash TEXT UNIQUE NOT NULL, git_tag TEXT, model_type TEXT, description TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, s3_path TEXT, metadata_json TEXT ) ''') cursor.execute(''' CREATE TABLE IF NOT EXISTS deployments ( id INTEGER PRIMARY KEY AUTOINCREMENT, environment TEXT, version_hash TEXT, deployed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, status TEXT, FOREIGN KEY (version_hash) REFERENCES model_versions (version_hash) ) ''') self.conn.commit() def register_version(self, version_hash: str, git_tag: str, model_type: str, s3_path: str, metadata: Dict, description: str = ""): """注册新版本""" cursor = self.conn.cursor() cursor.execute(''' INSERT OR REPLACE INTO model_versions (version_hash, git_tag, model_type, description, s3_path, metadata_json) VALUES (?, ?, ?, ?, ?, ?) ''', (version_hash, git_tag, model_type, description, s3_path, json.dumps(metadata))) self.conn.commit() def get_latest_version(self, model_type: str = "pi0"): """获取最新版本""" cursor = self.conn.cursor() cursor.execute(''' SELECT * FROM model_versions WHERE model_type = ? ORDER BY created_at DESC LIMIT 1 ''', (model_type,)) return cursor.fetchone() def record_deployment(self, environment: str, version_hash: str, status: str = "success"): """记录部署""" cursor = self.conn.cursor() cursor.execute(''' INSERT INTO deployments (environment, version_hash, status) VALUES (?, ?, ?) ''', (environment, version_hash, status)) self.conn.commit() def get_deployment_history(self, environment: str, limit: int = 10): """获取部署历史""" cursor = self.conn.cursor() cursor.execute(''' SELECT d.deployed_at, v.version_hash, v.git_tag, d.status FROM deployments d JOIN model_versions v ON d.version_hash = v.version_hash WHERE d.environment = ? ORDER BY d.deployed_at DESC LIMIT ? ''', (environment, limit)) return cursor.fetchall()5.2 挑战二：测试环境资源不足
问题：Pi0模型测试需要GPU，但我们的CI/CD Runner只有CPU。租用GPU实例成本又太高。
解决方案：我们搭建了混合测试环境。单元测试在CI Runner上跑，集成测试和性能测试通过API调用专门的测试集群。
# tests/integration/test_cluster.py import pytest import requests import time class TestClusterIntegration: """测试集群集成测试""" @pytest.fixture def cluster_client(self): """连接到测试集群""" # 从环境变量获取集群配置 cluster_url = os.getenv("TEST_CLUSTER_URL", "http://test-cluster:8080") return ClusterClient(cluster_url) def test_cluster_availability(self, cluster_client): """测试集群可用性""" # 获取集群状态 status = cluster_client.get_status() assert status['healthy'] == True assert status['gpu_available'] == True # 检查GPU型号和显存 gpu_info = status['gpu_info'] assert gpu_info['memory_total'] >= 8 * 1024 # 至少8GB显存 def test_remote_inference(self, cluster_client): """测试远程推理""" # 准备测试数据 test_data = { "images": { "cam_high": [...] # 模拟图像数据 }, "prompt": "pick up the red block" } # 提交推理任务 job_id = cluster_client.submit_inference(test_data) # 等待结果 max_wait = 30 # 最多等待30秒 for _ in range(max_wait): result = cluster_client.get_result(job_id) if result['status'] == 'completed': assert 'actions' in result assert result['inference_time'] < 1.0 # 推理时间应小于1秒 return time.sleep(1) pytest.fail(f"推理任务超时: {job_id}") def test_concurrent_requests(self, cluster_client): """测试并发请求""" import concurrent.futures # 准备多个并发请求 requests_data = [ {"prompt": f"test request {i}"} for i in range(10) ] # 并发提交 with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor: futures = [ executor.submit(cluster_client.submit_inference, data) for data in requests_data ] # 收集结果 results = [] for future in concurrent.futures.as_completed(futures): job_id = future.result() result = cluster_client.get_result(job_id, timeout=60) results.append(result) # 检查所有请求都成功 success_count = sum(1 for r in results if r['status'] == 'completed') assert success_count == len(requests_data), f"只有{success_count}/{len(requests_data)}个请求成功" # 计算平均推理时间 avg_time = sum(r['inference_time'] for r in results) / len(results) print(f"平均推理时间: {avg_time:.3f}秒") assert avg_time < 0.5 # 平均应小于500ms5.3 挑战三：回滚困难
问题：新版本上线后发现问题，回滚到旧版本很麻烦，需要手动修改配置、重启服务。
解决方案：我们实现了一键回滚机制，每次部署都自动备份旧版本，回滚时只需要执行一个命令。
#!/bin/bash # scripts/rollback.sh set -e ENV=${1:-"staging"} ROLLBACK_TO=${2:-"previous"} echo "开始回滚 ${ENV} 环境..." # 获取当前版本 CURRENT_VERSION=$(docker inspect pi0-inference-${ENV} --format='{{.Config.Image}}' | cut -d':' -f2) if [ "$ROLLBACK_TO" = "previous" ]; then # 获取上一个版本 PREVIOUS_VERSION=$(python scripts/get_previous_version.py --env "$ENV") if [ -z "$PREVIOUS_VERSION" ]; then echo "找不到上一个版本" exit 1 fi TARGET_VERSION=$PREVIOUS_VERSION else TARGET_VERSION=$ROLLBACK_TO fi echo "当前版本: ${CURRENT_VERSION}" echo "目标版本: ${TARGET_VERSION}" if [ "$CURRENT_VERSION" = "$TARGET_VERSION" ]; then echo "已经是目标版本，无需回滚" exit 0 fi # 检查目标版本是否存在 if ! docker pull ${DOCKER_REGISTRY}/pi0-model:${TARGET_VERSION} > /dev/null 2>&1; then echo "目标版本不存在: ${TARGET_VERSION}" exit 1 fi # 备份当前部署配置 BACKUP_DIR="/backup/deployments/${ENV}/$(date +%Y%m%d_%H%M%S)" mkdir -p "$BACKUP_DIR" cp docker-compose.${ENV}.yml "$BACKUP_DIR/" cp config/${ENV}.env "$BACKUP_DIR/" echo "备份已保存到: ${BACKUP_DIR}" # 执行回滚 echo "执行回滚..." ./scripts/deploy.sh --env "$ENV" --version "$TARGET_VERSION" # 验证回滚 sleep 10 if curl -s "http://localhost:${INFERENCE_PORT}/health" | grep -q "healthy"; then echo "回滚成功！" # 发送通知 python scripts/send_notification.py \ --type "rollback" \ --env "$ENV" \ --from_version "$CURRENT_VERSION" \ --to_version "$TARGET_VERSION" \ --reason "manual" else echo "回滚后健康检查失败" # 尝试恢复备份 echo "尝试恢复备份..." cp "$BACKUP_DIR/docker-compose.${ENV}.yml" . cp "$BACKUP_DIR/${ENV}.env" config/ ./scripts/deploy.sh --env "$ENV" --version "$CURRENT_VERSION" echo "已恢复到原始版本" exit 1 fi6. 最佳实践总结
经过多个项目的实践，我们总结了一些Pi0模型CI/CD的最佳实践：
1. 分层测试策略：不要所有测试都在CI流水线里跑。单元测试这种轻量级的可以在每次提交时跑，集成测试和性能测试可以按需触发，或者放在夜间跑。
2. 模型与代码分离：模型权重和代码一定要分开管理。代码用Git，模型用对象存储+注册表。这样既不会拖慢Git操作，也方便模型版本管理。
3. 环境一致性：用Docker保证开发、测试、生产环境的一致性。特别是CUDA版本、Python包版本这些容易出问题的地方。
4. 渐进式部署：新版本不要一次性全量上线。可以先部署到少量机器人上，观察一段时间没问题再逐步扩大范围。
5. 完善的监控：不仅要监控服务是否存活，还要监控推理延迟、成功率、资源使用率这些业务指标。设置合理的告警阈值，早发现问题早处理。
6. 文档和自动化：所有操作都要有文档，能自动化的尽量自动化。特别是回滚、扩容这些应急操作，一定要有脚本，关键时刻能快速执行。
7. 成本控制：GPU资源很贵，CI/CD流水线要优化资源使用。比如测试环境可以共用GPU实例，非工作时间自动缩容等。
总结
给Pi0这样的机器人模型搭建CI/CD流程，确实比传统软件要复杂一些。但一旦搭建好了，带来的收益也是巨大的：部署速度从小时级降到分钟级、版本管理清晰可控、问题能快速发现和修复。
我们现在的流程，从代码提交到生产环境部署，最快15分钟就能完成。而且因为有完整的测试和监控，上线后出问题的概率大大降低。即使出了问题，也能一键回滚，把影响降到最低。
当然，这套方案还在不断优化中。比如我们正在探索用Kubernetes来管理推理服务，实现更灵活的扩缩容；也在研究模型压缩技术，减少模型体积，加快部署速度。
如果你也在部署Pi0或者其他大模型，希望这篇文章能给你一些参考。最重要的是根据实际需求来设计，不用追求一步到位。可以先从自动化构建开始，然后加上测试，再完善部署和监控，一步步迭代完善。
获取更多AI镜像
想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。
