scripts/user_model_code/preprocess_sgd.py

import json
import os
import sys

from tqdm import tqdm

from crazyneuraluser.user_model_code.analysis_sgd import DATA_SPLIT, collect_data
from crazyneuraluser.user_model_code.utils_sgd import (
    add_str,
    compare_slot_values_in_state,
    dict2list,
    get_special_tokens,
    get_turn_intent,
    load_schema,
    split_intent,
    wrap_element,
)

"""pre-processing script for SGD

The annotations for a turn are grouped into frames, where each frame corresponds to a single service
The values of "slot_values" in user "state" is a list, where spoken variations are considered, e.g., tomorrow, 8/2
"""


class DialMetaData:
    def __init__(self, dial_id, dial):
        self.dial_id = dial_id
        self.turn_meta_list, self.scenario = self.parse(dial)  # None for system turn
        self.linearise_turns()

    def parse(self, dial):
        turn_meta_list = []
        scenario = []
        sys_turn = None  # dummy sys turn for first usr turn
        prev_intent = ""
        prev_usr_turn, prev_usr_turn_meta = (
            None,
            None,
        )  # dummpy for tracing goal change at first turn
        for turn_id, turn in enumerate(dial["turns"]):
            if turn["speaker"] == "SYSTEM":
                sys_turn = turn
                turn_meta_list.append(None)
                continue

            # init turn meta
            turn_meta = TurnMetaData(prev_intent, sys_turn, turn, self.dial_id)

            # get goal change label
            turn_meta.get_goal_change_label(prev_usr_turn, prev_usr_turn_meta)

            # update previous goal
            for prev_turn_meta in reversed(turn_meta_list):
                if prev_turn_meta is None:
                    continue
                prev_turn_meta.accumulate_constraints(turn_meta)

            # record task (intent) in scenario
            prev_intent = turn_meta.usr_intent
            if turn_meta.usr_intent not in scenario:
                scenario.append(turn_meta.usr_intent)

            turn_meta_list.append(turn_meta)
            prev_usr_turn, prev_usr_turn_meta = turn, turn_meta

        assert len(turn_meta_list) == len(dial["turns"])
        return turn_meta_list, scenario

    def linearise_turns(self):
        # linearise necessary meterials
        for turn_meta in self.turn_meta_list:
            if turn_meta is None:
                continue
            turn_meta._linearise(self.scenario)


