Repository navigation
Expand file tree
/
Copy pathrelevant_search.py
More file actions
179 lines (145 loc) · 6.29 KB
/
Copy pathrelevant_search.py
File metadata and controls
179 lines (145 loc) · 6.29 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
import timeit
from collections import defaultdict
class Rule:
def __init__(self, premises, conclusion):
self.premises = premises
self.conclusion = conclusion
def __repr__(self):
return f"Rule({self.premises} -> {self.conclusion})"
class OptimizedExpertSystem:
def __init__(self):
self.knowledge_base = []
self.facts = set()
# Index rules by conclusion for faster lookup
self.conclusion_to_rules = defaultdict(list)
# Cache for memoization
self.inference_cache = {}
def add_rule(self, premises, conclusion):
rule = Rule(premises, conclusion)
self.knowledge_base.append(rule)
self.conclusion_to_rules[conclusion].append(rule)
def add_fact(self, fact):
self.facts.add(fact)
# Clear cache when facts change
self.inference_cache = {}
def backward_inference(self, goal, path=None, depth=0, max_depth=100):
"""
Optimized backward chaining algorithm with:
- Memoization to avoid redundant computations
- Path tracking to avoid cycles
- Depth limiting to prevent infinite recursion
- Rule indexing for faster lookups
- Minimal rule path selection
"""
# Initialize path tracking if not provided
if path is None:
path = set()
self.used_rules = []
self.inference_cache = {} # Reset cache for new inference
self.best_rule_paths = {} # Track the best path to each goal
# Check depth limit to prevent infinite recursion
if depth > max_depth:
return False
# Check if goal is already a known fact
if goal in self.facts:
self.best_rule_paths[goal] = []
return True
# Check if we've already computed this goal
if goal in self.inference_cache:
return self.inference_cache[goal]
# Check for cycles in the inference path
if goal in path:
return False
# Add current goal to path
path.add(goal)
# Get rules that can derive this goal
relevant_rules = self.conclusion_to_rules.get(goal, [])
# Sort rules by number of premises (prefer simpler rules first)
relevant_rules.sort(key=lambda r: len(r.premises))
best_rule = None
min_rule_count = float('inf')
for rule in relevant_rules:
all_premises_true = True
current_path_rules = []
# Check each premise
for premise in rule.premises:
if not self.backward_inference(premise, path, depth + 1, max_depth):
all_premises_true = False
break
# Add rules used for this premise
if premise in self.best_rule_paths:
current_path_rules.extend(self.best_rule_paths[premise])
if all_premises_true:
# Add this rule to the current path
current_path_rules.append(rule)
# Check if this path uses fewer rules than our current best
if len(current_path_rules) < min_rule_count:
min_rule_count = len(current_path_rules)
best_rule = rule
self.best_rule_paths[goal] = current_path_rules
# If we found a valid path
if best_rule is not None:
# Update the used_rules list with unique rules from the best path
for rule in self.best_rule_paths[goal]:
if rule not in self.used_rules:
self.used_rules.append(rule)
# Remove goal from path before returning
path.remove(goal)
# Cache the result
self.inference_cache[goal] = True
return True
# Remove goal from path before returning
path.remove(goal)
# Cache the negative result
self.inference_cache[goal] = False
return False
def measure_optimized_inference_time(expert_system, goal):
result = expert_system.backward_inference(goal)
used_rules = expert_system.used_rules if hasattr(expert_system, 'used_rules') else []
return result, used_rules
# Example usage
if __name__ == "__main__":
# Original system for comparison
original_system = OptimizedExpertSystem()
original_system.add_fact("A")
original_system.add_fact("D")
original_system.add_rule(["A"], "B")
original_system.add_rule(["B"], "C")
original_system.add_rule(["D"], "E")
original_system.add_rule(["E", "B"], "F")
# Create a more complex knowledge base
complex_system = OptimizedExpertSystem()
# Add base facts
complex_system.add_fact("A")
complex_system.add_fact("D")
complex_system.add_fact("X")
# Add rules with various complexity
complex_system.add_rule(["A"], "B")
complex_system.add_rule(["B"], "C")
complex_system.add_rule(["D"], "E")
complex_system.add_rule(["E", "B"], "F")
complex_system.add_rule(["X"], "Y")
complex_system.add_rule(["Y"], "Z")
complex_system.add_rule(["Z", "C"], "G")
complex_system.add_rule(["F", "G"], "H")
complex_system.add_rule(["A", "B", "C", "D", "E"], "F") # Redundant complex rule
print("=== Original System ===")
goal = "F"
time_taken = timeit.timeit(lambda: measure_optimized_inference_time(original_system, goal), number=1000)
result, used_rules = measure_optimized_inference_time(original_system, goal)
print(f"Goal '{goal}' is {'proven' if result else 'not proven'}")
print(f"Time taken for 1000 runs: {time_taken:.6f} seconds")
print(f"Rules used: {len(used_rules)}")
print("Rules used to achieve the goal:")
for rule in used_rules:
print(f"Premises: {rule.premises} -> Conclusion: {rule.conclusion}")
print("\n=== Complex System ===")
goal = "H"
time_taken = timeit.timeit(lambda: measure_optimized_inference_time(complex_system, goal), number=1000)
result, used_rules = measure_optimized_inference_time(complex_system, goal)
print(f"Goal '{goal}' is {'proven' if result else 'not proven'}")
print(f"Time taken for 1000 runs: {time_taken:.6f} seconds")
print(f"Rules used: {len(used_rules)}")
print("Rules used to achieve the goal:")
for rule in used_rules:
print(f"Premises: {rule.premises} -> Conclusion: {rule.conclusion}")