Fixedpoint ― 再帰的なルールで到達可能性を解く

Fixedpoint ― 再帰的なルールで到達可能性を解く#

この章で学ぶこと#

Z3 の Fixedpoint は、再帰的なルールを扱うエンジンです。「A から B に行けて、B から C に行けるなら、A から C に行ける」といった推論を、そのままルールとして書けます。

  • ルール（rule）と事実（fact）で関係を書く

  • query() で「成り立つか」を聞く

  • 到達できるとき、実際の経路をどう取り出すか

デフォルトのエンジンはボトムアップ型の Datalog です。Datalog は関係データベースの問い合わせ（SQL の再帰クエリ）に似ていて、与えた事実からルールを繰り返し適用し、新しい事実を増やしていきます。

from z3 import *

例1：グラフの到達可能性#

次の無向グラフで、頂点1から頂点6へ行けるかどうかを調べます。

        flowchart LR
1 --- 2
1 --- 5
5 --- 2
4 --- 5
2 --- 3
4 --- 3
4 --- 6

使う述語は3つです。

  • edge(a, b) … a と b の間に辺がある（事実として与える）

  • connect(a, b) … a から b へ直接移動できる（辺があれば双方向に成り立つ）

  • path(a, b) … a から b へ到達できる（再帰的に決まる）

fp = Fixedpoint()
fp.set(engine='datalog')          # Datalog エンジンを使う

s = BitVecSort(8)                  # 頂点の番号は8ビットで表す

connect = Function('connect', s, s, BoolSort())
edge = Function('edge', s, s, BoolSort())
path = Function('path', s, s, BoolSort())

a, b, c = Consts('a b c', s)

fp.register_relation(path, edge, connect)
fp.declare_var(a, b, c)

# ルール：辺があれば直接移動できる（無向なので両方向）
fp.rule(connect(a, b), edge(a, b))
fp.rule(connect(a, b), edge(b, a))

# ルール：直接移動できるなら到達できる
fp.rule(path(a, b), connect(a, b))
# ルール：a→b 移動できて、b→c 到達できるなら、a→c 到達できる
fp.rule(path(a, c), [connect(a, b), path(b, c)])

# 事実：グラフの辺
edges = [(1, 2), (1, 5), (5, 2), (4, 5), (2, 3), (4, 3), (4, 6)]
for p, q in edges:
    fp.fact(edge(p, q))

コードの説明#

1. エンジン

fp = Fixedpoint()
fp.set(engine='datalog')

engine='datalog' で Datalog エンジンを選びます。

2. 頂点の型

s = BitVecSort(8)

頂点の番号は8ビットのビットベクトルで表します。Datalog は有限の型を必要とするので、BitVecSort や IntSort ではなく有限の型を使います。

3. 述語（関係）の定義

connect = Function('connect', s, s, BoolSort())

Function(名前, 引数の型..., 戻り値の型) で述語を作ります。「2つの頂点を受け取って真偽を返す関係」なので、戻り値は BoolSort() です。

4. 変数の宣言と述語の登録

fp.register_relation(path, edge, connect)
fp.declare_var(a, b, c)

register_relation() で述語をエンジンに登録し、declare_var() でルール内で使う変数を宣言します。

5. ルールと事実

fp.rule(path(a, c), [connect(a, b), path(b, c)])

rule(頭, 本体) は「本体がすべて成り立つなら、頭も成り立つ」というルールです。このルールは「a から b に移動でき、b から c に到達できるなら、a から c に到達できる」と読みます。自分自身（path）を本体に入れているのが再帰です。

fp.fact(edge(1, 2)) は、辺があることを事実として登録します。

問い合わせる#

print(fp.query(path(1, 6)))     # 到達できるか
print(fp.query(path(1, 7)))     # 存在しない頂点

sat
unsat

頂点1から6へは到達でき（sat）、頂点7へは到達できません（unsat）。この「成り立つかどうか」を答えるのが Fixedpoint の仕事です。

経路を再構成する#

「到達できる」は分かりましたが、どう行けばよいかはまだ分かりません。

