1. 环境准备：从零搭建昇腾Llama部署平台

想在昇腾NPU上跑Llama大模型，第一步就是把环境给搭起来。我刚开始接触昇腾的时候，最头疼的就是环境配置，各种驱动、框架版本搞得人头大。不过现在有了云上资源，整个过程已经简化了很多，就算你没有物理硬件也能玩起来。

1.1 硬件选择：云上资源 vs 自有硬件

如果你只是想体验一下，或者做初步的验证，我强烈建议从云上资源开始。GitCode平台提供了免费的昇腾Notebook实例，配置是1个昇腾910B NPU、32核CPU和64GB内存，跑Llama-7B或者Llama-8B模型完全够用。最关键的是，这个资源是免费的，虽然有时间限制，但足够你把整个流程跑通。

申请流程很简单：登录GitCode，找到Notebook服务，选择NPU basic配置，镜像选euler2.9-py38-torch2.1.0-cann8.0-openmind0.6-notebook这个版本。这个镜像已经把PyTorch、CANN、torch_npu这些关键组件都预装好了，省去了你自己安装的麻烦。

如果你是要在生产环境部署，那就需要考虑自有硬件了。我推荐Atlas 300T Pro训练卡或者Atlas 800训练服务器，内存至少128GB，最好256GB以上，因为大模型加载需要大量内存。存储空间建议500GB起步，毕竟一个Llama-7B的模型文件就要13GB左右，再加上数据集、日志什么的，空间小了根本不够用。

1.2 软件环境配置：避开那些坑

环境配置这块，我踩过的坑可不少。首先是版本兼容性问题，PyTorch、torch_npu、CANN这三个组件的版本必须严格对应。我刚开始没注意，随便装了个最新版的PyTorch，结果torch_npu根本识别不了，折腾了半天才发现是版本不匹配。

正确的做法是先检查CANN版本，然后用对应的torch_npu版本。在终端里运行npu-smi info，能看到CANN的版本号，然后去昇腾官方文档查兼容性矩阵，找到对应的PyTorch和torch_npu版本。

安装完基础环境后，一定要验证NPU是否可用。我写了个简单的验证脚本：

import torch
import torch_npu  # 这个导入顺序不能错，必须在torch之后

print(f"PyTorch版本: {torch.__version__}")
print(f"torch_npu版本: {torch_npu.__version__}")
print(f"NPU可用: {torch.npu.is_available()}")
print(f"设备名称: {torch.npu.get_device_name(0)}")

如果最后一行显示NPU可用，并且能打印出设备名称，那说明环境配置成功了。这里有个细节要注意：import torch_npu必须在import torch之后，顺序反了就会报错。

接下来安装模型相关的依赖。transformers库是必须的，我建议用国内镜像源安装，速度会快很多：

pip install transformers>=4.36.0 accelerate>=0.24.0 -i https://pypi.tuna.tsinghua.edu.cn/simple
pip install numpy scipy scikit-learn matplotlib seaborn

有时候还会遇到OpenBLAS的线程冲突问题，表现就是报错"can't start new thread"。这个问题的解决方法是在代码最开头设置环境变量：

import os
os.environ["OMP_NUM_THREADS"] = "1"
os.environ["MKL_NUM_THREADS"] = "1"
os.environ["OPENBLAS_NUM_THREADS"] = "1"
os.environ["VECLIB_MAXIMUM_THREADS"] = "1"
os.environ["NUMEXPR_NUM_THREADS"] = "1"

这几个环境变量限制了底层数学库的线程数，能有效避免多线程冲突导致的崩溃。我建议在所有昇腾NPU的代码里都加上这几行，算是预防性措施。

2. 模型获取与转换：让Llama在昇腾上跑起来

环境准备好了，接下来就是搞模型。Llama模型现在有很多版本，从7B到70B参数规模的都有。对于初次尝试，我建议从Llama-2-7B或者Llama-3-8B开始，这两个模型大小适中，对硬件要求不算太高，又能体验到足够强的能力。

2.1 模型下载：避开权限墙

直接从Meta官方下载Llama模型需要申请权限，流程比较麻烦。我推荐用开源社区的镜像版本，比如NousResearch/Llama-2-7b-hf或者LLM-Research/Meta-Llama-3-8B-Instruct，这些镜像不需要特殊权限，下载速度也稳定。

如果你在国内，访问HuggingFace可能不太稳定，可以设置镜像源：

export HF_ENDPOINT=https://hf-mirror.com

或者用ModelScope来下载，这是国内的模型社区，速度会快很多：

from modelscope import snapshot_download

model_dir = snapshot_download(
    'LLM-Research/Meta-Llama-3-8B-Instruct',
    cache_dir='./models',
    revision='master'
)

