AI模型互操作性：格式不够用了之后

模型互操作性格式转换协议标准实践经验

作者: 醉月思发布日期: 2023-04-24 分类: 技术实践

用户的需求其实很直接：

用 PyTorch 训练的 BERT 模型需要转成 ONNX 格式
ONNX 模型要在 TensorRT 上运行，获得更好的性能
最终对外提供的 API 要兼容 OpenAI 协议，方便现有工具调用

听着挺合理，但每一步都有坑。

当你想把一个训练好的模型从一个平台搬到另一个平台时，最头疼的往往

上个月遇到一个挺典型的场景：用户在 PyTorch 上训练了一个 BERT 模型，要在不同的推理平台上部署，涉及到 ONNX 转换、TensorRT 优化、最后还要对外提供 OpenAI 兼容的 API。

问题场景

用户的需求其实很直接：

用 PyTorch 训练的 BERT 模型需要转成 ONNX 格式
ONNX 模型要在 TensorRT 上运行，获得更好的性能
最终对外提供的 API 要兼容 OpenAI 协议，方便现有工具调用

听着挺合理，但每一步都有坑。

第一步：PyTorch 到 ONNX

这个看似简单的转换，实际操作中遇到的第一坑就是动态 batch size。

原本的训练脚本中，模型接收的是固定 batch size 的输入：

# 原始推理代码
def forward(self, input_ids, attention_mask):
    return self.bert(input_ids, attention_mask)

# 调用时
outputs = model(input_ids, attention_mask)

直接转换成 ONNX 会报错，因为 ONNX 需要明确的输入维度。需要改成动态维度：

# 修改后的转换代码
import torch
from transformers import BertModel

model = BertModel.from_pretrained('bert-base-uncased')
model.eval()

# 准备示例输入
dummy_input_ids = torch.randint(0, 1000, (1, 128))
dummy_attention_mask = torch.ones((1, 128))

# 定义动态维度
dynamic_axes = {
    'input_ids': {0: 'batch_size', 1: 'sequence_length'},
    'attention_mask': {0: 'batch_size', 1: 'sequence_length'},
    'output': {0: 'batch_size', 1: 'sequence_length'}
}

# 导出 ONNX 模型
torch.onnx.export(
    model,
    (dummy_input_ids, dummy_attention_mask),
    'bert.onnx',
    input_names=['input_ids', 'attention_mask'],
    output_names=['output'],
    dynamic_axes=dynamic_axes,
    opset_version=14  # 注意版本选择
)

print("ONNX 模型已导出")

这里需要注意 opset_version 的选择。选太老的话，某些操作符不支持；选太新的话，目标推理框架可能还没跟上。踩过几次坑后，现在通常选 opset 14 或 15，算是比较稳定的选择。

第二步：ONNX 到 TensorRT

拿到 ONNX 模型后，开始转换到 TensorRT。这步遇到的坑更多。

import tensorrt as trt
import pycuda.driver as cuda
import pycuda.autoinit

def build_engine(onnx_file_path):
    TRT_LOGGER = trt.Logger(trt.Logger.WARNING)
    
    builder = trt.Builder(TRT_LOGGER)
    network = builder.create_network(1 << int(trt.NetworkDefinitionCreationFlag.EXPLICIT_BATCH))
    parser = trt.OnnxParser(network, TRT_LOGGER)
    
    # 解析 ONNX 模型
    with open(onnx_file_path, 'rb') as model:
        parser.parse(model.read())
    
    # 构建配置
    config = builder.create_builder_config()
    config.max_workspace_size = 1 << 30  # 1GB
    config.set_flag(trt.BuilderFlag.FP16)  # 启用 FP16 加速
    
    # 构建引擎
    engine = builder.build_engine(network, config)
    return engine

engine = build_engine('bert.onnx')

这个过程中遇到的主要问题有几个：

算子支持问题

有些 ONNX 算子 TensorRT 不支持，或者支持但性能不好。比如某些版本的 GroupNorm、InstanceNorm 等，转换时会报错。解决办法通常是：

调整模型结构，换成 TensorRT 支持更好的算子
或者回退到 ONNX Runtime 上运行
或者手动实现不支持的算子

FP16 精度问题

启用 FP16 可以加速推理，但有些模型对精度敏感，输出结果会明显变差。实测中发现：

# 对比 FP32 和 FP16 的输出差异
config.set_flag(trt.BuilderFlag.FP16)  # FP16
# vs
config.clear_flag(trt.BuilderFlag.FP16)  # FP32

对于 NLP 模型，通常差异不大；但对于一些数值敏感的模型，可能需要测试后再决定。

动态 shape 处理

TensorRT 对动态 shape 的支持有限，需要明确指定 profile：

profile = builder.create_optimization_profile()

