案例9：边缘智能聊天机器人

1. 项目简介

本项目基于昇腾310B平台，构建一个可以在边缘设备上运行的智能聊天机器人。系统集成了文本嵌入、RAG（检索增强生成）、语音交互等AI技术，能够在离线环境下回答专业问题，也可接入云端大语言模型提升回复质量。

与云端聊天机器人相比，边缘端部署具有低延迟、隐私保护、离线可用三大优势。本项目的核心设计思想是：不强行在昇腾310B上运行大语言模型（这在实际硬件条件下既困难又低效），而是将NPU用于它最擅长的事情——文本嵌入的快速推理，再通过向量检索和模板/云端LLM两级策略生成回复。

项目的源代码可以从这里下载。

2. 内容大纲

2.1. 硬件准备

• 核心计算单元: 昇腾310B开发者套件
• 音频设备:
  • USB麦克风或USB耳机（语音输入）
  • USB音响或3.5mm耳机（语音输出）
  • 也可直接使用电脑自带麦克风和扬声器
• 网络连接:
  • 以太网或WiFi（仅云端LLM模式需要）
• 可选外设:
  • 触摸屏显示器（用于独立运行时的交互）
  • 键盘鼠标（开发调试用）

系统硬件架构

相比原方案中动辄列出"4麦克风阵列、NVMe SSD、触摸屏、锂电池组"等大量外围硬件，本项目的硬件需求非常简洁：一块昇腾310B开发板、一个USB麦克风和音响（或一个USB耳机），足以运行完整的聊天机器人系统。没有昇腾设备也可以——嵌入模型会自动回退到CPU推理。

2.2. 软件环境

• 操作系统: Ubuntu 20.04 / 22.04 LTS
• CANN版本: 7.0.RC1 或以上（仅NPU推理需要）
• Python版本: 3.9 或以上
• 核心依赖:
  • gradio — Web聊天界面
  • sentence-transformers — 嵌入模型（CPU推理 & 模型参考）
  • faiss-cpu — 向量相似度搜索
  • jieba — 中文分词（文本分块）
  • numpy — 数值计算
• 语音交互（可选）:
  • SpeechRecognition — 语音识别（调用Google Web Speech API）
  • pyaudio — 麦克风音频采集
  • pyttsx3 — 语音合成（espeak后端）
• 模型准备（仅开发/转换阶段）:
  • torch + transformers — ONNX模型导出
  • optimum — HuggingFace模型导出工具（推荐）
• 云端LLM（可选）:
  • requests — HTTP调用OpenAI兼容API

环境配置脚本 (setup.sh)

#!/bin/bash
set -e

echo "=== Ascend 310B 智能聊天机器人 - 环境安装 ==="

# 1. 系统依赖
sudo apt update
sudo apt install -y python3-dev python3-pip portaudio19-dev espeak

# 2. Python包
pip3 install gradio sentence-transformers faiss-cpu jieba numpy<2.0
pip3 install SpeechRecognition pyaudio pyttsx3

# 3. 模型准备（下载 + ONNX导出 + ATC转换）
python3 prepare_models.py

echo "安装完成！启动: python3 app.py"

与旧方案的关键区别：

• 不再安装 PyTorch / transformers / accelerate / peft 作为运行时依赖——这些加起来超过 5GB，对于嵌入模型来说完全不需要
• 不再安装 Redis / FastAPI / uvicorn / websockets——Gradio 一个库覆盖了 Web 界面和 API
• 语音库标注为可选，没有麦克风也不影响核心功能

2.3. 系统架构设计

本节是理解整个项目设计思路的核心。在动手写代码之前，需要先回答一个关键问题：昇腾310B到底能不能跑大语言模型？

2.3.1 为什么不在NPU上直接跑LLM

昇腾310B的NPU通过OM（Offline Model）格式执行推理。OM是一个静态计算图——输入张量的形状和数据类型在模型转换时就固定了，运行时不可改变。

而大语言模型的生成过程是自回归的：每生成一个token，要把它拼回输入序列，再跑一次模型。序列长度在逐token增长，输入形状在不断变化。这与OM的静态图假设根本矛盾。

加上KV-cache管理的复杂性、7B模型至少14GB的FP16内存需求（而昇腾310B通常只有4-8GB），直接部署LLM不切实际。

