Qwen2.5-1.5B开源大模型教程：自定义stop_token+生成截断策略配置-育师

Qwen2.5-1.5B开源大模型教程：自定义stop_token+生成截断策略配置

1. 引言：为什么需要控制生成停止？

当你使用Qwen2.5-1.5B这样的轻量级大模型进行对话时，可能会遇到一个常见问题：模型有时候会“说个不停”。

比如你问它“用Python写一个Hello World程序”，它可能不仅给出了代码，还开始详细解释每一行代码的含义、Python的历史、编程环境的搭建……虽然很热心，但这并不是你想要的。更糟糕的是，在生成长篇大论时，可能会耗尽预设的max_new_tokens（最大生成长度），导致回答在关键处被生硬截断，留下一句不完整的话。

这就是为什么我们需要掌握“生成停止控制”技术。通过自定义stop_token（停止词）和配置合理的截断策略，你可以告诉模型：“到这里就可以停了”，或者“如果发现它在重复自己，就主动结束”。这不仅能提升对话的精准度和用户体验，还能节省计算资源，让轻量级的1.5B模型用起来更“聪明”。

本教程将手把手教你，如何在你已经部署好的Qwen2.5-1.5B本地对话助手中，实现这两项关键配置。

2. 理解核心概念：停止词与截断

在深入代码之前，我们先花几分钟，用大白话搞清楚要对付的两个“家伙”是什么。

2.1 停止词 (Stop Token)：给模型设个“句号”

你可以把停止词想象成对话中的“暗号”。当模型在生成文本时，一旦它输出了你预设的停止词，就会立刻停止生成，就像人说话时看到了一个明确的“结束”标志。

停止词通常有两种：

字符串：比如“\n\n”（两个换行）。你可以告诉模型：“如果你连续输出了两个换行，通常意味着一个段落或思路结束，那你就可以停下来了。”这在生成格式化工整的文本（如代码、列表）时特别有用。
令牌ID (Token ID)：这是分词器将单词或字符转换成的数字。有些词，比如“<|endoftext|>”，本身就是模型训练时学到的特殊结束标记。使用它的Token ID作为停止词最为精准。

举个例子：你问：“列出三种水果。”

没有停止词：模型可能回答：“苹果、香蕉、橙子。水果富含维生素，对健康有益……（继续拓展）”
设置停止词“.”(句号)：模型在输出“苹果、香蕉、橙子。”后，看到句号，生成停止。回答干净利落。

2.2 生成截断策略：防止模型“鬼打墙”

有时候，模型会陷入循环，不断重复相似的短语或句子，比如：“这是一个好问题。这是一个好问题。这是一个好问题……” 或者，它生成的内容开始变得毫无逻辑、杂乱无章（称为“退化”）。

这时，光靠停止词可能无效，因为它并没有输出你设定的“暗号”。我们需要更主动的防御机制，这就是截断策略。Hugging Face的transformers库在调用model.generate()时，提供了诸如repetition_penalty、no_repeat_ngram_size等参数来从算法上抑制重复。而我们今天要配置的，是一种更直接的后处理策略：当检测到重复或退化时，提前终止生成。

我们将利用stopping_criteria（停止标准）功能来实现这一点。

3. 动手改造：为你的助手添加控制功能

现在，我们假设你已经成功运行了基于Streamlit的Qwen2.5-1.5B本地对话项目。我们将直接修改其核心生成部分的代码。

3.1 第一步：定义自定义停止标准

我们需要创建一个类，它能够监视模型生成过程中的令牌序列，并在满足条件时发出停止信号。

在你的项目代码中（通常是包含model.generate()函数的部分），添加以下代码块：

from transformers import StoppingCriteria, StoppingCriteriaList class CustomStoppingCriteria(StoppingCriteria): """ 自定义停止标准。 1. 检测到指定的停止词ID时停止。 2. 检测到序列重复时停止（简单示例）。 """ def __init__(self, stop_token_ids=None, max_repetition=5): super().__init__() self.stop_token_ids = stop_token_ids if stop_token_ids is not None else [] # 将停止词ID列表转换为集合便于快速查找 self.stop_token_id_set = set(self.stop_token_ids) self.max_repetition = max_repetition self.last_tokens = [] def __call__(self, input_ids, scores, **kwargs): # 获取最新生成的令牌ID current_token = input_ids[0, -1].item() # 条件1：如果当前令牌是停止词，则停止 if current_token in self.stop_token_id_set: return True # 条件2：简单重复检测（检查最近生成的令牌是否在重复） self.last_tokens.append(current_token) if len(self.last_tokens) > self.max_repetition: self.last_tokens.pop(0) # 如果最近 max_repetition 个令牌都相同，则停止 if len(set(self.last_tokens)) == 1: return True # 不停止 return False

