AbductiveLearning
/
ABLkit
Not watched
1
0
Fork 0
Code
Releases 0 Wiki evaluate Activity
Issues 0 Pull Requests 0 Datasets Model Cloudbrain HPC
Browse Source

Update abducer_base.py

pull/3/head
troyyyyy GitHub 3 years ago
parent
7e4102c709
commit
09083f9c2e
No known key found for this signature in database GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 90 additions and 102 deletions
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. +90
    -102
      abl/abducer/abducer_base.py

+ 90
- 102
abl/abducer/abducer_base.py View File

@@ -10,33 +10,16 @@
# #
# ================================================================# # ================================================================#


# import sys

# sys.path.append(".")
# sys.path.append("..")

import abc import abc
# TODO 尽量别用import *
from .kb import * from .kb import *
import numpy as np import numpy as np
from zoopt import Dimension, Objective, Parameter, Opt from zoopt import Dimension, Objective, Parameter, Opt
from ..utils.utils import confidence_dist, flatten, hamming_dist from ..utils.utils import confidence_dist, flatten, hamming_dist


import math
import time


class AbducerBase(abc.ABC): class AbducerBase(abc.ABC):
def __init__(
self,
kb,
dist_func="confidence",
zoopt=False,
multiple_predictions=False,
cache=True,
):
def __init__(self, kb, dist_func='confidence', zoopt=False, multiple_predictions=False, cache=True):
self.kb = kb self.kb = kb
assert dist_func == "hamming" or dist_func == "confidence"
assert dist_func == 'hamming' or dist_func == 'confidence'
self.dist_func = dist_func self.dist_func = dist_func
self.zoopt = zoopt self.zoopt = zoopt
self.multiple_predictions = multiple_predictions self.multiple_predictions = multiple_predictions
@@ -47,41 +30,42 @@ class AbducerBase(abc.ABC):
self.cache_candidates = {} self.cache_candidates = {}


def _get_cost_list(self, pred_res, pred_res_prob, candidates): def _get_cost_list(self, pred_res, pred_res_prob, candidates):
if self.dist_func == "hamming":
if self.dist_func == 'hamming':
if self.multiple_predictions:
pred_res = flatten(pred_res)
candidates = [flatten(c) for c in candidates]
return hamming_dist(pred_res, candidates) return hamming_dist(pred_res, candidates)
elif self.dist_func == "confidence":
mapping = dict(
zip(
self.kb.pseudo_label_list,
list(range(len(self.kb.pseudo_label_list))),
)
)
return confidence_dist(
pred_res_prob, [list(map(lambda x: mapping[x], c)) for c in candidates]
)
elif self.dist_func == 'confidence':
if self.multiple_predictions:
pred_res_prob = flatten(pred_res_prob)
candidates = [flatten(c) for c in candidates]
mapping = dict(zip(self.kb.pseudo_label_list, list(range(len(self.kb.pseudo_label_list)))))
candidates = [list(map(lambda x: mapping[x], c)) for c in candidates]
return confidence_dist(pred_res_prob, candidates)


def _get_one_candidate(self, pred_res, pred_res_prob, candidates): def _get_one_candidate(self, pred_res, pred_res_prob, candidates):
if len(candidates) == 0: if len(candidates) == 0:
return [] return []
elif len(candidates) == 1 or self.zoopt: elif len(candidates) == 1 or self.zoopt:
return candidates[0] return candidates[0]
else: else:
cost_list = self._get_cost_list(pred_res, pred_res_prob, candidates) cost_list = self._get_cost_list(pred_res, pred_res_prob, candidates)
min_address_num = np.min(cost_list) min_address_num = np.min(cost_list)
idxs = np.where(cost_list == min_address_num)[0] idxs = np.where(cost_list == min_address_num)[0]
return [candidates[idx] for idx in idxs][0]
candidate = [candidates[idx] for idx in idxs][0]
return candidate


