数字人开发入门必看：Heygem二次开发接口文档解析

如果你正在探索数字人视频生成技术，或者想为自己的项目添加AI数字人功能，那么今天这篇文章就是为你准备的。Heygem数字人视频生成系统提供了一个强大的批量处理WebUI，但你可能不知道，它背后还隐藏着更强大的二次开发能力。

我是科哥，这个系统的开发者。在开发过程中，我设计了完整的API接口，让开发者能够将数字人视频生成能力集成到自己的应用中。今天，我就来详细解析这些接口，带你从零开始掌握Heygem的二次开发。

1. 为什么需要二次开发接口？

在开始技术细节之前，先说说为什么要有二次开发接口。你可能已经用过Heygem的Web界面，上传音频和视频，点击按钮就能生成数字人视频。这很方便，但如果你需要：

• 将数字人视频生成集成到自己的网站或应用中
• 批量处理成千上万个视频，但不想手动操作
• 根据业务逻辑动态调整生成参数
• 将生成结果自动推送到其他系统

这时候，Web界面就不够用了。你需要的是程序化的接口，能够通过代码来控制整个流程。这就是二次开发接口的价值所在。

2. 系统架构概览

要理解接口，先要了解系统的整体架构。Heygem系统采用前后端分离的设计：

前端WebUI (Gradio) ←→ 后端API服务 ←→ 数字人视频生成引擎

前端负责用户交互，后端提供API接口，生成引擎处理实际的视频合成。当你进行二次开发时，你实际上是在与后端API服务交互。

系统支持两种主要模式：

• 批量处理模式：用同一段音频生成多个数字人视频
• 单个处理模式：快速生成单个数字人视频

两种模式都有对应的API接口，你可以根据需求选择。

3. 核心API接口详解

现在进入正题，看看具体的API接口。所有接口都通过HTTP请求调用，返回JSON格式的数据。

3.1 基础配置接口

在开始调用生成接口前，你可能需要获取系统状态或配置信息。

获取系统状态

import requests

# 获取系统当前状态
response = requests.get("http://localhost:7860/api/status")
status_data = response.json()

print(f"系统状态: {status_data['status']}")
print(f"当前任务数: {status_data['queue_length']}")
print(f"GPU可用: {status_data['gpu_available']}")

这个接口返回系统的运行状态、任务队列长度、GPU是否可用等信息。在开始大批量处理前，先检查系统状态是个好习惯。

获取支持的文件格式

# 获取系统支持的文件格式
response = requests.get("http://localhost:7860/api/supported_formats")
formats = response.json()

print("支持的音频格式:", formats['audio'])
print("支持的视频格式:", formats['video'])

了解系统支持哪些格式，可以避免上传不支持的文件导致错误。

3.2 单个视频生成接口

这是最基础的接口，用于生成单个数字人视频。

接口地址: POST http://localhost:7860/api/generate/single

请求参数:

import requests
import base64

# 准备请求数据
payload = {
    "audio_file": "base64编码的音频文件内容",
    "video_file": "base64编码的视频文件内容",
    "output_format": "mp4",  # 可选，默认mp4
    "quality": "high",  # 可选：low, medium, high
    "callback_url": "https://your-server.com/callback"  # 可选，处理完成后的回调地址
}

# 实际调用
response = requests.post(
    "http://localhost:7860/api/generate/single",
    json=payload,
    headers={"Content-Type": "application/json"}
)

result = response.json()
if result["success"]:
    print(f"任务ID: {result['task_id']}")
    print(f"预计完成时间: {result['estimated_time']}秒")
else:
    print(f"错误: {result['error']}")

参数说明:

• audio_file: Base64编码的音频文件内容
• video_file: Base64编码的视频文件内容
• output_format: 输出视频格式，默认mp4
• quality: 生成质量，影响处理时间和输出效果
• callback_url: 处理完成后的回调地址，系统会POST结果到这个地址

返回结果:

{
    "success": true,
    "task_id": "task_123456",
    "estimated_time": 120,
    "message": "任务已加入队列"
}

3.3 批量视频生成接口

批量处理是Heygem的强项，这个接口可以一次性处理多个视频。

接口地址: POST http://localhost:7860/api/generate/batch

请求参数:

import requests

# 准备批量处理数据
payload = {
    "audio_file": "base64编码的音频文件内容",
    "video_files": [
        "base64编码的视频1内容",
        "base64编码的视频2内容",
        "base64编码的视频3内容"
    ],
    "output_prefix": "batch_",  # 输出文件前缀
    "concurrent_limit": 2,  # 同时处理的任务数
    "notify_email": "user@example.com"  # 可选，处理完成通知邮箱
}

