Emotion2Vec+ Large语音情感识别系统Python读取结果文件示例代码

Emotion2Vec+ Large语音情感识别系统Python读取结果文件示例代码

1. 快速入门：理解结果文件结构

Emotion2Vec+ Large语音情感识别系统在完成音频分析后，会自动生成结构化的结果文件，存放在outputs/outputs_YYYYMMDD_HHMMSS/目录下。这些文件是后续二次开发和数据处理的基础。

系统默认生成两个核心文件：

• result.json：包含情感识别的全部结构化结果
• embedding.npy（可选）：当勾选"提取 Embedding 特征"时生成，包含音频的深度特征向量

对于大多数开发者而言，result.json是最常用的结果文件。它采用标准JSON格式，便于各种编程语言解析。文件内容清晰分层，包含了情感标签、置信度、详细得分分布等关键信息。

值得注意的是，该系统设计时充分考虑了工程落地需求——所有输出都遵循统一规范，无需额外的数据清洗步骤即可直接用于业务逻辑。这种"开箱即用"的设计理念，让开发者能够将精力集中在更高价值的应用创新上，而不是繁琐的数据格式适配工作。

2. result.json文件详解与Python解析

2.1 JSON结构解析

result.json文件采用简洁明了的JSON结构，包含以下核心字段：

{ "emotion": "happy", "confidence": 0.853, "scores": { "angry": 0.012, "disgusted": 0.008, "fearful": 0.015, "happy": 0.853, "neutral": 0.045, "other": 0.023, "sad": 0.018, "surprised": 0.021, "unknown": 0.005 }, "granularity": "utterance", "timestamp": "2024-01-04 22:30:00" }

各字段含义如下：

• emotion：主情感标签（英文小写）
• confidence：主情感置信度（0.0-1.0范围）
• scores：9种情感的详细得分字典，总和为1.0
• granularity：识别粒度（"utterance"整句级或"frame"帧级）
• timestamp：处理时间戳

2.2 基础Python读取代码

以下是最简洁实用的Python读取示例，适用于大多数应用场景：

import json from pathlib import Path def read_emotion_result(json_path): """ 读取Emotion2Vec+ Large系统生成的result.json文件 Args: json_path (str): result.json文件的完整路径 Returns: dict: 解析后的结果字典 """ try: # 使用Path对象确保跨平台兼容性 file_path = Path(json_path) # 检查文件是否存在 if not file_path.exists(): raise FileNotFoundError(f"结果文件不存在: {json_path}") # 读取并解析JSON with open(file_path, 'r', encoding='utf-8') as f: result = json.load(f) return result except json.JSONDecodeError as e: print(f"JSON解析错误: {e}") return None except Exception as e: print(f"读取文件时发生错误: {e}") return None # 使用示例 if __name__ == "__main__": # 替换为实际的result.json路径 result_file = "outputs/outputs_20240104_223000/result.json" # 读取结果 emotion_result = read_emotion_result(result_file) if emotion_result: print(" 成功读取情感识别结果") print(f"主要情感: {emotion_result['emotion']}") print(f"置信度: {emotion_result['confidence']:.3f}") print(f"识别粒度: {emotion_result['granularity']}") else: print(" 读取结果失败")

这段代码采用了生产环境推荐的最佳实践：

• 使用pathlib.Path确保Windows/Linux/macOS路径兼容
• 包含完整的异常处理机制
• 提供清晰的错误提示信息
• 返回值类型明确，便于后续调用

2.3 实用工具函数集合

在实际项目中，我们通常需要对结果进行更多操作。以下是几个常用的工具函数：

