Alpamayo-R1-10B开发者指南：Python调用API服务实现批量轨迹预测

1. 从WebUI到API：为什么需要批量预测？

如果你已经体验过Alpamayo-R1-10B的WebUI界面，可能会发现一个问题：每次只能处理一组图像和指令，然后等待模型推理，再手动点击下一次。这在开发自动驾驶算法、测试不同场景、或者需要处理大量数据时，效率实在太低了。

想象一下，你手头有1000个驾驶场景需要测试，每个场景包含前视、左侧、右侧三个摄像头的图像，还有对应的驾驶指令。如果用WebUI手动操作，可能需要好几天时间。这就是为什么我们需要通过Python调用API服务来实现批量轨迹预测。

批量预测能帮你做什么？简单来说就是三件事：

• 效率提升：一次处理成百上千个场景，而不是一个一个来
• 自动化流程：把预测过程集成到你的开发流水线中
• 数据驱动：基于大量预测结果进行统计分析，优化算法

2. 准备工作：确保API服务正常运行

在开始编写Python代码之前，我们需要先确认API服务已经启动并运行正常。根据你提供的文档，Alpamayo-R1-10B默认只开启了WebUI服务，API服务是禁用的。所以第一步就是启用它。

2.1 启用API服务

打开终端，执行以下命令：

# 首先检查当前服务状态
supervisorctl status

# 如果看到alpamayo-r1显示STOPPED，说明API服务没启动
# 临时启动API服务
supervisorctl start alpamayo-r1

# 设置API服务开机自启（这样下次重启服务器也会自动启动）
sed -i 's/autostart=false/autostart=true/' /etc/supervisor/conf.d/alpamayo-r1.conf
sed -i 's/autorestart=false/autorestart=true/' /etc/supervisor/conf.d/alpamayo-r1.conf
supervisorctl reread && supervisorctl update

# 再次检查状态，应该看到RUNNING
supervisorctl status alpamayo-r1

2.2 验证API服务

API服务默认运行在8000端口，我们可以用curl命令测试一下：

# 测试健康检查接口
curl http://localhost:8000/health

# 如果返回类似下面的JSON，说明服务正常
# {"status":"healthy","model_loaded":true}

如果遇到端口冲突或者想修改端口，可以编辑配置文件：

# 查看当前配置
cat /etc/supervisor/conf.d/alpamayo-r1.conf | grep PORT

# 如果需要修改端口，编辑配置文件
vi /etc/supervisor/conf.d/alpamayo-r1.conf
# 找到API_PORT="8000"这一行，修改为你想要的端口
# 然后重启服务
supervisorctl restart alpamayo-r1

3. Python调用API的基础方法

现在API服务已经运行起来了，我们来写第一个Python脚本。我会从最简单的单次预测开始，逐步扩展到批量处理。

3.1 安装必要的Python库

首先确保你的Python环境中有这些库：

pip install requests pillow numpy matplotlib

如果你还没有安装，可以用上面的命令安装。这些库的作用分别是：

• requests：用来发送HTTP请求到API
• pillow：处理图像文件
• numpy：处理数值数据
• matplotlib：可视化轨迹结果

3.2 基础的单次预测脚本

创建一个名为single_prediction.py的文件：

import requests
import json
import base64
from PIL import Image
import io

