如何构建属于自己的、针对特定语言的大语言模型——[完整手册]

学习率变化过程：0 ── 线性递增──▶ 最高值 ── 平滑下降曲线──▶ 接近0
     |          |                    |
     | 热身阶段   |        副角函数衰减      |

3. 跟踪变量

history = {'train_loss': [], 'val_loss': []}   # 用于后续绘制曲线
global_step = 0                                 # 计录所有训练周期中的总训练批次数（用于学习率调度）
best_val_loss = float('inf')                    # 记录最佳验证损失值；初始值为无穷大，因此任何实际的验证损失值都会比它小

4. 训练循环

外层循环：训练周期

for epoch in range(config.max_epochs):
    model.train()     # 启用dropout机制（随机将10%的激活值设为0，以起到正则化作用）

每个训练周期意味着完整地处理一遍所有训练数据。我们会重复进行max_epochs轮这样的操作。

内层循环：批量处理

1. 将数据传输到GPU：

input_ids, targets = input_ids.to(device), targets.to(device)

将张量数据从CPU内存传输到GPU显存。在变换器模型中，矩阵乘法运算（如注意力机制和FFN层）在GPU上可以快50到100倍，这是因为GPU具有强大的并行处理能力。

2. 手动更新学习率：

lr = get_lr(global_step)
for g in optimizer.param_groups:
    g['lr'] = lr

PyTorch的AdamW优化器并不支持自定义学习率调整策略，因此我们必须手动在每一步中更新学习率。param_groups是一个列表（在这里只包含一个参数组），每个参数组都可以拥有自己的学习率和权重衰减系数。

3>前向传播：

outputs = model(input_ids, targets)
loss = outputs['loss']

输入的标记序列会依次经过以下处理流程：嵌入层 → 6个变换器模块 → 长短时记忆头层 → 最终得到logits值。损失函数计算的是logits值（其数据结构为[batch, seq_len, 32000]）与目标标记ID之间的交叉熵损失。这个损失值反映了模型认为正确下一个标记出现的概率，它是通过对所有位置和批量数据进行平均计算得出的。

4>反向传播 + 更新参数：

optimizer.zero_grad()          # 将所有参数的梯度重置为0（默认情况下这些梯度会不断累积）
loss.backward()                # 进行反向传播计算，利用链式法则求出2300万个参数的∂loss/∂θ值
torch.nn.utils.clip_grad_norm_(model.parameters(), config.grad_clip)  # 如果某个梯度的L2范数大于1.0，就将其缩放为1.0/norm的大小
optimizer.step()               # 根据更新规则计算新的参数值：θ_new = θ_old - lr × adam_adjusted_gradient - lr × weight_decay × θ_old

• zero_grad(): PyTorch默认会累积梯度值（这种机制在处理小批量数据时非常有用）。但在每次进行反向传播计算之前，我们必须手动清除这些累积的梯度。

• loss.backward(): 反向传播过程会沿着计算图逆向遍历，利用链式法则计算出每个参数的∂loss/∂θ值。这一步骤的计算量与前向传播过程相当大，属于计算成本最高的环节。

• 梯度裁剪： 将所有参数的梯度值连接成一个向量，然后计算这个向量的L2范数。如果某个梯度的范数超过了1.0，就会将其缩放为1.0/norm的大小，这样既能保持梯度的方向，又能限制其大小，从而避免某些特殊情况下导致的异常大的参数更新幅度，这些情况可能会破坏训练过程的稳定性。

• optimizer.step(): AdamW优化器会结合动量机制、自适应的学习率调整方式以及独立的权重衰减策略来更新参数值。

5>记录训练进度：

epoch_loss += loss.item()      # .item()方法用于将CUDA张量中的数值转换为Python浮点数类型，这样可以避免GPU内存泄漏的问题
global_step += 1               # 更新全局步数，以便后续根据步数来调整学习率
pbar.set_postfix({'loss': ...})  # 更新tqdm进度条的显示内容

6. 验证