import json from pathlib import Path from typing import Dict, List, Optional class EmotionResultReader: """Emotion2Vec+ Large结果文件读取器""" # 情感映射表（中英文对照） EMOTION_MAP = { "angry": "愤怒", "disgusted": "厌恶", "fearful": "恐惧", "happy": "快乐", "neutral": "中性", "other": "其他", "sad": "悲伤", "surprised": "惊讶", "unknown": "未知" } def __init__(self, json_path: str): self.json_path = Path(json_path) self.data = self._load_data() def _load_data(self) -> Optional[Dict]: """加载并验证JSON数据""" try: if not self.json_path.exists(): raise FileNotFoundError(f"文件不存在: {self.json_path}") with open(self.json_path, 'r', encoding='utf-8') as f: data = json.load(f) # 验证必要字段 required_fields = ['emotion', 'confidence', 'scores', 'granularity'] for field in required_fields: if field not in data: raise ValueError(f"缺少必要字段: {field}") return data except Exception as e: print(f" 加载结果文件失败: {e}") return None def get_main_emotion(self) -> str: """获取主要情感的中文名称""" if not self.data: return "未知" return self.EMOTION_MAP.get(self.data['emotion'], "未知") def get_confidence_percentage(self) -> float: """获取置信度百分比（0-100）""" if not self.data: return 0.0 return round(self.data['confidence'] * 100, 1) def get_top_emotions(self, n: int = 3) -> List[Dict]: """获取置信度最高的前N种情感""" if not self.data: return [] # 将scores字典转换为列表并按得分排序 scores_list = [ {"emotion": emo, "score": score, "chinese": self.EMOTION_MAP.get(emo, emo)} for emo, score in self.data['scores'].items() ] # 按得分降序排列 scores_list.sort(key=lambda x: x['score'], reverse=True) return scores_list[:n] def is_positive_emotion(self) -> bool: """判断是否为积极情感（快乐）""" if not self.data: return False return self.data['emotion'] == 'happy' def get_detailed_report(self) -> str: """生成详细的情感分析报告""" if not self.data: return "无法生成报告：数据加载失败" report = [] report.append("=" * 40) report.append("🔊 Emotion2Vec+ Large情感分析报告") report.append("=" * 40) report.append(f"主要情感: {self.get_main_emotion()} ({self.data['emotion']})") report.append(f"置信度: {self.get_confidence_percentage()}%") report.append(f"识别粒度: {self.data['granularity']}") report.append("") report.append(" 详细得分分布:") # 按固定顺序显示所有情感（保持一致性） emotion_order = ['happy', 'neutral', 'surprised', 'angry', 'fearful', 'sad', 'disgusted', 'other', 'unknown'] for emo in emotion_order: score = self.data['scores'].get(emo, 0.0) chinese_name = self.EMOTION_MAP.get(emo, emo) percentage = round(score * 100, 1) marker = "" if emo == self.data['emotion'] else " " report.append(f"{marker}{chinese_name:8} {percentage:5.1f}%") report.append("=" * 40) return "\n".join(report) # 使用示例 if __name__ == "__main__": # 初始化读取器 reader = EmotionResultReader("outputs/outputs_20240104_223000/result.json") # 获取基本信息 print(f"主要情感: {reader.get_main_emotion()}") print(f"置信度: {reader.get_confidence_percentage()}%") print(f"是否积极情感: {reader.is_positive_emotion()}") # 获取前三名情感 top3 = reader.get_top_emotions(3) print("\n前三名情感:") for i, item in enumerate(top3, 1): print(f"{i}. {item['chinese']} ({item['emotion']}): {item['score']:.3f}") # 生成详细报告 print("\n" + reader.get_detailed_report())

这个类封装了常见操作，具有以下优势：

• 类型安全：使用类型提示提高代码可维护性
• 错误防御：内置数据验证和异常处理
• 业务友好：提供中文情感名称、百分比转换等实用功能
• 扩展性强：易于添加新功能而不影响现有代码

3. embedding.npy特征向量读取与应用

当系统启用"提取 Embedding 特征"选项时，会生成embedding.npy文件。这个NumPy数组包含了音频的深度语义特征，是进行高级分析和二次开发的关键。

3.1 基础读取与验证

import numpy as np from pathlib import Path def read_embedding(embedding_path: str) -> Optional[np.ndarray]: """ 读取Emotion2Vec+ Large生成的embedding.npy文件 Args: embedding_path (str): embedding.npy文件路径 Returns: np.ndarray: 特征向量数组，失败时返回None """ try: file_path = Path(embedding_path) if not file_path.exists(): print(f" 警告: embedding文件不存在 - {embedding_path}") return None # 使用numpy.load读取 embedding = np.load(file_path) # 验证数组维度 if embedding.ndim != 1: print(f" 警告: embedding维度异常 - 期望1维，得到{embedding.ndim}维") return None print(f" 成功读取embedding: {embedding.shape[0]}维向量") return embedding except Exception as e: print(f" 读取embedding失败: {e}") return None # 使用示例 embedding = read_embedding("outputs/outputs_20240104_223000/embedding.npy") if embedding is not None: print(f"Embedding形状: {embedding.shape}") print(f"前5个值: {embedding[:5]}")

3.2 特征向量的典型应用场景

场景1：情感相似度计算

from sklearn.metrics.pairwise import cosine_similarity import numpy as np def calculate_emotion_similarity(embedding1: np.ndarray, embedding2: np.ndarray) -> float: """ 计算两个音频情感特征的余弦相似度 Args: embedding1, embedding2: 两个embedding向量 Returns: float: 相似度分数（0.0-1.0） """ # 确保是二维数组用于sklearn vec1 = embedding1.reshape(1, -1) vec2 = embedding2.reshape(1, -1) similarity = cosine_similarity(vec1, vec2)[0][0] return float(similarity) # 示例：比较两个音频的情感相似度 emb1 = read_embedding("outputs/outputs_20240104_223000/embedding.npy") emb2 = read_embedding("outputs/outputs_20240104_223500/embedding.npy") if emb1 is not None and emb2 is not None: sim_score = calculate_emotion_similarity(emb1, emb2) print(f"情感相似度: {sim_score:.3f}") if sim_score > 0.8: print(" 两个音频情感高度相似") elif sim_score > 0.6: print("🟡 两个音频情感有一定相似性") else: print("🔴 两个音频情感差异较大")

