【案例】新媒体评论分类与信息抽取 · LoRA 训练与上线

# -*- coding:utf-8 -*-
"""alpha/r 这个缩放系数到底在缩放什么：纯标准库数值实验，可直接运行。

h = W·x + (alpha/r)·B·A·x
很多人以为调 alpha 就是调「LoRA 的强度」——对，但它和 rank 绑在一起：
rank 调大时如果 alpha 不动，等效强度会掉。
"""


def matvec(mat, vec):
    return [sum(m * v for m, v in zip(row, vec)) for row in mat]


def lora_delta(A, B, x, alpha, r):
    """(alpha/r) * B(A x)。A: (r, in)，B: (out, r)。"""
    ax = matvec(A, x)
    bax = matvec(B, ax)
    scale = alpha / r
    return [scale * v for v in bax], scale


if __name__ == '__main__':
    x = [1.0, 2.0, 3.0, 4.0]

    # r=2 的一组权重
    A2 = [[0.1, 0.0, 0.0, 0.0], [0.0, 0.1, 0.0, 0.0]]
    B2 = [[1.0, 0.0], [0.0, 1.0], [1.0, 1.0]]

    for alpha in (2, 8, 32):
        delta, scale = lora_delta(A2, B2, x, alpha, r=2)
        print('r=2 alpha=%-3d scale=%-5.1f delta=%s' % (alpha, scale, [round(v, 3) for v in delta]))

    # 性质一：delta 与 alpha 成正比
    d1, _ = lora_delta(A2, B2, x, 2, 2)
    d2, _ = lora_delta(A2, B2, x, 8, 2)
    assert all(abs(b - 4 * a) < 1e-9 for a, b in zip(d1, d2)), 'alpha 翻 4 倍，增量翻 4 倍'
    print('\nalpha 从 2 到 8，增量正好放大 4 倍')

    # 性质二：B 初始化为 0，接上去的那一刻输出完全不变
    B_zero = [[0.0, 0.0], [0.0, 0.0], [0.0, 0.0]]
    delta0, _ = lora_delta(A2, B_zero, x, 32, 2)
    assert delta0 == [0.0, 0.0, 0.0]
    print('B 初始化为 0 -> 训练第 0 步的输出与原模型逐元素相同，不会一上来就把模型带偏')

    # 性质三：rank 翻倍而 alpha 不动，缩放系数减半
    print('\nrank   alpha   scale')
    for r in (2, 4, 8, 16, 32):
        print('%-6d %-7d %.2f' % (r, 32, 32 / r))
    assert 32 / 8 == 4.0 and 32 / 32 == 1.0
    print('\n把 rank 从 8 调到 32 想「学得更多」，scale 却从 4 掉到 1 —— ')
    print('两个方向互相抵消，所以调 rank 时通常让 alpha 跟着一起调，保持 alpha/r 不变')

    # 常见配比
    print('\n常见配比: alpha = 2r（scale=2）或 alpha = 4r（scale=4）')
    for r in (8, 16, 32):
        print('  r=%-3d alpha=%-3d scale=%.1f' % (r, 4 * r, 4 * r / r))

# -*- coding:utf-8 -*-
"""LoRA 到底省了多少：把可训练参数一层一层算出来，纯标准库，可直接运行。

这些数字不是背来的，是从 ChatGLM-6B 的 config 里算出来的：
hidden_size=4096，num_layers=28，query_key_value 把 Q/K/V 合成一个
(4096, 3*4096) 的矩阵。
"""

HIDDEN = 4096
LAYERS = 28
TOTAL_PARAMS = 6_173_286_400   # ChatGLM-6B 约 62 亿


def lora_params_per_layer(in_dim, out_dim, rank):
    """A: (in, r)，B: (r, out)。"""
    return in_dim * rank + rank * out_dim


def full_params_per_layer(in_dim, out_dim):
    return in_dim * out_dim


