Source code for langchain_experimental.autonomous_agents.hugginggpt.task_executor
import copy
import uuid
from typing import Dict, List
import numpy as np
from langchain.tools.base import BaseTool
from langchain_experimental.autonomous_agents.hugginggpt.task_planner import Plan
[docs]class Task:
[docs]    def __init__(self, task: str, id: int, dep: List[int], args: Dict, tool: BaseTool):
        self.task = task
        self.id = id
        self.dep = dep
        self.args = args
        self.tool = tool
        self.status = "pending"
        self.message = ""
        self.result = ""
    def __str__(self) -> str:
        return f"{self.task}({self.args})"
[docs]    def save_product(self) -> None:
        import cv2
        if self.task == "video_generator":
            # ndarray to video
            product = np.array(self.product)
            nframe, height, width, _ = product.shape
            video_filename = uuid.uuid4().hex[:6] + ".mp4"
            fps = 30  # Frames per second
            fourcc = cv2.VideoWriter_fourcc(*"mp4v")  # type: ignore
            video_out = cv2.VideoWriter(video_filename, fourcc, fps, (width, height))
            for frame in self.product:
                video_out.write(frame)
            video_out.release()
            self.result = video_filename
        elif self.task == "image_generator":
            # PIL.Image to image
            filename = uuid.uuid4().hex[:6] + ".png"
            self.product.save(filename)  # type: ignore
            self.result = filename
[docs]    def completed(self) -> bool: