物語の展開によってactantsの要素が追加・削除する

前回の問題として、以下のようなものがあった。

• 登場人物がschedule_stackの最後の状態に合わない行動ばかりを提案すると、物語が永久に先に進まない。

• determine_next_actionをゴリゴリ回す部分を書いていない

• 物語の展開によってactantsの要素が追加・削除された場合に対応していない。

• global_logに集められた文章を小説の形に記述し直す必要がある

そのうちの、actantsの追加・削除に今日は取り組んだ。
そしてこのコード、なんと、書き散らしたまま、テストしていないのだ。
動いたらすごいね！

コード

OPENAI_API_KEY =　"YOUR_KEY"

import openai
import json
from langchain.chat_models import ChatOpenAI
from langchain.schema import AIMessage
from langchain.schema import HumanMessage
from langchain.schema import SystemMessage
from langchain import PromptTemplate

def is_valid_json(input_string):
    if input_string == None or input_string == "":
      return False

    try:
        json.loads(input_string)
        return True
    except:
        return False


def gpt_response_from_lists(system_content_list = [], user_content_list = [], assistant_content_list = [], max_tokens = None):
    chat = ChatOpenAI(model_name="gpt-3.5-turbo-16k", openai_api_key= OPENAI_API_KEY)

    if max_tokens != None:
      ChatOpenAI(model_name="gpt-3.5-turbo-16k", openai_api_key= OPENAI_API_KEY, max_tokens=max_tokens)
    else:
      ChatOpenAI(model_name="gpt-3.5-turbo-16k", openai_api_key= OPENAI_API_KEY)

    messages = []

    for system_content in system_content_list:
      messages.append(SystemMessage(content = system_content))

    list_length = max(len(assistant_content_list), len(user_content_list))

    for i in range(0, list_length):

      if i < len(user_content_list):
        messages.append(HumanMessage(content = user_content_list[i]))

      if i < len(assistant_content_list):
        messages.append(AIMessage(content = assistant_content_list[i]))

    response = chat(messages)

    return response

class Action_Object:
    def __init__(self, action = "", who_took_action = "Anonymous", consequence = "Undetermined", score = 0.0):
        self.action = action
        self.who_took_action = who_took_action
        self.consequence = consequence
        self.score = score
    
    def __str__(self):
        return str({"action": self.action, "who_took_action": self.who_took_action, "consequence": self.consequence})

global_log = dict()

