用Python+Spacy+Transformers构建5个NLP实战项目：从NLU到NLG的深度实践

自然语言处理（NLP）领域常被划分为自然语言理解（NLU）和自然语言生成（NLG）两大方向，但实际项目中这两者往往密不可分。本文将通过5个渐进式实战项目，带你用Python生态中最实用的工具链（Spacy+Transformers）打通NLU与NLG的任督二脉。不同于传统教程的理论堆砌，我们将从第一个代码单元格开始就进入实战状态——你会亲手搭建能处理真实业务的智能客服系统、可部署的情感分析API、能自动优化表达的文本润色工具等完整项目。过程中不仅会掌握BERT、GPT等Transformer模型的工程化应用技巧，更能理解如何让NLU与NLG在实际场景中协同工作。

1. 项目架构设计与环境配置

1.1 技术选型与工具链组合

现代NLP项目开发已形成稳定的工具栈分层架构：

• 基础层 ：Spacy（工业级文本处理）、NLTK（学术研究传统工具）
• 模型层 ：HuggingFace Transformers（预训练模型库）、Gensim（传统词向量）
• 部署层 ：FastAPI（轻量级API）、Streamlit（快速可视化）
• 辅助工具 ：Prodigy（数据标注）、Weights & Biases（实验跟踪）

推荐使用conda创建隔离环境：

conda create -n nlp_projects python=3.8
conda activate nlp_projects
pip install spacy transformers torch sentencepiece
python -m spacy download en_core_web_lg

1.2 硬件资源规划

不同规模项目对计算资源的需求差异显著：

项目类型	CPU核心	内存	GPU显存	预估训练时间
规则型系统	4	8GB	不需要	<1小时
微调BERT-base	8	32GB	12GB	2-4小时
运行GPT-3	16	64GB	24GB	需API调用

提示：本地开发时可使用Colab Pro的T4 GPU运行前四个项目，最后一个项目建议使用A100实例

2. 实战项目一：智能工单分类系统（纯NLU）

2.1 业务场景建模

假设我们需要处理电商平台的用户工单，原始数据如下：

tickets = [
    "我的订单#3012还没发货，已经逾期3天了",
    "刚收到的外套尺码不对想换M号",
    "支付成功后没收到积分奖励",
    "你们APP在iPhone12上老是闪退"
]

2.2 多标签分类实现

使用Spacy的TextCategorizer组件构建分类管道：

import spacy
from spacy.training import Example

nlp = spacy.blank("en")
config = {
    "model": {
        "@architectures": "spacy.TextCatEnsemble.v2",
        "tok2vec": {"@architectures": "spacy.Tok2Vec.v2", "width": 96},
        "nO": None,
    }
}
textcat = nlp.add_pipe("textcat_multilabel", config=config)

# 添加标签并准备训练数据
labels = ["物流问题", "退换货", "支付问题", "技术故障"]
for label in labels:
    textcat.add_label(label)

train_data = [
    ("我的订单还没发货", {"cats": {"物流问题": 1.0}}),
    ("想换大一号的", {"cats": {"退换货": 1.0}}),
    # 更多标注数据...
]

# 转换为Spacy示例
examples = [Example.from_dict(nlp.make_doc(text), annots) for text, annots in train_data]

2.3 模型训练与评估

使用早停策略防止过拟合：

from spacy.training import Example
import random

optimizer = nlp.initialize()
for epoch in range(10):
    random.shuffle(examples)
    losses = {}
    for batch in spacy.util.minibatch(examples, size=8):
        nlp.update(batch, drop=0.1, losses=losses, sgd=optimizer)
    print(f"Epoch {epoch}, Loss: {losses['textcat_multilabel']:.3f}")

评估时需注意多标签场景的特殊指标：

• 精确率-召回率曲线下面积（PR-AUC）
• 按样本精度（Exact Match Ratio）

3. 实战项目二：合同关键信息抽取（NLU+结构化输出）

3.1 法律文书中的实体识别

构建自定义NER模型识别合同要素：

from transformers import AutoTokenizer, AutoModelForTokenClassification

model_name = "bert-base-chinese"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForTokenClassification.from_pretrained(
    model_name, 
    num_labels=len(label_map)
)

# 示例标注数据
contract_text = "本合同由甲方：阿里巴巴（中国）有限公司与乙方：腾讯科技签订"
annotations = {
    "entities": [
        (5, 9, "PARTY"), 
        (10, 23, "COMPANY"),
        (26, 28, "PARTY"),
        (29, 33, "COMPANY")
    ]
}

3.2 关系抽取增强