# 设置最小、最优、最大尺寸
profile.set_shape("input_ids", 
                  (1, 1),          # 最小
                  (1, 128),        # 最优  
                  (8, 512))        # 最大

profile.set_shape("attention_mask",
                  (1, 1),
                  (1, 128),
                  (8, 512))

config.add_optimization_profile(profile)

不设置的话，模型只能按固定尺寸运行，灵活性很差。

第三步：OpenAI 协议适配

模型跑起来后，还有最后一道坎：对外提供 API 时要兼容 OpenAI 协议。

这听起来是个面子工程，但实际上挺重要。很多工具、库都默认支持 OpenAI 协议，如果自己的 API 能兼容，就能直接用这些现成的东西。

最简单的实现方式是写一个适配层：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
import uuid

app = FastAPI()

class ChatMessage(BaseModel):
    role: str
    content: str

class ChatCompletionRequest(BaseModel):
    model: str
    messages: List[ChatMessage]
    temperature: Optional[float] = 1.0
    max_tokens: Optional[int] = 100

class ChatCompletionResponse(BaseModel):
    id: str
    object: str = "chat.completion"
    created: int
    model: str
    choices: List[dict]

def run_inference(input_ids, attention_mask):
    # 这里调用 TensorRT 引擎进行推理
    # 实际实现中需要处理 CUDA 上下文、内存管理等
    outputs = engine.infer(input_ids, attention_mask)
    return outputs

@app.post("/v1/chat/completions")
async def chat_completions(request: ChatCompletionRequest):
    try:
        # 将聊天消息转换为模型输入
        # 这里简化处理，实际需要实现 tokenizer
        input_ids = encode_messages(request.messages)
        attention_mask = create_attention_mask(input_ids)
        
        # 运行推理
        outputs = run_inference(input_ids, attention_mask)
        
        # 将输出转换为 OpenAI 格式
        response = ChatCompletionResponse(
            id=str(uuid.uuid4()),
            created=int(time.time()),
            model=request.model,
            choices=[{
                "index": 0,
                "message": {
                    "role": "assistant",
                    "content": decode_outputs(outputs)
                },
                "finish_reason": "stop"
            }]
        )
        
        return response
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

这个适配层的核心是把内部的调用方式映射到 OpenAI 的协议上。看起来简单，但有几个细节要注意：

状态管理

OpenAI 的 chat completions 是无状态的，但很多模型需要上下文。如果需要对话功能，需要在应用层维护会话状态：

# 简单的会话管理
sessions = {}

@app.post("/v1/chat/completions")
async def chat_completions(request: ChatCompletionRequest):
    session_id = request.messages[0].content  # 从消息中提取会话ID
    
    if session_id not in sessions:
        sessions[session_id] = []
    
    # 更新对话历史
    sessions[session_id].extend(request.messages)
    
    # 使用历史上下文进行推理
    inputs = prepare_inputs(sessions[session_id])
    outputs = run_inference(inputs)
    
    # 返回响应
    return format_response(outputs)

流式输出

OpenAI 协议支持流式输出，这对用户体验很重要：

from fastapi.responses import StreamingResponse

@app.post("/v1/chat/completions")
async def chat_completions(request: ChatCompletionRequest):
    async def generate():
        inputs = prepare_inputs(request.messages)
        
        # 模拟流式输出
        for chunk in stream_inference(inputs):
            data = {
                "id": str(uuid.uuid4()),
                "object": "chat.completion.chunk",
                "created": int(time.time()),
                "model": request.model,
                "choices": [{
                    "index": 0,
                    "delta": {"content": chunk},
                    "finish_reason": None
                }]
            }
            yield f"data: {json.dumps(data)}\n\n"
        
        # 发送结束标记
        yield "data: [DONE]\n\n"
    
    return StreamingResponse(generate(), media_type="text/event-stream")

流式输出在用户体验上确实有明显提升，尤其是对于生成较长的文本时。

一些踩坑经验

整个过程中，踩过不少坑，总结几个比较典型的：

1. 版本兼容性

不同工具之间的版本依赖关系很复杂。比如 ONNX Runtime 和 TensorRT 对 CUDA 版本的要求不同，容易出现环境冲突。

解决办法通常是使用 Docker 容器隔离环境，每个框架用专门的容器。

2. 内存管理

TensorRT 的内存管理比较麻烦，尤其是多并发场景下。需要合理管理 GPU 内存，避免内存泄漏：

