用本地模型做文本分类与信息提取
2026/6/20 13:52:58 网站建设 项目流程

摘要:文本分类和信息提取是 NLP 最基础也最实用的两个任务——自动判断客户投诉的紧急程度、从简历中提取关键信息、对新闻按主题归类……在传统方法中,你需要针对每个任务训练专门的模型。而用大语言模型(LLM),你只需要写一段提示词就能完成所有这些任务。这篇文章基于本地 Qwen3 模型,演示如何用它完成 5 种分类任务和 3 种信息提取任务,全部代码可运行。


一、为什么用 LLM 做文本分类?

传统方法 vs LLM 方法

对比传统 ML 分类LLM 分类
开发周期数天到数周(标注+训练+调参)数分钟(写提示词)
标注数据需要大量标注样本不需要或少样本
类别变更需要重新训练改提示词即可
计算成本训练需要 GPU推理即可
可解释性特征权重能说出推理过程

适用场景

LLM 分类最适用的场景: ✅ 类别经常变化的场景(如:紧急程度分级规则每月调整) ✅ 没有标注数据的冷启动阶段 ✅ 需要解释"为什么分到这一类" ✅ 多标签分类(一个样本同时属于多个类) LLM 分类不如传统方法的场景: ❌ 每秒需要处理上万条的超高吞吐场景 ❌ 类别固定、数据量大的成熟业务(如:垃圾邮件过滤)

二、基础配置

加载本地模型

from transformers import AutoModelForCausalLM, AutoTokenizer import json import re # 加载 Qwen3-0.6B(本地路径) MODEL_PATH = "d:/ai/models/Qwen3-0.6B" tokenizer = AutoTokenizer.from_pretrained(MODEL_PATH, trust_remote_code=True) model = AutoModelForCausalLM.from_pretrained( MODEL_PATH, trust_remote_code=True, device_map="auto" ) def llm_classify(prompt, max_new_tokens=100, temperature=0.1): """ 统一推理函数 参数: prompt: 输入提示词 temperature: 分类任务用低 temperature(0.1)确保稳定性 """ messages = [{"role": "user", "content": prompt}] text = tokenizer.apply_chat_template( messages, tokenize=False, add_generation_prompt=True ) inputs = tokenizer(text, return_tensors="pt").to(model.device) outputs = model.generate( **inputs, max_new_tokens=max_new_tokens, temperature=temperature, do_sample=True, pad_token_id=tokenizer.eos_token_id, ) response = tokenizer.decode( outputs[0][inputs.input_ids.shape[1]:], skip_special_tokens=True ).strip() return response

三、五种分类任务实战

任务 1:二分类(情感分析)

def sentiment_analysis(text): """情感分析:判断评论是正面还是负面""" prompt = f"""请判断以下评论的情感倾向。 评论:{text} 情感(正面/负面):""" result = llm_classify(prompt) return result # 测试 test_reviews = [ "这家餐厅的菜品非常棒,服务也很周到!", "等了四十分钟才上菜,而且菜是凉的。", "环境还可以,但价格偏贵。", ] for review in test_reviews: result = sentiment_analysis(review) print(f"「{review[:20]}...」 → {result}")

输出

「这家餐厅的菜品非常棒...」 → 正面 「等了四十分钟才上菜...」 → 负面 「环境还可以,但价格偏贵...」 → 负面(如果非常明确才是正面)

任务 2:多分类(紧急程度分级)

def urgency_classification(text): """紧急程度分级:高/中/低""" prompt = f"""你是一个客服工单分类系统。请判断以下客户反馈的紧急程度。 分级标准: - 高(紧急):涉及安全问题、系统崩溃、资金损失、人身伤害、严重投诉 - 中(普通):功能异常、使用困难、需要人工介入 - 低(咨询):一般咨询、产品介绍、建议、非紧急问题 反馈:{text} 紧急程度(高/中/低):""" return llm_classify(prompt) # 测试 cases = [ "我的账户被不明人士登录了,里面的钱不见了!", "请问你们的产品支持Mac系统吗?", "导出功能报错,显示'系统繁忙',已经重试了三次都失败。", ] for case in cases: result = urgency_classification(case) print(f"紧急程度: {result} ← {case[:20]}")