使用SpanMarker识别实体间关系：

from span_marker import SpanMarkerModel

model = SpanMarkerModel.from_pretrained(
    "bert-base-chinese",
    labels=["甲方-公司", "乙方-公司", "签约方-日期"],
)

train_dataset = Dataset.from_dict({
    "tokens": [["本", "合同", "由",...]],
    "ner_tags": [[0, 0, 0, 1, 2,...]],
    "relation_tags": [[...]]
})

3.3 输出结构化JSON

将识别结果转换为业务系统可消费的格式：

{
  "contract_parties": [
    {
      "role": "甲方",
      "name": "阿里巴巴（中国）有限公司",
      "type": "企业"
    },
    {
      "role": "乙方", 
      "name": "腾讯科技",
      "type": "企业"
    }
  ],
  "sign_date": "2023-07-15",
  "effective_terms": "2年"
}

4. 实战项目三：新闻摘要生成器（NLG核心任务）

4.1 数据准备与清洗

使用CNN/DailyMail数据集：

from datasets import load_dataset

dataset = load_dataset("cnn_dailymail", "3.0.0")
example = dataset["train"][0]
print(f"""
原文长度: {len(example["article"].split())}词
摘要长度: {len(example["highlights"].split())}词
""")

4.2 PEGASUS模型微调

使用Google的预训练摘要模型：

from transformers import PegasusTokenizer, PegasusForConditionalGeneration

model_name = "google/pegasus-cnn_dailymail"
tokenizer = PegasusTokenizer.from_pretrained(model_name)
model = PegasusForConditionalGeneration.from_pretrained(model_name)

inputs = tokenizer(
    dataset["train"][:100]["article"],
    padding="max_length", 
    truncation=True,
    return_tensors="pt"
)

# 训练代码简化版
from transformers import Trainer, TrainingArguments

training_args = TrainingArguments(
    output_dir="./pegasus-finetuned",
    per_device_train_batch_size=4,
    predict_with_generate=True
)

trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=inputs
)
trainer.train()

4.3 生成结果优化

控制摘要生成质量的三个关键参数：

1. 长度惩罚 （length_penalty）：

   generate_kwargs = {
       "length_penalty": 1.5,  # >1鼓励长摘要，<1鼓励短摘要
       "max_length": 128,
       "num_beams": 8
   }
   

2. N-gram惩罚 （no_repeat_ngram_size）：

   generate_kwargs["no_repeat_ngram_size"] = 3  # 防止重复短语
   

3. 温度调节 （temperature）：

   generate_kwargs["temperature"] = 0.7  # 平衡创造性与准确性
   

5. 实战项目四：多轮对话系统（NLU+NLG联合）

5.1 对话状态跟踪

使用Rasa风格的状态管理：

class DialogState:
    def __init__(self):
        self.slots = {
            "product_type": None,
            "color_preference": None,
            "budget_range": None
        }
        self.history = []
    
    def update(self, user_utterance):
        # 使用NLU模型提取意图和实体
        doc = nlp(user_utterance)
        for ent in doc.ents:
            if ent.label_ in self.slots:
                self.slots[ent.label_] = ent.text
        self.history.append(user_utterance)

5.2 响应生成策略

混合规则与生成式方法：

from transformers import pipeline

generator = pipeline(
    "text-generation", 
    model="microsoft/DialoGPT-medium"
)

def generate_response(state):
    # 规则优先
    if not state.slots["product_type"]:
        return "请问您想购买什么类型的商品？"
    
    # 生成式响应
    prompt = f"用户想买{state.slots['product_type']}"
    if state.slots["color_preference"]:
        prompt += f"，偏好{state.slots['color_preference']}色"
    prompt += "。请生成亲切的推荐回复："
    
    return generator(
        prompt,
        max_length=100,
        do_sample=True,
        temperature=0.8
    )[0]["generated_text"]

5.3 评估指标设计

对话系统的特殊评估维度：

维度	评估方法	合格标准
连贯性	人工评分（1-5分）	≥4分
任务完成率	关键槽位填充率	≥90%
用户体验	平均对话轮次	≤5轮完成核心任务
安全性	敏感词触发率	≤1%

6. 实战项目五：代码注释生成（领域特定NLG）

6.1 代码解析与表示

使用Tree-sitter进行语法分析：

from tree_sitter import Language, Parser

PYTHON_LANGUAGE = Language('build/my-languages.so', 'python')
parser = Parser()
parser.set_language(PYTHON_LANGUAGE)

code = """
def factorial(n):
    if n <= 1:
        return 1
    return n * factorial(n-1)
"""

