整体感受
这段代码主要实现了一个可以在网页上交互的对话机器人，用到了昇思 MindSpore 框架和 Gradio 工具。简单来说，就是让我们能通过网页和模型聊天！下面我就一段一段给大家解释。
代码逐段解读
1. 导入需要的工具库
import gradio as grimport mindsporefrom mindnlp.transformers import AutoModelForCausalLM, AutoTokenizerfrom mindnlp.transformers import TextIteratorStreamerfrom threading import Threadfrom mindnlp.peft import PeftModel
这部分就像我们做菜前要准备好各种厨具，这里导入了运行程序需要的工具库：
-
gradio as gr：用来快速搭建网页交互界面的工具，有了它不用懂复杂的网页开发也能做出漂亮的交互界面
-
mindspore：昇思深度学习框架，是运行模型的基础
-
mindnlp.transformers里的工具：用来加载大模型（AutoModelForCausalLM）和处理文本（AutoTokenizer）
-
TextIteratorStreamer：让模型生成回答时能像人说话一样一个字一个字出来，不是等全部生成完才显示
-
Thread：多线程工具，让模型生成回答时网页不会卡住
-
PeftModel：用来加载微调后的模型（这里暂时没用到）
2. 环境设置（注释掉的部分）
# activate synchronization in pynative mode# mindspore.set_context(pynative_synchronize=True)
这部分是设置 MindSpore 运行模式的，现在注释掉了，我们知道有这么个设置就行，暂时不用管它～
3. 加载模型和分词器
tokenizer = AutoTokenizer.from_pretrained("MindSpore-Lab/DeepSeek-R1-Distill-Qwen-1.5B-FP16", mirror="modelers", ms_dtype=mindspore.float16)model = AutoModelForCausalLM.from_pretrained("MindSpore-Lab/DeepSeek-R1-Distill-Qwen-1.5B-FP16", mirror="modelers", ms_dtype=mindspore.float16)# model = PeftModel.from_pretrained(model, "./output/DeepSeek-R1-Distill-Qwen-1.5B/adapter_model_for_demo/") # adapter_model path
这部分是整个程序的核心，就像我们把 “大脑” 和 “翻译官” 请出来：
-
tokenizer：相当于翻译官，负责把我们说的话转换成模型能理解的数字格式
-
model：就是大模型本身，相当于机器人的 “大脑”，负责思考并生成回答
-
括号里的参数是告诉程序从哪里下载模型，用什么数据类型运行（float16 是一种节省内存的格式）
-
最后一行注释是加载微调模型用的，我们初学者先掌握基础的就行～
4. 设置系统提示词
system_prompt = "You are a helpful and friendly chatbot"
这个很简单，就是给机器人设定一个初始身份，告诉它 “你是一个乐于助人且友好的聊天机器人”，相当于给机器人定个性格～
5. 构建对话历史格式
def build_input_from_chat_history(chat_history, msg: str): messages = [{'role': 'system', 'content': system_prompt}] for user_msg, ai_msg in chat_history: messages.append({'role': 'user', 'content': user_msg}) messages.append({'role': 'assistant', 'content': ai_msg}) messages.append({'role': 'user', 'content': msg}) return messages
这个函数的作用是整理聊天记录，把之前的对话和当前的问题按照模型要求的格式组装起来：
-
先加入系统提示词
-
然后把历史对话一条一条加进去，区分用户说的话（user）和机器人说的话（assistant）
-
最后加上当前用户的问题
-
这样模型就能理解整个对话上下文啦
6. 生成回答的函数
def predict(message, history): history_transformer_format = history + [[message, ""]] messages = build_input_from_chat_history(history, message) input_ids = tokenizer.apply_chat_template( messages, add_generation_prompt=True, return_tensors="ms", tokenize=True ) streamer = TextIteratorStreamer(tokenizer, timeout=300, skip_prompt=True, skip_special_tokens=True) generate_kwargs = dict( input_ids=input_ids, streamer=streamer, max_new_tokens=1024, do_sample=True, top_p=0.9, temperature=0.1, num_beams=1, repetition_penalty=1.2 ) t = Thread(target=model.generate, kwargs=generate_kwargs) t.start() partial_message = "" for new_token in streamer: partial_message += new_token if '</s>' in partial_message: break yield partial_message
这是处理用户输入并生成回答的核心函数，有点复杂，我们慢慢看：
-
首先把历史对话和当前问题整理好
-
调用前面的函数生成模型需要的消息格式
-
用tokenizer.apply_chat_template把消息转换成模型能处理的数字
-
streamer是用来实现流式输出的，让回答一个字一个字显示出来
-
generate_kwargs里是生成回答的各种参数：
-
max_new_tokens=1024：最多生成 1024 个词
-
temperature=0.1：温度值越小，回答越确定；越大越随机
-
-
用多线程（Thread）来生成回答，这样网页不会卡住
-
最后把生成的内容一点点返回给界面显示
7. 搭建并启动界面
gr.ChatInterface(predict, title="DeepSeek-R1-Distill-Qwen-1.5B", description="问几个问题", examples=['你是谁？', '你能做什么？'] ).launch()
这部分就是搭建网页界面了：
-
gr.ChatInterface：创建一个聊天界面
-
predict：指定用上面的 predict 函数来处理输入和生成回答
-
title：网页标题
-
description：描述文字
-
examples：提供一些示例问题
-
.launch()：启动这个网页应用，运行后我们就能在浏览器里和机器人聊天啦
整体感受
这段代码同样是实现 DeepSeek-R1-Distill-Qwen-1.5B 模型的推理，但明显针对香橙派的硬件特点做了很多优化。比如增加了静态图加速、单 token 生成计时等功能，看起来更注重在资源有限的开发板上高效运行。
代码逐段解读
1. 导入工具库
import gradio as grimport mindsporefrom mindnlp.transformers import AutoTokenizer, AutoModelForCausalLM, StaticCachefrom mindnlp.core import opsfrom mindnlp.configs import set_pyboost, ON_ORANGE_PI import timeimport numpy as np
和上一段代码相比，多了几个新朋友：
-
StaticCache：静态缓存工具，能帮模型在生成文本时节省内存
-
ops：MindNLP 的操作工具库，提供各种计算功能
-
set_pyboost和ON_ORANGE_PI：专门用于香橙派环境配置的工具
-
time：用来计算生成每个 token 的时间
-
numpy：这里用来处理概率计算，在香橙派上比原生实现更快
2. 环境与参数设置
# 在香橙派上，开启 O2 级别的 jit 优化if ON_ORANGE_PI: mindspore.set_context( enable_graph_kernel=True, mode=mindspore.GRAPH_MODE, jit_config={ "jit_level": "O2", }, )# 生成参数配置NUM_TOKENS_TO_GENERATE = 100 # 每个输入要生成的 token 数量TEMPERATURE = 0.8 # 温度参数（控制生成多样性）TOP_P = 0.8 # Top‑p 采样阈值
这部分是针对香橙派的特别优化：
-
ON_ORANGE_PI是个开关，检测到在香橙派上运行时就会启用特殊配置
-
GRAPH_MODE和jit_level": "O2"是 MindSpore 的静态图模式和二级优化，能让模型跑得更快
-
下面三个参数控制生成文本的长度和风格，数值不同生成效果也会不一样
3. 模型与分词器加载
model_id = "MindSpore-Lab/DeepSeek-R1-Distill-Qwen-1.5B-FP16"tokenizer = AutoTokenizer.from_pretrained(model_id, mirror="modelers")model = AutoModelForCausalLM.from_pretrained( model_id, ms_dtype=mindspore.float16, low_cpu_mem_usage=True, mirror="modelers",)# 使用 model.jit() 将全图静态图化model.jit()set_pyboost(False)print("[Init] 模型加载完成！")
和之前的代码相比有几个新东西：
-
low_cpu_mem_usage=True：这个参数很实用，能减少加载模型时对内存的占用，适合香橙派这种资源有限的设备
-
model.jit()：把模型转换成静态图模式，虽然启动时会慢点，但运行起来速度更快
-
set_pyboost(False)：关闭 MindSpore 的一个加速功能，可能在香橙派上更稳定
4. Top-p 采样函数
def sample_top_p(probs, p=0.9): """Top-p采样函数，用于生成文本时选择下一个token""" probs_np = probs.asnumpy() # 按概率降序排序 sorted_indices = np.argsort(-probs_np, axis=-1) sorted_probs = np.take_along_axis(probs_np, sorted_indices, axis=-1) # 计算累积概率并创建掩码 cumulative_probs = np.cumsum(sorted_probs, axis=-1) mask = cumulative_probs - sorted_probs > p sorted_probs[mask] = 0.0 sorted_probs = sorted_probs / np.sum(sorted_probs, axis=-1, keepdims=True) # 转换回 MindSpore Tensor sorted_probs_tensor = mindspore.Tensor(sorted_probs, dtype=mindspore.float32) sorted_indices_tensor = mindspore.Tensor(sorted_indices, dtype=mindspore.int32) next_token_idx = ops.multinomial(sorted_probs_tensor, 1) batch_size = probs.shape[0] batch_indices = ops.arange(0, batch_size, dtype=mindspore.int32).reshape(-1, 1) next_token = mindspore.ops.gather(sorted_indices_tensor, next_token_idx, axis=1, batch_dims=1) return next_token
这个函数是用来从模型给出的多个可能的下一个词中做选择的：
-
简单说就是先把可能的词按概率排序，然后只选那些累积概率不超过 TOP_P 的词
-
这样既能保证生成的词比较合理，又能有一定的随机性
-
注释里说用 numpy 实现是因为在香橙派上效率更高，看来是经过实际测试的
5. 解码函数
@mindspore.jit(jit_config=mindspore.JitConfig(jit_syntax_level='STRICT'))def get_decode_one_tokens_logits(model, cur_token, input_pos, cache_position, past_key_values, temperature=TEMPERATURE, top_p=TOP_P): """单个token的解码函数，返回logits，可以使用jit进行优化""" logits = model( cur_token, position_ids=input_pos, cache_position=cache_position, past_key_values=past_key_values, return_dict=False, use_cache=True )[0] return logitsdef decode_one_tokens(model, cur_token, input_pos, cache_position, past_key_values, temperature=TEMPERATURE, top_p=TOP_P): """单个token的解码函数，由logits、温度和Top_p选择合适的token""" logits = get_decode_one_tokens_logits(model, cur_token, input_pos, cache_position, past_key_values, temperature, top_p) if temperature > 0: probs = mindspore.mint.softmax(logits[:, -1] / temperature, dim=-1) new_token = sample_top_p(probs, top_p) else: new_token = mindspore.mint.argmax(logits[:, -1], dim=-1)[:, None] return new_token
这两个函数是生成单个词的核心：
-
@mindspore.jit装饰器能让函数编译成静态图，运行更快
-
get_decode_one_tokens_logits负责从模型获取每个可能的下一个词的概率
-
decode_one_tokens根据概率和前面设置的参数（温度、TOP_P）选择最合适的下一个词
6. 生成主函数
def generate_completion(prompt: str) -> str: """核心推理函数：给定 prompt，生成补齐文本并返回""" if not prompt: return "" inputs = tokenizer([prompt], return_tensors="ms", padding=True) batch_size, seq_length = inputs["input_ids"].shape # 创建静态缓存（用于加速自回归生成） past_key_values = StaticCache( config=model.config, max_batch_size=batch_size, max_cache_len=512, dtype=model.dtype ) cache_position = ops.arange(seq_length) generated_ids = ops.zeros( batch_size, seq_length + NUM_TOKENS_TO_GENERATE + 1, dtype=mindspore.int32 ) generated_ids[:, cache_position] = inputs["input_ids"].to(mindspore.int32) # 初始前向传播获取首个 logits logits = model( **inputs, cache_position=cache_position, past_key_values=past_key_values, return_dict=False, use_cache=True )[0] # 生成第一个新 token if TEMPERATURE > 0: probs = mindspore.mint.softmax(logits[:, -1] / TEMPERATURE, dim=-1) next_token = sample_top_p(probs, TOP_P) else: next_token = mindspore.mint.argmax(logits[:, -1], dim=-1)[:, None] generated_ids[:, seq_length] = next_token[:, 0] # 自回归生成循环 cache_position = mindspore.tensor([seq_length + 1]) partial_message = "" for step in range(1, NUM_TOKENS_TO_GENERATE): t_start = time.time() next_token = decode_one_tokens( model, next_token, None, cache_position, past_key_values, temperature=TEMPERATURE, top_p=TOP_P ) generated_ids[:, cache_position] = next_token.int() cache_position += 1 t_elapsed = time.time() - t_start print(f"[Token {step:02d}] {t_elapsed:.4f}s") # 打印单步生成耗时 if tokenizer.eos_token_id in next_token.int()[0]: # 遇到结束符就停止 break new_word = tokenizer.decode(next_token.int()[0], skip_special_tokens=True) partial_message += new_word yield partial_message
这个函数是整个程序的 “发动机”：
-
首先把用户输入的文字转换成模型能理解的格式
-
StaticCache是个好东西，能缓存之前计算的结果，避免重复计算，节省资源
-
然后通过循环一个一个地生成词语（自回归生成）
-
每次生成一个词都会计算耗时并打印出来，方便我们了解模型在香橙派上的运行速度
-
用yield而不是return是为了能实时看到生成的内容，而不是等全部生成完
7. Gradio 界面设置
def gradio_predict(prompt: str) -> str: """Gradio 回调：直接调用 generate_completion 并返回结果""" return generate_completion(prompt)demo = gr.Interface( fn=generate_completion, inputs=gr.Textbox(lines=4, label="输入 Prompt", placeholder="请介绍一下自己。"), outputs=gr.Textbox(lines=12, label="生成结果"), title="DeepSeek‑R1‑Distill‑Qwen‑1.5B (使用MindSpore JIT推理优化)", description=( "基于MindSpore的DeepSeek‑R1‑Distill‑Qwen‑1.5B推理。生成每个token的耗时在终端进行显示。" ), examples=[ ["请介绍一下自己。"], ["My favorite all time favorite condiment is ketchup."] ],)if __name__ == "__main__": demo.launch()
这部分和之前的代码类似，创建了一个网页界面：
-
输入框让我们输入问题或提示
-
输出框显示模型生成的回答
-
还提供了几个示例问题，方便我们快速测试
总结
这段代码和上一段相比，主要是针对香橙派这种资源有限的设备做了很多优化：
-
启用了静态图模式和 JIT 优化，提高运行速度
-
增加了内存优化参数，减少资源占用
-
详细记录了每个词的生成时间，方便调试性能
-
优化了采样和生成过程，更适合在开发板上运行