LLM推理服务核心：请求与采样参数数据结构设计与实现

1. 项目概述与核心价值最近在折腾大语言模型推理服务发现很多框架在实现请求调度和采样参数管理时代码结构要么过于复杂要么耦合太紧调试起来特别费劲。这让我萌生了一个想法能不能从零开始构建一个清晰、高效且易于理解的推理服务核心这就是“从0实现SGLang”系列文章的由来。在上一篇我们搭建了基础骨架后本篇我们将深入最核心的基石请求Req与采样参数SamplingParams这两个数据结构的设计与实现。为什么这两个结构如此关键你可以把它们想象成餐厅的后厨订单系统。一个用户点餐发起一个推理请求这个“订单”里不仅包含了要做什么菜输入的文本提示词还详细指定了菜要怎么做——比如辣度、咸淡、要不要香菜这就是采样参数控制模型生成文本的风格、长度、随机性等。后厨我们的推理引擎必须能快速、准确地解析每一张“订单”并高效地分配给合适的厨师计算资源进行处理。如果订单格式混乱、信息不全或者后厨理解有歧义整个服务就会乱套轻则上错菜重则厨房瘫痪。因此一个设计良好的Req和SamplingParams直接决定了推理服务的吞吐量、延迟稳定性以及功能的可扩展性。我们将从实际需求出发一步步推导出它们应有的字段并解释每个字段背后的设计考量。本文不仅会给出可直接复用的代码结构更会分享我在实现过程中踩过的坑和性能优化的关键技巧目标是让你看完后能清晰地掌握如何为你的LLM推理引擎打造一颗强健的“心脏”。2. 核心数据结构设计思路拆解在动手写代码之前我们必须想清楚这两个数据结构要承载什么信息以及它们之间如何协作。这个设计过程本质上是在对LLM推理请求的生命周期进行建模。2.1 请求Req的生命周期与状态机一个推理请求从接收到最终完成会经历多个状态。Req对象需要完整地记录这个旅程。我们先定义它的核心状态等待中PENDING请求刚被接收放入等待队列等待有可用的计算资源如GPU来处理。运行中RUNNING请求已被调度到计算设备上正在执行前向传播生成token。已完成FINISHED请求正常结束可能是生成了结束符或者达到了最大生成长度。已中止CANCELLED请求被用户或系统主动取消。错误ERROR在处理过程中发生了不可恢复的错误。为什么需要状态机这是实现高效调度和资源管理的基础。调度器可以根据状态快速过滤出可运行的请求PENDING监控系统可以统计运行中请求的数量而客户端轮询时也能准确知道请求是还在计算、已经完成还是出了错。在设计时我们用一个枚举类来清晰定义这些状态。除了状态一个Req还必须包含以下核心信息请求IDreq_id全局唯一的标识符用于在日志、监控和客户端回调中追踪这个请求。输入提示词prompt用户输入的文本这是生成的起点。采样参数sampling_params一个SamplingParams对象实例包含了所有控制生成行为的参数。输出相关需要记录已生成的token序列output_tokens及其对应的文本output_text以及生成是否被停止的原因stop_reason如遇到停止词或达到最大长度。性能与统计记录请求创建时间、开始运行时间、结束时间用于计算排队延迟、计算延迟等关键指标。回调与元数据可包含一个结果回调函数以及用户自定义的元数据字典用于更灵活的业务集成。2.2 采样参数SamplingParams的职责边界SamplingParams是一个相对独立但至关重要的配置对象。它的职责是告诉推理引擎“如何生成”而不是“生成什么”。它的设计必须兼顾通用性和扩展性覆盖绝大多数主流模型的采样方式。我们可以将其参数分为几大类长度控制参数如max_tokens最大生成长度、min_tokens最小生成长度。max_tokens必须设置这是防止生成无限长文本的安全阀。停止条件参数如stop停止词列表。当生成的文本包含列表中的任何字符串时停止生成。这是一个列表因为用户可能设置多个停止条件。采样策略参数这是最复杂的部分主要包括temperature温度控制随机性。越高如1.0输出越随机、有创意越低如0.1输出越确定、保守。top_p核采样另一种控制随机性的方法从累积概率超过p的最小token集合中采样。常与温度一起使用。top_kTop-K采样仅从概率最高的k个token中采样。frequency_penalty和presence_penalty频率惩罚和存在惩罚用于降低重复token出现的概率。特殊token处理如是否在输出中包含输入提示词echo是否跳过特殊的开始tokenskip_special_tokens等。设计的关键在于SamplingParams应该提供合理的默认值并且能够验证用户传入的参数是否有效、是否互斥。例如temperature为0时通常表示贪婪解码总是选概率最高的token此时再设置top_p或top_k可能就没有意义需要在初始化时进行验证或给出明确的行为定义。2.3 数据结构间的交互关系Req和SamplingParams是典型的组合关系一个Req“拥有” 一个SamplingParams。这种设计的好处是解耦采样逻辑被封装在SamplingParams中Req只负责持有和传递它。未来如果我们增加新的采样方法如Beam Search只需要扩展SamplingParamsReq的结构基本不受影响。复用相同的采样参数配置可以被多个请求共享虽然通常是每个请求独立实例化。序列化友好SamplingParams可以很容易地被序列化为JSON通过网络API传递然后由服务端反序列化并附加到Req上。在实现时我们会将SamplingParams设计为一个不可变的数据类使用dataclass(frozenTrue)一旦创建其参数就不能被修改。这保证了在请求处理过程中生成策略不会意外改变避免了难以调试的并发问题。3. Req 类的详细实现与核心要点理论分析完毕现在我们进入实战环节。我们将使用Python的dataclass来定义Req因为它能自动生成__init__、__repr__等方法让代码更简洁。但要注意dataclass主要用于存储数据一些复杂的逻辑和方法需要我们手动添加。3.1 基础字段定义与初始化逻辑首先我们定义状态枚举和Req的基础结构。from enum import Enum from dataclasses import dataclass, field from typing import List, Optional, Callable, Any, Dict import time import uuid class ReqStatus(Enum): PENDING pending RUNNING running FINISHED finished CANCELLED cancelled ERROR error dataclass class SamplingParams: # 我们将在下一节详细实现它 max_tokens: int temperature: float 1.0 top_p: float 1.0 stop: Optional[List[str]] None # ... 其他参数 dataclass class Req: req_id: str prompt: str sampling_params: SamplingParams # 状态与输出 status: ReqStatus ReqStatus.PENDING output_tokens: List[int] field(default_factorylist) output_text: str stop_reason: Optional[str] None # 如 “length”, “stop”, “cancel” # 时间戳 created_time: float field(default_factorytime.time) start_time: Optional[float] None finish_time: Optional[float] None # 回调与元数据 callback: Optional[Callable[[‘Req’], None]] None metadata: Dict[str, Any] field(default_factorydict) def __post_init__(self): 数据类初始化后的校验和设置 if self.sampling_params.max_tokens 0: raise ValueError(fmax_tokens must be positive, got {self.sampling_params.max_tokens}) # 可以在此处添加更多业务逻辑校验关键点解析与实操心得req_id的生成在上面的代码中我们要求外部传入req_id。但在实际API服务器中更常见的做法是在Req的工厂方法或构造函数中自动生成。例如可以使用uuid.uuid4().hex来生成一个唯一的字符串。这样能保证ID的全局唯一性避免冲突。field(default_factory...)的使用对于像output_tokens列表、metadata字典这样的可变对象绝对不要使用default[]或default{}。这是因为dataclass的默认值在类定义时就会被创建所有未显式提供该字段的Req实例将共享同一个列表或字典对象导致灾难性的数据混乱。必须使用default_factorylist或default_factorydict这样每个实例都会获得一个全新的对象。__post_init__方法这是dataclass提供的一个钩子在对象初始化完成后自动调用。它是进行参数验证和衍生数据初始化的绝佳位置。比如这里我们检查了max_tokens是否合法。3.2 状态迁移与线程安全在并发环境下多个线程如调度线程、执行线程、取消请求的线程可能同时修改同一个Req的状态。不加保护的状态修改会导致数据竞争状态机混乱。import threading from dataclasses import dataclass, field from typing import Optional dataclass class Req: # ... 其他字段同上 ... status: ReqStatus ReqStatus.PENDING _status_lock: threading.Lock field(default_factorythreading.Lock, initFalse, reprFalse) def set_status(self, new_status: ReqStatus): 线程安全地更新状态并可添加状态迁移逻辑 with self._status_lock: # 简单的状态迁移校验可根据需要更复杂 if self.status ReqStatus.FINISHED and new_status ! ReqStatus.FINISHED: raise RuntimeError(fCannot change status from {self.status} to {new_status}) if self.status ReqStatus.CANCELLED and new_status not in [ReqStatus.CANCELLED, ReqStatus.FINISHED]: # 已取消的请求可能最终会被标记为完成但不能重新运行 raise RuntimeError(fCannot change status from {self.status} to {new_status}) old_status self.status self.status new_status # 记录状态变更的时间戳 if new_status ReqStatus.RUNNING and self.start_time is None: self.start_time time.time() elif new_status in [ReqStatus.FINISHED, ReqStatus.CANCELLED, ReqStatus.ERROR]: if self.finish_time is None: self.finish_time time.time() # 可以在这里触发状态变更事件或日志 # self._on_status_change(old_status, new_status) return old_status def get_status(self) - ReqStatus: 线程安全地获取状态 with self._status_lock: return self.status注意事项与避坑指南锁的对象我们为每个Req实例单独分配一个锁_status_lock。这意味着不同请求的状态更新不会相互阻塞只有对同一个请求的并发修改才会被序列化这对高并发场景至关重要。如果使用全局锁性能会成为瓶颈。状态迁移逻辑set_status方法中加入了简单的校验逻辑。例如一个已经完成FINISHED的请求不应该再被改为运行中RUNNING。这属于业务规则的守护能提前发现程序逻辑错误。你可以根据你的调度器设计定义更严谨的状态迁移图。时间戳记录在状态变更时记录时间点是非常有用的。start_time在进入RUNNING时记录finish_time在终态完成、取消、错误时记录。这样我们就能轻松计算出排队延迟 start_time - created_time和计算延迟 finish_time - start_time这是监控服务健康度的核心指标。隐藏内部锁使用initFalse, reprFalse将_status_lock排除在构造函数参数和字符串表示之外保持Req对外接口的简洁。3.3 输出管理与完成回调当推理引擎生成每一个新的token时需要将其追加到Req的输出中。同时当请求完成无论成功与否时应触发用户预设的回调函数。def append_token(self, token_id: int, token_text: str): 追加生成的token。注意此方法可能在多线程环境下被调用。 # 如果对output_tokens/output_text的修改非常频繁且需要考虑极高并发 # 这里也可能需要加锁。但在典型调度模式下一个Req在同一时刻只由一个工作线程处理 # 所以可能不需要。为安全起见我们可以加锁。 with self._status_lock: # 复用状态锁或者为输出单独设锁 self.output_tokens.append(token_id) self.output_text token_text def mark_finished(self, stop_reason: str): 标记请求完成并执行回调 old_status self.set_status(ReqStatus.FINISHED) self.stop_reason stop_reason if self.callback is not None: try: self.callback(self) except Exception as e: # 务必捕获回调中的异常避免影响主流程 print(fError executing callback for request {self.req_id}: {e}) # 最好使用日志库记录错误实操心得回调异常处理用户提供的回调函数可能抛出任何异常。绝对不能让这个异常向上传播导致整个请求处理线程崩溃。必须用try-except块包裹并妥善记录错误。这是服务稳定性的一个关键细节。流式输出考虑上述append_token方法适用于非流式一次返回所有结果的场景。如果支持流式响应Server-Sent Events你可能需要更复杂的机制比如一个线程安全的队列让生成线程不断放入token而另一个网络发送线程从中取出并推送给客户端。这时Req可能需要包含一个asyncio.Queue或类似的结构。4. SamplingParams 类的实现与参数校验现在我们来构建SamplingParams。它的目标是成为一个自包含、自验证的配置对象。4.1 参数定义与默认值from dataclasses import dataclass, field from typing import List, Optional import math dataclass(frozenTrue) # 设置为不可变 class SamplingParams: 控制文本生成的参数。 # 长度控制 max_tokens: int min_tokens: int 1 # 停止条件 stop: Optional[List[str]] None stop_token_ids: Optional[List[int]] None # 在token_id层面的停止条件 # 采样策略 temperature: float 1.0 top_p: float 1.0 top_k: int -1 # -1 表示禁用top-k repetition_penalty: float 1.0 frequency_penalty: float 0.0 presence_penalty: float 0.0 # 特殊处理 echo: bool False # 是否在输出中回显输入提示 skip_special_tokens: bool True # 是否跳过特殊token如|endoftext| # 用于logits处理的额外参数 logprobs: Optional[int] None # 返回top-n logprobs def __post_init__(self): 验证参数合法性 # 1. 基础范围校验 if self.max_tokens 0: raise ValueError(fmax_tokens must be positive, got {self.max_tokens}) if self.min_tokens 0: raise ValueError(fmin_tokens must be non-negative, got {self.min_tokens}) if self.min_tokens self.max_tokens: raise ValueError(fmin_tokens ({self.min_tokens}) cannot be greater than max_tokens ({self.max_tokens})) if not math.isfinite(self.temperature) or self.temperature 0: raise ValueError(ftemperature must be non-negative and finite, got {self.temperature}) if not 0.0 self.top_p 1.0: raise ValueError(ftop_p must be in (0.0, 1.0], got {self.top_p}) if self.top_k -1 or self.top_k 0: raise ValueError(ftop_k must be -1 (disable) or positive, got {self.top_k}) if self.repetition_penalty 0.0: raise ValueError(frepetition_penalty must be positive, got {self.repetition_penalty}) if self.frequency_penalty 0.0 or self.frequency_penalty 2.0: raise ValueError(ffrequency_penalty should be between 0 and 2, got {self.frequency_penalty}) if self.presence_penalty 0.0 or self.presence_penalty 2.0: raise ValueError(fpresence_penalty should be between 0 and 2, got {self.presence_penalty}) # 2. 互斥或特殊逻辑校验 if self.temperature 0.0: # 温度0意味着贪婪解码此时top_p/top_k无效可以警告或忽略 # 我们选择静默忽略因为有些用户可能仍会设置我们只需在采样时优先考虑temperature0 pass if self.top_k 0 and self.top_p 1.0: # 同时启用top-k和top-p是常见的无需报错但采样逻辑需要正确处理两者的结合。 pass # 3. 规范化stop字段可选 # 确保stop是列表且不为空列表空列表在语义上可能表示“无停止词” if self.stop is not None and len(self.stop) 0: # 将空列表转为None便于后续判断 object.__setattr__(self, stop, None)核心细节与设计考量frozenTrue不可变这是关键设计。一旦SamplingParams实例被创建其所有字段都不能被修改。这消除了在请求处理过程中参数被意外更改的风险对于并发安全至关重要也使得对象可以被安全地哈希和缓存。参数验证的完整性__post_init__方法进行了严格的输入校验。这遵循了“快速失败”原则错误在请求被加入队列之前就会被发现而不是在推理中途才崩溃便于客户端调试。默认值的选择默认值参考了主流API如OpenAI。例如temperature1.0,top_p1.0意味着默认使用多项式采样非贪婪。top_k-1用负数表示禁用比用None更直观因为int类型在后续计算中更方便。object.__setattr__技巧因为frozenTrue我们不能直接self.stop None。但__post_init__是特例允许我们使用object.__setattr__来修改字段以完成对空列表的清理工作。4.2 工具方法与参数转换为了让SamplingParams更好用我们可以添加一些工具方法。def to_dict(self) - dict: 将参数转换为字典便于序列化如返回给API调用者 # 注意要过滤掉值为None的字段或者保留取决于你的API设计 result {} for key, value in self.__dict__.items(): if value is not None and not (isinstance(value, list) and len(value) 0): # 这里简单处理可以根据需要定制 result[key] value return result classmethod def from_dict(cls, data: dict) - SamplingParams: 从字典创建SamplingParams用于反序列化如从API请求体解析 # 过滤掉None值因为dataclass的字段可能有默认值 filtered_data {k: v for k, v in data.items() if v is not None} # 可以在此处进行额外的数据清洗或转换 # 例如确保stop是列表 if stop in filtered_data and isinstance(filtered_data[stop], str): filtered_data[stop] [filtered_data[stop]] return cls(**filtered_data) def get_sampling_config(self) - dict: 提取用于实际采样函数的配置字典。 这个方法将‘SamplingParams’中的参数转换为底层采样库如transformers所需的格式。 config { max_new_tokens: self.max_tokens, min_new_tokens: self.min_tokens, temperature: self.temperature if self.temperature 0 else 1e-7, # 处理temperature0 top_p: self.top_p, repetition_penalty: self.repetition_penalty, do_sample: self.temperature 0, # 温度0才进行采样 } if self.top_k 0: config[top_k] self.top_k if self.stop_token_ids: config[stop_token_ids] self.stop_token_ids # frequency_penalty和presence_penalty可能需要特定库支持 # 如果底层库不支持可以在这里实现后处理逻辑 return config工具方法的价值序列化与反序列化to_dict和from_dict是连接HTTP API与核心逻辑的桥梁。API层接收到JSON请求体后可以轻松地将其转换为SamplingParams对象。to_dict在将推理结果返回给客户端时也很有用可以附带使用的参数。适配层get_sampling_config方法是一个关键的“适配器”。不同的推理后端如Hugging Face Transformers, vLLM, TGI的采样接口可能略有不同。这个方法负责将我们统一的参数格式翻译成后端引擎能理解的格式。这集中了转换逻辑使核心的Req和调度代码与后端解耦。5. 在调度器中的集成与应用设计好的数据结构最终要为调度服务。让我们看看Req和SamplingParams如何融入一个简化的调度循环。5.1 请求队列与状态管理调度器通常维护几个核心队列import heapq import threading from typing import Dict, List, Optional class SimpleScheduler: def __init__(self): self._pending_queue: List[Req] [] # 等待队列 self._running_requests: Dict[str, Req] {} # 正在运行的请求 self._request_registry: Dict[str, Req] {} # 所有请求的注册表 self._lock threading.Lock() def add_request(self, req: Req): 接收一个新请求 with self._lock: if req.req_id in self._request_registry: raise ValueError(fRequest ID {req.req_id} already exists.) self._request_registry[req.req_id] req # 这里可以加入优先级逻辑例如根据创建时间或参数 heapq.heappush(self._pending_queue, (req.created_time, req.req_id, req)) # 触发调度决策 self._schedule() def _schedule(self): 调度逻辑将等待队列中的请求分配到空闲的计算槽位 # 这是一个简化的示例实际中你需要知道当前有多少空闲的“槽位”如GPU批处理空位 available_slots self._get_available_slots() with self._lock: while available_slots 0 and self._pending_queue: _, req_id, req heapq.heappop(self._pending_queue) # 再次检查请求状态防止在队列中时被取消 if req.get_status() ReqStatus.PENDING: req.set_status(ReqStatus.RUNNING) self._running_requests[req_id] req # 这里应该将req交给具体的“执行器”Worker去运行 # self._assign_to_worker(req) available_slots - 1 # 如果请求已被取消则跳过继续调度下一个调度逻辑解析多队列设计_pending_queue通常用优先队列heapq实现可以根据优先级如创建时间、付费等级决定调度顺序。_running_requests字典便于快速查找正在运行的请求例如用于取消操作。_request_registry记录了所有请求用于全局查询。状态一致性在将请求从等待队列移入运行集之前必须再次检查其状态。因为用户可能在请求排队时发起取消操作此时状态已变为CANCELLED不应再被调度。锁的粒度调度器的方法如add_request,_schedule在修改共享数据结构队列、字典时都需要加锁。但Req内部的状态修改由其自己的锁保护两者配合实现线程安全。5.2 采样参数在推理循环中的使用当请求被分配给执行器后执行器会进入一个生成循环。以下是这个循环中如何使用SamplingParams的简化示意class InferenceWorker: def process_request(self, req: Req, model, tokenizer): 处理单个请求的生成循环简化版 try: input_ids tokenizer.encode(req.prompt) # 初始化生成状态 generated_ids [] stop_reason None for step in range(req.sampling_params.max_tokens): # 1. 前向传播获取下一个token的logits # logits model.forward(input_ids generated_ids) # 2. 应用SamplingParams中的参数对logits进行处理 # processed_logits self._apply_sampling_params(logits, req.sampling_params, generated_ids) # 3. 采样下一个token_id # next_token_id self._sample(processed_logits, req.sampling_params.temperature) # 4. 将token添加到生成序列 # generated_ids.append(next_token_id) # token_text tokenizer.decode([next_token_id], skip_special_tokensFalse) # req.append_token(next_token_id, token_text) # 5. 检查停止条件 # if self._check_stop_condition(generated_ids, req.sampling_params, tokenizer): # stop_reason stop # break # if len(generated_ids) req.sampling_params.max_tokens: # stop_reason length # break # 6. 如果是流式可以在这里将token发送给客户端 pass # 循环结束标记请求完成 req.mark_finished(stop_reason or length) except Exception as e: req.set_status(ReqStatus.ERROR) req.stop_reason ferror: {e} # 记录日志 raise finally: # 无论成功失败从运行集合中移除 scheduler._remove_from_running(req.req_id) def _apply_sampling_params(self, logits, params: SamplingParams, generated_ids: List[int]): 根据采样参数修改logits。这是一个概念性实现。 # 应用温度 if params.temperature 0: logits logits / params.temperature else: # temperature0贪婪解码直接取argmax无需采样分布 pass # 应用top-p (nucleus sampling) if params.top_p 1.0: # 对logits进行排序和累积概率计算将累积概率超过top_p的token置为负无穷 pass # 应用top-k if params.top_k 0: # 仅保留概率最高的k个token其余置为负无穷 pass # 应用重复惩罚 (repetition penalty) if params.repetition_penalty ! 1.0: for token_id in set(generated_ids): # 对已出现过的token进行惩罚 if logits[token_id] 0: logits[token_id] logits[token_id] / params.repetition_penalty else: logits[token_id] logits[token_id] * params.repetition_penalty # frequency_penalty和presence_penalty实现类似但通常基于出现频率 return logits在推理循环中的关键点参数传递整个生成循环都围绕着req.sampling_params这个对象进行。它集中了所有控制逻辑使得循环主体清晰。停止条件判断停止条件检查停止词、最大长度是循环中的关键一步它依赖于sampling_params中的stop列表和max_tokens。注意停止词检查通常在解码为文本后进行可能需要处理跨token的边界情况这是一个常见的难点。Logits后处理_apply_sampling_params函数展示了如何将抽象的采样参数转化为对模型输出logits的具体数学变换。这是采样策略的核心。在实际项目中这部分逻辑可能非常复杂且对性能敏感通常会使用优化过的CUDA内核或调用深度学习框架的高级API来实现。6. 常见问题、调试技巧与性能优化在实际实现和运行中你会遇到各种各样的问题。下面是我从实践中总结的一些典型问题和解决方案。6.1 状态不一致与并发Bug问题现象请求状态莫名卡住或者输出token丢失/重复。排查思路检查锁的覆盖范围确保所有修改Req内部状态status,output_tokens,output_text的地方都在锁的保护下。使用threading.Lock时要特别注意锁的范围是否足够。添加诊断日志在Req.set_status方法中可以添加详细的日志记录状态变更的时间、线程和从什么状态变为什么状态。这能帮你追踪状态机的异常跳转。使用断言在开发阶段可以在Req的关键方法开头加入断言检查状态是否处于预期。例如在append_token中可以断言self.status ReqStatus.RUNNING。一个实用的调试技巧实现一个Req的__repr__方法让它输出关键信息并确保它是线程安全的只读访问字段。def __repr__(self): with self._status_lock: return (fReq(req_id{self.req_id[:8]}..., status{self.status.value}, fprompt_len{len(self.prompt)}, output_len{len(self.output_tokens)}))6.2 采样参数验证的边界情况问题现象某些参数组合导致生成结果异常或程序崩溃。排查与解决温度temperature为0这表示贪婪解码。你需要确保当temperature0时采样函数是直接取argmax而不是进行概率采样。同时top_p和top_k参数在这种情况下应该被忽略。最好在SamplingParams的验证或get_sampling_config方法中明确处理这一点。停止词stop包含空字符串或长字符串空字符串会导致立即停止这可能是用户错误。长字符串可能跨越多个token需要实现一个高效的字符串匹配算法如Aho-Corasick自动机来在生成的文本流中进行检测。对于简单的场景可以在每生成一个token后将当前全部输出文本与停止词列表进行匹配但效率较低。max_tokens设置过大这可能导致生成时间过长占用资源。可以在API网关或调度器层面设置一个全局上限并优先使用用户设置的值但不能超过全局上限。6.3 性能优化点SamplingParams的哈希与缓存由于SamplingParams是不可变的我们可以实现它的__hash__方法。这样具有相同参数配置的请求可以共享同一个SamplingParams对象甚至可以对预处理结果如停止词构建的AC自动机进行缓存。dataclass(frozenTrue) class SamplingParams: # ... 字段定义 ... def __hash__(self): # 将关键字段组成一个元组来生成哈希值 # 注意对于列表如stop需要先转换为元组 stop_tuple tuple(self.stop) if self.stop else None stop_ids_tuple tuple(self.stop_token_ids) if self.stop_token_ids else None return hash((self.max_tokens, self.temperature, self.top_p, self.top_k, stop_tuple, stop_ids_tuple))批量请求的采样参数分组在批量推理如vLLM的连续批处理中一次前向传播处理多个请求。如果这些请求的采样参数如temperature,top_p不同底层计算会变得复杂。一种优化策略是将参数相同的请求尽量放在同一个批次中处理。调度器可以根据SamplingParams的哈希值对等待队列中的请求进行分组。避免频繁的内存分配在生成循环中append_token会频繁修改列表和字符串。对于超长文本生成字符串拼接可能效率较低。可以考虑使用io.StringIO或列表先收集token文本最后再一次性连接。对于极端性能场景甚至可以考虑使用预分配的数组。6.4 扩展性考虑当前设计已经预留了扩展点新的采样策略如需添加新的参数如typical_p,eta_cutoff只需在SamplingParams中添加字段并在__post_init__和get_sampling_config中相应处理即可。请求优先级可以在Req中添加一个priority字段并修改调度器的_pending_queue为基于优先级的堆队列。请求依赖更复杂的场景可能需要支持请求链前一个请求的输出作为后一个的输入。这可以通过在Req的metadata中存储前置请求的ID并由调度器解析依赖关系来实现。通过以上六个部分的详细拆解我们从设计理念、代码实现、集成应用到问题排查完整地构建了LLM推理服务核心的Req与SamplingParams数据结构。它们虽不直接涉及复杂的模型计算却是整个服务稳定、高效、易用的基石。一个好的基础设计能让后续添加流式输出、函数调用、多模态等高级功能变得顺理成章。在下一篇文章中我们将基于这个坚实的基础开始构建调度器的核心引擎。