Alpamayo-R1-10B实战教程:curl命令调用/predict API实现自动化轨迹测试

1. 从WebUI到自动化:为什么需要API调用?

如果你已经体验过Alpamayo-R1-10B的WebUI界面,可能会觉得手动上传图片、输入指令、点击按钮的方式很直观。但当你需要进行大量测试时,这种手动操作就显得效率低下了。想象一下,你需要测试100个不同的驾驶场景,每个场景都要手动操作一遍,这得花多少时间?

这就是我们今天要解决的问题:如何通过curl命令自动化调用Alpamayo-R1-10B的API,实现批量轨迹测试。

1.1 手动测试 vs 自动化测试

让我用一个简单的对比来说明两者的区别:

测试方式操作步骤测试10个场景所需时间适合场景
WebUI手动测试1. 打开浏览器
2. 上传3张图片
3. 输入指令
4. 点击推理按钮
5. 记录结果
约15-20分钟少量测试、演示、探索性测试
API自动化测试1. 准备测试数据文件
2. 运行脚本或命令
3. 自动获取并保存结果
约1-2分钟批量测试、回归测试、性能评估

可以看到,自动化测试的效率是手动测试的10倍以上。更重要的是,自动化测试可以确保每次测试的条件完全一致,避免了人为操作带来的误差。

1.2 API调用的实际应用场景

在实际的自动驾驶研发中,API调用有多个重要用途:

场景一:批量验证模型性能 假设你收集了1000个不同的交叉口场景,想要测试模型在各种情况下的表现。手动测试几乎不可能完成,而通过API可以编写一个简单的脚本,一晚上就能跑完所有测试。

场景二:集成到现有系统 如果你的团队已经有自己的自动驾驶测试框架,可以通过API将Alpamayo-R1-10B集成进去,作为轨迹预测的一个模块。

场景三:持续集成/持续部署(CI/CD) 在每次代码更新后,自动运行一组核心测试用例,确保新版本没有引入回归问题。

场景四:参数调优实验 想要测试不同温度参数对轨迹生成的影响?通过API可以轻松实现参数扫描,自动收集结果并分析。

2. 准备工作:启用API服务

在开始使用curl命令之前,我们需要先确保Alpamayo-R1-10B的API服务已经正确启用。默认情况下,为了节省资源,API服务可能是关闭的。

2.1 检查当前服务状态

首先,让我们看看系统当前运行了哪些服务:

# 查看所有服务的状态
supervisorctl status

你会看到类似这样的输出:

alpamayo-webui    RUNNING   pid 12345, uptime 1:23:45
alpamayo-r1       STOPPED   Not started

这里有两个关键服务:

  • alpamayo-webui:WebUI界面服务,运行在7860端口
  • alpamayo-r1:API服务,运行在8000端口

如果alpamayo-r1显示为STOPPED,说明API服务没有启动。

2.2 启动API服务

启动API服务非常简单,只需要一个命令:

# 启动API服务
supervisorctl start alpamayo-r1

# 再次检查状态
supervisorctl status alpamayo-r1

如果一切正常,你会看到:

alpamayo-r1       RUNNING   pid 23456, uptime 0:00:30

2.3 验证API服务是否正常工作

在开始复杂的调用之前,我们先做个简单的健康检查:

# 使用curl检查API服务是否正常
curl -X GET "http://localhost:8000/health"

如果API服务正常运行,你会看到类似这样的响应:

{
  "status": "healthy",
  "model_loaded": true,
  "timestamp": "2024-01-15T10:30:00Z"
}

如果看到这个响应,恭喜你!API服务已经准备就绪。

2.4 配置API服务开机自启(可选)

如果你希望每次重启服务器后API服务都能自动启动,可以这样配置:

# 编辑API服务的配置文件
sudo vi /etc/supervisor/conf.d/alpamayo-r1.conf

找到以下两行配置:

autostart=false
autorestart=false

将它们改为:

autostart=true
autorestart=true

保存文件后,重新加载配置:

# 重新加载supervisor配置
supervisorctl reread
supervisorctl update

# 重启API服务使配置生效
supervisorctl restart alpamayo-r1

现在,即使服务器重启,API服务也会自动启动。

3. 理解API接口:/predict端点详解

在开始编写curl命令之前,我们需要先了解Alpamayo-R1-10B的API接口设计。核心的预测接口是/predict端点,它接收特定的输入数据,返回轨迹预测结果。

3.1 API接口基本信息

让我们先看看这个接口的基本信息:

属性
端点地址http://localhost:8000/predict
请求方法POST
Content-Typeapplication/json
响应格式JSON

3.2 请求数据结构

/predict接口期望接收一个JSON格式的请求体,包含以下关键字段:

{
  "front_camera": "base64编码的前视摄像头图像",
  "left_camera": "base64编码的左侧摄像头图像", 
  "right_camera": "base64编码的右侧摄像头图像",
  "prompt": "驾驶指令文本",
  "top_p": 0.98,
  "temperature": 0.6,
  "num_samples": 1
}

让我详细解释每个字段:

1. 摄像头图像字段(front_camera, left_camera, right_camera)

  • 类型:字符串
  • 格式:Base64编码的JPEG或PNG图像数据
  • 要求:三张图像必须同时提供,即使某些场景下可能不需要所有视角
  • 示例:"data:image/jpeg;base64,/9j/4AAQSkZJRgABAQ...(很长的一串字符)"

2. 驾驶指令(prompt)

  • 类型:字符串
  • 说明:用自然语言描述驾驶任务
  • 示例:"Navigate through the intersection safely"
  • 其他常用指令:
    • "Turn left at the intersection"
    • "Follow the vehicle ahead"
    • "Merge into the right lane"
    • "Stop at the crosswalk"

3. 生成参数

  • top_p:核采样概率,范围0.0-1.0,默认0.98
    • 值越小,生成结果越保守、确定性越强
    • 值越大,生成结果越多样、创造性越强
  • temperature:采样温度,范围0.0-2.0,默认0.6
    • 值越小,结果越确定(类似argmax)
    • 值越大,结果越随机
  • num_samples:轨迹采样数量,范围1-6,默认1
    • 每次推理生成的轨迹数量
    • 多个样本可以评估模型的不确定性

3.3 响应数据结构

API调用成功后,会返回一个JSON格式的响应:

