算力申请：https://ai.gitcode.com/ascend-tribe/openPangu-Ultra-MoE-718B-V1.1

模型地址：https://gitcode.com/test-oh-kb/Meta-Llama-3-8B-Instruct

---

在当前的大模型落地实践中，最大的瓶颈往往不是算力，而是高质量的数据。无论是微调行业小模型，还是构建 RAG（检索增强生成）知识库，都需要大量的结构化问答（QA）数据。然而，人工标注成本高昂，而使用云端 API 处理私有文档又面临数据隐私泄露的风险。

我将分享如何在国产算力平台——昇腾（Ascend）Atlas 800T NPU 上，利用 AtomGit的notebook功能快速部署 Llama-3-8B-Instruct 模型，构建一个本地化的“合成数据工厂”。该方案来测试是否能够利用 Atlas 800TNPU 的稳定算力，批量将非结构化文档转化为标准的 JSON 格式训练数据，全程数据不出域，安全可控。

一、 环境与模型准备

在开始构建应用之前，我们首先需要确保软硬件环境就绪。

1. 硬件检查

本次测试运行在昇腾 Atlas 800T 。我们首先使用 npu-smi info 确认 NPU 状态：

可以看到芯片状态（Health）为 OK。

2. 依赖安装

昇腾环境对 PyTorch 生态的支持已经非常成熟。我们需要安装 transformers、accelerate 以及魔搭社区的 modelscope SDK。

Bash

pip install transformers accelerate modelscope

3. 模型下载

为了获得更稳定的下载体验，我们编写脚本从 ModelScope 社区下载 Meta-Llama-3-8B-Instruct 权重。

下载脚本 (download.py):

from modelscope import snapshot_download
 
print("正在从 ModelScope 下载 Llama-3-8B-Instruct...")
model_dir = snapshot_download(
    'LLM-Research/Meta-Llama-3-8B-Instruct',
    cache_dir='./models',
    revision='master'
)
print(f"模型已下载至: {model_dir}")

运行该脚本后，模型权重将被保存到本地，为接下来的推理做准备。

二、 核心实战：构建“合成数据工厂”

目前，环境已经跑通。接下来我希望用模型产生业务价值。从而测试该模型在 昇腾 NPU上部署的实际效果如何。

场景定义

在企业知识库构建中，我们通常有大量的技术文档（PDF/Markdown），需要将其转换为 “问题-答案” 对（QA Pairs），以便进行微调或向量检索。

难点分析

直接让模型生成数据，容易出现以下问题：

1. 格式不可控：模型喜欢闲聊，输出的一堆废话无法被代码解析。

2. 幻觉风险：模型可能脱离文档内容瞎编。

尤其是这种权重小的模型！！

解决方案

我们采用 Prompt Engineering（提示工程） + Regex Parsing（正则清洗） 的组合策略，并利用昇腾 NPU 进行本地批量推理。

三、 代码实现

为了更清晰进行逻辑展示，我将 data_factory.py 的实现拆解为三个核心步骤：环境初始化、核心功能定义、以及批量生产主流程。

步骤 1：环境初始化与模型加载

这一步是之后操作的基础。需要激活昇腾 NPU 环境，并解决 Llama-3 在 Tokenizer 上的一些兼容性小问题（如缺少的 Pad Token）。

import torch
import torch_npu  # 【关键】必须导入，用于激活 NPU 后端
from transformers import AutoTokenizer, AutoModelForCausalLM
import json
import re
import time
import os
 
# --- 配置中心 ---
MODEL_PATH = "./models/LLM-Research/Meta-Llama-3-8B-Instruct"
DEVICE = "npu"  # 指定计算设备为昇腾 NPU
OUTPUT_FILE = "synthetic_qa_dataset.json"
 
