手把手教你用Qwen2.5-0.5B构建智能信息提取系统
随着大模型在自然语言处理领域的广泛应用,基于预训练语言模型的信息提取能力正成为企业智能化转型的关键技术之一。本文将带你从零开始,使用阿里开源的轻量级大模型Qwen2.5-0.5B-Instruct构建一个高效的中文命名实体识别(NER)系统,实现对地址、公司、人名等关键信息的自动化提取。
本方案适用于需要快速部署、资源消耗低但精度较高的场景,特别适合中小型企业或边缘设备上的信息抽取任务。通过全参数微调的方式,我们将该模型打造成一个结构化输出专家,能够以 JSON 格式精准返回识别结果。
1. Qwen2.5-0.5B 模型与数据准备
1.1 Qwen2.5 系列核心优势
Qwen2.5是通义千问系列最新一代大语言模型,覆盖从0.5B到720B多个规模版本,具备以下显著改进:
- 知识增强:在数学、编程等领域引入专家模型训练,大幅提升逻辑推理能力。
- 结构化输出优化:支持高质量 JSON 输出,适用于 API 接口、数据库填充等结构化场景。
- 长上下文理解:最大支持 128K tokens 上下文,适合处理长文档。
- 多语言支持:涵盖中、英、法、西、日、韩等超过 29 种语言。
- 指令遵循更强:对 system prompt 更敏感,角色扮演和条件控制更稳定。
本文选用的是轻量级版本Qwen2.5-0.5B-Instruct,专为指令理解和生成任务设计,在保持高性能的同时显著降低算力需求,可在单卡 4090D 上完成训练与推理。
📌 官方 ModelScope 地址:https://modelscope.cn/models/Qwen/Qwen2.5-0.5B-Instruct
1.2 数据集选择与预处理
我们采用中文 NLP 基准数据集CLUENER2020,包含 10 类常见实体标签:
| 实体类别 | 示例 |
|---|---|
| address | 北京、上海 |
| company | 阿里巴巴、腾讯 |
| name | 张三、李四 |
| movie | 《流浪地球》 |
| game | 英雄联盟 |
| organization | 联合国儿童基金会 |
| government | 教育部 |
| book | 《三体》 |
| position | 总经理 |
| scene | 故宫博物院 |
原始数据格式如下:
{ "text": "浙商银行企业信贷部叶老桂博士则从另一个角度对五道门槛进行了解读。", "label": { "name": {"叶老桂": [[9, 11]]}, "company": {"浙商银行": [[0, 3]]} } }由于我们关注的是实体内容而非位置信息,因此对数据进行简化转换,仅保留实体名称列表:
import json def trans(file_path, save_path): with open(save_path, "a", encoding="utf-8") as w: with open(file_path, "r", encoding="utf-8") as r: for line in r: line = json.loads(line) text = line['text'] label = {} for key, items in line['label'].items(): label[key] = list(items.keys()) # 只保留实体值 w.write(json.dumps({"text": text, "label": label}, ensure_ascii=False) + "\n") w.flush() if __name__ == '__main__': trans("ner_data_origin/train.json", "ner_data/train.json") trans("ner_data_origin/dev.json", "ner_data/val.json")转换后格式示例:
{"text": "彭小军认为,国内银行现在走的是台湾的发卡模式...", "label": {"address": ["台湾"], "name": ["彭小军"]}}1.3 Token 分布分析
为合理设置输入输出长度,我们统计了训练集的 token 分布情况:
from transformers import AutoTokenizer import json import numpy as np def get_token_distribution(file_path, tokenizer): input_tokens, output_tokens = [], [] with open(file_path, "r", encoding="utf-8") as f: for line in f: data = json.loads(line) text = data["text"] label = json.dumps(data["label"], ensure_ascii=False) input_tokens.append(len(tokenizer(text).input_ids)) output_tokens.append(len(tokenizer(label).input_ids)) return np.min(input_tokens), np.max(input_tokens), np.mean(input_tokens), \ np.min(output_tokens), np.max(output_tokens), np.mean(output_tokens)结果显示: - 输入最大 token 数:50 - 输出最大 token 数:69
据此设定:
max_source_length = 50 # 输入文本长度 max_target_length = 140 # 输出 JSON 长度(预留空间)2. 微调训练全流程实现
2.1 自定义 Dataset 构建
我们继承 PyTorch 的Dataset类,封装 CLUENER 数据,并利用apply_chat_template构造标准对话格式提示词。
# ner_dataset.py from torch.utils.data import Dataset import torch import json import numpy as np class NerDataset(Dataset): def __init__(self, data_path, tokenizer, max_source_length, max_target_length): super().__init__() self.tokenizer = tokenizer self.max_source_length = max_source_length self.max_target_length = max_target_length self.data = [] if data_path: with open(data_path, "r", encoding='utf-8') as f: for line in f: if not line.strip(): continue item = json.loads(line) self.data.append({ "text": item["text"], "label": json.dumps(item["label"], ensure_ascii=False) }) print(f"data loaded, size: {len(self.data)}") def preprocess(self, text, label): messages = [ {"role": "system", "content": "你的任务是做Ner任务提取, 根据用户输入提取出完整的实体信息, 并以JSON格式输出。"}, {"role": "user", "content": text} ] prompt = self.tokenizer.apply_chat_template( messages, tokenize=False, add_generation_prompt=True ) # 编码输入部分 instruction = self.tokenizer( prompt, add_special_tokens=False, max_length=self.max_source_length, padding="max_length", truncation=True, return_tensors="pt" ) # 编码输出部分 response = self.tokenizer( label, add_special_tokens=False, max_length=self.max_target_length, padding="max_length", truncation=True, return_tensors="pt" ) input_ids = (instruction["input_ids"][0].tolist() + response["input_ids"][0].tolist() + [self.tokenizer.eos_token_id]) attention_mask = (instruction["attention_mask"][0].tolist() + response["attention_mask"][0].tolist() + [1]) labels = ([-100] * len(instruction["input_ids"][0]) + response["input_ids"][0].tolist() + [self.tokenizer.eos_token_id]) return input_ids, attention_mask, labels def __getitem__(self, index): input_ids, attention_mask, labels = self.preprocess(**self.data[index]) return { "input_ids": torch.LongTensor(input_ids), "attention_mask": torch.LongTensor(attention_mask), "labels": torch.LongTensor(labels) } def __len__(self): return len(self.data)💡 关键点说明: - 使用
-100掩码 loss,确保只计算输出部分损失 -apply_chat_template自动生成符合 Qwen 指令格式的 prompt - 输入输出拼接成完整序列用于因果语言建模
2.2 训练脚本配置与执行
# train.py import torch from torch.utils.data import DataLoader from torch.utils.tensorboard import SummaryWriter from transformers import AutoModelForCausalLM, AutoTokenizer from ner_dataset import NerDataset from tqdm import tqdm import time, sys def train_model(model, train_loader, val_loader, optimizer, device, num_epochs, model_output_dir, writer): batch_step = 0 for epoch in range(num_epochs): time1 = time.time() model.train() for index, data in enumerate(tqdm(train_loader, file=sys.stdout, desc=f"Train Epoch: {epoch}")): input_ids = data['input_ids'].to(device) attention_mask = data['attention_mask'].to(device) labels = data['labels'].to(device) optimizer.zero_grad() outputs = model(input_ids=input_ids, attention_mask=attention_mask, labels=labels) loss = outputs.loss loss.backward() optimizer.step() writer.add_scalar('Loss/train', loss.item(), batch_step) batch_step += 1 if index % 100 == 0: time2 = time.time() print(f"Step {index}, Loss: {loss:.4f}, Time per step: {(time2-time1)/(index+1):.4f}s") # Validation model.eval() val_loss = validate_model(model, device, val_loader) writer.add_scalar('Loss/val', val_loss, epoch) print(f"Epoch {epoch}, Val Loss: {val_loss:.4f}") model.save_pretrained(model_output_dir) def validate_model(model, device, val_loader): total_loss = 0.0 with torch.no_grad(): for data in tqdm(val_loader, file=sys.stdout, desc="Validating"): input_ids = data['input_ids'].to(device) attention_mask = data['attention_mask'].to(device) labels = data['labels'].to(device) outputs = model(input_ids=input_ids, attention_mask=attention_mask, labels=labels) total_loss += outputs.loss.item() return total_loss / len(val_loader) def main(): model_name = "model/Qwen2.5-0.5B-Instruct" train_path = "ner_data/train.json" val_path = "ner_data/val.json" max_source_len = 50 max_target_len = 140 epochs = 30 batch_size = 15 lr = 1e-4 output_dir = "output_ner" log_dir = "logs" device = torch.device("cuda" if torch.cuda.is_available() else "cpu") tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True) model = AutoModelForCausalLM.from_pretrained(model_name, trust_remote_code=True) # Datasets train_set = NerDataset(train_path, tokenizer, max_source_len, max_target_len) val_set = NerDataset(val_path, tokenizer, max_source_len, max_target_len) train_loader = DataLoader(train_set, batch_size=batch_size, shuffle=True, num_workers=4) val_loader = DataLoader(val_set, batch_size=batch_size, shuffle=False, num_workers=4) writer = SummaryWriter(log_dir) optimizer = torch.optim.AdamW(model.parameters(), lr=lr) model.to(device) train_model(model, train_loader, val_loader, optimizer, device, epochs, output_dir, writer) writer.close() if __name__ == '__main__': main()训练建议:
- 使用
AdamW优化器,学习率1e-4 - Batch Size 设置为 15(根据显存调整)
- Early Stopping 可结合验证 loss 判断是否过拟合
- TensorBoard 监控训练过程:
tensorboard --logdir=logs
3. 模型测试与效果验证
训练完成后,加载微调后的模型进行推理测试。
# test.py from transformers import AutoModelForCausalLM, AutoTokenizer import torch def main(): base_model = "model/Qwen2.5-0.5B-Instruct" fine_tuned_model = "output_ner" tokenizer = AutoTokenizer.from_pretrained(base_model, trust_remote_code=True) model = AutoModelForCausalLM.from_pretrained(fine_tuned_model, trust_remote_code=True) device = torch.device("cuda" if torch.cuda.is_available() else "cpu") model.to(device) test_cases = [ "三星WCG2011北京赛区魔兽争霸3最终名次", "新华网孟买3月10日电(记者聂云)印度国防部10日说,印度政府当天批准", "证券时报记者肖渔" ] for case in test_cases: messages = [ {"role": "system", "content": "你的任务是做Ner任务提取, 根据用户输入提取出完整的实体信息, 并以JSON格式输出。"}, {"role": "user", "content": case} ] prompt = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True) inputs = tokenizer(prompt, return_tensors="pt").to(device) outputs = model.generate( **inputs, max_new_tokens=140, top_k=1, pad_token_id=tokenizer.eos_token_id ) response = tokenizer.decode(outputs[0][inputs['input_ids'].shape[-1]:], skip_special_tokens=True) print("-" * 50) print(f"Input: {case}") print(f"Output: {response}") if __name__ == '__main__': main()测试结果示例:
Input: 新华网孟买3月10日电(记者聂云)印度国防部10日说... Output: {"organization": ["新华网"], "address": ["孟买"], "position": ["记者"], "name": ["聂云"], "government": ["印度国防部"]}✅ 成功识别出媒体、地点、职位、姓名、政府部门等多个实体类型。
4. 总结
本文详细介绍了如何使用Qwen2.5-0.5B-Instruct构建一个轻量级、高可用的中文信息提取系统。整个流程包括:
- 数据预处理:将 CLUENER 数据转为 JSON 结构化输出格式;
- 模型微调:基于全参数微调策略,提升模型在特定任务上的表现;
- 高效训练:适配小模型特性,合理设置超参,避免过拟合;
- 结构化输出:利用 Qwen 对 JSON 的强生成能力,直接输出可解析结果;
- 快速部署:模型体积小,可在消费级 GPU 上运行,适合生产环境落地。
✅最佳实践建议: - 若需更高性能,可尝试 LoRA 微调进一步节省资源; - 输出后增加 JSON 校验模块,防止格式错误; - 在实际业务中加入后处理规则引擎,提升召回率。
该方案不仅可用于 NER 任务,还可扩展至关系抽取、事件识别、表单填充等多种结构化信息提取场景,是构建企业级 AI 应用的理想起点。
💡获取更多AI镜像
想探索更多AI镜像和应用场景?访问 CSDN星图镜像广场,提供丰富的预置镜像,覆盖大模型推理、图像生成、视频生成、模型微调等多个领域,支持一键部署。