我用ModelScope下载Llama-3-8B，大概花了20分钟，13GB的模型文件分成了两个部分下载。下载过程中如果中断了，它会自动续传，不用重新开始，这个设计很贴心。

2.2 模型加载到NPU：关键参数设置

模型下载好后，就要加载到NPU上了。这里有几个关键参数需要注意，直接影响到性能和稳定性。

首先是精度选择。昇腾NPU对半精度（FP16）支持最好，计算速度快，显存占用也少。如果用全精度（FP32），不仅速度慢，显存可能也不够用。所以一定要设置torch_dtype=torch.float16。

然后是设备映射。我推荐用device_map="npu"这个参数，它会把模型的不同层自动分配到NPU上，比手动调用.to("npu")更智能，还能避免CPU内存被大量占用。

完整的加载代码长这样：

from transformers import AutoTokenizer, AutoModelForCausalLM
import torch
import torch_npu

model_path = "./models/LLM-Research/Meta-Llama-3-8B-Instruct"

# 加载tokenizer
tokenizer = AutoTokenizer.from_pretrained(model_path)

# 加载模型到NPU
model = AutoModelForCausalLM.from_pretrained(
    model_path,
    torch_dtype=torch.float16,  # 半精度，必须设置
    device_map="npu",  # 自动分配到NPU
    low_cpu_mem_usage=True  # 减少CPU内存占用
).eval()  # 切换到推理模式

print(f"模型加载完成，显存占用: {torch.npu.memory_allocated()/1e9:.2f}GB")

加载完成后，记得检查显存占用。Llama-3-8B在FP16精度下大概占用15-16GB显存，如果显示的数字差太多，可能是哪里出问题了。

2.3 Tokenizer配置：避免生成乱码

很多人在模型能跑起来后，发现生成的文本乱七八糟，要么是重复输出，要么是乱码。这往往是tokenizer配置的问题。

Llama模型的tokenizer默认没有pad_token，但在批处理或者某些生成场景下需要用到。如果不设置，就可能出现各种奇怪的问题。解决方法很简单：

# 设置padding相关参数
tokenizer.pad_token_id = tokenizer.eos_token_id  # 用eos_token作为pad_token
tokenizer.padding_side = "left"  # 左侧padding，保证生成质量

还有一个细节：如果要用chat模板，Llama-3的对话格式比较特殊，需要正确处理结束符：

# 构造对话输入
prompt = "你好，请用中文介绍一下你自己。"
messages = [
    {"role": "system", "content": "You are a helpful AI assistant."},
    {"role": "user", "content": prompt}
]

# 应用chat模板
input_ids = tokenizer.apply_chat_template(
    messages,
    add_generation_prompt=True,
    return_tensors="pt"
).to("npu")  # 记得转移到NPU

# 设置结束符
terminators = [
    tokenizer.eos_token_id,
    tokenizer.convert_tokens_to_ids("<|eot_id|>")  # Llama-3的特殊结束符
]

这些配置看起来琐碎，但实际使用中能避免很多莫名其妙的问题。我建议把这些配置封装成一个函数，每次加载模型时都调用，确保一致性。

3. 性能调优实战：从能用变好用

模型能跑起来只是第一步，要想在实际项目中用得好，还得做性能调优。我在昇腾NPU上调试Llama模型时，发现有几个性能瓶颈特别常见，也总结出了一套优化方法。

3.1 动态Shape：头号性能杀手

昇腾NPU有个特点：它喜欢静态的计算图。如果每次推理的输入长度都不一样（这就是动态Shape），NPU的编译器（ATC）就得每次都重新构建计算图，没法利用缓存，导致第一次推理特别慢，有时候能慢上几十倍。

我遇到过最夸张的情况：输入长度从50变成51，推理时间从200ms暴涨到5000ms。查看日志会发现大量[GE BUILD GRAPH]的记录，这就是在重新编译计算图。

解决方案是固定输入长度，或者用分桶策略。具体来说，就是把输入补齐到固定的几个长度档位，比如128、256、512、1024。这样NPU只需要为每个长度编译一次计算图，后续同样长度的输入就能直接复用，实现零编译开销。

def pad_to_fixed_length(input_ids, target_length=512):
    """将输入补齐到固定长度"""
    current_length = input_ids.shape[1]
    if current_length < target_length:
        # 左侧padding
        padding = torch.full((input_ids.shape[0], target_length - current_length), 
                           tokenizer.pad_token_id, 
                           device=input_ids.device)
        input_ids = torch.cat([padding, input_ids], dim=1)
    elif current_length > target_length:
        # 截断到最大长度
        input_ids = input_ids[:, :target_length]
    return input_ids

# 在实际推理前调用
padded_inputs = pad_to_fixed_length(input_ids, target_length=256)

