4、延迟优化（下）：流式输出优化、请求批处理（Dynamic Batching）、缓存策略（KV Cache、语义缓存）

好，咱们接着聊延迟优化。上一节我们把推理引擎、算子融合、量化这些硬核手段过了一遍。这一节，我打算聊聊更偏向「调度」和「复用」层面的优化——流式输出、请求批处理，还有缓存策略。

说实话，这三块在实际项目中，往往是见效最快、投入产出比最高的。你想想看，模型本身可能已经压到极限了，但如果你能让输出像流水一样流起来，让多个请求挤一辆车，让重复计算直接命中缓存……那延迟的下降，是肉眼可见的。

流式输出优化

先说说流式输出。很多 Agent 场景下，用户不希望等模型把整段话生成完才看到结果。比如你问一个复杂问题，模型要思考好几秒，如果等全部生成完再展示，用户体验就很差。

流式输出的核心思路，就是「边生成边推送」。模型每生成一个 token，就立刻通过 WebSocket 或 SSE（Server-Sent Events）推给前端。用户看到的是逐字出现的效果，心理等待时间会大幅缩短。

我在项目中遇到过一个问题：流式输出虽然降低了首 token 延迟，但整体吞吐反而下降了。为什么？因为每个 token 都要触发一次网络 IO，如果并发高，IO 开销就成了瓶颈。

我的建议：流式输出不要用 HTTP 长轮询，优先用 SSE。SSE 是单向通道，服务端主动推送，比 WebSocket 轻量。而且浏览器原生支持，不需要额外库。

另外，流式输出时要注意「攒批」策略。不是每个 token 都立刻推，而是攒 3-5 个 token 或者等 50ms 再推。这样能减少网络包数量，又不影响用户体验。嗯，这个阈值需要根据实际场景调，我一般从 50ms 开始试。

# 伪代码：流式输出攒批逻辑
async def stream_generate(prompt, batch_size=3, max_wait_ms=50):
    buffer = []
    async for token in model.generate_stream(prompt):
        buffer.append(token)
        if len(buffer) >= batch_size or time_elapsed >= max_wait_ms:
            yield buffer
            buffer = []
    if buffer:
        yield buffer

请求批处理（Dynamic Batching）

接下来是请求批处理。说白了，就是把多个用户的请求攒到一起，一次性喂给模型推理。GPU 的并行能力很强，处理一个 batch 的耗时，往往只比处理单个请求多一点点。所以批处理能大幅提升吞吐，降低平均延迟。

但这里有个坑：如果等太久才攒够 batch，单个请求的延迟反而会变高。所以需要「动态批处理」——设置一个最大等待时间，比如 200ms。时间到了，不管攒了多少请求，都立刻执行推理。

我曾经踩过的坑：一开始我把 batch 大小设得很大，想着能最大化吞吐。结果发现，有些请求的 token 长度差异太大，短的请求要等长的请求跑完才能返回。后来我改用「按 token 长度分组」的策略，把长度相近的请求放在同一个 batch 里，效果好了很多。

动态批处理的实现，一般需要一个调度器。它维护一个请求队列，定时触发推理。我习惯用 Python 的 asyncio 来实现，配合 asyncio.Queue 和 asyncio.Event，代码很简洁。

import asyncio

class DynamicBatcher:
    def __init__(self, model, max_batch_size=8, max_wait_ms=200):
        self.model = model
        self.max_batch_size = max_batch_size
        self.max_wait_ms = max_wait_ms
        self.queue = asyncio.Queue()
        self._running = True

    async def add_request(self, prompt):
        future = asyncio.Future()
        await self.queue.put((prompt, future))
        return await future

    async def _process_loop(self):
        while self._running:
            batch = []
            # 先取一个请求，然后等待更多请求或超时
            prompt, future = await self.queue.get()
            batch.append((prompt, future))
            deadline = asyncio.get_event_loop().time() + self.max_wait_ms / 1000
            while len(batch) < self.max_batch_size:
                timeout = deadline - asyncio.get_event_loop().time()
                if timeout <= 0:
                    break
                try:
                    prompt, future = await asyncio.wait_for(
                        self.queue.get(), timeout=timeout
                    )
                    batch.append((prompt, future))
                except asyncio.TimeoutError:
                    break
            # 执行推理
            prompts = [p for p, _ in batch]
            results = self.model.generate(prompts)
            for (_, future), result in zip(batch, results):
                future.set_result(result)

