microsoft--promptflow/scripts/tool/utils/tool_utils.py

import inspect
from enum import Enum, EnumMeta
from typing import Callable, Union, get_args, get_origin
from promptflow.contracts.tool import ConnectionType, InputDefinition, ValueType, ToolType
from promptflow.contracts.types import PromptTemplate


def value_to_str(val):
    if val is inspect.Parameter.empty:
        # For empty case, default field will be skipped when dumping to json
        return None
    if val is None:
        # Dump default: "" in json to avoid UI validation error
        return ""
    if isinstance(val, Enum):
        return val.value
    return str(val)


def resolve_annotation(anno) -> Union[str, list]:
    """Resolve the union annotation to type list."""
    origin = get_origin(anno)
    if origin != Union:
        return anno
    # Optional[Type] is Union[Type, NoneType], filter NoneType out
    args = [arg for arg in get_args(anno) if arg != type(None)]  # noqa: E721
    return args[0] if len(args) == 1 else args


def param_to_definition(param, value_type) -> (InputDefinition, bool):
    default_value = param.default
    enum = None
    custom_type = None
    # Get value type and enum from default if no annotation
    if default_value is not inspect.Parameter.empty and value_type == inspect.Parameter.empty:
        value_type = default_value.__class__ if isinstance(default_value, Enum) else type(default_value)
    # Extract enum for enum class
    if isinstance(value_type, EnumMeta):
        enum = [str(option.value) for option in value_type]
        value_type = str
    is_connection = False
    if ConnectionType.is_connection_value(value_type):
        if ConnectionType.is_custom_strong_type(value_type):
            typ = ["CustomConnection"]
            custom_type = [value_type.__name__]
        else:
            typ = [value_type.__name__]
        is_connection = True
    elif isinstance(value_type, list):
        if not all(ConnectionType.is_connection_value(t) for t in value_type):
            typ = [ValueType.OBJECT]
        else:
            custom_connection_added = False
            typ = []
            custom_type = []
            for t in value_type:
                if ConnectionType.is_custom_strong_type(t):
                    if not custom_connection_added:
                        custom_connection_added = True
                        typ.append("CustomConnection")
                    custom_type.append(t.__name__)
                else:
                    typ.append(t.__name__)
            is_connection = True
    else:
        typ = [ValueType.from_type(value_type)]
    return InputDefinition(type=typ, default=value_to_str(default_value),
                           description=None, enum=enum, custom_type=custom_type), is_connection


def function_to_interface(f: Callable, tool_type, initialize_inputs=None) -> tuple:
    sign = inspect.signature(f)
    all_inputs = {}
    input_defs = {}
    connection_types = []
    # Initialize the counter for prompt template
    prompt_template_count = 0
    # Collect all inputs from class and func
    if initialize_inputs:
        if any(k for k in initialize_inputs if k in sign.parameters):
            raise Exception(f'Duplicate inputs found from {f.__name__!r} and "__init__()"!')
        all_inputs = {**initialize_inputs}
    all_inputs.update(
        {
            k: v
            for k, v in sign.parameters.items()
            if k != "self" and v.kind != v.VAR_KEYWORD and v.kind != v.VAR_POSITIONAL  # TODO: Handle these cases
        }
    )
    # Resolve inputs to definitions.
    for k, v in all_inputs.items():
        # Get value type from annotation
        value_type = resolve_annotation(v.annotation)
        if value_type is PromptTemplate:
            # custom llm tool has prompt template as input, skip it
            prompt_template_count += 1
            continue
        input_def, is_connection = param_to_definition(v, value_type)
        input_defs[k] = input_def
        if is_connection:
            connection_types.append(input_def.type)

    # Check PromptTemplate input:
    # a. For custom llm tool, there should be exactly one PromptTemplate input
    # b. For python tool, PromptTemplate input is not supported
    if tool_type == ToolType.PYTHON and prompt_template_count > 0:
        raise Exception(f"Input of type 'PromptTemplate' not supported in python tool '{f.__name__}'. ")

    if tool_type == ToolType.CUSTOM_LLM and prompt_template_count == 0:
        raise Exception(f"No input of type 'PromptTemplate' was found in custom llm tool '{f.__name__}'. ")

    if tool_type == ToolType.CUSTOM_LLM and prompt_template_count > 1:
        raise Exception(f"Multiple inputs of type 'PromptTemplate' were found in '{f.__name__}'. "
                        "Only one input of this type is expected.")

    outputs = {}
    # Note: We don't have output definition now
    # outputs = {"output": OutputDefinition("output", [ValueType.from_type(type(sign.return_annotation))], "", True)}
    # if is_dataclass(sign.return_annotation):
    #     for f in fields(sign.return_annotation):
    #         outputs[f.name] = OutputDefinition(f.name, [ValueType.from_type(
    #             type(getattr(sign.return_annotation, f.name)))], "", False)
    return input_defs, outputs, connection_types