def load_model_environment():
    print(f"[*] 正在初始化 NPU 环境并加载模型: {MODEL_PATH}...")
    
    # 1. 加载 Tokenizer
    tokenizer = AutoTokenizer.from_pretrained(MODEL_PATH, trust_remote_code=True)
    
    # 【适配补丁】修复 Llama-3 缺失 pad_token 导致无法进行批量推理的问题
    if tokenizer.pad_token is None:
        tokenizer.pad_token = tokenizer.eos_token
        tokenizer.pad_token_id = tokenizer.eos_token_id
 
    # 2. 加载模型到 NPU
    # 注意：在 NPU 上强烈建议使用 float16，既能提升推理速度，又能减少显存占用
    model = AutoModelForCausalLM.from_pretrained(
        MODEL_PATH,
        torch_dtype=torch.float16, 
        device_map=DEVICE,
        trust_remote_code=True
    )
    
    print(f"[*] 模型加载成功！当前运行设备: {model.device}")
    return tokenizer, model
 
# 初始化全局模型变量
tokenizer, model = load_model_environment()
 
# 定义停止符，防止模型生成过多无用内容
terminators = [
    tokenizer.eos_token_id,
    tokenizer.convert_tokens_to_ids("<|eot_id|>")
]

步骤 2：Prompt构建与正则清洗

这是用来处理数据的逻辑代码。大模型直接生成的输出往往包含“废话”，我们需要通过 extract_json 函数配合正则表达式，从非结构化的回答中精准提取出 JSON 数据。

def extract_json(text):
    """
    鲁棒性清洗器：从模型可能包含“废话”的输出中，精准提取 JSON 对象
    """
    # 策略 1: 优先匹配 Markdown 代码块中的 JSON
    match = re.search(r"```json\s*(.*?)```", text, re.DOTALL)
    if match:
        json_str = match.group(1)
    else:
        # 策略 2: 兜底匹配，寻找最外层的花括号或方括号
        match = re.search(r"(\[.*\]|\{.*\})", text, re.DOTALL)
        if match:
            json_str = match.group(1)
        else:
            return None
    
    try:
        return json.loads(json_str)
    except json.JSONDecodeError:
        # 策略 3: 尝试简单的格式自动修复 (如去除尾部多余逗号)
        try:
            return json.loads(json_str.replace(",\n]", "\n]"))
        except:
            return None
 
def generate_qa_data(doc_text):
    """
    单次推理核心函数：输入文档片段 -> 输出结构化 QA 列表
    """
    # System Prompt: 设定“数据专家”人设，并强制约束输出格式
    system_prompt = """你是一个专业的数据合成助手。请根据提供的技术文档，生成 3 个高质量的中文问答对（QA Pair）。
    【输出要求】
    1. 必须严格输出标准的 JSON 列表格式：[{"question": "...", "answer": "..."}, ...]
    2. 问题要具体，答案要准确且完全来源于文档。
    3. 严禁包含任何 Markdown 标记以外的解释性文字。"""
    
    user_prompt = f"文档内容如下：\n{doc_text}\n\n请生成 QA JSON："
 
    # 构建对话模板
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_prompt}
    ]
    input_ids = tokenizer.apply_chat_template(
        messages, add_generation_prompt=True, return_tensors="pt"
    ).to(model.device)
 
    # NPU 推理生成
    outputs = model.generate(
        input_ids,
        max_new_tokens=1024,
        eos_token_id=terminators,
        do_sample=True,      # 启用采样，增加生成数据的多样性
        temperature=0.6,     # 温度适中，在创造性与准确性之间平衡
        top_p=0.9
    )
 
    # 解码与清洗
    response = outputs[0][input_ids.shape[-1]:]
    response_text = tokenizer.decode(response, skip_special_tokens=True)
    return extract_json(response_text)

步骤 3：批量生产流水线

最后，将上述模块组装起来，模拟从数据库读取文档，批量处理并保存结果的全过程。