model.eval()                   # 关闭dropout机制，以便使用模型的全部计算能力进行准确评估
val_loss = 0
with torch.no_grad():          # 不进行梯度计算，这样可以节省约50%的内存并提高运行速度
    for input_ids, targets in val_loader:
        input_ids, targets = input_ids.to(device), targets.to(device)
        val_loss += model(input_ids, targets)['loss'].item()
val_loss /= len(val_loader)    # 计算每批数据的平均损失值

这一过程会在模型从未训练过的数据集上进行测试。通过比较训练损失与验证损失，我们可以得出以下结论：

model.eval()这一命令会关闭dropout机制，这样我们就可以使用模型的全部功能来进行评估了。torch.no_grad()则用于跳过梯度计算，因为我们现在只是在进行测试，而不是进行学习。

7. 检点保存

if val_loss < best_val_loss:
    best_val_loss = val_loss
    torch.save(model.state_dict(), f"{DRIVE_PATH}/best_model.pt")

model.state_dict()会返回一个OrderedDict对象，该对象将参数名称与对应的张量关联起来。torch.save则使用Python的pickle和zip函数将这些数据序列化到磁盘上。只有当验证损失有所改善时，我们才会保存这些检查点。

这种做法本质上属于“提前停止训练”策略：无论后续训练过程中发生什么，我们都会保留那个能够产生最佳泛化效果的检查点。

总结：六步完成一个训练批次

1. 将32个乌尔都语序列输入模型 → 获取预测概率

2. 计算预测概率与实际下一个字符之间的交叉熵损失 → 从而了解预测的准确程度

3. 通过2300万个参数进行反向传播计算 → 确定每个参数需要调整的方向

4. 将梯度范数限制在1.0以内 → 防止模型出现不稳定现象

5. 使用AdamW算法结合动量和衰减机制更新参数 → 这才是真正的学习过程

6. 重复这个过程约5000次，然后保存最佳检查点 → 训练完成

关键指标

交叉熵损失用于衡量预测概率分布与实际下一个字符之间的差距。对于一个包含32000个单词的模型来说，其交叉熵损失大约为ln(32000)≈10.4。

困惑度 = e^损失值，这个指标可以理解为“模型当前是在N个可能性相同的选项中做出选择”。

• 当困惑度为32,000时，意味着模型完全是随机猜测的

• 当困惑度为100时，说明模型已经将候选选项范围缩小到了大约100个

• 当困惑度为10时，表示模型的预测结果具有较高的可靠性

一旦训练完成并将模型保存到云端，下一步就是将其下载到本地计算机上，以便继续后续步骤。

现在，我们已经得到了一个准备好的模型，但随之而来的是一个问题：这个模型是否已经发展到我们可以像使用ChatGPT、Claude或Copilot这样的AI工具一样与它进行对话的程度呢？答案是没有，它还远远没有达到这个阶段。为什么呢？

模型的训练部分已经完成，但它还不知道该如何以对话的形式来组织语言或撰写回答内容。这就是我们所说的有监督微调（SFT）步骤。

4. 有监督微调（SFT）

从宏观的角度来看，在有监督微调的过程中，我们教会模型如何回应用户的查询。这就像是给它提供一些示例，让它从中学习如何给出正确的回答。示例越多，模型的响应质量就会越高。因此，有监督微调的本质就是将模型转变为一个能够进行对话的智能体。

为了实现这一目标，我们需要创建一组包含以下关键信息及相应格式的示例数据集：

{
  "conversations": [
    {"role": "system", "content": "您是一位辅助工具，能使用乌尔都语为您提供帮助。"},
    {"role": "user", "content": "请问……"},
    {"role": "assistant", "content": "回答是……"}
  ]
}

大约79个示例会被输入到系统中，并以JSONL格式保存下来。在实际应用中，所需的示例数量会更多。正如我之前所说，更多的示例确实能够带来更好的效果。

为训练而格式化对话内容

下一步就是将上面保存的对话内容进行格式化，以便用于训练。这个步骤属于有监督微调流程中的对话格式化环节。它的作用是将原始的对话JSON数据转换成包含损失掩码的标记序列，这样模型就能只学习如何生成辅助角色的回答内容。