在实际项目中，我可以根据业务场景设计几个固定的长度档位。比如对话系统可以用256，文档处理用512，长文本生成用1024。用户输入短了就padding，长了就截断。虽然会损失一点信息，但性能提升是实实在在的。

3.2 数据搬运优化：减少CPU-NPU交互

另一个常见的性能问题是数据搬运。如果你在Python代码里频繁地在CPU和NPU之间传输数据，NPU的计算单元就会经常闲着等数据，利用率上不去。

用npu-smi info命令查看，如果NPU利用率长期低于50%，那很可能就是数据搬运成了瓶颈。解决方法很简单：提前搬运，减少交互。

# 不好的做法：在循环中频繁搬运
for prompt in prompts:
    inputs = tokenizer(prompt, return_tensors="pt")  # 在CPU上
    inputs = inputs.to("npu")  # 每次循环都搬运
    output = model.generate(**inputs)
    
# 好的做法：一次性搬运
all_inputs = []
for prompt in prompts:
    inputs = tokenizer(prompt, return_tensors="pt")
    all_inputs.append(inputs)

# 在CPU上完成所有预处理
batch_inputs = collate_fn(all_inputs)  # 自定义的批处理函数
batch_inputs = {k: v.to("npu") for k, v in batch_inputs.items()}  # 一次性搬运

# 在NPU上批量推理
outputs = model.generate(**batch_inputs)

还有一个技巧：尽量用torch_npu的原生算子替代Python层的循环。NPU对向量化操作优化得很好，但对Python层的控制流支持一般。比如矩阵乘法、卷积这些操作，用torch_npu的函数会比手写Python循环快很多。

3.3 显存管理：避免OOM的实用技巧

大模型最怕的就是显存溢出（OOM）。昇腾NPU对显存碎片比较敏感，如果频繁分配释放小块显存，很容易产生碎片，导致即使总显存够用，也分配不出连续的大块内存。

我常用的显存管理方法有几个。首先是在每次生成结束后主动清理缓存：

import gc

with torch.no_grad():
    outputs = model.generate(**inputs, max_new_tokens=512)

# 生成完成后立即清理
del outputs
gc.collect()
torch.npu.empty_cache()

其次是监控显存碎片情况：

def check_memory_fragmentation():
    stats = torch.npu.memory_stats()
    allocated = stats["allocated_bytes.all.current"]
    reserved = stats["reserved_bytes.all.current"]
    fragmentation = 1 - allocated / reserved if reserved > 0 else 0
    print(f"显存碎片率: {fragmentation:.2%}")
    return fragmentation

如果发现碎片率很高（比如超过30%），可以考虑重启服务，或者调整batch size和输入长度，减少显存的频繁分配释放。

还有一个高级技巧：使用torch.npu.memory_allocator设置缓存分配策略。昇腾NPU支持几种不同的分配器，有的对碎片更友好，有的分配速度更快。根据实际场景选择合适的分配器，能显著改善显存使用效率。

4. 实战案例：构建非结构化数据ETL管道

理论说再多，不如实际做一遍。我最近用昇腾NPU上的Llama-3-8B做了一个非结构化数据ETL（提取、转换、加载）的项目，效果很不错，这里分享一下具体实现。

4.1 项目背景与设计思路

在很多数据科学项目里，80%的时间都花在了数据清洗上。特别是从非结构化文本（比如医疗记录、简历、客服对话）中提取结构化信息，传统方法要么写复杂的规则，要么训练专门的模型，成本都很高。

我的想法是：用大语言模型来做这个事。让Llama-3扮演一个"信息录入员"，把杂乱的文本自动转换成规整的JSON。这样有几个好处：一是模型能理解上下文，提取更准确；二是适应性强，换一个领域只需要改改prompt，不用重新训练模型。

整个系统的架构很简单：输入是非结构化文本，经过Llama-3处理，输出结构化的JSON。关键是要设计好prompt，让模型知道我们要什么格式的数据。

4.2 核心代码实现

先来看系统prompt的设计，这是整个项目的灵魂：

system_instruction = """
你是一个专业的数据结构化助手。
请从用户的输入文本中提取关键实体信息，并严格以JSON格式输出。
不要包含任何多余的解释文字，只输出JSON。

如果是医疗文本，提取：{姓名, 年龄, 症状, 诊断, 建议}
如果是简历文本，提取：{姓名, 学历, 技能, 期望薪资, 联系方式}
"""

这个prompt有几个关键点：明确了角色（数据结构化助手）、限定了输出格式（只要JSON）、定义了不同文本类型的schema。模型会根据输入内容自动判断类型，然后按对应的schema提取信息。

数据预处理部分，我用了tokenizer的chat模板功能，这样能保证输入格式符合Llama-3的要求：

