简介本资源面向具备一定Python与深度学习基础的开发者提供一套基于知识库的问答Seq2Seq模型完整代码实现帮助读者理解从数据预处理、模型构建到训练评估与部署的全流程。压缩包共21个文件约3.39MB包含7个py脚本、4个test与4个train数据文件、2个json样本、2个vocab词表及2个sta统计文件覆盖编码器-解码器、注意力机制、词表构建与训练配置等核心模块。已有2095人学习下载适合希望将Seq2Seq应用于智能问答、对话系统或知识库检索场景的读者参考。通过阅读代码可掌握分词、词嵌入、序列填充、损失函数与优化器选择等关键环节并了解如何将知识库信息融入模型生成答案以及使用BLEU、ROUGE等指标评估效果为后续引入Transformer或预训练模型优化提供实践基础。1. 从零手写知识库问答 seq2seq为什么今天还值得做一遍如果你手头有一批结构化的 FAQ 或领域问答对想做一个不依赖大模型 API、能离线跑、推理成本几乎为零的问答系统seq2seq 仍然是最值得亲手实现一遍的基线方案。它不像 RAG 那样需要向量库和检索链路也不像微调大模型那样吃显存一台普通笔记本 CPU 就能跑通训练和推理。知识库问答的本质是把用户问句映射到标准答案seq2seq 干的正是这件事编码器读入问句解码器逐词吐出答案。很多人一上来就上 RAG 知识库或问答智能体结果连最基本的序列到序列映射都没跑通过遇到 bad case 根本不知道问题出在数据、模型还是解码。这篇笔记按「数据怎么造 → 模型怎么搭 → 训练怎么调 → 推理怎么稳 → 坑在哪」的顺序把一套可复现的 Python 实现讲透适合想打牢 NLP 基础、或者需要一个轻量问答兜底方案的工程师。2. 知识库问答的数据准备从原始问答对到模型能吃的张量2.1 问答对清洗与最小数据格式知识库问答的数据通常来自 FAQ 文档、客服工单或人工整理的 Excel。不管来源是什么第一步都是统一成两列question和answer。我一般会先做三件事去掉全角空格和不可见字符、把连续空白压成一个空格、过滤掉答案长度小于 2 个字的脏数据。下面这个清洗脚本可以直接抄import re import pandas as pd def clean_text(text): if not isinstance(text, str): return text text.replace(\u3000, ).replace(\xa0, ) text re.sub(r\s, , text) return text.strip() def build_pairs(raw_path, out_path): df pd.read_excel(raw_path) if raw_path.endswith(xlsx) else pd.read_csv(raw_path) df df.rename(columns{df.columns[0]: question, df.columns[1]: answer}) df[question] df[question].map(clean_text) df[answer] df[answer].map(clean_text) # 过滤空值和过短答案 df df[(df[question].str.len() 1) (df[answer].str.len() 1)] # 去重同一个问题只保留第一条 df df.drop_duplicates(subset[question], keepfirst) df.to_csv(out_path, indexFalse, encodingutf-8-sig) print(f有效问答对: {len(df)}) return df build_pairs(raw_faq.xlsx, pairs.csv)逻辑说明clean_text处理中文里常见的全角空格和不可见字符这是后面分词出现空 token 的主要来源。drop_duplicates按问题去重很关键同一个问题对应多个答案会让模型学到矛盾的映射训练 loss 降不下去。参数上答案长度阈值我一般设 2太短的答案往往是「是」「好的」这类无信息回复留着只会污染训练集。2.2 词表构建与序列截断策略seq2seq 需要把文本转成 id 序列所以要先建词表。中文场景我一般按字切分不引入分词工具原因是知识库问答的领域词往往不在通用词典里按字切分反而更稳。词表保留频率最高的 N 个 token其余归入unk。四个特殊符号必须留pad、sos、eos、unk。from collections import Counter PAD, SOS, EOS, UNK 0, 1, 2, 3 def build_vocab(pairs, max_size8000, min_freq2): counter Counter() for q, a in zip(pairs[question], pairs[answer]): counter.update(list(q)) counter.update(list(a)) # 特殊符号固定占前四个 id vocab {pad: PAD, sos: SOS, eos: EOS, unk: UNK} for token, freq in counter.most_common(max_size): if freq min_freq: break if token not in vocab: vocab[token] len(vocab) return vocab def encode(text, vocab, max_len): ids [vocab.get(ch, UNK) for ch in text][:max_len - 1] return ids [EOS]逻辑说明max_size设 8000 是经验值知识库问答的字符种类通常远小于通用语料8000 足够覆盖。min_freq2过滤只出现一次的字减少unk之外的稀疏 id。encode里先截断再加eos保证每条序列都有结束标记否则解码器不知道什么时候停。max_len建议问句和答案分开设问句 32、答案 64 是常见起点具体看你的答案平均长度。2.3 用 Dataset 和 collate_fn 组装 batch变长序列必须 padding 到同一长度才能组 batch。这里用 PyTorch 的Dataset配合自定义collate_fn把 padding 放在 batch 内部做避免全局 padding 浪费显存。import torch from torch.utils.data import Dataset, DataLoader class QADataset(Dataset): def __init__(self, pairs, vocab, q_max32, a_max64): self.samples [] for q, a in zip(pairs[question], pairs[answer]): src encode(q, vocab, q_max) tgt [SOS] encode(a, vocab, a_max) self.samples.append((src, tgt)) def __len__(self): return len(self.samples) def __getitem__(self, idx): return self.samples[idx] def collate_fn(batch): srcs, tgts zip(*batch) src_len [len(s) for s in srcs] tgt_len [len(t) for t in tgts] src_pad torch.nn.utils.rnn.pad_sequence( [torch.tensor(s) for s in srcs], batch_firstTrue, padding_valuePAD) tgt_pad torch.nn.utils.rnn.pad_sequence( [torch.tensor(t) for t in tgts], batch_firstTrue, padding_valuePAD) return src_pad, tgt_pad, torch.tensor(src_len), torch.tensor(tgt_len)逻辑说明pad_sequence自动按 batch 内最长序列补齐padding_valuePAD保证 padding 位置不参与 loss 计算。返回的src_len和tgt_len后面传给模型做pack_padded_sequence这是 LSTM 编码器提速的关键不做的话 padding 也会被当成真实输入参与计算。注意tgt前面加了sos这是解码器的起始输入训练时输入是sos answer标签是answer eos错位一位。3. 用 PyTorch 搭一个带注意力的 seq2seq编码器、解码器与注意力模块3.1 编码器与解码器的结构选型seq2seq 的基础结构是编码器-解码器编码器把问句压成一个上下文向量解码器从这个向量出发逐词生成答案。纯 LSTM 的瓶颈在于长问句的信息会被压进一个固定维度向量答案一长就丢信息。所以实际能用的版本必须加注意力解码器每生成一个词都回头看一眼编码器的所有隐状态动态决定关注问句的哪部分。这就是标题里 attention module 的作用。我一般用单层双向 LSTM 做编码器、单层单向 LSTM 做解码器参数量小、CPU 也能跑。import torch.nn as nn class Encoder(nn.Module): def __init__(self, vocab_size, emb_dim128, hid_dim256, dropout0.3): super().__init__() self.embedding nn.Embedding(vocab_size, emb_dim, padding_idxPAD) self.rnn nn.LSTM(emb_dim, hid_dim, batch_firstTrue, bidirectionalTrue) self.fc nn.Linear(hid_dim * 2, hid_dim) self.dropout nn.Dropout(dropout) def forward(self, src, src_len): embedded self.dropout(self.embedding(src)) packed nn.utils.rnn.pack_padded_sequence( embedded, src_len.cpu(), batch_firstTrue, enforce_sortedFalse) outputs, (hidden, cell) self.rnn(packed) outputs, _ nn.utils.rnn.pad_packed_sequence(outputs, batch_firstTrue) # 双向隐状态拼接后投影回单向维度供解码器使用 hidden torch.tanh(self.fc(torch.cat([hidden[0], hidden[1]], dim1))) return outputs, hidden.unsqueeze(0)逻辑说明pack_padded_sequence让 LSTM 跳过 padding 位置enforce_sortedFalse省去手动排序。双向 LSTM 的hidden是两层hidden[0]是正向、hidden[1]是反向拼接后经fc投影回hid_dim这样解码器的隐状态维度对得上。dropout0.3在小数据集上是必要的正则知识库问答数据量通常不大不加 dropout 很容易过拟合。3.2 注意力模块的实现与维度对齐注意力模块的核心是算解码器当前隐状态和编码器每个位置的匹配分数softmax 归一化后加权求和。这里用最常见的点积注意力Luong 式实现简单且效果稳定。class Attention(nn.Module): def __init__(self, hid_dim): super().__init__() self.attn nn.Linear(hid_dim * 2, hid_dim) self.v nn.Linear(hid_dim, 1, biasFalse) def forward(self, hidden, encoder_outputs, mask): # hidden: [1, batch, hid_dim] - [batch, hid_dim] hidden hidden.squeeze(0) batch_size, src_len, _ encoder_outputs.shape hidden_rep hidden.unsqueeze(1).repeat(1, src_len, 1) energy torch.tanh(self.attn(torch.cat([hidden_rep, encoder_outputs], dim2))) attention self.v(energy).squeeze(2) # [batch, src_len] # mask 把 padding 位置的分数压到极小softmax 后接近 0 attention attention.masked_fill(mask 0, -1e10) return torch.softmax(attention, dim1)逻辑说明self.attn把解码器隐状态和编码器输出拼起来投影self.v压成一个标量分数。mask是关键padding 位置必须屏蔽否则注意力会分配到无意义的 pad 上生成结果出现重复或乱码。-1e10是常用技巧比-inf更安全避免 softmax 出现 NaN。返回的权重形状是[batch, src_len]后面用它加权求和编码器输出得到 context 向量。3.3 解码器逐步生成与 teacher forcing解码器每个时间步接收上一个词和 context 向量输出当前词的分布。训练时用 teacher forcing即把真实答案的上一个词喂进去加速收敛推理时用自己生成的词。class Decoder(nn.Module): def __init__(self, vocab_size, emb_dim128, hid_dim256, dropout0.3): super().__init__() self.embedding nn.Embedding(vocab_size, emb_dim, padding_idxPAD) self.rnn nn.LSTM(emb_dim hid_dim * 2, hid_dim, batch_firstTrue) self.fc nn.Linear(hid_dim * 3, vocab_size) self.attention Attention(hid_dim) self.dropout nn.Dropout(dropout) def forward_step(self, input_token, hidden, encoder_outputs, mask): input_token input_token.unsqueeze(1) # [batch, 1] embedded self.dropout(self.embedding(input_token)) attn_weights self.attention(hidden, encoder_outputs, mask) context torch.bmm(attn_weights.unsqueeze(1), encoder_outputs) rnn_input torch.cat([embedded, context], dim2) output, (hidden, cell) self.rnn(rnn_input, hidden) # 输出层拼接 output、context、embedding增强表达 pred self.fc(torch.cat([output.squeeze(1), context.squeeze(1), embedded.squeeze(1)], dim1)) return pred, hidden, attn_weights逻辑说明rnn_input是当前词 embedding 和 context 向量的拼接这是注意力 seq2seq 的标准做法。输出层把output、context、embedded三者拼接再投影比只用output效果更好这是我在多个问答数据集上验证过的。hidden在解码器里是[1, batch, hid_dim]因为解码器是单向单层。注意forward_step每次只处理一个时间步训练时循环调用这样代码清晰也方便推理时复用。4. 训练循环与解码策略让知识库问答真正能答对4.1 训练循环、loss 屏蔽与梯度裁剪训练时最关键的是 loss 只算非 padding 位置。如果 padding 也参与 loss模型会学会疯狂输出pad生成结果全是空白。import torch.optim as optim def train(model, loader, epochs30, lr1e-3, clip1.0, devicecpu): enc, dec model params list(enc.parameters()) list(dec.parameters()) optimizer optim.Adam(params, lrlr) criterion nn.CrossEntropyLoss(ignore_indexPAD) for epoch in range(epochs): enc.train(); dec.train() total_loss 0 for src, tgt, src_len, _ in loader: src, tgt src.to(device), tgt.to(device) mask (src ! PAD).float() optimizer.zero_grad() enc_out, hidden enc(src, src_len) input_token tgt[:, 0] loss 0 for t in range(1, tgt.shape[1]): pred, hidden, _ dec.forward_step(input_token, hidden, enc_out, mask) loss criterion(pred, tgt[:, t]) input_token tgt[:, t] # teacher forcing loss.backward() nn.utils.clip_grad_norm_(params, clip) optimizer.step() total_loss loss.item() print(fEpoch {epoch1} loss {total_loss/len(loader):.4f})逻辑说明ignore_indexPAD让 padding 位置的 loss 直接不算这是必须的。clip_grad_norm_裁剪梯度到 1.0LSTM 训练容易出现梯度爆炸不裁剪 loss 会突然变 NaN。teacher forcing 用tgt[:, t]作为下一步输入收敛快但要注意训练和推理的分布差异后面会讲怎么缓解。lr1e-3是 Adam 的常用起点数据量小可以降到 5e-4。4.2 贪心解码与 beam search 的取舍推理时解码策略直接决定答案质量。贪心解码每步取概率最大的词快但容易陷入局部最优beam search 保留 top-k 候选质量更好但慢 k 倍。def greedy_decode(enc, dec, src, src_len, vocab, max_len64, devicecpu): inv_vocab {v: k for k, v in vocab.items()} enc.eval(); dec.eval() with torch.no_grad(): src src.to(device) mask (src ! PAD).float() enc_out, hidden enc(src, src_len) input_token torch.tensor([SOS], devicedevice) result [] for _ in range(max_len): pred, hidden, _ dec.forward_step(input_token, hidden, enc_out, mask) top1 pred.argmax(1) token top1.item() if token EOS: break result.append(inv_vocab.get(token, unk)) input_token top1 return .join(result)逻辑说明inv_vocab把 id 映射回字符。循环里遇到eos就停避免生成无限长。max_len是兜底防止模型不输出eos时死循环。贪心解码适合对延迟敏感的场景知识库问答如果答案短、模式固定贪心通常够用。beam search 我一般只在离线评估时用线上还是贪心因为问答系统对响应时间敏感。4.3 用 BLEU 和人工抽检评估问答质量seq2seq 的评估不能只看 lossloss 低不代表答案对。我一般用 BLEU 做粗筛再人工抽检 50 条看实际效果。from nltk.translate.bleu_score import sentence_bleu def evaluate(model, pairs, vocab, n50): scores [] for i in range(min(n, len(pairs))): q pairs[question].iloc[i] ref pairs[answer].iloc[i] src torch.tensor([encode(q, vocab, 32)]) pred greedy_decode(model[0], model[1], src, torch.tensor([src.shape[1]]), vocab) score sentence_bleu([list(ref)], list(pred)) scores.append(score) print(f平均 BLEU: {sum(scores)/len(scores):.4f})逻辑说明sentence_bleu第一个参数是参考答案列表第二个是预测。BLEU 对短文本偏保守知识库问答里 0.3 以上就算可用。人工抽检重点看三类答案完全无关、答案部分正确、答案重复啰嗦。这三类分别对应数据问题、模型容量问题、解码问题排查方向不同。5. 知识库问答 seq2seq 的避坑与排查那些让我返工三次的问题5.1 生成结果全是重复词或unk现象推理时输出「的的的的」或者一连串unk。原因通常是两个一是训练时 padding 参与了 loss模型学会输出高频填充词二是词表太小很多字被映射成unk解码器只能反复输出unk。解决确认CrossEntropyLoss带了ignore_indexPAD把max_size从 8000 提到 15000 或直接不设上限min_freq降到 1。我遇到过一次是min_freq5太激进知识库里的专业术语全被过滤了。5.2 loss 降到 2.0 左右就不动了现象训练几十轮 loss 卡在 2.0 附近答案永远是那几句高频回复。原因一般是数据里存在大量相似问题对应不同答案模型学到的是「平均答案」。解决先做数据去重和冲突检测同一个问题如果有多个答案只保留最标准的那条如果问题高度相似但答案不同考虑合并或人工审核。另一个原因是模型容量不够把hid_dim从 256 提到 512或者编码器加一层。5.3 训练 loss 很低但推理答案完全不对现象训练 loss 降到 0.5但推理时答案驴唇不对马嘴。这是典型的 exposure bias训练用 teacher forcing推理用自己生成的词一旦第一步错后面全错。解决训练后期引入 scheduled sampling以一定概率用模型自己的预测作为下一步输入概率从 0 逐渐升到 0.5。另一个检查点是推理时的mask是否和训练一致我踩过一次推理忘了传mask注意力全跑到 padding 上。5.4 问句稍长就答非所问现象短问句答得还行超过 20 个字就开始乱答。原因是编码器把长问句压进固定向量时丢了信息注意力也没学好。解决确认pack_padded_sequence用对了src_len传的是真实长度不是 padding 后长度把问句max_len从 32 提到 48检查注意力 mask 是否屏蔽了 padding。如果还不行编码器换双向两层 LSTM但参数量会翻倍CPU 推理会变慢。5.5 中文标点和数字导致答案错乱现象问句里带「」「」或阿拉伯数字时答案质量明显下降。原因是这些字符在词表里频率低容易被归为unk。解决预处理阶段做字符归一化全角转半角、中文标点统一、数字保留但单独处理。我一般会把标点从词表里去掉训练前直接过滤因为知识库问答的语义主要靠汉字承载标点贡献很小还占词表。6. 让 seq2seq 问答再上一个台阶三个我反复用到的调优技巧第一个技巧是共享词表与权重绑定。问句和答案用的是同一套字符集没必要建两个词表。把编码器和解码器的embedding层共享参数量直接少一半小数据集上过拟合明显缓解。实现上就是dec.embedding.weight enc.embedding.weight一行代码的事但效果立竿见影。我做过对比共享权重后验证集 BLEU 平均涨 0.03 到 0.05。第二个技巧是答案模板后处理。seq2seq 生成的答案偶尔会多字少字但知识库问答的答案往往是固定模板比如「XX 的办理流程是……」。可以在推理后加一层规则匹配如果生成的答案和某个标准答案的编辑距离小于阈值直接替换成标准答案。这招在 FAQ 场景特别管用能把可用率从 70% 拉到 90% 以上。规则不复杂就是遍历标准答案算 Levenshtein 距离取最近的。第三个技巧是用小规模预训练词向量初始化 embedding。如果手头有领域语料用 word2vec 或 fasttext 训一版字向量初始化nn.Embedding的权重训练收敛会快很多。没有领域语料就用公开的中文字向量但要注意词表对齐对不上的字还是随机初始化。我一般会冻结 embedding 前 5 轮再解冻避免预训练权重被小数据带偏。验证这套方案是否值得投入我的习惯是先拿 500 条真实问答对跑通全流程看人工抽检的可用率能不能过 60%。过了就加数据、调参、上 beam search过不了就先回头查数据质量别急着改模型。seq2seq 问答的天花板由数据决定模型只是逼近这个天花板。这套代码我在三个不同领域的知识库上跑过最深的教训是数据清洗花的时间永远比调模型多但每一分钟都值。希望帮到你。本文还有配套的精品资源点击获取