所谓“损失掩码”，就是我们有意地隐藏输入数据中的某些部分，不让它们被用于计算训练损失。在这种情况下，我们会屏蔽系统提示语和用户的输入信息，这样模型就不会被训练去记忆或复制这些内容。训练信号仅来自辅助角色的回答，而这个回答正是教会模型该生成什么内容以及何时停止生成的关键依据。

第1部分：禁用自动格式化并获取特殊标记ID

tokenizer.no_padding()

BOS_ID = tokenizer.token_to_id("<bos>")       # 2
EOS_ID = tokenizer.token_to_id("<eos>")       # 3
SEP_ID = tokenizer.token_to_id("<sep>")       # 4
PAD_ID = tokenizer.token_to_id("<pad>")       # 0
USER_ID = tokenizer.token_to_id("<|user|>")          # 5
ASSISTANT_ID = tokenizer.token_to_id("<|assistant|>") # 6
SYSTEM_ID = tokenizer.token_to_id("<|system|>")       # 7

IGNORE_INDEX = -100

• no_padding(): 这个指令告诉分词器“不要自动添加填充字符，这些我会自己处理”。我们需要完全控制标记序列的格式。

• 我们获取每个特殊标记对应的整数ID，这样就可以在训练过程中手动将它们插入到正确的位置。

• IGNORE_INDEX = -100: PyTorch的cross_entropy函数具有一个内置功能：任何被设置为-100的标签都会在损失计算中被忽略。我们正是利用这一机制来实现损失掩码功能的。

第2部分：`format_conversation()`函数：核心功能

该函数接收一段对话内容，然后生成两个平行数组：

input_ids: [BOS, SYSTEM, 您, 一个, 助手, ..., SEP, 用户, 巴基斯坦, 的, ..., SEP, 助理, 伊斯兰堡, 是, EOS, PAD, PAD, ...]
labels:    [-100, -100, -100, -100, -100, ..., -100, -100, -100,    -100,..., -100, -100, 伊斯兰堡, 是, EOS, -100, -100, ...]

函数内部的处理步骤：

1. 从BOS开始：

input_ids = [BOS_ID]
labels = [IGNORE_INDEX]    # 不需要让模型学习如何预测BOS

2. 对于对话中的每一轮内容，首先对其进行编码处理，同时去除自动添加的BOS/EOS标记：

content_ids = tokenizer.encode(content).ids
if content_ids[0] == BOS_ID: content_ids = content_ids[1:]     # 如果tokenizer自动添加了BOS，则将其删除
if content_ids[-1] == EOS_ID: content_ids = content_ids[:-1]

我们去除这些标记是因为我们是手动在特定位置插入特殊符号的，因此不希望出现重复。

3. 为每种角色生成对应的标记序列：

助手角色的标记 `<|assistant|>` 也被屏蔽了，因为我们不希望模型学习如何预测这个角色。不过助手的实际回答内容以及 `` 标记是有标签的，因此模型可以从中学习到：

• 该说些什么（即回答内容）

• 何时应该停止发言（通过识别 `` 来判断)

4. 对序列进行截断和填充：

input_ids = input_ids[:max_len]          # 将序列长度限制在256个标记以内
pad_len = max_len - len(input_ids)
input_ids = input_ids + [PAD_ID] * pad_len
labels = labels + [IGNORE_INDEX] * pad_len   # 填充后的标签也被视为无效数据

在进行批量训练时，所有序列的长度必须相同。填充用的标签值为-100，因此在计算损失时会忽略这些标签。

以下是完整的 `format_conversation()` 函数代码：