{
  "status": "success",
  "reasoning": "模型的推理过程文本",
  "trajectory": [
    [x1, y1, z1, t1],
    [x2, y2, z2, t2],
    // ... 共64个时间步
  ],
  "metadata": {
    "model": "Alpamayo-R1-10B",
    "inference_time": 1.23,
    "timestamp": "2024-01-15T10:30:00Z"
  }
}

响应字段说明:

  1. status:请求状态,成功时为"success"
  2. reasoning:模型的Chain-of-Causation推理过程
    • 这是Alpamayo-R1的核心特性之一
    • 以文本形式解释模型为什么做出这样的决策
    • 示例:"前方有行人正在过马路,需要减速让行;左侧车道有车辆,不适合变道;保持当前车道行驶是最安全的选择。"
  3. trajectory:预测的轨迹数据
    • 一个包含64个时间步的数组
    • 每个时间步是[x, y, z, t]四个值
    • x, y, z是三维坐标,t是时间戳
  4. metadata:元数据信息
    • 包含模型名称、推理时间、时间戳等信息

3.4 错误响应

如果API调用失败,会返回错误信息:

{
  "status": "error",
  "message": "错误描述",
  "error_code": "ERROR_CODE"
}

常见错误包括:

  • MODEL_NOT_LOADED:模型未加载
  • INVALID_IMAGE:图像数据格式错误
  • INVALID_PROMPT:驾驶指令为空或格式错误
  • GPU_OUT_OF_MEMORY:GPU显存不足

4. 实战开始:第一个curl命令

现在我们已经了解了API的基本信息,让我们开始编写第一个curl命令。我会从最简单的开始,逐步增加复杂度。

4.1 准备测试图像

首先,我们需要准备测试用的图像。为了简化,我先教你如何将本地图像转换为Base64编码。

方法一:使用Python脚本转换

创建一个名为convert_image.py的文件:

import base64
import json

def image_to_base64(image_path):
    """将图像文件转换为Base64编码字符串"""
    with open(image_path, "rb") as image_file:
        # 读取图像二进制数据
        image_data = image_file.read()
        # 转换为Base64
        base64_str = base64.b64encode(image_data).decode('utf-8')
        # 添加Data URL前缀
        return f"data:image/jpeg;base64,{base64_str}"

# 示例:转换三张测试图像
front_image = image_to_base64("front_camera.jpg")
left_image = image_to_base64("left_camera.jpg")  
right_image = image_to_base64("right_camera.jpg")

# 保存到JSON文件,方便后续使用
request_data = {
    "front_camera": front_image,
    "left_camera": left_image,
    "right_camera": right_image,
    "prompt": "Navigate through the intersection safely",
    "top_p": 0.98,
    "temperature": 0.6,
    "num_samples": 1
}

with open("test_request.json", "w") as f:
    json.dump(request_data, f, indent=2)

print("图像转换完成,已保存到 test_request.json")

运行这个脚本:

python convert_image.py

方法二:使用命令行工具(如果没有Python环境)

如果你不想写Python脚本,也可以用命令行工具:

# 将图像转换为Base64并保存到文件
echo -n "data:image/jpeg;base64," > front_base64.txt
base64 front_camera.jpg >> front_base64.txt

# 查看结果
cat front_base64.txt

4.2 基础curl命令

现在,让我们编写第一个curl命令。假设我们已经有了Base64编码的图像数据,保存在test_request.json文件中。

最简单的调用:

# 使用文件中的JSON数据调用API
curl -X POST "http://localhost:8000/predict" \
  -H "Content-Type: application/json" \
  -d @test_request.json

这个命令做了三件事:

  1. -X POST:指定使用POST方法
  2. -H "Content-Type: application/json":设置请求头,告诉服务器我们发送的是JSON数据
  3. -d @test_request.json:从文件读取JSON数据作为请求体

将响应保存到文件:

默认情况下,curl会将响应输出到终端。为了后续分析,我们通常需要保存到文件:

# 将响应保存到文件
curl -X POST "http://localhost:8000/predict" \
  -H "Content-Type: application/json" \
  -d @test_request.json \
  -o response.json

现在,响应数据就保存在response.json文件中了。

4.3 查看和分析响应

让我们看看响应文件里有什么:

# 查看响应文件内容
cat response.json

你会看到类似这样的输出(为了可读性,我做了简化):

{
  "status": "success",
  "reasoning": "分析场景:前方是十字路口,交通信号灯为绿色。左侧有车辆等待左转,右侧车道畅通。决策:保持当前车道直行,注意观察左侧车辆动态。执行:生成平滑的直行轨迹,保持适当车速。",
  "trajectory": [
    [1.23, 0.45, 0.0, 0.0],
    [1.25, 0.48, 0.0, 0.1],
    [1.28, 0.52, 0.0, 0.2],
    // ... 更多数据点
  ],
  "metadata": {
    "model": "Alpamayo-R1-10B",
    "inference_time": 1.45,
    "timestamp": "2024-01-15T10:30:00Z"
  }
}

使用jq工具美化输出:

如果你安装了jq工具(JSON处理器),可以更漂亮地查看响应:

# 安装jq(如果还没有)
# Ubuntu/Debian: sudo apt-get install jq
# CentOS/RHEL: sudo yum install jq

# 使用jq格式化JSON输出
curl -X POST "http://localhost:8000/predict" \
  -H "Content-Type: application/json" \
  -d @test_request.json | jq '.'

或者只查看特定字段:

# 只查看推理过程
curl -X POST "http://localhost:8000/predict" \
  -H "Content-Type: application/json" \
  -d @test_request.json | jq '.reasoning'

# 只查看推理时间
curl -X POST "http://localhost:8000/predict" \
  -H "Content-Type: application/json" \
  -d @test_request.json | jq '.metadata.inference_time'

# 查看轨迹的前5个点
curl -X POST "http://localhost:8000/predict" \
  -H "Content-Type: application/json" \
  -d @test_request.json | jq '.trajectory[0:5]'

5. 进阶技巧:实用的curl命令参数

现在你已经掌握了基础的curl调用,让我们学习一些更实用的技巧,这些技巧在实际工作中非常有用。

5.1 超时设置

API调用可能会因为各种原因变慢或挂起。为了避免长时间等待,我们可以设置超时:

# 设置连接超时和传输超时
curl -X POST "http://localhost:8000/predict" \
  -H "Content-Type: application/json" \
  -d @test_request.json \
  --connect-timeout 10 \    # 连接超时10秒
  --max-time 30 \           # 整个请求最大30秒
  -o response.json