class AlpamayoPredictor:
    def __init__(self, api_url="http://localhost:8000"):
        """
        初始化预测器
        :param api_url: API服务的地址，默认是localhost:8000
        """
        self.api_url = api_url
        self.predict_endpoint = f"{api_url}/predict"
        
    def image_to_base64(self, image_path):
        """
        将图片文件转换为base64编码的字符串
        :param image_path: 图片文件路径
        :return: base64编码的字符串
        """
        with open(image_path, "rb") as image_file:
            # 读取图片并转换为base64
            encoded_string = base64.b64encode(image_file.read()).decode('utf-8')
            # 获取图片格式（jpg, png等）
            img_format = image_path.split('.')[-1].upper()
            if img_format == 'JPG':
                img_format = 'JPEG'
            # 构建data URI格式
            return f"data:image/{img_format.lower()};base64,{encoded_string}"
    
    def predict_single(self, front_image_path, left_image_path, right_image_path, prompt):
        """
        单次预测：输入三张图片和一个指令，获取轨迹预测结果
        :param front_image_path: 前视摄像头图片路径
        :param left_image_path: 左侧摄像头图片路径
        :param right_image_path: 右侧摄像头图片路径
        :param prompt: 驾驶指令文本
        :return: 预测结果字典
        """
        # 准备请求数据
        payload = {
            "front_image": self.image_to_base64(front_image_path),
            "left_image": self.image_to_base64(left_image_path),
            "right_image": self.image_to_base64(right_image_path),
            "prompt": prompt,
            "top_p": 0.98,      # 核采样概率
            "temperature": 0.6, # 采样温度
            "num_samples": 1    # 轨迹采样数量
        }
        
        # 发送POST请求
        try:
            response = requests.post(
                self.predict_endpoint,
                json=payload,
                timeout=60  # 设置60秒超时
            )
            
            # 检查响应状态
            if response.status_code == 200:
                result = response.json()
                return result
            else:
                print(f"请求失败，状态码: {response.status_code}")
                print(f"错误信息: {response.text}")
                return None
                
        except requests.exceptions.RequestException as e:
            print(f"请求异常: {e}")
            return None
    
    def save_result(self, result, output_file="prediction_result.json"):
        """
        保存预测结果到JSON文件
        :param result: 预测结果字典
        :param output_file: 输出文件名
        """
        if result:
            with open(output_file, 'w', encoding='utf-8') as f:
                json.dump(result, f, indent=2, ensure_ascii=False)
            print(f"结果已保存到: {output_file}")
    
    def print_result_summary(self, result):
        """
        打印预测结果的摘要信息
        :param result: 预测结果字典
        """
        if not result:
            print("没有有效的结果")
            return
            
        print("\n=== 预测结果摘要 ===")
        print(f"推理状态: {result.get('status', 'unknown')}")
        
        # 打印推理过程
        reasoning = result.get('reasoning', '')
        if reasoning:
            print(f"\n推理过程:")
            print(f"{reasoning[:200]}...")  # 只打印前200个字符
        
        # 打印轨迹信息
        trajectory = result.get('trajectory', [])
        if trajectory:
            print(f"\n轨迹点数: {len(trajectory)}")
            if len(trajectory) > 0:
                print(f"第一个轨迹点: {trajectory[0]}")
                print(f"最后一个轨迹点: {trajectory[-1]}")
        
        # 如果有可视化图像
        if 'visualization' in result:
            print(f"\n轨迹可视化已生成")

# 使用示例
if __name__ == "__main__":
    # 创建预测器实例
    predictor = AlpamayoPredictor()
    
    # 设置图片路径和指令
    # 注意：这里需要替换为你的实际图片路径
    front_img = "path/to/front_camera.jpg"
    left_img = "path/to/left_camera.jpg"
    right_img = "path/to/right_camera.jpg"
    driving_prompt = "Navigate through the intersection safely"
    
    print("开始单次轨迹预测...")
    print(f"前视摄像头: {front_img}")
    print(f"左侧摄像头: {left_img}")
    print(f"右侧摄像头: {right_img}")
    print(f"驾驶指令: {driving_prompt}")
    
    # 执行预测
    result = predictor.predict_single(front_img, left_img, right_img, driving_prompt)
    
    if result:
        # 打印结果摘要
        predictor.print_result_summary(result)
        
        # 保存完整结果
        predictor.save_result(result)
        
        print("\n预测完成！")
    else:
        print("预测失败，请检查API服务是否正常运行")

这个脚本做了几件重要的事情：

1. 封装了API调用逻辑：把HTTP请求的细节隐藏起来，你只需要关心输入和输出
2. 处理图片编码：自动把图片文件转换成API需要的base64格式
3. 错误处理：如果API调用失败，会给出明确的错误信息
4. 结果保存：把预测结果保存为JSON文件，方便后续分析

4. 实现高效的批量预测

单次预测只是开始，真正的价值在于批量处理。下面我们来实现一个完整的批量预测系统。

4.1 批量预测的核心思路

批量预测不是简单地把单次预测循环多次，我们需要考虑：

• 并发处理：同时发送多个请求，而不是等一个完成再发下一个
• 错误重试：某个请求失败时自动重试
• 进度显示：实时显示处理进度
• 结果管理：妥善保存每个场景的预测结果
• 资源控制：避免同时发送太多请求把服务器压垮

4.2 完整的批量预测脚本

创建一个名为batch_predictor.py的文件：

import requests
import json
import base64
import os
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime
import csv