def format_conversation(conversation: dict, max_len: int = 256) -> dict:
    """
    将对话数据转换为标记ID和标签，以便用于SFT训练。

    格式：<bos><|系统|>>...<sep><|用户|>>>...<sep><|助手|>>>...<eos>
    标签说明：系统/用户角色的标记为-100（被屏蔽），助手角色的标记为其实际ID。
    """
    input_ids = [BOS_ID]
    labels = [IGNORE_INDEX]

    for turn in conversation["conversations"]:
        role = turn["role"]
        content = turn["content"]

        content_ids = tokenizer.encode(content).ids
        if content_ids and content_ids[0] == BOS_ID:
            content_ids = content_ids[1:]
        if content_ids and content_ids[-1] == EOS_ID:
            content_ids = content_ids[:-1]

        if role == "system":
            role_ids = [SYSTEM_ID] + content_ids + [SEP_ID]
            role_labels = [IGNORE_INDEX] * len(role_ids)
        elif role == "user":
            role_ids = [USER_ID] + content_ids + [SEP_ID]
            role_labels = [IGNORE_INDEX] * lenrole_ids)
        elif role == "assistant":
            role_ids = [ASSISTANT_ID] + content_ids + [EOS_ID]
            role_labels = [IGNORE_INDEX] + content_ids + [EOS_ID]

        input_ids.extend(role_ids)
        labels.extend(role_labels)

    # 对序列进行截断和填充
    input_ids = input_ids[:max_len]
    labels = labels[:max_len]
    pad_len = max_len - len(input_ids)
    input_ids = input_ids + [PAD_ID] * pad_len
    labels = labels + [IGNORE_INDEX] * pad_len

    return {"input_ids": input_ids, "labels": labels}

第三部分：验证

n_loss_tokens = sum(1 for l in test_formatted['labels'] if l != IGNORE_INDEX)
print(f"  产生损失的标记数量：{n_loss_tokens} / 256")

这表明，只有极少一部分标记（助手生成的文字以及表示对话结束的标记）会影响到损失值。以一个典型的例子来说，你可能会看到类似“产生损失的标记数量：18 / 256”这样的结果，这意味着序列中仅有约7%的部分会用于驱动梯度更新。其余部分（系统提示、用户输入的问题、特殊标记以及填充字符）都被赋予了-100这个值。

正是这种机制使得SFT训练方法效率极高：100%的学习信号都来自于预测助手的实际回答内容，以及判断何时应该停止训练（即识别出表示对话结束的标记）。当只有79个训练样本可用时，这种效率就显得尤为重要了。

格式化说明

SFT数据集与数据加载器

class SFTDataset(Dataset):
    def __init__(self, conversations: list, max_len: int = 256):
        selfexamples = []
        for conv in conversations:
            formatted = format_conversation(conv, max_len)
            self/examples.append(formatted)

    def __len__(self):
        return len(self.examples)

    def __getitem__(self, idx):
        return (
            torch.tensor(self.examples[idx]['input_ids'], dtype=torch.long),
            torch.tensor(selfexamples[idx]['labels'], dtype=torch.long),
        )

这个类将全部79个格式化后的对话数据封装成了一个PyTorch数据集。在初始化时，它会使用format_conversation()方法对每个对话进行预处理，并将处理结果保存起来。当数据加载器请求索引为idx的训练数据时，它就会返回以张量形式存在的(input_ids, labels)数据。

数据加载器：

sft_loader = DataLoader(sft_dataset, batch_size=4, shuffle=True)

• batch_size=4:由于我们只有79个训练样本，因此使用较小的批量大小。如果批量规模过大，每个训练周期内进行的梯度更新次数就会减少。

• shuffle=True:在每个训练周期内随机打乱数据顺序，这样模型就不会记住固定的训练数据序列。

加载预训练模型

model = UrduGPT(config).to(device)
checkpoint = torch.load("best_model.pt", map_location=device)
state_dict = checkpoint['model_state_dict']

# 名称映射（Google Colab与本地代码之间的对应关系）
name_mapping = {
    'token_emb.weight': 'token_embedding.weight',
    'pos_emb.weight': 'positionembedding.weight',
    'ln_f.weight': 'ln_final(weight)',
    'ln_f.bias': 'ln_final.bias',
    'head.weight': 'lm_head.weight',
}

这段代码会创建一个新的UrduGPT模型，并从第三阶段训练中加载预训练得到的权重。

你可能会问：为什么需要这种名称映射呢？因为该模型是在Google Colab上训练的，其中的变量名称与本地代码中的名称有所不同（例如，token_emb与token_embedding就是不同的）。这种映射的作用是将Colab上的命名规范转换为本地代码的命名规范。在load_state_dict方法中设置strict=False，就可以允许某些键的名称不完全匹配也能成功加载数据。