参数说明:

  • --connect-timeout 10:如果10秒内无法建立连接,就放弃
  • --max-time 30:整个请求(包括连接、传输、处理)最多30秒

5.2 重试机制

在网络不稳定的环境中,一次请求可能会失败。我们可以让curl自动重试:

# 失败时重试3次,每次间隔2秒
curl -X POST "http://localhost:8000/predict" \
  -H "Content-Type: application/json" \
  -d @test_request.json \
  --retry 3 \          # 重试3次
  --retry-delay 2 \    # 每次重试间隔2秒
  --retry-max-time 60 \ # 重试总时间不超过60秒
  -o response.json

5.3 显示详细信息和进度

调试时,你可能需要查看详细的请求信息:

# 显示详细请求信息
curl -X POST "http://localhost:8000/predict" \
  -H "Content-Type: application/json" \
  -d @test_request.json \
  -v \          # 显示详细输出
  -o response.json

-v参数会显示:

  • 建立的连接信息
  • 发送的请求头
  • 接收的响应头
  • 传输进度

如果你只关心进度,可以这样:

# 显示进度条
curl -X POST "http://localhost:8000/predict" \
  -H "Content-Type: application/json" \
  -d @test_request.json \
  --progress-bar \  # 显示进度条
  -o response.json

5.4 处理HTTP状态码

有时候我们需要根据HTTP状态码做不同的处理:

# 只显示HTTP状态码
curl -X POST "http://localhost:8000/predict" \
  -H "Content-Type: application/json" \
  -d @test_request.json \
  -w "HTTP Status: %{http_code}\n" \  # 写入HTTP状态码
  -o /dev/null  # 不保存响应体

或者更完整的状态信息:

# 显示完整的请求统计信息
curl -X POST "http://localhost:8000/predict" \
  -H "Content-Type: application/json" \
  -d @test_request.json \
  -w "\n\
请求统计:\n\
状态码: %{http_code}\n\
DNS解析时间: %{time_namelookup}秒\n\
连接建立时间: %{time_connect}秒\n\
SSL握手时间: %{time_appconnect}秒\n\
开始传输时间: %{time_pretransfer}秒\n\
重定向时间: %{time_redirect}秒\n\
第一个字节时间: %{time_starttransfer}秒\n\
总时间: %{time_total}秒\n\
下载大小: %{size_download}字节\n" \
  -o response.json

5.5 直接在命令行中指定JSON数据

如果不想使用文件,也可以直接在命令行中指定JSON数据:

# 直接在命令行中指定JSON数据
curl -X POST "http://localhost:8000/predict" \
  -H "Content-Type: application/json" \
  -d '{
    "front_camera": "data:image/jpeg;base64,/9j/4AAQSkZJRg...",
    "left_camera": "data:image/jpeg;base64,/9j/4AAQSkZJRg...",
    "right_camera": "data:image/jpeg;base64,/9j/4AAQSkZJRg...",
    "prompt": "Navigate through the intersection safely",
    "top_p": 0.98,
    "temperature": 0.6,
    "num_samples": 1
  }' \
  -o response.json

注意:这种方法适合数据量小的情况。如果Base64字符串很长,建议还是使用文件。

6. 自动化测试实战:编写测试脚本

单个curl命令很有用,但真正的自动化测试需要编写脚本。让我展示几个实用的脚本示例。

6.1 基础测试脚本

首先,创建一个简单的测试脚本test_single.sh

#!/bin/bash

# test_single.sh - 单个场景测试脚本

# 设置变量
API_URL="http://localhost:8000/predict"
REQUEST_FILE="test_request.json"
RESPONSE_FILE="response_$(date +%Y%m%d_%H%M%S).json"
LOG_FILE="test_log.txt"

echo "=== 开始测试 ==="
echo "时间: $(date)"
echo "API地址: $API_URL"
echo "请求文件: $REQUEST_FILE"
echo "响应文件: $RESPONSE_FILE"
echo ""

# 执行curl命令
echo "发送请求..."
curl -X POST "$API_URL" \
  -H "Content-Type: application/json" \
  -d @"$REQUEST_FILE" \
  --connect-timeout 10 \
  --max-time 30 \
  -w "\nHTTP状态码: %{http_code}\n总时间: %{time_total}秒\n" \
  -o "$RESPONSE_FILE" \
  2>> "$LOG_FILE"

CURL_EXIT_CODE=$?

echo ""
echo "=== 测试结果 ==="

if [ $CURL_EXIT_CODE -eq 0 ]; then
    echo "✅ 请求成功完成"
    
    # 检查HTTP状态码
    HTTP_STATUS=$(grep "HTTP状态码" "$LOG_FILE" | tail -1 | awk '{print $2}')
    if [ "$HTTP_STATUS" = "200" ]; then
        echo "✅ HTTP状态码: 200 (成功)"
        
        # 检查响应内容
        if grep -q '"status":"success"' "$RESPONSE_FILE"; then
            echo "✅ API返回状态: success"
            
            # 提取推理时间
            INFERENCE_TIME=$(grep -o '"inference_time":[0-9]*\.\?[0-9]*' "$RESPONSE_FILE" | cut -d':' -f2)
            echo "⏱️  推理时间: ${INFERENCE_TIME}秒"
            
            # 提取推理文本的前50个字符
            REASONING_PREVIEW=$(grep -o '"reasoning":"[^"]*"' "$RESPONSE_FILE" | cut -d'"' -f4 | head -c 50)
            echo "💭 推理预览: ${REASONING_PREVIEW}..."
            
        else
            echo "❌ API返回状态: error"
            ERROR_MSG=$(grep -o '"message":"[^"]*"' "$RESPONSE_FILE" | cut -d'"' -f4)
            echo "错误信息: $ERROR_MSG"
        fi
    else
        echo "❌ HTTP状态码: $HTTP_STATUS"
    fi
else
    echo "❌ curl命令执行失败,退出码: $CURL_EXIT_CODE"
    echo "请查看日志文件: $LOG_FILE"
fi

echo ""
echo "响应已保存到: $RESPONSE_FILE"
echo "=== 测试结束 ==="

使用这个脚本:

# 给脚本执行权限
chmod +x test_single.sh

# 运行脚本
./test_single.sh

6.2 批量测试脚本

