| # 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) - 安装和配置 |
|
|