if __name__ == "__main__":
    # --- 模拟数据源 (实际场景中可替换为读取 PDF/TXT 文件) ---
    RAW_DOCS = [
        {"title": "Atlas 800T", "content": "Atlas 800T是一款具有高算力..."},
        {"title": "CANN 架构", "content": "CANN是华为针对AI场景推出的异构计算架构..."},
        # ... 更多文档 ...
    ]
 
    total_dataset = []
    print(f"\n🚀 启动数据合成工厂 | 待处理任务: {len(RAW_DOCS)} 篇")
    print("-" * 60)
 
    start_time = time.time()
 
    # --- 批量处理循环 ---
    for idx, doc in enumerate(RAW_DOCS):
        print(f"[{idx+1}/{len(RAW_DOCS)}] 处理: 《{doc['title']}》...", end="", flush=True)
        
        try:
            # 调用核心生成逻辑
            qa_list = generate_qa_data(doc['content'])
            
            # 数据校验
            if qa_list and isinstance(qa_list, list):
                total_dataset.extend(qa_list)
                print(f" ✅ 成功 | 产出 {len(qa_list)} 条 QA")
            else:
                print(f" ⚠️ 格式错误 (JSON 解析失败)")
                
        except Exception as e:
            print(f" ❌ 异常: {e}")
 
    # --- 结果持久化 ---
    if total_dataset:
        with open(OUTPUT_FILE, "w", encoding="utf-8") as f:
            json.dump(total_dataset, f, ensure_ascii=False, indent=4)
        print(f"\n💾 任务完成！数据集已保存至: {os.path.abspath(OUTPUT_FILE)}")
        print(f"📊 总耗时: {time.time() - start_time:.2f}s")

以下为完整代码：

import torch
import torch_npu  # 【关键】激活 NPU 环境
from transformers import AutoTokenizer, AutoModelForCausalLM
import json
import re
import time
import os
 
# ================= 0. 全局配置 =================
# 请确保此路径是你实际下载模型的路径
MODEL_PATH = "./models/LLM-Research/Meta-Llama-3-8B-Instruct"
DEVICE = "npu"  # 指定使用昇腾 NPU
OUTPUT_FILE = "synthetic_qa_dataset.json"
 
# ================= 1. 模型加载 =================
print("=" * 60)
print(f"[*] 正在初始化 NPU 环境并加载模型: {MODEL_PATH}")
print("=" * 60)
 
try:
    tokenizer = AutoTokenizer.from_pretrained(MODEL_PATH, trust_remote_code=True)
    # 修复 Llama-3 可能缺失 pad_token 导致无法进行批量推理的问题
    if tokenizer.pad_token is None:
        tokenizer.pad_token = tokenizer.eos_token
        tokenizer.pad_token_id = tokenizer.eos_token_id
 
    model = AutoModelForCausalLM.from_pretrained(
        MODEL_PATH,
        torch_dtype=torch.float16,  # NPU 上强烈建议使用半精度 (fp16) 以提升速度并减少显存
        device_map=DEVICE,
        trust_remote_code=True
    )
    
    # 设定 Llama-3 特有的停止符
    terminators = [
        tokenizer.eos_token_id,
        tokenizer.convert_tokens_to_ids("<|eot_id|>")
    ]
    print(f"[*] 模型加载成功！当前设备: {model.device}")
 
except Exception as e:
    print(f"[!] 模型加载失败，请检查路径或 NPU 驱动: {e}")
    exit(1)
 
# ================= 2. 核心功能函数 =================
 
def extract_json(text):
    """
    鲁棒性 JSON 提取器：从模型可能包含废话的输出中提取纯净的 JSON
    """
    # 优先匹配 Markdown 代码块
    match = re.search(r"```json\s*(.*?)```", text, re.DOTALL)
    if match:
        json_str = match.group(1)
    else:
        # 兜底：匹配最外层的花括号或方括号
        match = re.search(r"(\[.*\]|\{.*\})", text, re.DOTALL)
        if match:
            json_str = match.group(1)
        else:
            return None
    
    try:
        # 尝试解析
        return json.loads(json_str)
    except json.JSONDecodeError:
        # 简单的自动修复尝试（处理常见尾部逗号错误）
        try:
            return json.loads(json_str.replace(",\n]", "\n]"))
        except:
            return None
 
