摘要:文本分类和信息提取是 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 输出 | 示例可以极大提升格式稳定性 |
| 命名实体识别 | 识别预定义类型的实体 | 小模型上效果好 |
| 批量处理 | 多条文本合并到一个提示 | 减少调用次数,提升吞吐 |
核心三句话:
- 用 LLM 做分类/提取的核心优势是"零样本迁移"——换任务只需改提示词,不用重新训练
- 分类任务用 low temperature(0.1),生成任务用 high temperature(0.7)——这是最重要的参数设置
- 本地模型在分类和提取任务上的效果通常超出预期——Qwen3-0.6B 虽然只有 6 亿参数,但在结构化任务上表现可靠