auto_chaos.chaos

  1"""
  2🦹🏼‍♂️
  3Chaos module
  4🦹🏼‍♂️
  5"""
  6import os
  7import pprint
  8
  9from openai.types.chat import ChatCompletionMessage
 10
 11from auto_chaos.gpt_utils import generate_text
 12from auto_chaos.base_system import BaseSystem
 13from auto_chaos.utility_system import UtilitySystem
 14
 15PROMPTS_PATH = os.path.join(os.path.dirname(__file__), "prompts")
 16
 17
 18class Chaos:
 19    """
 20    Here the chaos happen !
 21    """
 22
 23    def __init__(self, systems: list[BaseSystem], initial_state: dict) -> None:
 24        # Load chaos engineer prompt
 25        with open(
 26            os.path.join(
 27                PROMPTS_PATH,
 28                "chaos_engineer.txt",
 29            ),
 30            "r",
 31        ) as file:
 32            self.system_prompt = file.read()
 33
 34        # Prepare first system and use messages
 35        self.messages = [
 36            {"role": "system", "content": self.system_prompt},
 37            {"role": "user", "content": f"{initial_state}"},
 38        ]
 39
 40        # Prepare systems
 41        self.systems = systems
 42        self.systems.append(UtilitySystem())
 43
 44    def chaos(self, objective=5) -> None:
 45        """
 46        Do chaos
 47
 48        Args:
 49            objective (int, optional): Number of chaos iteration. Defaults to 5.
 50        """
 51        # This is the end of the chaos
 52        if objective < 0:
 53            return
 54
 55        # Let the chaos engineer do his things and give actions to be done
 56        response = None
 57        while not response:
 58            response = generate_text(
 59                self.messages,
 60                model=os.getenv("MODEL_NAME", "gpt-3.5-turbo-16k"),
 61                temperature=1,
 62            )
 63            response = response.choices[0].message
 64        self.messages.append(response.dict())
 65
 66        # Do actions in all systems
 67        print(f"🦹🏼 Doing {response.content}")
 68        system_result = []
 69        system_error = []
 70        for system in self.systems:
 71            result, error = system.do_action(response.content)
 72            system_result += result
 73            system_error += error
 74
 75        # Add the result to messages for the chaos engineer
 76        if system_result or system_error:
 77            self.messages.append(
 78                {
 79                    "role": "user",
 80                    "content": f"Result : {system_result}"
 81                    + (f", Error: {system_error}" if system_error else ""),
 82                }
 83            )
 84            print(f"        🚀 Results {system_result}")
 85            print(f"        😵 Errors {system_error}")
 86
 87        # Do chaos again
 88        self.chaos(objective - 1)
 89
 90    def report(self):
 91        """
 92        Do a report of the chaos
 93        """
 94        print(f"🦹🏼‍♂️ Doing report, please wait ...")
 95        report = []
 96
 97        # Do a description of all systems
 98        # for system in self.systems:
 99        #    system_description = f"{system.do_action('DESCRIBE')}"
100
101        # Load chaos report prompt
102        with open(
103            os.path.join(
104                PROMPTS_PATH,
105                "chaos_report.txt",
106            ),
107            "r",
108        ) as file:
109            system_prompt = file.read()
110        report.append({"role": "system", "content": system_prompt})
111
112        # Give chaos engineer all previous actions result
113        report_prompt = "Here are the results of chaos actions:"
114        for message in self.messages:
115            if isinstance(message, ChatCompletionMessage):
116                if message.role != "user":
117                    continue
118                report_prompt += f" {message.content}"
119                continue
120            if message["role"] != "user":
121                continue
122            report_prompt += f" {message['content']}"
123        report.append({"role": "user", "content": report_prompt})
124
125        # Ask chaos engineer to do a report of the chaos
126        response = generate_text(
127            report, model=os.getenv("MODEL_NAME", "gpt-3.5-turbo-16k"), temperature=1.0
128        )
129
130        # Add chaos report to messages
131        self.messages.append(
132            {"role": "assistant", "content": response.choices[0].message.content}
133        )
134
135        pp = pprint.PrettyPrinter(indent=4)
136        pp.pprint(response.choices[0].message.content)