Schema Induction using LLMs

Inducing patterns from articles using LLMs

DARPA’s KAIROS program focuses on automatically learning the structure of complex events with multiple subevents and using the results to predict subevents that were missed or that have not yet occurred.

Many events or actions in the world are “complex” in that they can be described as a sequence of distinct subevents or actions. A schema is the representation of a stereotypical series (temporally ordered) of events or actions that typically involves one or more entities (or person) which serve as event arguments, responsible for the event.

We are currently experimenting on inducing schemas with an innovative approach based on large language models (LLMs). In this approach, schemas are represented as Python classes. Since LLMs like GPT-4 and Codex have seen a large amount of both Python and English data, they are capable of representing patterns observed in natural language in Python code, given appropriate prompting. [1, 2]

In initial testing with ChatGPT, we have shown that, given a list of example schemas (usually 1-shot) in Python and a scenario name, the LM is capable of generating schemas (in Python) that resemble those produced by human curators. To achieve this, we have constructed code that can convert SDF to and from a Python representation and prompting techniques to elicit representations of this kind from ChatGPT.

These initial schemas are flat in structure. Our next steps will be to induce schemas with hierarchical structure and to develop a feedback loop between schema induction and curation.

A typical example of schema for food borne illness

The corresponding python representation of the schema used for prompting.

class FoodborneIllness:
    """
    Illness resulting from food that is spoiled or contaminated by pathogenic bacteria, viruses, parasites or toxins.
    It involves several steps:
    
    Contamination of food by some contaminant
    Consumption of contaminated food by people
    People getting sick with the onset of symptoms
    Patient are sent to a hospital for further diagnosis of the disease
    Patient undergoes treatment for food borne illness
    Patient recovers from the disease
    

    Args:
        consumer: A person who consumes food or drink that is contaminated with harmful bacteria, viruses, parasites, or toxins, leading to symptoms of foodborne illness.
        food: The food or drink that contains the contaminant and is responsible for causing the illness.
        restaurant: A business that serves food or drink to the public, and may be the source of the contaminated food or drink.
        contaminant: The harmful bacteria, viruses, parasites, or toxins that cause the illness. These contaminants can enter the food or drink during production, processing, or handling.
        hospital: A medical facility where individuals with severe cases of foodborne illness may need to receive treatment, including intravenous fluids, medications, and other forms of supportive care.
        doctor: A medical professional who can diagnose and treat the illness caused by foodborne contamination.
    """
    def __init__(self, consumer, food, restaurant, contaminant, hospital, doctor):
        self.consumer = consumer
        self.food = food
        self.restaurant = restaurant
        self.contaminant = contaminant
        self.hospital = hospital
        self.doctor = doctor

    def contamination(self, thing_becoming_contaminated, contaminant):
        """Harmful bacteria, viruses, parasites, or toxins contaminate the food or drink."""
        return 

    def consumption(self, consumer, consumed, location):
        """The contaminated food or drink is ingested and enters the body."""
        return

    def onset_of_symptoms(self, agent):
        """After consuming contaminated food, the consumer will typically experience symptoms of food poisoning within a few hours to a few days. Common symptoms include nausea, vomiting, diarrhea, stomach cramps, and fever."""
        return

    def diagnosis(self, doctor, patient, location):
        """The consumer visits a doctor and is diagnosed with food poisoning."""
        return

    def treatment(self, doctor, patient, location):
        """The doctor prescribes medication to treat the food poisoning."""
        return

    def recovery(self, agent):
        """The consumer recovers from the food poisoning."""
        return

    def execute(self):
        self.contamination(self.food, self.contaminant)
        self.consumption(self.consumer, self.food, self.restaurant)
        self.onset_of_symptoms(self.consumer)
        self.diagnosis(self.doctor, self.consumer, self.hospital)
        self.treatment(self.doctor, self.consumer, self.hospital)
        self.recovery(self.consumer)

An example of a resultant schema scenrio of airplane crash when used the above code based representation for prompting in 1-shot setting. Instruction used : “Complete the class definition of the following python class based on the example provided. Like the example first generate the steps and then generate the class definition based on that”

