動(dòng)態(tài)規(guī)劃是一種用來(lái)解決定義了一個(gè)狀態(tài)空間的問(wèn)題的算法策略。這些問(wèn)題可分解為新的子問(wèn)題，子問(wèn)題有自己的參數(shù)。為了解決它們，我們必須搜索這個(gè)狀態(tài)空間并且在每一步作決策時(shí)進(jìn)行求值。得益于這類(lèi)問(wèn)題會(huì)有大量相同的狀態(tài)的這個(gè)事實(shí)，這種技術(shù)不會(huì)在解決重疊的子問(wèn)題上浪費(fèi)時(shí)間。

正如我們看到的，它也會(huì)導(dǎo)致大量地使用遞歸，這通常會(huì)很有趣。

為了說(shuō)明這種算法策略，我會(huì)用一個(gè)很好玩的問(wèn)題來(lái)作為例子，這個(gè)問(wèn)題是我最近參加的 一個(gè)編程競(jìng)賽中的 Tuenti Challenge #4 中的第 14 個(gè)挑戰(zhàn)問(wèn)題。
Train Empire

我們面對(duì)的是一個(gè)叫 Train Empire 的棋盤(pán)游戲（Board Game）。在這個(gè)問(wèn)題中，你必須為火車(chē)規(guī)劃出一條最高效的路線(xiàn)來(lái)運(yùn)輸在每個(gè)火車(chē)站的貨車(chē)。規(guī)則很簡(jiǎn)單：

•     每個(gè)車(chē)站都有一個(gè)在等待著的將要運(yùn)送到其他的車(chē)站的貨車(chē)。
•     每個(gè)貨車(chē)被送到了目的地會(huì)獎(jiǎng)勵(lì)玩家一些分?jǐn)?shù)。貨車(chē)可以放在任意車(chē)站。
•     火車(chē)只在一條單一的路線(xiàn)上運(yùn)行，每次能裝一個(gè)貨車(chē)，因?yàn)槿剂嫌邢拗荒芤苿?dòng)一定的距離。

我們可以把我們的問(wèn)題原先的圖美化一下。為了在燃料限制下贏得最大的分?jǐn)?shù)，我們需要知道貨車(chē)在哪里裝載，以及在哪里卸載。

我們?cè)趫D片中可以看到，我們有兩條火車(chē)路線(xiàn)：紅色和藍(lán)色。車(chē)站位于某些坐標(biāo)點(diǎn)上，所以我們很容易就能算出它們之間的距離。每一個(gè)車(chē)站有一個(gè)以它的終點(diǎn)命名的貨車(chē)，以及當(dāng)我們成功送達(dá)它可以得到的分?jǐn)?shù)獎(jiǎng)勵(lì)。

現(xiàn)在，假定我們的貨車(chē)能跑3千米遠(yuǎn)。紅色路線(xiàn)上的火車(chē)可以把 A 車(chē)站的火車(chē)送到它的 終點(diǎn) E (5點(diǎn)分?jǐn)?shù))，藍(lán)色路線(xiàn)上的火車(chē)可以運(yùn)送貨車(chē) C(10點(diǎn)分?jǐn)?shù))，然后運(yùn)送貨車(chē) B(5點(diǎn)分?jǐn)?shù))。 可以取得最高分20分。
狀態(tài)表示

我們把火車(chē)的位置，以及火車(chē)所走的距離和每個(gè)車(chē)站的貨車(chē)表格叫做一個(gè)問(wèn)題狀態(tài)。 改變這些值我們得到的仍是相同的問(wèn)題，但是參數(shù)變了。我們可以看到每次我們移動(dòng) 一列火車(chē)，我們的問(wèn)題就演變到一個(gè)不同的子問(wèn)題。為了算出最佳的移動(dòng)方案，我們 必須遍歷這些狀態(tài)然后基于這些狀態(tài)作出決策。讓我們開(kāi)始把。

我們將從定義火車(chē)路線(xiàn)開(kāi)始。因?yàn)檫@些路線(xiàn)不是直線(xiàn)，所以圖是最好的表示方法。

import math
from decimal import Decimal
from collections import namedtuple, defaultdict
 
class TrainRoute:
 
  def __init__(self, start, connections):
    self.start = start
 
    self.E = defaultdict(set)
    self.stations = set()
    for u, v in connections:
      self.E[u].add(v)
      self.E[v].add(u)
      self.stations.add(u)
      self.stations.add(v)
 
  def next_stations(self, u):
    if u not in self.E:
      return
    yield from self.E[u]
 
  def fuel(self, u, v):
    x = abs(u.pos[0] - v.pos[0])
    y = abs(u.pos[1] - v.pos[1])
    return Decimal(math.sqrt(x * x + y * y))