if __name__ == '__main__':
    qkv_out = 3 * HIDDEN
    print('query_key_value 原权重: %d x %d = %d' % (HIDDEN, qkv_out, HIDDEN * qkv_out))

    rank = 8
    a = HIDDEN * rank
    b = rank * qkv_out
    per_layer = lora_params_per_layer(HIDDEN, qkv_out, rank)
    print('\nrank=%d 时:' % rank)
    print('  A: %d x %d = %d' % (HIDDEN, rank, a))
    print('  B: %d x %d = %d' % (rank, qkv_out, b))
    print('  单层合计 = %d' % per_layer)
    assert a == 32768 and b == 98304 and per_layer == 131072

    total_lora = per_layer * LAYERS
    print('  x %d 层 = %d' % (LAYERS, total_lora))
    assert total_lora == 3_670_016

    pct = 100 * total_lora / TOTAL_PARAMS
    print('\n可训练占比 = %d / %d = %.4f%%' % (total_lora, TOTAL_PARAMS, pct))
    assert pct < 0.1

    # 对照：只改这 28 层的 qkv 做全参微调，也要 14 亿参数
    full_qkv = full_params_per_layer(HIDDEN, qkv_out) * LAYERS
    print('\n只全参微调这 28 层 qkv: %d 个参数，是 LoRA 的 %.0f 倍'
          % (full_qkv, full_qkv / total_lora))
    assert full_qkv / total_lora == 384.0

    # rank 翻倍，参数就翻倍——线性关系，不是指数
    print('\nrank   可训练参数   占比')
    prev = None
    for r in (1, 2, 4, 8, 16, 32, 64):
        n = lora_params_per_layer(HIDDEN, qkv_out, r) * LAYERS
        print('%-6d %-12d %.4f%%' % (r, n, 100 * n / TOTAL_PARAMS))
        if prev:
            assert n == prev * 2, 'rank 翻倍参数正好翻倍'
        prev = n

    # 存盘体积：float32 每个参数 4 字节
    print('\n适配器存盘体积（float32）: %.1f MB' % (total_lora * 4 / 1024 / 1024))
    print('完整模型存盘体积（float16）: %.1f GB' % (TOTAL_PARAMS * 2 / 1024 ** 3))
    assert total_lora * 4 / 1024 / 1024 < 20
    print('\n一个任务一个适配器，14 MB 一份 —— 这就是能同时上线几十个业务的原因')

# -*- coding:utf-8 -*-
"""给 ChatGLM 套上 LoRA：从加载底座到打印可训练参数占比。

整个「参数高效微调」的落地就这么几行。难点不在写，在于
看懂每一行改变了什么，出问题时知道去哪儿查。
"""
import torch
from peft import LoraConfig, TaskType, get_peft_model
from transformers import AutoModel, AutoTokenizer


def build_lora_model(pre_model: str,
                     rank: int = 8,
                     alpha: int = 32,
                     dropout: float = 0.1,
                     target_modules=None):
    """加载底座并挂上 LoRA 适配器。"""
    # trust_remote_code=True：ChatGLM 的建模代码在模型仓库里
    model = AutoModel.from_pretrained(pre_model, trust_remote_code=True)
    tokenizer = AutoTokenizer.from_pretrained(pre_model, trust_remote_code=True)

    config = LoraConfig(
        task_type=TaskType.CAUSAL_LM,   # 生成任务；分类任务要换成 SEQ_CLS
        inference_mode=False,           # 训练阶段必须 False，True 会把 dropout 关掉
        r=rank,
        lora_alpha=alpha,
        lora_dropout=dropout,
        # ChatGLM 把 Q/K/V 合成了一个 query_key_value 大矩阵，
        # 所以这里只写一个名字就等于同时改了 Q、K、V 三处
        target_modules=target_modules or ['query_key_value'],
    )

    model = get_peft_model(model, config)
    # 这行必看：可训练参数占比应当在 0.1% 以下。
    # 如果打出来是 100%，说明 target_modules 没匹配上任何层，LoRA 等于没生效
    model.print_trainable_parameters()
    return model, tokenizer


def count_trainable(model):
    """手动数一遍，别只信 print_trainable_parameters。"""
    trainable = sum(p.numel() for p in model.parameters() if p.requires_grad)
    total = sum(p.numel() for p in model.parameters())
    return trainable, total, 100 * trainable / total


def prepare_for_training(model, device: str):
    """训练前的三个开关，少一个都可能出问题。"""
    # ① 半精度：6B 模型用 float32 存不下，float16 在 24G 卡上才够
    model = model.half().to(device)
    # ② 有些底座默认关掉了 cache，训练时开着会和梯度检查点冲突
    model.config.use_cache = False
    # ③ LoRA 层本身用 float32 更稳：低秩矩阵数值小，半精度容易下溢成 0
    for name, param in model.named_parameters():
        if 'lora' in name.lower():
            param.data = param.data.to(torch.float32)
    return model


