简介这是一份可基于自定义中文语料训练聊天机器人的Python项目合集覆盖Seq2Seq、SeqGAN、TensorFlow 2.0与PyTorch等多个实现版本并新增基于Horovod的大规模分布式训练方案适合具备一定深度学习基础、希望搭建智能客服或在线问答系统的开发者实践学习。压缩包共85个文件包含18个Python脚本、20个JavaScript文件、15个CSS样式等代码与前端展示文件兼备整体约37.94MB目录按不同框架版本清晰组织并附带README说明便于对照学习。目前已有1070人学习下载。从内容看不仅提供基础Seq2seq与SeqGAN聊天模型还更新了PyTorch的batch_size训练模式后续规划引入FAQ机器人及Transformer预训练模型实用性较强下载后可直接配置自己的语料开始训练也可参考分布式版本扩展至多机多卡环境。1. 能自己训练的中文聊天机器人Python 代码、自定义语料、本地跑通标题里的这个中文聊天机器人本质上是一套 Python 写的本地训练闭环你准备自己的对话语料跑数据清洗和格式化调用训练代码得到模型最后在同一个脚本里跟它聊天。它的价值不是让你绕开现成的对话接口而是让模型真的“学会”你喂进去的表达习惯。适合两类人一是想拿自己的聊天记录做一个说话风格接近自己的闲聊机器人二是刚入门 NLP 训练、暂时不想碰预训练语言模型和复杂训练框架的开发者。我的建议一直是先跑小模型几百条高质量语料就能看到明显的风格倾向比一上来就追求大模型更实在。2. 语料怎么准备先把聊天记录变成干净的 QA 对训练一个聊天机器人模型反而是最不需要花时间选的部分决定上限的是数据。我第一次做的时候把几百 MB 聊天记录直接喂进模型结果训出来的机器人只会回“嗯”“哈哈”和“不知道”。问题不在模型在语料原始记录里有大量图片占位符、链接、广告和单人自说自话这些都会变成训练噪声。所以这一章先把数据链路做扎实后面所有训练和调参才有意义。2.1 核心格式一问一答还是多轮上下文模型输入输出需要结构化的一问一答对原始聊天记录是一条时间流没法直接训练。第一版我建议用最简单的 JSON 格式每个样本包含context和reply两个字段[ {context: 你吃饭了吗, reply: 刚吃今天点的外卖特别咸}, {context: 周末去哪玩, reply: 准备去爬山你要不要一起} ]这种格式的好处是清洗和训练都直观。什么时候需要多轮上下文当你想让机器人记住“刚才说了什么”时把最近两轮对话拼成一个整体作为context再取最后一句作为reply。比如{context: 你吃饭了吗 我刚吃今天点的外卖特别咸 你呢, reply: 我还没吃在赶方案}多轮样本能让机器人具备上下文关联能力但代价是数据稀疏同样的语料量需要翻倍甚至更多才能训好。我一般建议第一版只做一问一答先把链路跑通再在最后一章说的验证阶段逐步加入上下文。2.2 清洗脚本从原始文本到 qa_train.json这里给一套可以直接抄的清洗脚本。假设你的原始聊天记录导出后是“昵称内容”的格式脚本会按发言顺序构造问答对。注意只处理你有权使用的数据不要拿别人的私聊记录做训练。import json import re from pathlib import Path def clean_text(text: str) - str: text re.sub(r\[图片\]|\[表情\]|\[链接\]|\[语音\], , text) text re.sub(rhttps?://\S, , text) text re.sub(r\d{3,}, num , text) # 连续数字替换成占位符 text re.sub(r\s, , text) return text.strip() def build_qa_pairs(raw_path: Path, out_path: Path): pattern re.compile(r^(.?)\s*[:]\s*(.*)$) msgs [] with open(raw_path, encodingutf-8) as f: for line in f: line line.strip() if not line: continue m pattern.match(line) if not m or not m.group(2): continue speaker m.group(1).strip() content clean_text(m.group(2)) if len(content) 2: continue msgs.append((speaker, content)) pairs [] for i in range(len(msgs) - 1): speaker, content msgs[i] next_speaker, next_content msgs[i 1] if speaker next_speaker: continue # 同一个人连续说话跳过 pairs.append({context: content, reply: next_content}) # 同一个问题保留多个回答防止某条万能回复被学死 seen {} for p in pairs: seen.setdefault(p[context], []).append(p[reply]) qa [] for context, replies in seen.items(): for reply in replies[:2]: # 每个问题最多保留 2 条不同回答 qa.append({context: context, reply: reply}) with open(out_path, w, encodingutf-8) as f: json.dump(qa, f, ensure_asciiFalse, indent2) print(f共生成 {len(qa)} 条问答对)这段脚本有几个关键参数值得说明。len(content) 2过滤掉单字回复因为“嗯”“哦”“好”这类内容会让模型学到偷懒策略。speaker next_speaker跳过同一个人连续说话的情况避免把一个人的独白拆成问答对。replies[:2]限制同一个问题最多保留两条不同回答否则一个高频问题出现 50 次模型会被某一句话带偏。脚本里把三位的数字串替换成num这样电话号码、金额这类数据不会被模型当成固定文本背下来。如果你手里的语料已经是“问题\t回答”这种两列格式把build_qa_pairs里的解析逻辑换成按\t切分即可其他清洗和过滤流程完全一样。2.3 词表按字切分限制在 8000 以内清洗完语料下一步是建词表。中文 NLP 里常见的选项是 jieba 分词但小语料场景我强烈建议按字切分。原因很直接几千条问答对按词切分后大量词只出现一次词表里一半是生僻词模型根本学不到信息按字切分后常用汉字就三千多个词表小、覆盖率反而高还天然能处理没见过的词。from collections import Counter class Vocab: def __init__(self, symbolsNone): self.itos list(symbols or [pad, bos, eos, unk]) self.stoi {s: i for i, s in enumerate(self.itos)} def add(self, symbol): if symbol not in self.stoi: self.stoi[symbol] len(self.itos) self.itos.append(symbol) def encode(self, text): return [self.stoi.get(ch, self.stoi[unk]) for ch in text] def decode(self, ids): return .join(self.itos[i] for i in ids) def save(self, path): with open(path, w, encodingutf-8) as f: json.dump(self.itos, f, ensure_asciiFalse) classmethod def load(cls, path): with open(path, encodingutf-8) as f: itos json.load(f) vocab cls() vocab.itos itos vocab.stoi {s: i for i, s in enumerate(itos)} return vocab def build_vocab(corpus_texts, min_freq2, max_vocab8000): counter Counter() for text in corpus_texts: counter.update(text) vocab Vocab() for ch, freq in counter.most_common(max_vocab): if freq min_freq: vocab.add(ch) return vocabmin_freq2表示只保留出现至少两次的字出现一次的单字基本是错别字或噪声。max_vocab8000是保险上限实际中文常用字通常到不了这个数但遇到语料里有大量num这类占位符时不会被截掉。词表前四个位置固定是padboseosunk后面所有训练代码都依赖这组固定索引不要自己重排。建完词表后把它保存成vocab.json问答对保存成qa_train.json数据准备阶段就算结束。下一步要决定用什么模型去学习这批数据。3. 训练方案怎么选生成式、检索式还是微调预训练模型很多人在这一步纠结该用什么网络结构其实先想清楚要什么效果更重要。对于“用自己的语料训练出自己想要的聊天机器人”常见做法有三条路从零训练一个 Seq2Seq 生成模型、做检索式回复、微调预训练语言模型。三条路对应完全不同的硬件需求和效果预期。3.1 生成式 Seq2Seq适合小语料能产生新句子Seq2Seq 是经典的编码器-解码器结构编码器读入用户的问题解码器逐字生成回答。它的核心价值在于“生成”模型不是从语料里拷贝原句而是根据学习到的字词组合习惯现编一句话。小语料下它产出的句子可能机械但风格会很明显比如你喂的是口语化聊天记录它就会倾向于用语气词和短句。为什么入门首选它而不是 Transformer因为 GRU 这类循环结构在小数据集上训练快、显存占用低CPU 也能跑。Transformer 效果上限更高但需要更多数据和调参技巧第一版就上 Transformer 容易在损失函数和收敛上浪费大量时间。我的做法是先用 GRU 把整条链路跑通确认数据没问题后再在同样的代码基础上替换成多头注意力。3.2 检索式回复先跑通流程、验证语料质量检索式方案不训练生成模型而是把用户输入和语料里的历史问题做相似度匹配找到最相似的问题直接返回它的答案。常见做法是先用 TF-IDF 或 BM25 把问题转成向量再算余弦相似度。这套方案的好处是几乎不需要训练几十条语料就能演示效果也不会出现模型胡说八道的情况因为它只能返回语料库里真实存在的话。坏处也明显它不能产生新句子用户换个说法就匹配不到。我的定位是把它当作“语料质量验证工具”如果你用检索式方案都匹配不出像样的回答说明语料清洗还不够干净这时候先别急着训练生成模型。3.3 微调预训练语言模型效果最好但环境门槛高如果你手里的语料超过一万条且机器有 GPU直接微调一个开源预训练语言模型是效果最稳的路。做法不是从头训练而是加载一个已经在海量文本上预训练过的基座模型再用你的问答语料做二次微调。为了控制显存占用常见做法是冻结大部分参数只训练新增的低秩适配层也就是 LoRA 这类技术。很多人问我用哪个训练框架从 SFTTrainerWrapper 到 GRPOTrainerWrapper 都是封装好的训练工具底层做的事情和第四章要写的反向传播没有本质区别。我建议刚上手时先用原生 PyTorch 把一个小模型训通知道梯度、损失函数、参数更新是怎么回事再去碰框架。否则环境一报错你分不清是数据问题、模型问题还是框架版本问题。3.4 三种方案对比方案是否需要 GPU语料量门槛能不能产生新句子1 万条 QA 对估算耗时Seq2Seq 小模型不需要CPU 可跑几百条就能起步能但句式偏机械30 分钟到 1 小时检索式回复不需要几十条就能演示不能只返回库内原句分钟级微调预训练语言模型建议显存 8G 以上1 万条起步更稳能更接近真人表达数小时到数天如果你的目标是快速看到一个能聊的机器人先做方案 B如果你希望它像人一样组织语言方案 A 是性价比最高的起点如果数据和硬件都充足方案 C 是最终形态。接下来的第四章以方案 A 为主线给出一套完整可运行的训练代码。4. 训练闭环数据加载、Seq2Seq 训练与本地交互这一章是从数据到模型的核心环节所有代码都可以直接保存成.py文件使用。我在设计时尽量精简去掉了不必要的封装保留最核心的dataset.py、train.py、chat.py三个脚本按顺序运行就能完成训练。4.1 数据加载器把 JSON 语料变成批量张量训练时不能一次把全部数据塞进模型需要按批次读取并给长短不一的句子做 padding。这里的数据加载器会读取第二章生成的qa_train.json把问题编码成张量同时给回答加上bos和eos标记。import json import torch from torch.utils.data import Dataset, DataLoader class ChatDataset(Dataset): def __init__(self, path, vocab, max_len32): self.data json.load(open(path, encodingutf-8)) self.vocab vocab self.max_len max_len self.pad_id vocab.stoi[pad] self.bos_id vocab.stoi[bos] self.eos_id vocab.stoi[eos] def __len__(self): return len(self.data) def __getitem__(self, idx): row self.data[idx] q self.vocab.encode(row[context])[:self.max_len] a self.vocab.encode(row[reply])[:self.max_len - 1] a [self.bos_id] a [self.eos_id] return torch.tensor(q), torch.tensor(a) def collate(batch, pad_id): questions, answers zip(*batch) q_lens [len(q) for q in questions] a_lens [len(a) for a in answers] max_q, max_a max(q_lens), max(a_lens) q_pad torch.full((len(questions), max_q), pad_id, dtypetorch.long) a_pad torch.full((len(questions), max_a), pad_id, dtypetorch.long) for i, (q, a) in enumerate(batch): q_pad[i, :len(q)] q a_pad[i, :len(a)] a return q_pad, a_pad, torch.tensor(q_lens)这里有两个细节值得注意。max_len32限制了问题和回答的最大长度实际聊天记录里超过 32 个字的长句很少截断后模型更专注学短句表达。回答做[:max_len - 1]是为了给eos留出位置否则回答被截满后结束标记会被挤掉模型就不知道什么时候该停止生成。collate函数返回q_lens是为了后续给编码器做序列打包不用手动排序PyTorch 的pack_padded_sequence支持乱序输入。4.2 模型与训练主循环GRU 编码器加解码器模型采用一个两层结构编码器用 GRU 读入问题解码器用 GRU 逐步生成回答。为了让第一版代码短一点这里没有做逐时间步的注意力计算而是把编码器的最终隐状态作为全局上下文拼到解码器的每一步输入里够用且好理解。import torch.nn as nn import torch.nn.utils.rnn as rnn_utils class Seq2Seq(nn.Module): def __init__(self, vocab_size, embed_size64, hidden_size128): super().__init__() self.embedding nn.Embedding(vocab_size, embed_size, padding_idx0) self.encoder_gru nn.GRU(embed_size, hidden_size, batch_firstTrue) self.decoder_gru nn.GRU(embed_size hidden_size, hidden_size, batch_firstTrue) self.fc_out nn.Linear(hidden_size, vocab_size) def forward(self, q_ids, q_len, a_ids): q_emb self.embedding(q_ids) packed rnn_utils.pack_padded_sequence( q_emb, q_len.cpu(), batch_firstTrue, enforce_sortedFalse ) _, hidden self.encoder_gru(packed) a_emb self.embedding(a_ids) context hidden[-1].unsqueeze(1).expand(-1, a_emb.size(1), -1) dec_out, _ self.decoder_gru(torch.cat([a_emb, context], dim-1), hidden) logits self.fc_out(dec_out) return logitsseq2seq的forward里hidden形状是[1, batch, hidden_size]取hidden[-1]得到最后一个时间步的隐状态。context被复制到回答序列的每一个时间步相当于告诉解码器“整段问题的概括信息在这里”。fc_out把解码器输出映射回词表大小得到每个位置上每个词的概率。整个模型加上torch.utils.data和优化器训练主循环如下import torch from torch import nn device torch.device(cuda if torch.cuda.is_available() else cpu) vocab Vocab.load(vocab.json) dataset ChatDataset(qa_train.json, vocab) loader DataLoader(dataset, batch_size32, shuffleTrue, collate_fnlambda b: collate(b, 0)) model Seq2Seq(len(vocab)).to(device) optimizer torch.optim.AdamW(model.parameters(), lr1e-3) criterion nn.CrossEntropyLoss(ignore_index0) for epoch in range(30): model.train() total_loss 0.0 for q_ids, a_ids, q_len in loader: q_ids, a_ids, q_len q_ids.to(device), a_ids.to(device), q_len.to(device) optimizer.zero_grad() decoder_input a_ids[:, :-1] target a_ids[:, 1:] logits model(q_ids, q_len, decoder_input) loss criterion( logits.reshape(-1, logits.size(-1)), target.reshape(-1) ) loss.backward() nn.utils.clip_grad_norm_(model.parameters(), 1.0) optimizer.step() total_loss loss.item() print(fepoch {epoch1}, loss {total_loss / len(loader):.4f}) torch.save(model.state_dict(), chatbot.pt)这里decoder_input a_ids[:, :-1]用回答去掉最后一个标记作为输入target a_ids[:, 1:]用回答去掉第一个标记作为预测目标这是标准的 teacher forcing 训练方式解码器每一步都看到正确的前一个词。ignore_index0告诉损失函数忽略所有 padding 位置不会因为补齐的空位影响梯度。clip_grad_norm_(..., 1.0)是梯度裁剪防止长序列训练时梯度爆炸损失突然变成nan的第一排查点就在这里。4.3 本地交互加载权重用采样生成回答训练完的模型要能直接聊天关键是推理时没有正确答案可看只能把上一步生成的词作为下一步输入。解码时用随机采样而不是贪心取最大概率可以明显减少复读问题。torch.no_grad() def predict(model, text, vocab, max_len32, temperature0.8): model.eval() ids vocab.encode(text)[:max_len] q torch.tensor([ids], dtypetorch.long, devicedevice) q_len torch.tensor([len(ids)], dtypetorch.long, devicedevice) q_emb model.embedding(q) packed rnn_utils.pack_padded_sequence( q_emb, q_len.cpu(), batch_firstTrue, enforce_sortedFalse ) _, hidden model.encoder_gru(packed) out_ids [vocab.stoi[bos]] for _ in range(max_len): last torch.tensor([[out_ids[-1]]], dtypetorch.long, devicedevice) last_emb model.embedding(last) context hidden[-1].unsqueeze(1) dec_out, hidden model.decoder_gru( torch.cat([last_emb, context], dim-1), hidden ) logits model.fc_out(dec_out)[0, -1] / temperature p torch.softmax(logits, dim-1) next_id torch.multinomial(p, 1).item() if next_id vocab.stoi[eos]: break out_ids.append(next_id) return vocab.decode(out_ids[1:]) if __name__ __main__: model Seq2Seq(len(vocab)).to(device) model.load_state_dict(torch.load(chatbot.pt, map_locationdevice)) while True: user_input input(you: ) if user_input in {exit, quit}: break print(bot:, predict(model, user_input, vocab))temperature是采样参数值越小越保守越大越发散。语料质量一般时用 0.8回答接近原文且不容易乱串语料干净时可以调到 1.0让模型多尝试新组合。torch.multinomial按概率分布随机抽取一个 token即使某一个词的分数最高也有机会选到第二名这就打破了复读循环。如果跑完训练后回答里频繁出现unk说明词表构建阶段min_freq2过滤得太狠可以降为 1 保留更多生僻字。5. 避坑与排查损失下降不代表模型学会了聊天训练代码能跑通只是第一步真正折磨人的是训练过程一切正常、loss 稳定下降聊天结果却完全不能用。这一章把最常见的五类问题整理成“现象 → 原因 → 解决”的记录照着排查能省下大量试错时间。5.1 语料里高频万能回复太多模型学会偷懒现象训练完成后无论用户问什么机器人都回答“嗯”“哈哈”“不知道”偶尔有一句像样的但很快又绕回高频短句。原因原始聊天记录里这类万能回复出现频率极高比如一个人敷衍时连续发几十个“嗯”。模型发现只要输出“嗯”无论什么输入都能拿到较低的损失于是把这条路走成了捷径。解决在清洗阶段单独统计高频回复把出现次数超过总样本 1% 的短回复要么删除要么降低采样数量。我在 2.2 节脚本里只做了长度过滤这一步需要额外加逻辑Counter统计所有reply的出现次数对频次过高的回复做截断。5.2 语料被某一方的说话风格主导现象机器人整体风格偏像聊天记录里话最多的那个人哪怕你希望它学的是双方对话。原因问答对是按“谁回复谁”构造的如果一方发言数量是另一方的五倍模型自然学到话多一方的用词习惯。解决构造问答对时按speaker统计发言次数把发言过多一方的样本做随机降采样尽量让两个角色比例接近 1:1。如果你只想学某个特定对象的说话风格那就只保留该对象作为回复方的样本把它的发言全部当作reply另一方发言当作context。5.3 损失一直不降或者突然变成 nan现象训练了十几轮loss 在 8 左右不降或者某一步开始变成nan之后全部是nan。原因学习率太大导致梯度更新越过最优区域嵌入层输入里混入了异常大的 token 值梯度在长序列反向传播中爆炸。解决先把学习率从1e-3降到1e-4试一轮确认 loss 能下降后再逐步调大。同时检查词表构建是否越界Vocab.encode会不会返回超出词表长度的索引。最后确认梯度裁剪已经生效裁剪阈值clip从 1.0 降到 0.5 也会有效。5.4 训练集太小模型把语料原句背下来现象训练完问一句训练集里出现过的原问题回答一字不差但换一种说法回答立刻变得语无伦次。原因语料量太少模型参数容量相对过剩走的是死记硬背而不是学语义组合。解决先看语料总条数少于 500 条的优先补数据而不是改模型。补数据的方式可以是从公开的闲聊语料里筛同风格内容也可以把同一问题的历史不同回复都保留而不是只留两条。模型侧的正则手段是加大 dropout但小语料下 dropout 治标不治本。5.5 推理速度慢CPU 环境下每句话要等很久现象训练在 CPU 上跑完轮到聊天时每生成一个字要卡几百毫秒。原因解码器是逐时间步生成每生成一个字都要过一次完整 GRU 计算回答越长耗时越大。解决降到hidden_size64、embed_size32推理速度能提升一倍左右。另一种做法是在predict函数里限制max_len20回答超过 20 个字直接截断闲聊场景完全够用。如果只是自己本地玩不必追求跑满 32 字的长回复。6. 验证与进阶用固定问题集做回归测试再做多轮上下文模型训练不是终点怎么验证改数据后效果变好还是变差才是关键。我现在的习惯是准备一组固定问题每次调整语料或参数后都跑一遍同样的输入对比前后回答风格和准确率。没有这步你很难判断一次清洗到底是改善了数据还是把原本正常的内容误删了。fixed_questions [你在吗, 你吃饭了吗, 今天天气怎么样, 周末去爬山吗] for q in fixed_questions: reply predict(model, q, vocab) print(q, -, reply)看结果时重点记录三类异常空回答、训练集原句复读、答非所问。固定测试集不需要很大十到二十个问题足够。下一步值得做的是多轮上下文。做法是在dataset.py里把样本的context变成“历史轮次 当前问题”拼接后的字符串训练逻辑不变。我的建议是只拼最近两轮历史太多会让问题变成一长串噪声。多轮数据至少要准备五百条才能看到明显效果如果训练集不够会让整组任务变得更稀疏反而损害单轮效果。如果小模型已经满足不了你最后一条优化路径是直接升级到预训练语言模型微调用 LoRA 控制显存占用。这条路会带来新的踩坑点比如基座模型加载和分词器对齐但你在前五章积累的数据清洗和损失排查经验完全能复用。我自己的教训是不要一上来就把模型换大先用这套小模型把语料验证到位模型换来换去最后卡住你的永远是数据。希望帮到你。本文还有配套的精品资源点击获取