导出器
将任意 PreTrainedModel 导出为 ONNX、ExecuTorch 或独立的 PyTorch 程序,无论目标运行时为何。
exporter = DynamoExporter() # or OnnxExporter, ExecutorchExporter
config = DynamoConfig(dynamic=True)
exported = exporter.export(model, inputs, config=config)导出器位于 Transformers 内部,而非下游库中,因此,只要架构变更、新的注意力模式或自定义缓存类型在建模代码中落地,导出时即可立即获得支持。
这些导出器目前处于实验阶段。本模块中的许多补丁用于规避特定的上游缺陷(Torch、ONNX Script、ONNX Runtime、ExecuTorch),一旦上游修复落地,这些补丁将被移除。在 API 稳定之前,请将补丁视为与测试套件中使用的版本绑定。在生产工具中固定这些版本,并预期随着上游变更的落地,会出现新的补丁,同时旧的补丁会消失。
| 导出器 | 输出 | 运行时 |
|---|---|---|
DynamoExporter | ExportedProgram | 任意 PyTorch 运行时,AOT 编译 |
OnnxExporter | ONNXProgram | 任何 ONNX 运行时(ORT、TensorRT、OpenVINO) |
ExecutorchExporter | ExecutorchProgramManager | 移动和边缘设备(ExecuTorch) |
OpenVINOExporter | openvino.Model | OpenVINO 运行时(Intel CPU/GPU/NPU) |
AutoHfExporter 根据配置选择正确的导出器,而 AutoExportConfig 则根据字典选择正确的配置类。两者都遵循 Transformers 中相同的自动类模式,这在后端是在运行时选择而非在调用点硬编码的情况下非常有用。
from transformers.exporters import AutoExportConfig, AutoHfExporter
export_config_dict = {"export_format": "onnx", "dynamic": True}
config = AutoExportConfig.from_dict(export_config_dict)
exporter = AutoHfExporter.from_config(config)
onnx_program = exporter.export(model, inputs, config=config)安装
安装您计划导出的后端所需的依赖项。
以下列出的版本是导出器测试套件所固定的版本。较新或较旧的版本通常也能正常工作,但导出器的补丁针对的是特定的 API 接口,因此在生产环境中应固定这些版本,并预期
HfExporter在检测到版本偏差时记录警告。
pip install transformers "torch==2.12.0"pip install transformers "torch==2.12.0" "onnx==1.21.0" "onnxscript==0.7.0" onnxruntimepip install transformers "torch==2.12.0" "executorch==1.3.1"在 Apple Silicon 上需要 macOS 14 或更高版本。MLX 已包含在 macOS ARM64 ExecuTorch 的 nightly 轮子中;它不是一个独立的包或额外组件。以下经过验证的 nightly 版本对已通过 MLX 导出器测试。请从标准的 ExecuTorch nightly 注册表安装:
pip install transformers
pip install \
"executorch==1.6.0.dev20260924" \
"torch==2.15.0.dev20260924" \
--extra-index-url https://download.pytorch.org/whl/nightly/cpu请显式安装 torch,因为 nightly 版本的 ExecuTorch 轮子未将其声明为依赖项。
有关从源码构建,请参阅 MLX 安装指南。
pip install transformers "torch==2.12.0" "openvino==2026.3.1"导出模型
所有导出器共享相同的接口。使用配置创建一个导出器,然后调用 export()。
通过替换导出器类来切换运行时。
from transformers import AutoModelForCausalLM, AutoTokenizer
from transformers.exporters import DynamoExporter, DynamoConfig
model = AutoModelForCausalLM.from_pretrained("Qwen/Qwen3-0.6B")
tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen3-0.6B")
inputs = tokenizer("Hello, world!", return_tensors="pt")
exporter = DynamoExporter()
config = DynamoConfig(dynamic=True)
exported = exporter.export(model, inputs, config=config)
# run the exported graph directly
outputs = exported.module()(**inputs)from transformers import AutoModelForCausalLM, AutoTokenizer
from transformers.exporters import OnnxExporter, OnnxConfig
model = AutoModelForCausalLM.from_pretrained("Qwen/Qwen3-0.6B")
tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen3-0.6B")
inputs = tokenizer("Hello, world!", return_tensors="pt")
exporter = OnnxExporter()
config = OnnxConfig(dynamic=True)
onnx_program = exporter.export(model, inputs, config=config)
# save and load with ONNX Runtime
onnx_program.save("model.onnx")
import onnxruntime as ort
session = ort.InferenceSession("model.onnx")
ort_inputs = {k: v.numpy() for k, v in inputs.items()}
outputs = session.run(None, ort_inputs)backend 用于选择目标:xnnpack(默认值)用于 CPU,mlx 用于 Apple Silicon GPU,或 cuda 用于支持 CUDA 的 GPU。请选择相应的安装选项。MLX 执行需要 MLX delegate 及其 Metal 库;
在没有 CUDA 支持的环境中请求 CUDA 会引发 RuntimeError。
from transformers import AutoModelForCausalLM, AutoTokenizer
from transformers.exporters import ExecutorchExporter, ExecutorchConfig
model = AutoModelForCausalLM.from_pretrained("Qwen/Qwen3-0.6B")
tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen3-0.6B")
inputs = tokenizer("Hello, world!", return_tensors="pt")
exporter = ExecutorchExporter()
config = ExecutorchConfig(backend="xnnpack", dynamic=True) # Use "mlx" on Apple Silicon.
et_program = exporter.export(model, inputs, config=config)
# save for on-device deployment
et_program.save("model.pte")
# load and run via the ExecuTorch Python runtime
from executorch.runtime import Runtime
program = Runtime.get().load_program("model.pte")
method = program.load_method("forward")
outputs = method.execute(list(inputs.values()))from transformers import AutoModelForCausalLM, AutoTokenizer
from transformers.exporters import OpenVINOExporter, OpenVINOConfig
model = AutoModelForCausalLM.from_pretrained("Qwen/Qwen3-0.6B")
tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen3-0.6B")
inputs = tokenizer("Hello, world!", return_tensors="pt")
exporter = OpenVINOExporter()
config = OpenVINOConfig(dynamic=True)
ov_model = exporter.export(model, inputs, config=config)
ov_model.save("model.xml")
# compile and run on CPU (or "GPU" / "NPU" if available)
import openvino as ov
compiled = ov.Core().compile_model(ov_model, "CPU")
ov_inputs = {k: v.numpy() for k, v in inputs.items()}
outputs = compiled(ov_inputs)动态形状
传入 dynamic=True 会将每个张量维度标记为动态,从而使导出的图在运行时能够接受任意大小的输入,而无需重新追踪。
如需对哪些维度是动态的进行细粒度控制,请改为传入显式的 dynamic_shapes,它会直接转发给 torch.export.export。
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
from transformers.exporters import DynamoExporter, DynamoConfig
model = AutoModelForCausalLM.from_pretrained("Qwen/Qwen3-0.6B")
tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen3-0.6B")
inputs = tokenizer(["Hello, world!", "Hi"], padding=True, return_tensors="pt")
batch = torch.export.Dim("batch", min=1, max=32)
seq = torch.export.Dim("seq", min=1, max=2048)
exporter = DynamoExporter()
config = DynamoConfig(
dynamic_shapes={"input_ids": {0: batch, 1: seq}, "attention_mask": {0: batch, 1: seq}},
# Emit data-dependent shape guards as runtime asserts instead of failing the export when a
# guard wouldn't hold across the explicit symbolic range. Most LLMs need this under fine-grained
# ``Dim(min=, max=)`` bounds. Not needed with ``dynamic=True`` / ``Dim.AUTO``, where torch.export
# infers shape relations instead of verifying them against user-stated bounds.
prefer_deferred_runtime_asserts_over_guards=True,
)
exported = exporter.export(model, inputs, config=config)import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
from transformers.exporters import OnnxExporter, OnnxConfig
model = AutoModelForCausalLM.from_pretrained("Qwen/Qwen3-0.6B")
tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen3-0.6B")
inputs = tokenizer(["Hello, world!", "Hi"], padding=True, return_tensors="pt")
batch = torch.export.Dim("batch", min=1, max=32)
seq = torch.export.Dim("seq", min=1, max=2048)
exporter = OnnxExporter()
config = OnnxConfig(
dynamic_shapes={"input_ids": {0: batch, 1: seq}, "attention_mask": {0: batch, 1: seq}},
# Emit data-dependent shape guards as runtime asserts instead of failing the export when a
# guard wouldn't hold across the explicit symbolic range. Most LLMs need this under fine-grained
# ``Dim(min=, max=)`` bounds. Not needed with ``dynamic=True`` / ``Dim.AUTO``, where torch.export
# infers shape relations instead of verifying them against user-stated bounds.
prefer_deferred_runtime_asserts_over_guards=True,
)
onnx_program = exporter.export(model, inputs, config=config)import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
from transformers.exporters import ExecutorchExporter, ExecutorchConfig
model = AutoModelForCausalLM.from_pretrained("Qwen/Qwen3-0.6B")
tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen3-0.6B")
inputs = tokenizer(["Hello, world!", "Hi"], padding=True, return_tensors="pt")
batch = torch.export.Dim("batch", min=1, max=32)
seq = torch.export.Dim("seq", min=1, max=2048)
exporter = ExecutorchExporter()
config = ExecutorchConfig(
backend="xnnpack", # Use "mlx" on Apple Silicon.
dynamic_shapes={"input_ids": {0: batch, 1: seq}, "attention_mask": {0: batch, 1: seq}},
# Emit data-dependent shape guards as runtime asserts instead of failing the export when a
# guard wouldn't hold across the explicit symbolic range. Most LLMs need this under fine-grained
# ``Dim(min=, max=)`` bounds. Not needed with ``dynamic=True`` / ``Dim.AUTO``, where torch.export
# infers shape relations instead of verifying them against user-stated bounds.
prefer_deferred_runtime_asserts_over_guards=True,
)
et_program = exporter.export(model, inputs, config=config)import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
from transformers.exporters import OpenVINOExporter, OpenVINOConfig
model = AutoModelForCausalLM.from_pretrained("Qwen/Qwen3-0.6B")
tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen3-0.6B")
inputs = tokenizer(["Hello, world!", "Hi"], padding=True, return_tensors="pt")
batch = torch.export.Dim("batch", min=1, max=32)
seq = torch.export.Dim("seq", min=1, max=2048)
exporter = OpenVINOExporter()
config = OpenVINOConfig(
dynamic_shapes={"input_ids": {0: batch, 1: seq}, "attention_mask": {0: batch, 1: seq}},
# Emit data-dependent shape guards as runtime asserts instead of failing the export when a
# guard wouldn't hold across the explicit symbolic range — most LLMs need this under fine-grained
# ``Dim(min=, max=)`` bounds. Not needed with ``dynamic=True`` / ``Dim.AUTO``, where torch.export
# infers shape relations instead of verifying them against user-stated bounds.
prefer_deferred_runtime_asserts_over_guards=True,
)
ov_model = exporter.export(model, inputs, config=config)生成式模型
对于自回归生成,模型的 forward 在预填充阶段(完整提示词,无 KV 缓存)和解码阶段(单个 token,已填充的 KV 缓存)具有不同的形状。导出器暴露了 ~HfExporter.export_for_generation,该方法会拆分这两个阶段并分别导出。
对于多模态生成模型,预填充(prefill)还会进一步拆分为图像或音频编码器、语言模型以及 lm_head。编码器和语言模型的发现使用 get_encoder()(modality="image" 或 "audio")和 get_decoder() 访问器,因此任何使用这些组件的新架构都可以开箱即用。
projector 组件仅在模型通过某个属性名(multi_modal_projector、connector、embed_vision、embed_audio)暴露该组件时才会出现。下方的 Qwen2-VL 将其 projector 整合进了视觉塔,因此其组件字典中没有单独的 multi_modal_projector 键。新的架构必须将其 projector 属性对齐到这些名称之一,而不是不断扩充该列表。
from transformers import AutoModelForImageTextToText, AutoProcessor
from transformers.exporters import DynamoExporter, DynamoConfig
model = AutoModelForImageTextToText.from_pretrained("Qwen/Qwen2-VL-2B-Instruct")
processor = AutoProcessor.from_pretrained("Qwen/Qwen2-VL-2B-Instruct")
messages = [{"role": "user", "content": [{"type": "image", "url": "https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/pipeline-cat-chonk.jpeg"}, {"type": "text", "text": "Describe this image."}]}]
text = processor.apply_chat_template(messages, add_generation_prompt=True, tokenize=False)
inputs = processor(text=text, images=messages[0]["content"][0]["url"], return_tensors="pt").to(model.device)
exporter = DynamoExporter()
config = DynamoConfig(dynamic=True)
components = exporter.export_for_generation(model, inputs, config=config)
# components = {"image_encoder": ExportedProgram, "language_model": ExportedProgram, "lm_head": ExportedProgram, "decode": ExportedProgram}from transformers import AutoModelForImageTextToText, AutoProcessor
from transformers.exporters import OnnxExporter, OnnxConfig
model = AutoModelForImageTextToText.from_pretrained("Qwen/Qwen2-VL-2B-Instruct")
processor = AutoProcessor.from_pretrained("Qwen/Qwen2-VL-2B-Instruct")
messages = [{"role": "user", "content": [{"type": "image", "url": "https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/pipeline-cat-chonk.jpeg"}, {"type": "text", "text": "Describe this image."}]}]
text = processor.apply_chat_template(messages, add_generation_prompt=True, tokenize=False)
inputs = processor(text=text, images=messages[0]["content"][0]["url"], return_tensors="pt").to(model.device)
exporter = OnnxExporter()
config = OnnxConfig(dynamic=True)
components = exporter.export_for_generation(model, inputs, config=config)
# components = {"image_encoder": ONNXProgram, "language_model": ONNXProgram, "lm_head": ONNXProgram, "decode": ONNXProgram}from transformers import AutoModelForImageTextToText, AutoProcessor
from transformers.exporters import ExecutorchExporter, ExecutorchConfig
model = AutoModelForImageTextToText.from_pretrained("Qwen/Qwen2-VL-2B-Instruct")
processor = AutoProcessor.from_pretrained("Qwen/Qwen2-VL-2B-Instruct")
messages = [{"role": "user", "content": [{"type": "image", "url": "https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/pipeline-cat-chonk.jpeg"}, {"type": "text", "text": "Describe this image."}]}]
text = processor.apply_chat_template(messages, add_generation_prompt=True, tokenize=False)
inputs = processor(text=text, images=messages[0]["content"][0]["url"], return_tensors="pt").to(model.device)
exporter = ExecutorchExporter()
config = ExecutorchConfig(backend="xnnpack", dynamic=True) # Use "mlx" on Apple Silicon.
components = exporter.export_for_generation(model, inputs, config=config)
# components = {"image_encoder": ExecutorchProgramManager, "language_model": ..., "lm_head": ..., "decode": ...}from transformers import AutoModelForImageTextToText, AutoProcessor
from transformers.exporters import OpenVINOExporter, OpenVINOConfig
model = AutoModelForImageTextToText.from_pretrained("Qwen/Qwen2-VL-2B-Instruct")
processor = AutoProcessor.from_pretrained("Qwen/Qwen2-VL-2B-Instruct")
messages = [{"role": "user", "content": [{"type": "image", "url": "https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/pipeline-cat-chonk.jpeg"}, {"type": "text", "text": "Describe this image."}]}]
text = processor.apply_chat_template(messages, add_generation_prompt=True, tokenize=False)
inputs = processor(text=text, images=messages[0]["content"][0]["url"], return_tensors="pt").to(model.device)
exporter = OpenVINOExporter()
config = OpenVINOConfig(dynamic=True)
components = exporter.export_for_generation(model, inputs, config=config)
# components = {"image_encoder": openvino.Model, "language_model": openvino.Model, "lm_head": openvino.Model, "decode": openvino.Model}导出的组件是独立的图,而非可直接运行的推理流水线。调用方负责运行每个编码器、投影嵌入以及编排生成循环。
export_for_generation 的工作原理
decompose_for_generation() 会执行一次 model.generate(**inputs, max_new_tokens=2),并挂钩 model.forward 以捕获真实的 prefill 和 decode kwargs(如果模型是多模态的,还会通过在每个编码器/投影器/语言模型上设置钩子来捕获各子模块的 kwargs)。正因如此,它适用于任何架构,包括仅解码器、SSM、编码器-解码器以及多模态模型,且无需针对每个模型编写额外的胶水代码。export_for_generation 只是基于它的一行封装。
捕获过程会以急切模式在 inputs 上运行模型,因此请传入较小但具有代表性的值,例如简短的提示词、单张小图片或几帧音频。导出的程序并不绑定于这些尺寸(动态形状仍然可以正常传递),但较小的捕获输入会使 decompose_for_generation 的开销更低,并保持符号形状推断的可处理性。
直接调用 decompose_for_generation 以在分解和导出之间执行操作,例如运行 eager 前向传播进行验证、替换子模块的输入,或跳过某个阶段。
from transformers.exporters.utils import decompose_for_generation
components = decompose_for_generation(model, inputs)
# {"image_encoder": (submodel, fwd_kwargs), "language_model": (...), ..., "decode": (...)}
exported = {}
for name, (submodel, subinputs) in components.items():
eager_outputs = submodel(**subinputs) # sanity-check the eager forward before exporting
exported[name] = exporter.export(submodel, subinputs, config=config)多令牌解码
默认情况下,decode 组件是一个 单 token 步骤——即针对 KV 缓存的一个查询 token——因此 torch.export 会将其查询序列轴特化为 1。传入 multi_token_decode=True 可将 decode 捕获为 多 token 解码:decompose_for_generation() 会将两个连续的解码步骤(以 max_new_tokens=3 进行捕获)合并为一次前向传播,从而使该轴保持符号化。这样,单个图即可服务于任意查询长度——单 token(常规解码)、一次处理多个 token(从过去继续,例如接受一批投机 token),以及在缓存为空时的普通 prefill。
from transformers.exporters import DynamoExporter, DynamoConfig
exporter = DynamoExporter()
config = DynamoConfig(dynamic=True)
components = exporter.export_for_generation(model, inputs, config=config, multi_token_decode=True)
# components["decode"] now accepts a variable number of query tokensfrom transformers.exporters import OnnxExporter, OnnxConfig
exporter = OnnxExporter()
config = OnnxConfig(dynamic=True)
components = exporter.export_for_generation(model, inputs, config=config, multi_token_decode=True)
# components["decode"] now accepts a variable number of query tokensfrom transformers.exporters import ExecutorchExporter, ExecutorchConfig
exporter = ExecutorchExporter()
config = ExecutorchConfig(backend="xnnpack", dynamic=True) # Use "mlx" on Apple Silicon.
components = exporter.export_for_generation(model, inputs, config=config, multi_token_decode=True)
# components["decode"] now accepts a variable number of query tokens查询轴仅在动态形状导出(dynamic=True)下保持符号化;静态导出会将其固定为捕获时的长度,从而生成一个固定的多 token 图。它与下文中的静态 KV 缓存相组合——合并后的解码过程会将每一步的 token 就地写入固定大小的缓存中,而缓存内部会处理这些 token 的落位。
静态 KV 缓存
本部分中的 ExecuTorch 示例(包括零拷贝更新和解码循环)仅支持 XNNPACK。MLX 导出器会拒绝
StaticCache;在 MLX 中请使用DynamicCache。
generate() 默认会构建一个 DynamicCache,并随着序列的延长而重新分配内存——这对于导出的计算图而言是一个不断变化的目标。静态缓存则是一个固定大小的缓冲区,仅分配一次,并在每一步中将数据就地写入当前位置。结合 多 token 解码,可以将生成过程收敛为单个导出的计算图:decode 图接收一个固定大小的缓存和 可变数量的查询 token,因此同一个图既能处理提示词(空缓存 → prefill),也能处理每个生成的 token(已填充的缓存 → decode)。通过传递一个包含 cache_implementation="static"(以及 max_cache_len)的 GenerationConfig,并设置 multi_token_decode=True,即可导出该图:
from transformers import GenerationConfig
from transformers.exporters import DynamoExporter, DynamoConfig
exporter = DynamoExporter()
gen_config = GenerationConfig(cache_implementation="static", max_cache_len=2048)
components = exporter.export_for_generation(
model, inputs, config=DynamoConfig(dynamic=True), generation_config=gen_config, multi_token_decode=True
)from transformers import GenerationConfig
from transformers.exporters import OnnxExporter, OnnxConfig
exporter = OnnxExporter()
gen_config = GenerationConfig(cache_implementation="static", max_cache_len=2048)
components = exporter.export_for_generation(
model, inputs, config=OnnxConfig(dynamic=True), generation_config=gen_config, multi_token_decode=True
)from transformers import GenerationConfig
from transformers.exporters import ExecutorchExporter, ExecutorchConfig
exporter = ExecutorchExporter()
gen_config = GenerationConfig(cache_implementation="static", max_cache_len=2048)
components = exporter.export_for_generation(
model, inputs, config=ExecutorchConfig(backend="xnnpack", dynamic=True), generation_config=gen_config, multi_token_decode=True
)decode 图现在包含两个符号轴——查询长度(即你输入的 token 数量)和缓存长度(max_cache_len,可在加载时调整)。dynamic=True 将这些轴(以及其他所有轴)标记为 Dim.AUTO,因此导出的图在加载时可接受任意提示长度和缓存大小。
零拷贝就地更新
静态缓存被传入并就地修改,因此一个缓冲区即可在解码步骤之间携带状态,无需主机端拷贝——前提是运行时绑定调用方的缓冲区,而不是通过其自身的内存区域进行拷贝。为此所需的工作是各后端中唯一剩余的部分:
Dynamo — 导出的程序将缓存写入建模为
USER_INPUT_MUTATION,因此调用components["decode"].module()(...)会直接更新你传入的缓存张量。每一步复用相同的张量即可,无需任何配置。ONNX Runtime — 解码图将缓存暴露为配对的
input.<name>/output.<name>。ORT 的CudaSession.set_buffer_sharing(onnxruntime.transformers.io_binding_helper)将每一对绑定到同一个设备缓冲区,因此缓存可以在循环中就地读取和更新,无需经过主机往返。ExecuTorch (XNNPACK) — 在
ExecutorchConfig中关闭内存规划分配,以便就地写入可以落在调用者自己的张量中(请参阅参考文档了解每个标志的作用):config = ExecutorchConfig( backend="xnnpack", dynamic=True, alloc_graph_input=False, alloc_graph_output=False, alloc_mutable_buffers=False, )零拷贝的原地写入同样要求调用方在运行时通过
Method::set_output_data_ptr绑定输出缓冲区——Python 运行时并未暴露此功能(executorch.runtime.Method仅暴露execute/set_inputs/get_outputs)。上述标志位完成了相关配置,但原地写入是一条 仅限 C++ 的路径(参见下文 ExecuTorch 解码循环示例)。在 Python 中,需在每一步从方法输出中读取更新后的缓存。
Decode-loop inference examples
该循环在所有后端上都具有相同的结构——整个过程中都是相同的图。从一个空的固定大小缓存开始,一次性输入整个提示词(空缓存 → prefill),然后逐个生成 token(已填充的缓存 → decode)。每次调用都会传入 input_ids、因果 attention_mask 以及 position_ids(每一步根据新增 token 的数量进行递增),外加缓存,并返回每个查询位置的 logits。每个 token 在缓存中的位置由静态缓存内部跟踪,因此无需在调用中额外传递任何信息。缓存的初始化方式因运行时而异(Dynamo 使用 StaticCache 对象,ONNX Runtime 使用原始设备缓冲区,ExecuTorch 在 C++ 中使用调用方数组),因此每个选项卡下方都会构建自己的实现。Dynamo 和 ONNX Runtime 选项卡会就地更新缓存;ExecuTorch 的就地更新路径是 C++ 实现的(如上所述,其 Python 运行时无法做到这一点)。
torch.export 将静态缓存的写入记录为 USER_INPUT_MUTATION,因此加载后的计算图的 module() 会 直接 更新你传入的 StaticCache —— 一个缓存即可在整个循环中承载状态,无需进行任何绑定或回传。通过 register_pytree_node(StaticCache),torch.export.load 可以反扁平化 StaticCache 输入。缓存必须 预先初始化(因为 torch.export 会将已分配的 K/V 固化到输入规范中,所以延迟初始化的空缓存无法匹配)—— 但保存的程序自带 example_inputs,因此可以复用该已初始化的 StaticCache 模板,并将其重置为空:
import copy
import torch
from transformers import StaticCache
from transformers.exporters.exporter_dynamo import register_pytree_node
register_pytree_node(StaticCache)
exported = torch.export.load("decode.pt2")
decode = exported.module() # runs on the device its inputs / cache live on (CUDA here)
# the artifact carries an initialized StaticCache template — reuse it (reset to empty)
_, example_kwargs = exported.example_inputs
past_key_values = copy.deepcopy(example_kwargs["past_key_values"])
past_key_values.reset()
def causal_mask(positions, cache_len): # [1, 1, len(positions), cache_len]
return (torch.arange(cache_len, device="cuda")[None, :] <= positions[:, None])[None, None]
# prefill: the whole prompt in one call
positions = torch.arange(prompt_len, device="cuda")
logits = decode(input_ids=prompt_ids, attention_mask=causal_mask(positions, max_cache_len),
position_ids=positions[None], past_key_values=past_key_values).logits
next_token = logits[:, -1:].argmax(-1)
# decode: query=1 buffers reused in place
input_ids = torch.empty((1, 1), dtype=torch.long, device="cuda")
position_ids = torch.empty((1, 1), dtype=torch.long, device="cuda")
attention_mask = torch.empty((1, 1, 1, max_cache_len), dtype=torch.bool, device="cuda")
slots = torch.arange(max_cache_len, device="cuda")
for position in range(prompt_len, max_cache_len):
input_ids.copy_(next_token)
position_ids.fill_(position)
attention_mask[0, 0, 0].copy_(slots <= position)
logits = decode(input_ids=input_ids, attention_mask=attention_mask,
position_ids=position_ids, past_key_values=past_key_values).logits
next_token = logits[:, -1:].argmax(-1)ONNX Runtime 会按原样执行计算图;原地缓存更新是通过 ORT 的 CudaSession(onnxruntime.transformers.io_binding_helper)完成的,它是 ORT io-binding 的一个轻量封装。set_buffer_sharing 将缓存的 input.<name> 及其对应的 output.<name> 绑定到同一个设备缓冲区,因此被修改的 K/V/计数器会直接写回输入;allocate_buffers 负责分配其余(非共享)的输出——在这里仅为 logits;而 infer(feed_dict) 则通过指针绑定你的 CUDA 张量并执行推理。缓存缓冲区直接来自图自身的输入元数据(get_inputs() 的形状和类型),因此无需模型配置——唯一的符号轴(缓存长度)变为 max_cache_len:
import torch
import onnxruntime as ort
from onnxruntime.transformers.io_binding_helper import CudaSession, TypeHelper
def causal_mask(positions, cache_len):
return (torch.arange(cache_len, device="cuda")[None, :] <= positions[:, None])[None, None]
session = ort.InferenceSession("decode.onnx", providers=["CUDAExecutionProvider"])
cuda = CudaSession(session, torch.device("cuda"))
# fresh device cache buffers built from each cache input's own shape/dtype; share each
# input.<name>/output.<name> pair on one buffer so the update lands in place
cache = {}
for i in session.get_inputs():
if not i.name.startswith("input."):
continue
name = i.name[len("input.") :]
dims = [max_cache_len if isinstance(d, str) and not d.isdigit() else int(d) for d in i.shape]
cache[name] = torch.zeros(dims, dtype=TypeHelper.ort_type_to_torch_type(i.type), device="cuda")
cuda.set_buffer_sharing(f"input.{name}", f"output.{name}")
cache_feed = {f"input.{name}": buf for name, buf in cache.items()}
vocab_size = next(o.shape[-1] for o in session.get_outputs() if o.name.endswith("logits"))
# prefill: the whole prompt in one call
positions = torch.arange(prompt_len, device="cuda")
cuda.allocate_buffers({"logits": (1, prompt_len, vocab_size)})
out = cuda.infer({"input_ids": prompt_ids, "attention_mask": causal_mask(positions, max_cache_len),
"position_ids": positions[None], **cache_feed})
next_token = out["logits"][:, -1:].argmax(-1)
# decode: query=1 buffers reused in place
cuda.allocate_buffers({"logits": (1, 1, vocab_size)})
input_ids = torch.empty((1, 1), dtype=torch.long, device="cuda")
position_ids = torch.empty((1, 1), dtype=torch.long, device="cuda")
attention_mask = torch.empty((1, 1, 1, max_cache_len), dtype=torch.bool, device="cuda")
slots = torch.arange(max_cache_len, device="cuda")
for position in range(prompt_len, max_cache_len):
input_ids.copy_(next_token)
position_ids.fill_(position)
attention_mask[0, 0, 0].copy_(slots <= position)
out = cuda.infer({"input_ids": input_ids, "attention_mask": attention_mask,
"position_ids": position_ids, **cache_feed})
next_token = out["logits"][:, -1:].argmax(-1)ExecuTorch 的端侧运行时基于 C++,而原地缓存更新依赖于 Method::set_output_data_ptr —— Python 运行时并未暴露此接口(executorch.runtime.Method 仅暴露 execute/set_inputs/get_outputs),因此零拷贝解码是仅限 C++ 的路径。将每个被修改的缓存 output 绑定到其对应的缓存 input 缓冲区,新的 K/V/计数器便会直接落入调用方的 StaticCache 缓冲区中,无需任何拷贝。与其他选项卡一样,形状和大小信息来自工件本身——此处即程序的 method_meta(input_tensor_meta/output_tensor_meta → TensorInfo::nbytes()),因为 C++ 端没有可供查询的 Python 运行时:
#include <executorch/extension/data_loader/file_data_loader.h>
#include <executorch/extension/tensor/tensor_ptr.h>
#include <executorch/runtime/executor/method.h>
#include <executorch/runtime/executor/program.h>
using namespace executorch::runtime;
using executorch::extension::FileDataLoader;
using executorch::extension::make_tensor_ptr;
// load the exported decode and its `forward` method (Result error-checks elided for brevity)
auto loader = FileDataLoader::from("decode.pte");
auto program = Program::load(&loader.get());
// method-execution memory: a fixed arena for bookkeeping + one buffer per the method's memory plan
std::array<uint8_t, 4 * 1024 * 1024> arena;
MemoryAllocator method_allocator(arena.size(), arena.data());
auto meta = program->method_meta("forward");
std::vector<std::vector<uint8_t>> planned(meta->num_memory_planned_buffers());
std::vector<Span<uint8_t>> planned_spans;
for (size_t i = 0; i < planned.size(); ++i) {
planned[i].resize(meta->memory_planned_buffer_size(i).get());
planned_spans.push_back({planned[i].data(), planned[i].size()});
}
HierarchicalAllocator planned_allocator({planned_spans.data(), planned_spans.size()});
MemoryManager memory_manager(&method_allocator, &planned_allocator);
auto decode = std::move(program->load_method("forward", &memory_manager).get());
// shapes and sizes come from method_meta — no external config. Inputs are [ids, mask, position_ids,
// cache×N]; outputs are [returned cache×N, logits, mutated cache×N].
const size_t num_cache_tensors = meta->num_inputs() - 3;
const size_t logits_out_idx = num_cache_tensors;
std::vector<size_t> cache_nbytes(num_cache_tensors), cache_out_idx(num_cache_tensors);
for (size_t i = 0; i < num_cache_tensors; ++i) {
cache_nbytes[i] = meta->input_tensor_meta(3 + i)->nbytes();
cache_out_idx[i] = num_cache_tensors + 1 + i; // mutated cache = the last N outputs
}
const size_t logits_nbytes = meta->output_tensor_meta(logits_out_idx)->nbytes();
// one set of StaticCache buffers (K/V + per-layer counters) is reused across steps, bound in place each call
auto forward = [&](const TensorPtr& input_ids, const TensorPtr& mask, const TensorPtr& position_ids) {
decode.set_input(EValue(*input_ids), 0);
decode.set_input(EValue(*mask), 1);
decode.set_input(EValue(*position_ids), 2);
for (int i = 0; i < num_cache_tensors; ++i)
decode.set_input(EValue(*cache_tensor[i]), 3 + i);
// bind each mutated-cache output onto that same input tensor's data → the write lands in place, zero copies
for (int i = 0; i < num_cache_tensors; ++i)
decode.set_output_data_ptr(cache_tensor[i]->mutable_data_ptr(), cache_nbytes[i], cache_out_idx[i]);
decode.set_output_data_ptr(logits_data, logits_nbytes, logits_out_idx);
decode.execute(); // cache updated in place; logits written to logits_data
return argmax_last(logits_data); // greedy pick
};
// prefill the whole prompt, then decode one token per step — the cache carries in place across all calls
int64_t next_token = forward(prompt_ids, prompt_mask, prompt_positions);
for (int64_t position = prompt_len; position < max_cache_len; ++position) {
next_token = forward(make_tensor_ptr({1, 1}, &next_token, ScalarType::Long),
causal_mask(position), // [1, 1, 1, max_cache_len] bool
make_tensor_ptr({1, 1}, &position, ScalarType::Long));
}量化
每个导出配置都接受一个 quantizer,用于在导出过程中对模型进行量化,量化对象可以是 torch.export FX 图(适用于所有后端的 PT2E 或 NNCF),也可以是转换后的 ONNX 或 OpenVINO 模型(适用于 ONNX Runtime 或 NNCF)。有关量化器、其支持的后端、校准方法以及如何添加量化器,请参阅 量化导出。
局限性与变通方案
torch.export、torch.onnx.export 和 ExecuTorch 在处理某些特定的 PyTorch 模式时都存在一些不足。导出器通过在导出流程中明确定义的节点应用一组可逆的补丁和 FX 级别的修复来规避这些问题。这些细节在公开的 export API 中是不可见的,但以下是需要了解的最常见事项:
- FlashAttention 和 FlexAttention 无法在任何后端上导出。
sdpa是首选设置,eager也可以使用(速度较慢)。如果模型当前使用的是其他设置,请在调用export之前将其设置为其中之一。 grouped_mm可以顺利被DynamoExporter追踪,并且会被OnnxExporter自动转换。 对于使用 XNNPACK 或 MLX 的ExecutorchExporter,导出器会在导出前将 MoE 专家替换为batched_mm。- 滑动窗口层(Mistral、Gemma 3)导出时带有完整的键值缓存;滑动掩码仍会应用窗口。