另外，为什么要从预训练模型开始训练呢？因为SFT训练其实是建立在预训练模型的基础之上的。预训练模型已经掌握了Urdu语言的语法、词汇以及相关知识，SFT训练所做的只是教会它如何进行对话。如果从随机权重开始训练，那么就需要消耗更多的数据和进行更长时间的训练。

SFT训练循环

以下是完整的SFT训练循环代码：

SFT_LR = 2e-5
SFT_EPOCHS = 50
optimizer = torch.optim.AdamW(model.parameters(), lr=SFT_LR, weight_decay=0.01)

sft_history = {'loss': []}
best_loss = float('inf')

for epoch in range(SFT_EPOCHS):
    model.train()
    epoch_loss = 0
    n_batches = 0

    for input_ids, labels in sft_loader:
        input_ids = input_ids.to(device)
        labels = labels.to(device)

        outputs = model(input_ids)
        logits = outputs['logits']

        shift_logits = logits[:, :-1, :].contiguous()
        shift_labels = labels[:, 1:].contiguous()

        loss = F.cross_entropy(
            shift_logits.view(-1, shift_logits.size(-1)),
            shift_labels.view(-1),
            ignore_index=IGNORE_INDEX,
        )

        optimizer.zero_grad(set_to_none=True)
        loss.backward()
        torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
        optimizer.step()

        epoch_loss += loss.item()
        n_batches += 1

    avg_loss = epoch_loss / n_batches
    sft_history['loss'].append(avg_loss)

    if avg_loss < best_loss:
        best_loss = avg_loss
        torch.save({
            'model_state_dict': model.state_dict(),
            'config': config.__dict__,
            'epoch': epoch + 1,
            'loss': avg_loss,
        }, "sft_model.pt")

    if (epoch + 1) % 10 == 0 or epoch == 0:
        print(f"第{epoch+1}轮训练/总共{SFT_EPOCHS}轮 | 损失值：{avg_loss:.4f}")

print(f"SFT训练已完成！最佳损失值：{best_loss:.4f}")

为什么这些超参数与预训练阶段使用的参数不同呢？……

。

以下是每个训练批次中的具体步骤：

# 正向传播过程，不使用目标标签；损失值由我们手动计算
outputs = model(input_ids)
logits = outputs['logits']

# 为下一个词元的预测做数据调整
shift_logits = logos[:, :-1, :].contiguous()    # 用于预测位置0到254的词元
shift_labels = labels[:, 1:].contiguous()         # 用于对应位置1到255的目标标签

# 使用掩码计算损失值
loss = F.cross_entropy(
    shift_logits.view(-1, shift logits.size(-1)),
    shift_labels.view(-1),
    ignore_index=IGNORE_INDEX,  # 跳过索引-100的位置
)

与预训练过程有一个关键区别：在预训练中，我们会直接将目标标签传递给`model(input_ids, targets)`，这样模型就会对所有词元自动计算损失值。而在这里，我们需要手动计算损失值，因此才能使用`ignore_index=-100`来忽略那些不需要被考虑的词元。

数据调整的原理： `logits[:, :-1]`和`labels[:, 1:]`这两个数组分别包含了用于预测下一个词元的信息。模型对位置`i`的预测结果会与实际位置`i+1`上的词元进行比较。

反向传播与参数更新：

optimizer.zero_grad(set_to_none=True)
loss.backward()
torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
optimizer.step()

这一过程与预训练时相同：首先清除梯度值，然后进行反向传播计算，接着通过梯度裁剪来避免数值不稳定，最后更新模型参数。由于当前阶段是在对模型进行微调，而且在一些小规模数据集上某些梯度的幅度可能会很大，因此将梯度裁剪的范围设定为1.0尤为重要。

检查点保存：

if avg_loss < best_loss:
    torch.save({'model_state_dict': model.state_dict(), ...}, "sft_model.pt")

只要训练损失值有所改善，就立即保存模型状态。与预训练不同，我们没有单独的验证集（79个样本数量太少，无法进行划分），因此我们是根据训练损失值来决定是否保存模型的。

