А если агенту не платить? Альтернативная механика обучения с подкреплением

NCHARS := 3      # x, y, dir
def create_rand_chars():
    chars := [0] * NCHARS
    chars[0] := random.randrange(5, 50)
    chars[1] := random.randrange(5, 50)
    chars[2] := random.randrange(0, 4)
    return chars

# изменяет хар-ки на заданные приращения
def update_chars(chars, deltas):
    for i, d in enumerate(deltas):
        chars[i] += d
        if i = 2:
            chars[i] := (chars[i] + 4) % 4  # periodic condition

# проверяет рядом ли коордианты с целью
def is_near(chars, aim):
    delta := abs(chars[0] - aim[0]) + abs(chars[1] - aim[1])
    return delta < 2

# генерирует цель от текущего положения агента (его характеристик)
def gen_aim(chars, radius):
    x := random.randrange(-radius, radius+1)
    y := random.randrange(-radius, radius+1)
    if x = 0 and y = 0:
        r := random.random()
        if r < 0.25:
            x += random.randrange(2, radius+1)
        elif r < 0.5:
            y += random.randrange(2, radius+1)
        elif r < 0.75:
            x -= random.randrange(2, radius+1)
        else:
            y -= random.randrange(2, radius+1)
    return {0: chars[0] + x, 1: chars[1] + y}

# делает туплу, как разницу координат до цели и направление - нужно для сохранения результатов движения
def aimed_chars_to_tuple(chars: list, aim: dict):
    dx = aim[0] - chars[0]
    dy = aim[1] - chars[1]
    return dx, dy, chars[2]
 [6]

ACTIONS := {'left': 0, 'right': 1, 'go': 2}
IDX_TO_ACT := {idx: action for action, idx in ACTIONS.items()}
ACT_LIST := list(ACTIONS.keys())

DX := [1, 0, -1, 0]
DY := [0, 1, 0, -1]

class Env:
    def apply_action(self, action, chars):
        affected = set()       # to save affected chars
        deltas = [0] * NCHARS
        if action = 'left':
            deltas[2] = -1
            affected.add(2)
        elif action = 'right':
            deltas[2] = 1
            affected.add(2)
        elif action = 'go':
            deltas[0] = DX[chars[2]]
            deltas[1] = DY[chars[2]]
            if deltas[0] <> 0:
                affected.add(0)
            if deltas[1] <> 0:
                affected.add(1)
        return deltas, affected
 [6]