response = requests.post(
    "http://localhost:7860/api/generate/batch",
    json=payload
)

result = response.json()
if result["success"]:
    print(f"批量任务ID: {result['batch_id']}")
    print(f"包含任务数: {len(result['task_ids'])}")
    for task_id in result['task_ids']:
        print(f"  - 子任务: {task_id}")

参数说明:

• audio_file: 共用的音频文件
• video_files: 视频文件列表，每个都会与音频合成
• output_prefix: 输出文件名的前缀
• concurrent_limit: 最大并发数，控制资源使用
• notify_email: 处理完成后的通知邮箱

返回结果:

{
    "success": true,
    "batch_id": "batch_789012",
    "task_ids": ["task_001", "task_002", "task_003"],
    "total_tasks": 3,
    "message": "批量任务已创建"
}

3.4 任务状态查询接口

提交任务后，你需要知道处理进度。

接口地址: GET http://localhost:7860/api/task/{task_id}

使用示例:

def check_task_status(task_id):
    """检查任务状态"""
    response = requests.get(f"http://localhost:7860/api/task/{task_id}")
    status = response.json()
    
    print(f"任务ID: {status['task_id']}")
    print(f"状态: {status['status']}")  # pending, processing, completed, failed
    print(f"进度: {status.get('progress', 0)}%")
    
    if status['status'] == 'completed':
        print(f"结果文件: {status['result_file']}")
        print(f"处理时间: {status['processing_time']}秒")
    elif status['status'] == 'failed':
        print(f"错误信息: {status['error_message']}")
    
    return status

状态说明:

• pending: 等待处理
• processing: 正在处理
• completed: 处理完成
• failed: 处理失败

3.5 结果下载接口

任务完成后，你需要下载生成的结果。

下载单个结果

def download_result(task_id, save_path):
    """下载任务结果"""
    # 先获取任务信息
    status_response = requests.get(f"http://localhost:7860/api/task/{task_id}")
    task_info = status_response.json()
    
    if task_info['status'] != 'completed':
        print("任务尚未完成")
        return False
    
    # 下载视频文件
    download_url = f"http://localhost:7860/api/download/{task_id}"
    response = requests.get(download_url, stream=True)
    
    with open(save_path, 'wb') as f:
        for chunk in response.iter_content(chunk_size=8192):
            f.write(chunk)
    
    print(f"文件已保存到: {save_path}")
    return True

批量下载结果

def download_batch_results(batch_id, output_dir):
    """下载批量任务的所有结果"""
    import os
    import zipfile
    
    # 创建输出目录
    os.makedirs(output_dir, exist_ok=True)
    
    # 下载ZIP包
    zip_url = f"http://localhost:7860/api/download/batch/{batch_id}"
    zip_path = os.path.join(output_dir, f"{batch_id}.zip")
    
    response = requests.get(zip_url, stream=True)
    with open(zip_path, 'wb') as f:
        for chunk in response.iter_content(chunk_size=8192):
            f.write(chunk)
    
    # 解压
    with zipfile.ZipFile(zip_path, 'r') as zip_ref:
        zip_ref.extractall(output_dir)
    
    print(f"批量结果已下载到: {output_dir}")
    return True

4. 实战：构建自动化处理系统

了解了基本接口后，我们来看一个完整的实战例子：构建一个自动化数字人视频生成系统。

4.1 系统设计思路

假设我们要为电商平台开发一个功能：为每个商品生成数字人介绍视频。需求如下：

• 每天处理数百个商品
• 每个商品有固定的介绍音频
• 需要为每个商品匹配不同的模特视频
• 生成后自动上传到CDN
• 失败的任务需要重试

4.2 核心代码实现

import requests
import base64
import time
import json
from pathlib import Path
from typing import List, Dict
import logging

