【图网络】如何用Python实现算法：规划图技术(GraphPlanner)

作者 | Debby Nirwan

编译 | VK
来源 | Towards Data Science

这篇文章涉及规划图技术分步算法实现：从伪代码和方程到Python代码。在本文中，我们将用Python实现规划图（Planning Graph）及其规划器GraphPlanner，以及AI规划的数据结构和搜索算法。

介绍

规划图是为了解决经典人工智能规划方法(又称STRIPS规划器)的复杂性问题而发展起来的。我们需要实施两个主要部分：

• 规划图：数据结构

• 图规划：搜索算法，为我们找到解决方案

如果你不熟悉规划图，想了解更多，请查看我的以下帖子：

https://towardsdatascience.com/improving-classical-ai-planning-complexity-with-planning-graph-c63d47f87018

领域和问题表示

在我们开始实现之前，我们需要知道我们将如何表示规划域和规划问题。

规划图及其规划器与许多STRIPS规划器有相同的表示，因此我们将使用PDDL(规划域定义语言)来表示它们。下面是PDDL域文件的一个示例。

;; Specification in PDDL1 of the DWR domain(define (domain dock-worker-robot-simple)(:requirements :strips :typing )(:typeslocation      ; there are several connected locations in the harborrobot         ; holds at most 1 container, only 1 robot per locationcontainer)(:predicates(adjacent ?l1  ?l2 - location)       ; location ?l1 is adjacent ot ?l2(atl ?r - robot ?l - location)       ; robot ?r is at location ?l(loaded ?r - robot ?c - container )  ; robot ?r is loaded with container ?c(unloaded ?r - robot)                ; robot ?r is empty(in ?c - container ?l - location)    ; container ?c is within location ?l);; there are 3 operators in this domain:;; moves a robot between two adjacent locations(:action move:parameters (?r - robot ?from ?to - location):precondition (and (adjacent ?from ?to) (atl ?r ?from) ):effect (and (atl ?r ?to)(not (atl ?r ?from)) ));; loads an empty robot with a container held by a nearby crane(:action load:parameters (?l - location ?c - container ?r - robot):precondition (and (atl ?r ?l) (in ?c ?l) (unloaded ?r)):effect (and (loaded ?r ?c)(not (in ?c ?l)) (not (unloaded ?r)) ));; unloads a robot holding a container with a nearby crane(:action unload:parameters (?l - location ?c - container ?r - robot):precondition (and (atl ?r ?l) (loaded ?r ?c) ):effect (and (unloaded ?r) (in ?c ?l)(not (loaded ?r ?c)) )) )

我们可以将PDDL看作JSON或XML之类的东西，这意味着我们需要一个解析器来反序列化用它编写的表示。当我在Github上搜索时，其中有几个出现了，但是有一个似乎很适合我们的项目pddlpy。

https://github.com/hfoffani/pddl-lib

但是，它在开发中不再活跃，我发现了一个bug和一些问题。因此，我决定使用它并编写一个适配器/包装器，这是一个薄层，我们添加它来修复错误并解决其他问题。

PDDL适配器

我们需要的代表如下：

• 世界的初始状态：数据类型为set

• 目标状态：数据类型为set

• 运算符(也称为操作)的列表，这些运算符已用实变量实例化：数据类型为List[Operator]

我们将只使用pddlpy库中的一个接口，DomainProblem()类构造函数。

我们需要提供上面列出的三个接口：初始状态、目标状态和运算符列表。

我们创建了一个名为PlanningProblem的类：

class PlanningProblem(object):def __init__(self, dom_file: str, problem_file: str):self._domain_file = dom_fileself._problem_file = problem_fileself._domain_problem = DomainProblem(self._domain_file,self._problem_file)self._initial_state = self._to_set_of_tuples(self._domain_problem.initialstate())self._goal_state = self._to_set_of_tuples(self._domain_problem.goals())self._actions = self._get_ground_operators()

