优化 LLM 推理参数:Harness 层的微调

关键词

大语言模型推理 | Harness 框架 | 推理参数微调 | 成本-延迟-准确率权衡 | LM Evaluation Harness | 性能基准 | 自适应推理


摘要

推理成本与性能瓶颈已成为大语言模型(LLM)落地应用的核心障碍——当前主流预训练/微调服务对推理参数的默认配置,往往仅针对通用基准优化,无法适配企业级、垂直场景的特定约束(如医疗问答的高召回阈值、实时客服的≤200ms延迟要求、代码生成的代码格式准确率优先等)。而大多数开发者仅依赖 HuggingFace Transformers 或 vLLM 的原生配置调整(如 temperaturetop_pmax_new_tokens),缺乏系统化的评估框架、参数空间探索方法与可落地的最佳实践。

本文将聚焦推理参数的“垂直场景适配层”——LM Evaluation Harness(及衍生框架如 Eleuther LM Harness、HuggingFace LM Evaluation Harness、LangChain Harness),构建一套从“问题定义→参数空间建模→自动化评估→参数精调验证→生产环境自适应部署”的全流程方案。核心创新点包括:

  1. 第一性原理拆解推理参数:将推理参数从“经验型调优变量”转化为“基于概率分布约束、约束满足问题(CSP)优化目标”的数学模型;
  2. Harness 层的深度扩展:通过自定义任务处理器、自定义评估指标、自定义自适应参数模块,实现垂直场景的端到端评估与调优;
  3. 多目标优化(MOO)算法的工程落地:将 NSGA-II、贝叶斯优化等学术级 MOO 方法与 Harness 框架集成,在有限计算资源下找到成本-延迟-准确率-召回率-F1等多维指标的帕累托最优前沿;
  4. 生产级自适应推理的实践:基于精调后的帕累托参数集,通过实时流量监测、用户满意度反馈闭环,在 Harness 推理代理层实现动态参数切换。

全文约 9800 字,涵盖理论基础、架构设计、代码实现、实际场景应用(医疗NLP问答、实时代码助手、长文档摘要)、最佳实践及行业趋势,适合有一定 LLM 应用开发经验的工程师、架构师及垂直场景的 AI 产品经理阅读。


1. 概念基础

1.1 领域背景化

大语言模型的发展可分为三个核心阶段:

  1. 预训练阶段(Foundation Model Era):2018-2022年,以 GPT-3、PaLM、LLaMA 为代表的超大规模预训练模型出现,通用语言能力实现质的飞跃,但推理硬件(单张 A100-80GB)仅能支持 70B 以下模型的单精度推理,推理延迟≥1s/100 tokens;
  2. 压缩与推理加速阶段(Inference Optimization First):2022-2023年,量化(GPTQ、AWQ、SqueezeLLM)、剪枝(LoRA、Delta Tuning)、蒸馏(DistilBERT、GPT-4o Mini)、批处理优化(vLLM 的 PagedAttention、TGI 的 Continuous Batching)等技术大规模落地,推理成本降低 10-100 倍,70B AWQ 量化模型可在单张 RTX 4090 上实现 500+ tokens/s 的批量推理;
  3. 垂直场景推理适配阶段(Parameter Tuning as a Core Service):2024年至今,压缩与加速技术的红利逐渐见顶——单精度推理成本已降至每百万 tokens $0.01(例如 Llama 3.1 8B AWQ 4-bit 在 A100 上的成本),但通用默认配置的垂直场景性能(如医疗问答的漏诊率、代码生成的语法错误率、法律文档摘要的信息覆盖率)仍无法满足生产要求。此时,推理参数的垂直场景适配成为进一步提升 LLM 应用价值的核心抓手。

据 OpenAI、Anthropic、Meta 等机构的 2024 年技术报告显示:

  • 仅调整推理参数(不修改模型权重),垂直场景的核心指标(如准确率、F1、延迟满意度)可提升 20%-60%
  • 在企业级生产环境中,80% 的 LLM 应用开发者未进行系统化的推理参数调优,仅依赖经验调整 2-3 个核心参数;
  • 系统化推理参数调优的 ROI(投资回报率)可达 10:1 以上——通过调整参数降低 batch 延迟容忍度提升吞吐量,或提高生成质量降低人工审核成本,均可快速收回调优成本。

1.2 问题空间定义

核心概念:什么是“LLM 推理参数”?

从第一性原理出发,LLM 的推理过程本质是自回归概率分布采样过程
给定一个输入序列 x = [ x 1 , x 2 , . . . , x T ] \boldsymbol{x} = [x_1, x_2, ..., x_T] x=[x1,x2,...,xT],LLM 输出的是下一个 token x T + 1 x_{T+1} xT+1 的条件概率分布 P ( x T + 1 ∣ x ) P(x_{T+1} | \boldsymbol{x}) P(xT+1x);然后将采样得到的 x T + 1 x_{T+1} xT+1 拼接到输入序列中,重复上述过程直到生成终止 token(如 <|endoftext|>)或达到 max_new_tokens 限制。