输出

紧急程度: 高 ← 我的账户被不明人士登录了... 紧急程度: 低 ← 请问你们的产品支持Mac系统吗? 紧急程度: 中 ← 导出功能报错,显示'系统繁忙'...

任务 3:多标签分类

一个样本可以同时属于多个类别:

def multi_label_classify(text): """多标签分类:一篇文章可能同时属于多个主题""" prompt = f"""请判断以下文本属于哪些类别(可多选)。 可选类别:科技、体育、娱乐、政治、财经、教育、健康、其他 文本:{text} 所属类别(用逗号分隔):""" return llm_classify(prompt) # 测试 texts = [ "OpenAI发布新模型,在教育领域引发广泛讨论", "国足2-1逆转战胜日本队,球迷沸腾", "新研究显示:每天运动30分钟可降低患癌风险", ] for text in texts: labels = multi_label_classify(text) print(f"标签: {labels} ← {text[:20]}")

输出

标签: 科技, 教育 ← OpenAI发布新模型... 标签: 体育 ← 国足2-1逆转战胜日本队... 标签: 健康 ← 新研究显示:每天运动30分钟...

任务 4:自定义类别分类

def custom_classify(text, categories, descriptions=None): """自定义类别分类——类别和规则由你定""" cat_desc = "" if descriptions: cat_desc = "\n".join([f"- {c}: {d}" for c, d in zip(categories, descriptions)]) else: cat_desc = "\n".join([f"- {c}" for c in categories]) prompt = f"""请将以下文本分类到最合适的类别。 可选类别: {cat_desc} 文本:{text} 类别:""" return llm_classify(prompt) # 示例:电商客服对话分类 categories = ["退货退款", "物流查询", "产品咨询", "投诉", "其他"] descriptions = [ "用户要求退货或退款", "查询订单物流状态", "询问产品功能、规格、使用方法", "对产品质量或服务表达不满", "其他类型的问题", ] queries = [ "你好,我上周买的鞋子尺码不对,想换一双", "我的快递显示已签收但我没收到", "这款手机的电池续航怎么样?", ] for q in queries: cat = custom_classify(q, categories, descriptions) print(f"{cat} ← {q[:20]}")

输出

退货退款 ← 你好,我上周买的鞋子尺码不对... 物流查询 ← 我的快递显示已签收但我没收到... 产品咨询 ← 这款手机的电池续航怎么样?

任务 5:带理由的分类

不仅给出分类结果,还解释原因:

def classify_with_reason(text, categories): """分类 + 给出理由""" prompt = f"""将以下文本分类,并解释你的判断理由。 可选类别:{', '.join(categories)} 文本:{text} 请用以下格式输出: 类别:[类别] 理由:[为什么分到这一类] """ return llm_classify(prompt, max_new_tokens=150) # 测试 text = "这已经是我第三次联系客服了,每次都说会解决,但到现在都没人处理!" categories = ["投诉", "咨询", "建议"] result = classify_with_reason(text, categories) print(result)

输出

类别:投诉 理由:用户表达了对客服重复联系未解决问题的不满,语气中带有明显的失望和愤怒,属于典型的服务投诉。

四、三种信息提取任务

任务 1:结构化字段提取

def extract_fields(text, fields, output_json=True): """从非结构化文本中提取指定字段""" fields_str = "、".join(fields) prompt = f"""从以下文本中提取信息。 文本:{text} 需要提取的字段:{fields_str} {'请用JSON格式输出。' if output_json else ''}""" if output_json: prompt += "\n请严格按照 JSON 格式输出,不要加额外说明。" result = llm_classify(prompt, max_new_tokens=150, temperature=0.1) if output_json: try: return json.loads(result) except: return {"raw": result} return result # 测试:从简历中提取信息 resume = """ 姓名:张三 电话:138-0000-1234 邮箱:zhangsan@email.com 工作经验:5年 技能:Python、Java、机器学习、数据分析 教育背景:北京大学计算机科学硕士 """ fields = ["姓名", "电话", "邮箱", "工作经验", "技能", "教育背景"] extracted = extract_fields(resume, fields) print(json.dumps(extracted, ensure_ascii=False, indent=2))

