一、NLP 模型概览

1.1 模型类型

NLP 模型三大类:
  编码器 (Encoder)
    BERT, RoBERTa, ALBERT, DeBERTa
    输入: 文本 → 输出: 文本表示
    适用: 分类、NER、相似度计算

  解码器 (Decoder)
    GPT, LLaMA, Qwen, ChatGLM
    输入: 文本 → 输出: 生成文本
    适用: 对话、文本生成、翻译

  编码器-解码器 (Encoder-Decoder)
    T5, BART, mBART
    输入: 文本 → 输出: 生成文本
    适用: 摘要、翻译、问答

1.2 部署挑战

NLP 模型部署难点:
  1. 动态 Shape: 输入长度不固定
  2. Token 匹配: 分词器与模型耦合
  3. Attention Mask: 注意力掩码处理
  4. 自回归生成: 逐 token 生成
  5. 大模型显存: 参数量大，显存需求高

二、BERT 模型部署

2.1 导出 ONNX

from transformers import BertTokenizer, BertForSequenceClassification
import torch

# 加载模型
model = BertForSequenceClassification.from_pretrained(
    "bert-base-chinese",
    num_labels=2
)
model.eval()

tokenizer = BertTokenizer.from_pretrained("bert-base-chinese")

# 导出 ONNX
dummy_input = tokenizer(
    "测试文本",
    return_tensors="pt",
    padding="max_length",
    max_length=128,
    truncation=True
)

torch.onnx.export(
    model,
    (
        dummy_input["input_ids"],
        dummy_input["attention_mask"],
        dummy_input["token_type_ids"]
    ),
    "bert_classifier.onnx",
    input_names=["input_ids", "attention_mask", "token_type_ids"],
    output_names=["logits"],
    dynamic_axes={
        "input_ids": {0: "batch_size", 1: "seq_len"},
        "attention_mask": {0: "batch_size", 1: "seq_len"},
        "token_type_ids": {0: "batch_size", 1: "seq_len"},
        "logits": {0: "batch_size"}
    },
    opset_version=14
)

print("BERT ONNX 已导出")

2.2 ATC 转换

# 转换 BERT 分类模型
atc --model=bert_classifier.onnx \
    --framework=5 \
    --output=bert_classifier \
    --input_shape="input_ids:1,128;attention_mask:1,128;token_type_ids:1,128" \
    --soc_version=Ascend310 \
    --log=info

2.3 BERT 推理实现

import numpy as np

class BERTClassifier:
    def __init__(self, model_path, vocab_path):
        self.model = self._load_model(model_path)
        self.tokenizer = self._load_tokenizer(vocab_path)
        self.max_length = 128
    
    def _load_tokenizer(self, vocab_path):
        """加载分词器"""
        # 简化的分词器实现
        return {"vocab": self._load_vocab(vocab_path)}
    
    def _load_vocab(self, vocab_path):
        """加载词表"""
        vocab = {}
        with open(vocab_path, 'r', encoding='utf-8') as f:
            for idx, line in enumerate(f):
                vocab[line.strip()] = idx
        return vocab
    
    def tokenize(self, text):
        """分词"""
        tokens = list(text)
        
        # 截断
        tokens = tokens[:self.max_length - 2]
        
        # 添加特殊 token
        tokens = ["[CLS]"] + tokens + ["[SEP]"]
        
        # 转换为 ID
        input_ids = [self.tokenizer["vocab"].get(t, 0) for t in tokens]
        
        # 填充
        attention_mask = [1] * len(input_ids)
        token_type_ids = [0] * len(input_ids)
        
        padding_length = self.max_length - len(input_ids)
        input_ids += [0] * padding_length
        attention_mask += [0] * padding_length
        token_type_ids += [0] * padding_length
        
        return {
            "input_ids": np.array([input_ids], dtype=np.int64),
            "attention_mask": np.array([attention_mask], dtype=np.int64),
            "token_type_ids": np.array([token_type_ids], dtype=np.int64)
        }
    
    def predict(self, text):
        """推理"""
        inputs = self.tokenize(text)
        
        output = self._run_model(
            inputs["input_ids"],
            inputs["attention_mask"],
            inputs["token_type_ids"]
        )
        
        # Softmax
        probs = np.exp(output[0]) / np.sum(np.exp(output[0]))
        
        return probs

# 使用示例
classifier = BERTClassifier("bert_classifier.om", "vocab.txt")

text = "这是一个正面评价"
probs = classifier.predict(text)
print(f"正面概率: {probs[1]:.4f}")
print(f"负面概率: {probs[0]:.4f}")

三、GPT 模型部署

3.1 自回归生成

class GPTGenerator:
    def __init__(self, model_path, vocab_size=50257):
        self.model = self._load_model(model_path)
        self.vocab_size = vocab_size
        self.max_length = 512
    
    def generate(self, prompt, max_new_tokens=100, temperature=1.0):
        """自回归生成"""
        # 编码 prompt
        input_ids = self._encode(prompt)
        
        # 生成
        for _ in range(max_new_tokens):
            # 推理下一个 token
            logits = self._run_model(input_ids)
            
            # 温度缩放
            logits = logits / temperature
            
            # Softmax
            probs = np.exp(logits[0]) / np.sum(np.exp(logits[0]))
            
            # 采样
            next_token = np.random.choice(self.vocab_size, p=probs)
            
            # 检查结束
            if next_token == self._get_eos_token():
                break
            
            # 拼接
            input_ids = np.concatenate([input_ids, [[next_token]]])
            
            # 截断
            if input_ids.shape[1] > self.max_length:
                input_ids = input_ids[:, -self.max_length:]
        
        # 解码
        output_text = self._decode(input_ids[0])
        
        return output_text
    
    def _encode(self, text):
        """编码文本"""
        # 简化的编码实现
        tokens = list(text)
        input_ids = [ord(t) % self.vocab_size for t in tokens]
        
        return np.array([input_ids], dtype=np.int64)
    
    def _decode(self, token_ids):
        """解码 token"""
        # 简化的解码实现
        return "".join([chr(t) for t in token_ids if t < 128])