class TerroristAttack:
    """
    Represents a terrorist attack, a violent attack targeted at civilians for political or ideological reasons.
    It involves several steps:

    1. The planning and preparation of the terrorist attack by individuals or groups.
    2. The execution of the attack, which may involve the use of weapons, explosives, or other harmful methods.
    3. The immediate response from the affected area and emergency services.
    4. The investigation and gathering of evidence to identify the perpetrators and their motives.
    5. The support and assistance provided to the victims and their families.
    6. The implementation of security measures to prevent future attacks.

    Args:
        attackers: Individuals or groups responsible for the terrorist attack.
        target: The specific target of the attack, usually civilians.
        weapons: Weapons or tools used in the attack.
        affected_area: The location or area affected by the attack.
        emergency_services: Services involved in the immediate response to the attack.
        investigators: Individuals or teams responsible for investigating the attack.
        victims: People who are directly affected or harmed by the attack.
        security_measures: Measures implemented to prevent future attacks.
    """

    def __init__(self, attackers, target, weapons, affected_area, emergency_services, investigators, victims, security_measures):
        self.attackers = attackers
        self.target = target
        self.weapons = weapons
        self.affected_area = affected_area
        self.emergency_services = emergency_services
        self.investigators = investigators
        self.victims = victims
        self.security_measures = security_measures

    def plan_attack(self, attackers, target, weapons):
        """Plan and prepare the terrorist attack."""
        return

    def execute_attack(self, weapons):
        """Execute the terrorist attack using weapons or harmful methods."""
        return

    def immediate_response(self, affected_area, emergency_services):
        """Immediate response from the affected area and emergency services."""
        return

    def investigate_attack(self, investigators):
        """Investigate the terrorist attack to identify the perpetrators and their motives."""
        return

    def support_victims(self, victims):
        """Provide support and assistance to the victims and their families."""
        return

    def implement_security_measures(self, security_measures):
        """Implement security measures to prevent future attacks."""
        return

    def execute(self):
        self.plan_attack(self.attackers, self.target, self.weapons)
        self.execute_attack(self.weapons)
        self.immediate_response(self.affected_area, self.emergency_services)
        self.investigate_attack(self.investigators)
        self.support_victims(self.victims)
        self.implement_security_measures(self.security_measures)

To get more complex hierarchical schemas we prompt the linear schema to be expanded into sub-events by using instruction : “ Can you expand each of the functions of the class into sub functions except the execute function wherever each function which is an even can be decomposed into sub events?”

