机器学习推理的延迟与成本怎么一起测固定输入、批次和设备推理服务的“快”和“省”不能靠两个孤立的平均值判断。输入长度、Batch 组成、设备型号和并发到达方式不一致延迟与显存占用就没有可比性。1. 固定一次测量的输入先冻结模型与权重版本、精度、输入长度分布、Batch 策略和预热次数。延迟记录 TTFT、单请求完成时间和排队时间成本记录占用的设备时长、模型调用量或实际账单口径。无法采集的项直接标为空不用经验值补齐。2. 把调度器当成待验证对象动态组 Batch 会提高部分负载下的设备利用率也可能让短请求被长请求拖住。测试时固定请求集分别运行串行、固定 Batch 与 Token 预算调度每轮只改一种策略并检查输出是否一致。如果触发显存不足应保存输入长度摘要和调度决策缩小 Token 预算后重跑。下面的日志与参数是构造输入真正的边界必须由目标设备上的测试得到。3. Token 预算调度器utilization.gpu [%], memory.used [MiB] 12 %, 18452 MiB 18 %, 19200 MiB 98 %, 79820 MiB 99 %, 79850 MiB -- 触发 OOM 擦边抛错排队上下文被清空 5 %, 8192 MiBimport asyncio import time from typing import List, Dict, Any class TokenBudgetController: 基于 Token 预算与 KV Cache 预测的动态 Batch 调度器。 通过限制单 Batch 的最大 Token 总量避免显存溢出与长短请求相互阻塞。 def __init__(self, max_token_budget: int 8192, max_wait_ms: float 20.0): self.max_token_budget max_token_budget self.max_wait_ms max_wait_ms self.pending_queue: asyncio.Queue asyncio.Queue() self._is_running False async def submit_request(self, req_id: str, prompt_tokens: int, max_new_tokens: int) - Dict[str, Any]: future asyncio.get_event_loop().create_future() req_item { id: req_id, prompt_tokens: prompt_tokens, max_new_tokens: max_new_tokens, est_peak_tokens: prompt_tokens max_new_tokens, future: future, ts: time.time() } await self.pending_queue.put(req_item) return await future async def schedule_loop(self): self._is_running True while self._is_running: batch: List[Dict[str, Any]] [] current_budget 0 start_time time.time() while True: try: # 计算剩余等待超时时间 elapsed_ms (time.time() - start_time) * 1000 remaining_timeout max(0.0, (self.max_wait_ms - elapsed_ms) / 1000.0) if self.pending_queue.empty() and batch: break item await asyncio.wait_for(self.pending_queue.get(), timeoutremaining_timeout) # 检查预算约束 if current_budget item[est_peak_tokens] self.max_token_budget: batch.append(item) current_budget item[est_peak_tokens] else: # 预算超限放回队列头部待下次调度 await self.pending_queue.put(item) break except asyncio.TimeoutError: # 到达组包窗口超时阈值强制截断出队 break except Exception as err: # 捕获调度异常避免主循环异常退出 print(f[Scheduler Error] 调度过程发生未预期异常: {str(err)}) break if batch: asyncio.create_task(self._process_batch(batch)) else: await asyncio.sleep(0.005) async def _process_batch(self, batch: List[Dict[str, Any]]): 模拟 GPU 推理执行引擎 try: # 模拟推理延时与 Batch 内总 Token 数呈非线性关系 batch_tokens sum(x[est_peak_tokens] for x in batch) simulated_latency 0.05 (batch_tokens / 10000.0) * 0.1 await asyncio.sleep(simulated_latency) for item in batch: if not item[future].done(): item[future].set_result({ status: SUCCESS, latency_ms: (time.time() - item[ts]) * 1000, processed_tokens: item[est_peak_tokens] }) except Exception as e: for item in batch: if not item[future].done(): item[future].set_exception(e)4. 复核清单请求集是否固定模型、设备、预热与到达方式。TTFT、排队、完成时间和 Token 是否分别记录。调度方案是否使用相同输入并校验输出一致。OOM 或超时是否保留请求尺寸与调度决策。总结“延迟和成本怎么一起看”应以清晰的条件和脚本复核。先记录边界再解释结果。