前阵子读 cannn-recipes-infer 源码，我习惯先把一条链路的入口文件铺在桌面上，再顺着日志顺序把类调用过一遍。写这系列笔记第三篇的时候，正好走到离线推理这条最核心的执行链路：OfflineInference - Scheduler - ExecutionEngine - ModelWorker。刚接触这套框架的读者，多半会有个困惑：一个离线推理脚本，看起来就是“读数据、进模型、出结果”三步，为什么内部要拆出这么多角色？搞懂这条链路之后，你会发现每一个参与者都有不可替代的位置，Scheduler 管“什么时候做、做哪批”，ExecutionEngine 管“怎么拆、交给谁”，ModelWorker 管“真刀真枪碰硬件”，而 OfflineInference 则是这一切的总开关。
这篇文章适合两类人：一是被业务需求推着要用昇腾卡做推理，想搞清楚 cannn-recipes-infer 到底怎么把请求送入模型的开发者；二是对服务化推理框架的调度与执行架构感兴趣，想拿一套实际工程实现做参照的同行。我会按执行顺序拆解每个角色的核心职责，把关键数据结构、接口设计、调度策略和调试方法都摊开来讲，尽量做到看完就能自己在源码里找到对应位置。
1. 整体设计：一条推理请求在框架里如何“走完全程”
1.1 什么是 OfflineInference，它和在线的区别在哪
在聊链路之前，得先明确 OfflineInference 的定位。它通常指“一次性喂一批输入，等待全部或部分结果返回”的推理模式，典型场景是数据清洗、批量打分、离线评测、离线增强，而不是对话服务那种“来一个问句快速回一个答案”的在线推理。在线推理对首 token 延迟敏感，对单请求响应时间有硬指标；离线推理则更关注吞吐，允许排队、允许动态组 batch，甚至允许把不同长度的序列拼在一起执行。
这个本质差异决定了代码结构。在线场景要专门处理持续连接、请求超时、会话状态，离线场景则可以把精力集中在“批量构造、调度、执行、收集结果”这条干净的主线上。cannn-recipes-infer 里 OfflineInference 的出现，本质上就是把这条主线做成一个可复用的编排器，把调度和执行细节下沉到下面的组件里。
1.2 四个核心类各自的职责边界
先看一张职责划分的速览表，后面再逐个展开：
| 组件 | 核心职责 | 关键产出 |
|---|---|---|
| OfflineInference | 入口编排、配置解析、请求初始化 | 初始化后的推理会话、执行结果 |
| Scheduler | 请求排队、批决策、显存/优先级管理 | 可执行的调度批次（ScheduleBatch） |
| ExecutionEngine | 调度批次到硬件任务的转换、资源协商 | 任务执行上下文、任务句柄 |
| ModelWorker | 加载模型、执行算子、维护运行时状态 | 推理结果、张量输出、日志回调 |
它们不是简单的“调用链”，而是围绕任务状态机互相协作。OfflineInference 更像“项目经理”，Scheduler 是“排产计划员”，ExecutionEngine 是“车间主任”，ModelWorker 是“操作机器的工人”。每个角色都有明确的产出物，下游只依赖上游的产出，不关心上游内部怎么实现。
1.3 为什么要拆成 Scheduler / ExecutionEngine / ModelWorker 三层
拆层不是增加复杂度，是为了让每个问题域都能独立演进。我自己的理解是：
- Scheduler 和服务态、批处理策略绑定，它会随负载特征变化不断调整，必须和硬件执行细节解耦。
- ExecutionEngine 要管理“调度出来的批次怎么落到具体的卡上”，它需要知道设备数量、通信拓扑、内存预算，但不需要关心算子怎么实现。
- ModelWorker 最贴近硬件，负责加载权重、运行模型、把张量结果搬回宿主内存，它只服务 ExecutionEngine 下达的命令。
这有点像后端开发里 controller -> service -> dao 的分层。不在 controller 里写 SQL，也不在 dao 里做权限校验，是因为边界清晰之后，每一层都能单独测试和替换。cannn-recipes-infer 真正工程的点在于，三层之间通过明确的数据对象通信，并没有出现层层透传大字典的坏味道。
2. OfflineInference 启动：入口初始化与任务分发
2.1 从执行脚本到 OfflineInference 构造
打开 cannn-recipes-infer 的仓库，常见的入口脚本大致是这样的形态：
def main(): config = load_config("configs/example.yaml") infer = OfflineInference(config) outputs = infer.infer(prompts, generation_config) save_outputs(outputs)看起来简单，但OfflineInference(config)这一步内部工作量很大。构造过程会做四件事：解析混合数据源配置、初始化日志和运行时环境、创建 ExecutionEngine、向底层注册默认的调度策略。如果配置里包含了模型并行参数，比如多卡场景下的 TP（Tensor Parallel）或 PP（Pipeline Parallel）配置，这一步还会做资源拓扑协商，确认当前机器有几张可用卡、显存多大、是否满足并发需求。
有一个容易忽略的细节：OfflineInference构造并不代表模型已经加载到显存。模型真正的加载发生在 ModelWorker 初始化阶段，而 ModelWorker 的启动又在 ExecutionEngine 的 lazy init 流程里。这样设计的考虑是，如果用户只做配置校验或者只构建任务图，就没必要白白占住显存。
2.2 配置解析与关键参数
离线推理和在线服务相比，配置文件里多了一批和批处理相关的参数。我整理过一组常用配置项，按作用分组：
| 参数分组 | 典型参数 | 作用说明 |
|---|---|---|
| 模型类 | model_path, tokenizer_path, dtype | 权重和分词器位置、精度 |
| 资源类 | device_ids, tensor_parallel_size, block_size | 选择卡、并行度、KV Cache 块大小 |
| 调度类 | max_num_batched_tokens, max_num_seqs, max_batch_size | 控制单轮调度能吃的请求量和 token 量 |
| 生成类 | max_new_tokens, temperature, top_p | 生成策略参数 |
| 运行时类 | log_level, cache_dir, stream_output | 日志、缓存、是否流式返回 |
max_num_batched_tokens这个值很关键，它决定一次调度最多往模型里塞多少个 token。过小会浪费多卡并行能力，过大会导致单次 step 耗时变长，甚至显存溢出。配参数的时候，不能只盯着最大并发数，要想清楚你手上的卡是 32G 还是 64G，模型是 7B 还是 13B。实测下来，7B 模型在 32G 单卡上，如果 max_num_batched_tokens 给到 8192，再用上 paged KV cache，是能稳定运行并保住吞吐的。
2.3 请求预处理与任务批次构建
拿到一列 prompts 之后，OfflineInference 不会立刻把所有数据原生塞给下游。它要做 tokenize、长度统计、按预分配策略分组。源码里通常能看到类似build_request_batch的过程，它会把每个 prompt 转换成“请求对象”，里面至少包含：
class InferRequest: request_id: str input_ids: List[int] sampling_params: dict arrival_time: float state: str # pending / running / finished带上arrival_time是很实用的设计，调度器可以据此实现先到先服务，或者结合优先级做老化处理，防止低优先级请求永远饿死。这个对象就是 Scheduler 眼里最小的任务单元。
预处理阶段还有一个值得注意的点：长文本截断和 batch 内长度对齐。离线推理场景里，prompt 长度方差往往很大，有的可能是几十 token，有的可能是上千 token。如果不做处理，短序列和长序列混在一个 batch 里执行，短序列会白白等待长序列跑完。处理方式有两种：一种是把超长样本放进单独的 batch；另一种是允许动态 batch，让短序列先执行，长序列慢慢跑完后再加入新请求。cannn-recipes-infer 的实现路径更接近后者，因为连续 batching 本身就是这类框架的主打能力。
2.4 触发执行：infer 方法的内部编排
用户调用infer()后，OfflineInference 首先把请求批次交给 Scheduler 注册，然后进入一个事件循环。这个循环的核心逻辑可以概括为：
def infer(self, prompts, generation_config): requests = [self._build_request(p, generation_config) for p in prompts] scheduler = self.engine.scheduler scheduler.add_requests(requests) while not scheduler.is_finished(): scheduler.schedule() engine.execute(scheduler.current_batch) engine.collect_results() return self._assemble_outputs()while not scheduler.is_finished()是离线推理的精髓。在线服务里，请求是源源不断进来的，循环永远不退出；离线推理则有一个明确的终止条件：所有请求都进入 finished 状态。每一轮循环里，Scheduler 决定当前该执行哪些请求，Engine 负责真正跑一版模型 forward，然后收集结果，如此反复直到全部完成。
我读源码时的一个体会是，这个循环看起来简单，写好不容易。关键是要处理好“跑完一轮后，哪些请求还需要继续生成下一个 token”这件事。不是所有请求一轮就结束，多数请求要迭代很多轮，每一轮都需要把新生成的 token 拼接回 KV cache 并继续调度。这套“迭代式生成 + 动态调度”的机制，正是离线推理框架吞吐高的基石。
3. Scheduler 调度核心：排队、合并与批次决策
3.1 调度器的内部状态与请求队列
Scheduler 内部通常维护两个主要集合：等待队列（waiting queue）和运行集合（running set）。等待队列里的请求还没有真正进入模型，运行集合里的请求正在被 batch 执行或者已经执行了一部分。
class Scheduler: def __init__(self, policy_type="fcfs"): self.waiting: List[InferRequest] = [] self.running: List[InferRequest] = [] self.finished: List[InferRequest] = [] self.policy = self._init_policy(policy_type)在进一步看 schedule 逻辑之前，要先理解“一个请求从 waiting 到 running 意味着什么”。进入 running 意味着这个请求已经拿到了第一轮推理所需的显存空间，包括 KV cache 里给它预留的块。显存不是无限的，所以 Scheduler 最大的约束不是 CPU 算力，而是“当前还能释放出多少显存块给新请求”。这也是为什么 Scheduler 要主动管理 KV cache 块，而不是简单地把请求丢给执行器。
3.2 连续 Batching：离线场景下的默认调度策略
早期离线推理框架常用静态 batching，也就是把所有请求预先分成固定大小的 batch，一个 batch 跑完再跑下一个。静态 batching 实现简单，但弊端很明显：一个 batch 里一旦有早完成生成的短请求，它占的显存和计算位置依然被锁住，直到整个 batch 都结束。
cannn-recipes-infer 主推的是连续 batching（continuous batching）思路：每一轮调度都重新扫描 waiting 和 running 集合，已经结束的请求立刻退出，把显存块释放给新请求；还在生成中的请求和新加入的请求按资源预算合并进本轮 batch。这样的效果就是，模型每执行一次 forward，batch 里的内容都可能发生变化，GPU 永远不会被“等队友”拖慢。
在源码里，这个逻辑通常表现为一个循环：
def schedule(self) -> Optional[ScheduleBatch]: self._preempt_overloaded_requests() batches = self._create_batches_by_policy() return batches_preempt_overloaded_requests是资源不足时的兜底策略。如果当前显存块不足，Scheduler 会选择挂起或抢占一些请求，把显存让给优先级更高或更接近完成的请求。这个处理方式让我想起操作系统里的页面置换，只不过这里的“页”变成了 KV cache 块。
3.3 调度决策的关键指标
调度器在决定“本轮要把哪些请求合并成一个 batch”时，不是简单地把 waiting 里的请求拉出来，而是要算一笔账。我摘录几个关键约束：
- 请求数上限：
max_num_seqs，防止 batch 里请求数过多导致图执行时间不可控。 - token 数上限：
max_num_batched_tokens，整批的输入 token 总数不能超过这个值。 - 显存块预算：当前空闲 KV cache 块能否覆盖新请求的
max_new_tokens需求。 - 优先级规则：默认先到先服务，也支持按优先级字段调整。
这些约束需要组合判断：即便请求数很少，如果每个请求都很长，token 总数也可能撞到上限。相反，如果请求全部很短，可能在请求数量上先撞限。写调度器的时候，最容易忽略的是显存块预算，因为它在单请求视角下看不出来，只有把模型参数、序列长度和可用显存放到一起才算得明白。
3.4 调度结果如何交给 ExecutionEngine
Scheduler 完成决策后，产出的是一个ScheduleBatch对象，里面不再直接放原始的字符串 prompt，而是放“准备好执行的 token 序列 + 对应的采样参数 + 显存块索引”。它相当于一份“施工图”：Engine 拿到之后，不需要关心为什么是这些序列在一起执行，只需要照着这份图纸把工作做掉。
我特别看重这个边界。如果 Scheduler 把原始请求直接传给 Executor，那么 Executor 里必然要再做一遍调度决策，职责就混乱了。现在这样设计，Scheduler 可以单独用一组纯 CPU 的单元测试覆盖各种队列场景，不需要拉起卡做验证；Engine 侧也能专注于执行和资源管理。调试的时候，只要确认ScheduleBatch内容符合预期，问题范围就能缩小很多。
4. ExecutionEngine：从调度计划到可执行任务
4.1 Engine 在整条链路中的坐标
ExecutionEngine 在 cannn-recipes-infer 里的定位，我总结成一句话：把“调度计划”翻译成“硬件能跑的任务”，并在任务执行过程中管理生命周期。它不是直接调用模型的模块，更接近“任务编排器”。
第一次看源码时，我被 ExecutionEngine 和 ModelWorker 的边界搞晕过。后来发现，简单场景下 Engine 可以直接驱动模型执行，但一旦涉及多卡、时序并行或者流水并行，Engine 就要跟多个 Worker 通信。所以源码里更常见的做法是 Engine 维护一组 Worker 句柄，按调度批次把子任务拆给不同 Worker，再汇总结果。
4.2 执行引擎的接口设计
从调用关系看，ExecutionEngine 至少暴露三个核心方法：
class ExecutionEngine: def __init__(self, config): self.device_ids = config.device_ids self.workers = self._init_workers(config) self.cache_blocks = self._init_cache_blocks(config) def execute_batch(self, batch: ScheduleBatch): tasks = self._split_batch_to_tasks(batch) futures = [worker.submit(task) for worker, task in zip(self.workers, tasks)] return futures def collect(self, futures): results = [] for f in futures: results.append(f.result()) return results_init_workers这一步值得展开。它根据传入的device_ids决定启动几个 Worker。比如你有 4 张卡，tensor_parallel_size=4，那通常会启动 4 个 Worker，每个 Worker 绑定一张卡；如果tensor_parallel_size=2，则可能启动 2 个 Worker，每个 Worker 管理两张卡的计算流程。初始化时如果显存不足或卡被占用，错误会在这里提前暴露，而不是等到跑 batch 时才报错，这一点对线上排障很友好。
4.3 与 ModelWorker 的通信：任务下发与回调
Engine 和 Worker 之间是典型的“生产-消费”模式。Engine 作为产出任务的消费者，把每个子任务封装成一个WorkerTask，里面包含 input_ids、block_tables、sampling_params 等。Worker 执行完任务后，返回WorkerOutput，包含生成的 token 序列、logprobs、耗时统计等。
如果只看代码，很容易把“Engine 调 Worker 方法”理解成同步调用。实际上在多卡场景下，任务下发通常走队列，Worker 之间还有可能做通信。一个有用的经验是，调试多卡任务时，不要只看主进程日志，最好把每个 Worker 的日志分开输出，否则你很难判断延迟到底卡在哪个设备上。
4.4 Cache 管理与显存块分配
ExecutionEngine 里还有一个常被忽视但很重要的模块：Cache Manager。它负责 KV cache 块的分发、回收、拷贝。Scheduler 做预算判断时，需要向 Engine 查询剩余块数；Engine 执行完一轮后，又会把不再使用的块标记为空闲。
KV cache 块的大小通常由block_size控制，常见值是 16 或 32。block_size 越大，管理开销越小，但内部碎片越多；block_size 越小，显存利用越灵活，但块表管理更复杂。这块调优对离线吞吐影响很大。我之前跑一批长文本评测时，把 block_size 从 32 改成 16，显著降低了显存碎片，批大小能往上抬，整体吞吐提升了将近一成。如果你的业务序列长度波动大，值得试试更小的块大小。
5. ModelWorker：真正碰硬件的那一环
5.1 Worker 初始化：加载模型、绑定设备、配置运行模式
ModelWorker 是整条链路里最重的一环。初始化时，它要做的事情包括：
class ModelWorker: def __init__(self, config, device_id): self.device_id = device_id self.model = self._load_model(config) self._set_device(device_id) self._prepare_cache_blocks(config)_load_model要用 CANN 环境下的模型加载接口，把权重从磁盘搬到设备内存中。这里有个容易忽略的问题：模型的权重复制不是一次写完就完了。分布式推理时，每个 Worker 只持有自己负责的权重分片；如果配置了开启图模式的算子融合，还会额外编译和缓存执行图。图模式的好处是运行时省去不少 Python 算子分发开销，代价是首次启动时间明显变长。你要是第一次跑脚本发现等了很久才有日志输出，多半是在编译融合图。
5.2 Worker 的执行循环
Worker 启动后通常进入一个事件循环，等待 Engine 下发任务。这个循环的设计直接决定了吞吐上限：
def worker_loop(self): while True: task = self.task_queue.get() if task is None: break output = self._execute_task(task) self.result_queue.put(output)_execute_task内部会完成：把 input_ids 搬到设备、创建图输入、执行模型 forward、采样生成下一个 token、更新 KV cache、把新 token 和设备上的采样结果搬回宿主内存。如果开启流式输出，这个函数还会通过回调把中间结果推给上层。
我第一次读到这里时，惊讶于“采样”居然也在 Worker 里做。后来想通了，如果采样放回 Python 高层做，每生成一个 token 就要设备到宿主拷贝一次，延迟和带宽都扛不住。在设备端直接完成采样，只把最终 token id 传回，是性能上的必然选择。
5.3 结果回传与流式输出
离线推理虽然不像在线服务那样每个 token 都要马上推给用户，但有些场景仍然需要流式回调。比如评测程序想要看每轮生成的进度，或者长文本生成任务希望实时落盘。Worker 通过回调接口上报中间状态：
def _execute_task(self, task): for step in range(task.max_new_tokens): logits = self.model.forward(input_ids, cache_blocks) next_token = self.sample(logits, task.sampling_params) if task.stream_callback: task.stream_callback(step, next_token) if next_token == EOS: breakstream_callback的设计让上层能拿到逐步生成状态，又不必每步都进 Python 主线程做同步，因为回调本身还是在 Worker 线程里执行的，避免了额外的进程切换开销。
5.4 单测与硬件隔离
我个人的经验是，ModelWorker 是整条链路里最难单独调试的模块，因为一跑就要占卡、占显存。为了快速验证 Worker 逻辑，可以在代码里加一个execution_mode开关，当它等于mock时，Worker 不真正调用模型，只返回基于规则生成的伪结果。这样既能验证调度链路，又避开了硬件依赖。
但要注意，mock 模式只能验证通信和状态机，不能验证算子融合和显存管理。真正上卡之前，专门预留一台干净的测试机器，跑一遍全链路，确认没有设备占用后才能放心做批量任务。
6. 串起来看：一次完整推理的时序与排障
6.1 关键源码位置速查
为了方便接着读源码，我把自己梳理的“模块与可能文件/类名”的对应关系列在下面。不同版本可能略有差异，但按职责去找一般都能定位到：
| 逻辑角色 | 常见命名线索 | 需要关注的重点函数/成员 |
|---|---|---|
| OfflineInference | offline_inference.py | infer,_build_request,_assemble_outputs |
| Scheduler | scheduler.py | schedule,add_request,_preempt |
| ExecutionEngine | execution_engine.py | execute_batch,_init_workers,collect |
| ModelWorker | model_worker.py | worker_loop,_execute_task,load_model |
| Cache Manager | cache_manager.py | allocate,free,block_tables |
拿到一份新代码时，我习惯先打开日志文件，按日志打印顺序反推调用链，再回到源码确认每个模块的边界。日志里如果出现schedule和execute_batch交替出现，基本说明进入了主循环。
6.2 调试手法：断点、日志与性能观察
调试这条链路，最实用的三个手段：
- 断点打在
Scheduler.schedule()返回前，观察ScheduleBatch里请求数和 token 数是否符合预期。 - 日志切到 debug 级别，能看到每次调度后的队列长度变化，这是判断调度策略是否生效最快的方式。
- 在 Engine 的
execute_batch前后各打一个时间戳，能一眼看出是“调度太慢”还是“执行太慢”。调度慢多半是 Python 层队列操作或显存计算的问题；执行慢则要往模型、设备通信方向排查。
还有一个压箱底的经验：大量使用device to host拷贝时，通过昇腾环境里的 profiling 工具看 Host 和 Device 之间是否有长时间同步等待。很多离线条慢的原因不是算力不够，而是 CPU 端数据准备好了但设备端还在忙，或者反过来设备端在干等 CPU 下发。用 profiling 数据对照调度时间戳，基本能立刻定位瓶颈。
6.3 常见问题与速查表
把我在实际跑批过程中遇到的高频问题整理成一张速查表，供你排查时直接对号入座：
| 现象 | 可能原因 | 建议处理 |
|---|---|---|
| 启动很慢，日志长时间无输出 | 图模式编译，或首次加载权重耗时 | 如果是首次编译，耐心等；反复出现则检查缓存目录 |
| 显存溢出 OOM | batch 参数配得过大或 KV cache 预算不足 | 调小 max_num_batched_tokens，或减小 block_size |
| 吞吐很低，但显存没满 | 调度批次过小，设备利用率不足 | 增大 max_num_seqs，检查是否有显存碎片 |
| 特定请求特别慢 | 长序列抢占短序列资源，或优先级策略不当 | 开启抢占/超时策略，把超长请求单独分桶 |
| 多任务并行时结果错乱 | Worker 之间共享了同一份显存块 | 检查块分配是否加锁，确认 device_id 绑定正确 |
| 流式回调收不到中间结果 | 回调在线程中异常退出 | 在 callback 里加 try/except，查看 Worker 日志 |
6.4 我的一些优化心得
这条链路跑顺之后，我最大的感受是优化调度参数比优化算子实现更划算。因为底层模型算子是 CANN 已经优化好的，你干预空间不大，但调度层的参数几乎是完全可以自己掌控的。
一个简单的调优套路是：先定显存上限，再定 block_size，然后逐步抬高 max_num_batched_tokens 和 max_num_seqs，观察吞吐曲线。注意吞吐不是单调上升的，batch 太大时单次 forward 耗时变长，整体吞吐反而下降。我实测中，曲线会有一个明显的平台期，取平台期的中间点作为最终配置，最稳妥。
另外一个让我受益很多的做法是，把每次离线推理的配置、请求数、序列长度分布、吞吐、耗时记成一张实验表。因为调度策略之间的差异，只有在大量样本上才能看出来，靠一两次直觉调整很容易误判。有了实验表，你就能用数据说服自己和同事，而不是靠“感觉这次更快”。
回到开头那个问题：为什么离线推理简单脚本内部要拆出这么多角色？现在我可以说，