def prepare_input(text, system_prompt):
    """准备模型输入"""
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": text}
    ]
    
    # 应用chat模板
    text_input = tokenizer.apply_chat_template(
        messages,
        tokenize=False,  # 先不tokenize，方便调试
        add_generation_prompt=True
    )
    
    # 转换为tensor并移到NPU
    inputs = tokenizer(text_input, return_tensors="pt").to("npu")
    return inputs

推理时的参数设置也很重要。因为我们要的是结构化输出，需要关闭随机采样，让输出更稳定：

def extract_structured_data(text, system_prompt):
    """从文本中提取结构化数据"""
    inputs = prepare_input(text, system_prompt)
    
    with torch.no_grad():
        outputs = model.generate(
            **inputs,
            max_new_tokens=256,  # JSON输出一般不会太长
            do_sample=False,  # 关闭随机采样，保证输出稳定
            temperature=0.1,  # 低温度，减少随机性
            eos_token_id=tokenizer.eos_token_id,
            pad_token_id=tokenizer.pad_token_id
        )
    
    # 解码输出
    generated = outputs[0][inputs.input_ids.shape[1]:]  # 去掉输入部分
    response = tokenizer.decode(generated, skip_special_tokens=True)
    
    # 尝试解析JSON
    try:
        import json
        result = json.loads(response.strip())
        return result
    except json.JSONDecodeError:
        # 如果解析失败，返回原始文本
        return {"raw_output": response, "error": "JSON解析失败"}

4.3 实际测试与效果评估

我准备了两类测试数据：医疗记录和求职简历。医疗记录是这样的：

患者张伟，男，45岁。昨日夜间出现急性腹痛，位置在右下腹。
体温38.5度，伴有恶心呕吐。白细胞计数12000。
既往有高血压病史，长期服用硝苯地平。
初步诊断为急性阑尾炎，建议立即手术。

简历是这样的：

我是李娜，2020年毕业于北京大学计算机系硕士。
之前在字节跳动工作了3年，担任高级算法工程师。
精通Python, PyTorch, C++。
期望薪资是50k，希望能去上海发展。
邮箱：lina_code@example.com

运行提取程序后，医疗记录被转换成了：

{
  "姓名": "张伟",
  "年龄": "45",
  "症状": "急性腹痛，位置在右下腹，体温38.5度，伴有恶心呕吐",
  "诊断": "急性阑尾炎",
  "建议": "立即手术"
}

简历被转换成了：

{
  "姓名": "李娜",
  "学历": "北京大学计算机系硕士",
  "技能": "Python, PyTorch, C++",
  "期望薪资": "50k",
  "联系方式": "lina_code@example.com"
}

准确率相当高，模型不仅能提取显式信息（比如姓名、年龄），还能理解隐含信息（"急性阑尾炎"是诊断，"立即手术"是建议）。而且对于简历中的技能列表，它能自动整理成数组格式，虽然我prompt里没明确要求，但模型自己理解了这是多个技能。

性能方面，在昇腾910B NPU上，单条记录的处理时间大概在1.5-2秒左右，包括模型推理和JSON解析。如果批量处理，速度还能提升。这个性能对于很多离线ETL场景已经够用了。

4.4 生产环境部署建议

如果要把这个系统用到生产环境，我还有几个建议。首先是错误处理要完善。大模型有时候会"胡言乱语"，输出不符合JSON格式，或者提取的信息明显错误。需要加上重试机制和人工审核流程。

def robust_extraction(text, system_prompt, max_retries=3):
    """带重试的稳健提取"""
    for attempt in range(max_retries):
        try:
            result = extract_structured_data(text, system_prompt)
            if validate_result(result):  # 自定义的验证函数
                return result
        except Exception as e:
            print(f"第{attempt+1}次尝试失败: {e}")
            time.sleep(1)  # 等待一下再重试
    
    # 所有重试都失败，返回错误
    return {"error": "提取失败", "text": text}

其次是监控和日志。要记录每次处理的耗时、显存使用情况、成功率等指标，方便后续优化。

import time
import psutil

def monitor_extraction(text, system_prompt):
    """监控提取过程"""
    start_time = time.time()
    start_memory = torch.npu.memory_allocated()
    
    result = extract_structured_data(text, system_prompt)
    
    end_time = time.time()
    end_memory = torch.npu.memory_allocated()
    
    metrics = {
        "processing_time": end_time - start_time,
        "memory_increase": (end_memory - start_memory) / 1e9,  # GB
        "success": "error" not in result,
        "timestamp": time.time()
    }
    
    # 记录到日志或数据库
    log_metrics(metrics)
    return result

最后是版本管理。模型会更新，prompt可能需要调整，schema也可能变化。要做好版本控制，确保可追溯、可回滚。

