【AI Agent系列】【MetaGPT多智能体学习】8. MetaGPT多智能体进阶练习 - 使用MetaGPT重构BabyAGI

class TaskCreation(Action):
    name: str = "TaskCreation"
    objective: str = ""
    async def run(self, result: str, task_description: str, task_list: str):
        if len(result) > 0:
            prompt = f"""
            You are to use the result from an execution agent to create new tasks with the following objective: {self.objective}.
            The last completed task has the result: \n{result}
            This result was based on this task description: {task_description}.\n
            """
            if task_list:
                prompt += f"These are incomplete tasks: {task_list}\n"
            prompt += "Based on the result, return a list of tasks to be completed in order to meet the objective. "
            if task_list:
                prompt += "These new tasks must not overlap with incomplete tasks. "
        else:
            prompt = f"""
            你需要根据给定的任务思考出一系列Tasks，以此来保证能够一步一步地实现该任务的目标。
            任务为: {self.objective}.
            生成的Tasks只要能实现目标就足够了，不要有其它额外的Tasks产生。
            生成的Tasks确保能用文字写出来
            """
        prompt += """
        Return one task per line in your response. The result must be a numbered list in the format:
        #. First task
        #. Second task
        The number of each entry must be followed by a period. If your list is empty, write "There are no tasks to add at this time."
        Unless your list is empty, do not include any headers before your numbered list or follow your numbered list with any other output. 
        OUTPUT IN CHINESE
        """
        prompt = prompt.replace('  ', '')
        print(f'\n*****TASK CREATION AGENT PROMPT****\n{prompt}\n')    
        rsp = await self._aask(prompt)
        return rsp

class TaskCreationAgent(Role):
    name: str = "TaskCreationAgent"
    profile: str = "TaskCreationAgent"
    objective: str = ""
    task_list: str = ""
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([TaskCreation(objective = self.objective)])
        self._watch([UserRequirement])
    async def _act(self) -> Message:
        logger.info(f"{self._setting}: ready to {self.rc.todo}")
        todo = self.rc.todo
        msg = self.get_memories(k=1)  # 获取最近的一条记忆
        
        result = ""
        task_description = ""
        
        try:
            result = msg[0].content.split("|")[0]
            task_description = msg[0].content.split("|")[1]
        except:
            result = ""
            task_description = ""
        task_creation_result = await todo.run(result = result, task_description = task_description, task_list = self.task_list)
        logger.info(f'TaskCreationAgent : {task_creation_result}')
        msg = Message(content=task_creation_result, role=self.profile,
                      cause_by=type(todo), send_to='TaskPrioritizationAgent')
        self.task_list += '\n' + task_creation_result ## 需去重
        return msg

class TaskPrioritization(Action):
    name: str = "TaskPrioritization"
    objective: str = ""
    PROMPT_TEMPLATE: str = """
    You are tasked with prioritizing the following tasks: {tasks}
    Consider the ultimate objective of your team: {objective}.
    Tasks should be sorted from highest to lowest priority, where higher-priority tasks are those that act as pre-requisites or are more essential for meeting the objective.
    Do not remove any tasks. Return the ranked tasks as a numbered list in the format:
    #. First task
    #. Second task
    The entries must be consecutively numbered, starting with 1. The number of each entry must be followed by a period.
    Do not include any headers before your ranked list or follow your list with any other output.
    OUTPUT IN CHINESE
    """
    async def run(self, tasks: str):
        prompt = self.PROMPT_TEMPLATE.format(objective = self.objective, tasks = tasks)
        prompt = prompt.replace('  ', '')
        rsp = await self._aask(prompt)
        return rsp

class TaskPrioritizationAgent(Role):
    name: str = "TaskPrioritizationAgent"
    profile: str = "TaskPrioritizationAgent"
    objective: str = ""
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([TaskPrioritization(objective = self.objective)])
        self._watch([TaskCreation])
    async def _act(self) -> Message:
        logger.info(f"{self._setting}: ready to {self.rc.todo}")
        todo = self.rc.todo
        msg = self.get_memories(k=1)  # 获取最新的一条记忆，TaskCreation刚刚生成的任务列表
        task_prioritization_result = await todo.run(tasks = msg[0].content)
        logger.info(f'TaskPrioritizationAgent : {task_prioritization_result}')
        msg = Message(content=task_prioritization_result, role=self.profile,
                      cause_by=type(todo), send_to='ExecutionAgent')
        return msg

class ExecuteAction(Action):
    name: str = "ExecuteAction"
    objective: str = ""
    async def run(self, context: str, task: str):
        prompt = f'OUTPUT IN CHINESE. Perform one task based on the following objective: {self.objective}.\n'
        if context:
            prompt += f'Take into account these previously completed tasks:\n{context}'
        prompt += f'\nYour task: {task}\nResponse:'
        prompt = prompt.replace('  ', '')
        print("ExecuteAction: ", prompt)
        rsp = await self._aask(prompt)
        return rsp

class ExecutionAgent(Role):
    name: str = "ExecutionAgent"
    profile: str = "ExecutionAgent"
    objective: str = ""
    complete_task: str = ""
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([ExecuteAction(objective = self.objective)])
        # self._watch([UserRequirement])
    async def _act(self) -> Message:
        logger.info(f"{self._setting}: ready to {self.rc.todo}")
        todo = self.rc.todo
        msg = self.get_memories(k=1)  # 获取最新的一条记忆
        # logger.info(msg)
        
        if len(msg) <= 0:
            print('Received empty response from priotritization agent. ')
            return ""
        new_tasks = msg[0].content.split("\n") if "\n" in msg[0].content else [msg[0].content]
        task_execute = ""
        for task_string in new_tasks:
            task_parts = task_string.strip().split(".", 1)
            if len(task_parts) == 2:
                task_execute = task_parts[1].strip()
                break # 找到第一个任务，退出循环
                
        execute_result = await todo.run(context = self.complete_task, task = task_execute)
        logger.info(f'ExecutionAgent : {execute_result}')
        msg = Message(content=execute_result + "|" + task_execute, role=self.profile,
                      cause_by=type(todo), send_to = "TaskCreationAgent")
        self.complete_task += '\n' + task_execute
        return msg

babyAGIEnv = Environment()
async def main(objective: str, n_round=10):
    babyAGIEnv.add_roles([
            ExecutionAgent(objective = objective), 
            TaskCreationAgent(objective = objective),
            TaskPrioritizationAgent(objective = objective)
        ])
    babyAGIEnv.publish_message(
        Message(role="Human", content=objective, cause_by=UserRequirement,
                send_to='TaskCreationAgent')
    )
    while n_round > 0:
        # self._save()
        n_round -= 1
        logger.debug(f"max {n_round=} left.")
        run_result = await babyAGIEnv.run()
        # print("run_result: ", babyAGIEnv.history)
    return babyAGIEnv.history
asyncio.run(main(objective='Plan for Solve world hunger'))