场景2：批量处理多个结果文件

from pathlib import Path import json import numpy as np from typing import List, Dict, Any def batch_process_results(output_dir: str) -> List[Dict[str, Any]]: """ 批量处理指定目录下的所有结果文件 Args: output_dir: outputs目录路径 Returns: 包含所有结果的列表 """ results_dir = Path(output_dir) all_results = [] # 查找所有result.json文件 result_files = list(results_dir.glob("outputs_*/result.json")) print(f" 发现 {len(result_files)} 个结果文件") for result_file in result_files: try: # 读取JSON结果 with open(result_file, 'r', encoding='utf-8') as f: result_data = json.load(f) # 尝试读取对应的embedding embedding_path = result_file.parent / "embedding.npy" embedding = None if embedding_path.exists(): embedding = np.load(embedding_path).tolist() # 构建完整结果记录 full_result = { "timestamp": result_data.get("timestamp", ""), "emotion": result_data.get("emotion", "unknown"), "confidence": result_data.get("confidence", 0.0), "granularity": result_data.get("granularity", "unknown"), "embedding_shape": len(embedding) if embedding else 0, "embedding": embedding } all_results.append(full_result) except Exception as e: print(f" 处理文件 {result_file} 时出错: {e}") continue return all_results # 使用示例 all_analysis = batch_process_results("outputs/") print(f" 总共处理了 {len(all_analysis)} 个音频分析结果") # 统计情感分布 if all_analysis: emotion_count = {} for r in all_analysis: emo = r["emotion"] emotion_count[emo] = emotion_count.get(emo, 0) + 1 print("\n 情感分布统计:") for emo, count in sorted(emotion_count.items(), key=lambda x: x[1], reverse=True): print(f" {emo:10}: {count} 次")

4. 完整的二次开发示例：构建情感分析仪表板

以下是一个实用的二次开发示例，展示如何将Emotion2Vec+ Large的结果集成到实际应用中：

import json import numpy as np from pathlib import Path from datetime import datetime import matplotlib.pyplot as plt import seaborn as sns class EmotionAnalysisDashboard: """情感分析结果可视化仪表板""" def __init__(self, outputs_dir: str): self.outputs_dir = Path(outputs_dir) self.results = self._load_all_results() def _load_all_results(self) -> list: """加载所有结果文件""" results = [] for result_file in self.outputs_dir.glob("outputs_*/result.json"): try: with open(result_file, 'r', encoding='utf-8') as f: data = json.load(f) # 添加时间戳信息 timestamp = datetime.fromisoformat(data["timestamp"].replace(' ', 'T')) data["datetime"] = timestamp results.append(data) except Exception as e: print(f"跳过文件 {result_file}: {e}") return sorted(results, key=lambda x: x["datetime"]) def plot_emotion_trend(self, days_back: int = 7): """绘制最近N天的情感趋势图""" if not self.results: print("没有可用的结果数据") return # 过滤最近N天的数据 cutoff_time = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) recent_results = [ r for r in self.results if (cutoff_time - r["datetime"]).days <= days_back ] if not recent_results: print(f"最近{days_back}天内没有分析结果") return # 准备数据 dates = [r["datetime"].strftime("%m/%d") for r in recent_results] emotions = [r["emotion"] for r in recent_results] # 绘制趋势图 plt.figure(figsize=(12, 6)) plt.plot(dates, emotions, 'o-', linewidth=2, markersize=6, alpha=0.7) plt.title(f'最近{days_back}天情感分析趋势', fontsize=14, fontweight='bold') plt.xlabel('日期', fontsize=12) plt.ylabel('主要情感', fontsize=12) plt.xticks(rotation=45) plt.grid(True, alpha=0.3) plt.tight_layout() plt.show() def generate_summary_report(self): """生成综合分析报告""" if not self.results: return " 没有分析结果数据" total_count = len(self.results) latest_result = self.results[-1] # 情感分布统计 emotion_counts = {} for r in self.results: emo = r["emotion"] emotion_counts[emo] = emotion_counts.get(emo, 0) + 1 # 置信度统计 confidences = [r["confidence"] for r in self.results] avg_confidence = sum(confidences) / len(confidences) # 生成报告 report = [] report.append("=" * 50) report.append(" Emotion2Vec+ Large 分析仪表板报告") report.append("=" * 50) report.append(f"总分析次数: {total_count}") report.append(f"最新分析时间: {latest_result['datetime'].strftime('%Y-%m-%d %H:%M:%S')}") report.append(f"平均置信度: {avg_confidence:.2%}") report.append("") report.append(" 情感分布:") # 按频率排序显示 sorted_emotions = sorted(emotion_counts.items(), key=lambda x: x[1], reverse=True) for emo, count in sorted_emotions: percentage = count / total_count * 100 report.append(f" {emo:10}: {count:3d} 次 ({percentage:.1f}%)") report.append("") report.append(" 分析建议:") if avg_confidence < 0.7: report.append(" 平均置信度偏低，建议检查音频质量") elif len(emotion_counts) == 1: report.append(" 情感分布单一，模型表现稳定") else: report.append(" 情感分布多样，适合进一步分析用户偏好") report.append("=" * 50) return "\n".join(report) # 使用示例 if __name__ == "__main__": # 创建仪表板实例 dashboard = EmotionAnalysisDashboard("outputs/") # 生成文本报告 print(dashboard.generate_summary_report()) # 绘制趋势图（如果matplotlib可用） try: dashboard.plot_emotion_trend(days_back=7) except ImportError: print(" matplotlib未安装，跳过图表绘制")