5. 高级优化技巧：让性能再上一个台阶

前面的优化已经能让模型跑得不错了，但如果对性能有更高要求，比如要支持高并发、低延迟的场景，还需要一些高级技巧。

5.1 模型量化：平衡精度与速度

模型量化是提升推理速度、减少显存占用的有效手段。Llama模型通常用FP16精度，我们可以进一步量化为INT8甚至INT4。量化后的模型速度能提升1.5-2倍，显存占用减少一半。

昇腾NPU支持多种量化方式，我比较推荐的是动态量化，因为它不需要校准数据，实现简单：

from transformers import BitsAndBytesConfig
import torch

# 配置8位量化
quantization_config = BitsAndBytesConfig(
    load_in_8bit=True,
    llm_int8_threshold=6.0,
    llm_int8_has_fp16_weight=False
)

# 加载量化模型
model = AutoModelForCausalLM.from_pretrained(
    model_path,
    quantization_config=quantization_config,
    torch_dtype=torch.float16,
    device_map="npu"
)

量化后需要测试一下精度损失。我对比过Llama-3-8B在FP16和INT8下的表现，在大多数任务上差异不大，但在一些需要精确数值计算的任务上（比如数学推理），INT8会有明显精度下降。所以要不要量化、量化到什么程度，得根据具体应用场景决定。

如果对精度要求高，但又想节省显存，可以试试混合精度：关键层用FP16，其他层用INT8。这个实现起来复杂一些，需要手动指定哪些层保持高精度。

5.2 批处理推理：提升吞吐量

单条推理的延迟很难再降，但提升吞吐量还有一个大招：批处理。把多个请求打包一起推理，能显著提高NPU的利用率。

批处理的关键是要处理不同长度的输入。我前面提到的padding到固定长度就是为批处理做准备的。实现批处理推理：

def batch_inference(texts, system_prompt, batch_size=4):
    """批量推理"""
    all_results = []
    
    # 按batch_size分批
    for i in range(0, len(texts), batch_size):
        batch_texts = texts[i:i+batch_size]
        
        # 准备批量输入
        batch_inputs = []
        for text in batch_texts:
            inputs = prepare_input(text, system_prompt)
            batch_inputs.append(inputs)
        
        # 批处理需要统一长度，找到本批最大长度
        max_len = max(inp.input_ids.shape[1] for inp in batch_inputs)
        
        # 补齐到相同长度
        padded_inputs = []
        attention_masks = []
        for inp in batch_inputs:
            current_len = inp.input_ids.shape[1]
            if current_len < max_len:
                # 左侧padding
                pad_len = max_len - current_len
                padding = torch.full((1, pad_len), tokenizer.pad_token_id, 
                                   device=inp.input_ids.device)
                padded = torch.cat([padding, inp.input_ids], dim=1)
                # attention mask，padding部分为0
                mask = torch.cat([torch.zeros((1, pad_len), device=inp.attention_mask.device),
                                inp.attention_mask], dim=1)
            else:
                padded = inp.input_ids[:, :max_len]
                mask = inp.attention_mask[:, :max_len]
            
            padded_inputs.append(padded)
            attention_masks.append(mask)
        
        # 堆叠成batch
        batch_input_ids = torch.cat(padded_inputs, dim=0)
        batch_attention_mask = torch.cat(attention_masks, dim=0)
        
        # 批量推理
        with torch.no_grad():
            outputs = model.generate(
                input_ids=batch_input_ids,
                attention_mask=batch_attention_mask,
                max_new_tokens=256,
                do_sample=False,
                temperature=0.1
            )
        
        # 解码每个结果
        for j in range(len(batch_texts)):
            # 去掉padding和输入部分
            start_idx = batch_input_ids[j].shape[0] - torch.sum(batch_attention_mask[j]).item()
            generated = outputs[j][start_idx:]
            response = tokenizer.decode(generated, skip_special_tokens=True)
            all_results.append(response)
    
    return all_results

批处理能大幅提升吞吐量，但延迟可能会增加（因为要等凑够一个batch）。在实际系统中，我通常用动态批处理：设置一个最大等待时间，比如100ms，在这段时间内积累请求，然后一起处理。这样能在吞吐量和延迟之间取得平衡。

5.3 使用vLLM推理框架

如果你觉得手写批处理太麻烦，或者想要更极致的性能，可以试试vLLM。这是一个专门为大语言模型推理优化的框架，支持连续批处理、PagedAttention等高级特性。

vLLM也支持昇腾NPU，虽然需要一些适配工作。基本的用法是这样的：

from vllm import LLM, SamplingParams