class HeyGemAutomation:
    """HeyGem自动化处理类"""
    
    def __init__(self, base_url="http://localhost:7860"):
        self.base_url = base_url
        self.session = requests.Session()
        self.logger = logging.getLogger(__name__)
    
    def file_to_base64(self, file_path: str) -> str:
        """将文件转换为Base64编码"""
        with open(file_path, 'rb') as f:
            file_data = f.read()
        return base64.b64encode(file_data).decode('utf-8')
    
    def submit_batch_job(self, audio_path: str, video_paths: List[str]) -> Dict:
        """提交批量任务"""
        # 读取音频文件
        audio_base64 = self.file_to_base64(audio_path)
        
        # 读取所有视频文件
        video_base64_list = []
        for video_path in video_paths:
            video_base64 = self.file_to_base64(video_path)
            video_base64_list.append(video_base64)
        
        # 准备请求数据
        payload = {
            "audio_file": audio_base64,
            "video_files": video_base64_list,
            "output_prefix": "product_",
            "concurrent_limit": 3,  # 同时处理3个
            "callback_url": f"{self.base_url}/api/callback"
        }
        
        # 提交任务
        try:
            response = self.session.post(
                f"{self.base_url}/api/generate/batch",
                json=payload,
                timeout=30
            )
            response.raise_for_status()
            return response.json()
        except Exception as e:
            self.logger.error(f"提交任务失败: {e}")
            return {"success": False, "error": str(e)}
    
    def monitor_tasks(self, task_ids: List[str], check_interval=10):
        """监控任务进度"""
        completed = []
        failed = []
        processing = task_ids.copy()
        
        while processing:
            for task_id in processing[:]:  # 复制列表进行遍历
                try:
                    status = self.get_task_status(task_id)
                    
                    if status['status'] == 'completed':
                        self.logger.info(f"任务 {task_id} 完成")
                        completed.append(task_id)
                        processing.remove(task_id)
                        
                        # 下载结果
                        self.download_result(task_id, f"outputs/{task_id}.mp4")
                        
                    elif status['status'] == 'failed':
                        self.logger.error(f"任务 {task_id} 失败: {status.get('error_message')}")
                        failed.append(task_id)
                        processing.remove(task_id)
                        
                    else:
                        progress = status.get('progress', 0)
                        self.logger.info(f"任务 {task_id} 进度: {progress}%")
                        
                except Exception as e:
                    self.logger.error(f"检查任务 {task_id} 状态失败: {e}")
            
            if processing:
                time.sleep(check_interval)
        
        return {
            "completed": completed,
            "failed": failed,
            "total": len(task_ids)
        }
    
    def get_task_status(self, task_id: str) -> Dict:
        """获取任务状态"""
        response = self.session.get(f"{self.base_url}/api/task/{task_id}")
        return response.json()
    
    def download_result(self, task_id: str, save_path: str):
        """下载任务结果"""
        # 确保目录存在
        Path(save_path).parent.mkdir(parents=True, exist_ok=True)
        
        # 下载文件
        response = self.session.get(
            f"{self.base_url}/api/download/{task_id}",
            stream=True
        )
        
        with open(save_path, 'wb') as f:
            for chunk in response.iter_content(chunk_size=8192):
                f.write(chunk)
        
        self.logger.info(f"文件已保存: {save_path}")
    
    def retry_failed_tasks(self, failed_tasks: List[Dict], max_retries=3):
        """重试失败的任务"""
        retry_results = []
        
        for task_info in failed_tasks:
            for attempt in range(max_retries):
                self.logger.info(f"重试任务 {task_info['task_id']}, 第{attempt+1}次尝试")
                
                # 重新提交任务
                result = self.submit_single_job(
                    task_info['audio_path'],
                    task_info['video_path']
                )
                
                if result['success']:
                    retry_results.append({
                        'original_task': task_info['task_id'],
                        'new_task': result['task_id'],
                        'attempt': attempt + 1,
                        'success': True
                    })
                    break
                else:
                    if attempt == max_retries - 1:  # 最后一次尝试也失败
                        retry_results.append({
                            'original_task': task_info['task_id'],
                            'attempts': attempt + 1,
                            'success': False,
                            'error': result.get('error')
                        })
        
        return retry_results

# 使用示例
def main():
    # 初始化自动化处理器
    processor = HeyGemAutomation(base_url="http://your-server:7860")
    
    # 准备数据
    audio_file = "product_intro.wav"
    video_files = [
        "model1.mp4",
        "model2.mp4", 
        "model3.mp4",
        "model4.mp4",
        "model5.mp4"
    ]
    
    # 提交批量任务
    batch_result = processor.submit_batch_job(audio_file, video_files)
    
    if batch_result['success']:
        print(f"批量任务创建成功，ID: {batch_result['batch_id']}")
        
        # 监控任务进度
        monitor_result = processor.monitor_tasks(batch_result['task_ids'])
        
        print(f"处理完成: {len(monitor_result['completed'])} 个成功")
        print(f"处理失败: {len(monitor_result['failed'])} 个失败")
        
        # 如果有失败的任务，可以重试
        if monitor_result['failed']:
            print("开始重试失败的任务...")
            # 这里需要根据实际情况准备重试数据
            # retry_results = processor.retry_failed_tasks(failed_tasks)
    else:
        print(f"任务提交失败: {batch_result.get('error')}")

if __name__ == "__main__":
    main()

4.3 错误处理与重试机制