if __name__ == '__main__':
    model, tokenizer = build_lora_model('./ChatGLM-6B', rank=8, alpha=32, dropout=0.1)
    trainable, total, pct = count_trainable(model)
    print('可训练 %d / 总计 %d = %.4f%%' % (trainable, total, pct))
    print('缩放系数 alpha/r =', 32 / 8)

# -*- coding:utf-8 -*-
"""ChatGLM + LoRA 训练主脚本。

和 BERT 那两页的训练循环比，这里多了两样东西：
① 梯度累积——显存塞不下大 batch，就攒几步再更新一次；
② 半精度与 use_cache 开关——6B 模型的工程约束。
损失不用自己算：把 labels 一起喂进去，模型内部就把 -100 的位置跳过了。
"""
import os
import time

import torch
from tqdm import tqdm
from transformers import AutoModel, AutoTokenizer, default_data_collator, get_scheduler
from peft import LoraConfig, TaskType, get_peft_model

from data_loader import get_data
from glm_config import ProjectConfig

pc = ProjectConfig()


def save_model(model, cur_save_dir: str):
    """LoRA 存的是适配器，不是整个模型——十几 MB，不是十几 GB。"""
    os.makedirs(cur_save_dir, exist_ok=True)
    if pc.use_lora:
        merged_model = {}
        lora_state = model.state_dict()
        for k, v in lora_state.items():
            if 'lora' in k:      # 只挑带 lora 的权重
                merged_model[k] = v
        torch.save(merged_model, os.path.join(cur_save_dir, 'adapter_model.bin'))
    else:
        model.save_pretrained(cur_save_dir)


def evaluate_model(model, dev_dataloader):
    """验证集平均损失。生成任务没有现成的 F1，先用 loss 看趋势。"""
    model.eval()
    total_loss, n = 0.0, 0
    with torch.no_grad():
        for batch in dev_dataloader:
            loss = model(input_ids=batch['input_ids'].to(dtype=torch.long, device=pc.device),
                         labels=batch['labels'].to(dtype=torch.long, device=pc.device)).loss
            total_loss += float(loss)
            n += 1
    model.train()
    return total_loss / max(1, n)