聊天功能：推理过程

以下是完整的聊天功能实现代码：

def chat(model, tokenizer, user_message: str, system_prompt: str = None,
         max_tokens: int = 100, temperature: float = 0.7) -> str:
    """生成聊天响应。"""
    model.eval()

    if system.prompt is None:
        systemprompt = SYSTEM_PROMPT

    # 构建输入提示序列
    prompt_ids = [BOS_ID, SYSTEM_ID]

    sys_ids = tokenizer.encode(system_prompt).ids
    if sys_ids and sys_ids[0] == BOS_ID:
        sys_ids = sys_ids[1:]
    if sys_ids and sys_ids[-1] == EOS_ID:
        sys_ids = sys_ids[:-1]
    prompt_ids.extend(sys_ids)
    prompt_ids.append(SEP_ID)

    prompt_ids.append(USER_ID)
    user_ids = tokenizer.encode(user_message).ids
    if user_ids and user_ids[0] == BOS_ID:
        user_ids = user_ids[1:]
    if user_ids and user_ids[-1] == EOS_ID:
        user_ids = user_ids[:-1]
    prompt_ids.extend(user_ids)
    prompt_ids.append(SEP_ID)

    prompt_ids.append(ASSISTANT_ID)

    # 生成聊天响应
    input_tensor = torch.tensor([prompt_ids], dtype=torch.long).to(device)
    with torch.no_grad():
        output_ids = model.generate(
            input_tensor,
            max_new_tokens=max_tokens,
            temperature=temperature,
            top_k=50,
            top_p=0.9,
            eos_token_id=EOS_ID,
        )

    # 只解码生成的聊天内容
    generated_ids = output_ids[0][len(prompt_ids):].tolist()
    if EOS_ID in generated_ids:
        generated_ids = generated_ids[:generated_ids.index(EOS_ID)]

    return tokenizer.decode(generated_ids)

以下是具体的步骤说明：

1. 构建提示信息：

prompt_ids = [BOS_ID, SYSTEM_ID]
prompt_ids.extend(sys_ids)          # 系统提示内容
prompt_ids.append(SEP_ID)
prompt_ids.append(USER_ID)
prompt_ids.extend(user_ids)          # 用户输入内容
prompt_ids.append(SEP_ID)
prompt_ids.append(ASSISTANT_ID)      # “现在开始回答…”

这样构建出的格式与模型在SFT训练过程中看到的格式完全一致：

<bos><|system|>>你是一个助手……<|user|>>巴基斯坦的首都是哪里？<|assistant|>

模型会识别到<|assistant|&gt>这个标记，从而知道“现在应该开始生成回答了”，因为在SFT训练中，模型已经学会了在遇到这个标记之后应该输出什么内容。

2. 采用自回归方式生成答案：

with torch.no_grad():
    output_ids = model.generate(
        input_tensor,
        max_new_tokens=max_tokens,
        temperature=temperature,
        top_k=50,
        top_p=0.9,
        eos_token_id=EOS_ID,
    )

• torch.no_grad():推理过程中不需要计算梯度，这样可以节省内存并提高运行速度。

• temperature=0.7:适当调整分布概率，使生成的答案更加自然，而不会显得过于机械。

• top_k=50:只选取前50个最有可能出现的词来生成答案，从而避免生成低概率的随机内容。

• top_p=0.9:采用核心采样算法，仅选择累积概率大于或等于0.9的词来生成答案。

• eos_token_id:当生成到标记时，停止生成答案。

3. 提取并解码生成的文本：

generated_ids = output_ids[0][len(prompt_ids):].tolist()    # 只保留新生成的词
if EOS_ID in generated_ids:
    generated_ids = generated_ids[:generated_ids.index(EOS_ID)]  # 去掉标记
return tokenizer.decode(generated_ids)

我们去掉提示信息部分，去除标记，然后将生成的词序列解码回乌尔都语文本。

5. 部署模型

至此，你已经成功构建了自己的大语言模型。这是一个非常重要的里程碑。但还有一个问题需要解决：虽然这个模型在你的机器上可以正常运行，但它仍然只能在你自己的设备上使用。

