Alpamayo-R1-10B实战教程:curl命令调用/predict API实现自动化轨迹测试
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-Type | application/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"
}
}
响应字段说明:
- status:请求状态,成功时为
"success" - reasoning:模型的Chain-of-Causation推理过程
- 这是Alpamayo-R1的核心特性之一
- 以文本形式解释模型为什么做出这样的决策
- 示例:
"前方有行人正在过马路,需要减速让行;左侧车道有车辆,不适合变道;保持当前车道行驶是最安全的选择。"
- trajectory:预测的轨迹数据
- 一个包含64个时间步的数组
- 每个时间步是
[x, y, z, t]四个值 - x, y, z是三维坐标,t是时间戳
- 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
这个命令做了三件事:
-X POST:指定使用POST方法-H "Content-Type: application/json":设置请求头,告诉服务器我们发送的是JSON数据-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
解决方案:
-
检查API服务是否运行
supervisorctl status alpamayo-r1 -
如果服务未运行,启动它
supervisorctl start alpamayo-r1 -
检查端口是否被占用
netstat -tlnp | grep 8000 -
检查防火墙设置
# 查看防火墙状态 sudo ufw status # 如果防火墙启用,添加端口规则 sudo ufw allow 8000/tcp -
如果使用远程服务器,检查网络连接
# 测试网络连通性 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"
}
解决方案:
-
检查当前GPU使用情况
nvidia-smi -
释放未使用的GPU内存
# 停止不必要的服务 supervisorctl stop alpamayo-webui # 等待几秒让内存释放 sleep 5 # 重新检查 nvidia-smi -
减少批量请求的大小
- 确保每次只发送一个请求
- 在请求之间添加延迟
# 在批量测试脚本中添加延迟 sleep 2 # 等待2秒再发送下一个请求 -
调整模型参数
- 减少
num_samples参数值 - 使用更小的图像分辨率(如果支持)
- 减少
8.3 图像格式问题
问题:图像Base64编码错误
# API可能返回的错误
{
"status": "error",
"message": "Invalid image data",
"error_code": "INVALID_IMAGE"
}
解决方案:
-
检查Base64编码格式
# 查看Base64字符串的前100个字符 head -c 100 test_request.json # 应该以"data:image/jpeg;base64,"或"data:image/png;base64,"开头 -
验证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}") -
重新生成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响应慢或超时
解决方案:
-
优化curl命令参数
# 增加超时时间 curl --connect-timeout 30 --max-time 120 ... # 启用HTTP持久连接 curl --keepalive-time 30 --max-time 120 ... # 使用压缩传输(如果服务器支持) curl --compressed ... -
批量请求优化
# 使用并行请求(谨慎使用,可能增加服务器负载) # 安装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 -
服务器端优化
# 监控服务器资源使用 top # 查看CPU和内存使用 nvidia-smi -l 1 # 每秒查看一次GPU使用情况 # 调整API服务配置 # 编辑 /etc/supervisor/conf.d/alpamayo-r1.conf # 增加工作进程数(如果支持) numprocs=2 process_name=alpamayo-r1_%(process_num)s -
请求数据优化
- 压缩图像大小(在保持质量的前提下)
- 减少不必要的元数据
- 使用更简洁的JSON结构
8.5 日志和调试
问题:不知道请求为什么失败
解决方案:启用详细日志
-
curl详细输出
# 使用-v参数查看详细输出 curl -v -X POST "http://localhost:8000/predict" ... # 保存详细日志到文件 curl -v -X POST "http://localhost:8000/predict" ... 2> curl_debug.log -
服务器端日志
# 查看API服务日志 tail -f /root/Alpamayo-R1-10B/logs/webui_stderr.log # 查看系统日志 sudo journalctl -u supervisor -f -
网络抓包(高级调试)
# 安装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 核心技能掌握
- API服务管理:学会了如何启动、停止和监控Alpamayo-R1的API服务
- curl命令使用:掌握了各种curl参数和技巧,从基础调用到高级调试
- 数据准备:了解了如何准备测试数据,特别是图像Base64编码的处理
- 脚本编写:学会了编写自动化测试脚本,包括单个测试、批量测试和参数扫描
- 结果分析:掌握了如何分析测试结果,生成报告和可视化图表
- 问题排查:学会了诊断和解决常见的连接、内存、性能问题
9.2 实际应用价值
通过自动化测试,你可以:
- 大幅提升测试效率:从手动测试的几分钟一个场景,到自动化测试的几秒钟一个场景
- 确保测试一致性:自动化测试消除了人为操作差异,确保每次测试条件相同
- 支持大规模测试:轻松测试数百甚至数千个场景,全面评估模型性能
- 集成到开发流程:将API测试集成到CI/CD流程,实现自动化质量保障
- 参数优化研究:系统性地研究不同参数对模型性能的影响
9.3 下一步建议
根据你的具体需求,可以考虑以下方向:
- 扩展测试场景:创建更多样化的测试用例,覆盖不同的天气、光照、交通条件
- 性能基准测试:建立性能基准,监控模型推理时间的变化趋势
- 集成到现有系统:将API调用集成到你的自动驾驶测试框架中
- 开发可视化工具:创建Web界面或桌面应用,更方便地管理和查看测试结果
- 探索高级功能:研究模型的其他功能,如多模态输入、实时推理等
9.4 资源推荐
- curl官方文档:https://curl.se/docs/
- jq工具教程:https://stedolan.github.io/jq/tutorial/
- Python requests库:如果觉得curl不够方便,可以考虑使用Python的requests库
- 自动化测试框架:如pytest、Robot Framework等,可以构建更复杂的测试套件
记住,自动化测试不是一次性的工作,而是一个持续改进的过程。随着测试用例的积累和测试流程的优化,你会越来越了解模型的性能特点,也能更早地发现潜在问题。
获取更多AI镜像
想探索更多AI镜像和应用场景?访问 CSDN星图镜像广场,提供丰富的预置镜像,覆盖大模型推理、图像生成、视频生成、模型微调等多个领域,支持一键部署。
更多推荐
所有评论(0)