Z3 の Datalog には、なぜその事実が成り立つかの**説明（explanations）**を出す機能があります（datalog.generate_explanations）。ところが、この機能は最近の Z3（この環境は 5.1.0）では不安定で、explanations are not supported with undefined predicates というエラーになったり、答えが unsat に化けたり、Z3 ごと落ちたりします。

そこで、この章では経路は自分で作ります。到達できるかどうかは Fixedpoint に聞き、具体的な経路は辺のリストを BFS（幅優先探索）すれば求まります。

from collections import deque


def bfs_route(edges, start, goal, undirected=True):
    """(from, to) のリストから、最短の経路を BFS で作る"""
    neighbors = {}
    for p, q in edges:
        neighbors.setdefault(p, []).append(q)
        if undirected:
            neighbors.setdefault(q, []).append(p)

    prev = {start: None}
    queue = deque([start])
    while queue:
        cur = queue.popleft()
        if cur == goal:
            break
        for nxt in neighbors.get(cur, []):
            if nxt not in prev:
                prev[nxt] = cur
                queue.append(nxt)

    if goal not in prev:
        return None

    route = [goal]
    while prev[route[-1]] is not None:
        route.append(prev[route[-1]])
    return route[::-1]


print(bfs_route(edges, 1, 6))

[1, 5, 4, 6]

経路は 1 → 5 → 4 → 6 です。edge(4, 5) は無向なので 5→4 に使えます。

例2：3つのバケツ問題#

8リットル、5リットル、3リットルのバケツがあります。最初は8リットルのバケツだけが一杯で、あとは空です。2つのバケツに4リットルずつ入れるには、どう注げばよいでしょうか。

これは「状態を頂点、注ぐ操作を辺とするグラフの到達可能性」と見なせます。まず、ありうる状態と、状態間の移行を Python で全部作ります。

  • 水の量の合計はいつも8リットル

  • 移行は「あるバケツから別のバケツへ注ぐ」だけ

  • 注ぐと、注ぎ先がいっぱいになるか、注ぎ元が空になるかのどちらか

from itertools import product

sizes = (8, 5, 3)

# ありうる状態（合計が8リットル）
states = [(a, b, c)
          for a, b, c in product(*[range(size + 1) for size in sizes])
          if a + b + c == 8]

# 1回の注ぎで移れる状態の組
transitions = []
for s1, s2 in product(states, states):
    changed = [(size, old, new) for size, old, new in zip(sizes, s1, s2) if old != new]
    if len(changed) != 2:
        continue        # 変わるバケツはちょうど2つ（注ぎ元と注ぎ先）
    if any(new == size or new == 0 for size, old, new in changed):
        transitions.append((s1, s2))    # 注ぎ先が満杯 or 注ぎ元が空

print(len(states), "通りの状態、", len(transitions), "通りの注ぎ方")

24 通りの状態、 106 通りの注ぎ方

次に、この遷移を pour 述語の事実として登録し、plan 述語で「初期状態から目標状態への手順があるか」を推論します。

  • pour(a, b, c, a2, b2, c2) … 状態 (a, b, c) から (a2, b2, c2) へ1回で移れる

  • plan(a, b, c, a2, b2, c2) … 状態 (a, b, c) から (a2, b2, c2) へ何回かで移れる

fp = Fixedpoint()
fp.set(engine='datalog')

s = BitVecSort(5)
pour = Function('pour', s, s, s, s, s, s, BoolSort())
plan = Function('plan', s, s, s, s, s, s, BoolSort())
a, b, c, a1, b1, c1, a2, b2, c2 = Consts(['a', 'b', 'c', 'a1', 'b1', 'c1', 'a2', 'b2', 'c2'], s)

fp.register_relation(pour, plan)
fp.declare_var(a, b, c, a1, b1, c1, a2, b2, c2)

# 1回で移れるなら、plan も成り立つ
fp.rule(plan(a, b, c, a2, b2, c2), pour(a, b, c, a2, b2, c2))
# 1回移って、そこから plan があるなら、plan が成り立つ（再帰）
fp.rule(plan(a, b, c, a2, b2, c2), [pour(a, b, c, a1, b1, c1), plan(a1, b1, c1, a2, b2, c2)])