# 初始化vLLM
llm = LLM(
    model="NousResearch/Llama-2-7b-hf",
    tensor_parallel_size=1,  # 单卡
    gpu_memory_utilization=0.9,  # 显存使用率
    max_model_len=4096,  # 最大上下文长度
    quantization="awq"  # 量化方式，可选
)

# 准备采样参数
sampling_params = SamplingParams(
    temperature=0.1,
    top_p=0.9,
    max_tokens=256
)

# 批量推理
prompts = ["问题1", "问题2", "问题3"]
outputs = llm.generate(prompts, sampling_params)

for output in outputs:
    print(f"Prompt: {output.prompt}")
    print(f"Generated: {output.outputs[0].text}")

vLLM的优点是吞吐量高，显存利用率好，而且API设计得很简洁。缺点是定制性不如直接写PyTorch代码强，有些高级功能可能不支持。

5.4 监控与调优工具

性能调优不能靠猜，得有数据支持。昇腾提供了一些监控工具，可以帮助我们找到性能瓶颈。

最基础的是npu-smi命令，可以实时查看NPU的使用情况：

# 查看NPU状态
npu-smi info

# 监控NPU利用率
npu-smi monitor -i 0 -c 1  # 监控0号设备，每秒刷新一次

在代码里，可以用torch的profiler来记录详细的性能数据：

from torch.profiler import profile, record_function, ProfilerActivity

with profile(
    activities=[ProfilerActivity.NPU],  # 记录NPU活动
    record_shapes=True,
    profile_memory=True,
    with_stack=True
) as prof:
    with record_function("model_inference"):
        outputs = model.generate(**inputs, max_new_tokens=256)

# 输出性能报告
print(prof.key_averages().table(sort_by="npu_time_total", row_limit=20))

这个性能报告会显示每个算子的耗时、显存使用情况，帮你找到最耗时的操作。我常用这个功能来优化数据预处理部分，有时候CPU到NPU的数据拷贝比模型推理还耗时，这时候就要考虑优化数据流了。

还有一个实用的监控指标是首token延迟和后续token延迟。在流式输出场景下，用户很关注首字速度。可以用这个代码测量：

import time

def measure_latency(prompt, generate_length=100):
    """测量生成延迟"""
    inputs = tokenizer(prompt, return_tensors="pt").to("npu")
    
    # 预热
    for _ in range(3):
        _ = model.generate(**inputs, max_new_tokens=1)
    
    # 测量首token延迟
    torch.npu.synchronize()
    start_time = time.time()
    outputs = model.generate(**inputs, max_new_tokens=1)
    torch.npu.synchronize()
    first_token_latency = time.time() - start_time
    
    # 测量后续token延迟
    torch.npu.synchronize()
    start_time = time.time()
    outputs = model.generate(**inputs, max_new_tokens=generate_length)
    torch.npu.synchronize()
    total_latency = time.time() - start_time
    
    avg_per_token = (total_latency - first_token_latency) / (generate_length - 1)
    
    return {
        "first_token_latency": first_token_latency * 1000,  # 毫秒
        "avg_per_token_latency": avg_per_token * 1000,  # 毫秒
        "total_tokens": generate_length
    }

根据我的测试，在昇腾910B上跑Llama-3-8B，首token延迟大概在300-500ms，后续每个token在30-50ms。如果发现首token延迟特别高，那可能是动态Shape的问题，需要检查输入是否固定长度。

6. 常见问题排查手册

在实际部署中，总会遇到各种问题。我整理了一个常见问题排查手册，都是我自己踩过的坑。

6.1 环境配置问题

问题1：torch.npu找不到

AttributeError: module 'torch' has no attribute 'npu'

原因：没有导入torch_npu，或者导入顺序不对。

解决：确保在import torch之后import torch_npu：

import torch
import torch_npu  # 必须导入这个

问题2：版本不兼容

ImportError: torch_npu版本与CANN版本不匹配

原因：torch_npu和CANN版本不对应。

解决：检查CANN版本（npu-smi info），然后安装对应的torch_npu版本。昇腾官网有兼容性矩阵，一定要对照着来。

问题3：内存不足

RuntimeError: NPU out of memory

原因：模型太大，或者batch size设得太大。

解决：

1. 用FP16而不是FP32
2. 减小batch size
3. 启用梯度检查点（训练时）
4. 使用模型并行（多卡时）

6.2 模型推理问题

问题4：生成结果乱码或重复

模型输出乱七八糟的文本，或者不断重复同一句话。

原因：tokenizer配置不对，或者生成参数设置有问题。

解决：

# 正确配置tokenizer
tokenizer.pad_token_id = tokenizer.eos_token_id
tokenizer.padding_side = "left"