输出

{ "姓名": "张三", "电话": "138-0000-1234", "邮箱": "zhangsan@email.com", "工作经验": "5年", "技能": "Python、Java、机器学习、数据分析", "教育背景": "北京大学计算机科学硕士" }

任务 2:命名实体识别(NER)

def extract_entities(text): """提取文本中的命名实体""" prompt = f"""从以下文本中识别人名、地名、组织名、时间、金额等实体。 文本:{text} 请按以下格式输出(如果没有某类实体则填"无"): 人名: 地名: 组织名: 时间: 金额: """ return llm_classify(prompt, max_new_tokens=200) text = "2026年6月15日,华为公司在深圳发布了新款MatePad,余承东主持了发布会,产品售价3999元起。" print(extract_entities(text))

输出

人名:余承东 地名:深圳 组织名:华为公司 时间:2026年6月15日 金额:3999元

任务 3:文本摘要与关键信息

def summarize_and_extract(text, max_length=3): """文本摘要 + 关键信息提取""" prompt = f"""请对以下文本做摘要并提取关键信息。 文本:{text} 请输出: 1. 摘要({max_length}句话以内): 2. 关键词(5个,逗号分隔): 3. 核心观点(1句话): """ return llm_classify(prompt, max_new_tokens=200) news = """ 2026年6月18日,全球AI开发者大会在杭州开幕。本次大会吸引了来自85个国家的超过3万名开发者参与, 创下历史新高。大会主题聚焦于"AI Agent的产业化应用",超过200家企业展示了最新的AI Agent产品。 其中,多家中国企业展示了基于开源模型的行业解决方案,引起了广泛关注。 大会将持续三天,期间将举办50余场技术分论坛。 """ print(summarize_and_extract(news))

输出

1. 摘要(3句话以内): 全球AI开发者大会在杭州开幕,创下历史新高。主题聚焦AI Agent产业化应用。中国企业的开源方案引起广泛关注。 2. 关键词(5个,逗号分隔): AI开发者大会, AI Agent, 开源模型, 杭州, 产业化 3. 核心观点(1句话): 2026年AI Agent正从概念走向产业化应用,中国开源方案在全球舞台上展现竞争力。

五、批处理与性能优化

批量分类

当需要处理大量文本时,可以用 batch 处理来加速:

def batch_classify(texts, task_type="sentiment", batch_size=4): """批量分类""" results = [] for i in range(0, len(texts), batch_size): batch = texts[i:i+batch_size] # 在同一个提示中包含多个样本(提升吞吐) batch_prompt = f"""请对以下每条文本进行{task_type}分析。 """ for idx, text in enumerate(batch): batch_prompt += f"{idx+1}. 文本:{text}\n" batch_prompt += f"\n请对每条文本分别给出{task_type}结果。" result = llm_classify(batch_prompt, max_new_tokens=200) results.append(result) return results # 批量处理 8 条 reviews = [ "非常好用,推荐!", "质量一般,性价比不高。", "客服态度很差。", "物流很快,第二天就到了。", "颜色和图片有色差。", "整体满意,会回购。", "包装有破损,但产品没问题。", "用了三个月出现故障。", ] results = batch_classify(reviews, task_type="情感分析(正面/负面)") for review, result in zip(reviews, results): print(f"{review[:15]:15s} → {result[:10]}")

性能对比

处理方式8 条耗时适用场景
单条串行处理~24 秒实时小量请求
批量合并提示~8 秒离线批量处理
并行多模型实例~3 秒高吞吐生产环境

六、生产环境注意事项

输出格式规范化

import re def parse_classification(raw_output, valid_labels): """解析模型输出,提取有效标签""" # 清理输出 output = raw_output.strip().lower() # 直接匹配 for label in valid_labels: if label.lower() in output: return label # 正则匹配可能的前缀 match = re.search(r'(?:类别|分类|情感|标签)[::]\s*(\S+)', output) if match: return match.group(1) # 默认返回 return "unknown"

