基于腾讯云TI平台的AI微头条生成实战：从模型微调到内容优化

# dataloader.py
import json
import re
from pyspark.sql import SparkSession
from pyspark.sql.functions import udf, col
from pyspark.sql.types import BooleanType

spark = SparkSession.builder.appName("MicroNewsClean").getOrCreate()

def is_valid(text):
    if not text or len(text) < 30 or len(text) > 200:
        return False
    # 去除HTML标签
    clean = re.sub(r'<[^>]+>', '', text)
    # 特殊字符比例 < 10%
    spec_char = sum(1 for ch in clean if not ch.isalnum() and ch not in '，。！？') / len(clean)
    if spec_char > 0.1:
        return False
    # 停用词占比（自定义stopwords.txt）
    with open('stopwords.txt', 'r', encoding='utf-8') as f:
        stops = set(f.read().splitlines())
    words = [w for w in clean if w.strip()]
    if not words:
        return False
    stop_ratio = sum(1 for w in words if w in stops) / len(words)
    return stop_ratio < 0.4

valid_udf = udf(is_valid, BooleanType())
df = spark.read.json("raw_data/*.jsonl")
df_filtered = df.filter(valid_udf(col("content")))
df_filtered.write.json("cleaned_data/")

pip install transformers==4.35.0 peft==0.7.0 accelerate datasets deepspeed

import torch
from transformers import AutoTokenizer, AutoModelForCausalLM, TrainingArguments, Trainer
from peft import LoraConfig, get_peft_model, TaskType
from datasets import load_dataset

# 加载数据集（已切分为train/val）
dataset = load_dataset("json", data_files={"train": "cleaned_data/train/*.jsonl", 
                                           "val": "cleaned_data/val/*.jsonl"})

tokenizer = AutoTokenizer.from_pretrained("uer/gpt2-chinese-cluecorpussmall")
tokenizer.pad_token = tokenizer.eos_token

def tokenize_func(examples):
    # 输入格式：标题 + [SEP] + 正文
    texts = [t + "[SEP]" + c for t, c in zip(examples["title"], examples["content"])]
    return tokenizer(texts, truncation=True, max_length=512, padding="max_length")

tokenized = dataset.map(tokenize_func, batched=True, remove_columns=dataset["train"].column_names)

# LoRA 配置
lora_config = LoraConfig(
    task_type=TaskType.CAUSAL_LM,
    r=8,
    lora_alpha=32,
    target_modules=["c_attn"],  # GPT-2的注意力投影层
    lora_dropout=0.1,
    bias="none",
)

model = AutoModelForCausalLM.from_pretrained("uer/gpt2-chinese-cluecorpussmall")
model = get_peft_model(model, lora_config)
model.print_trainable_parameters()  # 仅 ~0.3M 可训练

# 训练参数（腾讯云V100 16G，batch_size=8，gradient_accumulation=4）
training_args = TrainingArguments(
    output_dir="./gpt2_lora_micro",
    per_device_train_batch_size=8,
    per_device_eval_batch_size=16,
    gradient_accumulation_steps=4,
    num_train_epochs=3,
    learning_rate=5e-4,
    fp16=True,
    save_steps=500,
    eval_steps=500,
    logging_steps=100,
    report_to="none",
)

trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=tokenized["train"],
    eval_dataset=tokenized["val"],
    data_collator=lambda data: {"input_ids": torch.stack([d["input_ids"] for d in data]),
                                "attention_mask": torch.stack([d["attention_mask"] for d in data])}
)
trainer.train()
model.save_pretrained("./gpt2_lora_final")