class Agent:
    def __init__(self):
        self.chars = create_rand_chars()

        self.act_effects = {}      
        self.char_to_act = {}       
        self.opposite_actions = {}
 [6]

    def lstage1a(self, env: Env) -> dict:
        affected_chars := {}
        for _ in range(20):
            action := random.choice(ACT_LIST)
            deltas, changed := env.apply_action(action, self.chars)
            update_chars(self.chars, deltas)
            if changed:
                if action in affected_chars:
                    affected_chars[action] |= changed
                else:
                    affected_chars[action] := changed
        return affected_chars
 [6]

    def lstage1b(self, env: Env, affected_chars: dict):
        for action_of_interest, dims in affected_chars.items():
            data := []
            uniques := [set() for _ in range(NCHARS * 2)]
            i := 0
            while i < 30 * len(dims):
                action := random.choice(ACT_LIST)
                deltas, _ := env.apply_action(action, self.chars)
                if action = action_of_interest:
                    data.append(self.chars.copy() + deltas)
                    for j in range(NCHARS):
                        uniques[j].add(self.chars[j])
                        if deltas[j] <> 0:
                            uniques[j + NCHARS].add(deltas[j])
                    i += 1
                update_chars(self.chars, deltas)

            # describe effect: (condition, dimension, value)
            for dimension in dims:
                if len(uniques[NCHARS + dimension]) = 1:  # только одно значение => результат - константа
                    if action_of_interest in self.act_effects:
                        self.act_effects[action_of_interest].append((None, dimension, data[0][NCHARS + dimension]))
                    else:
                        self.act_effects[action_of_interest] = [(None, dimension, data[0][NCHARS + dimension])]

                # более сложная ситуация:
                #   есть несколько значений => вероятно есть какое-то условие, по которому они выбираются
                #   предположим, что это условие это определенное значение одного из параметров
                elif len(uniques[NCHARS + dimension]) > 1:
                    for val in uniques[NCHARS + dimension]:

                        # простая эвристика, считаем, что то измерение, где меньше уникальных - это и есть условие
                        subdata := [row[:NCHARS] for row in data if row[NCHARS + dimension] = val]
                        subuniques := [set() for _ in range(NCHARS)]

                        min_un := 1000
                        min_un_index := None
                        for j in range(NCHARS):
                            for row in subdata:
                                subuniques[j].add(row[j])
                            if len(subuniques[j]) < min_un:
                                min_un := len(subuniques[j])
                                min_un_index := j
                        for unique in subuniques[min_un_index]:
                            if action_of_interest in self.act_effects:
                                self.act_effects[action_of_interest].append(((min_un_index, unique), dimension, val))
                            else:
                                self.act_effects[action_of_interest] = [((min_un_index, unique), dimension, val)]

        self.cultivate_act_effects()
 [6]

    def cultivate_act_effects(self):
        # обращаем словарь действий: {action: effect} -> {affected_characteristic: action}
        for act, effects in self.act_effects.items():
            for eff in effects:
                char := eff[1]  # affected characteristic
                if char in self.char_to_act:
                    self.char_to_act[char].append((act, eff[2], eff[0]))
                else:
                    self.char_to_act[char] := [(act, eff[2], eff[0])]  # (action, value, condition)
 [6]

    def find_opp_actions(self):
        act_list := list(self.act_effects)
        for i, action in enumerate(act_list):
            if len(self.act_effects[action]) = 1:
                for j in range(i + 1, len(act_list)):
                    act2 := act_list[j]
                    if len(self.act_effects[act2]) = 1:
                        if self.act_effects[action][0][1] = self.act_effects[act2][0][1] and self.act_effects[action][0][2] = -self.act_effects[act2][0][2]:
                            self.opposite_actions[action] := act2
                            self.opposite_actions[act2] := action
 [6]

    def find_solution(self, aim, prev_action):
        actions = []

        for char, val in aim.items():
            delta := calc_delta(val, self.chars[char], char)
            actions.extend([list(data) for data in self.char_to_act[char] if (delta * data[1]) > 0])      # has the same sign

        for i, action in enumerate(actions):
            if action[2]:       # there is a condition
                char := action[2][0]
                val := action[2][1]
                if self.chars[char] <> val:     
                    delta := calc_delta(val, self.chars[char], char)
                    replace_actions := [list(data) for data in self.char_to_act[char] if (delta * data[1]) > 0]
                    if replace_actions:
                        actions[i] := replace_actions[0]     # the first approximation

        if prev_action in self.opposite_actions:
            actions := [action for action in actions if action[0] <> self.opposite_actions[prev_action]]

        action := random.choice(actions)[0] if actions else None
        return action
 [6]

    RADIUS := 5              # radius around agent coordinates to place aim
    N_EPISODES := 10000
    MAX_STEP := RADIUS * 3 + 5       # maximal number of steps to solve episode

    agent := Agent()
    env := Env()

    affected_chars := agent.lstage1a(env)
    agent.lstage1b(env, affected_chars)
    agent.find_opp_actions()

    memory := {}   
    n_success := 0
    traj_len := 0
    for ep in range(N_EPISODES):
        aim = gen_aim(agent.chars, RADIUS)
        nearest_step := None
        traj := []   # list of tuples (agent.chars, action)
        j_add := 0
        prev_action := None
        for step in range(MAX_STEP):
            if not nearest_step and is_near(agent.chars, aim):
                nearest_step := step
                j_add := 1
            action := agent.find_solution(aim, prev_action)
            traj.append((agent.chars.copy(), action))
            if action:
                deltas, _ := env.apply_action(action, agent.chars)
                update_chars(agent.chars, deltas)
            else:
                n_success += 1
                traj_len += step
                nearest_step := step
                break
            prev_action := action

        # only successful trajectories
        if nearest_step:
            i := nearest_step - 1
            j := 1 + j_add
            while i > 0:
                inds := aimed_chars_to_tuple(traj[i][0], aim)
                if inds not in memory:
                    memory[inds] := [j, traj[i][1]]
                elif memory[inds][0] = j:     # the same number of steps => add action if not presented
                    if traj[i][1] not in memory[inds]:
                        memory[inds].append(traj[i][1])
                elif memory[inds][0] > j:      # new variant is faster, replace old
                    memory[inds] = [j, traj[i][1]]
                else:       # old variant is better, skip current
                    break
                j += 1
                i -= 1

    print("n_success: {:.1f}%, mean length of traj: {:.1f}".format(100 * n_success / N_EPISODES, traj_len / n_success if n_success else 0))
    print("memory size:", len(memory))
    # теперь оставим только одно действие в словаре и уберем число шагов
    memory := {inds: actions[1] for inds, actions in memory.items()}
 [6]

class ExpNet(nn.Module):
    def __init__(self, input_dim, group_dim, n_actions):
        super().__init__()
        self.input_dim := input_dim
        self.group_dim := group_dim

        fact_inp_dim := input_dim + 4
        self.net := nn.Sequential(
            nn.Linear(fact_inp_dim * group_dim, n_actions)
        )

    def forward(self, comb):
        x := torch.sign(comb[:, :2])
        delt := torch.sign(comb[:, 0:1] - comb[:, 1:2] )
        sm := torch.sqrt( torch.sum(comb[:, :2]**2, dim=1))  + 0.1
        r := comb[:, :2] / sm.unsqueeze(1)
        rad := torch.log( sm.unsqueeze(1) + 1 )
        x := torch.cat((x, delt, r, rad), dim=1)
        g := comb[:, 2:]

        B := x.size(0)

        x_expanded := torch.einsum('bi,bj->bij', g, x)  
        x_expanded := x_expanded.reshape(B, -1)         
        return self.net(x_expanded)

model := ExpNet(2, 4, len(ACTIONS))
 [6]

    def solution_by_mem(self, aim, prev_action, memory):
        inds := aimed_chars_to_tuple(self.chars, aim)
        if inds in memory:
            return memory[inds]
        else:
            return self.find_solution(aim, prev_action)
 [6]