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

Code before:
    def infer_signature(self, entry: Union[Callable, FlexFlow, Flow, Prompty]) -> dict:
        """Extract signature for a callable class or a function or a flow. Signature indicates the ports of a flex flow
        using the callable as entry.

        For flex flow:
            If entry is a callable function, the signature includes inputs and outputs.
            If entry is a callable class, the signature includes inputs, outputs, and init.

        For prompty flow:
            The signature includes inputs, outputs, and init. Init refers to PromptyModelConfiguration.

        For dag flow:
            The signature includes inputs and outputs.

        Type of each port is inferred from the type hints of the callable and follows type system of json schema.
        Given flow accepts json input in batch run and serve, we support only a part of types for those ports.
        Complicated types must be decorated with dataclasses.dataclass.
        Errors will be raised if annotated types are not supported.

        :param entry: entry of the flow, should be a method name relative to code
        :type entry: Callable
        :return: signature of the flow
        :rtype: dict
        """
        # TODO: should we support string entry? If so, we should also add a parameter to specify the working directory
        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": "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, Flow):
            # Get dag flow infer signature
            executable_flow = entry._init_executable()
            flow_meta = {"inputs": {}, "outputs": {}}
            for k, v in executable_flow.inputs.items():
                flow_meta["inputs"][k] = asdict(v, dict_factory=lambda x: {k: v for (k, v) in x if v})
                flow_meta["inputs"][k]["type"] = flow_meta["inputs"][k]["type"].value
            expect_item = ["type", "is_chat_output"]
            for k, v in executable_flow.outputs.items():

Review comment (suggestion):
putting implementation to `_infer_signature` is to avoid creating public call telemetry. Please move below logic to `_infer_signature`

Code after:
                f, cls = entry.__call__, entry
            else:
                f, cls = entry, None

            # callable entry must be of python, so we directly import from promptflow._core locally here
            from promptflow._core.tool_meta_generator import generate_flow_meta_dict_by_object

            flow_meta = generate_flow_meta_dict_by_object(f, cls)
            source_path = Path(inspect.getfile(entry))
            code = source_path.parent
            # TODO: should we handle the case that entry is not defined in root level of the source?
            flow_meta["entry"] = f"{source_path.stem}:{entry.__name__}"
        else:
            raise UserErrorException("Entry must be a function or a class.")

        format_signature_type(flow_meta)

        if validate:
            # this path is actually not used
            flow = FlexFlow(path=code / FLOW_FLEX_YAML, code=code, data=flow_meta, entry=flow_meta["entry"])
            flow._validate(raise_error=True)

        if include_primitive_output and "outputs" not in flow_meta:
            flow_meta["outputs"] = {
                "output": {
                    "type": "string",
                }
            }

        if not keep_entry:
            flow_meta.pop("entry", None)
        return flow_meta, code, snapshot_list

    @monitor_operation(activity_name="pf.flows.infer_signature", activity_type=ActivityType.PUBLICAPI)
    def infer_signature(self, entry: Union[Callable, FlexFlow, Flow, Prompty], **kwargs) -> dict:
        """Extract signature for a callable class or a function or a flow. Signature indicates the ports of a flex flow
        using the callable as entry.

        For flex flow:
            If entry is a callable function, the signature includes inputs and outputs.
            If entry is a callable class, the signature includes inputs, outputs, and init.

        For prompty flow:
            The signature includes inputs, outputs, and init. Init refers to PromptyModelConfiguration.

        For dag flow:
            The signature includes inputs and outputs.

        Type of each port is inferred from the type hints of the callable and follows type system of json schema.
        Given flow accepts json input in batch run and serve, we support only a part of types for those ports.
        Complicated types must be decorated with dataclasses.dataclass.