缓存策略：KV Cache 与语义缓存

最后聊聊缓存。这是延迟优化的「终极大招」——如果能把计算省掉，延迟自然就降下来了。

KV Cache

KV Cache 是 Transformer 推理中的标准优化。在自回归生成时，每生成一个新 token，都需要计算当前 token 与之前所有 token 的注意力。但之前的 Key 和 Value 其实已经算过了，没必要重复算。

KV Cache 就是把每一步的 Key 和 Value 缓存起来，下次生成时直接复用。这样，生成第 N 个 token 时，只需要计算第 N 个 token 的 Query，然后从缓存里取出之前的 Key 和 Value 做注意力计算。

你想想看，如果没有 KV Cache，生成 100 个 token 的复杂度是 O(N²)，有了之后变成 O(N)。这个优化，在长文本生成场景下，效果极其明显。

重要提示：KV Cache 虽然好，但会占用显存。每个 token 的 KV 向量大小是 2 × hidden_size × num_layers × precision。对于 7B 模型，一个 token 的 KV Cache 大约需要 1MB。生成 2048 个 token，就是 2GB 显存。所以，KV Cache 的显存管理也很关键。

我习惯的做法是：在推理框架层面，支持「预分配缓存」和「动态扩容」。预分配可以避免频繁的内存分配开销，动态扩容则能应对变长输入。另外，对于多轮对话场景，可以复用上一轮的 KV Cache，只计算新输入的部分。

语义缓存

语义缓存，是我个人觉得最有意思的一个优化点。它和传统的 key-value 缓存不同，不是精确匹配，而是「语义相似」匹配。

举个例子：用户问「今天天气怎么样？」，另一个用户问「今天天气如何？」。这两句话语义相同，但字面不同。如果用传统缓存，会 miss。但语义缓存会把它们映射到同一个向量空间，计算余弦相似度，如果相似度超过阈值，就直接返回缓存结果。

这个方案，在 Agent 场景下特别有用。因为很多用户问的问题其实是重复的，只是表达方式不同。比如「帮我查一下订单状态」和「我的订单到哪了」，本质上是同一个意图。

# 语义缓存实现思路
class SemanticCache:
    def __init__(self, embedding_model, similarity_threshold=0.92):
        self.embedding_model = embedding_model
        self.threshold = similarity_threshold
        self.cache = {}  # key: embedding, value: response

    def get(self, query):
        query_emb = self.embedding_model.encode(query)
        for cached_emb, response in self.cache.items():
            similarity = cosine_similarity(query_emb, cached_emb)
            if similarity >= self.threshold:
                return response
        return None

    def set(self, query, response):
        query_emb = self.embedding_model.encode(query)
        self.cache[query_emb] = response
我曾经踩过的坑：语义缓存的阈值设得太低，导致「帮我查天气」和「帮我查快递」被当成同一个问题，返回了错误结果。后来我把阈值调到 0.92，并且对敏感场景（如支付、订单）做了精确匹配兜底。记住，语义缓存适合「非关键」的重复问题，关键操作还是走精确匹配。

另外，语义缓存的更新策略也很重要。我建议用 LRU（最近最少使用）淘汰策略，同时给每个缓存项加一个 TTL（过期时间）。比如天气查询，缓存 10 分钟就够了；但「公司介绍」这种静态内容，可以缓存 24 小时。

小结

好了，这一节的内容就这些。流式输出、动态批处理、KV Cache、语义缓存——这四个手段，分别从「输出方式」、「请求调度」、「计算复用」、「结果复用」四个维度来降低延迟。

我个人习惯是：先上 KV Cache（几乎零成本，效果显著），再上动态批处理（需要调参，但收益大），然后根据场景决定是否用流式输出和语义缓存。嗯，优化没有银弹，关键是理解每个手段的适用场景和 trade-off。

最后，用一张图总结一下本章的知识体系：

延迟优化（下）知识体系 延迟优化策略 流式输出优化 请求批处理 KV Cache 语义缓存 SSE / WebSocket 攒批推送策略 首 token 延迟 最大等待时间 按长度分组 asyncio 调度 显存占用管理 预分配 + 动态扩容 多轮对话复用 向量相似度匹配 LRU + TTL 淘汰 精确匹配兜底 核心思路：减少计算量 + 提高复用率 + 优化调度 流式输出（输出方式）→ 动态批处理（请求调度）→ KV Cache（计算复用）→ 语义缓存（结果复用）

公众号:蓝海资料掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →