Wan2.2-I2V-A14B Python API调用教程：requests封装批量视频生成脚本

Wan2.2-I2V-A14B Python API调用教程requests封装批量视频生成脚本1. 环境准备与快速部署在开始使用Wan2.2-I2V-A14B模型的Python API之前我们需要确保环境已经正确部署。这个私有部署镜像已经针对RTX 4090D 24GB显存进行了深度优化开箱即用。1.1 启动API服务首先我们需要启动API服务。进入工作目录并执行以下命令cd /workspace bash start_api.sh服务启动后默认会在8000端口提供API接口。你可以通过访问http://localhost:8000/docs查看完整的API文档。1.2 验证服务状态为了确保API服务正常运行我们可以发送一个简单的GET请求进行验证import requests response requests.get(http://localhost:8000/health) print(response.json())如果服务正常运行你会看到类似这样的响应{status: healthy, version: 1.0.0}2. 基础API调用方法2.1 单次视频生成请求让我们从最基本的单次视频生成开始。以下是一个完整的Python示例import requests import json api_url http://localhost:8000/generate headers {Content-Type: application/json} payload { prompt: 一只橘猫在阳光下的窗台上打盹尾巴轻轻摆动, duration: 5, resolution: 1280x720, output_path: /workspace/output/cat_napping.mp4 } response requests.post(api_url, headersheaders, datajson.dumps(payload)) print(response.json())这个请求会返回一个JSON响应包含任务ID和状态信息{ task_id: abc123, status: queued, estimated_time: 45 }2.2 查询任务状态视频生成可能需要一些时间我们可以通过任务ID查询进度def check_task_status(task_id): status_url fhttp://localhost:8000/tasks/{task_id} response requests.get(status_url) return response.json() # 使用上面返回的task_id task_status check_task_status(abc123) print(task_status)响应可能如下{ task_id: abc123, status: completed, progress: 100, output_path: /workspace/output/cat_napping.mp4 }3. 批量视频生成脚本封装现在我们将这些基础调用封装成一个更实用的批量视频生成脚本。3.1 创建批量生成类首先创建一个Python类来封装所有API调用逻辑import requests import json import time from typing import List, Dict class VideoBatchGenerator: def __init__(self, base_urlhttp://localhost:8000): self.base_url base_url self.headers {Content-Type: application/json} def generate_video(self, prompt: str, duration: int 5, resolution: str 1280x720, output_path: str None) - Dict: 生成单个视频 if not output_path: output_path f/workspace/output/{int(time.time())}.mp4 payload { prompt: prompt, duration: duration, resolution: resolution, output_path: output_path } response requests.post( f{self.base_url}/generate, headersself.headers, datajson.dumps(payload) ) return response.json() def batch_generate(self, tasks: List[Dict], max_concurrent: int 3) - List[Dict]: 批量生成视频 results [] active_tasks [] for task in tasks: # 控制并发数量 while len(active_tasks) max_concurrent: self._check_active_tasks(active_tasks, results) time.sleep(5) # 提交新任务 task_id self.generate_video(**task)[task_id] active_tasks.append({task_id: task_id, task_data: task}) # 等待剩余任务完成 while active_tasks: self._check_active_tasks(active_tasks, results) time.sleep(5) return results def _check_active_tasks(self, active_tasks: List, results: List) - None: 检查活动任务状态 for i in range(len(active_tasks)-1, -1, -1): task active_tasks[i] status self.check_task_status(task[task_id]) if status[status] completed: results.append({ task_id: task[task_id], status: completed, output_path: status[output_path], task_data: task[task_data] }) active_tasks.pop(i) elif status[status] failed: results.append({ task_id: task[task_id], status: failed, error: status.get(error, Unknown error), task_data: task[task_data] }) active_tasks.pop(i)3.2 使用批量生成脚本现在我们可以使用这个类来批量生成视频# 初始化生成器 generator VideoBatchGenerator() # 准备批量任务 video_tasks [ { prompt: 城市夜景高楼大厦灯光闪烁车流穿梭, duration: 8, resolution: 1920x1080, output_path: /workspace/output/city_night.mp4 }, { prompt: 春天公园里樱花飘落孩子们在草地上玩耍, duration: 6, resolution: 1280x720 }, { prompt: 科幻场景未来城市悬浮汽车穿梭全息广告牌, duration: 10, resolution: 1920x1080 } ] # 执行批量生成 results generator.batch_generate(video_tasks) # 打印结果 for result in results: print(fTask {result[task_id]} - {result[status]}) if result[status] completed: print(fVideo saved at: {result[output_path]}) print(---)4. 高级功能与优化4.1 进度回调功能我们可以为批量生成脚本添加进度回调功能实时监控生成进度class VideoBatchGenerator: # ... 之前的代码 ... def batch_generate(self, tasks: List[Dict], max_concurrent: int 3, progress_callbackNone) - List[Dict]: 批量生成视频支持进度回调 results [] active_tasks [] total_tasks len(tasks) for task in tasks: # 控制并发数量 while len(active_tasks) max_concurrent: self._check_active_tasks(active_tasks, results) if progress_callback: progress_callback(len(results), total_tasks) time.sleep(5) # 提交新任务 task_id self.generate_video(**task)[task_id] active_tasks.append({task_id: task_id, task_data: task}) # 等待剩余任务完成 while active_tasks: self._check_active_tasks(active_tasks, results) if progress_callback: progress_callback(len(results), total_tasks) time.sleep(5) return results # 使用示例 def print_progress(completed, total): print(fProgress: {completed}/{total} ({completed/total*100:.1f}%)) results generator.batch_generate(video_tasks, progress_callbackprint_progress)4.2 错误处理与重试机制在实际应用中我们需要更健壮的错误处理和重试机制class VideoBatchGenerator: # ... 之前的代码 ... def generate_video(self, prompt: str, duration: int 5, resolution: str 1280x720, output_path: str None, max_retries: int 3) - Dict: 生成单个视频带重试机制 if not output_path: output_path f/workspace/output/{int(time.time())}.mp4 payload { prompt: prompt, duration: duration, resolution: resolution, output_path: output_path } for attempt in range(max_retries): try: response requests.post( f{self.base_url}/generate, headersself.headers, datajson.dumps(payload), timeout30 ) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: if attempt max_retries - 1: raise print(fAttempt {attempt 1} failed, retrying...) time.sleep(5 * (attempt 1)) def _check_active_tasks(self, active_tasks: List, results: List) - None: 检查活动任务状态带错误处理 for i in range(len(active_tasks)-1, -1, -1): task active_tasks[i] try: status self.check_task_status(task[task_id]) if status[status] completed: results.append({ task_id: task[task_id], status: completed, output_path: status[output_path], task_data: task[task_data] }) active_tasks.pop(i) elif status[status] failed: results.append({ task_id: task[task_id], status: failed, error: status.get(error, Unknown error), task_data: task[task_data] }) active_tasks.pop(i) except requests.exceptions.RequestException as e: print(fError checking task {task[task_id]}: {str(e)}) # 保持任务在活动列表中下次再检查5. 总结与最佳实践5.1 关键要点回顾通过本教程我们实现了Wan2.2-I2V-A14B API服务的基本调用方法封装了一个健壮的批量视频生成Python类添加了进度监控和错误处理机制实现了并发控制和任务管理5.2 最佳实践建议在使用批量视频生成脚本时建议遵循以下最佳实践并发控制根据你的硬件配置合理设置max_concurrent参数。对于RTX 4090D 24GB显存建议并发数不超过3个。提示词优化清晰的提示词能生成更好的视频效果。例如明确场景描述夕阳下的海滩海浪拍打岸边指定主体动作一只鹰在雪山上方盘旋描述氛围温馨的圣诞夜壁炉燃烧参数选择分辨率1080P(1920x1080)适合大多数场景720P(1280x720)速度更快时长5-10秒的视频通常效果最佳错误处理监控显存使用情况避免OOM错误实现日志记录便于排查问题对于失败任务自动重试2-3次性能优化将频繁使用的提示词模板化批量处理相似主题的视频可能利用模型缓存提高速度考虑使用异步IO进一步提高效率5.3 扩展思路这个基础脚本可以进一步扩展为集成到内容生产流水线中添加自动上传到视频平台的功能实现视频后处理添加字幕、水印等开发Web界面方便非技术人员使用获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。