-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathvalidation_information_credibility.py
102 lines (83 loc) · 3.65 KB
/
validation_information_credibility.py
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
from functions import *
from add_synonyms import *
import ast
# Ensure an event loop is created before importing llamaapi
loop = get_or_create_event_loop()
# Now you can safely import llamaapi
from llamaapi import LlamaAPI
chatbot_on = True
answer = ''
if chatbot_on:
user_input = input("Enter some text: ")
llama = LlamaAPI("LL-oZZ3DbV8EBPzVAdh7ylv5pQP3y0ryH77l7x50ELwEzDymcHuVOA3BnhH66HdFIZh")
api_request_json = {
"model": "llama3-70b",
"max_tokens": 10000,
"temperature": 0.1,
"messages": [
{"role": "system", "content": f"For the given text provide all concepts and relations between them. Ignore the correctnes of the text. Present the result in a python list of tuples - subject, relation, object. Please use Wikidata labels for the results. Don't use Wikidata IRIs for the result. Don't add comments."},
{"role": "user", "content": f"Text: {user_input}"},
]
}
while True:
try:
response = llama.run(api_request_json)
response.raise_for_status()
answer_content = response.json()["choices"][0]["message"]["content"]
break
except requests.exceptions.JSONDecodeError as e:
print(f"Error: {e}")
save_answer_to_file(answer_content, 'response.txt')
answer = answer_content
else:
answer = get_answer_from_file("response.txt")
triples_list = eval(answer)
wikidata_mappings = []
for triple in triples_list:
wikidata_mappings.append(map_triple_to_wikidata(triple))
merged_triples = merge_triples(triples_list, wikidata_mappings)
true_triples = []
explanations_list = []
for triple in merged_triples:
is_true_triple, explanation = check_wikidata_relationship(triple)
explanations_list.append(explanation)
if is_true_triple:
new_triple = (triple[0][1], triple[1][1], triple[2][1])
true_triples.append(new_triple)
else:
synonyms = get_synonyms(synonyms_file_path, triple[1][0])
for synonym in synonyms:
new_triple = (triple[0][0], synonym, triple[2][0])
new_triple_map = map_triple_to_wikidata(new_triple)
mapped_triple = tuple(zip(new_triple, new_triple_map))
is_true_triple, explanation = check_wikidata_relationship(mapped_triple)
if is_true_triple:
true_triples.append(mapped_triple)
explanations_list[-1] = "A synonym relationship is correct"
break
else:
explanations_list[-1] = explanations_list[-1] + "\nAND\n" + explanation
with open("triples.txt", "w") as file:
file.write(str(merged_triples))
#print(true_triples)
forward_chaining_triples = forward_chaining(true_triples, rules)
with open("result.txt", "w") as file:
pass
#print(forward_chaining_triples)
for index, triple in enumerate(merged_triples):
subject = triple[0][1]
predicate = triple[1][1]
obj = triple[2][1]
fact = (subject, predicate, obj)
output_str = f"{triple[0][0]} {triple[1][0]} {triple[2][0]}"
if fact in forward_chaining_triples or explanations_list[index] == "The relationship is correct." or explanations_list[index] == "A synonym relationship is correct":
if not explanations_list[index] == "The relationship is correct.":
with open("result.txt", "a") as file:
file.write(output_str + " --> True, because: " + explanations_list[index] + "\n\n")
else:
with open("result.txt", "a") as file:
file.write(output_str + " --> True\n\n")
else:
with open("result.txt", "a") as file:
file.write(output_str + " --> False, because:\n" + explanations_list[index] + "\n\n")
print("Check result.txt")