但这不意味着NPU在NLP任务中没用。恰恰相反——文本嵌入（Text Embedding）是NPU的"甜点区"：

• 固定输入形状（256 tokens in，384维向量 out）
• 纯前向推理，一次执行完成
• 模型小（all-MiniLM-L6-v2 约90MB），轻松装入NPU内存
• 推理速度快（NPU上约10ms/条 vs CPU 50-80ms/条）

2.3.2 三层架构设计

基于上述分析，本项目采用三层混合架构，完整程序流程如下：

每一层的职责和设计考量：

层	运行位置	为什么放在这里
语音I/O	CPU	音频采集/播放是操作系统层面的事，与NPU无关
文本嵌入	NPU（主）/ CPU（回退）	NPU擅长矩阵运算，嵌入模型正好是纯矩阵计算；CPU作为无NPU时的保障
FAISS检索	CPU	向量搜索是索引结构的遍历，非矩阵运算，CPU+C++优化就足够快（<1ms）
回复生成	CPU	模板匹配在CPU上是O(1)；云端API走网络，与NPU无关

2.3.3 与纯云端/纯边缘方案的对比

维度	纯云端	纯边缘LLM	本项目（混合）
延迟	网络延迟 0.5-3s	NPU不可行，CPU上极慢	嵌入<50ms + 检索<1ms + 回复<10ms（模板）
隐私	数据上传云端	完全本地	完全本地（离线模式）
可用性	依赖网络	始终可用	始终可用（离线模式）
回复质量	高（大模型）	取决于模型大小	模板精准/云端高质量
硬件要求	无需NPU	需大内存+GPU	升腾310B或普通CPU
维护成本	API费用	模型更新复杂	只更新知识库文本

2.4. 文本嵌入模型与昇腾部署

本节详细介绍如何在昇腾310B上部署文本嵌入模型，这是整个RAG管道的基石。

2.4.1 什么是文本嵌入

文本嵌入（Text Embedding）将一段自然语言文本转换为固定长度的浮点数向量。其核心性质是：语义相近的文本，在向量空间中距离更近。

"昇腾310B是一款边缘AI芯片"  ->  [0.12, -0.34, 0.08, ..., 0.21]  (384维)
"华为Ascend 310B是边缘推理处理器" -> [0.13, -0.31, 0.06, ..., 0.19]  ← 余弦相似度 约等于 0.92
"今天天气很好适合出去玩"  -> [-0.45, 0.28, 0.73, ..., -0.55]  ← 余弦相似度 约等于 0.03

有了这个性质，"找到知识库里和用户问题最相关的内容"就变成了数学上的向量内积运算。

2.4.2 为什么选择 all-MiniLM-L6-v2

候选模型	参数量	向量维度	模型大小	推理速度 (CPU)
all-MiniLM-L6-v2	22M	384	~90 MB	~10ms
all-mpnet-base-v2	110M	768	~420 MB	~50ms
bge-large-zh-v1.5	326M	1024	~1.3 GB	~150ms
text2vec-large-chinese	326M	1024	~1.3 GB	~150ms

all-MiniLM-L6-v2 是由 Microsoft 的 Sentence-Transformers 团队发布的轻量级嵌入模型。它用知识蒸馏技术将 BERT 的 12 层压缩到 6 层，保留了原始模型约 95% 的嵌入质量，但体积缩小为原来的 1/3。

选择它的原因：

• 规格适中：90MB 的 ONNX 模型对昇腾310B的内存没有任何压力
• 固定输入输出：最大 256 tokens 输入，384 维向量输出，完美匹配 OM 静态图
• 多语言支持：虽然以英文为主训练，但对中文的嵌入质量在轻量模型中排名靠前
• 生态成熟：HuggingFace 直接支持，optimum 库可一行代码导出 ONNX

2.4.3 模型转换管线

ONNX 导出代码（prepare_models.py 核心逻辑）：

from transformers import AutoTokenizer, AutoModel
import torch

tokenizer = AutoTokenizer.from_pretrained("sentence-transformers/all-MiniLM-L6-v2")
model = AutoModel.from_pretrained("sentence-transformers/all-MiniLM-L6-v2")
model.eval()

