Emotion2Vec+ Large二次开发怎么搞？API调用入门必看教程

1. 引言：构建可扩展的语音情感识别系统

随着人机交互技术的发展，语音情感识别在智能客服、心理健康监测、车载系统等场景中展现出巨大潜力。Emotion2Vec+ Large 是由阿里达摩院发布的大规模语音情感识别模型，具备高精度、多语言支持和强鲁棒性等特点。科哥基于该模型构建了本地化部署的 WebUI 系统，极大降低了使用门槛。

然而，对于开发者而言，仅使用图形界面无法满足自动化、集成化的需求。本文将重点讲解如何对 Emotion2Vec+ Large 系统进行二次开发，特别是通过 API 接口实现程序化调用，帮助您将语音情感识别能力无缝嵌入到自有系统中。

本教程适用于希望实现以下目标的开发者：

将情感识别功能集成进后端服务
实现批量音频处理流水线
提取音频 embedding 特征用于聚类或相似度计算
构建自动化测试与评估框架

2. 系统架构与 API 设计原理

2.1 整体架构解析

Emotion2Vec+ Large 本地系统采用典型的前后端分离架构：

[客户端] ←HTTP→ [Flask后端] ←Python→ [Emotion2Vec+模型]

前端：Gradio 构建的 WebUI，提供可视化操作
后端：Flask 框架驱动的服务核心，负责音频处理、模型加载与推理调度
模型层：基于 HuggingFace Transformers 或 ModelScope SDK 加载的预训练模型

虽然原系统未显式暴露 RESTful API，但其内部逻辑天然支持接口化改造。

2.2 API 调用的本质机制

通过对run.sh和 Gradio 启动脚本分析可知，系统启动的是一个监听localhost:7860的 Python 服务。Gradio 底层基于 FastAPI/Flask，所有 UI 操作最终都转化为 HTTP 请求。

因此，我们可以通过逆向工程方式，模拟浏览器行为，直接向后端发送 POST 请求来触发情感识别任务。

2.3 关键接口路径识别

经调试确认，核心接口路径如下：

功能	URL 路径	请求方法
文件上传与推理	`/api/predict/`	POST
获取示例音频	`/api/duplicate/`	GET

其中/api/predict/是实现二次开发的核心入口。

3. 实现 API 调用：从零开始的完整实践

3.1 准备工作与环境配置

确保系统已正常运行：

/bin/bash /root/run.sh

等待服务启动完成后（出现 "Running on local URL: http://localhost:7860"），即可开始调用。

所需依赖库：

pip install requests numpy

3.2 构建标准 API 请求结构

Gradio 的 API 接口遵循特定的数据封装格式。以下是调用情感识别功能的标准 JSON 结构：

{ "data": [ "base64编码的音频数据", "utterance", false ], "event_data": null, "fn_index": 0, "trigger_id": 1 }

字段说明：

data[0]：音频文件内容，需转为 Base64 编码字符串
data[1]：粒度选择，可选"utterance"或"frame"
data[2]：是否提取 Embedding，true/false

3.3 完整调用代码示例

import requests import base64 import json import numpy as np def call_emotion2vec_api(audio_path, granularity="utterance", extract_embedding=False): """ 调用 Emotion2Vec+ Large 的情感识别 API 参数: audio_path: 音频文件路径 granularity: 分析粒度 ("utterance" 或 "frame") extract_embedding: 是否导出特征向量 返回: 响应字典 """ # 读取音频文件并编码 with open(audio_path, 'rb') as f: audio_b64 = base64.b64encode(f.read()).decode('utf-8') # 构造 payload payload = { "data": [ f"data:audio/wav;base64,{audio_b64}", granularity, extract_embedding ], "event_data": None, "fn_index": 0, "trigger_id": 1 } # 发送请求 url = "http://localhost:7860/api/predict/" headers = {"Content-Type": "application/json"} response = requests.post(url, data=json.dumps(payload), headers=headers) if response.status_code == 200: return response.json() else: raise Exception(f"API 调用失败: {response.status_code}, {response.text}") # 使用示例 if __name__ == "__main__": result = call_emotion2vec_api( audio_path="./test.wav", granularity="utterance", extract_embedding=True ) print("原始响应:", result)

3.4 解析返回结果

API 返回结构如下：

{ "data": [ "😊 快乐 (Happy)\n置信度: 85.3%", "{\"emotion\": \"happy\", ...}", "/path/to/embedding.npy" ] }

解析代码：

def parse_result(api_response): """解析 API 返回结果""" raw_text = api_response['data'][0] # 主要情感文本 json_str = api_response['data'][1] # JSON 结果 embedding_path = api_response['data'][2] # 特征路径（可能为空） result_json = json.loads(json_str) print(f"主要情感: {result_json['emotion']}") print(f"置信度: {result_json['confidence']:.3f}") print("详细得分:") for emo, score in result_json['scores'].items(): print(f" {emo}: {score:.3f}") if embedding_path: embedding = np.load(embedding_path) print(f"Embedding 形状: {embedding.shape}") return result_json, embedding if embedding_path else None # 调用解析 parsed_result, emb = parse_result(result)

4. 高级技巧与工程优化建议

4.1 批量处理多个音频文件

import os from concurrent.futures import ThreadPoolExecutor def batch_process_audio_folder(folder_path): results = [] audio_files = [f for f in os.listdir(folder_path) if f.endswith(('.wav', '.mp3'))] def process_single(file): try: res = call_emotion2vec_api(os.path.join(folder_path, file)) parsed, _ = parse_result(res) return {"file": file, "result": parsed} except Exception as e: return {"file": file, "error": str(e)} with ThreadPoolExecutor(max_workers=4) as executor: results = list(executor.map(process_single, audio_files)) return results

提示：由于模型加载在内存中，多线程可显著提升吞吐量。

4.2 构建轻量级代理 API 服务

为避免直接依赖 Gradio 内部接口（可能随版本变化），建议封装一层稳定 API：

from flask import Flask, request, jsonify app = Flask(__name__) @app.route('/analyze', methods=['POST']) def analyze_emotion(): if 'audio' not in request.files: return jsonify({"error": "缺少音频文件"}), 400 file = request.files['audio'] temp_path = "/tmp/temp_upload." + file.filename.split('.')[-1] file.save(temp_path) try: result = call_emotion2vec_api( temp_path, granularity=request.form.get('granularity', 'utterance'), extract_embedding='extract_embedding' in request.form ) parsed, emb = parse_result(result) # 可选：返回 embedding 的 base64 编码 if emb is not None: from io import BytesIO buf = BytesIO() np.save(buf, emb) emb_b64 = base64.b64encode(buf.getvalue()).decode('utf-8') parsed['embedding_base64'] = emb_b64 return jsonify(parsed) except Exception as e: return jsonify({"error": str(e)}), 500 finally: os.remove(temp_path) if __name__ == '__main__': app.run(host='0.0.0.0', port=5000)

启动后可通过POST /analyze统一接入。