    def _load_operator_from_path(self, path: Union[str, Path], function: str, arg: List[Any], kws: Dict[str, Any], tag: str = None) -> Operator:
        """
        Load operator form local path.
        Args:
            path (`Union[str, Path]`):
                Path to the operator python file.
            arg (`List[str, Any]`):
                The init args for OperatorClass.
            kws (`Dict[str, Any]`):
                The init kwargs for OperatorClass.
        Returns
            (`typing.Any`)
                The `Operator` output.
        """
        path = Path(path)
        fname = function.split('/')[1].replace('-', '_')
        op_name = function.replace('-', '_').replace('/', '.')
        if not tag:
            tag = hashlib.sha256(fname.encode('utf-8')).hexdigest()
        modname = 'towhee.operator.' + op_name + '.' + tag

        all_pkg = [item.project_name for item in list(pkg_resources.working_set)]
        if 'requirements.txt' in (i.name for i in path.iterdir()):
            with open(path / 'requirements.txt', 'r', encoding='utf-8') as f:
                reqs = f.read().split('\n')
            for req in reqs:
                need_install = []
                if not req:
                    continue
                pkg_name = re.split(r'(~|>|<|=|!|\]|\[| )', req)[0]
                pkg_name = pkg_name.replace('_', '-')
                if pkg_name not in all_pkg:
                    need_install.append(req)
            if need_install:
                subprocess.check_call([sys.executable, '-m', 'pip', 'install', *need_install])

        op = self._load_op(modname, path, fname)
        if not op:
            engine_log.warning('Load operator %s:%s:%s failed, try to use the legacy type' , modname, path, fname)
            op = self._load_legacy_op(modname, path, fname)

        return self._instance_operator(op, arg, kws) if op is not None else None
Load operator form local path.
Args:
path (`Union[str, Path]`):
Path to the operator python file.
arg (`List[str, Any]`):
The init args for OperatorClass.
kws (`Dict[str, Any]`):
The init kwargs for OperatorClass.
Returns
(`typing.Any`)
The `Operator` output.
towhee/runtime/operator_manager/operator_loader.py