# 准备一个dummy输入来确定动态轴
dummy = tokenizer("Hello world", padding="max_length",
                   truncation=True, max_length=256, return_tensors="pt")

torch.onnx.export(
    model,
    (dummy["input_ids"], dummy["attention_mask"],
     dummy.get("token_type_ids", torch.zeros_like(dummy["input_ids"]))),
    "models/embedding_model.onnx",
    input_names=["input_ids", "attention_mask", "token_type_ids"],
    output_names=["sentence_embedding"],
    dynamic_axes={
        "input_ids": {0: "batch_size"},
        "attention_mask": {0: "batch_size"},
        "token_type_ids": {0: "batch_size"},
        "sentence_embedding": {0: "batch_size"},
    },
    opset_version=14,
)

ATC 转换命令：

atc --model=models/embedding_model.onnx \
    --framework=5 \
    --output=models/embedding_model \
    --input_shape="input_ids:1,256;attention_mask:1,256;token_type_ids:1,256" \
    --soc_version=Ascend310B4

参数说明：

• --framework=5：5 代表 ONNX 格式
• --input_shape：明确指定每个输入的维度，避免 ATC 推导失败
• --soc_version：根据实际芯片选择，npu-smi info 可查看

转换完成后，models/ 目录下会生成 embedding_model.om（约 45MB，FP16），这就是昇腾 NPU 可以直接执行的格式。

2.4.4 NPU 推理封装

昇腾的 Python API（PyACL）提供了底层的设备管理、内存拷贝和模型执行接口。ascend_inference.py 沿用了案例1中成熟的 AscendSystem / AscendModel 模式。

AscendSystem — 管理 NPU 设备生命周期：

class AscendSystem:
    def __init__(self, device_id=0):
        ret = acl.init()                          # 初始化 ACL 运行时
        ret = acl.rt.set_device(device_id)         # 选择 NPU 设备
        self.context, ret = acl.rt.create_context(device_id)  # 创建执行上下文
        self.stream, ret = acl.rt.create_stream()  # 创建任务流

初始化顺序是固定的：acl.init -> set_device -> create_context -> create_stream。释放时严格反向：destroy_stream -> destroy_context -> reset_device -> finalize。

AscendModel — 加载和执行 OM 模型：

class AscendModel:
    def _load_model(self):
        self.model_id, ret = acl.mdl.load_from_file(self.model_path)  # 加载 OM
        self.desc = acl.mdl.create_desc()         # 获取模型描述
        acl.mdl.get_desc(self.desc, self.model_id)
        self._init_buffers()                       # 预分配输入/输出内存

    def execute(self, input_data_list):
        # 1. Host -> Device: 将 numpy 数组拷贝到 NPU 内存
        for i, data in enumerate(input_data_list):
            acl.rt.memcpy(dev_ptr, size, host_ptr, size, 1)  # 1 = H2D

        # 2. 执行推理
        acl.mdl.execute(self.model_id, self.input_dataset, self.output_dataset)

        # 3. Device -> Host: 将结果拷回主机
        for i in range(len(self.output_buffers)):
            acl.rt.memcpy(host_ptr, size, dev_ptr, size, 2)  # 2 = D2H

EmbeddingModel — 业务层封装，整合 tokenizer 和 NPU 推理：

class EmbeddingModel:
    def encode(self, texts):
        """输入文本列表，返回归一化的嵌入向量 (N, 384)"""
        if self._use_npu:
            return self._encode_npu(texts)
        else:
            return self._encode_cpu(texts)  # 自动回退

    def _encode_npu(self, texts):
        for text in texts:
            encoded = tokenizer(text, padding="max_length",
                                truncation=True, max_length=256,
                                return_tensors="np")
            outputs = self._ascend_model.execute([
                encoded["input_ids"].astype(np.int64),
                encoded["attention_mask"].astype(np.int64),
                encoded["token_type_ids"].astype(np.int64),
            ])
            embedding = mean_pool(outputs[0], attention_mask)
            embeddings = L2_normalize(embeddings)
        return embeddings

需要特别注意的细节：