class TurnMetaData:
    def __init__(self, prev_intent, sys_turn, usr_turn, dial_id):
        self.dial_id = dial_id
        self.sys_turn, self.usr_turn = sys_turn, usr_turn
        self.empty_token = "_Empty_"
        assert self.empty_token in SPECIAL_TOKENS["additional_special_tokens"]

        # intent
        self.usr_intent, self.service = self._get_intent(usr_turn, prev_intent)

        # utterances
        self.utt = {}
        self.utt["sys"], self.utt["usr"] = self._get_utt(sys_turn), self._get_utt(
            usr_turn
        )

        # action
        self.act2sv = {}
        self.act2sv["sys"], _ = self._parse_action(sys_turn)
        self.act2sv["usr"], self.usr_constraints = self._parse_action(usr_turn)

        # task boundary
        self._get_new_task_label(prev_intent)

        # req_alts
        self._get_req_alts_label(self.act2sv["usr"])

    def _get_intent(self, turn, prev_intent):
        """manually set the `NONE` intent to the intent of previous turn"""
        active_intent, service = get_turn_intent(
            turn
        )  # intent annotation (migt be `NONE`)
        if active_intent == "NONE":
            active_intent = prev_intent
        return active_intent, service

    def _get_utt(self, turn):
        if turn is None:
            return ""
        return turn["utterance"]

    def _parse_action(self, turn):
        """
        parse action annotation to collect turn level information
        1) act to slot-value pairs, dict{act: {slot: value}}
        2) turn level constraints, dict{'informable': dict{slot: value}, 'requestable': set(slot)}
        """
        # get mapping from act to slot-value pairs
        act2sv = {}
        info_req = {"informable": dict(), "requestable": set()}  # constraints

        if turn is None:
            return None, info_req

        for frame in turn["frames"]:
            for action in frame["actions"]:
                act, slot, values = action["act"], action["slot"], action["values"]

                # deal with empty slot or value
                if turn["speaker"] == "USER":
                    assert len(values) in [0, 1]
                if slot == "":
                    slot = self.empty_token
                value = values[0] if len(values) > 0 else self.empty_token

                # act to slot-value pairs
                if act not in act2sv:
                    act2sv[act] = {}
                assert slot not in act2sv[act]
                act2sv[act][slot] = value

                # collect constraints
                if slot in [
                    "",
                    self.empty_token,
                ]:  # only act but no constraints, e.g., AFFIRM, NEGATE
                    continue

                # turn level informalable and requestable info
                if act == "REQUEST":
                    assert slot != ""
                    info_req["requestable"].add(slot)
                else:
                    if turn["speaker"] == "USER":
                        assert act in [
                            "INFORM_INTENT",
                            "INFORM",
                            "SELECT",
                        ]  # not apply to system side
                    if (
                        act != "SELECT"
                    ):  # result offered by system is part of initial user goal
                        assert slot not in info_req["informable"]
                        info_req["informable"][slot] = value
        return act2sv, info_req

    def accumulate_constraints(self, new_turn_meta):
        """
        Add slot, slot-value pairs from a given following turn
        This function is used to form user goal by accumulating constraints backward
        """
        # only accumulate constraints with the same task/intent
        if new_turn_meta.usr_intent != self.usr_intent:
            return

        if (
            new_turn_meta.goal_change
        ):  # if goal changes at a new turn, these constraints should not be put in previous turns
            return

        # only accumulate constraints without goal change
        # if the value of a slot is changed (goal change) in a new turn,
        # this slot-value pair is not part of initial goal and should not be added into the goal of previous turns
        new_constraints = new_turn_meta.usr_constraints
        self.usr_constraints["requestable"] = self.usr_constraints["requestable"].union(
            new_constraints["requestable"]
        )
        for slot, value in new_constraints["informable"].items():
            if slot not in self.usr_constraints["informable"]:
                self.usr_constraints["informable"][slot] = value

    def _get_new_task_label(self, prev_intent):
        """get a binary label indicating if a turn starts a new task (intent) in dialogue"""
        assert prev_intent != "NONE" and self.usr_intent != "NONE"
        if self.usr_intent != prev_intent:
            self.start_new_task = True
        else:
            self.start_new_task = False

    def _get_req_alts_label(self, act2sv):
        """get a binary label indicating if usr requests alternatives"""
        if "REQUEST_ALTS" in act2sv:
            self.req_alts = True
        else:
            self.req_alts = False

    def get_goal_change_label(self, prev_usr_turn, prev_turn_meta):
        """check if goal changed (value of slot changes) between two turn states"""
        if prev_usr_turn is None:  # first usr turn
            self.goal_change = False
            return

        if (
            len(self.usr_turn["frames"]) == 1
            and self.usr_turn["frames"][0]["state"]["active_intent"] == "NONE"
        ):  # `NONE` intent
            self.goal_change = False
            return

        if self.usr_intent != prev_turn_meta.usr_intent:  # new task
            self.goal_change = False
            return

        assert prev_usr_turn["speaker"] == "USER"
        prev_state_sv, curr_state_sv = None, None
        for frame in prev_usr_turn["frames"]:
            if frame["state"]["active_intent"] == self.usr_intent:
                prev_state_sv = frame["state"]["slot_values"]

        # fix some weird cases (count very few, around 30 turns)
        if prev_state_sv is None:
            assert (
                len(prev_usr_turn["frames"]) == 1
                and prev_usr_turn["frames"][0]["state"]["active_intent"] == "NONE"
            )
            prev_state_sv = prev_usr_turn["frames"][0]["state"]["slot_values"]

        for frame in self.usr_turn["frames"]:
            if frame["state"]["active_intent"] == self.usr_intent:
                curr_state_sv = frame["state"]["slot_values"]

        assert prev_state_sv is not None and curr_state_sv is not None
        self.goal_change = compare_slot_values_in_state(
            prev_state_sv, curr_state_sv
        )  # True if goal changes

    def _linearise(self, scenario):
        self.linear_act = {}
        self.linear_act["sys"] = self._linearise_act(self.act2sv["sys"])
        self.linear_act["usr"] = self._linearise_act(self.act2sv["usr"])
        self.linear_goal = self._linearise_goal(self.usr_constraints, scenario)

    def _linearise_act(self, act2sv):
        """
        NOTE: 1) split slot/value if "_"; 2) special tokens of acts; 3) empty slot or empty value
        NOTE: filer too many values (e.g., 10 movie names) but make sure the one the user chose is present

        Return: ordered (slots sorted within act, acts sorted) linearised act sequence,
                        e.g., <ACT/> <INFORM> </ACT> <SLOT/> area </SLOT> <VALUE/> Cambridge </VALUE> ...
                        e.g., <ACT/> <REQUEST> </ACT> <SLOT/> _Empty_ </SLOT> <VALUE/> _Empty_ </VALUE>
        """
        res = ""
        if act2sv is None:
            return res

        for act in sorted(act2sv.keys()):  # sort act
            sv = act2sv[act]  # dict{slot: value}

            act = "_{}_".format(act)  # act is special token
            assert act in SPECIAL_TOKENS["additional_special_tokens"]
            act_wrap = wrap_element("ACT", act)
            res = add_str(res, act_wrap)

            sorted_sv = dict2list(sv)  # sorted sv list, [slot=value]
            for sv_pair in sorted_sv:
                slot, value = sv_pair.split("=")
                slot, value = self._basic_normalise_slot(
                    slot
                ), self._basic_normalise_value(value, slot)

                # slot
                slot_wrap = wrap_element("SLOT", slot)
                res = add_str(res, slot_wrap)

                # value
                value_wrap = wrap_element("VALUE", value)
                res = add_str(res, value_wrap)
        return res[1:]  # remove first space

    def _basic_normalise_value(self, value, slot):
        # intent value
        if slot == "intent":
            value = split_intent(value)
            return value

        # special token value
        if value in ["True", "False"]:  # Empty is already in the form of "_Empty_"
            value = "_{}_".format(value)
            assert value in SPECIAL_TOKENS["additional_special_tokens"]
            return value
        return value

    def _basic_normalise_slot(self, slot):
        if slot not in SPECIAL_TOKENS["additional_special_tokens"]:
            slot = slot.replace(
                "_", " "
            )  # e.g., `date_of_journey` -> `date of journey`
        return slot

    def _linearise_goal(self, constraints, scenario):
        """
        linearise goal representation which consists of several parts:
        scenario, task (intent), task description, constraints with informable and requestable
        e.g., <SCENARIO/> task1 task2 .. </SCENARIO>
                  <TASK/> current task </TASK> <DESC/> task description </DESC>
                  <INFORM/> <SLOT/> slot1 </SLOT> <VALUE> value1 </VALUE> .. </INFORM>
                  <REQUEST/> <SLOT> slot1 </SLOT> <SLOT> slot2 </SLOT> .. </REQUEST>
        """
        res = ""
        # scenario
        assert isinstance(scenario, list) and len(scenario) > 0
        scenario = " ".join(
            [wrap_element("INTENT", split_intent(intent)) for intent in scenario]
        )
        scenario_wrap = wrap_element("SCENARIO", scenario)
        res = add_str(res, scenario_wrap)

        # task name
        intent = split_intent(self.usr_intent)
        assert intent in scenario
        intent_wrap = wrap_element("TASK", intent)
        res = add_str(res, intent_wrap)

        # task description
        description = SERVICE2META[self.service]["intents"][self.usr_intent][
            "description"
        ]
        description_warp = wrap_element("DESC", description)
        res = add_str(res, description_warp)

        # informable
        informable = dict2list(
            constraints["informable"]
        )  # sorted sv pair list [slot=value]
        res = add_str(res, "<INFORM/>")
        for sv_pair in informable:
            slot, value = sv_pair.split("=")
            slot, value = self._basic_normalise_slot(slot), self._basic_normalise_value(
                value, slot
            )
            # slot
            slot_wrap = wrap_element("SLOT", slot)
            res = add_str(res, slot_wrap)
            # value
            value_wrap = wrap_element("VALUE", value)
            res = add_str(res, value_wrap)
        res = add_str(res, "</INFORM>")

        # requestable
        requestable = sorted(
            list(constraints["requestable"])
        )  # sorted slot list [slot]
        res = add_str(res, "<REQUEST/>")
        for slot in requestable:
            slot = self._basic_normalise_slot(slot)
            slot_wrap = wrap_element("SLOT", slot)
            res = add_str(res, slot_wrap)
        res = add_str(res, "</REQUEST>")
        return res[1:]  # remove first space