错误处理与重试

def safe_classify(text, max_retries=2): """带重试机制的稳定分类""" for attempt in range(max_retries + 1): try: result = llm_classify(text, temperature=0.1) # 检查输出是否合理 if any(label in result for label in valid_labels): return parse_classification(result, valid_labels) # 如果输出不符合预期,重试 if attempt < max_retries: continue except Exception as e: if attempt < max_retries: continue return "error" return "unknown"

缓存机制

from functools import lru_cache @lru_cache(maxsize=1000) def cached_classify(text: str, task: str = "sentiment"): """缓存相同输入的分类结果,避免重复计算""" prompt = f"请判断以下文本的情感(正面/负面):\n{text}\n情感:" return llm_classify(prompt) # 使用:相同的文本不会重复调用模型 result1 = cached_classify("产品很好") # 调用模型 result2 = cached_classify("产品很好") # 命中缓存,瞬间返回

七、全套工具函数

class LocalLLMClassifier: """完整的本地 LLM 分类器工具类""" def __init__(self, model_path="d:/ai/models/Qwen3-0.6B"): self.tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True) self.model = AutoModelForCausalLM.from_pretrained( model_path, trust_remote_code=True, device_map="auto" ) self.cache = {} def predict(self, text, categories, task_type="分类", return_reason=False, temperature=0.1): """统一分类接口""" cache_key = (text, str(categories), task_type) if cache_key in self.cache: return self.cache[cache_key] if return_reason: prompt = f"""请将以下文本分类到最合适的类别。 可选类别:{', '.join(categories)} 文本:{text} 请用以下格式输出: 类别: 理由:""" else: prompt = f"""请将以下文本分类到最合适的类别。 可选类别:{', '.join(categories)} 文本:{text} 类别:""" result = self._generate(prompt, max_new_tokens=150, temperature=temperature) self.cache[cache_key] = result return result def extract(self, text, fields): """统一信息提取接口""" prompt = f"""从以下文本中提取信息。 文本:{text} 需要提取的字段:{'、'.join(fields)} 请用JSON格式输出。""" result = self._generate(prompt, max_new_tokens=200, temperature=0.1) try: return json.loads(result) except: return {"raw": result} def _generate(self, prompt, max_new_tokens=150, temperature=0.1): messages = [{"role": "user", "content": prompt}] text = self.tokenizer.apply_chat_template( messages, tokenize=False, add_generation_prompt=True ) inputs = self.tokenizer(text, return_tensors="pt").to(self.model.device) outputs = self.model.generate( **inputs, max_new_tokens=max_new_tokens, temperature=temperature, do_sample=True, ) return self.tokenizer.decode( outputs[0][inputs.input_ids.shape[1]:], skip_special_tokens=True ).strip() # 使用示例 clf = LocalLLMClassifier() # 分类 print(clf.predict("这个产品质量太差了", categories=["好评", "差评", "中性"], task_type="评价分类")) # 提取 print(clf.extract("我叫李四,电话是13912345678,住在北京市海淀区。", fields=["姓名", "电话", "地址"]))

八、总结

任务类型实现方式核心技巧
二分类直接提示,指定类别temperature=0.1 确保稳定
多分类给出类别列表 + 分类标准类别描述越清晰,结果越准
多标签分类允许多选输出明确说明用逗号分隔
信息提取指定字段名,要求 JSON 输出示例可以极大提升格式稳定性
命名实体识别识别预定义类型的实体小模型上效果好
批量处理多条文本合并到一个提示减少调用次数,提升吞吐

核心三句话

  1. 用 LLM 做分类/提取的核心优势是"零样本迁移"——换任务只需改提示词,不用重新训练
  2. 分类任务用 low temperature(0.1),生成任务用 high temperature(0.7)——这是最重要的参数设置
  3. 本地模型在分类和提取任务上的效果通常超出预期——Qwen3-0.6B 虽然只有 6 亿参数,但在结构化任务上表现可靠

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询