template_generate_ucdp.py (13682B)
1 import numpy 2 import re 3 import sys 4 5 6 INPUT_STYLE_SET = ['event_type', 'event_type_sent', 'keywords', 'triggers', 'template'] 7 OUTPUT_STYLE_SET = ['argument:sentence'] 8 9 BREAK_PENALTY = 10 # Penalty if variable is not preceded by space or followed by space/period. 10 TEMPLATE_MARKER = '\1' # Template marker for variables 11 POST_TEMPLATE_VALID = " .," # Unpenalized characters after template variable 12 PRE_TEMPLATE_VALID = " " # Unpenalized characters before template variable 13 14 UCDP_TEMPLATE_FORMAT = """The conflict between {side_a_name} (A) and {side_b_name} (B) resulted in causalities between {start_date} and {end_date} at {location_where_name} in {location_adm2_name} in the region of {location_adm1_name} in {location_root_name}. {deaths_side_a} people died on side A, {deaths_side_b} people died on side B, {deaths_civilian} civilians died and {deaths_unknown} unidentified people died. The low estimate for the total number of people that died is {deaths_low}. The high estimate for the total number of people that died is {deaths_high}.""" 15 16 UCDP_FIELDS = re.findall(r"{([^}]+)}", UCDP_TEMPLATE_FORMAT) 17 UCDP_TEMPLATE = re.sub(" {[^}]*}", TEMPLATE_MARKER, UCDP_TEMPLATE_FORMAT) 18 19 ROLE_TO_TYPE = { 20 "side_a_name": "Actor", 21 "side_b_name": "Actor", 22 "start_date": "Date", 23 "end_date": "Date", 24 "location_root_name": "Place", 25 "location_adm1_name": "Place", 26 "location_adm2_name": "Place", 27 "location_where_name": "Place", 28 "deaths_side_a": "Deaths", 29 "deaths_side_b": "Deaths", 30 "deaths_civilian": "Deaths", 31 "deaths_unknown": "Deaths", 32 "deaths_low": "Deaths", 33 "deaths_high": "Deaths", 34 } 35 36 TYPE_PH_MAP = { 37 "Actor": "somebody", 38 "Date": "sometime", 39 "Place": "somewhere", 40 "Deaths": "some number", 41 } 42 43 44 def fuzzy_align(prediction): 45 """ 46 Use a Levenshtein algorithm to fuzzily align the output of a model with its template. 47 48 As currently written, the variables must always be preceded by a space in the template. 49 Furthermore, the following character is expected to be one of POST_TEMPLATE_VALID. 50 In any case, when matching, a PRE_TEMPLATE_VALID will be enforced before a variable and any POST_TEMPLATE_VALID character will match the template just after a variable. 51 """ 52 m = numpy.empty((len(prediction)+1, len(UCDP_TEMPLATE)+1), dtype=numpy.int64) 53 parents = numpy.zeros((len(prediction)+1, len(UCDP_TEMPLATE)+1), dtype=numpy.int8) 54 55 m[:, 0] = numpy.arange(len(prediction)+1) 56 parents[1:, 0] = 1 57 58 template_init = (numpy.array(list(UCDP_TEMPLATE))==TEMPLATE_MARKER)*(BREAK_PENALTY-1)+1 59 m[0, 1:] = template_init.cumsum() 60 parents[0, 1:] = 2 61 62 63 for y in range(1, len(prediction)+1): 64 for x in range(1, len(UCDP_TEMPLATE)+1): 65 if x>1 and UCDP_TEMPLATE[x-2] == TEMPLATE_MARKER: 66 addition = 1 67 deletion = BREAK_PENALTY 68 transformation = (0 if prediction[y-1] in POST_TEMPLATE_VALID else BREAK_PENALTY) 69 elif UCDP_TEMPLATE[x-1] == TEMPLATE_MARKER: 70 addition = 0 71 deletion = BREAK_PENALTY 72 transformation = (0 if prediction[y-1] in PRE_TEMPLATE_VALID else BREAK_PENALTY) 73 else: 74 addition = 1 75 deletion = 1 76 transformation = (UCDP_TEMPLATE[x-1] != prediction[y-1]) 77 78 m[y][x] = min(m[y-1][x] + addition, m[y][x-1] + deletion, m[y-1][x-1] + transformation) 79 if m[y-1][x] + addition == m[y][x]: 80 parents[y][x] |= 1 81 if m[y][x-1] + deletion == m[y][x]: 82 parents[y][x] |= 2 83 if m[y-1][x-1] + transformation == m[y][x]: 84 parents[y][x] |= 4 85 86 matches = {location+1: [] for location, value in enumerate(UCDP_TEMPLATE) if value == TEMPLATE_MARKER} 87 y, x = len(prediction), len(UCDP_TEMPLATE) 88 while (y, x) != (0, 0): 89 if UCDP_TEMPLATE[x-1] == TEMPLATE_MARKER: 90 if parents[y][x] & 1: 91 y -= 1 92 matches[x].append(prediction[y]) 93 elif parents[y][x] & 4: 94 y -= 1 95 x -= 1 96 elif parents[y][x] & 2: 97 x -= 1 98 else: 99 if parents[y][x] & 4: 100 y -= 1 101 x -= 1 102 elif parents[y][x] & 1: 103 y -= 1 104 elif parents[y][x] & 2: 105 x -= 1 106 107 return { field: "".join(match[::-1]).strip() for field, (_, match) in zip(UCDP_FIELDS, sorted(matches.items())) } 108 109 110 class eve_template_generator(): 111 def __init__(self, passage, triggers, roles, input_style, output_style, vocab, instance_base=False): 112 """ 113 generate strctured information for events 114 115 args: 116 passage(List): a list of tokens 117 triggers(List): a list of triggers 118 roles(List): a list of Roles 119 input_style(List): List of elements; elements belongs to INPUT_STYLE_SET 120 input_style(List): List of elements; elements belongs to OUTPUT_STYLE_SET 121 instance_base(Bool): if instance_base, we generate only one pair (use for trigger generation), else, we generate trigger_base (use for argument generation) 122 """ 123 self.raw_passage = passage 124 self.triggers = triggers 125 self.roles = roles 126 self.events = self.process_events(passage, triggers, roles) 127 self.input_style = input_style 128 self.output_style = output_style 129 self.vocab = vocab 130 self.event_templates = [] 131 if instance_base: 132 for e_type in self.vocab['event_type_itos']: 133 theclass = getattr(sys.modules[__name__], e_type.replace(':', '_').replace('-', '_'), False) 134 if theclass: 135 self.event_templates.append(theclass(self.input_style, self.output_style, passage, e_type, self.events)) 136 else: 137 print(e_type) 138 139 else: 140 for event in self.events: 141 theclass = getattr(sys.modules[__name__], event['event type'].replace(':', '_').replace('-', '_'), False) 142 assert theclass 143 self.event_templates.append(theclass(self.input_style, self.output_style, event['tokens'], event['event type'], event)) 144 self.data = [x.generate_pair(x.trigger_text) for x in self.event_templates] 145 self.data = [x for x in self.data if x] 146 147 def get_training_data(self): 148 return self.data 149 150 def process_events(self, passage, triggers, roles): 151 """ 152 Given a list of token and event annotation, return a list of structured event 153 154 structured_event: 155 { 156 'trigger text': str, 157 'trigger span': (start, end), 158 'event type': EVENT_TYPE(str), 159 'arguments':{ 160 ROLE_TYPE(str):[{ 161 'argument text': str, 162 'argument span': (start, end) 163 }], 164 ROLE_TYPE(str):..., 165 ROLE_TYPE(str):.... 166 } 167 'passage': PASSAGE 168 } 169 """ 170 171 events = {trigger: [] for trigger in triggers} 172 173 for argument in roles: 174 trigger = argument[0] 175 events[trigger].append(argument) 176 177 event_structures = [] 178 for trigger, arguments in events.items(): 179 eve_type = trigger[2] 180 eve_text = "UNK" 181 eve_span = (trigger[0], trigger[1]) 182 argus = {} 183 for argument in arguments: 184 role_type = argument[1][2] 185 if role_type not in argus.keys(): 186 argus[role_type] = [] 187 argus[role_type].append({ 188 'argument text': argument[1][3], 189 'argument span': (argument[1][0], argument[1][1]), 190 }) 191 event_structures.append({ 192 'trigger text': eve_text, 193 'trigger span': eve_span, 194 'event type': eve_type, 195 'arguments': argus, 196 'passage': ' '.join(passage), 197 'tokens': passage 198 }) 199 return event_structures 200 201 class event_template(): 202 def __init__(self, input_style, output_style, passage, event_type, gold_event=None): 203 self.input_style = input_style 204 self.output_style = output_style 205 self.output_template = self.get_output_template() 206 self.passage = ' '.join(passage) 207 self.tokens = passage 208 self.event_type = event_type 209 if gold_event is not None: 210 self.gold_event = gold_event 211 if isinstance(gold_event, list): 212 # instance base 213 self.trigger_text = " and ".join([x['trigger text'] for x in gold_event if x['event type']==event_type]) 214 self.trigger_span = [x['trigger span'] for x in gold_event if x['event type']==event_type] 215 self.arguments = [x['arguments'] for x in gold_event if x['event type']==event_type] 216 else: 217 # trigger base 218 self.trigger_text = gold_event['trigger text'] 219 self.trigger_span = [gold_event['trigger span']] 220 self.arguments = [gold_event['arguments']] 221 else: 222 self.gold_event = None 223 224 @classmethod 225 def get_keywords(self): 226 pass 227 228 def generate_pair(self, query_trigger): 229 """ 230 Generate model input sentence and output sentence pair 231 """ 232 input_str = self.generate_input_str(query_trigger) 233 output_str, gold_sample = self.generate_output_str(query_trigger) 234 return (input_str, output_str, self.gold_event, gold_sample, self.event_type, self.tokens) 235 236 def generate_input_str(self, query_trigger): 237 return None 238 239 def generate_output_str(self, query_trigger): 240 return (None, False) 241 242 def decode(self, prediction): 243 pass 244 245 def evaluate(self, predict_output): 246 assert self.gold_event is not None 247 converted_gold = self.get_converted_gold() 248 correct = 0 249 total = 0 250 per_type = { rtype: [] for rtype in ROLE_TO_TYPE.values() } 251 for role, argument in converted_gold.items(): 252 total += 1 253 value = (argument == predict_output.get(role)) 254 correct += value 255 per_type[ROLE_TO_TYPE[role]].append(value) 256 return { 257 'gold_tri_num': 0, 258 'pred_tri_num': 0, 259 'match_tri_num': 0, 260 'gold_arg_num': len(converted_gold), 261 'pred_arg_num': len(predict_output), 262 'match_arg_id': correct, 263 'match_arg_cls': correct, 264 'aec_agg': sum(sum(values)/len(values) for values in per_type.values())/len(per_type) 265 } 266 267 def get_converted_gold(self): 268 converted_gold = {} 269 for argu in self.arguments: 270 for arg_type, arg_list in argu.items(): 271 for arg in arg_list: 272 converted_gold[arg_type] = arg['argument text'].strip() 273 return converted_gold 274 275 276 class UCDPDeath(event_template): 277 278 def __init__(self, input_style, output_style, passage, event_type, gold_event=None): 279 super().__init__(input_style, output_style, passage, event_type, gold_event) 280 281 @classmethod 282 def get_keywords(self): 283 return ['die', 'kill'] 284 285 def get_output_template(self): 286 output_template = '' 287 for o_style in OUTPUT_STYLE_SET: 288 if o_style in self.output_style: 289 if o_style == 'argument:sentence': 290 output_template += ' \n {}'.format(UCDP_TEMPLATE.format(**{field: TYPE_PH_MAP[role] for field, role in ROLE_TO_TYPE.items()})) 291 return ('\n'.join(output_template.split('\n')[1:])).strip() 292 293 def generate_input_str(self, query_trigger): 294 input_str = self.passage 295 for i_style in INPUT_STYLE_SET: 296 if i_style in self.input_style: 297 if i_style == 'event_type': 298 input_str += ' \n {}'.format('death in conflict event') 299 if i_style == 'event_type_sent': 300 input_str += ' \n {}'.format('The event is related to conflict where someone is being killed.') 301 if i_style == 'keywords': 302 input_str += ' \n Similar triggers such as {}'.format(', '.join(self.get_keywords())) 303 if i_style == 'triggers': 304 input_str += ' \n The event trigger word is {}'.format(query_trigger) 305 if i_style == 'template': 306 input_str += ' \n {}'.format(self.output_template) 307 308 return input_str 309 310 def generate_output_str(self, query_trigger): 311 assert self.gold_event is not None 312 output_str = '' 313 gold_sample = False 314 for o_style in OUTPUT_STYLE_SET: 315 if o_style in self.output_style: 316 if o_style == 'argument:sentence': 317 output_texts = [] 318 for argu in self.arguments: 319 filler = { 320 # In our case, the role should always be in argu 321 role: argu[role][0]["argument text"] if role in argu else TYPE_PH_MAP[rtype] 322 for role, rtype in ROLE_TO_TYPE.items() 323 } 324 output_texts.append(UCDP_TEMPLATE.format(**filler)) 325 gold_sample = True 326 output_str += ' \n {}'.format(' <sep> '.join(output_texts)) 327 328 output_str = ('\n'.join(output_str.split('\n')[1:])).strip() 329 return (output_str, gold_sample) 330 331 def decode(self, preds): 332 return fuzzy_align(preds)