TrainRoute 類(lèi)實(shí)現(xiàn)了一個(gè)非?；镜挠邢驁D，它把頂點(diǎn)作為車(chē)站存在一個(gè)集合中，把車(chē)站間 的連接存在一個(gè)字典中。請(qǐng)注意我們把 (u, v) 和 (v, u) 兩條邊都加上了，因?yàn)榛疖?chē)可以 向前向后移動(dòng)。

在 next_stations 方法中有一個(gè)有趣東西，在這里我使用了一個(gè)很酷的 Python 3 的特性yield from。這允許一個(gè)生成器 可以委派到另外一個(gè)生成器或者迭代器中。因?yàn)槊恳粋€(gè)車(chē)站都映射到一個(gè)車(chē)站的集合，我們只 需要迭代它就可以了。

讓我們來(lái)看一下 main class:

TrainWagon = namedtuple('TrainWagon', ('dest', 'value'))
TrainStation = namedtuple('TrainStation', ('name', 'pos', 'wagons'))
 
class TrainEmpire:
 
  def __init__(self, fuel, stations, routes):
    self.fuel = fuel
    self.stations = self._build_stations(stations)
    self.routes = self._build_routes(routes)
 
  def _build_stations(self, station_lines):
    # ...
 
  def _build_routes(self, route_lines):
    # ...
 
  def maximum_route_score(self, route):
 
    def score(state):
      return sum(w.value for (w, s) in state.wgs if w.dest == s.name)
 
    def wagon_choices(state, t):
      # ...
 
    def delivered(state):
      # ...
 
    def next_states(state):
      # ...
 
    def backtrack(state):
      # ...
 
    # ...
 
  def maximum_score(self):
    return sum(self.maximum_route_score(r) for r in self.routes)

我省略了一些代碼，但是我們可以看到一些有趣的東西。兩個(gè) 命名元組 將會(huì)幫助保持我們的數(shù)據(jù)整齊而簡(jiǎn)單。main class 有我們的火車(chē)能夠運(yùn)行的最長(zhǎng)的距離，燃料， 和路線(xiàn)以及車(chē)站這些參數(shù)。maximum_score 方法計(jì)算每條路線(xiàn)的分?jǐn)?shù)的總和，將成為解決問(wèn)題的 接口，所以我們有：

•     一個(gè) main class 持有路線(xiàn)和車(chē)站之間的連接
•     一個(gè)車(chē)站元組，存有名字，位置和當(dāng)前存在的貨車(chē)列表
•     一個(gè)帶有一個(gè)值和目的車(chē)站的貨車(chē)

動(dòng)態(tài)規(guī)劃

我已經(jīng)嘗試解釋了動(dòng)態(tài)規(guī)劃如何高效地搜索狀態(tài)空間的關(guān)鍵，以及基于已有的狀態(tài)進(jìn)行最優(yōu)的決策。 我們有一個(gè)定義了火車(chē)的位置，火車(chē)剩余的燃料，以及每個(gè)貨車(chē)的位置的狀態(tài)空間——所以我們已經(jīng)可以表示初始狀態(tài)。

我們現(xiàn)在必須考慮在每個(gè)車(chē)站的每一種決策。我們應(yīng)該裝載一個(gè)貨車(chē)然后把它送到目的地嗎？ 如果我們?cè)谙乱粋€(gè)車(chē)站發(fā)現(xiàn)了一個(gè)更有價(jià)值的貨車(chē)怎么辦？我們應(yīng)該把它送回去或者還是往前 移動(dòng)？或者還是不帶著貨車(chē)移動(dòng)？

很顯然，這些問(wèn)題的答案是那個(gè)可以使我們獲得更多的分?jǐn)?shù)的那個(gè)。為了得到答案，我們必須求出 所有可能的情形下的前一個(gè)狀態(tài)和后一個(gè)狀態(tài)的值。當(dāng)然我們用求分函數(shù) score 來(lái)求每個(gè)狀態(tài)的值。
 

def maximum_score(self):
  return sum(self.maximum_route_score(r) for r in self.routes)
 
State = namedtuple('State', ('s', 'f', 'wgs'))
 
wgs = set()
for s in route.stations:
  for w in s.wagons:
    wgs.add((w, s))
initial = State(route.start, self.fuel, tuple(wgs))