在实际生产环境中，网络波动、资源不足等问题可能导致任务失败。一个好的自动化系统需要有完善的错误处理机制。

class RobustHeyGemClient:
    """增强版的HeyGem客户端，包含错误处理和重试"""
    
    def __init__(self, base_url, max_retries=3, retry_delay=5):
        self.base_url = base_url
        self.max_retries = max_retries
        self.retry_delay = retry_delay
        self.session = requests.Session()
        
        # 设置重试策略
        retry_strategy = requests.packages.urllib3.util.retry.Retry(
            total=max_retries,
            backoff_factor=retry_delay,
            status_forcelist=[500, 502, 503, 504]
        )
        adapter = requests.adapters.HTTPAdapter(max_retries=retry_strategy)
        self.session.mount("http://", adapter)
        self.session.mount("https://", adapter)
    
    def submit_with_retry(self, audio_path, video_paths):
        """带重试的任务提交"""
        for attempt in range(self.max_retries):
            try:
                result = self.submit_batch_job(audio_path, video_paths)
                
                if result['success']:
                    return result
                else:
                    print(f"第{attempt+1}次尝试失败: {result.get('error')}")
                    
            except requests.exceptions.RequestException as e:
                print(f"第{attempt+1}次尝试网络错误: {e}")
            
            # 如果不是最后一次尝试，等待后重试
            if attempt < self.max_retries - 1:
                time.sleep(self.retry_delay * (attempt + 1))
        
        # 所有重试都失败
        return {
            "success": False,
            "error": f"所有{self.max_retries}次尝试都失败"
        }
    
    def check_system_health(self):
        """检查系统健康状态"""
        try:
            # 检查API服务
            status_response = self.session.get(
                f"{self.base_url}/api/status",
                timeout=5
            )
            status_data = status_response.json()
            
            # 检查磁盘空间（通过自定义端点）
            disk_response = self.session.get(
                f"{self.base_url}/api/system/disk",
                timeout=5
            )
            disk_data = disk_response.json()
            
            health_status = {
                "api_available": status_data.get("status") == "running",
                "queue_length": status_data.get("queue_length", 0),
                "gpu_available": status_data.get("gpu_available", False),
                "disk_free_gb": disk_data.get("free_gb", 0),
                "disk_usage_percent": disk_data.get("usage_percent", 0)
            }
            
            # 判断是否健康
            is_healthy = (
                health_status["api_available"] and
                health_status["queue_length"] < 100 and  # 队列不要太长
                health_status["disk_free_gb"] > 10  # 至少10GB空闲空间
            )
            
            health_status["is_healthy"] = is_healthy
            return health_status
            
        except Exception as e:
            print(f"健康检查失败: {e}")
            return {
                "is_healthy": False,
                "error": str(e)
            }

5. 高级功能与优化建议

5.1 性能优化技巧

并发控制

# 根据系统资源动态调整并发数
def calculate_optimal_concurrent():
    """计算最优并发数"""
    health = check_system_health()
    
    if not health["is_healthy"]:
        return 1  # 系统不健康时降低并发
    
    if health["gpu_available"]:
        # 有GPU时可以提高并发
        base_concurrent = 3
    else:
        # 只有CPU时降低并发
        base_concurrent = 1
    
    # 根据队列长度调整
    queue_length = health["queue_length"]
    if queue_length > 50:
        base_concurrent = max(1, base_concurrent - 1)
    
    return base_concurrent

批量处理优化

def optimize_batch_processing(video_paths, max_batch_size=10):
    """优化批量处理，避免单次请求太大"""
    optimized_batches = []
    
    # 按文件大小分组
    small_files = []
    large_files = []
    
    for path in video_paths:
        size_mb = os.path.getsize(path) / (1024 * 1024)
        if size_mb < 50:  # 小于50MB为小文件
            small_files.append(path)
        else:
            large_files.append(path)
    
    # 小文件可以批量处理
    for i in range(0, len(small_files), max_batch_size):
        batch = small_files[i:i + max_batch_size]
        optimized_batches.append({
            "type": "batch",
            "files": batch,
            "estimated_size": sum(os.path.getsize(f) for f in batch) / (1024 * 1024)
        })
    
    # 大文件单独处理
    for large_file in large_files:
        optimized_batches.append({
            "type": "single",
            "files": [large_file],
            "estimated_size": os.path.getsize(large_file) / (1024 * 1024)
        })
    
    return optimized_batches

5.2 回调机制实现

对于长时间运行的任务，使用回调机制可以让系统在任务完成后主动通知你。