def generate_qa_data(doc_text):
    """
    单次推理函数：输入文档 -> 输出 QA 列表
    """
    system_prompt = """你是一个专业的数据合成助手。请根据提供的技术文档，生成 3 个高质量的中文问答对（QA Pair）。
    
    【输出要求】
    1. 必须严格输出标准的 JSON 列表格式：[{"question": "...", "answer": "..."}, ...]
    2. 问题要具体，答案要准确且完全来源于文档。
    3. 严禁包含任何 Markdown 标记以外的解释性文字。
    """
    
    user_prompt = f"文档内容如下：\n{doc_text}\n\n请生成 QA JSON："
 
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_prompt}
    ]
 
    # 构建 Prompt
    input_ids = tokenizer.apply_chat_template(
        messages,
        add_generation_prompt=True,
        return_tensors="pt"
    ).to(model.device)
 
    # NPU 推理生成
    outputs = model.generate(
        input_ids,
        max_new_tokens=1024, # 允许生成较长文本
        eos_token_id=terminators,
        do_sample=True,      # 采样模式增加多样性
        temperature=0.6,     # 温度适中，保证准确性
        top_p=0.9
    )
 
    # 解码
    response = outputs[0][input_ids.shape[-1]:]
    response_text = tokenizer.decode(response, skip_special_tokens=True)
    
    # 清洗
    structured_data = extract_json(response_text)
    return structured_data
 
# ================= 3. 批量生产流水线 (Mock Data) =================
 
if __name__ == "__main__":
    # 模拟从数据库或文件夹读取的 5 篇技术文档
    # 这些文本涵盖了昇腾全栈，适合作为测试素材
    RAW_DOCS = [
        {
            "title": "Atlas 800T",
            "content": "Atlas 800T是一款具有高算力、高能效比的AI处理器。它采用达芬奇架构，集成了32个达芬奇核心。昇腾910半精度（FP16）算力达到320 TFLOPS，整数精度（INT8）算力达到640 TOPS，功耗为310W。该芯片采用7nm工艺，主要应用于云端训练场景，支持大规模分布式训练集群构建。"
        },
        {
            "title": "CANN 异构计算架构",
            "content": "CANN（Compute Architecture for Neural Networks）是华为针对AI场景推出的异构计算架构。作为连接上层AI框架与底层AI硬件的桥梁，CANN向下适配昇腾AI处理器，向上支持多种AI框架。它包含算子库（AOE）、图编译器（GE）、调优引擎等核心组件。开发者可以通过ACL（Ascend Computing Language）接口调用CANN的能力进行应用开发。"
        },
        {
            "title": "MindSpore AI框架",
            "content": "MindSpore是华为开源的全场景AI计算框架。它支持端、边、云全场景协同，提供自动微分、自动并行、自动调优等核心能力。MindSpore采用动静统一的编程模式，既兼顾了开发效率，又保证了运行性能。其生成的模型可以通过MindSpore Lite在手机等端侧设备上高效推理，实现了原本难以做到的端云一体化。"
        },
        {
            "title": "Atlas 800 推理服务器",
            "content": "Atlas 800 推理服务器（型号：3000）基于昇腾310处理器，最多支持8张Atlas 300推理卡，提供强大的实时推理能力。它广泛应用于智慧城市、智慧交通等场景的视频分析任务。该服务器支持风冷和液冷两种散热方式，具有高密度、低功耗的特点，单机最大INT8算力可达 2048 TOPS。"
        },
        {
            "title": "openEuler 操作系统",
            "content": "openEuler 是一个开源、免费的 Linux 发行版平台，将通过开放的社区形式与全球的开发者共同构建一个开放、多元和架构包容的软件生态体系。openEuler 完美支持 ARM64 架构（包括鲲鹏处理器），针对多核高并发场景进行了深度优化，是运行昇腾 AI 软件栈的推荐操作系统底座。"
        }
    ]
 
    total_dataset = []
    success_count = 0
    start_time = time.time()
 
    print(f"\n🚀 启动数据合成工厂 | 任务队列: {len(RAW_DOCS)} 篇文档")
    print("-" * 60)
 
    for idx, doc in enumerate(RAW_DOCS):
        print(f"[{idx+1}/{len(RAW_DOCS)}] 正在处理: 《{doc['title']}》...", end="", flush=True)
        
        loop_start = time.time()
        try:
            # 核心调用
            qa_list = generate_qa_data(doc['content'])
            
            if qa_list and isinstance(qa_list, list):
                # 简单校验数据质量
                valid_qa = [item for item in qa_list if 'question' in item and 'answer' in item]
                
                if valid_qa:
                    total_dataset.extend(valid_qa)
                    success_count += 1
                    print(f" ✅ 完成 ({time.time() - loop_start:.2f}s) | 产出 {len(valid_qa)} 条数据")
                else:
                    print(f" ⚠️ 格式错误 (JSON有效但不包含QA字段)")
            else:
                print(f" ❌ 解析失败 (模型输出非 JSON)")
                
        except Exception as e:
            print(f" ❌ 运行时异常: {e}")
 
    total_time = time.time() - start_time
 
    # ================= 4. 结果持久化与统计 =================
    
    print("-" * 60)
    print("📊 生产任务报告 (Summary)")
    print(f"Total Time  : {total_time:.2f} s")
    print(f"Avg Speed   : {total_time / len(RAW_DOCS):.2f} s/doc")
    print(f"Success Rate: {success_count}/{len(RAW_DOCS)}")
    print(f"Total QA Pairs: {len(total_dataset)}")
 
    # 保存文件
    if total_dataset:
        with open(OUTPUT_FILE, "w", encoding="utf-8") as f:
            json.dump(total_dataset, f, ensure_ascii=False, indent=4)
        print(f"\n💾 数据集已保存至: {os.path.abspath(OUTPUT_FILE)}")
    else:
        print("\n⚠️ 未生成有效数据，请检查日志。")