class Character:
    initial_system_content_template = """
    I am trying to write a story. You will assist by playing the role of a character in this story.
    I will provide information about the character you are to portray as follows. From then on, please respond in character.

    Character Name: {name}
    Character Personality: {personality}
    Charecter Goal: {goal}
    Current Need: {current_need}
    Log: {log}
    """

    initial_system_content = ""
    current_need = ""
    name = ""
    personality = ""
    log = dict()
    
    def __init__(self, name = "", personality = "" , goal = "", current_need = "", log = [],):
        self.name = name
        self.personality = personality
        self.goal = goal
        self.current_need = current_need
        for i in range(len(log)):
            self.log[i] = log[i]

        self.refresh_initial_system_content()

    def __str__(self):
        return self.name

    def refresh_initial_system_content(self):
        log = json.dumps(self.log)

        prompt_template = PromptTemplate(input_variables=["name", "personality", "goal", "current_need", "log"], template=self.initial_system_content_template)
        self.initial_system_content = prompt_template.format(name=self.name, personality=self.personality, goal=self.goal, current_need=self.current_need, log=log,)

    def set_current_need(self, current_need):
        self.current_need = current_need
        self.refresh_initial_system_content()
    
    def set_name(self, name):
        self.name = name
        self.refresh_initial_system_content()
    
    def set_personality(self, personality):
        self.personality = personality
        self.refresh_initial_system_content()

    def set_goal(self, goal):
        self.goal = goal
        self.refresh_initial_system_content()

    def add_to_logs(self, message):
        global global_log
        self.log[len(self.log)] = message
        global_log[len(global_log)] = message
        self.refresh_initial_system_content()

    def add_to_self_log(self, message):
        self.log[len(self.log)] = message
        self.refresh_initial_system_content()

    def evaluate_action_and_reassess_need(self, action_object):
        action = action_object.action
        who_took_action = action_object.who_took_action
        consequence = action_object.consequence

        # if action's type is not string, stringfy it
        if type(action) != str:
            action = str(action)

        log = json.dumps(self.log)

        user_content_template = """
        Given the following infomation, please infer the purpose of the action.
        Action: {action}
        Who took action: {who_took_action}
        Consequence: {consequence}

        The output should be a string that describes the intent of the action from your perspective.
        sample output: "I was beaten up by the police, and I assume that the police wanted to intimidate me."
        """
        
        prompt_template = PromptTemplate(input_variables=["action", "who_took_action", "consequence"], template=user_content_template)
        user_content = prompt_template.format(action=action, who_took_action=who_took_action, consequence=consequence,)
        inferred_purpose = gpt_response_from_lists(system_content_list=[self.initial_system_content], user_content_list=[user_content]).content
        self.add_to_self_log(f"{who_took_action}'s action: {action}.")
        self.add_to_self_log(f"Consequence of previous action: {consequence}.")
        self.add_to_logs(f"{self.name}'s thought: " + inferred_purpose)

        user_content_template = """
        Given the following information, suggest what your immediate need might now be. It is acceptable for the immediate need to be the same as the current need.
        Action: {action}
        Who took action: {who_took_action}
        Consequence: {consequence}

        The output should be a string that describes your immediate need.
        """
        
        prompt_template = PromptTemplate(input_variables=["action", "who_took_action", "consequence",], template=user_content_template)
        user_content = prompt_template.format(action=action, who_took_action=who_took_action, consequence=consequence,)
        new_need = gpt_response_from_lists(system_content_list=[self.initial_system_content], user_content_list=[user_content]).content
        old_need = self.current_need
        self.set_current_need(new_need)
        self.add_to_logs(f"{self.name}'s need renewed: {new_need}")

        user_content_template = """
        Your current need is {new_need}. Your previous need was {old_need}.
        Tell me what you think the difference between the two needs is and why you did (not) change your need.
        """

        prompt_template = PromptTemplate(input_variables=["new_need", "old_need"], template=user_content_template)
        user_content = prompt_template.format(new_need=new_need, old_need=old_need)
        determination = gpt_response_from_lists(system_content_list=[self.initial_system_content], user_content_list=[user_content]).content
        self.add_to_logs(f"{self.name}'s thought: " + determination)

        response = {"inferred_purpose": inferred_purpose, "new_need": new_need, "determination": determination,}

        return response

    def consider_next_actions(self, actants = []):
        actions = dict()

        if len(actants) == 0:
            actants.append(self.name)

        action_user_content_template = """
        Consider your next move in the current scene where there is {actant} involved.
        If this element aids you in achieving your goals or satisfies your needs, figure out a way to incorporate it into your action.
        But, if it hinders your goal or needs, think about a strategy to eliminate or neutralize it.
        To determine whether the element is helpful or harmful, it might helpful to refer to the log.

        The output should descrive and only describe your next action, and don't describe the consequence of the action.
        output example: "I will go to the police station and ask for the release of my friend."
        """

        consequence_user_content_template = """
        Consider the consequence of your action.
        To determine the consequence, it might helpful to refer to the log.
        output example: "I went to the police station and ask for the release of my friend, then I was arrested."

        Action you took: {action}
        """
        
        for actant in actants:
            # if actant is not string, stringfy it
            if type(actant) != str:
                actant = str(actant)

            # determine the action
            prompt_template = PromptTemplate(input_variables=["actant"], template=action_user_content_template)
            user_content = prompt_template.format(actant = actant)
            action = gpt_response_from_lists(system_content_list=[self.initial_system_content], user_content_list=[user_content], max_tokens=10).content

            # determine the consequence
            prompt_template = PromptTemplate(input_variables=["action"], template=consequence_user_content_template)
            user_content = prompt_template.format(action = action)
            consequence = gpt_response_from_lists(system_content_list=[self.initial_system_content], user_content_list=[user_content], max_tokens=10).content

            # store the action to the actions dictionary
            actions[actant] = Action_Object(action=action, who_took_action=self.name, consequence=consequence)

        return actions