def model2train():
    model = AutoModel.from_pretrained(pc.pre_model, trust_remote_code=True)
    tokenizer = AutoTokenizer.from_pretrained(pc.pre_model, trust_remote_code=True)

    if pc.use_lora:
        config = LoraConfig(task_type=TaskType.CAUSAL_LM,
                            inference_mode=False,
                            r=pc.lora_rank,
                            lora_alpha=pc.lora_alpha,
                            lora_dropout=pc.lora_dropout,
                            target_modules=['query_key_value'])
        model = get_peft_model(model, config)
        model.print_trainable_parameters()   # 占比应当远小于 1%

    # 半精度上卡；LoRA 层单独拉回 float32，低秩矩阵数值小，fp16 容易下溢
    model = model.half().to(pc.device)
    model.config.use_cache = False
    if pc.use_lora:
        for name, param in model.named_parameters():
            if 'lora' in name.lower():
                param.data = param.data.to(torch.float32)

    no_decay = ['bias', 'LayerNorm.weight']
    optimizer_grouped_parameters = [
        {'params': [p for n, p in model.named_parameters()
                    if not any(nd in n for nd in no_decay) and p.requires_grad],
         'weight_decay': pc.weight_decay},
        {'params': [p for n, p in model.named_parameters()
                    if any(nd in n for nd in no_decay) and p.requires_grad],
         'weight_decay': 0.0},
    ]
    optimizer = torch.optim.AdamW(optimizer_grouped_parameters, lr=pc.learning_rate)

    train_dataloader, dev_dataloader = get_data()
    # 有梯度累积时，「多少步更新一次」才是调度器该看的步数
    num_update_steps_per_epoch = max(
        1, len(train_dataloader) // pc.gradient_accumulation_steps)
    max_train_steps = pc.epochs * num_update_steps_per_epoch
    warm_steps = int(pc.warmup_ratio * max_train_steps)
    lr_scheduler = get_scheduler(name='linear',
                                 optimizer=optimizer,
                                 num_warmup_steps=warm_steps,
                                 num_training_steps=max_train_steps)

    print('每轮更新 %d 次，共 %d 次，预热 %d 次'
          % (num_update_steps_per_epoch, max_train_steps, warm_steps))

    loss_list, global_step, tic_train = [], 0, time.time()
    for epoch in range(1, pc.epochs + 1):
        for batch_idx, batch in enumerate(tqdm(train_dataloader, desc='epoch %d' % epoch)):
            # labels 一起传进去，损失由模型内部算，-100 的位置自动跳过
            loss = model(
                input_ids=batch['input_ids'].to(dtype=torch.long, device=pc.device),
                labels=batch['labels'].to(dtype=torch.long, device=pc.device)).loss
            loss_list.append(float(loss.detach()))

            # 除以累积步数：不除的话等效学习率会被放大 8 倍
            (loss / pc.gradient_accumulation_steps).backward()

            if (batch_idx + 1) % pc.gradient_accumulation_steps == 0:
                optimizer.step()
                lr_scheduler.step()
                optimizer.zero_grad()
                global_step += 1

                if global_step % pc.logging_steps == 0:
                    time_diff = time.time() - tic_train
                    print('step %d, epoch %d, loss %.5f, speed %.2f step/s'
                          % (global_step, epoch, sum(loss_list) / len(loss_list),
                             pc.logging_steps / time_diff))
                    tic_train = time.time()

                if global_step % pc.save_freq == 0:
                    save_model(model, os.path.join(pc.save_dir, 'model_%d' % global_step))

        dev_loss = evaluate_model(model, dev_dataloader)
        print('epoch %d 结束，dev loss = %.5f' % (epoch, dev_loss))
        save_model(model, os.path.join(pc.save_dir, 'model_epoch_%d' % epoch))

    print('训练结束')


if __name__ == '__main__':
    model2train()

# -*- coding:utf-8 -*-
"""加载 LoRA 适配器做推理，并把输出过一道解析闸。

生成式模型返回的是一段文本，不是一个类别。
「能解析」和「抽对了」是两件事，所以这里设两道闸：
① json.loads 能不能过；② 字段齐不齐。两道都记失败率，不要静默丢弃。
"""
import json
import os
import re
from typing import List

CODE_FENCE = re.compile(r'^```(?:json)?\s*|\s*```$', re.MULTILINE)
REQUIRED_SPO_KEYS = {'subject', 'predicate', 'object'}


def load_model(adapter_dir: str):
    """底座 + 适配器。底座只加载一次，适配器可以随时换成另一个业务的。

    torch / transformers / peft 都在函数内部导入：这样没装深度学习环境的机器
    也能直接跑本文件底部的解析闸自检。
    """
    from peft import PeftModel
    from transformers import AutoModel, AutoTokenizer

    from glm_config import ProjectConfig

    pc = ProjectConfig()
    tokenizer = AutoTokenizer.from_pretrained(pc.pre_model, trust_remote_code=True)
    model = AutoModel.from_pretrained(pc.pre_model, trust_remote_code=True)
    model = PeftModel.from_pretrained(model, adapter_dir)
    # merge_and_unload 把 BA 合并回原权重，推理不再多走一条支路，延迟回到底座水平。
    # 代价：合并后就换不了适配器了，要多任务共存就别合并
    model = model.merge_and_unload()
    return model.half().to(pc.device).eval(), tokenizer


def _device():
    from glm_config import ProjectConfig

    return ProjectConfig().device


def strip_fence(text: str) -> str:
    return CODE_FENCE.sub('', text.strip()).strip()


def parse_answer(raw: str, task: str = 'spo'):
    """把模型输出解析成结构化结果，返回 (ok, 结果或失败原因)。"""
    # ① 只要 Answer: 之后那一段。模型常把整个提示复读一遍
    if 'Answer:' in raw:
        raw = raw.split('Answer:')[-1]
    body = strip_fence(raw)

    # ② 第一道闸：能不能解析成 JSON
    try:
        data = json.loads(body)
    except json.JSONDecodeError as e:
        return False, 'JSON 解析失败: %s | 原文前 60 字: %s' % (e.msg, body[:60])

    # ③ 第二道闸：结构与字段
    if task == 'spo':
        if not isinstance(data, list):
            return False, '期望 list，实际 %s' % type(data).__name__
        for item in data:
            if not isinstance(item, dict):
                return False, '列表元素不是 dict'
            missing = REQUIRED_SPO_KEYS - set(item)
            if missing:
                return False, '缺字段 %s' % sorted(missing)
    elif task == 'classify':
        if not isinstance(data, list) or not all(isinstance(x, str) for x in data):
            return False, '分类任务期望字符串列表'
    return True, data


def inference(model, tokenizer, contexts: List[str], task='spo', max_new_tokens=300):
    import torch

    device = _device()
    results, fail_parse, fail_field = [], 0, 0
    with torch.no_grad():
        for context in contexts:
            ids = tokenizer.encode(context, return_tensors='pt').to(device)
            out = model.generate(input_ids=ids,
                                 max_new_tokens=max_new_tokens,
                                 do_sample=False)   # 抽取任务别采样，要可复现
            raw = tokenizer.decode(out[0], skip_special_tokens=True)
            ok, data = parse_answer(raw, task)
            if not ok:
                if 'JSON 解析失败' in str(data):
                    fail_parse += 1
                else:
                    fail_field += 1
            results.append({'ok': ok, 'result': data, 'raw': raw})
    total = max(1, len(contexts))
    print('解析失败率 %.1f%%，字段校验失败率 %.1f%%'
          % (100 * fail_parse / total, 100 * fail_field / total))
    return results


if __name__ == '__main__':
    # 不依赖模型也能验证解析闸本身是对的
    cases = [
        ('```json\n[{"predicate": "主演", "subject": "霸王别姬", "object": "张国荣"}]\n```', True),
        ('Answer: [{"predicate": "主演", "subject": "霸王别姬", "object": "张国荣"}]', True),
        ('[{"predicate": "主演", "subject": "霸王别姬"}]', False),      # 缺 object
        ('[{"predicate": "主演", "subject": "霸王别姬",', False),        # 被截断
        ('好的，我来帮你抽取：主演是张国荣', False),                      # 根本没输出 JSON
    ]
    for raw, expect in cases:
        ok, data = parse_answer(raw, 'spo')
        assert ok == expect, raw
        print('%s  %s' % ('通过' if ok else '拦下', data if ok else data))
    print('\n解析闸自检通过：5 条样例的判定与预期一致')

# -*- coding:utf-8 -*-
"""一个底座挂多个适配器：显存账与切换策略，纯标准库，可直接运行。

LoRA 真正改变上线方式的地方在这里：
底座只装一份，每个业务各自一个十几 MB 的适配器，按请求切。
前提是**不要 merge_and_unload**——合并了就换不回来了。
"""

BASE_GB = 11.5          # ChatGLM-6B float16 权重约 11.5 GB
ADAPTER_MB = 14.0       # rank=8 时适配器约 14 MB


class AdapterRegistry:
    """把「哪个业务用哪个适配器」这件事管起来，不要散在 if-else 里。"""

    def __init__(self):
        self._adapters = {}
        self._active = None

    def register(self, name: str, path: str, note: str = ''):
        if name in self._adapters:
            raise ValueError('适配器 %s 已注册，换个名字或先 unregister' % name)
        self._adapters[name] = {'path': path, 'note': note}

    def unregister(self, name: str):
        self._adapters.pop(name, None)
        if self._active == name:
            self._active = None

    def activate(self, name: str):
        """真实代码里对应 model.set_adapter(name)。"""
        if name not in self._adapters:
            raise KeyError('未注册的适配器: %s' % name)
        self._active = name
        return self._adapters[name]['path']

    @property
    def active(self):
        return self._active

    def memory_gb(self):
        """底座一份 + 每个适配器一份。"""
        return BASE_GB + len(self._adapters) * ADAPTER_MB / 1024

    def table(self):
        rows = [('业务', '适配器路径', '备注')]
        for name, info in self._adapters.items():
            rows.append((name, info['path'], info['note']))
        return rows


if __name__ == '__main__':
    reg = AdapterRegistry()
    reg.register('comment_cls', './checkpoints/cls/model_epoch_2', '评论分类')
    reg.register('spo_extract', './checkpoints/spo/model_epoch_2', '三元组抽取')
    reg.register('title_gen', './checkpoints/title/model_epoch_3', '标题生成')

    for row in reg.table():
        print('%-14s %-40s %s' % row)

    print('\n当前激活:', reg.active)
    path = reg.activate('spo_extract')
    print('切到 spo_extract ->', path)
    assert reg.active == 'spo_extract'

    # 重名直接拦住，不要覆盖别人的适配器
    try:
        reg.register('spo_extract', './other')
    except ValueError as e:
        print('\n重名被拦住:', e)

    # 显存账
    print('\n适配器数   显存占用')
    for n in (1, 3, 10, 50):
        r = AdapterRegistry()
        for i in range(n):
            r.register('task_%d' % i, './ckpt/%d' % i)
        print('%-10d %.3f GB' % (n, r.memory_gb()))

    r50 = AdapterRegistry()
    for i in range(50):
        r50.register('t%d' % i, './c/%d' % i)
    assert r50.memory_gb() < BASE_GB + 1
    print('\n50 个业务共存，比单独部署 50 份 6B 模型省了 %.0f GB' % (50 * BASE_GB - r50.memory_gb()))

    # 合并与不合并的取舍
    print('\n%-22s %-16s %s' % ('策略', '推理延迟', '能否切换适配器'))
    print('%-22s %-16s %s' % ('merge_and_unload', '与底座相同', '不能，已经焊死'))
    print('%-22s %-16s %s' % ('保留适配器', '略高（多一条支路）', '能，set_adapter 即可'))
    print('\n单业务高并发选合并；多业务共用一张卡选保留。')

# -*- coding:utf-8 -*-
"""上线骨架：从适配器目录到一个可调用的函数，四个 TODO 填完即可。

把「加载 -> 拼提示 -> 生成 -> 解析 -> 兜底」串成一条链，
每一环都留了失败出口，不让异常直接抛到调用方。
"""
import json
import os

# TODO-1: 底座路径
BASE_MODEL = os.environ.get('GLM_BASE', './ChatGLM-6B')
# TODO-2: 适配器目录
ADAPTER_DIR = os.environ.get('GLM_ADAPTER', './checkpoints/ptune/model_epoch_2')
# TODO-3: 与训练时逐字一致的提示模板
SYSTEM = '你现在是一个很厉害的阅读理解器，严格按照人类指令进行回答。'
INSTRUCTION = '找到句子中的三元组信息并输出成json给我：'
# TODO-4: 兜底返回值。宁可返回空结果，也不要把异常抛给上游
FALLBACK = []


def build_context(sentence: str) -> str:
    return ('Instruction: %s\nInput: %s\n\n%s\nAnswer: '
            % (SYSTEM, INSTRUCTION, sentence))


def load(base=BASE_MODEL, adapter=ADAPTER_DIR, merge=True):
    """返回 (model, tokenizer)。merge=True 延迟更低但不能再换适配器。"""
    from peft import PeftModel
    from transformers import AutoModel, AutoTokenizer

    tokenizer = AutoTokenizer.from_pretrained(base, trust_remote_code=True)
    model = AutoModel.from_pretrained(base, trust_remote_code=True)
    model = PeftModel.from_pretrained(model, adapter)
    if merge:
        model = model.merge_and_unload()
    return model.half().cuda().eval(), tokenizer


def extract(model, tokenizer, sentence: str, max_new_tokens=300):
    """一条句子进，结构化结果出。任何一环失败都返回 FALLBACK 并记日志。"""
    import torch

    context = build_context(sentence)
    try:
        with torch.no_grad():
            ids = tokenizer.encode(context, return_tensors='pt').cuda()
            out = model.generate(input_ids=ids,
                                 max_new_tokens=max_new_tokens,
                                 do_sample=False)
            raw = tokenizer.decode(out[0], skip_special_tokens=True)
    except Exception as e:
        print('[generate 失败] %s | 句子: %s' % (e, sentence[:30]))
        return FALLBACK

    body = raw.split('Answer:')[-1].strip()
    if body.startswith('```'):
        body = body.strip('`')
        body = body[4:] if body.lower().startswith('json') else body
    try:
        data = json.loads(body.strip())
    except json.JSONDecodeError as e:
        # 记原文，不要只记「解析失败」——排查时全靠这一行
        print('[解析失败] %s | 原文: %s' % (e.msg, body[:80]))
        return FALLBACK
    return data


def health_check():
    """上线前的三项检查，任何一项不过就别放量。"""
    problems = []
    if not os.path.isdir(ADAPTER_DIR):
        problems.append('适配器目录不存在: %s' % ADAPTER_DIR)
    if 'Answer: ' not in build_context('测试'):
        problems.append('提示模板与训练时不一致')
    if not isinstance(FALLBACK, (list, dict)):
        problems.append('兜底值类型不对，上游会崩')
    return problems


if __name__ == '__main__':
    print('提示模板:')
    print(repr(build_context('《霸王别姬》是由张国荣主演的一部电影')))
    problems = health_check()
    if problems:
        print('\n上线前检查未通过:')
        for p in problems:
            print('  -', p)
    else:
        print('\n上线前检查通过')