lxcxjxhx's picture
Upload API.md with huggingface_hub
28d30c3 verified
|
Raw
History Blame Contribute Delete
35.3 kB
# API 文档
本文档提供 HOS Model Optimizer 的 Python API 参考,涵盖所有核心模块的类、方法和函数。
## 目录
- [量化模块 API](#量化模块-api)
- [推理模块 API](#推理模块-api)
- [训练模块 API](#训练模块-api)
- [部署模块 API](#部署模块-api)
- [配置模块 API](#配置模块-api)
- [评测模块 API](#评测模块-api)
- [工具函数 API](#工具函数-api)
---
## 量化模块 API
量化模块位于 `hos_optimizer.quantize`,提供多种量化方法和评估工具。
### 异常类
#### `QuantizationError`
量化过程中的异常。
```python
from hos_optimizer.quantize import QuantizationError
try:
quantize_gguf(...)
except QuantizationError as e:
print(f"量化失败: {e}")
```
### 核心函数
#### `quantize_gguf()`
GGUF 量化 - 使用 llama.cpp 工具链。
```python
def quantize_gguf(
model_path: str,
output_path: str,
quant_type: str = "Q4_K_M",
llama_cpp_path: Optional[str] = None
) -> str
```
**参数**
- `model_path` (str): 输入模型路径(HuggingFace 格式)
- `output_path` (str): 输出 GGUF 文件路径
- `quant_type` (str): 量化类型,如 `Q4_K_M`, `Q5_K_M`, `Q8_0` 等,默认 `Q4_K_M`
- `llama_cpp_path` (Optional[str]): llama.cpp 安装路径,为 None 则从 PATH 查找
**返回**`str` - 输出文件路径
**异常**`QuantizationError` - 量化失败时抛出
**示例**
```python
from hos_optimizer.quantize import quantize_gguf
output = quantize_gguf(
model_path="./model",
output_path="./model.gguf",
quant_type="Q4_K_M"
)
```
#### `quantize_awq()`
AWQ 4-bit 量化 - 激活感知量化,精度损失最小。
```python
def quantize_awq(
model_path: str,
output_path: str,
bits: int = 4,
group_size: int = 128
) -> str
```
**参数**
- `model_path` (str): 输入模型路径
- `output_path` (str): 输出模型路径
- `bits` (int): 量化位数,默认 4
- `group_size` (int): 分组大小,默认 128
**返回**`str` - 输出模型路径
**异常**`QuantizationError` - 量化失败时抛出
**示例**
```python
from hos_optimizer.quantize import quantize_awq
output = quantize_awq(
model_path="./model",
output_path="./model-awq",
bits=4,
group_size=128
)
```
#### `quantize_gptq()`
GPTQ 量化 - 基于 Optimal Brain Quantization 框架。
```python
def quantize_gptq(
model_path: str,
output_path: str,
bits: int = 4,
group_size: int = 128,
desc_act: bool = False
) -> str
```
**参数**
- `model_path` (str): 输入模型路径
- `output_path` (str): 输出模型路径
- `bits` (int): 量化位数,4 或 8
- `group_size` (int): 分组大小,默认 128
- `desc_act` (bool): 是否按激活值排序,默认 False
**返回**`str` - 输出模型路径
**异常**`QuantizationError` - 量化失败时抛出
**示例**
```python
from hos_optimizer.quantize import quantize_gptq
output = quantize_gptq(
model_path="./model",
output_path="./model-gptq",
bits=4,
group_size=128
)
```
#### `evaluate_perplexity()`
评估量化模型的 PPL(Perplexity)。
```python
def evaluate_perplexity(
model_path: str,
dataset: str = "wikitext",
max_samples: int = 100,
stride: int = 512
) -> float
```
**参数**
- `model_path` (str): 模型路径
- `dataset` (str): 评估数据集名称,默认 `wikitext`
- `max_samples` (int): 最大评估样本数
- `stride` (int): 滑动窗口步长
**返回**`float` - PPL 值
**异常**`QuantizationError` - 评估失败时抛出
**示例**
```python
from hos_optimizer.quantize import evaluate_perplexity
ppl = evaluate_perplexity(
model_path="./model-awq",
dataset="wikitext",
max_samples=100
)
print(f"PPL: {ppl:.2f}")
```
#### `convert_format()`
量化格式转换工具。
```python
def convert_format(
model_path: str,
output_path: str,
from_format: str,
to_format: str,
**kwargs
) -> str
```
**参数**
- `model_path` (str): 输入模型路径
- `output_path` (str): 输出模型路径
- `from_format` (str): 源格式 (`gguf`, `awq`, `gptq`, `hf`)
- `to_format` (str): 目标格式 (`gguf`, `awq`, `gptq`, `hf`)
- `**kwargs`: 其他参数(如 `bits`, `quant_type`
**返回**`str` - 输出模型路径
**异常**`QuantizationError` - 转换失败时抛出
**示例**
```python
from hos_optimizer.quantize import convert_format
output = convert_format(
model_path="./model",
output_path="./model-awq",
from_format="hf",
to_format="awq",
bits=4
)
```
### 辅助函数
#### `check_vram_availability()`
检查 GPU VRAM 可用性。
```python
def check_vram_availability() -> Dict[str, Any]
```
**返回**`Dict[str, Any]` - 包含 VRAM 信息的字典
```python
{
"available": True,
"total_vram_gb": 8.0,
"free_vram_gb": 6.5,
"device": "NVIDIA GeForce RTX 3070"
}
```
#### `optimize_for_low_vram()`
根据 VRAM 限制优化配置。
```python
def optimize_for_low_vram(config: Dict[str, Any]) -> Dict[str, Any]
```
**参数**:
- `config` (Dict[str, Any]): 原始配置
**返回**:`Dict[str, Any]` - 优化后的配置
#### `get_model_size()`
获取模型文件大小(GB)。
```python
def get_model_size(model_path: str) -> float
```
**参数**
- `model_path` (str): 模型路径
**返回**`float` - 模型大小(GB)
---
## 推理模块 API
推理模块位于 `hos_optimizer.inference`,提供统一的推理接口和三个后端实现。
### 数据结构
#### `InferenceRequest`
推理请求数据结构。
```python
@dataclass
class InferenceRequest:
prompt: str
max_tokens: int = 256
temperature: float = 0.7
top_p: float = 0.9
top_k: int = 50
stop: Optional[List[str]] = None
json_schema: Optional[Dict[str, Any]] = None # SGLang 约束生成
extra: Optional[Dict[str, Any]] = None
```
#### `InferenceResult`
推理结果数据结构。
```python
@dataclass
class InferenceResult:
text: str
token_ids: List[int] = field(default_factory=list)
prompt: str = ""
latency_ms: float = 0.0
tokens_per_second: float = 0.0
prompt_tokens: int = 0
completion_tokens: int = 0
metadata: Dict[str, Any] = field(default_factory=dict)
```
#### `PerformanceStats`
性能统计数据结构。
```python
@dataclass
class PerformanceStats:
total_requests: int = 0
total_prompt_tokens: int = 0
total_completion_tokens: int = 0
total_latency_ms: float = 0.0
peak_vram_mb: float = 0.0
wall_time_s: float = 0.0
@property
def avg_latency_ms(self) -> float: ...
@property
def throughput_tokens_per_s(self) -> float: ...
@property
def requests_per_s(self) -> float: ...
def summary(self) -> str: ...
```
### 核心类
#### `UnifiedInferenceEngine`
统一推理引擎,封装后端选择和配置优化逻辑。
```python
class UnifiedInferenceEngine:
def __init__(
self,
model_path: str,
backend: Optional[str] = None,
auto_load: bool = True,
**kwargs
)
```
**参数**
- `model_path` (str): 模型路径或 HF 仓库 ID
- `backend` (Optional[str]): 推理后端名称(`llama_cpp`/`vllm`/`sglang`),为 None 时自动检测
- `auto_load` (bool): 是否自动加载模型,默认 True
- `**kwargs`: 传递给后端的额外参数
**方法**
##### `generate()`
单次推理生成。
```python
def generate(
self,
prompt: str,
max_tokens: int = 256,
temperature: float = 0.7,
top_p: float = 0.9,
top_k: int = 50,
stop: Optional[List[str]] = None,
json_schema: Optional[Dict[str, Any]] = None,
**extra
) -> InferenceResult
```
**参数**
- `prompt` (str): 输入提示文本
- `max_tokens` (int): 最大生成 token 数
- `temperature` (float): 采样温度(0 = 贪婪解码)
- `top_p` (float): nucleus sampling 参数
- `top_k` (int): top-k sampling 参数
- `stop` (Optional[List[str]]): 停止词列表
- `json_schema` (Optional[Dict[str, Any]]): JSON Schema 约束(仅 SGLang 支持)
- `**extra`: 传递给后端的额外参数
**返回**`InferenceResult` - 推理结果
##### `generate_batch()`
批量推理生成。
```python
def generate_batch(
self,
prompts: List[str],
max_tokens: int = 256,
temperature: float = 0.7,
top_p: float = 0.9,
**extra
) -> List[InferenceResult]
```
**参数**
- `prompts` (List[str]): 输入提示文本列表
- `max_tokens` (int): 最大生成 token 数
- `temperature` (float): 采样温度
- `top_p` (float): nucleus sampling 参数
- `**extra`: 传递给后端的额外参数
**返回**`List[InferenceResult]` - 推理结果列表
##### `get_stats()`
获取性能统计。
```python
def get_stats(self) -> PerformanceStats
```
**返回**:`PerformanceStats` - 性能统计
##### `shutdown()`
关闭引擎,释放资源。
```python
def shutdown(self) -> None
```
**属性**:
##### `backend_name`
当前使用的后端名称。
```python
@property
def backend_name(self) -> str
```
**示例**
```python
from hos_optimizer.inference import UnifiedInferenceEngine
# 创建引擎
engine = UnifiedInferenceEngine(
model_path="./model",
backend="vllm",
gpu_memory_utilization=0.9
)
# 单次推理
result = engine.generate("你好", max_tokens=256)
print(result.text)
# 批量推理
results = engine.generate_batch(["问题1", "问题2"])
# 性能统计
stats = engine.get_stats()
print(stats.summary())
# 关闭
engine.shutdown()
```
### 后端类
#### `LlamaCppBackend`
llama-cpp-python 推理后端。
```python
class LlamaCppBackend(InferenceBackend):
def __init__(self, model_path: str, **kwargs)
```
**参数**:
- `model_path` (str): GGUF 模型路径
- `n_gpu_layers` (int): GPU offload 层数,-1 表示全部,默认 -1
- `n_ctx` (int): 上下文长度,默认 4096
- `n_threads` (int): CPU 线程数
- `use_mmap` (bool): 是否使用 mmap,默认 True
- `use_mlock` (bool): 是否使用 mlock,默认 False
- `n_batch` (int): 批处理大小,默认 512
**方法**:
- `load()`: 加载模型
- `generate(request: InferenceRequest) -> InferenceResult`: 单次推理
- `generate_batch(requests: List[InferenceRequest]) -> List[InferenceResult]`: 批量推理
- `serve(host: str = "0.0.0.0", port: int = 8080)`: 启动 API 服务
- `shutdown()`: 释放资源
#### `VLLMBackend`
vLLM 推理后端。
```python
class VLLMBackend(InferenceBackend):
def __init__(self, model_path: str, **kwargs)
```
**参数**:
- `model_path` (str): 模型路径
- `gpu_memory_utilization` (float): 显存利用率,默认 0.9
- `max_model_len` (int): 最大模型长度,默认 4096
- `max_num_seqs` (int): 最大并发序列数,默认 128
- `dtype` (str): 推理精度,默认 `float16`
- `tensor_parallel_size` (int): tensor parallel 数量,默认 1
- `enforce_eager` (bool): 是否使用 eager 模式,默认 False
- `trust_remote_code` (bool): 是否信任远程代码,默认 True
**方法**:
- `load()`: 加载模型
- `generate(request: InferenceRequest) -> InferenceResult`: 单次推理
- `generate_batch(requests: List[InferenceRequest]) -> List[InferenceResult]`: 批量推理
- `serve(host: str = "0.0.0.0", port: int = 8000)`: 启动 API 服务
- `shutdown()`: 释放资源
#### `SGLangBackend`
SGLang 推理后端。
```python
class SGLangBackend(InferenceBackend):
def __init__(self, model_path: str, **kwargs)
```
**参数**
- `model_path` (str): 模型路径
- `mem_fraction_static` (float): 静态显存分配比例,默认 0.9
- `context_length` (int): 上下文长度,默认 4096
- `tp_size` (int): tensor parallel 数量,默认 1
- `trust_remote_code` (bool): 是否信任远程代码,默认 True
**方法**
- `load()`: 加载模型
- `generate(request: InferenceRequest) -> InferenceResult`: 单次推理(支持 JSON Schema 约束)
- `generate_batch(requests: List[InferenceRequest]) -> List[InferenceResult]`: 批量推理
- `serve(host: str = "0.0.0.0", port: int = 30000)`: 启动 API 服务
- `shutdown()`: 释放资源
### 性能监控
#### `PerformanceMonitor`
性能监控器,跟踪吞吐量、延迟和显存占用。
```python
class PerformanceMonitor:
def __init__(self)
def start(self) -> None
def end(self) -> None
def record_request(self, prompt_tokens: int, completion_tokens: int, latency_ms: float) -> None
def update_peak_vram(self) -> None
def reset(self) -> None
@property
def stats(self) -> PerformanceStats
```
### 工具函数
#### `detect_best_backend()`
自动检测最优推理后端。
```python
def detect_best_backend() -> str
```
**返回**`str` - 后端名称(`vllm`/`sglang`/`llama_cpp`
#### `get_gpu_memory_usage_mb()`
获取当前 GPU 显存占用(MB)。
```python
def get_gpu_memory_usage_mb() -> float
```
**返回**`float` - 显存占用(MB)
#### `get_total_gpu_memory_mb()`
获取 GPU 总显存(MB)。
```python
def get_total_gpu_memory_mb() -> float
```
**返回**`float` - 总显存(MB)
---
## 训练模块 API
训练模块位于 `hos_optimizer.train`,提供 QLoRA/LoRA 微调功能。
### 核心类
#### `TrainingConfig`
训练配置类。
```python
@dataclass
class TrainingConfig:
# 模型配置
model_name_or_path: str = "Qwen/Qwen2.5-0.5B"
trust_remote_code: bool = True
# 训练方法
finetuning_type: str = "qlora" # "qlora" 或 "lora"
# 数据集配置
dataset_path: str = ""
dataset_format: str = "alpaca" # "alpaca" 或 "sharegpt"
max_seq_length: int = 2048
preprocessing_num_workers: int = 4
# QLoRA 量化配置
use_4bit: bool = True
bnb_4bit_quant_type: str = "nf4"
bnb_4bit_use_double_quant: bool = True
bnb_4bit_compute_dtype: str = "bfloat16"
# LoRA 配置
lora_rank: int = 16
lora_alpha: int = 32
lora_dropout: float = 0.05
lora_target_modules: List[str] = field(default_factory=lambda: ["all"])
# 训练参数
output_dir: str = "./output"
num_train_epochs: int = 3
per_device_train_batch_size: int = 2
gradient_accumulation_steps: int = 8
learning_rate: float = 2e-4
lr_scheduler_type: str = "cosine"
warmup_ratio: float = 0.1
max_grad_norm: float = 1.0
weight_decay: float = 0.01
# 日志和保存
logging_steps: int = 10
save_steps: int = 500
save_total_limit: int = 3
# 精度配置
bf16: bool = True
fp16: bool = False
# 性能优化
gradient_checkpointing: bool = True
optim: str = "adamw_torch"
# 其他
seed: int = 42
use_unsloth: bool = True
```
#### `DatasetProcessor`
数据集处理器。
```python
class DatasetProcessor:
def __init__(self, tokenizer, max_seq_length: int = 2048)
def format_alpaca(self, example: Dict) -> Dict
def format_sharegpt(self, example: Dict) -> Dict
def tokenize_function(self, example: Dict) -> Dict
def process_dataset(self, dataset: Dataset, dataset_format: str = "alpaca") -> Dataset
```
### 核心函数
#### `train()`
执行训练。
```python
def train(config: TrainingConfig) -> None
```
**参数**
- `config` (TrainingConfig): 训练配置
**示例**
```python
from hos_optimizer.train import TrainingConfig, train
config = TrainingConfig(
model_name_or_path="./model",
dataset_path="./data.json",
finetuning_type="qlora",
num_train_epochs=3
)
train(config)
```
#### `merge_model()`
合并 LoRA 权重到基础模型。
```python
def merge_model(
base_model_path: str,
adapter_path: str,
output_path: str,
trust_remote_code: bool = True
) -> None
```
**参数**
- `base_model_path` (str): 基础模型路径
- `adapter_path` (str): LoRA adapter 路径
- `output_path` (str): 输出路径
- `trust_remote_code` (bool): 是否信任远程代码
**示例**
```python
from hos_optimizer.train import merge_model
merge_model(
base_model_path="./base",
adapter_path="./adapter",
output_path="./merged"
)
```
#### `load_and_process_dataset()`
加载并处理数据集。
```python
def load_and_process_dataset(
dataset_path: str,
tokenizer,
dataset_format: str = "alpaca",
max_seq_length: int = 2048,
test_size: float = 0.05
) -> DatasetDict
```
**参数**
- `dataset_path` (str): 数据集路径(JSON 文件)
- `tokenizer`: 分词器
- `dataset_format` (str): 数据格式(`alpaca``sharegpt`
- `max_seq_length` (int): 最大序列长度
- `test_size` (float): 测试集比例
**返回**`DatasetDict` - 包含 train 和 test 的 DatasetDict
---
## 部署模块 API
部署模块位于 `hos_optimizer.deploy`,提供硬件检测、自动配置和服务启动功能。
### 数据结构
#### `HardwareInfo`
硬件信息数据类。
```python
@dataclass
class HardwareInfo:
gpu_name: str
gpu_memory_gb: float
cuda_version: Optional[str]
cpu_cores: int
system_memory_gb: float
```
#### `DeploymentConfig`
部署配置数据类。
```python
@dataclass
class DeploymentConfig:
backend: Backend
quantization: Quantization
max_model_size_gb: float
recommended_for: str
startup_args: Dict
```
### 核心类
#### `HardwareDetector`
硬件检测器。
```python
class HardwareDetector:
@staticmethod
def detect_gpu() -> Tuple[str, float, Optional[str]]
@staticmethod
def detect_cpu() -> int
@staticmethod
def detect_system_memory() -> float
@classmethod
def get_hardware_info(cls) -> HardwareInfo
```
**示例**
```python
from hos_optimizer.deploy import HardwareDetector
hardware = HardwareDetector.get_hardware_info()
print(f"GPU: {hardware.gpu_name}")
print(f"VRAM: {hardware.gpu_memory_gb:.1f}GB")
print(f"CPU cores: {hardware.cpu_cores}")
```
#### `ConfigSelector`
自动配置选择器。
```python
class ConfigSelector:
@staticmethod
def select_config(
hardware: HardwareInfo,
model_size_b: float,
use_case: str = "general"
) -> DeploymentConfig
```
**参数**
- `hardware` (HardwareInfo): 硬件信息
- `model_size_b` (float): 模型大小(十亿参数)
- `use_case` (str): 使用场景(`general`, `high_concurrency`, `multi_turn`
**返回**`DeploymentConfig` - 推荐的部署配置
**示例**
```python
from hos_optimizer.deploy import HardwareDetector, ConfigSelector
hardware = HardwareDetector.get_hardware_info()
config = ConfigSelector.select_config(hardware, model_size_b=7.0, use_case="general")
print(f"推荐: {config.recommended_for}")
```
#### `ServiceLauncher`
服务启动器。
```python
class ServiceLauncher:
def __init__(self, config: DeploymentConfig, model_path: str, host: str = "0.0.0.0", port: int = 8000)
def start(self) -> bool
def stop(self) -> None
```
**方法**
- `start()`: 启动服务,返回是否成功
- `stop()`: 停止服务
**示例**
```python
from hos_optimizer.deploy import ServiceLauncher, ConfigSelector, HardwareDetector
hardware = HardwareDetector.get_hardware_info()
config = ConfigSelector.select_config(hardware, 7.0)
launcher = ServiceLauncher(config, "./model.gguf", "0.0.0.0", 8000)
if launcher.start():
print("服务已启动")
# ...
launcher.stop()
```
#### `HealthChecker`
服务健康检查器。
```python
class HealthChecker:
def __init__(self, host: str = "localhost", port: int = 8000)
def check_health(self, timeout: int = 5) -> bool
def check_model_loaded(self, timeout: int = 300) -> bool
def get_service_info(self) -> Optional[Dict]
```
**方法**
- `check_health()`: 检查服务健康状态
- `check_model_loaded()`: 检查模型是否加载完成
- `get_service_info()`: 获取服务信息
### 核心函数
#### `deploy_model()`
一键部署模型。
```python
def deploy_model(
model_path: str,
model_size_b: float = 7.0,
use_case: str = "general",
host: str = "0.0.0.0",
port: int = 8000,
auto_start: bool = True
) -> Optional[ServiceLauncher]
```
**参数**
- `model_path` (str): 模型文件路径
- `model_size_b` (float): 模型大小(十亿参数)
- `use_case` (str): 使用场景
- `host` (str): 服务主机地址
- `port` (int): 服务端口
- `auto_start` (bool): 是否自动启动服务
**返回**`Optional[ServiceLauncher]` - 服务启动器实例
**示例**
```python
from hos_optimizer.deploy import deploy_model
launcher = deploy_model(
model_path="./model.gguf",
model_size_b=7.0,
use_case="general",
auto_start=True
)
```
---
## 配置模块 API
配置模块位于 `hos_optimizer.config`,提供配置管理和优化功能。
### 异常类
#### `ConfigError`
配置相关异常的基类。
#### `ConfigValidationError`
配置验证失败时抛出。
#### `ConfigConflictError`
配置项之间存在冲突时抛出。
#### `TemplateNotFoundError`
请求的模板不存在时抛出。
### 核心类
#### `ConfigManager`
配置管理器,负责加载、生成、验证和管理配置。
```python
class ConfigManager:
def __init__(self, config_dir: Optional[str] = None)
```
**参数**
- `config_dir` (Optional[str]): 配置文件目录,默认为项目根目录下的 `configs/`
**方法**
##### `load_config()`
从 YAML 文件加载配置。
```python
def load_config(self, config_path: str) -> Dict[str, Any]
```
**参数**
- `config_path` (str): YAML 配置文件路径
**返回**`Dict[str, Any]` - 解析后的配置字典
**异常**`ConfigError` - 文件不存在或解析失败
##### `save_config()`
将配置保存为 YAML 文件。
```python
def save_config(self, config: Dict[str, Any], output_path: str) -> None
```
**参数**
- `config` (Dict[str, Any]): 配置字典
- `output_path` (str): 输出文件路径
##### `merge_configs()`
深度合并两个配置字典。
```python
def merge_configs(self, base: Dict[str, Any], override: Dict[str, Any]) -> Dict[str, Any]
```
**参数**:
- `base` (Dict[str, Any]): 基础配置
- `override` (Dict[str, Any]): 覆盖配置
**返回**:`Dict[str, Any]` - 合并后的新配置字典
##### `generate_optimal_config()`
根据使用场景和硬件条件生成最优配置。
```python
def generate_optimal_config(
self,
scenario: str,
model_path: str = "",
vram_gb: float = 8.0
) -> Dict[str, Any]
```
**参数**
- `scenario` (str): 场景名称,可选值:
- `inference_0.8b`
- `inference_7b`
- `training_0.8b`
- `training_7b`
- `high_throughput_serving`
- `multi_turn_dialogue`
- `model_path` (str): 模型路径
- `vram_gb` (float): 可用显存大小(GB)
**返回**`Dict[str, Any]` - 最优配置字典
**异常**`ConfigError` - 不支持的场景或显存不足
**示例**
```python
from hos_optimizer.config import ConfigManager
manager = ConfigManager()
config = manager.generate_optimal_config(
scenario="inference_7b",
model_path="./model",
vram_gb=8.0
)
```
##### `auto_select_scenario()`
根据模型大小和任务类型自动选择最优场景并生成配置。
```python
def auto_select_scenario(
self,
model_size_b: float,
task: str = "inference",
vram_gb: float = 8.0
) -> Dict[str, Any]
```
**参数**
- `model_size_b` (float): 模型大小(十亿参数)
- `task` (str): 任务类型(`inference`/`training`/`serving`/`dialogue`
- `vram_gb` (float): 可用显存(GB)
**返回**`Dict[str, Any]` - 自动选择的最优配置
##### `validate_config()`
验证配置并返回所有发现的问题列表。
```python
def validate_config(self, config: Dict[str, Any]) -> List[str]
```
**参数**:
- `config` (Dict[str, Any]): 待验证的配置字典
**返回**:`List[str]` - 问题描述字符串列表,空列表表示验证通过
##### `validate_and_raise()`
验证配置,如果发现问题则抛出异常。
```python
def validate_and_raise(self, config: Dict[str, Any]) -> None
```
**参数**:
- `config` (Dict[str, Any]): 待验证的配置
**异常**:
- `ConfigValidationError`: 存在验证问题
- `ConfigConflictError`: 存在配置冲突
##### `list_templates()`
列出所有可用的配置模板名称。
```python
def list_templates(self) -> List[str]
```
**返回**:`List[str]` - 模板名称列表
##### `get_template()`
获取指定名称的配置模板。
```python
def get_template(self, name: str) -> Dict[str, Any]
```
**参数**
- `name` (str): 模板名称
**返回**`Dict[str, Any]` - 模板配置字典
**异常**`TemplateNotFoundError` - 模板不存在
##### `register_template()`
注册自定义配置模板。
```python
def register_template(self, name: str, template: Dict[str, Any]) -> None
```
**参数**:
- `name` (str): 模板名称
- `template` (Dict[str, Any]): 模板配置字典
##### `unregister_template()`
注销自定义配置模板。
```python
def unregister_template(self, name: str) -> None
```
**参数**
- `name` (str): 模板名称
**异常**
- `ConfigError`: 尝试注销内置模板
- `TemplateNotFoundError`: 模板不存在
##### `export_template()`
将模板导出为 YAML 文件。
```python
def export_template(self, name: str, output_path: str) -> None
```
**参数**
- `name` (str): 模板名称
- `output_path` (str): 输出文件路径
##### `load_template_from_file()`
从 YAML 文件加载并注册为自定义模板。
```python
def load_template_from_file(self, name: str, file_path: str) -> None
```
**参数**
- `name` (str): 注册时使用的模板名称
- `file_path` (str): YAML 文件路径
---
## 评测模块 API
评测模块位于 `hos_optimizer.evaluate`,提供模型质量评测和多模型对比功能。
### 异常类
#### `EvaluationError`
评测相关异常的基类。
#### `DatasetFormatError`
数据集格式不正确时抛出。
#### `MetricComputeError`
指标计算失败时抛出。
### 数据结构
#### `EvaluationConfig`
评测配置数据类。
```python
@dataclass
class EvaluationConfig:
# 模型配置
model_path: str = ""
tokenizer_path: Optional[str] = None
trust_remote_code: bool = True
# 数据集配置
dataset_path: str = ""
dataset_format: Optional[str] = None # 自动检测: alpaca / sharegpt / messages
max_samples: Optional[int] = None
max_seq_length: int = 512
# 评测指标
metrics: List[str] = field(default_factory=lambda: ["bleu", "rouge"])
# 任务类型
task_type: str = "text_generation"
# 生成参数
max_new_tokens: int = 256
temperature: float = 0.7
top_p: float = 0.9
batch_size: int = 1
# 输出配置
output_format: str = "json"
output_path: Optional[str] = None
# 8GB VRAM 优化
load_in_4bit: bool = False
device_map: str = "auto"
# 其他
seed: int = 42
verbose: bool = False
```
#### `SampleResult`
单条样本的评测结果。
```python
@dataclass
class SampleResult:
index: int
prompt: str
reference: str
prediction: str
metrics: Dict[str, float] = field(default_factory=dict)
```
#### `EvaluationResult`
评测结果数据类。
```python
@dataclass
class EvaluationResult:
model_path: str
dataset_path: str
task_type: str
metrics_summary: Dict[str, float] = field(default_factory=dict)
sample_results: List[SampleResult] = field(default_factory=list)
total_samples: int = 0
elapsed_seconds: float = 0.0
timestamp: str = ""
metadata: Dict[str, Any] = field(default_factory=dict)
```
### 核心类
#### `MetricLoader`
评测指标加载器,封装 HuggingFace evaluate 库和内置指标实现。
```python
class MetricLoader:
SUPPORTED_METRICS = ("ppl", "bleu", "rouge", "exact_match", "f1")
def __init__(self)
def compute(self, metric_name: str, predictions: List[str],
references: List[str], model=None, tokenizer=None) -> float
```
**支持的指标**
- `ppl`: 困惑度(Perplexity),基于模型 logits 计算
- `bleu`: BLEU 分数(n-gram 精确率)
- `rouge`: ROUGE-L F1 分数
- `exact_match`: 精确匹配率
- `f1`: Token 级 F1 分数
**示例**
```python
from hos_optimizer.evaluate import MetricLoader
loader = MetricLoader()
score = loader.compute(
metric_name="bleu",
predictions=["模型输出文本"],
references=["参考文本"]
)
print(f"BLEU: {score}")
```
#### `DatasetLoader`
数据集加载器,支持 JSON/JSONL 格式,自动检测数据格式。
```python
class DatasetLoader:
SUPPORTED_FORMATS = ("alpaca", "sharegpt", "messages")
def __init__(self)
def load(self, dataset_path: str, dataset_format: Optional[str] = None,
max_samples: Optional[int] = None) -> List[Tuple[str, str]]
```
**参数**
- `dataset_path` (str): 数据集文件路径(JSON 或 JSONL)
- `dataset_format` (Optional[str]): 数据格式名称,None 表示自动检测
- `max_samples` (Optional[int]): 最大样本数,None 表示全部
**返回**`List[Tuple[str, str]]` - (prompt, reference) 元组列表
**示例**
```python
from hos_optimizer.evaluate import DatasetLoader
loader = DatasetLoader()
samples = loader.load(
dataset_path="./test.json",
dataset_format=None, # 自动检测
max_samples=100
)
for prompt, reference in samples:
print(f"Prompt: {prompt[:50]}...")
print(f"Reference: {reference[:50]}...")
```
#### `EvaluationEngine`
评测执行引擎,负责加载模型并执行推理生成。
```python
class EvaluationEngine:
def __init__(self, config: EvaluationConfig)
def load_model(self) -> None
def generate_predictions(self, samples: List[Tuple[str, str]]) -> List[str]
def compute_metrics(self, predictions: List[str],
references: List[str]) -> Dict[str, float]
def shutdown(self) -> None
```
**示例**
```python
from hos_optimizer.evaluate import EvaluationConfig, EvaluationEngine
config = EvaluationConfig(
model_path="./model",
dataset_path="./test.json",
metrics=["bleu", "rouge", "f1"],
max_new_tokens=256,
batch_size=1
)
engine = EvaluationEngine(config)
engine.load_model()
# 加载数据
from hos_optimizer.evaluate import DatasetLoader
loader = DatasetLoader()
samples = loader.load("./test.json")
# 生成预测
predictions = engine.generate_predictions(samples)
references = [ref for _, ref in samples]
# 计算指标
metrics = engine.compute_metrics(predictions, references)
print(metrics)
engine.shutdown()
```
#### `ResultExporter`
评测结果导出器,支持 JSON 和 Markdown 格式。
```python
class ResultExporter:
@staticmethod
def export_json(result: EvaluationResult, output_path: str) -> None
@staticmethod
def export_markdown(result: EvaluationResult, output_path: str) -> None
@staticmethod
def export_comparison_json(results: List[EvaluationResult], output_path: str) -> None
@staticmethod
def export_comparison_markdown(results: List[EvaluationResult], output_path: str) -> None
```
### 核心函数
#### `evaluate_model()`
统一评测接口,执行完整评测流程。
```python
def evaluate_model(config: EvaluationConfig) -> EvaluationResult
```
**参数**:
- `config` (EvaluationConfig): 评测配置
**返回**:`EvaluationResult` - 评测结果
**示例**:
```python
from hos_optimizer.evaluate import EvaluationConfig, evaluate_model
config = EvaluationConfig(
model_path="./model",
dataset_path="./test.json",
metrics=["bleu", "rouge"],
output_format="json",
output_path="./result.json"
)
result = evaluate_model(config)
print(f"BLEU: {result.metrics_summary.get('bleu', 0)}")
print(f"ROUGE: {result.metrics_summary.get('rouge', 0)}")
```
#### `compare_models()`
多模型对比评测。
```python
def compare_models(model_paths: List[str], config: EvaluationConfig) -> List[EvaluationResult]
```
**参数**
- `model_paths` (List[str]): 模型路径列表
- `config` (EvaluationConfig): 评测配置(model_path 会被忽略)
**返回**:`List[EvaluationResult]` - 各模型的评测结果列表
**示例**
```python
from hos_optimizer.evaluate import EvaluationConfig, compare_models
config = EvaluationConfig(
dataset_path="./test.json",
metrics=["bleu", "rouge", "f1"],
output_format="markdown",
output_path="./comparison.md"
)
results = compare_models(
model_paths=["./model_a", "./model_b", "./model_c"],
config=config
)
for r in results:
print(f"{r.model_path}: {r.metrics_summary}")
```
---
## 工具函数 API
工具函数位于 `hos_optimizer.utils`,提供日志配置、文件操作和模型路径处理等通用功能。
### 日志配置
#### `setup_logger()`
配置并返回日志记录器。
```python
def setup_logger(
name: str = "hos_optimizer",
level: int = logging.INFO,
log_file: Optional[str] = None,
fmt: str = "%(asctime)s [%(levelname)s] %(name)s: %(message)s",
datefmt: str = "%H:%M:%S"
) -> logging.Logger
```
**参数**
- `name` (str): 日志记录器名称
- `level` (int): 日志级别
- `log_file` (Optional[str]): 日志文件路径(可选)
- `fmt` (str): 日志格式
- `datefmt` (str): 日期格式
**返回**`logging.Logger` - 配置好的 Logger 实例
**示例**
```python
from hos_optimizer.utils import setup_logger
logger = setup_logger(
name="my_app",
level=logging.DEBUG,
log_file="app.log"
)
logger.info("应用启动")
```
### 文件操作工具
#### `ensure_dir()`
确保目录存在,不存在则创建。
```python
def ensure_dir(path: str) -> str
```
**参数**:
- `path` (str): 目录路径
**返回**:`str` - 目录路径
#### `get_file_size_gb()`
获取文件大小(GB)。
```python
def get_file_size_gb(path: str) -> float
```
**参数**
- `path` (str): 文件路径
**返回**:`float` - 文件大小(GB)
#### `get_dir_size_gb()`
获取目录总大小(GB)。
```python
def get_dir_size_gb(path: str) -> float
```
**参数**
- `path` (str): 目录路径
**返回**`float` - 目录总大小(GB)
#### `find_model_files()`
在目录中查找模型文件。
```python
def find_model_files(path: str) -> List[str]
```
**参数**
- `path` (str): 搜索路径
**返回**`List[str]` - 模型文件路径列表
### 模型路径处理
#### `resolve_model_path()`
解析模型路径,支持相对路径和环境变量展开。
```python
def resolve_model_path(path: str) -> str
```
**参数**
- `path` (str): 原始路径
**返回**`str` - 解析后的绝对路径
#### `is_model_path()`
判断路径是否为有效的模型路径。
```python
def is_model_path(path: str) -> bool
```
**参数**
- `path` (str): 路径字符串
**返回**`bool` - 是否为有效模型路径
#### `get_model_format()`
推断模型格式。
```python
def get_model_format(path: str) -> str
```
**参数**
- `path` (str): 模型路径
**返回**`str` - 格式字符串(`gguf`/`safetensors`/`pytorch`/`unknown`
---
## 更多资源
- [使用示例](EXAMPLES.md) - 完整使用示例
- [架构文档](docs/architecture.md) - 系统设计说明
- [安装指南](INSTALL.md) - 安装和配置