tree = parser.parse(bytes(code, "utf8"))
nodes_to_comments = {
    "function_definition": "描述函数整体功能",
    "if_statement": "说明条件分支逻辑",
    "return_statement": "解释返回值含义"
}

6.2 CodeT5模型应用

Salesforce的代码生成模型：

from transformers import T5ForConditionalGeneration, AutoTokenizer

model = T5ForConditionalGeneration.from_pretrained("Salesforce/codet5-base-multi-sum")
tokenizer = AutoTokenizer.from_pretrained("Salesforce/codet5-base-multi-sum")

inputs = tokenizer(
    "Generate Python docstring: " + code,
    return_tensors="pt",
    max_length=512,
    truncation=True
)

outputs = model.generate(
    inputs.input_ids,
    max_length=100,
    num_beams=5,
    early_stopping=True
)

print(tokenizer.decode(outputs[0], skip_special_tokens=True))

6.3 生成质量提升技巧

通过后处理优化注释可读性：

1. 术语一致性检查 ：

   glossary = {
       "factorial": "阶乘函数",
       "recursive": "递归实现"
   }
   

2. 代码元素保留 ：

   def retain_code_terms(comment, code):
       for token in code.split():
           if token in comment:
               comment = comment.replace(token, f"`{token}`")
       return comment
   

3. 风格规范化 ：

   def format_as_docstring(comment):
       lines = [line.strip() for line in comment.split("\n")]
       return '"""\n' + "\n".join(lines) + '\n"""'
   

7. 工程化部署与性能优化

7.1 模型量化加速

使用ONNX Runtime提升推理速度：

from transformers import convert_graph_to_onnx

convert_graph_to_onnx.convert(
    framework="pt",
    model=model,
    output_path="model.onnx",
    opset=12,
    tokenizer=tokenizer
)

import onnxruntime as ort

sess = ort.InferenceSession("model.onnx")
inputs = tokenizer("Text to analyze", return_tensors="np")
outputs = sess.run(None, dict(inputs))

7.2 缓存策略设计

针对高频请求的缓存方案：

from diskcache import Cache
from hashlib import md5

cache = Cache("nlp_cache")

def get_cached_response(text, model):
    key = md5(f"{model}:{text}".encode()).hexdigest()
    if key in cache:
        return cache[key]
    
    result = model(text)
    cache.set(key, result, expire=3600)
    return result

7.3 监控指标实现

Prometheus监控关键指标：

from prometheus_client import Counter, Gauge

REQUEST_COUNT = Counter(
    'nlp_request_total',
    'Total NLP API requests',
    ['model', 'status']
)

LATENCY = Gauge(
    'nlp_latency_seconds',
    'Request processing latency',
    ['model']
)

def process_request(text, model):
    start = time.time()
    try:
        result = model(text)
        REQUEST_COUNT.labels(model=model, status="success").inc()
        return result
    except Exception:
        REQUEST_COUNT.labels(model=model, status="fail").inc()
        raise
    finally:
        LATENCY.labels(model=model).set(time.time() - start)

8. 前沿技术融合与项目扩展

8.1 大模型时代的工作流革新

当引入GPT-4级别模型后的架构调整：

1. 提示工程 取代传统微调

   def build_prompt(task_description, examples, new_input):
       return f"""
   {task_description}
   
   Examples:
   {examples}
   
   New Input:
   {new_input}
   
   Output:
   """
   

2. 小样本学习 （Few-shot Learning）：

   few_shot_examples = [
       ("文本1", "标签1"),
       ("文本2", "标签2")
   ]
   

3. 人类反馈强化学习 （RLHF）：

   from trl import PPOTrainer
   
   trainer = PPOTrainer(
       model=model,
       tokenizer=tokenizer,
       reward_model=reward_model
   )
   

8.2 多模态扩展

结合CLIP等视觉模型：

from transformers import CLIPProcessor, CLIPModel

clip_model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")
processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")

def image_to_text(image_path, candidate_labels):
    image = Image.open(image_path)
    inputs = processor(
        text=candidate_labels,
        images=image,
        return_tensors="pt",
        padding=True
    )
    outputs = clip_model(**inputs)
    logits = outputs.logits_per_image
    return candidate_labels[logits.argmax()]

8.3 持续学习策略

防止模型遗忘的增量学习：

from continual import ContinualLearner

learner = ContinualLearner(
    core_model=model,
    memory_size=1000,
    replay_strategy="reservoir"
)

while new_data:
    learner.observe(batch)
    if learner.should_update():
        learner.update()