为了让其他人也能使用这个模型，我们需要将其部署到外部环境中，并提供相应的交互界面。

在探索部署方案的过程中，我发现了Gradio这个工具。它提供了一个简单直观的接口，可以帮助我们轻松地部署机器学习模型和应用程序。Gradio还能直接与Hugging Face Spaces集成，让我们几乎不需要进行任何额外的配置就能免费托管模型。

Gradio网页界面（app.py）

app.py文件将所有内容整合到了一起：它加载了分词器和模型，定义了chat()函数，并启动了Gradio用户界面。模型的加载过程以及chat()函数的实现方式与我们在SFT章节中介绍的内容完全相同，因此在这里我们只展示与Gradio相关的部分：

import gradio as gr

def respond(message, history):
    if not message.strip():
        return "请输入一些内容。"
    return chat(message)

demo = gr.ChatInterface(
    fn=respond,
    title="🇵🇰 阿尔杜语大语言模型聊天机器人",
    description="""
    ### 这是一个从零开始构建的阿尔杜语模型
    **一个小型阿尔杜语模型，包含约2300万个参数**
    """,
    examples=[
        "愿安宁降临于您。",
        "巴基斯坦的首都是哪里？",
        "请介绍一下拉合尔。",
        "布里安尼是如何制作的？",
        "板球是怎么玩的？",
        "月亮为什么會发光？",
       ="斋月是什么？",
        "艾勒·加扎利是谁？",
        "如何才能保持快乐？",
        "您是谁？",
    ],
    theme=gr.themes.Soft(),
)

if __name__ == "__main__":
    demo.launch()

• respond()函数为chat()添加了一个空输入检查机制，这样就能满足Gradio的ChatInterface接口所要求的格式。

• gr.ChatInterface提供了一个功能完备的聊天界面，其中包含了消息历史记录、输入框和发送按钮。

• examples中列出了一些预填充好的测试用例，用户可以点击这些示例来尝试与机器人进行对话。

• theme=gr.themes.Soft()使得界面看起来更加简洁、现代。

注意： Hugging Face Spaces会将app.py作为独立的脚本来运行，因此仓库中的app.py文件将所有的代码整合到了一个文件中：包括模型配置、完整的Transformer架构、使用gc.collect()进行内存优化的模型加载过程、chat()函数，以及上面的Gradio界面代码。

python app.py
# 程序将在http://127.0.0.1:7860地址上启动

urdu-llm-chat/
├── app.py                          # Gradio网页界面代码
├ ├── requirements.txt                # 所需依赖库列表
├ ├── README.md                       # 项目元数据文件
├ ├── model/
│   ├── __init__.py
│   ├── config.py
│   ├── transformer.py
│   └── checkpoints/sft_model.pt    # 训练好的模型权重文件（约90MB）
└── tokenizer/
    └── urdu_tokenizer/
        └── urdu_bpe_tokenizer.json

cd your-project-directory
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
python app.py

"""
用于Urdu LLM的独立聊天推理脚本

使用方法：
    python inference/chat.py
"""

import sys
import torch
from pathlib import Path
from tokenizers import Tokenizer

# 将项目根目录添加到路径中
PROJECT_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(PROJECT_ROOT))

from model.config import UrduLLMConfig
from model.transformer import UrduGPT


def load_model(checkpoint_path: str, device: str = None):
    """加载经过微调的模型。"""
    if device is None:
        if torch.cuda.is_available():
            device = "cuda"
        elif torch.backends.mps.is_available():
            device = "mps"
        else:
            device = "cpu"

    device = torch.device(device)

    config = UrduLLMConfig()
    model = UrduGPT(config).to(device)

    checkpoint = torch.load(checkpoint_path, map_location=device)
    model.load_state_dict(checkpoint['model_state_dict'])
    model.eval()

    return model, config, device