# 调整生成参数
outputs = model.generate(
    **inputs,
    max_new_tokens=256,
    do_sample=True,  # 启用采样
    temperature=0.7,  # 控制随机性，0.7比较平衡
    top_p=0.9,  # 核采样
    repetition_penalty=1.1,  # 抑制重复
    no_repeat_ngram_size=3  # 禁止3-gram重复
)

问题5：推理速度慢

第一次推理特别慢，后续正常。

原因：动态Shape导致每次都要重新编译计算图。

解决：固定输入长度，或者用我前面提到的分桶策略。

问题6：精度异常

和GPU上的结果对比，差异很大。

原因：可能是数据类型问题，或者某些算子回退到了CPU。

解决：

1. 确保模型用FP16加载
2. 检查输入数据类型，必须是torch.int64
3. 查看日志，搜索"fallback"，看是否有算子回退到CPU执行

6.3 多卡训练/推理问题

问题7：多卡通信失败

单卡正常，多卡就挂死。

原因：HCCL（昇腾的通信库）配置问题。

解决：

# 设置环境变量
export HCCL_WHITELIST_DISABLE=1
export HCCL_IF_IP=你的IP地址

# 用torchrun启动分布式训练
torchrun --nproc_per_node=2 train.py

问题8：负载不均衡

多卡中有的卡利用率高，有的卡利用率低。

原因：模型并行划分不均匀，或者数据分布不均。

解决：

1. 检查模型是否均匀划分到各个卡上
2. 使用更均衡的并行策略，比如流水线并行
3. 调整数据加载器，确保每个卡处理的数据量差不多

6.4 性能调优问题

问题9：NPU利用率低

用npu-smi查看，利用率长期低于50%。

原因：数据搬运瓶颈，或者计算密度不够。

解决：

1. 增加batch size，提高计算密度
2. 使用异步数据加载
3. 预取数据到NPU内存
4. 使用更高效的数据格式，比如torch.channels_last

问题10：显存碎片化

batch size不大，但频繁OOM。

原因：频繁分配释放显存，导致碎片化。

解决：

1. 使用内存池
2. 固定输入输出tensor的大小
3. 定期清理缓存：torch.npu.empty_cache()
4. 重启服务（最直接的方法）

6.5 调试技巧

遇到问题不知道怎么下手时，可以按这个流程排查：

1. 最小化复现：先写一个最简单的测试脚本，排除业务代码的影响
2. 逐模块验证：从数据加载→模型加载→前向推理→结果输出，一步步验证
3. 查看日志：昇腾的日志很详细，设置export ASCEND_GLOBAL_LOG_LEVEL=3可以看到更多信息
4. 对比验证：在CPU或GPU上跑同样的代码，看是否正常
5. 社区求助：昇腾社区有很多热心开发者，描述清楚问题现象和复现步骤

我习惯在代码里加一些调试开关，方便排查问题：

DEBUG = True

def debug_print(*args, **kwargs):
    if DEBUG:
        print(*args, **kwargs)

# 在关键位置添加调试信息
debug_print(f"输入shape: {input_ids.shape}")
debug_print(f"显存使用: {torch.npu.memory_allocated()/1e9:.2f}GB")

这些调试信息在生产环境可以关掉，在测试环境打开，能快速定位问题。

7. 生产环境部署建议

最后聊聊生产环境部署要注意的事情。实验室里跑通只是第一步，要真正上线服务，还有很多工程化的工作要做。

7.1 容器化部署

我强烈建议用Docker容器化部署。昇腾提供了官方的基础镜像，里面已经配置好了CANN、PyTorch、torch_npu等组件，比自己从头配置省事多了。

Dockerfile大概长这样：

FROM ascendhub.huaweicloud.com/public-ascendhub/pytorch:2.1.0-cann8.0-ubuntu20.04

# 设置工作目录
WORKDIR /app

# 复制代码
COPY . .

# 安装依赖
RUN pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

# 设置环境变量
ENV OMP_NUM_THREADS=1
ENV MKL_NUM_THREADS=1
ENV PYTHONUNBUFFERED=1

# 启动命令
CMD ["python", "app.py"]

用Docker Compose管理多服务：

version: '3.8'

services:
  llama-service:
    build: .
    container_name: llama-service
    runtime: ascend  # 指定昇腾运行时
    devices:
      - "/dev/davinci0"  # 挂载NPU设备
    volumes:
      - ./models:/app/models  # 挂载模型目录
      - ./logs:/app/logs      # 挂载日志目录
    environment:
      - NPU_VISIBLE_DEVICES=0
      - ASCEND_GLOBAL_LOG_LEVEL=3
    ports:
      - "8000:8000"
    restart: unless-stopped

容器化的好处是环境一致，容易扩展，也方便做CI/CD。

7.2 服务化与API设计

直接跑Python脚本不适合生产环境，需要封装成服务。我常用FastAPI来做API服务：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List
import uvicorn