1. 数据类型：PyACL 要求 int64 输入。numpy 默认的 int32 会导致数据错位。
2. 内存对齐：np.ascontiguousarray() 确保 C-contiguous 内存布局，否则 acl.rt.memcpy 会失败。
3. Mean Pooling：Transformer 输出是每个 token 的隐藏状态，需要对其求均值得到句子级向量。注意要用 attention_mask 屏蔽 padding token。

def _pool_output(self, outputs, attention_mask):
    # outputs: [batch, seq_len, hidden_size]
    # mask:    [batch, seq_len]
    mask = np.expand_dims(attention_mask.astype(np.float32), axis=-1)
    summed = (token_embeddings * mask).sum(axis=1)   # 有效token求和
    counts = mask.sum(axis=1).clip(min=1)             # 有效token数量
    return summed / counts                             # 均值

4. L2 归一化：将向量归一化到单位球面上。归一化后，内积即等价于余弦相似度，这是 FAISS IndexFlatIP 的工作前提。

@staticmethod
def _normalize(embeddings):
    norms = np.linalg.norm(embeddings, axis=-1, keepdims=True)
    norms = np.clip(norms, 1e-12, None)   # 防止除零
    return embeddings / norms

2.5. 知识库与向量检索

有了嵌入模型，下一步是构建知识库并实现高效的向量检索。

2.5.1 RAG 的核心思想

RAG（Retrieval-Augmented Generation，检索增强生成）的工作流程：

RAG 解决的核心问题：让 AI 的回答有据可查。没有 RAG 时，模型要么凭空编造（大模型的"幻觉"），要么只能靠训练数据中记住的知识。有了 RAG 后，回答可以被溯源到知识库中的具体文档片段。

2.5.2 中文文本分块策略

知识库文档需要被切成合适大小的"块"（chunk）。块太大，检索精度下降（一个块里混入太多不相关内容）；块太小，语义不完整。

knowledge_base.py 中的分块策略考虑了中文的特点：

def _split_text(self, text):
    paragraphs = [p.strip() for p in text.split("\n") if p.strip()]
    chunks = []
    for para in paragraphs:
        if len(para) <= self._chunk_size:      # 短段落直接作为一块
            chunks.append(para)
            continue
        sentences = self._split_sentences(para)  # 按句号/问号/感叹号切分
        current = ""
        for sent in sentences:
            if len(current) + len(sent) <= self._chunk_size:
                current += sent
            else:
                chunks.append(current)
                # 重叠：保留上一块的末尾，避免语义断裂
                current = current[-self._chunk_overlap:] + sent
        if current:
            chunks.append(current)
    return chunks

中文分句不使用 jieba 分词，而是通过正则表达式匹配句末标点（。！？；）。这比用分词器更轻量，且对中文段落的分句准确度足够。

2.5.3 FAISS 向量索引

FAISS（Facebook AI Similarity Search）是 Meta 开源的向量相似度搜索库。核心操作极其简单：

import faiss
import numpy as np

# 创建索引
dim = 384
index = faiss.IndexFlatIP(dim)         # 内积索引

# 添加向量
embeddings = model.encode(documents)   # shape: (N, 384)
index.add(embeddings)                  # 加入索引

# 搜索
query_vec = model.encode(["昇腾310B的算力是多少？"])
scores, indices = index.search(query_vec, k=3)  # 返回最相似的3个

IndexFlatIP 的含义：

• Flat：暴力搜索，不做任何近似。对 N 个文档、384 维向量，复杂度 O(N×384)。
• IP：Inner Product（内积）。配合 L2 归一化的向量，内积等价于余弦相似度。

对于几千到几万条文档的规模，暴力搜索完全够用（<1ms）。当知识库扩大到百万级别时，FAISS 提供了 IVF（倒排索引）、HNSW（图索引）等近似搜索方案，只需改一行代码即可切换。

2.5.4 知识库管理

KnowledgeBase 类提供了完整的增删查改和持久化：

kb = KnowledgeBase(embedding_model)

# 添加知识
kb.add_texts(["昇腾310B支持FP16和INT8推理..."], [{"source": "manual"}])
kb.add_document("path/to/knowledge.txt")  # 自动分块 + 编码 + 入库

# 检索
results = kb.search("什么是CANN？", k=3)
# [{"text": "...", "score": 0.87, "metadata": {...}}, ...]