代码解释：

这个类继承自StoppingCriteria。它的__call__方法会在模型生成每一个新令牌后被调用。
stop_token_ids: 传入一个列表，包含你想要作为停止信号的令牌ID。
max_repetition: 设置一个重复检测的窗口大小。这里是一个简化示例，当连续生成max_repetition个相同的令牌时，判定为重复循环，触发停止。
在__call__方法中，我们检查两个条件，满足任一即返回True（要求停止）。

3.2 第二步：准备停止词令牌ID

停止词需要以令牌ID的形式提供给上面的类。我们需要使用分词器来转换。

找到你加载分词器（tokenizer）的地方，在后面添加：

# 定义你想要用作停止词的字符串列表 stop_strings = ["\n\n", "。", "！", "？", "<|endoftext|>"] # 示例：双换行、中文句号、感叹号、问号及模型自带结束符 # 将字符串转换为令牌ID。注意：一个字符串可能对应多个ID，我们取第一个（或展平列表）。 stop_token_ids = [] for stop_str in stop_strings: token_ids = tokenizer.encode(stop_str, add_special_tokens=False) if token_ids: # 这里选择该停止词对应的第一个令牌ID作为停止信号 stop_token_ids.append(token_ids[0]) print(f"配置的停止词ID: {stop_token_ids}")

提示：“<|endoftext|>”通常是模型内建的结束标记，用它作为停止词非常可靠。你可以根据中文对话习惯，添加“。”、“！”等标点。

3.3 第三步：整合到模型生成调用中

现在，找到你项目中执行model.generate()的函数。它可能看起来像这样：

def generate_response(input_text, chat_history): # ... 之前的代码：用apply_chat_template格式化输入 ... inputs = tokenizer(formatted_prompt, return_tensors="pt").to(model.device) # 旧的生成参数 generation_config = { "max_new_tokens": 1024, "temperature": 0.7, "top_p": 0.9, "do_sample": True, } with torch.no_grad(): outputs = model.generate(**inputs, **generation_config) # ... 后续解码输出 ...

我们需要修改它，加入我们的停止标准：

def generate_response(input_text, chat_history): # ... 之前的代码：用apply_chat_template格式化输入 ... inputs = tokenizer(formatted_prompt, return_tensors="pt").to(model.device) # 1. 实例化我们的自定义停止标准 custom_stopper = CustomStoppingCriteria(stop_token_ids=stop_token_ids, max_repetition=5) stopping_criteria = StoppingCriteriaList([custom_stopper]) # 2. 更新生成参数，加入 stopping_criteria generation_config = { "max_new_tokens": 1024, "temperature": 0.7, "top_p": 0.9, "do_sample": True, "stopping_criteria": stopping_criteria, # 关键添加项 # 可选：同时使用内置的重复惩罚 "repetition_penalty": 1.1, } with torch.no_grad(): outputs = model.generate(**inputs, **generation_config) # 3. 解码时，主动移除停止词（避免它出现在最终回复里） generated_tokens = outputs[0, inputs['input_ids'].shape[1]:] # 获取新生成的部分 # 如果最后一个令牌是停止词，则去掉它 if generated_tokens[-1].item() in stop_token_ids: generated_tokens = generated_tokens[:-1] response = tokenizer.decode(generated_tokens, skip_special_tokens=True) # ... 返回 response ...

关键改动说明：

创建停止标准列表：将我们定义的CustomStoppingCriteria实例放入StoppingCriteriaList。
注入参数：将stopping_criteria加入到generation_config中。
后处理：解码后，检查生成的最后一个令牌是否是停止词，如果是则将其剔除，确保回复内容干净。

3.4 第四步：测试与调整

完成代码修改后，重启你的Streamlit应用。

进行如下测试：

测试停止词：问一个事实性问题，如“中国的首都是哪里？”。观察模型在输出“北京。”后是否立即停止，而不是继续补充地理知识。
测试列表生成：输入“请写出三个Python关键字。”。检查模型在列出三个关键字（很可能以换行或句号分隔）后是否停止。
测试截断（可能需要诱导）：尝试问一个非常开放、容易导致重复的问题，或者故意在对话历史中制造循环。观察max_repetition机制是否生效。

调整技巧：

停止词不生效？检查stop_token_ids是否正确获取。用tokenizer.decode([token_id])反查一下ID对应的到底是什么词。
回答被过早截断？可能是停止词设置得太常见。例如，如果“。”作为停止词，那么模型在回答中任何一句结束时都可能停止。可以考虑使用更独特的标记组合，如“\n\n”或主要依赖“<|endoftext|>”。
重复检测太敏感/不敏感？调整max_repetition参数。也可以在我们的CustomStoppingCriteria中实现更复杂的重复检测逻辑，比如检测重复的n-gram（词组）。