class BatchAlpamayoPredictor:
    def __init__(self, api_url="http://localhost:8000", max_workers=3):
        """
        初始化批量预测器
        :param api_url: API服务地址
        :param max_workers: 最大并发数，根据服务器性能调整
        """
        self.api_url = api_url
        self.predict_endpoint = f"{api_url}/predict"
        self.max_workers = max_workers
        self.results = []
        self.failed_cases = []
        
    def image_to_base64(self, image_path):
        """将图片转换为base64格式"""
        if not os.path.exists(image_path):
            raise FileNotFoundError(f"图片文件不存在: {image_path}")
            
        with open(image_path, "rb") as f:
            encoded = base64.b64encode(f.read()).decode('utf-8')
            img_format = image_path.split('.')[-1].upper()
            if img_format == 'JPG':
                img_format = 'JPEG'
            return f"data:image/{img_format.lower()};base64,{encoded}"
    
    def prepare_payload(self, scene_data):
        """
        准备单个场景的请求数据
        :param scene_data: 包含图片路径和指令的字典
        :return: 请求payload
        """
        payload = {
            "front_image": self.image_to_base64(scene_data['front']),
            "left_image": self.image_to_base64(scene_data['left']),
            "right_image": self.image_to_base64(scene_data['right']),
            "prompt": scene_data['prompt'],
            "top_p": scene_data.get('top_p', 0.98),
            "temperature": scene_data.get('temperature', 0.6),
            "num_samples": scene_data.get('num_samples', 1)
        }
        return payload
    
    def predict_single_scene(self, scene_id, scene_data, max_retries=3):
        """
        预测单个场景，支持重试机制
        :param scene_id: 场景ID
        :param scene_data: 场景数据
        :param max_retries: 最大重试次数
        :return: (scene_id, 结果或错误信息)
        """
        for attempt in range(max_retries):
            try:
                payload = self.prepare_payload(scene_data)
                
                response = requests.post(
                    self.predict_endpoint,
                    json=payload,
                    timeout=120  # 批量处理时设置更长超时
                )
                
                if response.status_code == 200:
                    result = response.json()
                    result['scene_id'] = scene_id
                    result['timestamp'] = datetime.now().isoformat()
                    return (scene_id, 'success', result)
                else:
                    error_msg = f"HTTP {response.status_code}: {response.text[:100]}"
                    if attempt < max_retries - 1:
                        time.sleep(2 ** attempt)  # 指数退避
                        continue
                    return (scene_id, 'failed', error_msg)
                    
            except Exception as e:
                error_msg = str(e)
                if attempt < max_retries - 1:
                    time.sleep(2 ** attempt)
                    continue
                return (scene_id, 'failed', error_msg)
        
        return (scene_id, 'failed', 'Max retries exceeded')
    
    def load_scenes_from_csv(self, csv_file):
        """
        从CSV文件加载场景数据
        CSV格式：scene_id,front_image,left_image,right_image,prompt
        """
        scenes = []
        with open(csv_file, 'r', encoding='utf-8') as f:
            reader = csv.DictReader(f)
            for row in reader:
                scene = {
                    'scene_id': row['scene_id'],
                    'front': row['front_image'],
                    'left': row['left_image'],
                    'right': row['right_image'],
                    'prompt': row['prompt']
                }
                # 可选参数
                if 'top_p' in row:
                    scene['top_p'] = float(row['top_p'])
                if 'temperature' in row:
                    scene['temperature'] = float(row['temperature'])
                scenes.append(scene)
        return scenes
    
    def load_scenes_from_folder(self, folder_path, prompt_template="Navigate safely"):
        """
        从文件夹结构加载场景数据
        文件夹结构：
        scenes/
          scene_001/
            front.jpg
            left.jpg
            right.jpg
            prompt.txt  # 可选，包含指令
          scene_002/
            ...
        """
        scenes = []
        
        for scene_dir in sorted(os.listdir(folder_path)):
            scene_path = os.path.join(folder_path, scene_dir)
            if not os.path.isdir(scene_path):
                continue
                
            # 检查必要的图片文件
            front_img = os.path.join(scene_path, "front.jpg")
            left_img = os.path.join(scene_path, "left.jpg")
            right_img = os.path.join(scene_path, "right.jpg")
            
            if not all(os.path.exists(img) for img in [front_img, left_img, right_img]):
                print(f"警告：场景 {scene_dir} 缺少必要的图片文件")
                continue
            
            # 读取指令文件，如果没有则使用模板
            prompt_file = os.path.join(scene_path, "prompt.txt")
            if os.path.exists(prompt_file):
                with open(prompt_file, 'r', encoding='utf-8') as f:
                    prompt = f.read().strip()
            else:
                prompt = prompt_template
            
            scenes.append({
                'scene_id': scene_dir,
                'front': front_img,
                'left': left_img,
                'right': right_img,
                'prompt': prompt
            })
        
        return scenes
    
    def predict_batch(self, scenes, output_dir="batch_results"):
        """
        批量预测主函数
        :param scenes: 场景列表
        :param output_dir: 输出目录
        """
        # 创建输出目录
        os.makedirs(output_dir, exist_ok=True)
        
        total_scenes = len(scenes)
        print(f"开始批量预测，共 {total_scenes} 个场景")
        print(f"并发数: {self.max_workers}")
        print(f"输出目录: {output_dir}")
        print("-" * 50)
        
        start_time = time.time()
        completed = 0
        failed = 0
        
        # 使用线程池并发处理
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            # 提交所有任务
            future_to_scene = {
                executor.submit(
                    self.predict_single_scene,
                    scene['scene_id'],
                    scene
                ): scene['scene_id']
                for scene in scenes
            }
            
            # 处理完成的任务
            for future in as_completed(future_to_scene):
                scene_id = future_to_scene[future]
                try:
                    scene_id, status, result = future.result()
                    
                    if status == 'success':
                        # 保存单个场景的结果
                        result_file = os.path.join(output_dir, f"{scene_id}_result.json")
                        with open(result_file, 'w', encoding='utf-8') as f:
                            json.dump(result, f, indent=2, ensure_ascii=False)
                        
                        self.results.append(result)
                        completed += 1
                        
                        # 打印进度
                        progress = (completed + failed) / total_scenes * 100
                        print(f"[{progress:.1f}%] 完成: {scene_id} - 推理成功")
                        
                    else:
                        self.failed_cases.append({
                            'scene_id': scene_id,
                            'error': result
                        })
                        failed += 1
                        print(f"[{progress:.1f}%] 失败: {scene_id} - {result}")
                        
                except Exception as e:
                    self.failed_cases.append({
                        'scene_id': scene_id,
                        'error': str(e)
                    })
                    failed += 1
                    print(f"异常: {scene_id} - {e}")
        
        # 保存汇总结果
        self.save_summary(output_dir, start_time, total_scenes, completed, failed)
        
        return self.results
    
    def save_summary(self, output_dir, start_time, total, completed, failed):
        """保存批量处理的汇总信息"""
        end_time = time.time()
        duration = end_time - start_time
        
        summary = {
            'batch_info': {
                'total_scenes': total,
                'completed': completed,
                'failed': failed,
                'success_rate': completed / total * 100 if total > 0 else 0,
                'start_time': datetime.fromtimestamp(start_time).isoformat(),
                'end_time': datetime.fromtimestamp(end_time).isoformat(),
                'duration_seconds': duration,
                'avg_time_per_scene': duration / completed if completed > 0 else 0
            },
            'failed_cases': self.failed_cases,
            'results_summary': []
        }
        
        # 添加每个结果的简要信息
        for result in self.results:
            summary['results_summary'].append({
                'scene_id': result.get('scene_id'),
                'status': result.get('status'),
                'timestamp': result.get('timestamp'),
                'trajectory_points': len(result.get('trajectory', [])),
                'has_visualization': 'visualization' in result
            })
        
        # 保存汇总文件
        summary_file = os.path.join(output_dir, "batch_summary.json")
        with open(summary_file, 'w', encoding='utf-8') as f:
            json.dump(summary, f, indent=2, ensure_ascii=False)
        
        # 保存失败案例
        if self.failed_cases:
            failed_file = os.path.join(output_dir, "failed_cases.json")
            with open(failed_file, 'w', encoding='utf-8') as f:
                json.dump(self.failed_cases, f, indent=2, ensure_ascii=False)
        
        # 打印最终统计
        print("\n" + "="*50)
        print("批量预测完成！")
        print(f"总场景数: {total}")
        print(f"成功: {completed}")
        print(f"失败: {failed}")
        print(f"成功率: {summary['batch_info']['success_rate']:.1f}%")
        print(f"总耗时: {duration:.1f} 秒")
        print(f"平均每个场景: {summary['batch_info']['avg_time_per_scene']:.1f} 秒")
        print(f"详细结果保存在: {output_dir}")
        print("="*50)

