Repository: microsoft/promptflow
File: src/promptflow-devkit/promptflow/_sdk/operations/_flow_operations.py (Python)

Code before:
    @staticmethod
    def _infer_signature(entry: Union[Callable, FlexFlow, Flow, Prompty]):
        if isinstance(entry, Prompty):
            from promptflow._sdk.schemas._flow import ALLOWED_TYPES
            from promptflow.contracts.tool import ValueType
            from promptflow.core._model_configuration import PromptyModelConfiguration

            flow_meta = {
                "inputs": entry._data.get("inputs", {}),
                "outputs": entry._data.get("outputs", {"output": {"type": "string"}}),
            }
            init_dict = {}
            for field in fields(PromptyModelConfiguration):
                filed_type = type(field.type).__name__
                init_dict[field.name] = {"type": filed_type if filed_type in ALLOWED_TYPES else ValueType.OBJECT.value}
                if field.default != MISSING:
                    init_dict[field.name]["default"] = field.default
            flow_meta["init"] = init_dict
            format_signature_type(flow_meta)
        elif isinstance(entry, FlexFlow):
            entry_func = entry_string_to_callable(entry.entry_file, entry.entry)
            flow_meta, _, _ = FlowOperations._infer_signature_flex_flow(
                entry=entry_func, language=entry.language, include_primitive_output=True
            )
        elif inspect.isclass(entry) or inspect.isfunction(entry):
            flow_meta, _, _ = FlowOperations._infer_signature_flex_flow(entry=entry, include_primitive_output=True)
        else:
            # TODO support to get infer signature of dag flow
            raise UserErrorException(f"Invalid entry {type(entry).__name__}, only support callable object or prompty.")
        return flow_meta

    @staticmethod
    def _infer_signature_flex_flow(
        entry: Union[Callable, str],
        *,
        code: str = None,
        keep_entry: bool = False,
        validate: bool = True,
        language: str = FlowLanguage.Python,
        include_primitive_output: bool = False,
    ) -> Tuple[dict, Path, List[str]]:
        """Infer signature of a flow entry.

        Note that this is a Python only feature.
        """
        snapshot_list = None
        # resolve entry and code
        if isinstance(entry, str):
            if not code:
                raise UserErrorException("Code path is required when entry is a string.")
            code = Path(code)

Review comment (refactor):
this part is duplicated to origin `_infer_signature`? I would recommend that you directly add `elif` for `Prompty` and `FlexFlow` in origin `_infer_signature`.

Code after:
    @staticmethod
    def _infer_signature(entry: Union[Callable, FlexFlow, Flow, Prompty], include_primitive_output: bool = False):
        if isinstance(entry, Prompty):
            from promptflow._sdk.schemas._flow import ALLOWED_TYPES
            from promptflow.contracts.tool import ValueType
            from promptflow.core._model_configuration import PromptyModelConfiguration

            flow_meta = {"inputs": entry._data.get("inputs", {})}
            if "outputs" in entry._data:
                flow_meta["outputs"] = entry._data.get("outputs")
            elif include_primitive_output:
                flow_meta["outputs"] = {"output": {"type": "string"}}
            init_dict = {}
            for field in fields(PromptyModelConfiguration):
                filed_type = type(field.type).__name__
                init_dict[field.name] = {"type": filed_type if filed_type in ALLOWED_TYPES else ValueType.OBJECT.value}
                if field.default != MISSING:
                    init_dict[field.name]["default"] = field.default
            flow_meta["init"] = init_dict
            format_signature_type(flow_meta)
        elif isinstance(entry, FlexFlow):
            entry_func = entry_string_to_callable(entry.entry_file, entry.entry)
            flow_meta, _, _ = FlowOperations._infer_signature_flex_flow(
                entry=entry_func, language=entry.language, include_primitive_output=include_primitive_output
            )
        elif inspect.isclass(entry) or inspect.isfunction(entry):
            flow_meta, _, _ = FlowOperations._infer_signature_flex_flow(
                entry=entry, include_primitive_output=include_primitive_output
            )
        else:
            # TODO support to get infer signature of dag flow
            raise UserErrorException(f"Invalid entry {type(entry).__name__}, only support callable object or prompty.")
        return flow_meta

    @staticmethod
    def _infer_signature_flex_flow(
        entry: Union[Callable, str],
        *,
        code: str = None,
        keep_entry: bool = False,
        validate: bool = True,
        language: str = FlowLanguage.Python,
        include_primitive_output: bool = False,
    ) -> Tuple[dict, Path, List[str]]:
        """Infer signature of a flow entry.

        Note that this is a Python only feature.
        """
        snapshot_list = None
        # resolve entry and code
        if isinstance(entry, str):