库提供的状态不是我们想要的正确数据类型，因此需要将它们转换为一组元组。

我们使用set()数据类型，以便以后实现数据结构和算法。因为在经典的人工智能规划中，我们大量使用集合论，我们应该使用set()数据类型来利用内置函数来加速我们的实现。我们将在下一节看到更多内容。

我们还必须创建一个运算符列表，我们称之为操作。下面是适配器的最终代码。

class PlanningProblem(object):def __init__(self, dom_file: str, problem_file: str):self._domain_file = dom_fileself._problem_file = problem_fileself._domain_problem = DomainProblem(self._domain_file,self._problem_file)self._initial_state = self._to_set_of_tuples(self._domain_problem.initialstate())self._goal_state = self._to_set_of_tuples(self._domain_problem.goals())self._actions = self._get_ground_operators()@staticmethoddef _type_symbols(variable_type, world_objects: dict):# 如果变量类型是在world对象中找到的，# 返回对象名称列表，如robr, robqreturn (k for k, v in world_objects.items() if v == variable_type)def _instantiate(self, variables, world_objects: dict):variable_ground_space = []for variable_name, variable_type in variables:c = []for symbol in self._type_symbols(variable_type, world_objects):c.append((variable_name, symbol))variable_ground_space.append(c)return itertools.product(*variable_ground_space)def _get_ground_operators(self) -> List[Operator]:ground_operators = []for operator in self._domain_problem.operators():op = self._domain_problem.domain.operators[operator]for ground in self._instantiate(op.variable_list.items(),self._domain_problem.worldobjects()):st = dict(ground)gop = Operator(operator)gop.variable_list = stgop.precondition_pos = set([a.ground(st) for a in op.precondition_pos])gop.precondition_neg = set([a.ground(st) for a in op.precondition_neg])gop.effect_pos = set([a.ground(st) for a in op.effect_pos])gop.effect_neg = set([a.ground(st) for a in op.effect_neg])ground_operators.append(gop)return ground_operators@staticmethoddef _to_set_of_tuples(state: Set[Atom]) -> Set[Tuple]:set_of_tuples = set()for atom in state:tup = tuple(atom.predicate)set_of_tuples.add(tup)return set_of_tuples@propertydef initial_state(self):return self._initial_state@propertydef goal_state(self):return self._goal_state@propertydef actions(self):return self._actions

我们现在可以使用这个类来解决规划领域和规划问题，并将重点放在数据结构和算法实现上。

规划图：数据结构

在这里我们将只看伪代码和方程，然后接下来的重点是将它们翻译成代码

以下是我们如何构建规划图的伪代码：

有四个步骤，我们一个接一个地讲。

计算动作

这是指以下步骤：

其中包括两部分：

• 对于PDDL适配器提供的所有操作，我们在当前状态下搜索可用的操作

• 我们确保那些可应用的操作的前提条件不在前提条件的互斥体中

class PlanningGraph(object):def __init__(self, dom_file: str, problem_file: str):self._planning_problem = PlanningProblem(dom_file, problem_file)self._graph: Graph = Graph(visualize)def compute_actions(self, gr: Graph):graph_result = grlevel = gr.num_of_levels# 计算 Aiaction_list = []for action in self._planning_problem.actions:if self._applicable(action, graph_result.prop[level - 1],graph_result.prop_mutexes[level - 1]):action_list.append(action)graph_result.act[level] = action_list@staticmethoddef _applicable(action, state, preconditions_mutex) -> bool:if action.precondition_pos.issubset(state) and \action.precondition_neg.isdisjoint(state):applicable = Trueif preconditions_mutex is not None:for precondition in list(permutations(action.precondition_pos, 2)):if precondition in preconditions_mutex:applicable = Falsebreakelse:applicable = Falsereturn applicable)

计算前提条件

下一步是计算前提条件，也就是这一步：

这一步非常简单：