# 使用示例
if __name__ == "__main__":
    # 创建批量预测器
    # 注意：max_workers不要设置太大，避免压垮服务器
    predictor = BatchAlpamayoPredictor(max_workers=2)
    
    # 方法1：从CSV文件加载场景
    # scenes = predictor.load_scenes_from_csv("scenes.csv")
    
    # 方法2：从文件夹结构加载场景
    scenes = predictor.load_scenes_from_folder(
        folder_path="./scenes",
        prompt_template="Navigate through the intersection safely"
    )
    
    if not scenes:
        print("没有找到可用的场景数据")
        exit(1)
    
    print(f"加载了 {len(scenes)} 个场景")
    
    # 执行批量预测
    results = predictor.predict_batch(
        scenes=scenes,
        output_dir="./batch_results"
    )
    
    # 可选：分析结果
    if results:
        print(f"\n成功预测了 {len(results)} 个场景")
        print("前3个场景的推理状态:")
        for i, result in enumerate(results[:3]):
            print(f"{i+1}. {result.get('scene_id')}: {result.get('status')}")

这个批量预测脚本提供了完整的功能：

1. 多种数据加载方式：支持从CSV文件或文件夹结构加载场景数据
2. 并发处理：使用线程池同时处理多个请求，大幅提升效率
3. 错误重试：自动重试失败的请求，采用指数退避策略
4. 进度显示：实时显示处理进度和成功率
5. 完整的结果管理：每个场景的结果单独保存，还有汇总报告
6. 资源控制：通过max_workers控制并发数，避免服务器过载