# for zoopt # for zoopt
def _zoopt_score_multiple(self, pred_res, key, solution): def _zoopt_score_multiple(self, pred_res, key, solution):
all_address_flag = reform_idx(solution, pred_res) all_address_flag = reform_idx(solution, pred_res)
score = 0 score = 0
for idx in range(len(pred_res)): for idx in range(len(pred_res)):
address_idx = [
i for i, flag in enumerate(all_address_flag[idx]) if flag != 0
]
candidate = self.kb.address_by_idx(
[pred_res[idx]], key[idx], address_idx, True
)
address_idx = [i for i, flag in enumerate(all_address_flag[idx]) if flag != 0]
candidate = self.kb.address_by_idx([pred_res[idx]], key[idx], address_idx, True)
if len(candidate) > 0: if len(candidate) > 0:
score += 1 score += 1
return score return score
@@ -89,9 +73,7 @@ class AbducerBase(abc.ABC):
def _zoopt_address_score(self, pred_res, key, sol): def _zoopt_address_score(self, pred_res, key, sol):
if not self.multiple_predictions: if not self.multiple_predictions:
address_idx = [idx for idx, i in enumerate(sol.get_x()) if i != 0] address_idx = [idx for idx, i in enumerate(sol.get_x()) if i != 0]
candidates = self.kb.address_by_idx(
pred_res, key, address_idx, self.multiple_predictions
)
candidates = self.kb.address_by_idx(pred_res, key, address_idx, self.multiple_predictions)
return 1 if len(candidates) > 0 else 0 return 1 if len(candidates) > 0 else 0
else: else:
return self._zoopt_score_multiple(pred_res, key, sol.get_x()) return self._zoopt_score_multiple(pred_res, key, sol.get_x())
@@ -108,7 +90,7 @@ class AbducerBase(abc.ABC):
dim=dimension, dim=dimension,
constraint=lambda sol: self._constrain_address_num(sol, max_address_num), constraint=lambda sol: self._constrain_address_num(sol, max_address_num),
) )
parameter = Parameter(budget=100, autoset=True)
parameter = Parameter(budget=100, intermediate_result=False, autoset=True)
solution = Opt.min(objective, parameter).get_x() solution = Opt.min(objective, parameter).get_x()


return solution return solution
@@ -119,11 +101,7 @@ class AbducerBase(abc.ABC):
pred_res = flatten(pred_res) pred_res = flatten(pred_res)
key = tuple(key) key = tuple(key)
if (tuple(pred_res), key) in self.cache_min_address_num: if (tuple(pred_res), key) in self.cache_min_address_num:
address_num = min(
max_address_num,
self.cache_min_address_num[(tuple(pred_res), key)]
+ require_more_address,
)
address_num = min(max_address_num, self.cache_min_address_num[(tuple(pred_res), key)] + require_more_address)
if (tuple(pred_res), key, address_num) in self.cache_candidates: if (tuple(pred_res), key, address_num) in self.cache_candidates:
candidates = self.cache_candidates[(tuple(pred_res), key, address_num)] candidates = self.cache_candidates[(tuple(pred_res), key, address_num)]
if self.zoopt: if self.zoopt:
@@ -152,18 +130,12 @@ class AbducerBase(abc.ABC):
if self.zoopt: if self.zoopt:
solution = self.zoopt_get_solution(pred_res, key, max_address_num) solution = self.zoopt_get_solution(pred_res, key, max_address_num)
address_idx = [idx for idx, i in enumerate(solution) if i != 0] address_idx = [idx for idx, i in enumerate(solution) if i != 0]
candidates = self.kb.address_by_idx(
pred_res, key, address_idx, self.multiple_predictions
)
candidates = self.kb.address_by_idx(pred_res, key, address_idx, self.multiple_predictions)
address_num = int(solution.sum()) address_num = int(solution.sum())
min_address_num = address_num min_address_num = address_num
else: else:
candidates, min_address_num, address_num = self.kb.abduce_candidates( candidates, min_address_num, address_num = self.kb.abduce_candidates(
pred_res,
key,
max_address_num,
require_more_address,
self.multiple_predictions,
pred_res, key, max_address_num, require_more_address, self.multiple_predictions
) )


candidate = self._get_one_candidate(pred_res, pred_res_prob, candidates) candidate = self._get_one_candidate(pred_res, pred_res_prob, candidates)
@@ -177,32 +149,21 @@ class AbducerBase(abc.ABC):
return self.kb.abduce_rules(pred_res) return self.kb.abduce_rules(pred_res)


def batch_abduce(self, Z, Y, max_address_num=-1, require_more_address=0): def batch_abduce(self, Z, Y, max_address_num=-1, require_more_address=0):
if self.multiple_predictions:
return self.abduce(
(Z["cls"], Z["prob"], Y), max_address_num, require_more_address
)
else:
return [
self.abduce((z, prob, y), max_address_num, require_more_address)
for z, prob, y in zip(Z["cls"], Z["prob"], Y)
]
# if self.multiple_predictions:
return self.abduce((Z['cls'], Z['prob'], Y), max_address_num, require_more_address)
# else:
# return [self.abduce((z, prob, y), max_address_num, require_more_address) for z, prob, y in zip(Z['cls'], Z['prob'], Y)]