# 使用示例
gpt = GPTGenerator("gpt.om")

prompt = "今天天气真"
output = gpt.generate(prompt, max_new_tokens=50, temperature=0.8)
print(f"生成文本: {output}")

3.2 KV Cache 优化

class GPTWithKVCache:
    """带 KV Cache 的 GPT 推理"""
    
    def __init__(self, model_path):
        self.model = self._load_model(model_path)
        self.kv_cache = None
    
    def generate(self, prompt, max_new_tokens=100):
        """带 KV Cache 的生成"""
        # 编码 prompt
        input_ids = self._encode(prompt)
        
        # 首次推理 (计算所有 token 的 KV)
        logits, self.kv_cache = self._run_model_with_cache(input_ids)
        
        # 逐 token 生成
        for _ in range(max_new_tokens):
            # 只推理最后一个 token
            last_token = input_ids[:, -1:]
            
            # 使用 KV Cache
            logits, self.kv_cache = self._run_model_with_cache(
                last_token,
                past_key_values=self.kv_cache
            )
            
            # 采样
            next_token = self._sample(logits)
            
            # 拼接
            input_ids = np.concatenate([input_ids, [[next_token]]], axis=1)
        
        return self._decode(input_ids[0])
    
    def _run_model_with_cache(self, input_ids, past_key_values=None):
        """带 KV Cache 的推理"""
        # 实际实现中，这里会处理 KV Cache
        logits = self._run_model(input_ids)
        
        # 更新 KV Cache
        new_cache = self._update_cache(past_key_values, input_ids)
        
        return logits, new_cache
    
    def _update_cache(self, old_cache, new_tokens):
        """更新 KV Cache"""
        # 简化实现
        return new_tokens

# 使用示例
gpt_cache = GPTWithKVCache("gpt.om")
output = gpt_cache.generate("请解释什么是机器学习", max_new_tokens=200)

四、Token 匹配

4.1 分词器适配

class TokenizerAdapter:
    """分词器适配器"""
    
    def __init__(self, tokenizer_type="bert"):
        self.tokenizer_type = tokenizer_type
    
    def tokenize(self, text, max_length=128):
        """分词"""
        if self.tokenizer_type == "bert":
            return self._bert_tokenize(text, max_length)
        elif self.tokenizer_type == "gpt2":
            return self._gpt2_tokenize(text, max_length)
        elif self.tokenizer_type == "llama":
            return self._llama_tokenize(text, max_length)
        else:
            raise ValueError(f"不支持的分词器类型: {self.tokenizer_type}")
    
    def _bert_tokenize(self, text, max_length):
        """BERT 分词"""
        # 使用 BERT 分词器
        from transformers import BertTokenizer
        
        tokenizer = BertTokenizer.from_pretrained("bert-base-chinese")
        
        tokens = tokenizer(
            text,
            padding="max_length",
            max_length=max_length,
            truncation=True,
            return_tensors="np"
        )
        
        return {
            "input_ids": tokens["input_ids"],
            "attention_mask": tokens["attention_mask"],
            "token_type_ids": tokens.get("token_type_ids", np.zeros_like(tokens["input_ids"]))
        }
    
    def _gpt2_tokenize(self, text, max_length):
        """GPT-2 分词"""
        from transformers import GPT2Tokenizer
        
        tokenizer = GPT2Tokenizer.from_pretrained("gpt2")
        
        tokens = tokenizer(
            text,
            padding="max_length",
            max_length=max_length,
            truncation=True,
            return_tensors="np"
        )
        
        return {
            "input_ids": tokens["input_ids"],
            "attention_mask": tokens["attention_mask"]
        }
    
    def _llama_tokenize(self, text, max_length):
        """LLaMA 分词"""
        from transformers import LlamaTokenizer
        
        tokenizer = LlamaTokenizer.from_pretrained("llama-7b")
        
        tokens = tokenizer(
            text,
            padding="max_length",
            max_length=max_length,
            truncation=True,
            return_tensors="np"
        )
        
        return {
            "input_ids": tokens["input_ids"],
            "attention_mask": tokens["attention_mask"]
        }

# 使用示例
tokenizer_adapter = TokenizerAdapter("bert")

text = "这是测试文本"
tokens = tokenizer_adapter.tokenize(text, max_length=128)

print(f"input_ids: {tokens['input_ids'].shape}")
print(f"attention_mask: {tokens['attention_mask'].shape}")

五、常见问题

相关仓库

• transformers - NLP 模型库 https://gitee.com/huggingface/transformers
• sentencepiece - 分词器 https://gitee.com/google/sentencepiece
• torch_npu - 昇腾推理 https://gitee.com/ascend/torch_npu