5. 高级功能：结果分析与可视化

批量预测完成后，我们还需要对结果进行分析和可视化。下面是一些实用的分析工具。

5.1 轨迹数据分析脚本

创建一个名为analyze_results.py的文件：

import json
import os
import numpy as np
import matplotlib.pyplot as plt
from matplotlib.patches import Rectangle, Circle
import seaborn as sns
from collections import defaultdict

class TrajectoryAnalyzer:
    def __init__(self, results_dir):
        """
        初始化轨迹分析器
        :param results_dir: 包含预测结果的目录
        """
        self.results_dir = results_dir
        self.results = []
        self.load_results()
    
    def load_results(self):
        """加载所有结果文件"""
        for filename in os.listdir(self.results_dir):
            if filename.endswith("_result.json"):
                filepath = os.path.join(self.results_dir, filename)
                try:
                    with open(filepath, 'r', encoding='utf-8') as f:
                        result = json.load(f)
                        self.results.append(result)
                except Exception as e:
                    print(f"加载文件失败 {filename}: {e}")
        
        print(f"加载了 {len(self.results)} 个预测结果")
    
    def analyze_trajectory_statistics(self):
        """分析轨迹的统计特征"""
        if not self.results:
            print("没有可分析的结果")
            return
        
        stats = {
            'total_scenes': len(self.results),
            'successful_predictions': 0,
            'failed_predictions': 0,
            'trajectory_lengths': [],
            'reasoning_lengths': [],
            'scene_types': defaultdict(int)
        }
        
        for result in self.results:
            status = result.get('status', 'unknown')
            if status == 'success':
                stats['successful_predictions'] += 1
                
                # 轨迹长度
                trajectory = result.get('trajectory', [])
                stats['trajectory_lengths'].append(len(trajectory))
                
                # 推理文本长度
                reasoning = result.get('reasoning', '')
                stats['reasoning_lengths'].append(len(reasoning))
                
                # 场景类型分析（简单基于指令）
                prompt = result.get('prompt', '').lower()
                if 'intersection' in prompt:
                    stats['scene_types']['intersection'] += 1
                elif 'lane' in prompt:
                    stats['scene_types']['lane_change'] += 1
                elif 'follow' in prompt:
                    stats['scene_types']['following'] += 1
                elif 'merge' in prompt:
                    stats['scene_types']['merging'] += 1
                else:
                    stats['scene_types']['other'] += 1
            else:
                stats['failed_predictions'] += 1
        
        # 计算平均值
        if stats['trajectory_lengths']:
            stats['avg_trajectory_length'] = np.mean(stats['trajectory_lengths'])
            stats['std_trajectory_length'] = np.std(stats['trajectory_lengths'])
        
        if stats['reasoning_lengths']:
            stats['avg_reasoning_length'] = np.mean(stats['reasoning_lengths'])
            stats['std_reasoning_length'] = np.std(stats['reasoning_lengths'])
        
        stats['success_rate'] = stats['successful_predictions'] / stats['total_scenes'] * 100
        
        return stats
    
    def plot_trajectory_distribution(self, save_path=None):
        """绘制轨迹长度分布图"""
        lengths = []
        for result in self.results:
            if result.get('status') == 'success':
                trajectory = result.get('trajectory', [])
                lengths.append(len(trajectory))
        
        if not lengths:
            print("没有成功的轨迹数据")
            return
        
        plt.figure(figsize=(10, 6))
        plt.hist(lengths, bins=20, alpha=0.7, color='skyblue', edgecolor='black')
        plt.axvline(np.mean(lengths), color='red', linestyle='--', 
                   label=f'平均长度: {np.mean(lengths):.1f}')
        plt.xlabel('轨迹点数')
        plt.ylabel('场景数量')
        plt.title('轨迹长度分布')
        plt.legend()
        plt.grid(True, alpha=0.3)
        
        if save_path:
            plt.savefig(save_path, dpi=300, bbox_inches='tight')
            print(f"分布图已保存到: {save_path}")
        plt.show()
    
    def visualize_sample_trajectories(self, num_samples=5, save_dir=None):
        """可视化样本轨迹"""
        successful_results = [r for r in self.results if r.get('status') == 'success']
        
        if not successful_results:
            print("没有成功的轨迹数据")
            return
        
        # 随机选择样本
        import random
        samples = random.sample(successful_results, min(num_samples, len(successful_results)))
        
        fig, axes = plt.subplots(1, len(samples), figsize=(5*len(samples), 5))
        if len(samples) == 1:
            axes = [axes]
        
        for idx, (ax, result) in enumerate(zip(axes, samples)):
            trajectory = result.get('trajectory', [])
            scene_id = result.get('scene_id', f'scene_{idx}')
            prompt = result.get('prompt', '')[:50] + '...' if len(result.get('prompt', '')) > 50 else result.get('prompt', '')
            
            if trajectory:
                # 提取x, y坐标（假设轨迹点是[x, y, z]格式）
                x_coords = [point[0] for point in trajectory if len(point) >= 2]
                y_coords = [point[1] for point in trajectory if len(point) >= 2]
                
                # 绘制轨迹
                ax.plot(x_coords, y_coords, 'b-', linewidth=2, label='预测轨迹')
                ax.plot(x_coords, y_coords, 'ro', markersize=3, alpha=0.5)
                
                # 标记起点和终点
                if x_coords and y_coords:
                    ax.plot(x_coords[0], y_coords[0], 'go', markersize=8, label='起点')
                    ax.plot(x_coords[-1], y_coords[-1], 'rs', markersize=8, label='终点')
                
                ax.set_xlabel('X坐标')
                ax.set_ylabel('Y坐标')
                ax.set_title(f'{scene_id}\n{prompt}')
                ax.legend()
                ax.grid(True, alpha=0.3)
                ax.axis('equal')
            else:
                ax.text(0.5, 0.5, '无轨迹数据', 
                       ha='center', va='center', transform=ax.transAxes)
                ax.set_title(f'{scene_id}')
        
        plt.tight_layout()
        
        if save_dir:
            os.makedirs(save_dir, exist_ok=True)
            save_path = os.path.join(save_dir, 'sample_trajectories.png')
            plt.savefig(save_path, dpi=300, bbox_inches='tight')
            print(f"样本轨迹图已保存到: {save_path}")
        plt.show()
    
    def generate_report(self, output_file="analysis_report.md"):
        """生成分析报告"""
        stats = self.analyze_trajectory_statistics()
        
        report = f"""# Alpamayo-R1 批量预测分析报告

## 总体统计
- **总场景数**: {stats['total_scenes']}
- **成功预测**: {stats['successful_predictions']}
- **失败预测**: {stats['failed_predictions']}
- **成功率**: {stats.get('success_rate', 0):.1f}%

## 轨迹统计
- **平均轨迹点数**: {stats.get('avg_trajectory_length', 0):.1f}
- **轨迹点标准差**: {stats.get('std_trajectory_length', 0):.1f}
- **最短轨迹**: {min(stats['trajectory_lengths']) if stats['trajectory_lengths'] else 0}
- **最长轨迹**: {max(stats['trajectory_lengths']) if stats['trajectory_lengths'] else 0}

## 推理文本统计
- **平均推理长度**: {stats.get('avg_reasoning_length', 0):.1f} 字符
- **推理长度标准差**: {stats.get('std_reasoning_length', 0):.1f}

## 场景类型分布
"""
        
        for scene_type, count in stats['scene_types'].items():
            percentage = count / stats['successful_predictions'] * 100 if stats['successful_predictions'] > 0 else 0
            report += f"- **{scene_type}**: {count} ({percentage:.1f}%)\n"
        
        # 添加建议
        report += """
## 建议与观察

1. **成功率分析**: 如果成功率低于预期，检查失败案例的日志
2. **轨迹长度**: 关注异常短的轨迹（可能表示模型不确定）
3. **场景分布**: 确保测试场景覆盖各种驾驶情况
4. **推理质量**: 手动检查推理文本的逻辑性和合理性
"""
        
        with open(output_file, 'w', encoding='utf-8') as f:
            f.write(report)
        
        print(f"分析报告已生成: {output_file}")
        return report

