多模型推理 Pipeline 的性能优化——批处理、模型并行与响应缓存

多模型推理 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 BlockingQueue<InferenceRequest> 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 { List<InferenceRequest> batch = collectBatch(); if (batch.isEmpty()) { continue; } long startTime = System.nanoTime(); List<InferenceResult> 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 List<InferenceRequest> collectBatch() throws InterruptedException { List<InferenceRequest> 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(List<InferenceRequest> requests, List<InferenceResult> 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 Map<String, ModelExecutor> executorMap; private final ExecutorService parallelExecutor; public ModelPipelineOrchestrator(Map<String, ModelExecutor> executorMap) { this.executorMap = executorMap; // 为什么使用 VirtualThread 而非固定线程池： // 模型调用主要是 IO 等待（GPU 推理），虚拟线程避免阻塞平台线程 this.parallelExecutor = Executors.newVirtualThreadPerTaskExecutor(); } /** * 根据 DAG 拓扑执行多模型推理 Pipeline。 * * @param pipelineDef Pipeline定义，描述模型节点及依赖关系 * @param input 原始输入 * @return 所有节点的推理结果 */ public Map<String, Object> execute(PipelineDefinition pipelineDef, Object input) { if (pipelineDef.getNodes().isEmpty()) { return Collections.emptyMap(); } // 拓扑排序，确定执行顺序 List<String> topologicalOrder = topologicalSort(pipelineDef); Map<String, 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)) { // 所有依赖已满足——并行执行 List<CompletableFuture<Void>> 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, Map<String, 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 List<CompletableFuture<Void>> collectParallelFutures( PipelineNode node, PipelineDefinition pipelineDef, Map<String, 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 List<String> topologicalSort(PipelineDefinition pipelineDef) { // 基于 Kahn 算法进行拓扑排序 Map<String, Integer> inDegree = new HashMap<>(); Map<String, List<String>> 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); } } Queue<String> queue = new LinkedList<>(); for (Map.Entry<String, Integer> entry : inDegree.entrySet()) { if (entry.getValue() == 0) { queue.offer(entry.getKey()); } } List<String> 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, Map<String, Object> context) { return node.getParallelGroup() != null && node.getDependencies().stream() .allMatch(context::containsKey); } private Object resolveInput(PipelineNode node, Map<String, Object> context) { // 解析节点输入：可能引用上游节点输出或原始输入 return context.get(node.getInputSource()); } }

四、响应缓存策略

对于相似度较高的输入，可以直接复用历史推理结果，减少 GPU 计算开销。

/** * 推理结果缓存——基于输入向量相似度的语义缓存。 * * 为什么用语义缓存而非精确匹配：两个语义相同但表述不同的请求 * （如"今天天气怎么样"与"今天天气如何"），在 LLM 推理中结果应当一致。 */ @Component public class SemanticInferenceCache { private final Cache<String, 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 Optional<InferenceResult> 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%，显著降低因模型切换导致的请求延迟毛刺。