现在,让我们创建一个更强大的批量测试脚本batch_test.sh

#!/bin/bash

# batch_test.sh - 批量测试脚本

# 配置
API_URL="http://localhost:8000/predict"
TEST_CASES_DIR="test_cases"
RESULTS_DIR="results"
LOG_FILE="batch_test_$(date +%Y%m%d_%H%M%S).log"

# 创建目录
mkdir -p "$RESULTS_DIR"

echo "=== 批量测试开始 ===" | tee "$LOG_FILE"
echo "开始时间: $(date)" | tee -a "$LOG_FILE"
echo "测试用例目录: $TEST_CASES_DIR" | tee -a "$LOG_FILE"
echo "结果目录: $RESULTS_DIR" | tee -a "$LOG_FILE"
echo "" | tee -a "$LOG_FILE"

# 统计变量
TOTAL_TESTS=0
SUCCESS_TESTS=0
FAILED_TESTS=0
TOTAL_TIME=0

# 遍历所有测试用例
for TEST_FILE in "$TEST_CASES_DIR"/*.json; do
    if [ -f "$TEST_FILE" ]; then
        TOTAL_TESTS=$((TOTAL_TESTS + 1))
        TEST_NAME=$(basename "$TEST_FILE" .json)
        RESPONSE_FILE="$RESULTS_DIR/${TEST_NAME}_response.json"
        
        echo "测试 $TOTAL_TESTS: $TEST_NAME" | tee -a "$LOG_FILE"
        echo "请求文件: $TEST_FILE" | tee -a "$LOG_FILE"
        
        # 记录开始时间
        START_TIME=$(date +%s.%N)
        
        # 执行curl命令
        curl -X POST "$API_URL" \
          -H "Content-Type: application/json" \
          -d @"$TEST_FILE" \
          --connect-timeout 15 \
          --max-time 45 \
          --silent \
          -o "$RESPONSE_FILE" \
          -w "HTTP状态码: %{http_code}\n总时间: %{time_total}秒\n" \
          2>> "$LOG_FILE"
        
        CURL_EXIT_CODE=$?
        
        # 记录结束时间
        END_TIME=$(date +%s.%N)
        ELAPSED_TIME=$(echo "$END_TIME - $START_TIME" | bc)
        TOTAL_TIME=$(echo "$TOTAL_TIME + $ELAPSED_TIME" | bc)
        
        # 检查结果
        if [ $CURL_EXIT_CODE -eq 0 ]; then
            # 检查HTTP状态码
            HTTP_STATUS=$(tail -2 "$LOG_FILE" | grep "HTTP状态码" | awk '{print $2}')
            
            if [ "$HTTP_STATUS" = "200" ]; then
                # 检查API响应状态
                if grep -q '"status":"success"' "$RESPONSE_FILE"; then
                    echo "✅ 测试通过" | tee -a "$LOG_FILE"
                    SUCCESS_TESTS=$((SUCCESS_TESTS + 1))
                    
                    # 提取推理时间
                    INFERENCE_TIME=$(grep -o '"inference_time":[0-9]*\.\?[0-9]*' "$RESPONSE_FILE" | cut -d':' -f2)
                    echo "  推理时间: ${INFERENCE_TIME}秒" | tee -a "$LOG_FILE"
                else
                    echo "❌ API返回错误" | tee -a "$LOG_FILE"
                    FAILED_TESTS=$((FAILED_TESTS + 1))
                fi
            else
                echo "❌ HTTP错误: $HTTP_STATUS" | tee -a "$LOG_FILE"
                FAILED_TESTS=$((FAILED_TESTS + 1))
            fi
        else
            echo "❌ curl执行失败" | tee -a "$LOG_FILE"
            FAILED_TESTS=$((FAILED_TESTS + 1))
        fi
        
        echo "  耗时: ${ELAPSED_TIME}秒" | tee -a "$LOG_FILE"
        echo "" | tee -a "$LOG_FILE"
        
        # 避免请求过于频繁
        sleep 1
    fi
done

echo "=== 批量测试完成 ===" | tee -a "$LOG_FILE"
echo "统计信息:" | tee -a "$LOG_FILE"
echo "总测试数: $TOTAL_TESTS" | tee -a "$LOG_FILE"
echo "成功: $SUCCESS_TESTS" | tee -a "$LOG_FILE"
echo "失败: $FAILED_TESTS" | tee -a "$LOG_FILE"
echo "成功率: $(echo "scale=2; $SUCCESS_TESTS * 100 / $TOTAL_TESTS" | bc)%" | tee -a "$LOG_FILE"
echo "总耗时: ${TOTAL_TIME}秒" | tee -a "$LOG_FILE"
echo "平均耗时: $(echo "scale=2; $TOTAL_TIME / $TOTAL_TESTS" | bc)秒/测试" | tee -a "$LOG_FILE"
echo "完成时间: $(date)" | tee -a "$LOG_FILE"

使用这个脚本前,需要准备测试用例:

# 创建测试用例目录
mkdir -p test_cases

# 创建几个测试用例文件
# test_cases/intersection.json
# test_cases/highway.json  
# test_cases/pedestrian.json
# test_cases/parking.json

# 运行批量测试
chmod +x batch_test.sh
./batch_test.sh

6.3 参数扫描测试脚本

如果你想测试不同参数对结果的影响,可以使用这个参数扫描脚本:

#!/bin/bash

# parameter_scan.sh - 参数扫描测试脚本

API_URL="http://localhost:8000/predict"
BASE_REQUEST="base_request.json"  # 基础请求模板
RESULTS_DIR="parameter_results"
LOG_FILE="parameter_scan_$(date +%Y%m%d_%H%M%S).log"

mkdir -p "$RESULTS_DIR"

echo "=== 参数扫描测试开始 ===" | tee "$LOG_FILE"

# 定义要测试的参数值
TEMPERATURES=(0.1 0.3 0.6 0.9 1.2)
TOP_PS=(0.9 0.95 0.98 0.99 1.0)
PROMPTS=(
    "Navigate through the intersection safely"
    "Turn left at the intersection" 
    "Follow the vehicle ahead"
    "Merge into the right lane"
)

# 读取基础请求模板
BASE_JSON=$(cat "$BASE_REQUEST")

test_count=0

for temp in "${TEMPERATURES[@]}"; do
    for top_p in "${TOP_PS[@]}"; do
        for prompt in "${PROMPTS[@]}"; do
            test_count=$((test_count + 1))
            
            # 生成测试名称
            test_name="temp_${temp}_topp_${top_p}_prompt_${prompt:0:20}"
            test_name=$(echo "$test_name" | tr ' ' '_' | tr '/' '_')
            
            echo "测试 $test_count: $test_name" | tee -a "$LOG_FILE"
            echo "参数: temperature=$temp, top_p=$top_p" | tee -a "$LOG_FILE"
            echo "指令: $prompt" | tee -a "$LOG_FILE"
            
            # 替换参数
            request_json=$(echo "$BASE_JSON" | jq --arg temp "$temp" --arg top_p "$top_p" --arg prompt "$prompt" \
                '.temperature = ($temp | tonumber) | .top_p = ($top_p | tonumber) | .prompt = $prompt')
            
            # 保存临时请求文件
            echo "$request_json" > "temp_request.json"
            
            # 执行测试
            response_file="$RESULTS_DIR/${test_name}.json"
            
            curl -X POST "$API_URL" \
              -H "Content-Type: application/json" \
              -d @"temp_request.json" \
              --connect-timeout 15 \
              --max-time 45 \
              --silent \
              -o "$response_file" \
              2>> "$LOG_FILE"
            
            # 检查结果
            if [ $? -eq 0 ] && grep -q '"status":"success"' "$response_file"; then
                echo "✅ 成功" | tee -a "$LOG_FILE"
                
                # 提取推理时间
                inference_time=$(grep -o '"inference_time":[0-9]*\.\?[0-9]*' "$response_file" | cut -d':' -f2)
                echo "  推理时间: ${inference_time}秒" | tee -a "$LOG_FILE"
            else
                echo "❌ 失败" | tee -a "$LOG_FILE"
            fi
            
            echo "" | tee -a "$LOG_FILE"
            
            # 清理临时文件
            rm -f "temp_request.json"
            
            # 避免请求过于频繁
            sleep 0.5
        done
    done
done

echo "=== 参数扫描完成 ===" | tee -a "$LOG_FILE"
echo "总测试数: $test_count" | tee -a "$LOG_FILE"
echo "结果保存在: $RESULTS_DIR/" | tee -a "$LOG_FILE"

7. 实际应用:完整的自动化测试流程

现在,让我们把这些知识整合起来,创建一个完整的自动化测试流程。这个流程包括数据准备、测试执行、结果分析和报告生成。

7.1 创建测试数据准备脚本

首先,我们需要准备测试数据。创建一个脚本来自动生成测试用例:

#!/usr/bin/env python3
# prepare_test_data.py - 测试数据准备脚本

import json
import base64
import os
from pathlib import Path

def prepare_test_cases():
    """准备测试用例"""
    
    # 测试用例配置
    test_cases = [
        {
            "name": "intersection_straight",
            "front_image": "test_images/intersection/front.jpg",
            "left_image": "test_images/intersection/left.jpg",
            "right_image": "test_images/intersection/right.jpg",
            "prompt": "Navigate through the intersection safely",
            "description": "十字路口直行场景"
        },
        {
            "name": "intersection_left_turn",
            "front_image": "test_images/intersection/front.jpg",
            "left_image": "test_images/intersection/left.jpg",
            "right_image": "test_images/intersection/right.jpg",
            "prompt": "Turn left at the intersection",
            "description": "十字路口左转场景"
        },
        {
            "name": "highway_follow",
            "front_image": "test_images/highway/front.jpg",
            "left_image": "test_images/highway/left.jpg",
            "right_image": "test_images/highway/right.jpg",
            "prompt": "Follow the vehicle ahead at a safe distance",
            "description": "高速公路跟车场景"
        },
        {
            "name": "pedestrian_crossing",
            "front_image": "test_images/pedestrian/front.jpg",
            "left_image": "test_images/pedestrian/left.jpg",
            "right_image": "test_images/pedestrian/right.jpg",
            "prompt": "Stop for pedestrians at the crosswalk",
            "description": "人行横道礼让行人场景"
        },
        {
            "name": "lane_change",
            "front_image": "test_images/highway/front.jpg",
            "left_image": "test_images/highway/left.jpg",
            "right_image": "test_images/highway/right.jpg",
            "prompt": "Merge into the right lane when safe",
            "description": "高速公路变道场景"
        }
    ]
    
    # 创建测试用例目录
    test_cases_dir = Path("test_cases")
    test_cases_dir.mkdir(exist_ok=True)
    
    # 基础请求模板
    base_request = {
        "top_p": 0.98,
        "temperature": 0.6,
        "num_samples": 1
    }
    
    # 为每个测试用例创建请求文件
    for test_case in test_cases:
        # 读取图像文件并转换为Base64
        try:
            with open(test_case["front_image"], "rb") as f:
                front_base64 = f"data:image/jpeg;base64,{base64.b64encode(f.read()).decode('utf-8')}"
            
            with open(test_case["left_image"], "rb") as f:
                left_base64 = f"data:image/jpeg;base64,{base64.b64encode(f.read()).decode('utf-8')}"
            
            with open(test_case["right_image"], "rb") as f:
                right_base64 = f"data:image/jpeg;base64,{base64.b64encode(f.read()).decode('utf-8')}"
        except FileNotFoundError as e:
            print(f"警告: 图像文件不存在 - {e}")
            # 使用占位符
            front_base64 = "data:image/jpeg;base64,PLACEHOLDER"
            left_base64 = "data:image/jpeg;base64,PLACEHOLDER"
            right_base64 = "data:image/jpeg;base64,PLACEHOLDER"
        
        # 构建请求数据
        request_data = {
            **base_request,
            "front_camera": front_base64,
            "left_camera": left_base64,
            "right_camera": right_base64,
            "prompt": test_case["prompt"],
            "metadata": {
                "test_case": test_case["name"],
                "description": test_case["description"],
                "created_at": "2024-01-15T10:00:00Z"
            }
        }
        
        # 保存到文件
        output_file = test_cases_dir / f"{test_case['name']}.json"
        with open(output_file, "w") as f:
            json.dump(request_data, f, indent=2)
        
        print(f"创建测试用例: {test_case['name']} -> {output_file}")
    
    print(f"\n共创建 {len(test_cases)} 个测试用例")
    print("测试用例保存在: test_cases/")

if __name__ == "__main__":
    prepare_test_cases()

7.2 创建结果分析脚本

测试完成后,我们需要分析结果。创建一个分析脚本:

#!/usr/bin/env python3
# analyze_results.py - 结果分析脚本

import json
import pandas as pd
from pathlib import Path
import matplotlib.pyplot as plt
import numpy as np

def analyze_results():
    """分析测试结果"""
    
    results_dir = Path("results")
    if not results_dir.exists():
        print("错误: results目录不存在")
        return
    
    # 收集所有结果
    results = []
    
    for result_file in results_dir.glob("*.json"):
        try:
            with open(result_file, "r") as f:
                data = json.load(f)
            
            # 提取关键信息
            result = {
                "test_case": result_file.stem.replace("_response", ""),
                "status": data.get("status", "unknown"),
                "inference_time": data.get("metadata", {}).get("inference_time", 0),
                "timestamp": data.get("metadata", {}).get("timestamp", ""),
                "has_trajectory": "trajectory" in data and len(data["trajectory"]) > 0,
                "trajectory_length": len(data.get("trajectory", [])),
                "reasoning_length": len(data.get("reasoning", ""))
            }
            
            # 如果有错误信息,也记录下来
            if data.get("status") == "error":
                result["error_message"] = data.get("message", "")
            
            results.append(result)
            
        except (json.JSONDecodeError, KeyError) as e:
            print(f"警告: 无法解析文件 {result_file} - {e}")
    
    if not results:
        print("没有找到可分析的结果")
        return
    
    # 转换为DataFrame
    df = pd.DataFrame(results)
    
    print("=== 测试结果分析 ===")
    print(f"总测试数: {len(df)}")
    print(f"成功数: {len(df[df['status'] == 'success'])}")
    print(f"失败数: {len(df[df['status'] == 'error'])}")
    print(f"成功率: {len(df[df['status'] == 'success']) / len(df) * 100:.1f}%")
    print()
    
    # 统计推理时间
    if 'inference_time' in df.columns:
        success_df = df[df['status'] == 'success']
        if len(success_df) > 0:
            print("推理时间统计 (成功案例):")
            print(f"  平均时间: {success_df['inference_time'].mean():.2f}秒")
            print(f"  最短时间: {success_df['inference_time'].min():.2f}秒")
            print(f"  最长时间: {success_df['inference_time'].max():.2f}秒")
            print(f"  标准差: {success_df['inference_time'].std():.2f}秒")
            print()
    
    # 轨迹数据统计
    print("轨迹数据统计:")
    print(f"  有轨迹数据的测试: {len(df[df['has_trajectory']])}")
    print(f"  平均轨迹长度: {df['trajectory_length'].mean():.1f}个点")
    print(f"  平均推理文本长度: {df['reasoning_length'].mean():.1f}字符")
    print()
    
    # 错误分析
    error_df = df[df['status'] == 'error']
    if len(error_df) > 0:
        print("错误分析:")
        for _, row in error_df.iterrows():
            print(f"  {row['test_case']}: {row.get('error_message', '未知错误')}")
        print()
    
    # 生成报告
    generate_report(df)
    
    # 生成可视化图表
    generate_visualizations(df)

def generate_report(df):
    """生成详细报告"""
    
    report_file = "test_report.md"
    
    with open(report_file, "w") as f:
        f.write("# Alpamayo-R1-10B 自动化测试报告\n\n")
        f.write(f"**生成时间**: {pd.Timestamp.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n")
        
        f.write("## 测试概览\n\n")
        f.write(f"- 总测试数: {len(df)}\n")
        f.write(f"- 成功数: {len(df[df['status'] == 'success'])}\n")
        f.write(f"- 失败数: {len(df[df['status'] == 'error'])}\n")
        f.write(f"- 成功率: {len(df[df['status'] == 'success']) / len(df) * 100:.1f}%\n\n")
        
        f.write("## 性能统计\n\n")
        success_df = df[df['status'] == 'success']
        if len(success_df) > 0:
            f.write("### 推理时间\n\n")
            f.write(f"- 平均时间: {success_df['inference_time'].mean():.2f}秒\n")
            f.write(f"- 最短时间: {success_df['inference_time'].min():.2f}秒\n")
            f.write(f"- 最长时间: {success_df['inference_time'].max():.2f}秒\n")
            f.write(f"- 标准差: {success_df['inference_time'].std():.2f}秒\n\n")
        
        f.write("## 详细结果\n\n")
        f.write("| 测试用例 | 状态 | 推理时间(秒) | 轨迹长度 | 推理文本长度 |\n")
        f.write("|---------|------|-------------|----------|-------------|\n")
        
        for _, row in df.iterrows():
            status_emoji = "✅" if row['status'] == 'success' else "❌"
            f.write(f"| {row['test_case']} | {status_emoji} {row['status']} | {row.get('inference_time', 'N/A'):.2f} | {row['trajectory_length']} | {row['reasoning_length']} |\n")
        
        f.write("\n## 错误详情\n\n")
        error_df = df[df['status'] == 'error']
        if len(error_df) > 0:
            for _, row in error_df.iterrows():
                f.write(f"### {row['test_case']}\n")
                f.write(f"- 错误信息: {row.get('error_message', '未知错误')}\n\n")
        else:
            f.write("无错误\n")
    
    print(f"详细报告已生成: {report_file}")

def generate_visualizations(df):
    """生成可视化图表"""
    
    # 只处理成功案例
    success_df = df[df['status'] == 'success']
    if len(success_df) <= 1:
        print("成功案例不足,跳过图表生成")
        return
    
    # 创建图表目录
    plots_dir = Path("plots")
    plots_dir.mkdir(exist_ok=True)
    
    # 1. 推理时间分布图
    plt.figure(figsize=(10, 6))
    plt.bar(range(len(success_df)), success_df['inference_time'].values)
    plt.xlabel('测试用例')
    plt.ylabel('推理时间 (秒)')
    plt.title('Alpamayo-R1-10B 推理时间分布')
    plt.xticks(range(len(success_df)), success_df['test_case'].values, rotation=45, ha='right')
    plt.tight_layout()
    plt.savefig(plots_dir / 'inference_time_distribution.png', dpi=150)
    plt.close()
    
    # 2. 轨迹长度分布图
    plt.figure(figsize=(10, 6))
    plt.bar(range(len(success_df)), success_df['trajectory_length'].values)
    plt.xlabel('测试用例')
    plt.ylabel('轨迹长度 (点数)')
    plt.title('轨迹长度分布')
    plt.xticks(range(len(success_df)), success_df['test_case'].values, rotation=45, ha='right')
    plt.tight_layout()
    plt.savefig(plots_dir / 'trajectory_length_distribution.png', dpi=150)
    plt.close()
    
    # 3. 推理文本长度分布图
    plt.figure(figsize=(10, 6))
    plt.bar(range(len(success_df)), success_df['reasoning_length'].values)
    plt.xlabel('测试用例')
    plt.ylabel('推理文本长度 (字符数)')
    plt.title('推理文本长度分布')
    plt.xticks(range(len(success_df)), success_df['test_case'].values, rotation=45, ha='right')
    plt.tight_layout()
    plt.savefig(plots_dir / 'reasoning_length_distribution.png', dpi=150)
    plt.close()
    
    print(f"图表已生成到: {plots_dir}/")

if __name__ == "__main__":
    analyze_results()

7.3 创建主控制脚本

最后,创建一个主控制脚本,把整个流程串起来:

#!/bin/bash

# run_automated_test.sh - 自动化测试主脚本

echo "=== Alpamayo-R1-10B 自动化测试流程 ==="
echo "开始时间: $(date)"
echo ""

# 步骤1: 准备测试数据
echo "步骤1: 准备测试数据..."
python3 prepare_test_data.py
if [ $? -ne 0 ]; then
    echo "❌ 测试数据准备失败"
    exit 1
fi
echo "✅ 测试数据准备完成"
echo ""

# 步骤2: 检查API服务
echo "步骤2: 检查API服务状态..."
curl -s -o /dev/null -w "%{http_code}" http://localhost:8000/health > /tmp/health_check.txt
HTTP_STATUS=$(cat /tmp/health_check.txt)

if [ "$HTTP_STATUS" = "200" ]; then
    echo "✅ API服务正常运行"
else
    echo "⚠️  API服务可能未启动,尝试启动..."
    supervisorctl start alpamayo-r1
    sleep 5
    
    # 再次检查
    curl -s -o /dev/null -w "%{http_code}" http://localhost:8000/health > /tmp/health_check.txt
    HTTP_STATUS=$(cat /tmp/health_check.txt)
    
    if [ "$HTTP_STATUS" = "200" ]; then
        echo "✅ API服务启动成功"
    else
        echo "❌ API服务启动失败,请手动检查"
        exit 1
    fi
fi
echo ""

# 步骤3: 运行批量测试
echo "步骤3: 运行批量测试..."
./batch_test.sh
if [ $? -ne 0 ]; then
    echo "⚠️  批量测试执行过程中可能出现问题,继续执行分析..."
fi
echo ""

# 步骤4: 分析测试结果
echo "步骤4: 分析测试结果..."
python3 analyze_results.py
if [ $? -ne 0 ]; then
    echo "❌ 结果分析失败"
    exit 1
fi
echo ""

# 步骤5: 生成总结报告
echo "步骤5: 生成测试总结..."
echo ""
echo "=== 测试流程完成 ==="
echo "完成时间: $(date)"
echo ""
echo "生成的文件:"
echo "1. 测试用例: test_cases/ 目录"
echo "2. 测试结果: results/ 目录"
echo "3. 测试日志: batch_test_*.log"
echo "4. 分析报告: test_report.md"
echo "5. 可视化图表: plots/ 目录"
echo ""
echo "下一步建议:"
echo "1. 查看 test_report.md 了解详细结果"
echo "2. 检查 plots/ 目录中的图表"
echo "3. 根据结果调整测试用例或模型参数"
echo "4. 可以修改 prepare_test_data.py 添加更多测试场景"

8. 常见问题与解决方案

在实际使用curl调用API的过程中,你可能会遇到一些问题。这里我总结了一些常见问题及其解决方案。

8.1 连接问题

问题:连接被拒绝或超时

# 错误信息示例
curl: (7) Failed to connect to localhost port 8000: Connection refused
# 或
curl: (28) Connection timed out after 10001 milliseconds

解决方案:

  1. 检查API服务是否运行

    supervisorctl status alpamayo-r1
    
  2. 如果服务未运行,启动它

    supervisorctl start alpamayo-r1
    
  3. 检查端口是否被占用

    netstat -tlnp | grep 8000
    
  4. 检查防火墙设置

    # 查看防火墙状态
    sudo ufw status
    
    # 如果防火墙启用,添加端口规则
    sudo ufw allow 8000/tcp
    
  5. 如果使用远程服务器,检查网络连接

    # 测试网络连通性
    ping 服务器IP
    
    # 测试端口连通性
    telnet 服务器IP 8000
    # 或
    nc -zv 服务器IP 8000
    

8.2 内存不足问题

问题:GPU显存不足

# API可能返回的错误
{
  "status": "error",
  "message": "GPU out of memory",
  "error_code": "GPU_OUT_OF_MEMORY"
}

解决方案:

  1. 检查当前GPU使用情况

    nvidia-smi
    
  2. 释放未使用的GPU内存

    # 停止不必要的服务
    supervisorctl stop alpamayo-webui
    
    # 等待几秒让内存释放
    sleep 5
    
    # 重新检查
    nvidia-smi
    
  3. 减少批量请求的大小

    • 确保每次只发送一个请求
    • 在请求之间添加延迟
    # 在批量测试脚本中添加延迟
    sleep 2  # 等待2秒再发送下一个请求
    
  4. 调整模型参数

    • 减少num_samples参数值
    • 使用更小的图像分辨率(如果支持)

8.3 图像格式问题

问题:图像Base64编码错误

# API可能返回的错误
{
  "status": "error", 
  "message": "Invalid image data",
  "error_code": "INVALID_IMAGE"
}

解决方案:

  1. 检查Base64编码格式

    # 查看Base64字符串的前100个字符
    head -c 100 test_request.json
    
    # 应该以"data:image/jpeg;base64,"或"data:image/png;base64,"开头
    
  2. 验证Base64编码是否正确

    import base64
    import json
    
    # 读取请求文件
    with open("test_request.json", "r") as f:
        data = json.load(f)
    
    # 检查前视摄像头图像
    front_data = data["front_camera"]
    if not front_data.startswith("data:image/"):
        print("错误: Base64数据缺少Data URL前缀")
    
    # 提取Base64部分
    if front_data.startswith("data:image/jpeg;base64,"):
        base64_str = front_data.replace("data:image/jpeg;base64,", "")
    elif front_data.startswith("data:image/png;base64,"):
        base64_str = front_data.replace("data:image/png;base64,", "")
    else:
        print("错误: 不支持的图像格式")
    
    # 尝试解码
    try:
        decoded = base64.b64decode(base64_str)
        print(f"Base64解码成功,数据大小: {len(decoded)}字节")
    except Exception as e:
        print(f"Base64解码失败: {e}")
    
  3. 重新生成Base64编码

    # 使用正确的格式重新编码
    def encode_image_correctly(image_path):
        with open(image_path, "rb") as f:
            image_data = f.read()
        
        base64_str = base64.b64encode(image_data).decode('utf-8')
        
        # 根据图像类型添加正确的前缀
        if image_path.lower().endswith('.jpg') or image_path.lower().endswith('.jpeg'):
            return f"data:image/jpeg;base64,{base64_str}"
        elif image_path.lower().endswith('.png'):
            return f"data:image/png;base64,{base64_str}"
        else:
            raise ValueError(f"不支持的图像格式: {image_path}")
    

8.4 性能优化建议

问题:API响应慢或超时

解决方案:

  1. 优化curl命令参数

    # 增加超时时间
    curl --connect-timeout 30 --max-time 120 ...
    
    # 启用HTTP持久连接
    curl --keepalive-time 30 --max-time 120 ...
    
    # 使用压缩传输(如果服务器支持)
    curl --compressed ...
    
  2. 批量请求优化

    # 使用并行请求(谨慎使用,可能增加服务器负载)
    # 安装GNU parallel
    # Ubuntu/Debian: sudo apt-get install parallel
    
    # 并行执行多个请求
    cat request_list.txt | parallel -j 4 curl -X POST http://localhost:8000/predict \
      -H "Content-Type: application/json" \
      -d {} \
      --connect-timeout 30 \
      --max-time 120 \
      -o response_{#}.json
    
  3. 服务器端优化

    # 监控服务器资源使用
    top  # 查看CPU和内存使用
    nvidia-smi -l 1  # 每秒查看一次GPU使用情况
    
    # 调整API服务配置
    # 编辑 /etc/supervisor/conf.d/alpamayo-r1.conf
    # 增加工作进程数(如果支持)
    numprocs=2
    process_name=alpamayo-r1_%(process_num)s
    
  4. 请求数据优化

    • 压缩图像大小(在保持质量的前提下)
    • 减少不必要的元数据
    • 使用更简洁的JSON结构

8.5 日志和调试

问题:不知道请求为什么失败

解决方案:启用详细日志

  1. curl详细输出

    # 使用-v参数查看详细输出
    curl -v -X POST "http://localhost:8000/predict" ...
    
    # 保存详细日志到文件
    curl -v -X POST "http://localhost:8000/predict" ... 2> curl_debug.log
    
  2. 服务器端日志

    # 查看API服务日志
    tail -f /root/Alpamayo-R1-10B/logs/webui_stderr.log
    
    # 查看系统日志
    sudo journalctl -u supervisor -f
    
  3. 网络抓包(高级调试)

    # 安装tcpdump
    # Ubuntu/Debian: sudo apt-get install tcpdump
    
    # 捕获API流量
    sudo tcpdump -i lo port 8000 -w api_traffic.pcap
    
    # 在另一个终端执行curl命令
    # 然后停止tcpdump (Ctrl+C)
    
    # 使用Wireshark分析
    # 或使用tcpdump读取
    tcpdump -r api_traffic.pcap -A | grep -A 10 -B 10 "predict"
    

9. 总结

通过本教程,你已经掌握了使用curl命令调用Alpamayo-R1-10B的/predict API实现自动化轨迹测试的完整流程。让我们回顾一下关键要点:

9.1 核心技能掌握

  1. API服务管理:学会了如何启动、停止和监控Alpamayo-R1的API服务
  2. curl命令使用:掌握了各种curl参数和技巧,从基础调用到高级调试
  3. 数据准备:了解了如何准备测试数据,特别是图像Base64编码的处理
  4. 脚本编写:学会了编写自动化测试脚本,包括单个测试、批量测试和参数扫描
  5. 结果分析:掌握了如何分析测试结果,生成报告和可视化图表
  6. 问题排查:学会了诊断和解决常见的连接、内存、性能问题

9.2 实际应用价值

通过自动化测试,你可以:

  1. 大幅提升测试效率:从手动测试的几分钟一个场景,到自动化测试的几秒钟一个场景
  2. 确保测试一致性:自动化测试消除了人为操作差异,确保每次测试条件相同
  3. 支持大规模测试:轻松测试数百甚至数千个场景,全面评估模型性能
  4. 集成到开发流程:将API测试集成到CI/CD流程,实现自动化质量保障
  5. 参数优化研究:系统性地研究不同参数对模型性能的影响

9.3 下一步建议

根据你的具体需求,可以考虑以下方向:

  1. 扩展测试场景:创建更多样化的测试用例,覆盖不同的天气、光照、交通条件
  2. 性能基准测试:建立性能基准,监控模型推理时间的变化趋势
  3. 集成到现有系统:将API调用集成到你的自动驾驶测试框架中
  4. 开发可视化工具:创建Web界面或桌面应用,更方便地管理和查看测试结果
  5. 探索高级功能:研究模型的其他功能,如多模态输入、实时推理等

9.4 资源推荐

  1. curl官方文档:https://curl.se/docs/
  2. jq工具教程:https://stedolan.github.io/jq/tutorial/
  3. Python requests库:如果觉得curl不够方便,可以考虑使用Python的requests库
  4. 自动化测试框架:如pytest、Robot Framework等,可以构建更复杂的测试套件

记住,自动化测试不是一次性的工作,而是一个持续改进的过程。随着测试用例的积累和测试流程的优化,你会越来越了解模型的性能特点,也能更早地发现潜在问题。


获取更多AI镜像

想探索更多AI镜像和应用场景?访问 CSDN星图镜像广场,提供丰富的预置镜像,覆盖大模型推理、图像生成、视频生成、模型微调等多个领域,支持一键部署。

Logo

腾讯云面向开发者汇聚海量精品云计算使用和开发经验,营造开放的云计算技术生态圈。

更多推荐