# 使用示例
if __name__ == "__main__":
    # 分析批量预测结果
    analyzer = TrajectoryAnalyzer("./batch_results")
    
    # 生成统计报告
    stats = analyzer.analyze_trajectory_statistics()
    print("统计结果:")
    for key, value in stats.items():
        if isinstance(value, (int, float)):
            print(f"{key}: {value}")
    
    # 绘制轨迹分布图
    analyzer.plot_trajectory_distribution(save_path="./analysis/trajectory_distribution.png")
    
    # 可视化样本轨迹
    analyzer.visualize_sample_trajectories(
        num_samples=3,
        save_dir="./analysis"
    )
    
    # 生成详细报告
    analyzer.generate_report("./analysis/report.md")

5.2 批量结果对比脚本

如果你有不同参数或不同模型的预测结果，可以进行比较分析：

import json
import os
import numpy as np
import matplotlib.pyplot as plt

class ResultComparator:
    def __init__(self, result_dirs, labels=None):
        """
        初始化结果比较器
        :param result_dirs: 不同实验结果的目录列表
        :param labels: 对应的标签列表
        """
        self.result_dirs = result_dirs
        self.labels = labels or [f"实验{i+1}" for i in range(len(result_dirs))]
        self.all_results = []
        
        # 加载所有结果
        for dir_path in result_dirs:
            results = []
            for filename in os.listdir(dir_path):
                if filename.endswith("_result.json"):
                    filepath = os.path.join(dir_path, filename)
                    try:
                        with open(filepath, 'r', encoding='utf-8') as f:
                            results.append(json.load(f))
                    except:
                        pass
            self.all_results.append(results)
    
    def compare_success_rates(self):
        """比较成功率"""
        success_rates = []
        for results in self.all_results:
            if not results:
                success_rates.append(0)
                continue
            
            successful = sum(1 for r in results if r.get('status') == 'success')
            success_rate = successful / len(results) * 100
            success_rates.append(success_rate)
        
        return success_rates
    
    def compare_trajectory_consistency(self):
        """比较轨迹一致性（相同场景不同实验的轨迹差异）"""
        # 这里需要相同场景在不同实验中的结果
        # 实现略，根据实际需求定制
        pass
    
    def plot_comparison_chart(self, save_path=None):
        """绘制比较图表"""
        success_rates = self.compare_success_rates()
        
        plt.figure(figsize=(10, 6))
        bars = plt.bar(range(len(success_rates)), success_rates, 
                      color=['skyblue', 'lightgreen', 'lightcoral', 'gold'])
        
        plt.xlabel('实验组')
        plt.ylabel('成功率 (%)')
        plt.title('不同实验组成功率比较')
        plt.xticks(range(len(success_rates)), self.labels)
        plt.ylim(0, 100)
        
        # 在柱子上显示数值
        for bar, rate in zip(bars, success_rates):
            height = bar.get_height()
            plt.text(bar.get_x() + bar.get_width()/2., height + 1,
                    f'{rate:.1f}%', ha='center', va='bottom')
        
        plt.grid(True, alpha=0.3, axis='y')
        
        if save_path:
            plt.savefig(save_path, dpi=300, bbox_inches='tight')
        plt.show()