for s1, s2 in transitions:
    fp.fact(pour(*s1, *s2))

start, goal = (8, 0, 0), (4, 4, 0)

%time print(fp.query(plan(*start, *goal)))

sat
CPU times: total: 0 ns
Wall time: 1.49 ms

sat なので、8,0,0 から 4,4,0 へ移る手順があります。

手順は、transitions（辺のリスト）を BFS すれば求まります。例1 の bfs_route() がそのまま使えます（こちらは有向なので undirected=False）。

route = bfs_route(transitions, start, goal, undirected=False)

for s1, s2 in zip(route, route[1:]):
    print(s1, "->", s2)

(8, 0, 0) -> (3, 5, 0)
(3, 5, 0) -> (3, 2, 3)
(3, 2, 3) -> (6, 2, 0)
(6, 2, 0) -> (6, 0, 2)
(6, 0, 2) -> (1, 5, 2)
(1, 5, 2) -> (1, 4, 3)
(1, 4, 3) -> (4, 4, 0)

7回の注ぎで 4,4,0 に到達しました。人間が解くと試行錯誤が必要な問題も、状態と遷移さえ書き出せば、あとは機械的に解けます。

登録したルールと事実も確認できます。

for rule in fp.get_rules()[:5]:
    print(rule)
print("...")

ForAll([A, B, C, D, E, F],
       Implies(pour(A, B, C, D, E, F),
               plan(A, B, C, D, E, F)))
ForAll([A, B, C, D, E, F, G, H, I],
       Implies(And(plan(D, E, F, G, H, I),
                   pour(A, B, C, D, E, F)),
               plan(A, B, C, G, H, I)))
pour(0, 5, 3, 3, 5, 0)
pour(0, 5, 3, 5, 0, 3)
pour(1, 4, 3, 0, 5, 3)
...

例3：pour をルールにする#

例2 では、Python ですべての遷移を計算して pour を事実として登録しました。こんどは、pour をルールで書きます。ありうる状態 state(a, b, c) を事実として与え、注ぎ方の条件をルールにします。

条件は「変わったバケツが2つ」で、どちらかが満杯か空になることです。

fp = Fixedpoint()
fp.set(engine='datalog')

s = BitVecSort(5)
state = Function('state', s, s, s, BoolSort())
pour = Function('pour', s, s, s, s, s, s, BoolSort())
plan = Function('plan', s, s, s, s, s, s, BoolSort())
a, b, c, a1, b1, c1, a2, b2, c2 = Consts(['a', 'b', 'c', 'a1', 'b1', 'c1', 'a2', 'b2', 'c2'], s)

fp.register_relation(state, pour, plan)
fp.declare_var(a, b, c, a1, b1, c1, a2, b2, c2)

fp.rule(plan(a, b, c, a2, b2, c2), pour(a, b, c, a2, b2, c2))
fp.rule(plan(a, b, c, a2, b2, c2), [pour(a, b, c, a1, b1, c1), plan(a1, b1, c1, a2, b2, c2)])

# 注ぎ方のルール：変わるバケツは2つ、どちらかが満杯か空になる
fp.rule(
    pour(a, b, c, a1, b1, c1),
    [
        state(a, b, c),
        state(a1, b1, c1),
        Or(
            And(a == a1, b != b1, Or(b1 == 0, b1 == 5, c1 == 0, c1 == 3)),
            And(b == b1, a != a1, Or(a1 == 0, a1 == 8, c1 == 0, c1 == 3)),
            And(c == c1, a != a1, Or(a1 == 0, a1 == 8, b1 == 0, b1 == 5)),
        ),
    ],
)

# ありうる状態（合計8リットル）を事実として登録
for p in states:
    fp.fact(state(*p))

%time print(fp.query(plan(*start, *goal)))

sat
CPU times: total: 46.9 ms
Wall time: 38.3 ms