class TerroristAttack:
    """
    Represents a terrorist attack, a violent attack targeted at civilians for political or ideological reasons.
    It involves several steps:

    1. The planning and preparation of the terrorist attack by individuals or groups.
    2. The execution of the attack, which may involve the use of weapons, explosives, or other harmful methods.
    3. The immediate response from the affected area and emergency services.
    4. The investigation and gathering of evidence to identify the perpetrators and their motives.
    5. The support and assistance provided to the victims and their families.
    6. The implementation of security measures to prevent future attacks.

    Args:
        attackers: Individuals or groups responsible for the terrorist attack.
        target: The specific target of the attack, usually civilians.
        weapons: Weapons or tools used in the attack.
        affected_area: The location or area affected by the attack.
        emergency_services: Services involved in the immediate response to the attack.
        investigators: Individuals or teams responsible for investigating the attack.
        victims: People who are directly affected or harmed by the attack.
        security_measures: Measures implemented to prevent future attacks.
    """

    def __init__(self, attackers, target, weapons, affected_area, emergency_services, investigators, victims, security_measures):
        self.attackers = attackers
        self.target = target
        self.weapons = weapons
        self.affected_area = affected_area
        self.emergency_services = emergency_services
        self.investigators = investigators
        self.victims = victims
        self.security_measures = security_measures

    def plan_attack(self, attackers, target, weapons):
        """Plan and prepare the terrorist attack."""
        self.identify_objectives(target)
        self.recruit_attackers(attackers)
        self.acquire_weapons(weapons)
        self.prepare_logistics()

    def identify_objectives(self, target):
        """Identify specific objectives or targets for the attack."""
        return

    def recruit_attackers(self, attackers):
        """Recruit and organize individuals or groups responsible for the attack."""
        return

    def acquire_weapons(self, weapons):
        """Obtain or acquire weapons or tools for the attack."""
        return

    def prepare_logistics(self):
        """Arrange logistical aspects of the attack, such as transportation or safe houses."""
        return

    def execute_attack(self, weapons):
        """Execute the terrorist attack using weapons or harmful methods."""
        self.activate_explosives(weapons)
        self.open_fire(weapons)
        self.launch_coordinated_attacks()

    def activate_explosives(self, weapons):
        """Activate or detonate explosives as part of the attack."""
        return

    def open_fire(self, weapons):
        """Use firearms or other weapons to initiate the attack."""
        return

    def launch_coordinated_attacks(self):
        """Coordinate simultaneous or sequential attacks at different locations."""
        return

    def immediate_response(self, affected_area, emergency_services):
        """Immediate response from the affected area and emergency services."""
        self.evacuate_civilians(affected_area)
        self.call_emergency_services(emergency_services)
        self_treat_injured()

    def evacuate_civilians(self, affected_area):
        """Initiate the evacuation of civilians from the affected area."""
        return

    def call_emergency_services(self, emergency_services):
        """Notify and request assistance from emergency services."""
        return

    def treat_injured(self):
        """Provide immediate medical treatment to the injured individuals."""
        return

        def investigate_attack(self, investigators):
        """Investigate the terrorist attack to identify the perpetrators and their motives."""
        self.collect_evidence()
        self.interview_witnesses()
        self_analyze_forensic_data()
        self.identify_suspects(investigators)

    def collect_evidence(self):
        """Gather physical and digital evidence related to the attack."""
        return

    def interview_witnesses(self):
        """Conduct interviews with witnesses to gather information and testimonies."""
        return

    def analyze_forensic_data(self):
        """Analyze forensic data, such as DNA samples or fingerprints, to identify leads."""
        return

    def identify_suspects(self, investigators):
        """Identify potential suspects based on gathered evidence and intelligence."""
        return

    def support_victims(self, victims):
        """Provide support and assistance to the victims and their families."""
        self_offer_medical_aid(victims)
        self_provide_counseling(victims)
        self_assist_with_recovery(victims)

    def offer_medical_aid(self, victims):
        """Provide immediate medical aid and treatment to the injured victims."""
        return

    def provide_counseling(self, victims):
        """Offer psychological counseling and support to the victims and their families."""
        return

    def assist_with_recovery(self, victims):
        """Assist the victims in their recovery process, including financial support and rehabilitation."""
        return

    def implement_security_measures(self, security_measures):
        """Implement security measures to prevent future attacks."""
        self_increase_surveillance(security_measures)
        self_enhance_intelligence_gathering(security_measures)
        self_strengthen_border_security(security_measures)

    def increase_surveillance(self, security_measures):
        """Enhance surveillance systems and monitoring in potential target areas."""
        return

    def enhance_intelligence_gathering(self, security_measures):
        """Improve intelligence gathering and analysis capabilities to detect threats."""
        return

    def strengthen_border_security(self, security_measures):
        """Implement measures to strengthen border security and prevent illegal activities."""
        return

    def execute(self):
        self.plan_attack(self.attackers, self.target, self.weapons)
        self.execute_attack(self.weapons)
        self.immediate_response(self.affected_area, self.emergency_services)
        self.investigate_attack(self.investigators)
        self.support_victims(self.victims)
        self.implement_security_measures(self.security_measures)
       

References

[1] X Wang, S Li, H Ji, Code4Struct: Code Generation for Few-Shot Structured Prediction from Natural Language

[2] A Madaan, S Zhou, U Alon, Y Yang, G Neubig, Language Models of Code are Few-Shot Commonsense Learners