Chordia / docs /API_REFERENCE.md
Corolin's picture
first commit
0a6452f
# API参考文档
本文档详细介绍了情绪与生理状态变化预测模型的所有API接口、类和函数。
## 目录
1. [模型类](#模型类)
2. [数据处理类](#数据处理类)
3. [工具类](#工具类)
4. [损失函数](#损失函数)
5. [评估指标](#评估指标)
6. [工厂函数](#工厂函数)
7. [命令行接口](#命令行接口)
## 模型类
### `PADPredictor`
基于多层感知机的情绪与生理状态变化预测器。
```python
class PADPredictor(nn.Module):
def __init__(self,
input_dim: int = 7,
output_dim: int = 3,
hidden_dims: list = [512, 256, 128],
dropout_rate: float = 0.3,
weight_init: str = "xavier_uniform",
bias_init: str = "zeros")
```
#### 参数
- `input_dim` (int): 输入维度,默认为7(用户PAD 3维 + Vitality 1维 + AI当前PAD 3维)
- `output_dim` (int): 输出维度,默认为3(ΔPAD 3维,压力通过公式动态计算)
- `hidden_dims` (list): 隐藏层维度列表,默认为[512, 256, 128]
- `dropout_rate` (float): Dropout概率,默认为0.3
- `weight_init` (str): 权重初始化方法,默认为"xavier_uniform"
- `bias_init` (str): 偏置初始化方法,默认为"zeros"
#### 方法
##### `forward(self, x: torch.Tensor) -> torch.Tensor`
前向传播。
**参数:**
- `x` (torch.Tensor): 输入张量,形状为 (batch_size, input_dim)
**返回:**
- `torch.Tensor`: 输出张量,形状为 (batch_size, output_dim)
**示例:**
```python
import torch
from src.models.pad_predictor import PADPredictor
model = PADPredictor()
input_data = torch.randn(4, 7) # batch_size=4, input_dim=7
output = model(input_data)
print(f"Output shape: {output.shape}") # torch.Size([4, 3])
```
##### `predict_components(self, x: torch.Tensor) -> Dict[str, torch.Tensor]`
预测并分解输出组件。
**参数:**
- `x` (torch.Tensor): 输入张量
**返回:**
- `Dict[str, torch.Tensor]`: 包含各组件的字典
- `'delta_pad'`: ΔPAD (3维)
- `'delta_pressure'`: ΔPressure (1维,动态计算)
- `'confidence'`: Confidence (1维,可选)
**示例:**
```python
components = model.predict_components(input_data)
print(f"ΔPAD shape: {components['delta_pad'].shape}") # torch.Size([4, 3])
print(f"ΔPressure shape: {components['delta_pressure'].shape}") # torch.Size([4, 1])
print(f"Confidence shape: {components['confidence'].shape}") # torch.Size([4, 1])
```
##### `get_model_info(self) -> Dict[str, Any]`
获取模型信息。
**返回:**
- `Dict[str, Any]`: 包含模型信息的字典
**示例:**
```python
info = model.get_model_info()
print(f"Model type: {info['model_type']}")
print(f"Total parameters: {info['total_parameters']}")
print(f"Trainable parameters: {info['trainable_parameters']}")
```
##### `save_model(self, filepath: str, include_optimizer: bool = False, optimizer: Optional[torch.optim.Optimizer] = None)`
保存模型到文件。
**参数:**
- `filepath` (str): 保存路径
- `include_optimizer` (bool): 是否包含优化器状态,默认为False
- `optimizer` (Optional[torch.optim.Optimizer]): 优化器对象
**示例:**
```python
model.save_model("model.pth", include_optimizer=True, optimizer=optimizer)
```
##### `load_model(cls, filepath: str, device: str = 'cpu') -> 'PADPredictor'`
从文件加载模型。
**参数:**
- `filepath` (str): 模型文件路径
- `device` (str): 设备类型,默认为'cpu'
**返回:**
- `PADPredictor`: 加载的模型实例
**示例:**
```python
loaded_model = PADPredictor.load_model("model.pth", device='cuda')
```
##### `freeze_layers(self, layer_names: list = None)`
冻结指定层的参数。
**参数:**
- `layer_names` (list): 要冻结的层名称列表,如果为None则冻结所有层
**示例:**
```python
# 冻结所有层
model.freeze_layers()
# 冻结特定层
model.freeze_layers(['network.0.weight', 'network.2.weight'])
```
##### `unfreeze_layers(self, layer_names: list = None)`
解冻指定层的参数。
**参数:**
- `layer_names` (list): 要解冻的层名称列表,如果为None则解冻所有层
## 数据处理类
### `DataPreprocessor`
数据预处理器,负责特征标准化和标签处理。
```python
class DataPreprocessor:
def __init__(self,
feature_scaler: str = "standard",
label_scaler: str = "standard",
feature_range: tuple = None,
label_range: tuple = None)
```
#### 参数
- `feature_scaler` (str): 特征标准化方法,默认为"standard"
- `label_scaler` (str): 标签标准化方法,默认为"standard"
- `feature_range` (tuple): 特征范围,用于MinMax缩放
- `label_range` (tuple): 标签范围,用于MinMax缩放
#### 方法
##### `fit(self, features: np.ndarray, labels: np.ndarray) -> 'DataPreprocessor'`
拟合预处理器参数。
**参数:**
- `features` (np.ndarray): 训练特征数据
- `labels` (np.ndarray): 训练标签数据
**返回:**
- `DataPreprocessor`: 自身实例
##### `transform(self, features: np.ndarray, labels: np.ndarray = None) -> tuple`
转换数据。
**参数:**
- `features` (np.ndarray): 输入特征数据
- `labels` (np.ndarray, optional): 输入标签数据
**返回:**
- `tuple`: (转换后的特征, 转换后的标签)
##### `fit_transform(self, features: np.ndarray, labels: np.ndarray = None) -> tuple`
拟合并转换数据。
##### `inverse_transform(self, features: np.ndarray, labels: np.ndarray = None) -> tuple`
逆转换数据。
##### `save(self, filepath: str)`
保存预处理器到文件。
##### `load(cls, filepath: str) -> 'DataPreprocessor'`
从文件加载预处理器。
**示例:**
```python
from src.data.preprocessor import DataPreprocessor
# 创建预处理器
preprocessor = DataPreprocessor(
feature_scaler="standard",
label_scaler="standard"
)
# 拟合和转换数据
processed_features, processed_labels = preprocessor.fit_transform(train_features, train_labels)
# 保存预处理器
preprocessor.save("preprocessor.pkl")
# 加载预处理器
loaded_preprocessor = DataPreprocessor.load("preprocessor.pkl")
```
### `SyntheticDataGenerator`
合成数据生成器,用于生成训练和测试数据。
```python
class SyntheticDataGenerator:
def __init__(self,
num_samples: int = 1000,
seed: int = 42,
noise_level: float = 0.1,
correlation_strength: float = 0.5)
```
#### 参数
- `num_samples` (int): 生成的样本数量,默认为1000
- `seed` (int): 随机种子,默认为42
- `noise_level` (float): 噪声水平,默认为0.1
- `correlation_strength` (float): 相关性强度,默认为0.5
#### 方法
##### `generate_data(self) -> tuple`
生成合成数据。
**返回:**
- `tuple`: (特征数据, 标签数据)
##### `save_data(self, features: np.ndarray, labels: np.ndarray, filepath: str, format: str = 'csv')`
保存数据到文件。
**示例:**
```python
from src.data.synthetic_generator import SyntheticDataGenerator
# 创建数据生成器
generator = SyntheticDataGenerator(num_samples=1000, seed=42)
# 生成数据
features, labels = generator.generate_data()
# 保存数据
generator.save_data(features, labels, "synthetic_data.csv", format='csv')
```
### `EmotionDataset`
PyTorch数据集类,用于情绪预测任务。
```python
class EmotionDataset(Dataset):
def __init__(self,
features: np.ndarray,
labels: np.ndarray,
transform: callable = None)
```
#### 参数
- `features` (np.ndarray): 特征数据
- `labels` (np.ndarray): 标签数据
- `transform` (callable): 数据变换函数
## 工具类
### `InferenceEngine`
推理引擎,提供高性能的模型推理功能。
```python
class InferenceEngine:
def __init__(self,
model: nn.Module,
preprocessor: DataPreprocessor = None,
device: str = 'auto')
```
#### 方法
##### `predict(self, input_data: Union[list, np.ndarray]) -> Dict[str, Any]`
单样本预测。
**参数:**
- `input_data`: 输入数据,可以是列表或NumPy数组
**返回:**
- `Dict[str, Any]`: 预测结果字典
**示例:**
```python
from src.utils.inference_engine import create_inference_engine
# 创建推理引擎
engine = create_inference_engine(
model_path="model.pth",
preprocessor_path="preprocessor.pkl"
)
# 单样本预测
input_data = [0.5, 0.3, -0.2, 75.0, 0.1, 0.4, -0.1]
result = engine.predict(input_data)
print(f"ΔPAD: {result['delta_pad']}")
print(f"Confidence: {result['confidence']}")
```
##### `predict_batch(self, input_batch: Union[list, np.ndarray]) -> List[Dict[str, Any]]`
批量预测。
##### `benchmark(self, num_samples: int = 1000, batch_size: int = 32) -> Dict[str, float]`
性能基准测试。
**返回:**
- `Dict[str, float]`: 性能统计信息
**示例:**
```python
# 性能基准测试
stats = engine.benchmark(num_samples=1000, batch_size=32)
print(f"Throughput: {stats['throughput']:.2f} samples/sec")
print(f"Average latency: {stats['avg_latency']:.2f}ms")
```
### `ModelTrainer`
模型训练器,提供完整的训练流程管理。
```python
class ModelTrainer:
def __init__(self,
model: nn.Module,
preprocessor: DataPreprocessor = None,
device: str = 'auto')
```
#### 方法
##### `train(self, train_loader: DataLoader, val_loader: DataLoader, config: Dict[str, Any]) -> Dict[str, Any]`
训练模型。
**参数:**
- `train_loader` (DataLoader): 训练数据加载器
- `val_loader` (DataLoader): 验证数据加载器
- `config` (Dict[str, Any]): 训练配置
**返回:**
- `Dict[str, Any]`: 训练历史记录
**示例:**
```python
from src.utils.trainer import ModelTrainer
# 创建训练器
trainer = ModelTrainer(model, preprocessor)
# 训练配置
config = {
'epochs': 100,
'learning_rate': 0.001,
'weight_decay': 1e-4,
'patience': 10,
'save_dir': './models'
}
# 开始训练
history = trainer.train(train_loader, val_loader, config)
```
##### `evaluate(self, test_loader: DataLoader) -> Dict[str, float]`
评估模型。
## 损失函数
### `WeightedMSELoss`
加权均方误差损失函数。
```python
class WeightedMSELoss(nn.Module):
def __init__(self,
delta_pad_weight: float = 1.0,
delta_pressure_weight: float = 1.0,
confidence_weight: float = 0.5,
reduction: str = 'mean')
```
#### 参数
- `delta_pad_weight` (float): ΔPAD损失权重,默认为1.0
- `delta_pressure_weight` (float): ΔPressure损失权重,默认为1.0
- `confidence_weight` (float): 置信度损失权重,默认为0.5
- `reduction` (str): 损失缩减方式,默认为'mean'
**示例:**
```python
from src.models.loss_functions import WeightedMSELoss
criterion = WeightedMSELoss(
delta_pad_weight=1.0,
delta_pressure_weight=1.0,
confidence_weight=0.5
)
loss = criterion(predictions, targets)
```
### `ConfidenceLoss`
置信度损失函数。
```python
class ConfidenceLoss(nn.Module):
def __init__(self, reduction: str = 'mean')
```
## 评估指标
### `RegressionMetrics`
回归评估指标计算器。
```python
class RegressionMetrics:
def __init__(self)
```
#### 方法
##### `calculate_all_metrics(self, y_true: np.ndarray, y_pred: np.ndarray) -> Dict[str, float]`
计算所有回归指标。
**参数:**
- `y_true` (np.ndarray): 真实值
- `y_pred` (np.ndarray): 预测值
**返回:**
- `Dict[str, float]`: 包含所有指标的字典
**示例:**
```python
from src.models.metrics import RegressionMetrics
metrics_calculator = RegressionMetrics()
metrics = metrics_calculator.calculate_all_metrics(true_labels, predictions)
print(f"MSE: {metrics['mse']:.4f}")
print(f"MAE: {metrics['mae']:.4f}")
print(f"R²: {metrics['r2']:.4f}")
```
### `PADMetrics`
PAD专用评估指标。
```python
class PADMetrics:
def __init__(self)
```
#### 方法
##### `evaluate_predictions(self, predictions: np.ndarray, targets: np.ndarray) -> Dict[str, Any]`
评估PAD预测结果。
## 工厂函数
### `create_pad_predictor(config: Optional[Dict[str, Any]] = None) -> PADPredictor`
创建PAD预测器的工厂函数。
**参数:**
- `config` (Dict[str, Any], optional): 配置字典
**返回:**
- `PADPredictor`: PAD预测器实例
**示例:**
```python
from src.models.pad_predictor import create_pad_predictor
# 使用默认配置
model = create_pad_predictor()
# 使用自定义配置
config = {
'dimensions': {
'input_dim': 7,
'output_dim': 4或3
},
'architecture': {
'hidden_layers': [
{'size': 256, 'activation': 'ReLU', 'dropout': 0.3},
{'size': 128, 'activation': 'ReLU', 'dropout': 0.2}
]
}
}
model = create_pad_predictor(config)
```
### `create_inference_engine(model_path: str, preprocessor_path: str = None, device: str = 'auto') -> InferenceEngine`
创建推理引擎的工厂函数。
**参数:**
- `model_path` (str): 模型文件路径
- `preprocessor_path` (str, optional): 预处理器文件路径
- `device` (str): 设备类型
**返回:**
- `InferenceEngine`: 推理引擎实例
### `create_training_setup(config: Dict[str, Any]) -> tuple`
创建训练设置的工厂函数。
**参数:**
- `config` (Dict[str, Any]): 训练配置
**返回:**
- `tuple`: (模型, 训练器, 数据加载器)
## 命令行接口
### 主CLI工具
项目提供了统一的命令行接口,支持多种操作:
```bash
emotion-prediction <command> [options]
```
#### 可用命令
- `train`: 训练模型
- `predict`: 进行预测
- `evaluate`: 评估模型
- `inference`: 推理脚本
- `benchmark`: 性能基准测试
#### 训练命令
```bash
emotion-prediction train --config CONFIG_FILE [OPTIONS]
```
**参数:**
- `--config, -c`: 训练配置文件路径(必需)
- `--output-dir, -o`: 输出目录(默认: ./outputs)
- `--device`: 计算设备(auto/cpu/cuda,默认: auto)
- `--resume`: 从检查点恢复训练
- `--epochs`: 覆盖训练轮数
- `--batch-size`: 覆盖批次大小
- `--learning-rate`: 覆盖学习率
- `--seed`: 随机种子(默认: 42)
- `--verbose, -v`: 详细输出
- `--log-level`: 日志级别(DEBUG/INFO/WARNING/ERROR)
**示例:**
```bash
# 基础训练
emotion-prediction train --config configs/training_config.yaml
# GPU训练
emotion-prediction train --config configs/training_config.yaml --device cuda
# 从检查点恢复
emotion-prediction train --config configs/training_config.yaml --resume checkpoint.pth
```
#### 预测命令
```bash
emotion-prediction predict --model MODEL_FILE [OPTIONS]
```
**参数:**
- `--model, -m`: 模型文件路径(必需)
- `--preprocessor, -p`: 预处理器文件路径
- `--interactive, -i`: 交互式模式
- `--quick`: 快速预测模式(7个数值)
- `--batch`: 批量预测模式(输入文件)
- `--output, -o`: 输出文件路径
- `--device`: 计算设备
- `--verbose, -v`: 详细输出
- `--log-level`: 日志级别
**示例:**
```bash
# 交互式预测
emotion-prediction predict --model model.pth --interactive
# 快速预测
emotion-prediction predict --model model.pth --quick 0.5 0.3 -0.2 75.0 0.1 0.4 -0.1
# 批量预测
emotion-prediction predict --model model.pth --batch input.csv --output results.csv
```
#### 评估命令
```bash
emotion-prediction evaluate --model MODEL_FILE --data DATA_FILE [OPTIONS]
```
**参数:**
- `--model, -m`: 模型文件路径(必需)
- `--data, -d`: 测试数据文件路径(必需)
- `--preprocessor, -p`: 预处理器文件路径
- `--output, -o`: 评估结果输出路径
- `--report`: 生成详细报告文件路径
- `--metrics`: 评估指标列表(默认: mse mae r2)
- `--batch-size`: 批次大小(默认: 32)
- `--device`: 计算设备
- `--verbose, -v`: 详细输出
- `--log-level`: 日志级别
**示例:**
```bash
# 基础评估
emotion-prediction evaluate --model model.pth --data test_data.csv
# 生成详细报告
emotion-prediction evaluate --model model.pth --data test_data.csv --report report.html
```
#### 基准测试命令
```bash
emotion-prediction benchmark --model MODEL_FILE [OPTIONS]
```
**参数:**
- `--model, -m`: 模型文件路径(必需)
- `--preprocessor, -p`: 预处理器文件路径
- `--num-samples`: 测试样本数量(默认: 1000)
- `--batch-size`: 批次大小(默认: 32)
- `--device`: 计算设备
- `--report`: 生成性能报告文件路径
- `--warmup`: 预热轮数(默认: 10)
- `--verbose, -v`: 详细输出
- `--log-level`: 日志级别
**示例:**
```bash
# 标准基准测试
emotion-prediction benchmark --model model.pth
# 自定义测试
emotion-prediction benchmark --model model.pth --num-samples 5000 --batch-size 64
```
## 配置文件API
### 模型配置
模型配置文件使用YAML格式,支持以下参数:
```yaml
# 模型基本信息
model_info:
name: str # 模型名称
type: str # 模型类型
version: str # 模型版本
# 输入输出维度
dimensions:
input_dim: int # 输入维度
output_dim: int # 输出维度
# 网络架构
architecture:
hidden_layers:
- size: int # 层大小
activation: str # 激活函数
dropout: float # Dropout率
output_layer:
activation: str # 输出激活函数
use_batch_norm: bool # 是否使用批归一化
use_layer_norm: bool # 是否使用层归一化
# 初始化参数
initialization:
weight_init: str # 权重初始化方法
bias_init: str # 偏置初始化方法
# 正则化
regularization:
weight_decay: float # L2正则化系数
dropout_config:
type: str # Dropout类型
rate: float # Dropout率
```
### 训练配置
训练配置文件支持以下参数:
```yaml
# 训练信息
training_info:
experiment_name: str # 实验名称
description: str # 实验描述
seed: int # 随机种子
# 训练超参数
training:
optimizer:
type: str # 优化器类型
learning_rate: float # 学习率
weight_decay: float # 权重衰减
scheduler:
type: str # 调度器类型
epochs: int # 训练轮数
early_stopping:
enabled: bool # 是否启用早停
patience: int # 耐心值
min_delta: float # 最小改善
```
## 异常处理
项目定义了以下自定义异常:
### `ModelLoadError`
模型加载错误。
### `DataPreprocessingError`
数据预处理错误。
### `InferenceError`
推理过程错误。
### `ConfigurationError`
配置文件错误。
**示例:**
```python
from src.utils.exceptions import ModelLoadError, InferenceError
try:
model = PADPredictor.load_model("invalid_model.pth")
except ModelLoadError as e:
print(f"模型加载失败: {e}")
try:
result = engine.predict(invalid_input)
except InferenceError as e:
print(f"推理失败: {e}")
```
## 日志系统
项目使用结构化日志系统:
```python
from src.utils.logger import setup_logger
import logging
# 设置日志
setup_logger(level='INFO', log_file='training.log')
logger = logging.getLogger(__name__)
# 使用日志
logger.info("训练开始")
logger.debug(f"批次大小: {batch_size}")
logger.warning("检测到潜在的过拟合")
logger.error("训练过程中发生错误")
```
## 类型提示
项目完全支持类型提示,所有公共API都有详细的类型注解:
```python
from typing import Dict, List, Optional, Union, Tuple
import numpy as np
import torch
def predict_emotion(
input_data: Union[List[float], np.ndarray],
model_path: str,
preprocessor_path: Optional[str] = None,
device: str = 'auto'
) -> Dict[str, Any]:
"""
预测情绪变化
Args:
input_data: 输入数据,7维向量
model_path: 模型文件路径
preprocessor_path: 预处理器文件路径
device: 计算设备
Returns:
包含预测结果的字典
Raises:
InferenceError: 推理失败时抛出
"""
pass
```
---
更多详细信息请参考源代码和示例文件。如有问题,请查看[故障排除指南](TUTORIAL.md#故障排除)或提交Issue。