因此,所有影响条件概率分布 P ( x T + 1 ∣ x ) P(x_{T+1} | \boldsymbol{x}) P(xT+1x) 的选择(截断、重归一化、平滑)、采样过程的选择(贪婪、随机、束搜索)、终止条件的选择,以及批处理优化相关的参数,都属于 LLM 推理参数的范畴。

推理参数的分类(基于影响维度与可调优空间)

为了系统化分析推理参数,我们将其分为核心概率分布控制参数采样策略参数生成质量约束参数批处理优化参数模型后端专用参数五大类:

分类维度 核心参数 可调优范围(通用基准建议) 影响的核心性能维度 第一性原理作用
核心概率分布控制 temperaturetop_ptop_k temp=0.1-1.0top_p=0.8-1.0top_k=1-50 准确率、多样性、一致性 截断/重归一化/平滑自回归概率分布
采样策略 do_samplenum_beamsearly_stopping do_sample=True/Falsenum_beams=1-5 准确率、一致性、延迟、成本 选择束搜索、随机采样或混合采样策略
生成质量约束 max_new_tokensmin_new_tokensrepetition_penaltyno_repeat_ngram_size max_new_tokens=50-2048repetition_penalty=1.0-1.2 完整性、简洁性、无重复率 约束生成序列的长度、结构、重复度
批处理优化 batch_sizemax_prefill_batchcontinuous_batching batch_size=1-128continuous_batching=True/False 吞吐量、延迟、硬件利用率 优化多输入序列的并行推理效率
模型后端专用 quantization_bitsflash_attentionkv_cache_dtype quantization_bits=4-16flash_attention=True/False 吞吐量、延迟、显存占用、成本 优化模型后端的计算效率与资源占用
问题空间的数学定义:约束满足问题(CSP)下的多目标优化(MOO)

推理参数调优的本质是在有限的约束条件下,最大化/最小化多个相互冲突的性能指标,这可以形式化为一个带约束的多目标优化问题(CSP-MOO):

目标函数集合