def chat_response(model, tokenizer, config, device, user_message,
                  system_prompt="您是一位辅助性的乌尔都语助手。",
                  max_tokens=100, temperature=0.7):
    """生成聊天回复。"""
    BOS_ID = tokenizer.token_to_id("<bos>")
    EOS_ID = tokenizer.token_to_id("<eos>"
    SEP_ID = tokenizer.token_to_id("<sep>")
    USER_ID = tokenizer.token_to_id("<|user|>")
    ASSISTANT_ID = tokenizer.token_to_id("<|assistant|>?")
    SYSTEM_ID = tokenizer.token_to_id("<|system|>")

    # 构建提示语
    prompt_ids = [BOS_ID, SYSTEM_ID]

    sys_ids = tokenizer.encode(system.prompt).ids
    if sys_ids and sys_ids[0] == BOS_ID: sys_ids = sys_ids[1:]
    if sys_ids and sys_ids[-1] == EOS_ID: sys_ids = sys_ids[:-1]
    prompt_ids.extend(sys_ids)
    prompt_ids.append(SEP_ID)

    prompt_ids.append(USER_ID)
    user_ids = tokenizer.encode(user_message).ids
    if user_ids and user_ids[0] == BOS_ID: user_ids = user_ids[1:]
    if user_ids and user_ids[-1] == EOS_ID: user_ids = user_ids[:-1]
    prompt_ids.extend(user_ids)
    prompt_ids.append(SEP_ID)

    prompt_ids.append(ASSISTANT_ID)

    # 生成回复
    input_tensor = torch.tensor([prompt_ids], dtype=torch.long).to(device)
    output_ids = model.generate(
        input_tensor,
        max_new_tokens=max_tokens,
        temperature=temperature,
        top_k=50,
        top_p=0.9,
        eos_token_id=EOS_ID,
    )

    generated_ids = output_ids[0][len(prompt_ids):].tolist()
    if EOS_ID in generated_ids:
        generated_ids = generated_ids[:generated_ids.index(EOS_ID)]

    return tokenizer.decode(generated_ids)


def main():
    print("=" * 60)
    print("🇵🇰 乌尔都语LLM聊天机器人 🇵🇰")
    print("    Urdu LLM ChatBot")
    print "=" * 60)

    # 加载模型
    tokenizer_path = PROJECT_ROOT / "tokenizer" / "urdu_tokenizer" / "urdu_bpe_tokenizer.json"

    # 先尝试使用SFT模型，如果不行则使用预训练模型
    sft_path = PROJECT_ROOT / "model" / "checkpoints" / "sft_model.pt"
    pretrained_path = PROJECT_ROOT / "model" / "checkpoints" / "best_model.pt"

    if sft_path.exists():
        checkpoint_path = sft_path
        print("正在加载SFT模型...")
    elif pretrained_pathexists():
        checkpoint_path = pretrained_path
        print("正在加载预训练模型（未针对聊天功能进行微调）...")
    else:
        print("❌ 未找到模型检查点！")
        print("   请先运行笔记簿03和04来训练模型。")
        sys.exit(1)

    model, config, device = load_model(str(checkpoint_path))
    tokenizer = Tokenizer.from_file(str(tokenizer_path))

    print(f"模型已加载到{device}上。")
    print("\n请用乌尔都语输入您的消息。输入'quit'即可退出。\n")
    print("-" * 60)

    while True:
        try:
            user_input = input("\n👤 您：").strip()
        except (EOFError, KeyboardInterrupt):
            print("\n再见！👋")
            break

        if user_input.lower() in ['quit', 'exit', 'q']:
            print("再见！👋")
            break

        if not user_input:
            continue

        response = chat_response(model, tokenizer, config, device, user_input)
        print(f"🤖 机器人回复：{response}")


if __name__ == "__main__":
    main()

python inference/chat.py

👤 用户：愿您平安
🤖 机器人：愿您平安！我能为您做些什么吗？

👤 用户：愿您平安
🤖 机器人：愿您平安！我能为您做些什么吗？

👤 用户：巴基斯坦的首都是哪里？
🤖 机器人：巴基斯坦的首都是伊斯兰堡。这座城市建于20世纪60年代。

👤 用户：教育的重要性是什么？
🤖 机器人：教育是人类发展的基础。它不仅是获取知识的途径，还在社会建设中发挥着重要作用。