# 使用示例
if __name__ == "__main__":
    # 比较不同参数设置的结果
    comparator = ResultComparator(
        result_dirs=[
            "./results_default_params",
            "./results_low_temp",
            "./results_high_top_p"
        ],
        labels=["默认参数", "低温(0.3)", "高Top-p(0.99)"]
    )
    
    success_rates = comparator.compare_success_rates()
    print("成功率比较:")
    for label, rate in zip(comparator.labels, success_rates):
        print(f"{label}: {rate:.1f}%")
    
    comparator.plot_comparison_chart(save_path="./comparison/success_rate.png")

6. 实战技巧与最佳实践

在实际使用中，有几个技巧能让你的批量预测更加高效可靠：

6.1 优化并发数

# 根据服务器性能调整并发数
# 一般建议从2开始，逐步增加
def find_optimal_workers(api_url, test_scenes, max_test=5):
    """测试找到最优的并发数"""
    best_workers = 2
    best_time = float('inf')
    
    for workers in [2, 3, 4, 5]:
        if workers > max_test:
            break
            
        predictor = BatchAlpamayoPredictor(api_url, max_workers=workers)
        
        start_time = time.time()
        results = predictor.predict_batch(test_scenes[:10])  # 用少量场景测试
        end_time = time.time()
        
        duration = end_time - start_time
        success_rate = len([r for r in results if r]) / len(test_scenes[:10]) * 100
        
        print(f"并发数 {workers}: 耗时 {duration:.1f}s, 成功率 {success_rate:.1f}%")
        
        if duration < best_time and success_rate > 90:
            best_time = duration
            best_workers = workers
    
    print(f"\n推荐并发数: {best_workers}")
    return best_workers

