LingBot-VA 2.0 推理速度慢，通常先要在数据加载、模型推理、后处理三个环节加上时间戳，量化耗时占比。用日志和统计找出峰值所在，再决定用并行加载、批处理还是缓存。以下方案基于通用推理链路，具体接口可能需按实际环境调整。

先给推理流程加时间戳并运行多次记录耗时，若某阶段耗时稳定占比高，则针对该阶段优化；若波动大，先查资源竞争或I/O。优先用批处理合并小请求，用缓存消去重复计算。优化后必须重新测试，并对比输出一致性，避免只快但不对。

在关键阶段添加时间戳

没有时间戳，只能凭感觉猜测瓶颈。先在LingBot-VA 2.0的推理入口和内部关键函数上埋点。以下装饰器可以复用到任意函数。

import time
import logging

def timed(func):
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        cost = time.perf_counter() - start
        logging.info(f"[{func.__name__}] cost {cost:.4f}s")
        return result
    return wrapper

然后在调用链上手动记录。注意，装饰器只对单个函数生效，如果推理主函数包含多个步骤，需要分别在各步骤函数上加装饰器。

@timed
def load_data(inputs): ...
@timed
def model_infer(features): ...
@timed
def post_process(raw): ...

def infer(inputs):
    data = load_data(inputs)
    raw = model_infer(data)
    return post_process(raw)

如果不想改动原函数，也可以把整个推理体包在一个计时上下文里，用变量分别记录各段耗时。

多次运行并统计耗时分布

单次运行不足以判断，因为资源调度和缓存状态会干扰。建议连续运行10次以上，收集各阶段耗时。可以先把耗时追加到列表，再统计。

import numpy as np

times = {k: [] for k in ["load_data", "model_infer", "post_process"]}
for _ in range(10):
    # 每次推理后把各阶段耗时写进相应列表
    times["load_data"].append(...)
    times["model_infer"].append(...)
    times["post_process"].append(...)

for k, v in times.items():
    arr = np.array(v)
    print(f"{k}: mean={arr.mean():.4f}s p95={np.percentile(arr, 95):.4f}s")

将每次原始日志同时保存到文件，便于复盘。如果某阶段P95远高于平均值，说明有间歇性慢请求，可能是锁竞争或I/O抖动。此时优化该阶段的低效路径比单纯加缓存更有效。

根据瓶颈选择优化方案

从统计结果看瓶颈。通常，如果数据加载耗时占比高，先做并行加载或预加载；如果是模型计算耗时高，批处理合并请求能提高吞吐；如果是后处理高，看看是否有重复计算可以加缓存。

• 数据加载耗时高：用多线程/异步预取，或把数据缓存到内存。适用条件：数据读取频繁，且数据尺寸不大。
• 模型推理耗时高：引入批处理队列，把多个请求合成一个batch。适用条件：模型支持动态batch，或输入形状可对齐。
• 后处理耗时高：对相同输入输出建立缓存，或优化后处理逻辑。适用条件：存在重复输入，或后处理中有大量重复计算。

如果多个阶段耗时都高，先优化占比最大的阶段，避免一次改动多个地方导致难以定位收益来源。

实现批处理队列或输出缓存

这里给出一个队列批处理代码骨架，假设模型推理函数支持批量输入。

import queue
import threading

class BatchInfer:
    def __init__(self, model, batch_size=8, timeout=0.1):
        self.model = model
        self.batch_size = batch_size
        self.timeout = timeout
        self.q = queue.Queue()
        self.thread = threading.Thread(target=self._run, daemon=True)
        self.thread.start()

    def predict(self, data):
        result_q = queue.Queue()
        self.q.put((data, result_q))
        return result_q.get()

    def _run(self):
        while True:
            batch = [self.q.get()]
            try:
                while len(batch) < self.batch_size:
                    item = self.q.get(timeout=self.timeout)
                    batch.append(item)
            except queue.Empty:
                pass
            datas = [x[0] for x in batch]
            outputs = self.model.infer(datas)
            for (_, rq), out in zip(batch, outputs):
                rq.put(out)

原调用中把 model.infer(data) 替换成 batch_infer.predict(data) 即可。注意，如果模型不直接支持批量，需要在内部循环处理或者选择缓存方案。

输出缓存装饰器示例：

import functools
import hashlib

def cache_output(func):
    _cache = {}
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        key = hashlib.md5(repr((args, kwargs)).encode()).hexdigest()
        if key in _cache:
            return _cache[key]
        out = func(*args, **kwargs)
        _cache[key] = out
        return out
    return wrapper

把缓存装饰器加到后处理函数上，但要注意输入必须可哈希序列化，且缓存不会导致内存无限增长。如果请求参数复杂，可以手动构造更稳定的key。

重新测试并对比耗时

优化后，用同样的测试脚本重新运行10次以上，记录各阶段耗时。下表是模板，把实际数值填上。

同时，把优化前后相同输入的结果保存下来，用文件对比或直接打印比较，确认输出一致。如果输出有差异，检查批处理形状对齐或缓存键冲突。只有耗时下降且输出正确，优化才算有效。