def __call__(self, Z, Y, max_address_num=-1, require_more_address=0): def __call__(self, Z, Y, max_address_num=-1, require_more_address=0):
return self.batch_abduce(Z, Y, max_address_num, require_more_address) return self.batch_abduce(Z, Y, max_address_num, require_more_address)




if __name__ == "__main__":
prob1 = [
[0, 0.99, 0.01, 0, 0, 0, 0, 0, 0, 0],
[0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1],
]
prob2 = [
[0, 0, 0.01, 0, 0, 0, 0, 0.99, 0, 0],
[0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1],
]
if __name__ == '__main__':
prob1 = [[0, 0.99, 0.01, 0, 0, 0, 0, 0, 0, 0], [0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1]]
prob2 = [[0, 0, 0.01, 0, 0, 0, 0, 0.99, 0, 0], [0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1]]


kb = add_KB()
abd = AbducerBase(kb, "confidence")
kb = add_KB(True)
abd = AbducerBase(kb, 'confidence')
res = abd.abduce(([1, 1], prob1, 8), max_address_num=2, require_more_address=0) res = abd.abduce(([1, 1], prob1, 8), max_address_num=2, require_more_address=0)
print(res) print(res)
res = abd.abduce(([1, 1], prob2, 8), max_address_num=2, require_more_address=0) res = abd.abduce(([1, 1], prob2, 8), max_address_num=2, require_more_address=0)
@@ -214,9 +175,23 @@ if __name__ == "__main__":
res = abd.abduce(([1, 1], prob1, 20), max_address_num=2, require_more_address=0) res = abd.abduce(([1, 1], prob1, 20), max_address_num=2, require_more_address=0)
print(res) print(res)
print() print()
multiple_prob = [[[0, 0.99, 0.01, 0, 0, 0, 0, 0, 0, 0], [0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1]],
[[0, 0, 0.01, 0, 0, 0, 0, 0.99, 0, 0], [0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1]]]
kb = add_KB()
abd = AbducerBase(kb, 'confidence', multiple_predictions=True)
res = abd.abduce(([[1, 1], [1, 2]], multiple_prob, [4, 8]), max_address_num=4, require_more_address=0)
print(res)
res = abd.abduce(([[1, 1], [1, 2]], multiple_prob, [4, 8]), max_address_num=4, require_more_address=1)
print(res)
print()


kb = add_prolog_KB() kb = add_prolog_KB()
abd = AbducerBase(kb, "confidence")
abd = AbducerBase(kb, 'confidence')
res = abd.abduce(([1, 1], prob1, 8), max_address_num=2, require_more_address=0) res = abd.abduce(([1, 1], prob1, 8), max_address_num=2, require_more_address=0)
print(res) print(res)
res = abd.abduce(([1, 1], prob2, 8), max_address_num=2, require_more_address=0) res = abd.abduce(([1, 1], prob2, 8), max_address_num=2, require_more_address=0)
@@ -230,7 +205,7 @@ if __name__ == "__main__":
print() print()


kb = add_prolog_KB() kb = add_prolog_KB()
abd = AbducerBase(kb, "confidence", zoopt=True)
abd = AbducerBase(kb, 'confidence', zoopt=True)
res = abd.abduce(([1, 1], prob1, 8), max_address_num=2, require_more_address=0) res = abd.abduce(([1, 1], prob1, 8), max_address_num=2, require_more_address=0)
print(res) print(res)
res = abd.abduce(([1, 1], prob2, 8), max_address_num=2, require_more_address=0) res = abd.abduce(([1, 1], prob2, 8), max_address_num=2, require_more_address=0)
@@ -243,42 +218,55 @@ if __name__ == "__main__":
print(res) print(res)
print() print()