# 持久化
kb.save("data/index.faiss", "data/documents.json")
kb.load("data/index.faiss", "data/documents.json")

2.6. 对话管理与响应生成

2.6.1 对话状态机

DialogueManager 维护一个简单的四状态机：

状态转换规则：

• 首次交互自动进入 GREETING（根据时间段返回不同问候语）
• 检测到"再见/拜拜/bye"等关键词进入 FAREWELL
• 其他情况保持在 ACTIVE 状态，走完整的 RAG 处理流程

2.6.2 两级回复生成

默认模式：模板 + RAG 检索上下文

查询经过 RAG 检索后，将匹配到的知识片段直接组织成回复：

def _generate_template(self, query, context):
    if context:
        pieces = ["根据我的知识库，以下是相关内容：\n"]
        for i, item in enumerate(context, 1):
            pieces.append(f"{i}. {item['text']}\n")
        pieces.append("\n请问还有什么想了解的吗？")
        return "".join(pieces)
    return "抱歉，我在知识库中没有找到相关的信息..."

这种方式虽然简单，但结合了 RAG 的检索能力——回复中的每一条信息都直接来自知识库，零幻觉。对于昇腾310B、CANN、边缘计算等专业问题，FAQ + 知识库的模板回复比大模型凭空生成更可靠。

云端增强模式：LLM + RAG

当用户配置了云端 API Key，同样的检索结果会作为 system prompt 注入 LLM：

def _generate_cloud(self, query, context, history):
    system_prompt = (
        "你是一个运行在昇腾310B边缘设备上的AI助手。"
        "请基于提供的知识库内容回答用户问题。"
        "如果知识库中没有相关信息，请诚实告知，不要编造。"
    )
    # 将检索到的文档拼接为上下文
    context_block = "\n".join(f"[{i}] {item['text']}"
                              for i, item in enumerate(context, 1))
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "system", "content": f"相关知识库内容：{context_block}"},
        {"role": "user", "content": query},
    ]
    # 调用 OpenAI 兼容 API
    resp = requests.post(endpoint, headers={...}, json={
        "model": "gpt-3.5-turbo",
        "messages": messages,
        "max_tokens": 512,
        "temperature": 0.7,
    })

兼容所有使用 OpenAI API 格式的服务：

• 云端：OpenAI、Azure OpenAI、通义千问、DeepSeek
• 本地：Ollama、vLLM、llama.cpp server、LocalAI

2.6.3 对话历史管理

ConversationHistory 是一个固定容量的环形缓冲区：

class ConversationHistory:
    def __init__(self, max_turns=10):
        self._turns = []
        self._max = max_turns

    def add(self, role, text):
        self._turns.append(Turn(role=role, text=text, timestamp=time.time()))
        if len(self._turns) > self._max:
            self._turns.pop(0)  # 超出容量时丢弃最早的

    def get_recent(self, n=4):
        return self._turns[-n:]   # 获取最近n轮

保留最近 10 轮对话，云端模式下将最近 4 轮作为上下文发送给 LLM。这样做既维持了多轮对话的连贯性，又控制了 API 调用的 token 消耗。

2.7. 语音交互系统

语音交互采用浏览器端录音 + 服务端识别的架构。Gradio 的 Audio 组件负责在浏览器中通过 MediaRecorder API 录制音频，然后将 WAV 数据发送到服务端。服务端的 SpeechRecognizer 接收音频文件，调用 Google Web Speech API 完成语音识别。

2.7.1 语音识别

class SpeechRecognizer:
    def recognize_from_file(self, audio_path):
        recognizer = sr.Recognizer()
        with sr.AudioFile(audio_path) as source:
            audio = recognizer.record(source)
        return recognizer.recognize_google(audio, language="zh-CN")

Google Web Speech API 的优势是免费、无需注册、中英文识别准确率不错。局限是需要网络连接，且可能有频率限制。在生产环境中可以替换为：

• Vosk — 离线、支持中文的小型识别模型
• Whisper.cpp — OpenAI Whisper 的 C++ 移植，在 CPU 上可运行
• FunASR — 阿里达摩院出品，中文识别效果优秀

2.7.2 语音合成