设置回调端点

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/api/callback', methods=['POST'])
def heygem_callback():
    """处理HeyGem回调"""
    data = request.json
    
    task_id = data.get('task_id')
    status = data.get('status')
    result_url = data.get('result_url')
    error_message = data.get('error_message')
    
    if status == 'completed':
        print(f"任务 {task_id} 已完成")
        print(f"结果地址: {result_url}")
        
        # 这里可以触发后续处理，比如：
        # 1. 下载文件到本地
        # 2. 上传到CDN
        # 3. 更新数据库状态
        # 4. 发送通知
        
    elif status == 'failed':
        print(f"任务 {task_id} 失败: {error_message}")
        
        # 失败处理逻辑
        # 1. 记录错误日志
        # 2. 触发重试机制
        # 3. 发送告警
    
    return jsonify({"success": True})

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

5.3 监控与日志

完善的监控和日志系统对于生产环境至关重要。

import logging
from datetime import datetime
import json

class HeyGemMonitor:
    """HeyGem任务监控器"""
    
    def __init__(self, log_file="heygem_monitor.log"):
        # 设置日志
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler(log_file),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
        
        # 任务统计
        self.stats = {
            "total_tasks": 0,
            "completed_tasks": 0,
            "failed_tasks": 0,
            "total_processing_time": 0
        }
    
    def log_task_start(self, task_id, audio_size, video_count):
        """记录任务开始"""
        self.stats["total_tasks"] += 1
        
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "event": "task_start",
            "task_id": task_id,
            "audio_size_mb": audio_size,
            "video_count": video_count
        }
        
        self.logger.info(f"任务开始: {json.dumps(log_entry)}")
    
    def log_task_complete(self, task_id, processing_time, output_size):
        """记录任务完成"""
        self.stats["completed_tasks"] += 1
        self.stats["total_processing_time"] += processing_time
        
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "event": "task_complete",
            "task_id": task_id,
            "processing_time_seconds": processing_time,
            "output_size_mb": output_size,
            "avg_processing_time": self.stats["total_processing_time"] / self.stats["completed_tasks"]
        }
        
        self.logger.info(f"任务完成: {json.dumps(log_entry)}")
    
    def log_task_failure(self, task_id, error_message, retry_count=0):
        """记录任务失败"""
        self.stats["failed_tasks"] += 1
        
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "event": "task_failure",
            "task_id": task_id,
            "error": error_message,
            "retry_count": retry_count,
            "success_rate": self.stats["completed_tasks"] / self.stats["total_tasks"] * 100
        }
        
        self.logger.error(f"任务失败: {json.dumps(log_entry)}")
    
    def generate_report(self):
        """生成统计报告"""
        report = {
            "timestamp": datetime.now().isoformat(),
            "statistics": self.stats.copy(),
            "success_rate": (self.stats["completed_tasks"] / self.stats["total_tasks"] * 100) if self.stats["total_tasks"] > 0 else 0,
            "avg_processing_time": self.stats["total_processing_time"] / self.stats["completed_tasks"] if self.stats["completed_tasks"] > 0 else 0
        }
        
        self.logger.info(f"统计报告: {json.dumps(report, indent=2)}")
        return report

6. 总结

通过本文的详细解析，你应该已经对Heygem数字人视频生成系统的二次开发接口有了全面的了解。从基础的单任务处理到复杂的批量自动化系统，这些接口提供了灵活而强大的集成能力。

关键要点回顾：

1. 接口设计合理：Heygem的API设计考虑了实际使用场景，提供了从任务提交、状态查询到结果下载的完整流程

2. 批量处理是核心优势：通过批量接口，你可以高效处理大量视频，显著提升工作效率

3. 错误处理很重要：在生产环境中，完善的错误处理和重试机制是系统稳定性的保障

4. 监控不能少：通过日志和监控系统，你可以实时了解系统状态，快速发现问题

5. 回调机制提升体验：对于长时间任务，使用回调可以让你的应用更响应式

下一步建议：

如果你准备开始二次开发，我建议：

1. 先从单个任务接口开始，熟悉基本流程
2. 实现简单的错误处理和重试逻辑
3. 逐步扩展到批量处理
4. 添加监控和日志功能
5. 根据业务需求优化性能

数字人视频生成技术正在快速发展，通过Heygem的二次开发接口，你可以将这项技术快速集成到自己的产品中。无论是电商、教育、娱乐还是其他领域，AI数字人都能为你带来全新的用户体验。

记住，好的系统不是一蹴而就的。从简单开始，逐步完善，根据实际使用反馈不断优化。如果你在开发过程中遇到问题，或者有新的功能需求，欢迎随时交流。

---

获取更多AI镜像

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