class PlanningGraph(object):def __init__(self, dom_file: str, problem_file: str):self._planning_problem = PlanningProblem(dom_file, problem_file)self._graph: Graph = Graph(visualize)def compute_preconditions(self, gr: Graph):graph_result = grlevel = gr.num_of_levelsproposition_list = set()for action in action_list:for effect in action.effect_pos:proposition_list.add(effect)graph_result.prop[level] = proposition_list

我们只存储计算出的动作效果。

计算操作互斥

该算法的下一步是计算Actions Mutex（操作互斥），这是一组相互抵消效果的操作。

在这个等式中有三个部分，对于动作中所有可能的排列，我们希望在我们的列表中包括以下内容：

• 行动的消极影响会干扰另一方的积极影响或先决条件

• 第二部分是相同的，只是在另一个方向(b到a)

• 第三部分是它们的前提条件是互斥

class PlanningGraph(object):def __init__(self, dom_file: str, problem_file: str):self._planning_problem = PlanningProblem(dom_file, problem_file)self._graph: Graph = Graph(visualize)def compute_actions_mutex(self, gr: Graph):graph_result = grlevel = gr.num_of_levelsaction_mutex_list = []for action_pair in list(permutations(action_list, 2)):if self.compute_action_mutex(action_pair,graph_result.prop_mutexes[level - 1]):action_mutex_list.append(action_pair)graph_result.act_mutexes[level] = action_mutex_list@staticmethoddef compute_action_mutex(pair, preconds_mutex) -> bool:a = pair[0]b = pair[1]# 两个动作是相互依存的if a.effect_neg.interp(b.precondition_pos.union(b.effect_pos)) \!= set():return Trueif b.effect_neg.interp(a.precondition_pos.union(a.effect_pos)) \!= set():return True# 它们的先决条件互斥if preconds_mutex is not None:for mutex in preconds_mutex:# (p, q)p = mutex[0]q = mutex[1]if p in a.precondition_pos and q in b.precondition_pos:return Truereturn False

计算互斥的前提条件

建立规划图的算法的最后一步是计算预条件互斥

这意味着我们要寻找一对互斥的前提条件。它们是互斥的当且仅当：

• 对于同时产生p和q的所有操作对，它们都在actions Mutex列表中

• 没有同时产生p和q的单一操作

class PlanningGraph(object):def __init__(self, dom_file: str, problem_file: str):self._planning_problem = PlanningProblem(dom_file, problem_file)self._graph: Graph = Graph()def compute_preconditions_mutex(self, gr: Graph):proposition_mutex_list = []for proposition_pair in list(permutations(proposition_list, 2)):if self.compute_precondition_mutex(proposition_pair, action_list, action_mutex_list):if proposition_pair not in proposition_mutex_list:swapped = (proposition_pair[1], proposition_pair[0])if swapped not in proposition_mutex_list:proposition_mutex_list.append(proposition_pair)graph_result.prop_mutexes[level] = proposition_mutex_list@staticmethoddef compute_precondition_mutex(proposition_pair, action_list, action_mutex):p = proposition_pair[0]q = proposition_pair[1]for action in action_list:if p in action.effect_pos and q in action.effect_pos:# (p, q)不是互斥对象，如果它们都是由同一个动作产生的return False# 每一个产生p的动作actions_with_p = set()for action in action_list:if p in action.effect_pos:actions_with_p.add(action)# 每一个产生q的动作actions_with_q = set()for action in action_list:if q in action.effect_pos:actions_with_q.add(action)all_mutex = Truefor p_action in actions_with_p:for q_action in actions_with_q:if p_action == q_action:return Falseif (p_action, q_action) not in action_mutex:all_mutex = Falsebreakif not all_mutex:breakreturn all_mutex

我们现在已经完成了构建数据结构的代码，规划图。为了帮助调试，可以使用pydot扩充代码以生成图形可视化。下面是一个例子。