app = FastAPI(title="Llama推理服务")

class GenerationRequest(BaseModel):
    prompt: str
    max_tokens: int = 256
    temperature: float = 0.7

class GenerationResponse(BaseModel):
    text: str
    tokens_generated: int
    time_cost: float

@app.post("/generate", response_model=GenerationResponse)
async def generate_text(request: GenerationRequest):
    """文本生成接口"""
    try:
        start_time = time.time()
        
        # 预处理输入
        inputs = tokenizer(request.prompt, return_tensors="pt").to("npu")
        
        # 生成
        with torch.no_grad():
            outputs = model.generate(
                **inputs,
                max_new_tokens=request.max_tokens,
                temperature=request.temperature,
                do_sample=True
            )
        
        # 解码
        generated = outputs[0][inputs.input_ids.shape[1]:]
        text = tokenizer.decode(generated, skip_special_tokens=True)
        
        time_cost = time.time() - start_time
        
        return GenerationResponse(
            text=text,
            tokens_generated=len(generated),
            time_cost=time_cost
        )
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health_check():
    """健康检查"""
    return {
        "status": "healthy",
        "npu_available": torch.npu.is_available(),
        "model_loaded": model is not None
    }

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

这个服务提供了生成接口和健康检查接口，可以用nginx做负载均衡，部署多个实例。

7.3 监控与告警

上线后要监控服务状态。我通常监控这几个指标：

1. 服务可用性：HTTP状态码、响应时间
2. 资源使用：NPU利用率、显存使用、温度
3. 业务指标：请求量、成功率、平均生成长度

用Prometheus收集指标，Grafana展示，关键指标设置告警：

from prometheus_client import Counter, Histogram, Gauge

# 定义指标
REQUEST_COUNT = Counter('llama_requests_total', 'Total requests')
REQUEST_LATENCY = Histogram('llama_request_latency_seconds', 'Request latency')
GPU_UTILIZATION = Gauge('npu_utilization_percent', 'NPU utilization')
GPU_MEMORY = Gauge('npu_memory_used_gb', 'NPU memory used')

@app.post("/generate")
async def generate_text(request: GenerationRequest):
    REQUEST_COUNT.inc()
    
    with REQUEST_LATENCY.time():
        result = await do_generate(request)
    
    # 更新NPU指标
    GPU_UTILIZATION.set(torch.npu.utilization())
    GPU_MEMORY.set(torch.npu.memory_allocated() / 1e9)
    
    return result

7.4 版本管理与回滚

模型服务更新时要有回滚方案。我用的策略是：

1. 蓝绿部署：新旧版本同时运行，流量逐步切换
2. 模型版本化：每个模型有唯一版本号，请求时指定
3. 配置中心：生成参数、prompt模板等放在配置中心，热更新

class ModelManager:
    def __init__(self):
        self.models = {}  # 版本->模型映射
        self.current_version = None
    
    def load_model(self, version, model_path):
        """加载新版本模型"""
        if version in self.models:
            return
        
        # 在新进程中加载，避免影响当前服务
        model = load_model_in_subprocess(model_path)
        self.models[version] = model
        
        # 预热
        self.warmup_model(version)
    
    def switch_version(self, version):
        """切换当前版本"""
        if version not in self.models:
            raise ValueError(f"版本{version}未加载")
        
        old_version = self.current_version
        self.current_version = version
        
        # 异步卸载旧版本
        if old_version:
            self.unload_model_async(old_version)
    
    def get_model(self, version=None):
        """获取模型实例"""
        if version is None:
            version = self.current_version
        return self.models.get(version)

这套方案虽然复杂，但能保证服务平稳更新，出现问题时能快速回滚。

7.5 成本优化

最后说说成本优化。昇腾NPU虽然比同性能的GPU便宜，但用量大了也是一笔开销。

优化方向1：请求合并。很多场景下用户的请求可以合并处理，比如聊天历史相似的对话、同一批次的文档处理。合并后一次推理，分摊成本。

优化方向2：缓存策略。相似的请求结果可以缓存，设置合适的TTL。比如常见问题回答、模板化内容生成。

优化方向3：动态扩缩容。根据请求量自动调整实例数，低峰期减少实例节省成本。

优化方向4：模型蒸馏。用大模型蒸馏出小模型，简单任务用小模型，复杂任务用大模型，混合使用降低成本。

这些优化需要根据业务特点来设计，没有通用的方案。关键是要监控成本指标，持续优化。

我在实际项目中用这套方案部署了Llama-3-8B服务，支撑了每天几十万的请求，稳定运行了半年多。昇腾NPU的性价比确实不错，虽然生态还在完善中，但基本功能都很稳定。如果你也在考虑国产AI芯片，昇腾是个值得尝试的选择。