gpt_enterprise.scrum_master

  1"""
  2\U0001F3C2
  3Scrum Master
  4\U0001F3C2
  5"""
  6
  7import os
  8import ast
  9import time
 10import json
 11import asyncio
 12from typing import List, Dict
 13
 14from gpt_enterprise.gpt_utils import generate_text
 15from gpt_enterprise.employee import Employee
 16
 17
 18MANAGER_PROMPTS_PATH = os.path.join(os.path.dirname(__file__), "prompts", "managers")
 19
 20
 21class ScrumMaster:
 22    """
 23    A scrum master will try to achieve CEO guidelines by creating tasks.
 24    The scrum master will plan tasks and assign them to employees to achieve
 25    CEO guidelines inside the given enterprise.
 26    """
 27
 28    def __init__(
 29        self,
 30        ceo_guidelines: str,
 31        manager_retry: int,
 32        output_directory: str,
 33        interactive: bool = False,
 34    ):
 35        """
 36
 37        Args:
 38            ceo_guidelines (str): _description_
 39            manager_retry (int): _description_
 40            output_directory (str): _description_
 41            interactive (bool): Defaults to False
 42        """
 43        with open(
 44            os.path.join(
 45                (
 46                    MANAGER_PROMPTS_PATH
 47                    if not os.getenv("CUSTOM_MANAGER_PROMPTS_PATH")
 48                    else os.getenv("CUSTOM_MANAGER_PROMPTS_PATH")
 49                ),
 50                "scrum_master.txt",
 51            ),
 52            "r",
 53        ) as file:
 54            self.role = file.read()
 55        self.ceo_guidelines = ceo_guidelines
 56        self.manager_retry = manager_retry
 57        self.output_directory = output_directory
 58        self.interactive = interactive
 59        self.emoji = "\U0001F3C2"
 60        self.tasks = None
 61
 62    def plan_tasks(self, employees: List[Employee]) -> List[Dict]:
 63        """
 64        Tell the scrum master to make create and plan tasks to achieve guidelines
 65
 66        Returns:
 67            List[object]: A list with all tasks
 68        """
 69        # Build the list of hired employees with their role name and name
 70        hired_employees = [
 71            {"name": employee.name, "role_name": employee.role_name}
 72            for employee in employees
 73            if employee.name != "GUY"
 74        ]
 75        self.ceo_guidelines += f"AVAILABLE_EMPLOYEES: {hired_employees}"
 76        for _ in range(self.manager_retry):
 77            try:
 78                print(
 79                    f"\n {self.emoji} Hey, I'm doing plans to achieve your guidelines !\n"
 80                )
 81                response = generate_text(
 82                    system_prompt=self.role,
 83                    user_prompt=f"Here are the CEO guidelines : {self.ceo_guidelines}",
 84                    model=os.getenv("GPT_VERSION", "gpt-3.5-turbo"),  # TODO
 85                    temperature=1.0,
 86                )
 87                # Convert to dict
 88                task_plan = ast.literal_eval(response.choices[0].message.content)
 89                print(json.dumps(task_plan, indent=4))
 90                if self.interactive:
 91                    if "y" in (
 92                        input(
 93                            f"\n {self.emoji} Is that task plan good for you ?"
 94                        ).lower()
 95                        or "y"
 96                    ):
 97                        return task_plan
 98                    else:
 99                        continue