class TextToSpeech:
    def speak(self, text):
        def _run():
            engine = pyttsx3.init()
            engine.setProperty("rate", 160)    # 语速
            engine.setProperty("volume", 0.8)  # 音量
            engine.say(text)
            engine.runAndWait()
        threading.Thread(target=_run, daemon=True).start()

pyttsx3 在 Linux 上使用 espeak 作为后端，离线可用、零成本。缺点是中文发音偏机械，自然度不及商业 TTS 服务。如需更好的中文语音效果，可替换为 Piper TTS（开源、离线、中文音色更自然）。

2.7.3 浏览器端录音

Gradio 的 gr.Audio(sources=["microphone"], type="numpy") 返回 (sample_rate, audio_array)，其中 audio_array 是 float32 numpy 数组。需要在服务端将其转换为 WAV 格式，因为 speech_recognition 只接受文件路径：

def voice_input_fn(audio):
    sr_val, audio_data = audio
    audio_int16 = (audio_data * 32767).astype(np.int16)

    # 写入临时 WAV 文件
    with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp:
        with wave.open(tmp, "wb") as wf:
            wf.setnchannels(1)
            wf.setsampwidth(2)
            wf.setframerate(sr_val)
            wf.writeframes(audio_int16.tobytes())
        tmp_path = tmp.name

    text = asr.recognize_from_file(tmp_path)
    os.unlink(tmp_path)
    return text, text  # 返回识别结果，自动填入聊天输入框

2.8. Web界面与API服务

本项目使用 Gradio 构建 Web 界面。Gradio 是 HuggingFace 出品的 Python 库，专为机器学习模型演示而设计。与 Flask/FastAPI 相比，Gradio 的优势在于：不需要写 HTML/CSS/JS 代码，Python 类定义直接映射为 Web 组件，内置 WebSocket 连接管理、队列系统和错误处理。

2.8.1 界面设计

with gr.Blocks(theme=gr.themes.Soft(), title="Ascend 310B 智能聊天机器人") as demo:
    with gr.Tabs():
        with gr.TabItem(" 对话"):
            chatbot = gr.Chatbot(height=500)
            msg_box = gr.Textbox(label="输入消息")
            send_btn = gr.Button("发送", variant="primary")
            audio_in = gr.Audio(sources=["microphone"], type="numpy")

            # 文本对话
            msg_box.submit(chat_fn, [msg_box, chatbot], [msg_box, chatbot])
            send_btn.click(chat_fn, [msg_box, chatbot], [msg_box, chatbot])

            # 语音输入
            audio_in.stop_recording(voice_input_fn, audio_in, [status, msg_box])

        with gr.TabItem(" 设置"):
            cloud_toggle = gr.Checkbox(label="启用云端 LLM")
            ...

gr.Chatbot 组件内置了聊天气泡样式和自动滚动，交互逻辑只需关注 handler 函数的输入输出映射。

2.8.2 事件流

Gradio 的事件驱动模型：

2.9. 用户手册

2.9.1 系统部署

1. 克隆代码：进入 samples/case9/ 目录
2. 安装环境：bash setup.sh
3. 准备模型：python3 prepare_models.py（自动下载、导出ONNX、转换为OM）
4. 启动服务：python3 app.py
5. 访问界面：打开浏览器访问 http://127.0.0.1:7860

如果没有昇腾设备，跳过第3步的 ATC 转换即可。系统会自动使用 CPU 模式运行，所有功能不受影响。

2.9.2 配置说明

配置项	位置	说明
嵌入模型	config.py	默认 all-MiniLM-L6-v2，可换其他 sentence-transformers 模型
检索数量	config.py: TOP_K_RETRIEVAL	默认 3，增大可提供更多上下文
相似度阈值	config.py: SIMILARITY_THRESHOLD	默认 0.3，提高可过滤不相关结果
云端 API	设置面板	支持任何 OpenAI 兼容接口（Ollama/vLLM/通义千问等）
语音开关	设置面板	关闭后仅文本交互

2.9.3 使用指南

• 文本对话：直接在输入框打字，回车发送
• 语音输入：点击 语音 按钮开始录音，再次点击停止，自动识别并发送
• 知识问答：询问昇腾310B、CANN、边缘计算、RAG 等相关问题，系统会从知识库检索后回答
• 云端增强：在设置面板填入 API Key 并启用，回复质量大幅提升
• 扩展知识库：编辑 data/sample_knowledge.txt 添加自己的知识条目，重启服务即可生效