例2 と同じく sat です。pour をルールで書いても、状態を事実として与えれば同じ結論になります。この環境では 0.04 秒ほどでした（例2 は 1 ミリ秒ほど）。

pour のルールを読むときのポイントは、state(a, b, c) と state(a1, b1, c1) の両方を本体に入れていることです。注いだ後の状態も、ありうる状態でなければいけません。

例4：state もルールにする#

最後に、state もルールにします。バケツごとの取りうる量を bucket_a, bucket_b, bucket_c の事実として与え、state はその組み合わせ（合計8リットル）として導きます。

fp = Fixedpoint()
fp.set(engine='datalog')

s = BitVecSort(5)
bucket_a = Function('bucket_a', s, BoolSort())
bucket_b = Function('bucket_b', s, BoolSort())
bucket_c = Function('bucket_c', s, BoolSort())
state = Function('state', s, s, s, BoolSort())
pour = Function('pour', s, s, s, s, s, s, BoolSort())
plan = Function('plan', s, s, s, s, s, s, BoolSort())
a, b, c, a1, b1, c1, a2, b2, c2 = Consts(['a', 'b', 'c', 'a1', 'b1', 'c1', 'a2', 'b2', 'c2'], s)

fp.register_relation(state, pour, plan, bucket_a, bucket_b, bucket_c)
fp.declare_var(a, b, c, a1, b1, c1, a2, b2, c2)

fp.rule(plan(a, b, c, a2, b2, c2), pour(a, b, c, a2, b2, c2))
fp.rule(plan(a, b, c, a2, b2, c2), [pour(a, b, c, a1, b1, c1), plan(a1, b1, c1, a2, b2, c2)])

fp.rule(
    pour(a, b, c, a1, b1, c1),
    [
        state(a, b, c),
        state(a1, b1, c1),
        Or(
            And(a == a1, b != b1, Or(b1 == 0, b1 == 5, c1 == 0, c1 == 3)),
            And(b == b1, a != a1, Or(a1 == 0, a1 == 8, c1 == 0, c1 == 3)),
            And(c == c1, a != a1, Or(a1 == 0, a1 == 8, b1 == 0, b1 == 5)),
        ),
    ],
)

# state は「各バケツの量の組み合わせ」で、合計が8リットルになるもの
fp.rule(state(a, b, c), [bucket_a(a), bucket_b(b), bucket_c(c), a + b + c == 8])

for i in range(sizes[0] + 1):
    fp.fact(bucket_a(i), str(i))
for i in range(sizes[1] + 1):
    fp.fact(bucket_b(i), str(i))
for i in range(sizes[2] + 1):
    fp.fact(bucket_c(i), str(i))

%time print(fp.query(plan(*start, *goal)))

sat
CPU times: total: 625 ms
Wall time: 614 ms

こちらも sat ですが、ずっと時間がかかりました。この環境では 0.6 秒で、例3 の 0.04 秒と比べて15倍ほどです。

state をルールにすると、pour のルールを適用するたびに state の組み合わせを考え直すことになります。その分だけ探索が増えるわけです。「ルールで書ける」ことと「速い」ことは別で、分かっている関係は事実として与えたほうが速いという例です。

まとめ#

使うもの

役割

Fixedpoint()

エンジンを作る

set(engine='datalog')

Datalog エンジンを選ぶ

Function(名前, 型..., BoolSort())

述語を作る

register_relation(...)

述語をエンジンに登録する

declare_var(...)

ルールで使う変数を宣言する

rule(頭, 本体)

「本体なら頭」というルールを追加する

fact(...)

事実を追加する

query(...)

成り立つか問い合わせる（sat / unsat）

get_rules()

登録したルールと事実を確認する

  • 到達可能性は「1回で移れる」＋「再帰」の2つのルールで書ける

  • Fixedpoint が答えるのは「到達できるか」。具体的な経路は、辺のリストを BFS すれば作れる

  • 説明（generate_explanations）で経路を取り出す機能は、最近の Z3 では不安定なので、あてにしないほうがよい

  • 状態と遷移が分かっているなら、事実として与えるほうが速い。すべてルールで書くと柔軟だが遅くなる