这个仪表板展示了如何将基础读取功能升级为实际业务应用：

• 数据聚合：自动发现和加载所有结果文件
• 时间序列分析：支持按时间维度分析情感变化趋势
• 统计报告：生成可读性强的业务报告
• 智能建议：基于数据分析结果提供实用建议

5. 最佳实践与常见问题解决

5.1 生产环境最佳实践

在将Emotion2Vec+ Large集成到生产系统时，建议遵循以下最佳实践：

import logging from pathlib import Path import json import time # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('emotion_analysis.log'), logging.StreamHandler() ] ) class ProductionEmotionReader: """生产环境就绪的情感结果读取器""" def __init__(self, timeout_seconds: int = 30): self.timeout_seconds = timeout_seconds self.logger = logging.getLogger(__name__) def safe_read_result(self, json_path: str, max_retries: int = 3) -> dict: """ 安全读取结果文件，包含重试机制和超时控制 """ file_path = Path(json_path) for attempt in range(max_retries): try: # 检查文件是否已完全写入（避免读取不完整文件） if not file_path.exists(): self.logger.warning(f"文件不存在，尝试 {attempt + 1}/{max_retries}: {json_path}") time.sleep(1) continue # 等待文件大小稳定（防止读取正在写入的文件） initial_size = file_path.stat().st_size time.sleep(0.1) final_size = file_path.stat().st_size if initial_size != final_size: self.logger.info(f"文件仍在写入中，重试 {attempt + 1}/{max_retries}") time.sleep(0.5) continue # 读取文件 with open(file_path, 'r', encoding='utf-8') as f: result = json.load(f) self.logger.info(f" 成功读取结果: {json_path}") return result except json.JSONDecodeError as e: self.logger.error(f"JSON解析错误 {json_path}: {e}") break except Exception as e: self.logger.error(f"读取文件错误 {json_path}: {e}") if attempt < max_retries - 1: time.sleep(1) raise RuntimeError(f"无法在{max_retries}次尝试内读取结果文件: {json_path}") # 使用示例 reader = ProductionEmotionReader() try: result = reader.safe_read_result("outputs/outputs_20240104_223000/result.json") print(f"处理成功: {result['emotion']}") except Exception as e: print(f"处理失败: {e}")

5.2 常见问题排查指南

5.3 性能优化建议

对于高并发场景，可以采用以下优化策略：

import threading from functools import lru_cache from pathlib import Path class OptimizedEmotionReader: """优化版读取器，支持缓存和并发""" def __init__(self, cache_size: int = 128): self.cache_size = cache_size self._lock = threading.Lock() @lru_cache(maxsize=128) def _cached_read_json(self, file_path: str) -> dict: """带缓存的JSON读取""" with open(file_path, 'r', encoding='utf-8') as f: return json.load(f) def read_with_cache(self, json_path: str) -> dict: """线程安全的缓存读取""" file_path = str(Path(json_path).resolve()) return self._cached_read_json(file_path) def batch_read_parallel(self, json_paths: list) -> list: """并行批量读取""" import concurrent.futures results = [] with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor: future_to_path = { executor.submit(self.read_with_cache, path): path for path in json_paths } for future in concurrent.futures.as_completed(future_to_path): try: result = future.result() results.append(result) except Exception as e: print(f"读取失败: {e}") return results

获取更多AI镜像

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