搜索算法：GraphPlanner

我们现在已经准备好了数据结构，我们可以开始实现搜索算法，为我们的计划问题找到解决方案。

该算法是递归的，分为三个部分：

• 计划

• 提取

• 搜索

提取和搜索

这两个步骤是递归的，算法如下。第一部分是Extract：

下一部分是Search：

这是这两个函数如何递归工作的示例：

它递归地调用Search()，直到所有的命题都被解析，并调用Extract()进入规划图的下一级。

Python编写如下：

class GraphPlanner(object):def __init__(self):self._layered_plan: LayeredPlan = LayeredPlan()self._mutex = {}def _extract(self, gr: Graph, g: set, index: int):if index == 0:return Plan()return self._search(gr, g, Plan(), index)def _search(self, gr: Graph, g: set, plan: Plan, index: int):if g == set():new_goals = set()for action in plan.plan:for proposition in action.precondition_pos:if 'adjacent' not in proposition:new_goals.add(proposition)extracted_plan = self._extract(gr, new_goals, index-1)if extracted_plan is None:return Noneelse:self._layered_plan[index-1] = extracted_planself._layered_plan[index] = planreturn planelse:# 选择g中的任意pproposition = g.pop()# 计算解析器resolvers = []for action in gr.act[index]:if proposition in action.effect_pos:if plan.plan:mutex = Falsefor action2 in plan.plan:if (action, action2) in \gr.act_mutexes[index]:mutex = Truebreakif not mutex:resolvers.append(action)else:resolvers.append(action)# 没有解析器if not resolvers:return None# 选择非确定性，如果失败则回溯while resolvers:resolver = resolvers.pop()plan.append(resolver)plan_result = self._search(gr, g - resolver.effect_pos,plan, index)if plan_result is not None:return plan_resultelse:plan.remove(resolver)g.add(proposition)return None

主程序

最后，我们到达了算法的最后一步、以下是主要步骤和入口点：

在某些情况下，我们需要计划更多的步骤来创建解决方案计划，我们需要展开规划图并重试搜索。

我们还需要添加一个额外的步骤，以确保算法在没有可能的解决方案时终止。下面是我们的最后一段代码：

class GraphPlanner(object):def __init__(self):self._layered_plan: LayeredPlan = LayeredPlan()self._mutex = {}def plan(self, gr: Graph, g: set):index = gr.num_of_levels - 1if not g.issubset(gr.prop[index]):return Noneplan = self._extract(gr, g, index)if plan:return self._layered_planif gr.fixed_point:n = 0try:props_mutex = self._mutex[gr.num_of_levels - 1]except KeyError:props_mutex = Noneif props_mutex:n = len(props_mutex)else:n = 0while True:index += 1gr = PlanningGraph.expand(gr)plan = self._extract(gr, g, index)if plan:return self._layered_planelif gr.fixed_point:try:props_mutex = self._mutex[gr.num_of_levels-1]except KeyError:props_mutex = Noneif props_mutex:if n == len(props_mutex):# 这意味着它已经稳定下来return Noneelse:n = len(props_mutex)

结尾

我意识到要描述实现这个算法的思想过程并不容易。但我希望至少你能对如何实现从等式、伪代码到Python代码的算法有一些基本的理解。

完整代码可在我的Github上获得，如下所示：

https://github.com/debbynirwan/planning_graph

往期精彩回顾适合初学者入门人工智能的路线及资料下载机器学习及深度学习笔记等资料打印机器学习在线手册深度学习笔记专辑《统计学习方法》的代码复现专辑
AI基础下载机器学习的数学基础专辑温州大学《机器学习课程》视频
本站qq群851320808，加入微信群请扫码：

本文来自互联网用户投稿，文章观点仅代表作者本人，不代表本站立场，不承担相关法律责任。如若转载，请注明出处。 如若内容造成侵权/违法违规/事实不符，请点击【内容举报】进行投诉反馈！