def collect_examples(dial_id, dial_meta, examples):
    num = 0
    examples[dial_id] = {}
    for turn_meta in dial_meta.turn_meta_list:
        if turn_meta is None:  # sys turn
            continue

        example_id = "{}-{}".format(dial_id, num)
        example = {
            "utterances": turn_meta.utt,
            "actions": turn_meta.linear_act,
            "goal": turn_meta.linear_goal,
            "service": turn_meta.service,
            "intent": turn_meta.usr_intent,
            "goal_change": turn_meta.goal_change,
            "start_new_task": turn_meta.start_new_task,
            "req_alts": turn_meta.req_alts,
        }
        examples[dial_id][example_id] = example
        num += 1


def prepare_data_seq(data, out_data_path):
    for split in DATA_SPLIT:
        examples = {}
        for dial_num, dial_id in enumerate(tqdm(sorted(data[split].keys()))):
            dial = data[split][dial_id]
            dial_meta = DialMetaData(dial_id, dial)
            collect_examples(dial_id, dial_meta, examples)

        with open("{}/{}.json".format(out_data_path, split), "w") as f:
            json.dump(examples, f, sort_keys=True, indent=4)
        print("Done process {} {} dialogues".format(split, len(examples)))


if __name__ == "__main__":
    if len(sys.argv) == 1:
        print("wrong arguments!")
        print("usage: python utils/preprocess_sgd.py sgd-data-path")
        sys.exit(1)

    # Set data path
    data_path = sys.argv[1]
    out_data_path = "./processed_data/sgd/"
    os.makedirs(out_data_path, exist_ok=True)

    # Load data and material as global var
    SERVICE2META, INTENTS, SLOTS = load_schema(data_path)
    SPECIAL_TOKENS = get_special_tokens()
    data = collect_data(data_path, remove_dial_switch=True)

    # Process data
    prepare_data_seq(data, out_data_path)