2.9.4 维护与扩展

• 添加自定义知识：在 data/ 目录下创建文本文件，修改 app.py 中 get_knowledge_base() 的加载逻辑
• 切换嵌入模型：修改 config.py 中的 MODEL_NAME，重新运行 prepare_models.py
• 自定义回复模板：编辑 data/sample_faq.json 增加 FAQ 问答对
• 接入离线 ASR：将 voice_io.py 中的 recognize_google 替换为 Vosk 或 Whisper
• 日志监控：终端输出包含请求日志和检索上下文，方便调试

3. 源代码结构

samples/case9/
├── app.py                   # Gradio Web 界面入口，事件绑定
├── ascend_inference.py      # Ascend NPU 推理封装
│   ├── AscendSystem         #   NPU 设备初始化 / 释放
│   ├── AscendModel          #   OM 模型加载 / H2D / execute / D2H
│   ├── EmbeddingModel       #   文本嵌入业务层（tokenize -> NPU/CPU -> 归一化）
│   └── create_embedding_model()  # 工厂函数，自动尝试 NPU -> 回退 CPU
├── knowledge_base.py        # RAG 检索引擎
│   ├── Document             #   文档数据类
│   └── KnowledgeBase        #   FAISS 索引管理 / 分块 / 检索 / 持久化
├── dialogue.py              # 对话管理器
│   ├── State                #   对话状态枚举
│   ├── ConversationHistory  #   环形缓冲对话历史
│   └── DialogueManager      #   意图检测 -> RAG -> 模板/API 回复
├── voice_io.py              # 语音交互
│   ├── SpeechRecognizer     #   语音识别（Google API / 文件）
│   └── TextToSpeech         #   语音合成（pyttsx3 / espeak）
├── config.py                # 全局配置常量
├── prepare_models.py        # 模型准备（ONNX 导出 + ATC 转换）
├── setup.sh                 # 一键环境安装
├── requirements.txt         # Python 依赖清单
├── data/
│   ├── sample_knowledge.txt # 示例知识库（约 20 条昇腾/边缘计算知识）
│   └── sample_faq.json      # 示例 FAQ 问答对（15 组模式匹配）
├── models/                  # 模型文件目录
└── README.md                # 快速开始指南

各模块的调用关系：

4. 效果演示

基础对话 — 离线模板模式

用户: 什么是昇腾310B？
机器人: 根据我的知识库，以下是相关内容：
  1. 昇腾310B是华为推出的面向边缘计算场景的AI推理处理器...
  2. 昇腾310B支持FP16和INT8两种计算精度...
  请问还有什么想了解的吗？

RAG 检索 — 专业问题

用户: CANN的ATC工具怎么用？
机器人: 根据我的知识库，以下是相关内容：
  1. ATC（Ascend Tensor Compiler）是昇腾的模型转换工具...
  2. 使用ATC进行模型转换的基本流程是：首先将训练好的模型导出为ONNX格式...
  请问还有什么想了解的吗？

语音交互

用户: （点击录音按钮）"什么是边缘计算？"
机器人: （识别文字 -> RAG检索 -> 返回回复 -> TTS朗读）
       "根据我的知识库，边缘计算是一种将计算和数据存储从云端推到..."

性能指标

指标	CPU 模式	NPU 模式	说明
嵌入延迟 (单条)	50-80ms	10-15ms	all-MiniLM-L6-v2, 256 tokens
嵌入延迟 (批量32条)	200ms	80ms	知识库批量入库
FAISS 检索 (1000条)	< 1ms	< 1ms	IndexFlatIP
端到端响应 (模板)	~100ms	~30ms	不含语音
端到端响应 (云端LLM)	1-3s	1-3s	取决于网络和API
语音识别延迟	1-2s	1-2s	Google Web Speech API

关键观察：NPU 对嵌入计算有约 4-5x 的加速，这对批量入库知识库时效果明显。端到端延迟的主要瓶颈在云端 API 和语音识别服务，而非本地推理。这也印证了架构设计的合理性——NPU 被用在它最擅长的密集矩阵计算上，而系统的其他部分不受 NPU 限制。