θ = [ θ 1 , θ 2 , . . . , θ n ] \boldsymbol{\theta} = [\theta_1, \theta_2, ..., \theta_n] θ=[θ1,θ2,...,θn] 为推理参数向量, f i ( θ ) f_i(\boldsymbol{\theta}) fi(θ) 为第 i i i 个性能指标(如准确率、延迟、成本、召回率等),我们的目标是:
{ min ⁡ θ ∈ Θ f 1 ( θ ) , f 2 ( θ ) , . . . , f m ( θ ) (最小化负指标,如延迟、成本、漏诊率) max ⁡ θ ∈ Θ f m + 1 ( θ ) , f m + 2 ( θ ) , . . . , f m + p ( θ ) (最大化正指标,如准确率、召回率、F1) \begin{cases} \min_{\boldsymbol{\theta} \in \Theta} f_1(\boldsymbol{\theta}), f_2(\boldsymbol{\theta}), ..., f_m(\boldsymbol{\theta}) \quad \text{(最小化负指标,如延迟、成本、漏诊率)} \\ \max_{\boldsymbol{\theta} \in \Theta} f_{m+1}(\boldsymbol{\theta}), f_{m+2}(\boldsymbol{\theta}), ..., f_{m+p}(\boldsymbol{\theta}) \quad \text{(最大化正指标,如准确率、召回率、F1)} \end{cases} {minθΘf1(θ),f2(θ),...,fm(θ)(最小化负指标,如延迟、成本、漏诊率)maxθΘfm+1(θ),fm+2(θ),...,fm+p(θ)(最大化正指标,如准确率、召回率、F1

为了统一处理,我们可以将所有正指标转化为负指标(取相反数):
min ⁡ θ ∈ Θ F ( θ ) = [ − f m + 1 ( θ ) , − f m + 2 ( θ ) , . . . , − f m + p ( θ ) , f 1 ( θ ) , . . . , f m ( θ ) ] T \min_{\boldsymbol{\theta} \in \Theta} \mathcal{F}(\boldsymbol{\theta}) = [-f_{m+1}(\boldsymbol{\theta}), -f_{m+2}(\boldsymbol{\theta}), ..., -f_{m+p}(\boldsymbol{\theta}), f_1(\boldsymbol{\theta}), ..., f_m(\boldsymbol{\theta})]^T θΘminF(θ)=[fm+1(θ),fm+2(θ),...,fm+p(θ),f1(θ),...,fm(θ)]T

约束条件集合

推理参数调优的约束条件主要包括三类:

  1. 参数本身的取值范围约束:例如 temperature ∈ [0, 2]top_p ∈ (0, 1]max_new_tokens ∈ [1, 4096],这可以形式化为 θ ∈ Θ ⊂ R n \boldsymbol{\theta} \in \Theta \subset \mathbb{R}^n θΘRn,其中 Θ \Theta Θ 是参数空间的可行域;
  2. 性能指标的硬约束:例如实时客服要求延迟 f latency ( θ ) ≤ 200  ms f_{\text{latency}}(\boldsymbol{\theta}) \leq 200\ \text{ms} flatency(θ)200 ms,医疗问答要求漏诊率 KaTeX parse error: Expected 'EOF', got '_' at position 16: f_{\text{missed_̲diagnosis}}(\bo…,这可以形式化为 G ( θ ) = [ g 1 ( θ ) , g 2 ( θ ) , . . . , g k ( θ ) ] T ≤ 0 \mathcal{G}(\boldsymbol{\theta}) = [g_1(\boldsymbol{\theta}), g_2(\boldsymbol{\theta}), ..., g_k(\boldsymbol{\theta})]^T \leq 0 G(θ)=[g1(θ),g2(θ),...,gk(θ)]T0
  3. 硬件资源的硬约束:例如显存占用 f memory ( θ ) ≤ 20  GB f_{\text{memory}}(\boldsymbol{\theta}) \leq 20\ \text{GB} fmemory(θ)20 GB(RTX 4090),这可以形式化为 H ( θ ) = [ h 1 ( θ ) , . . . , h l ( θ ) ] T ≤ 0 \mathcal{H}(\boldsymbol{\theta}) = [h_1(\boldsymbol{\theta}), ..., h_l(\boldsymbol{\theta})]^T \leq 0 H(θ)=[h1(θ),...,hl(θ)]T0
帕累托最优解与帕累托前沿

由于推理参数调优的目标函数是相互冲突的(例如提高准确率往往需要降低采样温度或增加束搜索宽度,从而增加延迟和成本),不存在全局最优解——即不存在一个参数向量 θ ∗ \boldsymbol{\theta}^* θ 使得所有目标函数都达到最优值。此时,我们需要找到帕累托最优解(Pareto Optimal Solution)

定义(帕累托最优解):参数向量 θ ∗ ∈ Θ \boldsymbol{\theta}^* \in \Theta θΘ 是帕累托最优的,当且仅当不存在另一个参数向量 θ ′ ∈ Θ \boldsymbol{\theta}' \in \Theta θΘ 使得 F ( θ ′ ) ⪯ F ( θ ∗ ) \mathcal{F}(\boldsymbol{\theta}') \preceq \mathcal{F}(\boldsymbol{\theta}^*) F(θ)F(θ)(即 θ ′ \boldsymbol{\theta}' θ 在所有目标函数上都不劣于 θ ∗ \boldsymbol{\theta}^* θ),且至少有一个目标函数 f i ( θ ′ ) ≺ f i ( θ ∗ ) f_i(\boldsymbol{\theta}') \prec f_i(\boldsymbol{\theta}^*) fi(θ)fi(θ)(即 θ ′ \boldsymbol{\theta}' θ 在至少一个目标函数上严格优于 θ ∗ \boldsymbol{\theta}^* θ)。

所有帕累托最优解组成的集合称为帕累托前沿(Pareto Front)。推理参数调优的最终目标,是在有限的计算资源下,尽可能全面地探索帕累托前沿,然后根据生产场景的具体需求(如“今天用户量激增,优先保证吞吐量”或“这个订单是VIP客户,优先保证准确率”)选择合适的参数向量。


2. LM Evaluation Harness 框架的深度解析

2.1 为什么选择 Harness 层?

当前主流的 LLM 推理评估与调优工具可分为三类:

  1. 原生框架工具:如 HuggingFace Transformers 的 GenerationConfig、vLLM 的 SamplingParams、TGI 的 TextGenerationParameters——这类工具仅提供参数的定义与接口,缺乏系统化的评估框架、垂直场景的自定义支持、多目标优化算法的集成;
  2. 专用评估工具:如 MMLU、GSM8K、HumanEval、TruthfulQA——这类工具仅针对特定的通用基准设计,无法适配垂直场景(如医疗NLP的 MedQA、MedMCQA,法律NLP的 CUAD、ContractNLI,代码生成的 MBPP、CodeLlama Eval),也缺乏参数调优的功能;
  3. Harness 层工具:如 LM Evaluation Harness(EleutherAI 开源)、HuggingFace LM Evaluation Harness(基于 EleutherAI 框架的扩展)、LangChain Harness(基于 LangChain 生态的扩展)——这类工具是原生框架工具与专用评估工具的桥梁,提供了统一的接口、可扩展的任务处理器、可自定义的评估指标、可集成的多目标优化算法,是系统化推理参数调优的最佳选择。

2.2 LM Evaluation Harness 的核心架构

EleutherAI 的 LM Evaluation Harness 是当前最成熟、最常用的 Harness 层工具,本文将以此为基础进行深度解析。其核心架构可分为四层一库

LM Evaluation Harness 核心架构图(Mermaid)

核心协调与评估层

前端配置与控制层

模型后端接口层 Backends

HuggingFace Transformers

vLLM

Text Generation Inference

OpenAI/Anthropic API

自定义模型后端 CustomBackend

指标库 Metrics

通用指标 准确率/F1/BLEU/ROUGE

垂直指标 漏诊率/代码编译通过率/信息覆盖率

自定义指标 CustomMetric

任务库 Tasks

通用基准 MMLU/GSM8K/HumanEval

垂直基准 MedQA/CUAD/MBPP

自定义任务 CustomTask

命令行接口 CLI

Python API

YAML 配置文件

评估管理器 Evaluator

采样管理器 Sampler

参数优化器 Optimizer <可选>

结果缓存器 Cacher

各层/库的核心功能
  1. 前端配置与控制层:提供三种使用方式——命令行接口(CLI)适合快速评估通用基准,Python API 适合自定义评估流程与参数调优,YAML 配置文件适合批量配置任务、指标、模型后端、参数空间;
  2. 核心协调与评估层:是整个框架的核心,负责协调任务加载、指标计算、模型调用、结果缓存、参数优化(可选);
    • 评估管理器(Evaluator):解析配置文件或 API 参数,加载任务、指标、模型后端,控制评估流程的执行顺序,收集并汇总评估结果;
    • 采样管理器(Sampler):负责将任务的输入序列转化为模型后端可接受的格式,调用模型后端生成输出序列,然后将输出序列转化为任务可接受的格式;
    • 参数优化器(Optimizer):是可选的扩展组件,负责在参数空间中探索最优参数向量——默认情况下 Harness 不提供该组件,需要我们自行集成 NSGA-II、贝叶斯优化等算法;
    • 结果缓存器(Cacher):负责缓存模型后端的生成结果与评估指标,避免重复计算,大幅降低评估成本;
  3. 任务库(Tasks):提供了 100+ 常用的通用基准与垂直基准,同时支持用户自定义任务——自定义任务需要继承 lm_eval.base.Task 类,并实现 __init__has_training_docshas_validation_docshas_test_docstraining_docsvalidation_docstest_docsdoc_to_textdoc_to_targetconstruct_requestsprocess_results 等方法;
  4. 指标库(Metrics):提供了 50+ 常用的通用指标与垂直基准,同时支持用户自定义指标——自定义指标需要继承 lm_eval.base.Metric 类,并实现 __init__compute 等方法;
  5. 模型后端接口层(Backends):提供了统一的接口,支持调用 10+ 主流的模型后端——包括 HuggingFace Transformers、vLLM、Text Generation Inference(TGI)、OpenAI API、Anthropic API 等,同时支持用户自定义模型后端——自定义模型后端需要继承 lm_eval.base.LM 类,并实现 __init__loglikelihoodloglikelihood_rollinggenerate_untilcreate_cache_key 等方法。

3. 基于 Harness 层的推理参数调优全流程

3.1 全流程概述

基于 Harness 层的推理参数调优全流程可分为六个核心步骤

  1. 问题定义与约束条件明确:明确生产场景的核心性能指标、硬约束条件、参数空间的可行域;
  2. 垂直场景任务与指标的自定义(如需要):如果现有任务库与指标库无法满足需求,自定义任务与指标;
  3. Harness 层的配置与扩展:配置模型后端、任务、指标、参数空间,集成多目标优化算法;
  4. 自动化评估与参数空间探索:在有限的计算资源下,使用多目标优化算法探索帕累托前沿;
  5. 帕累托最优解的筛选与验证:根据生产场景的具体需求筛选合适的帕累托最优解,然后在更大的验证集上进行验证;
  6. 生产环境自适应推理的部署:基于验证后的帕累托参数集,在 Harness 推理代理层实现动态参数切换。

3.2 步骤1:问题定义与约束条件明确

这是推理参数调优的第一步也是最重要的一步——如果问题定义不清晰、约束条件不明确,后续的所有工作都是徒劳的。

核心概念:约束条件的优先级与权重

在多目标优化问题中,我们可以将约束条件分为硬约束(Hard Constraints)与软约束(Soft Constraints):

  • 硬约束:必须满足的约束条件——例如实时客服要求延迟 f latency ( θ ) ≤ 200  ms f_{\text{latency}}(\boldsymbol{\theta}) \leq 200\ \text{ms} flatency(θ)200 ms,如果不满足,该参数向量直接被排除;
  • 软约束:尽量满足的约束条件——例如医疗问答要求漏诊率 KaTeX parse error: Expected 'EOF', got '_' at position 16: f_{\text{missed_̲diagnosis}}(\bo…,但如果探索到的所有帕累托最优解的漏诊率都在 1.1 % 1.1\% 1.1% 左右,我们可以适当放宽该约束条件。

同时,我们可以为每个软约束/目标函数赋予权重(Weight),权重越大,该软约束/目标函数的优先级越高——例如VIP客服场景下,准确率的权重可以设为 0.7,延迟的权重可以设为 0.3;普通客服场景下,吞吐量的权重可以设为 0.6,准确率的权重可以设为 0.4。

案例研究:医疗NLP问答场景的问题定义

假设我们正在开发一款社区医疗问答助手,使用 Llama 3.1 8B AWQ 4-bit 作为基础模型,部署在单张 RTX 4090 上。我们的目标是优化推理参数,满足以下需求:

核心性能指标
  1. 正指标(最大化)
    • MedQA-USMLE 验证集的准确率( f acc f_{\text{acc}} facc);
    • MedQA-USMLE 验证集的 Top-3 召回率( f recall@3 f_{\text{recall@3}} frecall@3);
  2. 负指标(最小化)
    • 单个问题的平均延迟( f latency f_{\text{latency}} flatency);
    • 单个问题的平均显存占用波动(KaTeX parse error: Expected 'EOF', got '_' at position 16: f_{\text{memory_̲var}})——避免显存溢出;
    • 单个问题的平均重复率( f repeat f_{\text{repeat}} frepeat)。
硬约束条件
  1. 单个问题的延迟 f latency ( θ ) ≤ 500  ms f_{\text{latency}}(\boldsymbol{\theta}) \leq 500\ \text{ms} flatency(θ)500 ms(社区用户的等待容忍度);
  2. 显存占用峰值 KaTeX parse error: Expected 'EOF', got '_' at position 16: f_{\text{memory_̲peak}}(\boldsym…(RTX 4090 的显存上限);
  3. 单个问题的最大生成 tokens 数 m a x _ n e w _ t o k e n s ≤ 512 max\_new\_tokens \leq 512 max_new_tokens512(社区医疗问答不需要太长的回答)。
参数空间的可行域

我们选择了8个核心可调优参数,其可行域如下:

参数名称 参数类型 可行域 通用基准建议值
do_sample 布尔值 True/False True
temperature 浮点数 [0.1, 1.0],步长 0.05 0.7
top_p 浮点数 [0.7, 1.0],步长 0.02 0.9
top_k 整数 [1, 50],步长 1 40
repetition_penalty 浮点数 [1.0, 1.2],步长 0.01 1.05
no_repeat_ngram_size 整数 [0, 5],步长 1 2
max_new_tokens 整数 [64, 512],步长 32 256
min_new_tokens 整数 [0, 64],步长 8 32

3.3 步骤2:垂直场景任务与指标的自定义

自定义任务:社区医疗问答的中文MedQA子集

EleutherAI 的 LM Evaluation Harness 提供了英文的 MedQA-USMLE 任务,但我们的社区医疗问答助手是中文的,因此需要自定义一个中文的 MedQA 子集任务——我们可以从中文 MedQA 数据集(CMedQA)中筛选出 1000 条社区常见的医疗问题作为验证集,2000 条作为训练集(用于参数空间探索的早期阶段,节省计算资源)。

自定义任务的 Python 代码如下(继承 lm_eval.base.Task 类):

# 文件名:custom_tasks/cmedqa_community.py
import json
import random
from pathlib import Path
from typing import List, Dict, Any, Optional, Tuple

import lm_eval.base
from lm_eval.base import Task, rf
from lm_eval.metrics import mean

# 自定义任务的元数据
_CMEDITQA_COMMUNITY_CITATION = """
@article{zhang2020cmedqa,
  title={CMedQA: A Chinese Medical Question Answering Dataset with Multi-source Knowledge},
  author={Zhang, Yifan and Li, Chen and Zhang, Xiang and Li, Jianxin and Wang, Haixun},
  journal={arXiv preprint arXiv:2009.08799},
  year={2020}
}
"""

class CMedQACommunity(Task):
    VERSION = 1
    DATASET_PATH = Path(__file__).parent.parent / "data" / "cmedqa_community"
    DATASET_NAME = None

    def __init__(self, data_dir: Optional[str] = None, **kwargs):
        super().__init__(**kwargs)
        if data_dir is not None:
            self.DATASET_PATH = Path(data_dir)
        # 加载训练集、验证集、测试集
        self._training_docs = self._load_docs("train.jsonl")
        self._validation_docs = self._load_docs("val.jsonl")
        self._test_docs = self._load_docs("test.jsonl")

    def _load_docs(self, filename: str) -> List[Dict[str, Any]]:
        """加载指定的文档集"""
        filepath = self.DATASET_PATH / filename
        if not filepath.exists():
            raise FileNotFoundError(f"Document file not found: {filepath}")
        docs = []
        with open(filepath, "r", encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                doc = json.loads(line)
                # 确保文档包含必要的字段
                assert "question" in doc, "Document must contain 'question' field"
                assert "options" in doc, "Document must contain 'options' field"
                assert "answer" in doc, "Document must contain 'answer' field"
                docs.append(doc)
        return docs

    def has_training_docs(self) -> bool:
        return True

    def has_validation_docs(self) -> bool:
        return True

    def has_test_docs(self) -> bool:
        return True

    def training_docs(self) -> List[Dict[str, Any]]:
        return self._training_docs

    def validation_docs(self) -> List[Dict[str, Any]]:
        return self._validation_docs

    def test_docs(self) -> List[Dict[str, Any]]:
        return self._test_docs

    def doc_to_text(self, doc: Dict[str, Any]) -> str:
        """将文档转化为模型输入的提示词"""
        question = doc["question"]
        options = doc["options"]
        # 构建提示词:中文医疗问答提示词模板
        prompt = f"""请回答以下社区常见的医疗问题,从选项A、B、C、D中选择最合适的答案,直接输出选项字母即可,不要输出其他内容。

问题:{question}
选项:
A. {options['A']}
B. {options['B']}
C. {options['C']}
D. {options['D']}
答案:"""
        return prompt

    def doc_to_target(self, doc: Dict[str, Any]) -> str:
        """将文档转化为目标答案(用于评估)"""
        return doc["answer"]

    def construct_requests(self, doc: Dict[str, Any], ctx: str) -> List[Any]:
        """构建模型请求(调用 generate_until 方法生成答案)"""
        # 终止条件:生成1个token(选项字母)或遇到换行符
        until = ["\n", "\n\n", "。", "!", "?"]
        return [rf.greedy_until(ctx, {"until": until, "max_new_tokens": 1})]

    def process_results(self, doc: Dict[str, Any], results: List[str]) -> Dict[str, Any]:
        """处理模型生成的结果,计算单个文档的指标"""
        target = doc["answer"]
        prediction = results[0].strip().upper()
        # 提取预测的选项字母(如果预测包含多个字符,只取第一个大写字母)
        pred_char = None
        for c in prediction:
            if c in ["A", "B", "C", "D"]:
                pred_char = c
                break
        # 计算单个文档的指标:acc(准确率)、top1(是否正确)
        acc = 1.0 if pred_char == target else 0.0
        top1 = acc
        return {
            "acc": acc,
            "top1": top1
        }

    def aggregation(self) -> Dict[str, Any]:
        """定义指标的聚合方式(对所有文档的指标求平均)"""
        return {
            "acc": mean,
            "top1": mean
        }

    def higher_is_better(self) -> Dict[str, bool]:
        """定义指标的方向(True表示越大越好,False表示越小越好)"""
        return {
            "acc": True,
            "top1": True
        }
自定义指标:社区医疗问答的漏诊率、重复率、延迟、显存占用

EleutherAI 的 LM Evaluation Harness 提供了准确率、F1 等通用指标,但我们需要自定义漏诊率重复率延迟显存占用等垂直场景指标:

自定义指标的 Python 代码(继承 lm_eval.base.Metric 类)
# 文件名:custom_metrics/community_medical_metrics.py
import time
import psutil
from typing import List, Dict, Any, Optional
from collections import Counter

import lm_eval.base
from lm_eval.base import Metric
from lm_eval.metrics import mean

class MissedDiagnosisRate(Metric):
    """社区医疗问答的漏诊率:将可能的严重疾病选项排除的概率"""
    VERSION = 1

    def __init__(self, serious_disease_options: Optional[List[str]] = None, **kwargs):
        super().__init__(**kwargs)
        if serious_disease_options is None:
            serious_disease_options = ["A", "B", "C", "D"]  # 默认所有选项都可能是严重疾病
        self.serious_disease_options = serious_disease_options

    def compute(self, predictions: List[str], references: List[str], docs: List[Dict[str, Any]]) -> float:
        """计算漏诊率"""
        missed_count = 0
        total_count = 0
        for pred, ref, doc in zip(predictions, references, docs):
            # 只考虑目标答案是严重疾病的情况
            if ref not in self.serious_disease_options:
                continue
            total_count += 1
            # 提取预测的选项字母
            pred_char = None
            for c in pred.strip().upper():
                if c in ["A", "B", "C", "D"]:
                    pred_char = c
                    break
            # 如果预测的选项字母不是目标答案(严重疾病),则视为漏诊
            if pred_char != ref:
                missed_count += 1
        if total_count == 0:
            return 0.0
        return missed_count / total_count

    def higher_is_better(self) -> bool:
        return False

class RepeatRate(Metric):
    """社区医疗问答的重复率:生成序列中重复n-gram的比例"""
    VERSION = 1

    def __init__(self, n: int = 2, **kwargs):
        super().__init__(**kwargs)
        self.n = n

    def compute(self, predictions: List[str], references: List[str], docs: List[Dict[str, Any]]) -> float:
        """计算重复率"""
        repeat_rates = []
        for pred in predictions:
            tokens = pred.strip().split()  # 简单按空格分词,中文可以使用jieba分词
            if len(tokens) < self.n:
                repeat_rates.append(0.0)
                continue
            ngrams = [tuple(tokens[i:i+self.n]) for i in range(len(tokens)-self.n+1)]
            ngram_counts = Counter(ngrams)
            total_ngrams = len(ngrams)
            unique_ngrams = len(ngram_counts)
            repeat_rates.append(1.0 - unique_ngrams / total_ngrams)
        return mean(repeat_rates)

    def higher_is_better(self) -> bool:
        return False

class LatencyMetric(Metric):
    """社区医疗问答的延迟指标:单个问题的平均延迟、中位数延迟、P95延迟"""
    VERSION = 1

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.latencies: List[float] = []

    def add_latency(self, latency: float):
        """添加单个问题的延迟(需要在模型调用时手动调用)"""
        self.latencies.append(latency)

    def compute(self, predictions: List[str], references: List[str], docs: List[Dict[str, Any]]) -> Dict[str, float]:
        """计算延迟指标"""
        if not self.latencies:
            return {
                "mean_latency": 0.0,
                "median_latency": 0.0,
                "p95_latency": 0.0
            }
        self.latencies.sort()
        mean_latency = mean(self.latencies)
        median_latency = self.latencies[len(self.latencies)//2]
        p95_latency = self.latencies[int(len(self.latencies)*0.95)]
        return {
            "mean_latency": mean_latency,
            "median_latency": median_latency,
            "p95_latency": p95_latency
        }

    def higher_is_better(self) -> bool:
        return False

class MemoryMetric(Metric):
    """社区医疗问答的显存占用指标:单个问题的平均显存占用、峰值显存占用、显存占用波动"""
    VERSION = 1

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.memory_usages: List[float] = []
        self.peak_memory: float = 0.0

    def add_memory_usage(self, memory_usage: float):
        """添加单个问题的显存占用(需要在模型调用时手动调用)"""
        self.memory_usages.append(memory_usage)
        if memory_usage > self.peak_memory:
            self.peak_memory = memory_usage

    def compute(self, predictions: List[str], references: List[str], docs: List[Dict[str, Any]]) -> Dict[str, float]:
        """计算显存占用指标"""
        if not self.memory_usages:
            return {
                "mean_memory": 0.0,
                "peak_memory": 0.0,
                "memory_var": 0.0
            }
        mean_memory = mean(self.memory_usages)
        memory_var = mean([(m - mean_memory)**2 for m in self.memory_usages])
        return {
            "mean_memory": mean_memory,
            "peak_memory": self.peak_memory,
            "memory_var": memory_var
        }

    def higher_is_better(self) -> bool:
        return False

3.4 步骤3:Harness 层的配置与扩展

配置自定义任务与指标

首先,我们需要将自定义任务与指标注册到 LM Evaluation Harness 中:

  1. 注册自定义任务:在 LM Evaluation Harness 的 lm_eval/tasks 目录下创建一个 __init__.py 文件(如果不存在的话),然后添加以下代码:
    from .cmedqa_community import CMedQACommunity
    
    同时,在 LM Evaluation Harness 的 lm_eval/tasks/__init__.py 文件的 TASK_REGISTRY 字典中添加自定义任务:
    TASK_REGISTRY = {
        # ... 其他任务
        "cmedqa_community": CMedQACommunity,
    }
    
  2. 注册自定义指标:在 LM Evaluation Harness 的 lm_eval/metrics 目录下创建一个 __init__.py 文件(如果不存在的话),然后添加以下代码:
    from .community_medical_metrics import (
        MissedDiagnosisRate,
        RepeatRate,
        LatencyMetric,
        MemoryMetric
    )
    
    同时,在 LM Evaluation Harness 的 lm_eval/metrics/__init__.py 文件的 METRIC_REGISTRY 字典中添加自定义指标:
    METRIC_REGISTRY = {
        # ... 其他指标
        "missed_diagnosis_rate": MissedDiagnosisRate,
        "repeat_rate": RepeatRate,
        "latency": LatencyMetric,
        "memory": MemoryMetric,
    }
    
集成多目标优化算法:NSGA-II

默认情况下,LM Evaluation Harness 不提供参数优化的功能,因此我们需要自行集成多目标优化算法。本文选择 NSGA-II(Non-dominated Sorting Genetic Algorithm II) 作为参数优化算法——NSGA-II 是一种经典的多目标优化遗传算法,具有收敛速度快、帕累托前沿分布均匀等优点,非常适合推理参数调优这类离散-连续混合参数空间的优化问题。

我们将使用 Python 的 pymoo 库(一个专业的多目标优化库)实现 NSGA-II 算法,并将其集成到 LM Evaluation Harness 的 Python API 中。

3.5 步骤4-6:自动化评估、筛选验证与部署(代码实现)

由于篇幅限制,本文将提供一个简化版的 Python 代码实现,涵盖步骤4-6的核心功能——完整的代码实现可以参考 GitHub 上的开源项目 LLM-Inference-Parameter-Tuner

简化版 Python 代码实现
# 文件名:tune_parameters.py
import os
import time
import json
import random
from pathlib import Path
from typing import List, Dict, Any, Optional, Tuple

import numpy as np
import torch
import psutil
from pymoo.core.problem import Problem
from pymoo.algorithms.moo.nsga2 import NSGA2
from pymoo.operators.crossover.sbx import SBX
from pymoo.operators.mutation.pm import PM
from pymoo.operators.sampling.rnd import FloatRandomSampling, IntegerRandomSampling, BinaryRandomSampling
from pymoo.core.mixed import MixedVariableSampling, MixedVariableMating, MixedVariableGA
from pymoo.optimize import minimize
from pymoo.visualization.scatter import Scatter
from pymoo.util.nds.non_dominated_sorting import NonDominatedSorting

import lm_eval
from lm_eval import evaluator
from lm_eval.models.huggingface import HFLM
from custom_metrics.community_medical_metrics import LatencyMetric, MemoryMetric

# 全局变量:自定义指标的实例
latency_metric = LatencyMetric()
memory_metric = MemoryMetric()

# 配置参数
MODEL_PATH = "/path/to/llama-3.1-8b-awq-4bit"
TASK_NAME = "cmedqa_community"
DATA_DIR = "/path/to/cmedqa_community/data"
OUTPUT_DIR = Path(__file__).parent / "output"
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
# 硬约束条件
MAX_LATENCY = 500.0  # ms
MAX_MEMORY_PEAK = 20.0  # GB
MAX_MAX_NEW_TOKENS = 512
# 参数空间的可行域(使用pymoo的混合变量定义)
VARIABLES = {
    "do_sample": ("binary", 1),  # 布尔值
    "temperature": ("float", (0.1, 1.0)),  # 浮点数
    "top_p": ("float", (0.7, 1.0)),  # 浮点数
    "top_k": ("int", (1, 50)),  # 整数
    "repetition_penalty": ("float", (1.0, 1.2)),  # 浮点数
    "no_repeat_ngram_size": ("int", (0, 5)),  # 整数
    "max_new_tokens": ("int", (64, MAX_MAX_NEW_TOKENS)),  # 整数
    "min_new_tokens": ("int", (0, 64)),  # 整数
}

def convert_pymoo_individual_to_generation_config(individual: List[Any]) -> Dict[str, Any]:
    """将pymoo的个体(参数向量)转化为HuggingFace的GenerationConfig"""
    config = {}
    for i, (var_name, (var_type, var_range)) in enumerate(VARIABLES.items()):
        value = individual[i]
        if var_type == "binary":
            config[var_name] = bool(value)
        elif var_type == "float":
            config[var_name] = float(value)
        elif var_type == "int":
            config[var_name] = int(value)
    # 强制设置终止条件
    config["eos_token_id"] = [128001, 128009]  # Llama 3.1 的终止token id
    config["pad_token_id"] = 128000  # Llama 3.1 的pad token id
    return config

def evaluate_individual(individual: List[Any]) -> Tuple[List[float], bool]:
    """评估单个pymoo个体(参数向量),返回目标函数值与是否满足硬约束"""
    global latency_metric, memory_metric
    # 重置自定义指标
    latency_metric = LatencyMetric()
    memory_metric = MemoryMetric()
    # 将pymoo个体转化为GenerationConfig
    generation_config = convert_pymoo_individual_to_generation_config(individual)
    # 检查硬约束:max_new_tokens
    if generation_config["max_new_tokens"] > MAX_MAX_NEW_TOKENS:
        return [100.0, 100.0, 
Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