PEA* 알고리즘 - 이렇게 간단하다구요? 그럼에도 메모리 효율적이라구요?

안녕하세요. 오늘은 A* 알고리즘의 변형 중 하나인 Partial Expansion A*(PEA*) 알고리즘을 공부해보려고 해요!

• YOSHIZUMI, Takayuki; MIURA, Teruhisa; ISHIDA, Toru. A* with Partial Expansion for Large Branching Factor Problems. In: AAAI/IAAI. 2000. p. 923-929.

CBS 알고리즘을 공부하면서 알게되었던 EPEA* 알고리즘을 공부하기 위해 기본이 되는 PEA* 알고리즘을 공부했던 것인데요.

이제야 정리하게 되네요...

 

PEA* 알고리즘은 메모리를 매우 많이 사용하게 되는 Multiple Sequence Alignment(MSA) 문제를 풀기 위해 고안되었습니다.

PEA* 알고리즘이 어떻게 메모리 복잡도를 놓쳤는지 자세히 알아보도록 하겠습니다!

---

0. PEA* 알고리즘 개요

PEA* 알고리즘은 Cut Off와 F ` 이라는 개념을 추가하여 OPEN 리스트가 무분별하게 커지는 것을 막음으로써 메모리 사용량을 줄였습니다.

 

간단하게 설명드리면 Cut Off는 OPEN 리스트에 추가할 노드들을 필터링 하기 위해 필요한 개념이고,

F `은 필터링되어서 OPEN 리스트에 추가되지 못한 노드들을 나중에라도 탐색하기 위하여 필요한 개념입니다.

 

A* 알고리즘에서 OPEN 리스트에서 노드를 선택한 후, 해당 노드와 연결된 모든 노드가 OPEN 리스트에 추가될 수 있는 반면, PEA* 알고리즘에서는 OPEN 리스트에서 선택된 노드의 F ` + C(Cut Off 값) 보다 f ` 값이 작은 노드들만 OPEN 리스트에 추가될 수 있습니다.

 

PEA* 알고리즘을 진짜 간단하게 설명드렸는데요.

아마 F `값이 f ` 값과 어떻게 다른지, F `이 언제 갱신되는지 등이 궁금하실 것 같습니다.

PEA* 알고리즘의 동작 방식을 구체적으로 살펴보도록 하겠습니다!

 

1. PEA* 알고리즘 동작 방식

1. Cut Off 값 C를 설정합니다.
2. 시작 노드 s를 OPEN에 추가하고 평가 함수 f `(n)을 이용해서 f ` 값을 계산합니다.
3. OPEN에서 F ` 값이 가장 작은 노드 n을 추출합니다.
   • f ` 값이 최소인 노드들이 있다면 랜덤하게 아무 노드나 선택합니다.
   • 하지만 G에 포함된 노드가 있다면 해당 노드를 먼저 고릅니다.
4. 만약 선택된 노드 n이 G에 포함된다면 알고리즘을 종료합니다.
5. Connected(n)을 f ` 값이 F `(n) + C보다 작거나 같은 집합 S_le, f `값이  F `(n) + C보다 큰 집합 S_gt 로 나눕니다.
6. S_le의 모든 노드들(n_l)에 대해서 다음 과정을 수행합니다.
   • 만약 n_l이 OPEN 리스트에도 포함되어 있지 않고, CLOSED에도 포함되어 있지 않다면
     - g `(n_l)을 g `(n) + c(n, n_l)로 계산합니다.
     - F `(n_l)을 g `(n_l) + h `(n_l)로 계산합니다.
     - n_l을 OPEN 리스트에 추가합니다.
   • 만약 n_l이 OPEN 리스트에 포함되어 있고, g(n) + c(n, n_l) < g(n_l)이라면
     - g `(n_l)을 g `(n) + c `(n, n_l)로 계산합니다.
     - F `(n_l)을 g `(n_l) + h `(n_l)로 계산합니다.
   • 만약 n_l이 CLOSED 리스트에 포함되어 있고, g(n) + c(n, n_l) < g(n_l)이라면
     - g `(n_l)을 g `(n) + c `(n, n_l)로 계산합니다.
     - F `(n_l)을 g `(n_l) + h `(n_l)로 갱신한합니다.
     - n_l을 CLOSED 리스트에서 제거하고 OPEN 리스트에 추가합니다.