6.2 处理大文件时的内存优化

def process_large_dataset_in_chunks(dataset_path, chunk_size=100, output_dir="results"):
    """分块处理大型数据集，避免内存溢出"""
    import pandas as pd
    
    # 读取整个数据集
    df = pd.read_csv(dataset_path)
    total_rows = len(df)
    
    # 分块处理
    for chunk_start in range(0, total_rows, chunk_size):
        chunk_end = min(chunk_start + chunk_size, total_rows)
        chunk = df.iloc[chunk_start:chunk_end]
        
        print(f"处理块 {chunk_start//chunk_size + 1}/{(total_rows-1)//chunk_size + 1}")
        
        # 转换为场景列表
        scenes = []
        for _, row in chunk.iterrows():
            scenes.append({
                'scene_id': row['scene_id'],
                'front': row['front_image'],
                'left': row['left_image'],
                'right': row['right_image'],
                'prompt': row['prompt']
            })
        
        # 创建分块输出目录
        chunk_dir = os.path.join(output_dir, f"chunk_{chunk_start:04d}")
        os.makedirs(chunk_dir, exist_ok=True)
        
        # 处理当前块
        predictor = BatchAlpamayoPredictor(max_workers=3)
        predictor.predict_batch(scenes, chunk_dir)
        
        # 可选：清理内存
        del scenes
        import gc
        gc.collect()

6.3 监控与日志记录

import logging
from datetime import datetime

def setup_logging(log_dir="logs"):
    """设置详细的日志记录"""
    os.makedirs(log_dir, exist_ok=True)
    
    log_file = os.path.join(log_dir, f"batch_predict_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log")
    
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
        handlers=[
            logging.FileHandler(log_file),
            logging.StreamHandler()  # 同时输出到控制台
        ]
    )
    
    return logging.getLogger(__name__)

# 在批量预测中使用
logger = setup_logging()

def predict_with_logging(scene_data):
    """带日志记录的预测函数"""
    try:
        logger.info(f"开始处理场景: {scene_data['scene_id']}")
        # ... 预测逻辑 ...
        logger.info(f"场景 {scene_data['scene_id']} 处理成功")
        return result
    except Exception as e:
        logger.error(f"场景 {scene_data['scene_id']} 处理失败: {e}")
        raise

7. 总结

通过Python调用Alpamayo-R1-10B的API服务实现批量轨迹预测，你可以：

1. 大幅提升效率：从手动逐个处理到自动化批量处理
2. 集成到开发流程：将预测作为算法开发的一部分
3. 进行大规模测试：轻松测试成百上千个场景
4. 数据驱动优化：基于大量预测结果优化算法

关键要点回顾：

1. API服务启用：记得先启用API服务并验证连接
2. 并发控制：根据服务器性能调整并发数，避免过载
3. 错误处理：实现重试机制，处理网络波动和服务器错误
4. 结果管理：妥善保存每个场景的结果，便于后续分析
5. 监控日志：详细的日志记录帮助调试和优化

下一步建议：

1. 从简单开始：先用少量场景测试，确保整个流程工作正常
2. 逐步扩展：成功后再增加场景数量，调整并发参数
3. 结果分析：使用提供的分析工具理解预测结果的质量和模式
4. 优化迭代：根据分析结果调整参数，优化预测效果

批量预测不仅仅是技术实现，更是一种工作方式的转变。它让你能够以数据驱动的方式开发和测试自动驾驶算法，真正发挥Alpamayo-R1-10B这类大模型的潜力。

---

获取更多AI镜像

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