kb = HWF_KB(len_list=[1, 3, 5])
abd = AbducerBase(kb, "hamming")
res = abd.abduce(
(["5", "+", "2"], None, 3), max_address_num=2, require_more_address=0
)
kb = HWF_KB(True, len_list=[1, 3, 5], max_err = 0.1)
abd = AbducerBase(kb, 'hamming')
res = abd.abduce((['5', '+', '2'], None, 3), max_address_num=2, require_more_address=0)
print(res)
res = abd.abduce((['5', '+', '9'], None, 64), max_address_num=3, require_more_address=0)
print(res)
kb = HWF_KB(True, len_list=[1, 3, 5], max_err = 1)
abd = AbducerBase(kb, 'hamming')
res = abd.abduce((['5', '+', '9'], None, 64), max_address_num=3, require_more_address=0)
print(res)
res = abd.abduce((['5', '+', '2'], None, 1.67), max_address_num=3, require_more_address=0)
print(res)
res = abd.abduce((['5', '8', '8', '8', '8'], None, 3.17), max_address_num=5, require_more_address=3)
print(res)
print()
kb = HWF_KB(len_list=[1, 3, 5], max_err = 0.1)
abd = AbducerBase(kb, 'hamming', multiple_predictions=True)
res = abd.abduce(([['5', '+', '2'], ['5', '+', '9']], None, [3, 64]), max_address_num=6, require_more_address=0)
print(res)
print()
kb = HWF_KB(len_list=[1, 3, 5], max_err = 0.1)
abd = AbducerBase(kb, 'hamming')
res = abd.abduce((['5', '+', '2'], None, 3), max_address_num=2, require_more_address=0)
print(res)
res = abd.abduce((['5', '+', '9'], None, 64), max_address_num=3, require_more_address=0)
print(res) print(res)
res = abd.abduce(
(["5", "+", "2"], None, 64), max_address_num=3, require_more_address=0
)
kb = HWF_KB(len_list=[1, 3, 5], max_err = 1)
abd = AbducerBase(kb, 'hamming')
res = abd.abduce((['5', '+', '9'], None, 64), max_address_num=3, require_more_address=0)
print(res) print(res)
res = abd.abduce(
(["5", "+", "2"], None, 1.67), max_address_num=3, require_more_address=0
)
res = abd.abduce((['5', '+', '2'], None, 1.67), max_address_num=3, require_more_address=0)
print(res) print(res)
res = abd.abduce(
(["5", "8", "8", "8", "8"], None, 3.17),
max_address_num=5,
require_more_address=3,
)
res = abd.abduce((['5', '8', '8', '8', '8'], None, 3.17), max_address_num=5, require_more_address=3)
print(res) print(res)
print() print()


kb = HED_prolog_KB() kb = HED_prolog_KB()
abd = AbducerBase(kb, zoopt=True, multiple_predictions=True) abd = AbducerBase(kb, zoopt=True, multiple_predictions=True)
consist_exs = [[1, "+", 0, "=", 0], [1, "+", 1, "=", 0], [0, "+", 0, "=", 1, 1]]
consist_exs2 = [
[1, "+", 0, "=", 0],
[1, "+", 1, "=", 0],
[0, "+", 1, "=", 1, 1],
] # not consistent with rules
inconsist_exs = [[1, "+", 0, "=", 0], [1, "=", 1, "=", 0], [0, "=", 0, "=", 1, 1]]
consist_exs = [[1, 1, '+', 0, '=', 1, 1], [1, '+', 1, '=', 1, 0], [0, '+', 0, '=', 0]]
inconsist_exs = [[1, '+', 0, '=', 0], [1, '=', 1, '=', 0], [0, '=', 0, '=', 1, 1]]
# inconsist_exs = [[1, '+', 0, '=', 0], ['=', '=', '=', '=', 0], ['=', '=', 0, '=', '=', '=']] # inconsist_exs = [[1, '+', 0, '=', 0], ['=', '=', '=', '=', 0], ['=', '=', 0, '=', '=', '=']]
rules = ["my_op([0], [0], [1, 1])", "my_op([1], [1], [0])", "my_op([1], [0], [0])"]
rules = ['my_op([0], [0], [0])', 'my_op([1], [1], [1, 0])']


print(kb.logic_forward(consist_exs), kb.logic_forward(inconsist_exs))
print(kb.consist_rule(consist_exs, rules), kb.consist_rule(consist_exs2, rules))
print(kb._logic_forward(consist_exs, True), kb._logic_forward(inconsist_exs, True))
print(kb.consist_rule([1, '+', 1, '=', 1, 0], rules), kb.consist_rule([1, '+', 1, '=', 1, 1], rules))
print() print()


res = abd.abduce((consist_exs, None, [1] * len(consist_exs))) res = abd.abduce((consist_exs, None, [1] * len(consist_exs)))


Write Preview
Loading…
Cancel
Save