self.result = filename
[docs]    def completed(self) -> bool:
        return self.status == "completed"
[docs]    def failed(self) -> bool:
        return self.status == "failed"
[docs]    def pending(self) -> bool:
        return self.status == "pending"
[docs]    def run(self) -> str:
        from diffusers.utils import load_image
        try:
            new_args = copy.deepcopy(self.args)
            for k, v in new_args.items():
                if k == "image":
                    new_args["image"] = load_image(v)
            if self.task in ["video_generator", "image_generator", "text_reader"]:
                self.product = self.tool(**new_args)
            else:
                self.result = self.tool(**new_args)
        except Exception as e:
            self.status = "failed"
            self.message = str(e)
        self.status = "completed"
        self.save_product()
        return self.result
[docs]class TaskExecutor:
    """Load tools to execute tasks."""
[docs]    def __init__(self, plan: Plan):
        self.plan = plan
        self.tasks = []
        self.id_task_map = {}
        self.status = "pending"
        for step in self.plan.steps:
            task = Task(step.task, step.id, step.dep, step.args, step.tool)
            self.tasks.append(task)
            self.id_task_map[step.id] = task
[docs]    def completed(self) -> bool:
        return all(task.completed() for task in self.tasks)
[docs]    def failed(self) -> bool:
        return any(task.failed() for task in self.tasks)
[docs]    def pending(self) -> bool: