2025年自然语言处理核心技术与应用实战

# 2025年基于大型预训练语言模型的Token Classification示例实现
import torch
import torch.nn as nn
import torch.nn.functional as F
from transformers import AutoTokenizer, AutoModelForTokenClassification, pipeline
from datasets import load_dataset, load_metric
import numpy as np
from tqdm import tqdm
import os

class AdvancedTokenClassifier:
    def __init__(self, model_name="google/gemma-2-9b-it", num_labels=3, device=None):
        # 加载预训练的语言模型和分词器
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForTokenClassification.from_pretrained(model_name, num_labels=num_labels, ignore_mismatched_sizes=True)
        
        # 设置设备
        self.device = device if device is not None else ("cuda" if torch.cuda.is_available() else "cpu")
        self.model.to(self.device)
        
        # 设置模型为评估模式
        self.model.eval()
        
        # 获取模型信息
        self.model_name = model_name
        self.num_labels = num_labels
        
        # 初始化pipeline用于快速推理
        self.pipeline = pipeline(
            "token-classification",
            model=self.model,
            tokenizer=self.tokenizer,
            device=self.device,
            aggregation_strategy="simple"  # 设置聚合策略
        )
        
        # 设置标签映射（可以根据实际任务进行调整）
        self.label_map = {0: "O", 1: "B-ENT", 2: "I-ENT"}  # 示例：O表示非实体，B-ENT表示实体开始，I-ENT表示实体内部
        self.id2label = {i: label for i, label in enumerate(self.label_map.values())}
        self.label2id = {label: i for i, label in enumerate(self.label_map.values())}
    
    def tokenize_text(self, texts, max_length=512, padding=True, truncation=True):
        # 分词处理
        tokenized_inputs = self.tokenizer(
            texts,
            max_length=max_length,
            padding=padding,
            truncation=truncation,
            return_tensors="pt",
            return_offsets_mapping=True  # 返回偏移量映射，用于对齐原始文本
        )
        
        # 将数据移至指定设备
        for key, value in tokenized_inputs.items():
            if isinstance(value, torch.Tensor):
                tokenized_inputs[key] = value.to(self.device)
        
        return tokenized_inputs
    
    def predict(self, texts, threshold=0.5, max_length=512):
        # 预测文本中的token类别
        # 确保输入是列表格式
        if isinstance(texts, str):
            texts = [texts]
        
        # 使用pipeline进行预测
        predictions = self.pipeline(texts, max_length=max_length, truncation=True)
        
        # 处理预测结果
        results = []
        for i, text in enumerate(texts):
            # 获取当前文本的预测
            text_predictions = predictions[i] if isinstance(predictions, list) else predictions
            
            # 过滤低置信度的预测
            filtered_predictions = [p for p in text_predictions if p["score"] >= threshold]
            
            # 提取实体信息
            entities = []
            for pred in filtered_predictions:
                entity = {
                    "entity": pred["entity_group"],
                    "word": pred["word"],
                    "start": pred["start"],
                    "end": pred["end"],
                    "score": pred["score"]
                }
                entities.append(entity)
            
            results.append({
                "text": text,
                "entities": entities,
                "predictions": filtered_predictions
            })
        
        return results if len(results) > 1 else results[0]
    
    def batch_predict(self, texts, batch_size=8, threshold=0.5, max_length=512):
        # 批量预测文本中的token类别
        results = []
        
        # 将文本分成批次
        for i in tqdm(range(0, len(texts), batch_size), desc="Batch Predicting"):
            batch_texts = texts[i:i+batch_size]
            
            # 使用pipeline进行批量预测
            batch_predictions = self.pipeline(batch_texts, max_length=max_length, truncation=True)
            
            # 处理批次预测结果
            for j, text in enumerate(batch_texts):
                # 获取当前文本的预测
                text_predictions = batch_predictions[j] if isinstance(batch_predictions, list) else batch_predictions
                
                # 过滤低置信度的预测
                filtered_predictions = [p for p in text_predictions if p["score"] >= threshold]
                
                # 提取实体信息
                entities = []
                for pred in filtered_predictions:
                    entity = {
                        "entity": pred["entity_group"],
                        "word": pred["word"],
                        "start": pred["start"],
                        "end": pred["end"],
                        "score": pred["score"]
                    }
                    entities.append(entity)
                
                results.append({
                    "text": text,
                    "entities": entities,
                    "predictions": filtered_predictions
                })
        
        return results

from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline
import torch

# 加载模型和分词器
def load_model(model_name="meta-llama/Llama-4-70B-Chat-hf"):
    print(f"正在加载模型: {model_name}")
    # 设置缓存目录，避免重复下载
    cache_dir="./models_cache"
    
    # 加载分词器
    tokenizer = AutoTokenizer.from_pretrained(model_name, cache_dir=cache_dir)
    
    # 加载模型，使用GPU加速
    model = AutoModelForCausalLM.from_pretrained(
        model_name,
        cache_dir=cache_dir,
        torch_dtype=torch.bfloat16,  # 使用bfloat16精度减少显存占用
        device_map="auto",  # 自动分配设备
        load_in_4bit=True,  # 启用4bit量化
    )
    
    return tokenizer, model