7. S_gt가 공집합이라면 n을 CLOSED에 추가합니다.
8. S_gt가 공집합이 아니라면 F `(n) 을 S_gt 에 속한 노드 중 가장 작은 f `값으로 치환하고 노드 n을 다시 OPEN 리스트에 추가합니다.
9. 3로 돌아갑니다.

 

2. 예제 코드

이번에도 heuristic 함수 h `(x) = 0으로 준비했습니다...!

from dataclasses import dataclass
from typing import List
import heapq


@dataclass
class Node:
    id: int
    f_val: float = float("inf")
    F_val: float = float("inf")
    parent_id: int = None

    def __lt__(self, other):
        return self.get_F_val() < other.get_F_val()

    def get_F_val(self):
        if self.F_val == float("inf"):
            return self.f_val
        return self.F_val


@dataclass
class Edge:
    start_node: Node
    end_node: Node
    weight: float

    def get_other_node(self, node_id: int) -> Node:
        if self.start_node.id == node_id:
            return self.end_node
        else:
            return self.start_node


@dataclass
class Graph:
    nodes: dict
    adjacent_edges: dict

    def get_node(self, node_id: int) -> Node:
        return self.nodes[node_id]

    def get_connected_nodes(self, node_id: int) -> List[Node]:
        adjacent_edges = self.adjacent_edges.get(node_id, [])
        return [edge.get_other_node(node_id) for edge in adjacent_edges]

    def get_adjacent_edges(self, node_id: int) -> List[Edge]:
        return self.adjacent_edges.get(node_id, [])

    def set_edge_weight(self, edge_id: int, weight: float):
        self.edges[edge_id].set_weight(weight)

    def trace(self, node_id):
        parent_id = self.nodes[node_id].parent_id
        if parent_id != None:
            return self.trace(parent_id) + [node_id]
        return [node_id]


def pea(graph, start_node_id, end_node_id, C=0):
    open_heap = []
    open_dict = {}
    closed_dict = {}

    global num_of_expanded_open_list, num_of_maximum_expanded_node
    num_of_expanded_open_list = 0
    num_of_maximum_expanded_node = 0

    def add_open(node_id):
        heapq.heappush(open_heap, graph.get_node(node_id))
        open_dict[node_id] = True

        global num_of_expanded_open_list, num_of_maximum_expanded_node
        num_of_expanded_open_list += 1
        num_of_maximum_expanded_node = max(len(open_dict), num_of_maximum_expanded_node)

    def pop_open():
        node = heapq.heappop(open_heap)
        del open_dict[node.id]
        return node.id

    def add_close(node_id):
        closed_dict[node_id] = True

    def remove_close(node_id):
        del closed_dict[node_id]

    start_node = graph.get_node(start_node_id)
    start_node.f_val = 0
    start_node.F_val = 0
    add_open(start_node_id)

    while len(open_heap) > 0:
        node = graph.get_node(pop_open())
        if node.id == end_node_id:
            break

        s_le, s_gt = [], []
        edges = graph.get_adjacent_edges(node.id)
        for edge in edges:
            other_node = edge.get_other_node(node.id)
            if other_node.f_val == float("inf"):
                other_node.f_val = node.f_val + edge.weight

            if other_node.f_val <= node.get_F_val() + C:
                heapq.heappush(s_le, [other_node.f_val, other_node, edge.weight])
            else:
                heapq.heappush(s_gt, [other_node.f_val, other_node, edge.weight])

        for [_, node_l, c_n1_n2] in s_le:
            if node_l.id not in open_dict and node_l.id not in closed_dict:
                node_l.f_val = node.f_val + c_n1_n2
                node_l.F_val = node_l.f_val
                node_l.parent_id = node.id
                add_open(node_l.id)
            elif node_l.id in open_dict and node.f_val + c_n1_n2 < node_l.f_val:
                node_l.f_val = node.f_val + c_n1_n2
                node_l.F_val = node_l.f_val
                node_l.parent_id = node.id
            elif node_l.id in closed_dict and node.f_val + c_n1_n2 < node_l.f_val:
                node_l.f_val = node.f_val + c_n1_n2
                node_l.F_val = node_l.f_val
                node_l.parent_id = node.id
                add_open(node_l.id)
                remove_close(node_l.id)

        if len(s_gt) > 0:
            [f_val, _, _] = heapq.heappop(s_gt)
            node.F_val = f_val
            add_open(node.id)
        else:
            add_close(node.id)

    print("Cut Off(C) 값:", C)
    print("최단 경로:", graph.trace(end_node_id))
    print("OPEN 리스트에 노드가 추가된 횟수:", num_of_expanded_open_list)
    print("OPEN 리스트에 저장된 노드 개수의 최대값:", num_of_maximum_expanded_node)
    print("====================")


if __name__ == "__main__":
    nodes = {}
    for i in range(16):
        nodes[i + 1] = Node(id=i + 1)

    adjacent_edges = {}
    adjacent_edges[1] = [Edge(start_node=nodes[1], end_node=nodes[2], weight=1)]
    adjacent_edges[1] += [Edge(start_node=nodes[1], end_node=nodes[3], weight=2)]
    adjacent_edges[1] += [Edge(start_node=nodes[1], end_node=nodes[4], weight=3)]
    adjacent_edges[1] += [Edge(start_node=nodes[1], end_node=nodes[5], weight=4)]
    adjacent_edges[1] += [Edge(start_node=nodes[1], end_node=nodes[6], weight=5)]
    adjacent_edges[2] = [Edge(start_node=nodes[2], end_node=nodes[7], weight=1)]
    adjacent_edges[2] += [Edge(start_node=nodes[2], end_node=nodes[8], weight=2)]
    adjacent_edges[2] += [Edge(start_node=nodes[2], end_node=nodes[9], weight=3)]
    adjacent_edges[2] += [Edge(start_node=nodes[2], end_node=nodes[10], weight=4)]
    adjacent_edges[2] += [Edge(start_node=nodes[2], end_node=nodes[11], weight=5)]
    adjacent_edges[7] = [Edge(start_node=nodes[7], end_node=nodes[12], weight=1)]
    adjacent_edges[7] += [Edge(start_node=nodes[7], end_node=nodes[13], weight=2)]
    adjacent_edges[7] += [Edge(start_node=nodes[7], end_node=nodes[14], weight=3)]
    adjacent_edges[7] += [Edge(start_node=nodes[7], end_node=nodes[15], weight=4)]
    adjacent_edges[7] += [Edge(start_node=nodes[7], end_node=nodes[16], weight=5)]

    graph = Graph(nodes=nodes, adjacent_edges=adjacent_edges)

    for val in range(5):
        pea(graph, 1, 16, C=val)

 

3. 예시

Cut Off 값에 따라서 OPEN 리스트에 저장되는 최대 노드 개수가 어떻게 변하는지 살펴보기 위해서 자식 노드가 많은 그래프를 준비했습니다!

자식 노드가 많은 그래프

 

실행 결과

실제로 Cut Off 값이 커질수록 OPEN 리스트에 저장되는 노드 개수의 최대 값이 증가하고,

최단 경로를 찾기까지 탐색한 노드 수가 줄어드는 것을 확인할 수 있었습니다 👍👍👍

실행 결과!

---

PEA*알고리즘은 어떠셨나요?

생각보다 너무 간단하여 "진짜 이게 끝이야?" 하면서 논문을 여러 번 읽었던 기억이 나네요... ㅋㅋㅋ

그리고 예시로 훨씬 큰 그래프를 준비했으면 좋았을텐데 아쉬움이 남네요... 흑흑

 

슬슬 작업 배정 관련 알고리즘들도 공부해보려고 하는데 A* 변형 알고리즘이 너무 많아서 언제쯤 시작할 수 있을지 모르겠습니다 ㅜ