Agent 训练与反馈学习

分数 z = w × x + b
直接回答概率 p = 1 / (1 + e^(-z))
先检索概率 = 1 - p

单条损失 = -[y × ln(p) + (1-y) × ln(1-p)]
平均损失 = 各条损失之和 / 样本数量

mkdir training-demo
cd training-demo

"""从人工反馈训练一个两参数动作选择器。"""
import math


def probability(weights, features):
    # 特征一表示资料是否充分，特征二固定为 1，用来学习偏置。
    if len(weights) != 2 or len(features) != 2:
        raise ValueError("权重和特征都需要两个数")
    if not all(type(x) in (int, float) and math.isfinite(x)
               for x in [*weights, *features]):
        raise ValueError("输入需要有限数值")
    score = sum(w * x for w, x in zip(weights, features))
    # 分支写法避免很大的负分数导致指数溢出。
    return 1 / (1 + math.exp(-score)) if score >= 0 else math.exp(score) / (1 + math.exp(score))


def loss(weights, rows):
    # 交叉熵衡量：真实应当选择的动作，获得了多少概率。
    total = 0.0
    for features, label in rows:
        p = min(max(probability(weights, features), 1e-12), 1 - 1e-12)
        total -= label * math.log(p) + (1 - label) * math.log(1 - p)
    return total / len(rows)


def train(rows, epochs=50, rate=0.2):
    if not rows or type(epochs) is not int or epochs <= 0:
        raise ValueError("训练需要样本和正整数轮数")
    if type(rate) not in (int, float) or not math.isfinite(rate) or rate <= 0:
        raise ValueError("学习率需要正数")
    for features, label in rows:
        if type(label) is not int or label not in (0, 1):
            raise ValueError("标签需要整数 0 或 1")
        probability([0.0, 0.0], features)
    weights = [0.0, 0.0]
    for _ in range(epochs):
        gradient = [0.0, 0.0]
        for features, label in rows:
            error = probability(weights, features) - label
            for i in range(2):
                gradient[i] += error * features[i] / len(rows)
        # 按梯度的反方向走一小步，让下一轮损失降低。
        weights = [w - rate * g for w, g in zip(weights, gradient)]
    return weights


def split_groups(records, held_out):
    # 同一任务的不同轨迹必须在同一侧，防止训练答案混入评测。
    return ([value for task, value in records if task not in held_out],
            [value for task, value in records if task in held_out])


def main():
    # 1 表示直接回答；0 表示先检索资料。标签来自预先写好的反馈。
    rows = [([1.0, 1.0], 1), ([-1.0, 1.0], 0)]
    weights = train(rows)
    print(f"训练前损失：{loss([0.0, 0.0], rows):.4f}")
    print(f"训练后损失：{loss(weights, rows):.4f}")
    print(f"资料充分时直接回答概率：{probability(weights, [0.8, 1.0]):.4f}")
    print(f"资料不足时直接回答概率：{probability(weights, [-0.8, 1.0]):.4f}")


if __name__ == "__main__":
    main()

ls
python -X utf8 training_demo.py
python -X utf8 -m unittest -v test_training_demo.py

训练前损失：0.6931
训练后损失：0.1057
资料充分时直接回答概率：0.8526
资料不足时直接回答概率：0.1474

python -X utf8 training_demo.py