def generate_micro(keyword, model, tokenizer, max_new_tokens=150, min_length=40):
    prompt = f"【{keyword}】"
    inputs = tokenizer(prompt, return_tensors="pt").to(model.device)
    outputs = model.generate(
        **inputs,
        max_new_tokens=max_new_tokens,
        min_new_tokens=min_length,
        do_sample=True,
        top_k=50,
        top_p=0.92,
        repetition_penalty=1.1,
        no_repeat_ngram_size=3,
        early_stopping=True,
        eos_token_id=tokenizer.eos_token_id,
        pad_token_id=tokenizer.pad_token_id,
    )
    full_text = tokenizer.decode(outputs[0], skip_special_tokens=True)
    # 截取[SEP]后的部分
    if "[SEP]" in full_text:
        return full_text.split("[SEP]")[1].strip()
    return full_text[len(prompt):].strip()

from transformers import AutoTokenizer, AutoModelForSequenceClassification, Trainer, TrainingArguments
from datasets import Dataset
import pandas as pd

df = pd.read_csv("labeled_interaction.csv")  # columns: text, label (1=高互动)
dataset = Dataset.from_pandas(df)

tokenizer = AutoTokenizer.from_pretrained("hfl/chinese-roberta-wwm-ext")
def tok_func(batch):
    return tokenizer(batch["text"], truncation=True, padding=True, max_length=128)
dataset = dataset.map(tok_func, batched=True)

model = AutoModelForSequenceClassification.from_pretrained("hfl/chinese-roberta-wwm-ext", num_labels=2)
args = TrainingArguments(
    output_dir="./scorer",
    per_device_train_batch_size=32,
    num_train_epochs=5,
    learning_rate=2e-5,
    evaluation_strategy="epoch",
    save_strategy="epoch",
    load_best_model_at_end=True,
)
trainer = Trainer(model=model, args=args, train_dataset=dataset, 
                  eval_dataset=dataset.shuffle().select(range(3000)))
trainer.train()
model.save_pretrained("./micro_scorer")

from transformers import pipeline
scorer = pipeline("text-classification", model="./micro_scorer", device=0)

def rank_and_select(keyword, generator, scorer, n_candidates=20, top_k=3, diversity_lambda=0.3):
    candidates = [generate_micro(keyword, generator, tokenizer) for _ in range(n_candidates)]
    scores = [scorer(c)[0]['score'] for c in candidates]  # 正类概率
    
    # MMR 排序（最大化 relevance - diversity）
    selected = []
    remaining = list(range(n_candidates))
    for _ in range(top_k):
        best_idx = -1
        best_score = -1
        for idx in remaining:
            rel = scores[idx]
            if selected:
                # 计算与已选句子的最大相似度（基于TF-IDF）
                sim = max([compute_cosine(candidates[idx], candidates[s]) for s in selected])
                mmr = rel - diversity_lambda * sim
            else:
                mmr = rel
            if mmr > best_score:
                best_score = mmr
                best_idx = idx
        selected.append(best_idx)
        remaining.remove(best_idx)
    return [candidates[i] for i in selected]

# 合并LoRA权重
from peft import PeftModel
base = AutoModelForCausalLM.from_pretrained("uer/gpt2-chinese-cluecorpussmall")
lora = PeftModel.from_pretrained(base, "./gpt2_lora_final")
merged = lora.merge_and_unload()
merged.eval()

# 导出为TorchScript（使用trace）
dummy_input = tokenizer("【新能源】", return_tensors="pt")["input_ids"]
traced = torch.jit.trace(merged, dummy_input)
traced.save("generator_ts.pt")

# service.py
from fastapi import FastAPI, Request
import torch
from transformers import AutoTokenizer
from scorer_pipeline import load_scorer
from mmr_utils import rank_and_select

app = FastAPI()
generator = torch.jit.load("generator_ts.pt")
tokenizer = AutoTokenizer.from_pretrained("uer/gpt2-chinese-cluecorpussmall")
scorer = load_scorer("./micro_scorer")

@app.post("/generate")
async def generate(request: Request):
    data = await request.json()
    keyword = data.get("keyword", "")
    candidates = [generate_micro(keyword, generator, tokenizer) for _ in range(20)]
    ranked = rank_and_select(candidates, scorer)
    return {"keyword": keyword, "top3": ranked}

# 健康检查
@app.get("/health")
async def health():
    return {"status": "ok"}