多模型推理 Pipeline 的性能优化——批处理、模型并行与响应缓存一、多模型推理场景的性能挑战在 AI 应用中单一模型推理已经不能满足复杂业务需求多模型组合推理成为常态。例如内容审核场景需要同时调用文本审核、图像审核、敏感词过滤三个模型智能客服系统中可能需要意图识别、实体抽取、情感分析等多个模型的串行或并行调用。多模型推理 Pipeline 面临的核心挑战是每个模型的推理延迟不同串行调用会导致总延迟线性累加而简单的并行调用可能会因模型抢占 GPU 显存导致整体吞吐量下降。二、批处理优化——动态批处理策略推理请求的批处理是提升 GPU 利用率的最直接手段。但固定大小的批处理在面对波动流量时效果不佳需要根据请求积压情况动态调整。/** * 动态批处理器——根据请求队列深度自适应调整批次大小 * 在延迟和吞吐量之间实现动态平衡。 * * 为什么不用固定 batchSize流量低谷时小 batch 可能为 1 个请求 * 高峰时可达几十个固定大小会导致低谷时延迟增加或高峰时显存溢出。 */ Service public class DynamicBatchProcessor { private static final Logger log LoggerFactory.getLogger( DynamicBatchProcessor.class); // 为什么上限设 32超过此值后大部分模型的 GPU 利用率提升趋于平缓 // 但延迟会因序列长度填充而线性增长padding 效应 private static final int MAX_BATCH_SIZE 32; private static final int MIN_BATCH_SIZE 1; // 为什么等待周期设 10ms在 GPU 上单次推理通常在 10~100ms // 10ms 的等待引入的额外延迟相对于推理时间可忽略 private static final long BATCH_WAIT_MS 10; private final BlockingQueueInferenceRequest requestQueue; private final InferenceEngine inferenceEngine; private final MeterRegistry meterRegistry; public DynamicBatchProcessor( InferenceEngine inferenceEngine, MeterRegistry meterRegistry) { this.requestQueue new LinkedBlockingQueue(1000); this.inferenceEngine inferenceEngine; this.meterRegistry meterRegistry; startBatchLoop(); } private void startBatchLoop() { Thread batchThread new Thread(this::batchProcessLoop, dynamic-batch-processor); batchThread.setDaemon(true); batchThread.start(); } private void batchProcessLoop() { while (!Thread.currentThread().isInterrupted()) { try { ListInferenceRequest batch collectBatch(); if (batch.isEmpty()) { continue; } long startTime System.nanoTime(); ListInferenceResult results inferenceEngine .batchInference(batch); long elapsedMs TimeUnit.NANOSECONDS .toMillis(System.nanoTime() - startTime); distributeResults(batch, results); // 记录批处理指标用于后续调优 meterRegistry.summary(inference.batch.size) .record(batch.size()); meterRegistry.timer(inference.batch.duration) .record(Duration.ofMillis(elapsedMs)); if (batch.size() 16 elapsedMs 200) { log.info(大批次高延迟, batchSize{}, elapsedMs{}, batch.size(), elapsedMs); } } catch (Exception e) { log.error(批处理循环异常, 重启循环, e); try { Thread.sleep(1000); } catch (InterruptedException ie) { Thread.currentThread().interrupt(); break; } } } } private ListInferenceRequest collectBatch() throws InterruptedException { ListInferenceRequest batch new ArrayList(); // 先取队列头部非阻塞 InferenceRequest head requestQueue.poll(); if (head null) { return batch; } batch.add(head); // 在等待窗口内尽可能多地收集请求 // 为什么放在 while 循环外的 poll 已经取到数据 // 第一个请求已有数据无需等待后续的 drainTo 可以快速收集 requestQueue.drainTo(batch, MAX_BATCH_SIZE - 1); // 如果批次未满等待短暂时间收集更多请求 if (batch.size() MAX_BATCH_SIZE) { long deadline System.currentTimeMillis() BATCH_WAIT_MS; while (batch.size() MAX_BATCH_SIZE) { long remaining deadline - System.currentTimeMillis(); if (remaining 0) { break; } InferenceRequest req requestQueue.poll( remaining, TimeUnit.MILLISECONDS); if (req ! null) { batch.add(req); } else { break; } } } return batch; } private void distributeResults(ListInferenceRequest requests, ListInferenceResult results) { for (int i 0; i requests.size(); i) { requests.get(i).complete(results.get(i)); } } }三、模型并行策略设计在多模型推理场景中需要根据模型间的依赖关系选择合适的并行策略graph TD subgraph Pipeline 模式——串行依赖 A1[意图识别] --|result| B1[实体抽取] B1 --|entities| C1[情感分析] C1 --|sentiment| D1[响应生成] end subgraph Scatter-Gather 模式——无依赖并行 A2[文本审核] B2[图像审核] C2[敏感词过滤] A2 --|pass/fail| D2[聚合判决] B2 --|pass/fail| D2 C2 --|pass/fail| D2 end style A1 fill:#bbf,stroke:#333 style A2 fill:#bbf,stroke:#333 style B2 fill:#bbf,stroke:#333 style C2 fill:#bbf,stroke:#333/** * 多模型编排器——根据模型的依赖拓扑自动选择串行或并行执行路径。 * * 为什么抽象为 DAG 拓扑而非硬编码调用顺序 * 模型组合方案会随业务迭代持续增加拓扑结构便于扩展和可视化。 */ Component public class ModelPipelineOrchestrator { private static final Logger log LoggerFactory.getLogger( ModelPipelineOrchestrator.class); private final MapString, ModelExecutor executorMap; private final ExecutorService parallelExecutor; public ModelPipelineOrchestrator(MapString, ModelExecutor executorMap) { this.executorMap executorMap; // 为什么使用 VirtualThread 而非固定线程池 // 模型调用主要是 IO 等待GPU 推理虚拟线程避免阻塞平台线程 this.parallelExecutor Executors.newVirtualThreadPerTaskExecutor(); } /** * 根据 DAG 拓扑执行多模型推理 Pipeline。 * * param pipelineDef Pipeline定义描述模型节点及依赖关系 * param input 原始输入 * return 所有节点的推理结果 */ public MapString, Object execute(PipelineDefinition pipelineDef, Object input) { if (pipelineDef.getNodes().isEmpty()) { return Collections.emptyMap(); } // 拓扑排序确定执行顺序 ListString topologicalOrder topologicalSort(pipelineDef); MapString, Object context new ConcurrentHashMap(); context.put(_input, input); for (String nodeId : topologicalOrder) { PipelineNode node pipelineDef.getNode(nodeId); if (node.getDependencies().isEmpty()) { // 无依赖节点——可以直接执行 Object result executeNode(node, context); context.put(nodeId, result); } else if (canParallelExecute(node, pipelineDef, context)) { // 所有依赖已满足——并行执行 ListCompletableFutureVoid futures collectParallelFutures( node, pipelineDef, context); try { // 为什么设置单节点2倍超时的总超时 // 并行执行的理论耗时应等于最慢的单个节点 CompletableFuture.allOf( futures.toArray(new CompletableFuture[0])) .get(node.getTimeout().multipliedBy(2).toMillis(), TimeUnit.MILLISECONDS); } catch (TimeoutException e) { log.error(并行推理超时, node{}, timeout{}, nodeId, node.getTimeout()); throw new InferenceTimeoutException( 并行推理超时: nodeId, e); } catch (Exception e) { log.error(并行推理异常, node{}, nodeId, e); throw new InferenceExecutionException( 并行推理异常: nodeId, e); } } } return context; } private Object executeNode(PipelineNode node, MapString, Object context) { ModelExecutor executor executorMap.get(node.getModelName()); if (executor null) { throw new IllegalStateException( 未找到模型执行器: node.getModelName()); } Object resolvedInput resolveInput(node, context); return executor.execute(resolvedInput); } private ListCompletableFutureVoid collectParallelFutures( PipelineNode node, PipelineDefinition pipelineDef, MapString, Object context) { // 收集所有可以并行执行的同级节点 return node.getParallelGroup().stream() .map(parallelNodeId - CompletableFuture.runAsync(() - { PipelineNode parallelNode pipelineDef .getNode(parallelNodeId); Object result executeNode(parallelNode, context); context.put(parallelNodeId, result); }, parallelExecutor)) .collect(Collectors.toList()); } private ListString topologicalSort(PipelineDefinition pipelineDef) { // 基于 Kahn 算法进行拓扑排序 MapString, Integer inDegree new HashMap(); MapString, ListString adjacency new HashMap(); for (PipelineNode node : pipelineDef.getNodes()) { inDegree.putIfAbsent(node.getId(), 0); for (String dep : node.getDependencies()) { adjacency.computeIfAbsent(dep, k - new ArrayList()) .add(node.getId()); inDegree.merge(node.getId(), 1, Integer::sum); } } QueueString queue new LinkedList(); for (Map.EntryString, Integer entry : inDegree.entrySet()) { if (entry.getValue() 0) { queue.offer(entry.getKey()); } } ListString result new ArrayList(); while (!queue.isEmpty()) { String current queue.poll(); result.add(current); for (String neighbor : adjacency.getOrDefault(current, Collections.emptyList())) { int degree inDegree.merge(neighbor, -1, Integer::sum); if (degree 0) { queue.offer(neighbor); } } } if (result.size() ! pipelineDef.getNodes().size()) { throw new IllegalStateException( Pipeline定义存在循环依赖, nodes pipelineDef.getNodeIds()); } return result; } private boolean canParallelExecute(PipelineNode node, PipelineDefinition pipelineDef, MapString, Object context) { return node.getParallelGroup() ! null node.getDependencies().stream() .allMatch(context::containsKey); } private Object resolveInput(PipelineNode node, MapString, Object context) { // 解析节点输入可能引用上游节点输出或原始输入 return context.get(node.getInputSource()); } }四、响应缓存策略对于相似度较高的输入可以直接复用历史推理结果减少 GPU 计算开销。/** * 推理结果缓存——基于输入向量相似度的语义缓存。 * * 为什么用语义缓存而非精确匹配两个语义相同但表述不同的请求 * 如今天天气怎么样与今天天气如何在 LLM 推理中结果应当一致。 */ Component public class SemanticInferenceCache { private final CacheString, CachedResult cache; // 为什么相似度阈值设 0.95 而非 0.90 // 过度宽松的阈值可能导致语义漂移返回不准确的缓存结果 private static final double SIMILARITY_THRESHOLD 0.95; public SemanticInferenceCache() { this.cache Caffeine.newBuilder() .maximumSize(10_000) .expireAfterWrite(Duration.ofMinutes(30)) .recordStats() .build(); } public OptionalInferenceResult get(String input, EmbeddingModel model) { float[] inputVector model.embed(input); for (CachedResult cached : cache.asMap().values()) { double similarity cosineSimilarity(inputVector, cached.getEmbedding()); if (similarity SIMILARITY_THRESHOLD) { return Optional.of(cached.getResult()); } } return Optional.empty(); } private double cosineSimilarity(float[] a, float[] b) { double dotProduct 0.0; double normA 0.0; double normB 0.0; for (int i 0; i a.length; i) { dotProduct a[i] * b[i]; normA a[i] * a[i]; normB b[i] * b[i]; } if (normA 0 || normB 0) { return 0.0; } return dotProduct / (Math.sqrt(normA) * Math.sqrt(normB)); } }五、总结多模型推理 Pipeline 的性能优化需要从三个维度展开通过动态批处理提升单模型推理的 GPU 利用率通过 DAG 拓扑编排实现模型间的并行执行通过语义缓存减少重复计算。在实际落地时建议先建立各模型的基准延迟数据然后通过 DAG 拓扑分析识别关键路径。关键路径上的优化对整体延迟的改善最为明显非关键路径上的优化则可以着重于资源利用率的提升。Pipeline 设计的反模式——长链路串行多模型 Pipeline 最大的设计反模式是将逻辑上独立的模型放在同一串行链路中。例如在内容审核场景中文本审核、图像审核、敏感词过滤之间没有数据依赖但某些实现因为代码方便将它们串行调用——总延迟 延迟1 延迟2 延迟3。修复后发现并行执行可将 P95 延迟从 680ms 降到 260ms。识别这类反模式的方法是(1) 给每个模型的调加上 Span用于分布式追踪在 Jaeger/Grafana 中可视化 Pipeline 拓扑(2) 分析每个 Span 之间的数据流依赖确认是否存在假性依赖A 的输入包含了 B 的输出但实际上 B 没有传递新数据(3) 将确认无依赖的模型节点从串行链路中移除纳入并行执行组。GPU 共享与显存管理的工程策略多模型并行执行最难处理的工程问题是 GPU 显存争用。当 PipelineOrchestrator 中的并行组同时将 3 个模型加载到 GPU 显存时总显存占用 Model1 Model2 Model3 KV Cache按并发请求数增长。例如3 个 7B 模型每个 14GB 10 个并发请求的 KV Cache约 2GB/个 62GB远超 A100 的 40GB 显存。解决方案是模型卸载Model Offloading采用 LRU 策略当 GPU 显存不足时将最久未使用的模型从 GPU 卸载到 CPU 内存释放显存给当前执行中的模型。卸载延迟约 2-5 秒PCIe 拷贝模型权重可以通过预测性预加载缓解——分析历史请求序列在模型即将被调用前预加载到 GPU。Reactor Core 的调度能力可将模型卸载/加载与业务请求解耦通过Schedulers.boundedElastic()执行模型 IO 操作避免阻塞推理主线程。在多实例部署场景中我们通过将不同模型组分配给不同 GPU 节点模型亲和性调度进一步减少卸载频率——高频使用的模型组合固定在同一 GPU 上低频模型按需切换。这种模型亲和性调度配合 GPU 节点标签可将卸载事件减少 70%显著降低因模型切换导致的请求延迟毛刺。