运行效果预期

执行 python data_factory.py，你将看到如下输出：

[
    {
        "question": "昇腾采用的是什么架构？",
        "answer": "昇腾采用达芬奇架构，集成了32个达芬奇核心。"
    },
    {
        "question": "Atlas 800T的FP16算力是多少？",
        "answer": "Atlas 800T的半精度（FP16）算力达到320 TFLOPS。"
    },
    {
        "question": "该芯片主要应用于什么场景？",
        "answer": "该芯片主要应用于云端训练场景，支持大规模分布式训练。"
    }
]
✅ 成功生成 3 条 QA 数据！

实际运行：

我们来进行打印输出内容查看效果如何：

在面对较多的数据数据处理情况下，模型处理依旧稳定，效果还不错!

这证明了在 Atlas 800T 上运行的 Llama-3 不仅具备逻辑理解能力，还能很好地遵循复杂的格式指令。

四、 部署中的问题总结与解决

在实操过程中，我们根据终端日志捕捉到了一些 Warning 信息。

1. torch_dtype 弃用警告

● 现象：FutureWarning: torch_dtype is deprecated! Use dtype instead!

● 解决：在 from_pretrained 函数中，将参数名 torch_dtype 修改为 dtype。虽然旧参数目前仍兼容，但更新代码规范更好。

2. Pad Token 缺失警告

● 现象：The attention mask and the pad token id were not set…

● 分析：Llama-3 原生 Tokenizer 没有定义 Padding Token，直接推理可能导致生成发散或报错。

● 解决：在代码中显式指定：

tokenizer.pad_token = tokenizer.eos_token
tokenizer.pad_token_id = tokenizer.eos_token_id

3. NPU 插件导入

在任何涉及模型的 Python 脚本首行，务必保留 import torch_npu。这是 PyTorch 能够识别 device=“npu” 的关键，否则会报错 RuntimeError: Expected one of cpu, cuda…。

五、 总结

该次测试，实践了在昇腾 Atlas 800T NPU 上构建本地化数据工厂的全过程，其中有几个技术要点值得关注：

1. 跨平台部署无缝衔接：通过引入 torch_npu 插件，我们以极低的代码侵入成本实现了 Llama-3 模型在国产硬件上的原生运行，验证了跨架构部署的便捷性。

2. 本地推理的鲁棒性：在数据生成的长序列推理场景中，NPU 表现出了优秀的稳定性。相比于云端 API，本地部署方案让我们能够更灵活地控制推理参数（如 Temperature、Top-P），从而在数据多样性与准确性之间找到最佳的技术平衡点。

3. 数据处理流程自动化：通过 Python 脚本将模型推理与数据清洗逻辑解耦，构建了一套可复用的 ETL（Extract, Transform, Load）流水线。这不仅验证了 NPU 处理复杂指令的能力，也为后续引入更复杂的 RAG 或微调任务打下了坚实的算力基础。