從每個(gè)狀態(tài)出發(fā)都有幾個(gè)選擇：要么帶著貨車(chē)移動(dòng)到下一個(gè)車(chē)站，要么不帶貨車(chē)移動(dòng)。停留不動(dòng)不會(huì)進(jìn)入一個(gè)新的 狀態(tài)，因?yàn)槭裁礀|西都沒(méi)改變。如果當(dāng)前的車(chē)站有多個(gè)貨車(chē)，移動(dòng)它們中的一個(gè)都將會(huì)進(jìn)入一個(gè)不同的狀態(tài)。

def wagon_choices(state, t):
  yield state.wgs # not moving wagons is an option too
 
  wgs = set(state.wgs)
  other_wagons = {(w, s) for (w, s) in wgs if s != state.s}
  state_wagons = wgs - other_wagons
  for (w, s) in state_wagons:
    parked = state_wagons - {(w, s)}
    twgs = other_wagons | parked | {(w, t)}
    yield tuple(twgs)
 
def delivered(state):
  return all(w.dest == s.name for (w, s) in state.wgs)
 
def next_states(state):
  if delivered(state):
    return
  for s in route.next_stations(state.s):
    f = state.f - route.fuel(state.s, s)
    if f < 0:
      continue
    for wgs in wagon_choices(state, s):
      yield State(s, f, wgs)

next_states 是一個(gè)以一個(gè)狀態(tài)為參數(shù)然后返回所有這個(gè)狀態(tài)能到達(dá)的狀態(tài)的生成器。 注意它是如何在所有的貨車(chē)都移動(dòng)到了目的地后停止的，或者它只進(jìn)入到那些燃料仍然足夠的狀態(tài)。wagon_choices 函數(shù)可能看起來(lái)有點(diǎn)復(fù)雜，其實(shí)它僅僅返回那些可以從當(dāng)前車(chē)站到下一個(gè)車(chē)站的貨車(chē)集合。

這樣我們就有了實(shí)現(xiàn)動(dòng)態(tài)規(guī)劃算法需要的所有東西。我們從初始狀態(tài)開(kāi)始搜索我們的決策，然后選擇 一個(gè)最有策略。看！初始狀態(tài)將會(huì)演變到一個(gè)不同的狀態(tài)，這個(gè)狀態(tài)也會(huì)演變到一個(gè)不同的狀態(tài)！ 我們正在設(shè)計(jì)的是一個(gè)遞歸算法：

•     獲取狀態(tài)
•     計(jì)算我們的決策
•     做出最優(yōu)決策

顯然每個(gè)下一個(gè)狀態(tài)都將做這一系列的同樣的事情。我們的遞歸函數(shù)將會(huì)在燃料用盡或者所有的貨車(chē)都被運(yùn)送都目的地了時(shí)停止。
 

max_score = {}
 
def backtrack(state):
  if state.f <= 0:
    return state
  choices = []
  for s in next_states(state):
    if s not in max_score:
      max_score[s] = backtrack(s)
    choices.append(max_score[s])
  if not choices:
    return state
  return max(choices, key=lambda s: score(s))
 
max_score[initial] = backtrack(initial)
return score(max_score[initial])

完成動(dòng)態(tài)規(guī)劃策略的最后一個(gè)陷阱：在代碼中，你可以看到我使用了一個(gè) max_score 字典， 它實(shí)際上緩存著算法經(jīng)歷的每一個(gè)狀態(tài)。這樣我們就不會(huì)重復(fù)一遍又一遍地遍歷我們的我們?cè)缇鸵呀?jīng) 經(jīng)歷過(guò)的狀態(tài)的決策。

當(dāng)我們搜索狀態(tài)空間的時(shí)候，一個(gè)車(chē)站可能會(huì)到達(dá)多次，這其中的一些可能會(huì)導(dǎo)致相同的燃料，相同的貨車(chē)。 火車(chē)怎么到達(dá)這里的沒(méi)關(guān)系，只有在那個(gè)時(shí)候做的決策有影響。如果我們我們計(jì)算過(guò)那個(gè)狀態(tài)一次并且保存了 結(jié)果，我們就不在需要再搜索一遍這個(gè)子空間了。

如果我們沒(méi)有用這種記憶化技術(shù)，我們會(huì)做大量完全相同的搜索。 這通常會(huì)導(dǎo)致我們的算法很難高效地解決我們的問(wèn)題。
總結(jié)

Train Empire 提供了一個(gè)絕佳的的例子，以展示動(dòng)態(tài)規(guī)劃是如何在有重疊子問(wèn)題的問(wèn)題做出最優(yōu)決策。 Python 強(qiáng)大的表達(dá)能力再一次讓我們很簡(jiǎn)單地就能把想法實(shí)現(xiàn)，并且寫(xiě)出清晰且高效的算法。

完整的代碼在 contest repository。

最新評(píng)論