class TensorRTInference:
    def __init__(self, engine_path):
        self.engine = load_engine(engine_path)
        self.context = self.engine.create_execution_context()
        
        # 预分配 GPU 内存
        self.bindings = []
        for binding in self.engine:
            size = trt.volume(self.engine.get_binding_shape(binding))
            dtype = trt.nptype(self.engine.get_binding_dtype(binding))
            host_mem = cuda.pagelocked_empty(size, dtype)
            device_mem = cuda.mem_alloc(host_mem.nbytes)
            self.bindings.append({
                'host': host_mem,
                'device': device_mem
            })
    
    def infer(self, inputs):
        # 将输入数据拷贝到 GPU
        for i, input_data in enumerate(inputs):
            np.copyto(self.bindings[i]['host'], input_data.ravel())
            cuda.memcpy_htod(self.bindings[i]['device'], self.bindings[i]['host'])
        
        # 执行推理
        self.context.execute_v2([int(b['device']) for b in self.bindings])
        
        # 将输出数据拷回 CPU
        outputs = []
        for i in range(len(inputs), len(self.bindings)):
            cuda.memcpy_dtoh(self.bindings[i]['host'], self.bindings[i]['device'])
            outputs.append(self.bindings[i]['host'].reshape(self.engine.get_binding_shape(i)))
        
        return outputs

3. 错误处理

模型推理过程中各种错误都可能发生：输入格式不对、模型加载失败、内存不足等。需要做好错误处理，避免服务崩溃：

@app.post("/v1/chat/completions")
async def chat_completions(request: ChatCompletionRequest):
    try:
        # 参数验证
        if len(request.messages) == 0:
            raise HTTPException(status_code=400, detail="至少需要一条消息")
        
        if request.max_tokens and request.max_tokens > 4096:
            raise HTTPException(status_code=400, detail="max_tokens 不能超过 4096")
        
        # 推理
        outputs = run_inference(request)
        
        # 返回结果
        return format_response(outputs)
        
    except cuda.Error as e:
        raise HTTPException(status_code=500, detail="GPU 内存不足")
    except RuntimeError as e:
        raise HTTPException(status_code=500, detail="模型推理错误")
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"未知错误: {str(e)}")

性能对比

折腾完整个流程后，简单测了一下性能。在相同的硬件上（NVIDIA T4 GPU）：

平台	推理延迟 (ms)	吞吐量 (QPS)
PyTorch	45	22
ONNX Runtime	38	26
TensorRT (FP32)	32	31
TensorRT (FP16)	18	55

TensorRT FP16 的性能提升确实明显，但也需要根据实际场景权衡精度和速度。

小结

模型互操作性这个话题，说起来挺宏大，实际操作中就是解决一个又一个具体的兼容性问题。

从这次实践来看，几个关键点：

格式转换要谨慎：ONNX 作为中间格式不错，但不是万能的，某些操作可能不被目标平台支持
性能优化要有度：FP16、INT8 能提升性能，但会影响精度，需要测试验证
协议适配有价值：OpenAI 协议虽然简单，但能带来生态上的便利，值得投入
工具链要稳定：版本依赖、环境兼容这些问题很烦人，但不得不面对

技术上没有银弹，每个方案都有适用场景和边界。知道什么时候用什么工具，比掌握单一工具更重要。

最后，模型互操作性背后其实是生态系统的博弈。今天支持这个协议，明天适配那个格式，本质上是为了在不同生态之间找到自己的位置。

折腾到现在，算是明白了一件事：技术选择从来不是单纯的工程问题，还涉及到生态、标准、话语权这些更复杂的因素。我们能做的，就是在这些约束条件下，找到最适合自己的方案。

版权声明: 本文首发于指尖魔法屋-AI模型互操作性：格式不够用了之后（https://blog.thinkmoon.cn/post/211-ai-model-interoperability-practice/）转载或引用必须申明原指尖魔法屋来源及源地址！

利用 Cursor ACP 模式搭建飞书助手机器人，无需任何大模型接口

用 lark-cli 收飞书消息，Python 桥接 Cursor ACP（agent acp），项目代码里不集成任何大模型 API；适合已有 Cursor 订阅、想在本机跑 Agent 的场景。

📁 技术实践

📅 2026-07-07

用 GolemBot 把 Codex 接到飞书机器人

这篇文章记录一种少写代码的方案：用 GolemBot 接飞书机器人，用 Codex 做 Agent，再通过飞书 CLI Skills 读取文档、需求清单和 bug 清单。

📁 技术实践

📅 2026-06-09

调试 OpenAI Responses API 的 Web Search 踩坑记录

本文记录一次在投资研究 Agent 中调试 OpenAI Responses API Web Search 工具的过程，重点梳理请求格式、流式 SSE 解析、工具名兼容性和结构化数据源设计。

📁 技术实践

📅 2026-04-30

AI模型互操作性：格式不够用了之后

问题场景

第一步：PyTorch 到 ONNX

第二步：ONNX 到 TensorRT

算子支持问题

FP16 精度问题

动态 shape 处理

第三步：OpenAI 协议适配

状态管理

流式输出

一些踩坑经验

1. 版本兼容性

2. 内存管理

3. 错误处理

性能对比

小结

评论

相关阅读