100                return task_plan
101            except Exception as err:
102                print(err)
103                print(
104                    f"\n {self.emoji} I've messed up, retrying to plan tasks... \n Error : \n {response.choices[0].message.content}\n"
105                )
106        print(f"\n {self.emoji} I've messed up, I'm not able to do this...\n")
107        raise err
108
109    def do_plan(self, tasks: List[object], employees: List[Employee]) -> List[Dict]:
110        """
111        Do the given sequence of tasks
112
113        Args:
114            tasks (List[object]): List of tasks to execute
115            Object must have fields task_name, employee_name, todo, type, requirements ("yes" or "no")
116            employees (List[Employee]): List of hired employees
117
118        Returns:
119            List[Dict]: Production of the team ! All tasks have the result field
120        """
121        self.tasks = tasks
122        print(f"\n {self.emoji} Hey, I'm doing plan, just wait for the result !\n")
123        # Do until all tasks have the result field set
124        while not all(task.get("result") for task in self.tasks):
125            for index, task in enumerate(self.tasks):
126                # Do tasks without any requirements
127                # TODO: Do it asynchronously
128                if "no" in str(task.get("requirements", "")) and not task.get("result"):
129                    if self.interactive and "y" not in (
130                        input(
131                            f"{self.emoji} Ask {task['employee_name']} to go on with the task \n {task} ?"
132                        ).lower()
133                        or "y"
134                    ):
135                        continue
136                    self._employee_task(
137                        index,
138                        task,
139                        employees.get(task["employee_name"], employees.get("helpful")),
140                    )
141                # Do tasks that requires result of other tasks
142                elif (
143                    "no" not in str(task.get("requirements", "no"))
144                    and not task.get("result")
145                    and self.tasks[int(task["requirements"])].get("result")
146                ):
147                    # Add the previous employee work to the current task to
148                    # be used by the assigned employee.
149                    task[
150                        "todo"
151                    ] += f" Here is the work done by {self.tasks[int(task['requirements'])]['employee_name']} : {self.tasks[int(task['requirements'])]['result']}"
152                    if self.interactive and "y" not in (
153                        input(
154                            f"{self.emoji} Ask {task['employee_name']} to go on with the task \n {task} ?"
155                        ).lower()
156                        or "y"
157                    ):
158                        continue
159                    self._employee_task(
160                        index,
161                        task,
162                        employees.get(task["employee_name"], employees.get("helpful")),
163                    )
164        return self.tasks
165
166    async def do_plan_async(
167        self, tasks: List[object], employees: List[Employee]
168    ) -> List[Dict]:
169        """
170        Do the given sequence of tasks asynchronously (not compatible with interactive mode).
171
172        Args:
173            tasks (List[object]): List of tasks to execute
174            Object must have fields task_name, employee_name, todo, type, requirements ("yes" or "no")
175            employees (List[Employee]): List of hired employees
176
177        Returns:
178            List[Dict]: Production of the team ! All tasks have the result field
179        """
180        print(f"\n {self.emoji} Hey, I'm doing plan, just wait for the result !\n")
181        self.tasks = tasks
182        # Create futures
183        all_tasks = []
184        for task_index, task in enumerate(tasks):
185            # employee_name field may be a list of employe name
186            if isinstance(task["employee_name"], list):
187                for employee_name in task["employee_name"]:
188                    all_tasks.append(
189                        asyncio.to_thread(
190                            self._wait_for_result,
191                            task_index,
192                            task,
193                            employees.get(employee_name, employees.get("helpful")),
194                        )
195                    )
196            else:
197                all_tasks.append(
198                    asyncio.to_thread(
199                        self._wait_for_result,
200                        task_index,
201                        task,
202                        employees.get(task["employee_name"], employees.get("helpful")),
203                    )
204                )
205        # Run all futures in parallel
206        await asyncio.gather(*all_tasks)
207        return self.tasks
208
209    def _wait_for_result(self, task_index: int, task: dict, employee: Employee) -> dict:
210        """
211        Wait for required task result to be available if the current task requires it.
212
213        Args:
214            task (dict): _description_
215            employee (Employee): _description_
216
217        Returns:
218            dict: _description_
219        """
220        if "no" not in str(task.get("requirements", "no")):
221            counter = 0
222            # Test if requirements is an int or a list
223            try:
224                task_requirements = list(task["requirements"])
225            except TypeError:
226                task_requirements = [int(task["requirements"])]
227            for task_idx in task_requirements:
228                # Test if it is an int, if not continue
229                try:
230                    task_idx = int(task_idx)
231                except Exception as error:
232                    continue
233                while not self.tasks[task_idx].get("result"):
234                    time.sleep(0.01)
235                    counter += 1
236                    if counter >= 2000:
237                        print(
238                            f"Waiting for {self.tasks[int(task['requirements'])]['employee_name']} to finish..."
239                        )
240                        counter = 0
241                # Add the previous employee work to the current task to
242                # be used by the assigned employee.
243                task[
244                    "todo"
245                ] += f" Here is the work done by {self.tasks[task_idx]['employee_name']} : {self.tasks[task_idx]['result']}"
246
247        print(
248            f"{self.emoji} {employee.name} is doing task {task_index} : {task['todo']}"
249        )
250        self._employee_task(task_index, task, employee)
251
252    def _employee_task(self, task_index: int, task: dict, employee: Employee):
253        """
254        Ask given employee to do the task and set the result field.
255        The global list of tasks is then updated with the modified task.
256
257        Args:
258            task_index (int): _description_
259            task (dict): _description_
260            employee (Employee): _description_
261        """
262        todo = task["todo"]
263        if task["type"] == "image":
264            # Get number of images to ask from the task todo
265            nb_image = 2
266            if "NB_IMAGES" in todo:
267                nb_image = int(todo.split("NB_IMAGES")[1][1])
268                todo = todo.split("NB_IMAGES")[0]
269            task["result"] = employee.ask_image(
270                manager_request=todo,
271                output_directory=self.output_directory,
272                base_name=employee.name,
273                nb_image=min(nb_image, 5),
274            )
275        elif task["type"] == "text":
276            task["result"] = employee.ask_task(todo)
277        self.tasks[task_index] = task