# 创建文本生成管道
def create_text_generator(tokenizer, model):
    text_generator = pipeline(
        "text-generation",
        model=model,
        tokenizer=tokenizer,
        device_map="auto",
        max_length=1024,  # 设置最大生成长度
        do_sample=True,  # 启用采样生成
        temperature=0.7,  # 控制生成的随机性
        top_p=0.95,  # 控制词汇多样性
        repetition_penalty=1.1  # 防止重复
    )
    return text_generator

# 生成文本
def generate_text(text_generator, prompt, max_new_tokens=512):
    print(f"\n提示词: {prompt}")
    
    # 生成文本
    result = text_generator(
        prompt,
        max_new_tokens=max_new_tokens
    )
    
    # 提取生成的文本
    generated_text = result[0]["generated_text"][len(prompt):].strip()
    print(f"\n生成结果:\n{generated_text}")
    
    return generated_text

# 示例用法
if __name__ == "__main__":
    # 加载模型
    tokenizer, model = load_model()
    
    # 创建文本生成管道
    text_generator = create_text_generator(tokenizer, model)
    
    # 示例1：生成博客文章
    prompt1 = "写一篇关于2025年人工智能发展趋势的博客文章，面向普通读者，要求通俗易懂，包含具体例子。"
    generate_text(text_generator, prompt1)
    
    # 示例2：生成营销文案
    prompt2 = "为一款智能家居产品写一段吸引人的营销文案，突出其智能化、便捷性和安全性。"
    generate_text(text_generator, prompt2)
    
    # 示例3：生成代码注释
    prompt3 = "为以下Python代码添加详细的注释:\ndef calculate_factorial(n):\n    if n == 0 or n == 1:\n        return 1\n    else:\n        return n * calculate_factorial(n-1)"
    generate_text(text_generator, prompt3)

from transformers import AutoTokenizer, AutoModelForSequenceClassification, pipeline
import torch
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.metrics import classification_report, confusion_matrix

# 加载情感分析模型
def load_sentiment_model(model_name="cardiffnlp/twitter-roberta-base-sentiment-latest", use_4bit=True):
    print(f"正在加载情感分析模型: {model_name}")
    
    # 加载分词器
    tokenizer = AutoTokenizer.from_pretrained(model_name)
    
    # 配置模型参数
    model_kwargs = {
        "torch_dtype": torch.bfloat16,
        "use_safetensors": True,
    }
    
    # 如果启用4bit量化
    if use_4bit:
        from transformers import BitsAndBytesConfig
        quantization_config = BitsAndBytesConfig(
            load_in_4bit=True,
            bnb_4bit_compute_dtype=torch.bfloat16,
            bnb_4bit_quant_type="nf4",
            bnb_4bit_use_double_quant=True,
        )
        model_kwargs["quantization_config"] = quantization_config
    
    # 加载模型
    model = AutoModelForSequenceClassification.from_pretrained(
        model_name,
        **model_kwargs
    )
    
    # 将模型移至GPU（如果可用）
    if torch.cuda.is_available():
        model = model.to("cuda")
    
    return model, tokenizer

# 创建情感分析管道
def create_sentiment_pipeline(model, tokenizer):
    print("正在创建情感分析管道...")
    
    # 创建情感分析管道
    sentiment_analyzer = pipeline(
        "sentiment-analysis",
        model=model,
        tokenizer=tokenizer,
        device=0 if torch.cuda.is_available() else -1,
        return_all_scores=True
    )
    
    # 创建细粒度情感分析函数
    def fine_grained_sentiment_analysis(text, threshold=0.5):
        # 获取情感分析结果
        results = sentiment_analyzer(text)[0]
        
        # 格式化结果
        formatted_results = {}
        for result in results:
            formatted_results[result["label"]] = result["score"]
        
        # 确定主要情感和强度
        main_sentiment = max(formatted_results, key=formatted_results.get)
        sentiment_intensity = formatted_results[main_sentiment]
        
        # 检测混合情感
        mixed_sentiments = []
        for label, score in formatted_results.items():
            if label != main_sentiment and score > threshold:
                mixed_sentiments.append((label, score))
        
        return {
            "main_sentiment": main_sentiment,
            "sentiment_intensity": sentiment_intensity,
            "mixed_sentiments": mixed_sentiments,
            "all_sentiments": formatted_results
        }
    
    return fine_grained_sentiment_analysis

# 示例用法
if __name__ == "__main__":
    # 加载模型
    model, tokenizer = load_sentiment_model()
    
    # 创建情感分析管道
    sentiment_analyzer = create_sentiment_pipeline(model, tokenizer)
    
    # 示例1：分析社交媒体评论
    text1 = "这款新产品真的太棒了，功能强大且易于使用，强烈推荐给大家！"
    result1 = sentiment_analyzer(text1)
    print(f"\n示例1 - 社交媒体评论情感分析:\n{result1}")
    
    # 示例2：分析客户反馈
    text2 = "服务态度很好，但产品质量有待提高，希望能够尽快改进。"
    result2 = sentiment_analyzer(text2)
    print(f"\n示例2 - 客户反馈情感分析:\n{result2}")