class Game_Master:
    def __init__(self, actants = [], schedule_stack = [],):
        self.actants = actants
        self.schedule_stack = schedule_stack

    def add_actant(self, actant):
        self.actants.append(actant)

    def add_schedule_stack(self, situation):
        self.schedule_stack.append(situation)

    def remove_actant(self, name):
        for i in range(len(self.actants)):
            if str(self.actants[i]) == str(name)
                self.actants.pop(i)
                break
    
    def pop_schedule_stack(self, situation = None):
        if situation == None:
            return self.schedule_stack.pop()

        for i in range(len(self.schedule_stack)):
            if str(self.schedule_stack[i]) == str(situation):
                return self.schedule_stack.pop(i)

    def aggregate_action_candidates(self, action_object):
        action_candidates = []
        for actant in self.actants:
            if type(actant) == Character:
                action_candidates.append(actant.evaluate_action_and_reassess_need(action_object))

        return action_candidates


    def determine_next_action(self, action_candidates):
        #　Consequenseとsheduled_situationが最もマッチするものを選択する。
        # sheduled_situationを取得する。
        sheduled_situation = self.schedule_stack[-1]

        # 点数をつける。点数が付けられていない場合は、繰り返す。ここで、どのactionが取られるかが決定する。
        max_score = 0
        max_scored_action = None
        for key in list(action_candidates.keys()):
            action_candidates[key] = self.calculate_score(sheduled_situation, action_candidates[key])

            if action_candidates[key].score >= max_score:
                max_score = action_candidates[key].score
                max_scored_action = action_candidates[key]

        # 点数が最も高いものが9点未満の場合、スタックはそのままにする。9点の場合、スタックからpopする。
        if max_score == 9:
            self.pop_schedule_stack()

        # max_scored_actionによって削除・退場するactantの処理を行う。
        removed_actants = self.remove_actant_with_action_object(max_scored_action)


        # consequenceによって追加されたactantを追加する。
        added_actants = self.add_actant_with_action_object(max_scored_action)

        return {"max_scored_action": max_scored_action, "removed_actants": removed_actants, "added_actants": added_actants}

    def remove_actant_with_action_object(self, action_object):
        consequence = action_object.consequence

        system_content = """
        You will be presented with a list of elements of actants, and the next consequence of the scene.
        Based on that information, list out people or things that have exited, been destroyed, died, or no longer function.
        The listed actants should be separated by commas (,). Also, the output should only be a comma-separated string of actant names, do not output any other strings.
        """

        user_content_example_1 = """
        actants: Bocchi,Guitar,room
        consequence: Bocchi awkwardly bowed and then quickly left the room.
        """

        assistant_content_example_1 = """
        Bocchi
        """

        user_content_example_2 = """
        actants: Frogman,cigarette
        consequence: Surprisingly, Frogman ate the cigarette.
        """

        assistant_content_example_2 = """
        cigarette
        """

        user_content_example_3 = """
        actants: Dolton,Anna,village
        consequence: A helicopter and a passenger plane crashed into the village. The village chief, Dolton, was caught in it and lost his life. Anna, the wife of the village chief, was terrified and fled.
        """

        assistant_content_example_3 = """
        Dolton,Anna
        """

        actants_names = ",".join([str(actant) for actant in self.actants])

        user_content = f"""
        actants: {actants_names}
        consequence: {consequence}
        """

        system_contents = [system_content]
        user_contents = [user_content_example_1, user_content_example_2, user_content_example_3, user_content]
        assistant_contents = [assistant_content_example_1, assistant_content_example_2, assistant_content_example_3]

        actants_to_remove = gpt_response_from_lists(system_content_list=system_contents, user_content_list=user_contents, assistant_content_list=assistant_contents).content.split(",")

        for actant in actants_to_remove:
            actant = actant.strip().strip("\"").strip("\n")
            self.remove_actant(actant)

        return actants_to_remove

    def add_actant_with_action_object(self, action_object):
        consequence = action_object.consequence

        system_content = """
        You will be presented with a list of elements of actants, and the next consequence of the scene.
        Based on that information, list out people or things that have newly entered, been created, or have begun to function.
        Then, determine the actant has will or not and list them in json format.
        The actant that is already in the actants list may not be included in the list, but it is not bad to include it.
        output example: {"Alice" : {"has_will": true}, "Bycycle" : {"has_will": false}, "cat" : {"has_will": true}, "Doraemon" : {"has_will": true}}
        """

        user_content_example_1 = """
        actants: Bocchi,room
        consequence: Bocchi awkwardly bowed and then quickly left the room, and a guitar was left behind.
        """

        assistant_content_example_1 = """
        {"Bocchi's guitar" : {"has_will": false}}'
        """

        user_content_example_2 = """
        actants: Laboratory
        consequence: Suddenly, Frogman advented and vomited Spiderman and a cigarette.
        """

        assistant_content_example_2 = """
        {"Frogman" : {"has_will": true}, "Spiderman" : {"has_will": true}, "cigarette" : {"has_will": false}}
        """

        user_content_example_3 = """
        actants: Dolton,Anna,village
        consequence: A helicopter and a passenger plane crashed into the village. The village chief, Dolton, was caught in it and lost his life. Anna, the wife of the village chief, was terrified and fled.
        """

        assistant_content_example_3 = """
        {"helicopter" : {"has_will": false}, "passenger plane" : {"has_will": false}}
        """

        actants_names = ",".join([str(actant) for actant in self.actants])

        user_content = f"""
        actants: {actants_names}
        consequence: {consequence}
        """

        system_contents = [system_content]
        user_contents = [user_content_example_1, user_content_example_2, user_content_example_3, user_content]
        assistant_contents = [assistant_content_example_1, assistant_content_example_2, assistant_content_example_3]

        success = False
        trial_count = 0

        while not success and trial_count < 5:
            trial_count += 1
            actants_to_add = gpt_response_from_lists(system_content_list=system_contents, user_content_list=user_contents, assistant_content_list=assistant_contents).content
            if is_valid_json(actants_to_add):
                success = True

        if trial_count >= 5:
            print("Failed to get valid json from GPT-3.")
            return

        actants_to_add_json = json.loads(actants_to_add)

        existing_actants_names = [str(actant) for actant in self.actants]
        added_actants = []

        for key in list(actants_to_add_json.keys()):
            if key in existing_actants_names:
                continue

            if actants_to_add_json[key]["has_will"] == False :
                self.add_actant(key)
                added_actants.append(key)
            else:
                new_character = self.create_character(name = key, first_log = consequence)
                self.add_actant(new_character)
                added_actants.append(new_character)
        
        return added_actants

    def create_character(self, name, first_log = None):
        system_content = """
            You will be provided with a character's name (name), the record of the story so far (global_log), and the first record (first_log) that character holds in this story. The first_log also represents the most recent event for that character.
            Based on this information, please determine the character's personality, goals, and current desires.
            The output should be in JSON format.
            Output example: {'name': 'Alice', 'personality': 'kind', 'goal': 'to become a better person', 'current_need': 'to feel loved'}
            """

        user_content_example_1 = """
            name: Alice
            global_log: {"1": {"action": "Seiji saw a girl attempting suicide and falling into a river", "consequence": "the girl fell into the river"},
                        "2": {"action": "Alice was saved by Seiji.", "consequence": "Alice was saved by Seiji."},
                        "3": {"Seiji's thought": "What a beautiful girl she is! Why did she fall into the river?"}}
            first_log: Alice was saved by Seiji.
            """


        assistant_content_example_1 = """
            {'name': 'Alice', 'personality': 'grateful', 'goal': 'to repay Seiji', 'current_need': 'to understand her own feelings'}
            """


        user_content = f"""
            name:{name}
            global_log: {str{global_log}}
            first_log: {first_log}
            """

        system_contents = [system_content]
        user_contents = [user_content_example_1, user_content]
        assistant_contents = [assistant_content_example_1]

        success = False
        trial_count = 0
        new_character = None

        while not success and trial_count < 5:
            trial_count += 1
            character_content = gpt_response_from_lists(system_content_list=system_contents, user_content_list=user_contents, assistant_content_list=assistant_contents).content
            if is_valid_json(character_content):
                character_content_json = json.loads(character_content)
                # check if character_content_json has all the keys
                if "name" in character_content_json and "personality" in character_content_json and "goal" in character_content_json and "current_need" in character_content_json:
                    new_character = Character(name = character_content_json["name"],
                                            personality = character_content_json["personality"],
                                            goal = character_content_json["goal"],
                                            current_need = character_content_json["current_need"]
                                            log = [first_log])
                    success = True

        if trial_count >= 5:
            print("Failed to get valid json from GPT-3.")
            return

        return new_character


    def calculate_score(self, sheduled_situation, action_candidate):
        system_content = """
        The information provided to you includes the scheduled_situation, action_candidate, and global_log.
        scheduled_situation: This is the situation that should be accomplished next in the narrative. It is typically provided as a string.
        action_candidate: This is a candidate for the next action to be taken in the narrative. It is given as a parsed string in JSON format.
            An example of an action_candidate: {"action": "This item represents the action taken.", "who_took_action": "This represents who took the action.", "consequence": "This roughs out the result of the action."}
        global_log: This is a record of the actions that have occurred in the narrative. It is given as a parsed string in JSON format.

        You need to consider the combination of the scheduled_situation and action_candidates and provide a score. If the scheduled_situation has been achieved, give a score of 9. If there has been no progress towards achieving the scheduled_situation, give a score of 0. The output must always be parsable as JSON.

        Output example: '{"matchness": "The scheduled_situation has been achieved. However, it is inconsistent with the global_log", "score": 5}'
        Output example: '{"matchness": "It is consistent with the global_log. However, the scheduled_situation has not been achieved.", "score": 5}'
        Output example: '{"matchness": "It is inconsistent with the global_log, and the scheduled_situation has not been achieved.", "score": 0}'
        Output example: '{"matchness": "It is inconsistent with the global_log, and the scheduled_situation has not been achieved. However, it seems likely to approach a situation where the scheduled_situation can be achieved.", "score": 7}'
        Output example: '{"matchness": "It is consistent with the global_log. In addition, the scheduled_situation has been achieved.", "score": 9}'
        """

        # 点数をつける。点数が付けられていない場合は、繰り返す。
        system_contents = [system_content]
        stringified_action_candidate = str(action_candidate)

        user_content_template = """
        sheduled_situation: {scheduled_situation}
        action_candidate: {action_candidate}
        global_log: {global_log}
        """

        user_content = user_content_template.format(scheduled_situation = sheduled_situation, action_candidate = stringified_action_candidate, global_log = str(global_log))
        
        has_result_valid_score = False
        score = None

        trial_count = 0

        while not has_result_valid_score and trial_count < 1:
            trial_count += 1
            print("trial_count: ", trial_count)

            result = gpt_response_from_lists(system_contents, [user_content]).content
            print("result: ", result)
            if is_valid_json(result):
                result = json.loads(result)
                print("type(result): ", type(result))

                #check if "score" is in result
                if "score" not in result:
                    continue

                #check if result["score"] can be interpreted as a integer
                try:
                    score = int(result["score"])
                    has_result_valid_score = True
                except:
                    continue

        action_candidate.score = score

        return action_candidate

    

明日やること

明日は、動くかどうかをチェックする。
万一うごいたら、schedule_stackが尽きるまで物語を生成するようなコードを書く。