diff --git a/python/semantic_kernel/functions/kernel_function_decorator.py b/python/semantic_kernel/functions/kernel_function_decorator.py index 2e597099b9e8..108a614b0cdd 100644 --- a/python/semantic_kernel/functions/kernel_function_decorator.py +++ b/python/semantic_kernel/functions/kernel_function_decorator.py @@ -6,6 +6,8 @@ from inspect import Parameter, Signature, isasyncgenfunction, isclass, isgeneratorfunction, signature from typing import Annotated, Any, ForwardRef, Union, get_args, get_origin +from semantic_kernel.schema.kernel_json_schema_builder import KernelJsonSchemaBuilder + NoneType = type(None) logger = logging.getLogger(__name__) @@ -65,7 +67,7 @@ def decorator(func: Callable[..., object]) -> Callable[..., object]: logger.debug(f"Parsing decorator for function: {getattr(func, '__kernel_function_name__')}") func_sig = signature(func, eval_str=True) - annotations = _process_signature(func_sig) + annotations = _process_signature(func_sig, getattr(func, "__globals__", None)) logger.debug(f"{annotations=}") setattr(func, "__kernel_function_parameters__", annotations) @@ -114,8 +116,15 @@ def _get_underlying_type(annotation: Any) -> Any: return annotation -def _process_signature(func_sig: Signature) -> list[dict[str, Any]]: - """Process the signature of the function.""" +def _process_signature(func_sig: Signature, globalns: dict[str, Any] | None = None) -> list[dict[str, Any]]: + """Process the signature of the function. + + Args: + func_sig: The signature of the function. + globalns: The function's module globals, used to resolve forward references such as + ``list["Model"]`` in parameter annotations. ``signature(..., eval_str=True)`` evaluates an + annotation that is a string, but not a string nested inside a generic alias. + """ annotations = [] for arg in func_sig.parameters.values(): @@ -129,6 +138,8 @@ def _process_signature(func_sig: Signature) -> list[dict[str, Any]]: underlying_type = _get_underlying_type(annotation) else: underlying_type = annotation + if globalns is not None: + underlying_type = KernelJsonSchemaBuilder.resolve_forward_refs(underlying_type, globalns) parsed_annotation["type_object"] = underlying_type annotations.append(parsed_annotation) diff --git a/python/semantic_kernel/schema/kernel_json_schema_builder.py b/python/semantic_kernel/schema/kernel_json_schema_builder.py index aaaeed0f390d..e1548eac8e4e 100644 --- a/python/semantic_kernel/schema/kernel_json_schema_builder.py +++ b/python/semantic_kernel/schema/kernel_json_schema_builder.py @@ -3,7 +3,7 @@ import sys import types from enum import Enum -from typing import Any, Literal, Union, get_args, get_origin, get_type_hints +from typing import Annotated, Any, ForwardRef, Literal, Union, get_args, get_origin, get_type_hints from semantic_kernel.const import PARSED_ANNOTATION_UNION_DELIMITER from semantic_kernel.exceptions.function_exceptions import FunctionInvalidParameterConfiguration @@ -86,6 +86,7 @@ def build_model_schema( hints = get_type_hints(model, globalns=model_module_globals, localns={}) for field_name, field_type in hints.items(): + field_type = cls.resolve_forward_refs(field_type, model_module_globals) field_description = None if hasattr(model, "model_fields") and field_name in model.model_fields: field_info = model.model_fields[field_name] @@ -109,6 +110,88 @@ def build_model_schema( return schema + @classmethod + def resolve_forward_refs(cls, annotation: Any, globalns: dict[str, Any]) -> Any: + """Resolves forward references sitting inside a generic alias. + + On Python 3.10, ``get_type_hints`` evaluates an annotation that *is* a string and + ``ForwardRef`` objects nested inside a generic alias, but not a bare ``str`` sitting in a + PEP 585 alias's ``__args__``. ``list["Inner"]`` goes through ``list.__class_getitem__``, + which stores ``"Inner"`` verbatim with no ``ForwardRef`` wrapper, so nothing resolves it and + the element type falls through to the ``{"type": "object"}`` placeholder. + ``typing.Optional["Inner"]`` does wrap the string, which is why the top-level and + ``Optional`` forms already work. + + Python 3.11 made ``get_type_hints`` wrap those strings itself (gh-85542), so on 3.11+ the + hints arrive fully resolved and this is a no-op that returns the annotation unchanged. + + Args: + annotation: The annotation to resolve. + globalns: The namespace of the module owning the annotation. + + Returns: + Any: The annotation, rebuilt with resolvable string arguments replaced by their types. + """ + origin = get_origin(annotation) + args = get_args(annotation) + if origin is None or not args: + return annotation + if origin is Literal: + # Literal members are values, not names; a Literal["Inner"] must stay a string. + return annotation + if origin is Annotated: + # Only the annotated type is a type; the metadata that follows it is left alone. + return Annotated[(cls._resolve_forward_ref_arg(args[0], globalns), *args[1:])] + + resolved_args = tuple(cls._resolve_forward_ref_arg(arg, globalns) for arg in args) + if resolved_args == args: + return annotation + + try: + return origin[resolved_args] + except TypeError: + # Not every alias can be rebuilt from its origin and args; keep the original. + return annotation + + @classmethod + def _resolve_forward_ref_arg(cls, arg: Any, globalns: dict[str, Any]) -> Any: + """Resolves a single generic argument, recursing into nested generic aliases. + + Args: + arg: The generic argument to resolve. + globalns: The namespace to resolve the reference against. + + Returns: + Any: The resolved type, or the argument unchanged when it cannot be resolved. + """ + if isinstance(arg, str): + return cls._lookup_type(arg, globalns, default=arg) + if isinstance(arg, ForwardRef): + return cls._lookup_type(arg.__forward_arg__, globalns, default=arg) + return cls.resolve_forward_refs(arg, globalns) + + @staticmethod + def _lookup_type(name: str, globalns: dict[str, Any], default: Any) -> Any: + """Looks a name up in the namespace, substituting it only when it names a type. + + A forward reference that happens to share its name with a module, a function or a plain + value must not be rebuilt around that object: a module reaches ``build_model_schema`` and + fails on ``__module__``, and a function or value would produce a misleading schema. Keeping + the original reference lets the existing placeholder fallback apply instead. + + Args: + name: The referenced name. + globalns: The namespace to look it up in. + default: What to return when the name is missing or does not name a type. + + Returns: + Any: The class or typing construct the name refers to, or ``default``. + """ + value = globalns.get(name, default) + if isinstance(value, type) or get_origin(value) is not None: + return value + return default + @classmethod def _is_optional(cls, field_type: Any) -> bool: return get_origin(field_type) in {types.UnionType, Union} and type(None) in get_args(field_type) diff --git a/python/tests/unit/functions/test_kernel_function_decorators.py b/python/tests/unit/functions/test_kernel_function_decorators.py index 048ea903200e..70faaf87c4d9 100644 --- a/python/tests/unit/functions/test_kernel_function_decorators.py +++ b/python/tests/unit/functions/test_kernel_function_decorators.py @@ -38,6 +38,14 @@ def func_docstring_as_description(self, input): """Description.""" return input + @kernel_function + def func_forward_ref_list(self, items: list["InputObject"]): + return items + + @kernel_function + def func_forward_ref_colliding_with_module(self, items: list["pytest"]): + return items + @kernel_function def func_input_annotated(self, input: Annotated[str, "input description"]): return input @@ -306,3 +314,61 @@ def test_annotation_parsing(name, annotation, description, type_, is_required): assert description == annotation_dict.get("description") assert type_ == annotation_dict["type_"] assert is_required == annotation_dict["is_required"] + + +def test_kernel_function_resolves_forward_ref_in_list_parameter(): + """list["InputObject"] on a kernel function parameter resolves to list[InputObject].""" + my_func = getattr(MiscClass(), "func_forward_ref_list") + param = my_func.__kernel_function_parameters__[0] + + assert param["name"] == "items" + assert param["type_"] == "list[InputObject]" + assert param["type_object"] == list[InputObject] + + +@pytest.mark.parametrize( + ("annotation", "expected_type_object", "expected_description"), + [ + (list["InputObject"], list[InputObject], None), + (dict[str, "InputObject"], dict[str, InputObject], None), + (Annotated[list["InputObject"], "description"], list[InputObject], "description"), + (list["InputObject"] | None, list[InputObject], None), + ], +) +def test_process_signature_resolves_forward_refs_with_globalns(annotation, expected_type_object, expected_description): + param = Parameter( + name="items", + annotation=annotation, + default=Parameter.empty, + kind=Parameter.POSITIONAL_OR_KEYWORD, + ) + func_sig = Signature(parameters=[param]) + + annotations = _process_signature(func_sig, globals()) + + assert annotations[0]["type_object"] == expected_type_object + assert annotations[0].get("description") == expected_description + + +def test_process_signature_without_globalns_leaves_forward_refs_unchanged(): + """Callers that pass no namespace get the previous behaviour.""" + param = Parameter( + name="items", + annotation=list["InputObject"], + default=Parameter.empty, + kind=Parameter.POSITIONAL_OR_KEYWORD, + ) + func_sig = Signature(parameters=[param]) + + annotations = _process_signature(func_sig) + + assert annotations[0]["type_object"] == list["InputObject"] + + +def test_kernel_function_forward_ref_that_names_a_module_is_left_unresolved(): + """A reference colliding with a non type global must not be substituted or raise at decoration.""" + my_func = getattr(MiscClass(), "func_forward_ref_colliding_with_module") + param = my_func.__kernel_function_parameters__[0] + + assert param["type_"] == "list[pytest]" + assert param["type_object"] == list["pytest"] diff --git a/python/tests/unit/schema/test_schema_builder.py b/python/tests/unit/schema/test_schema_builder.py index 22e725ac2045..dd64c9082185 100644 --- a/python/tests/unit/schema/test_schema_builder.py +++ b/python/tests/unit/schema/test_schema_builder.py @@ -1,6 +1,7 @@ # Copyright (c) Microsoft. All rights reserved. import json +import sys from enum import Enum from typing import Annotated, Any, Literal, Optional, Union from unittest.mock import Mock @@ -499,3 +500,123 @@ def test_build_model_with_literal_fields(): }, "required": ["level"], } + + +class ForwardRefTarget(KernelBaseModel): + value: int + label: str + + +class DirectListHolder(KernelBaseModel): + items: list[ForwardRefTarget] = [] + + +class ForwardRefListHolder(KernelBaseModel): + items: list["ForwardRefTarget"] = [] + + +class ForwardRefDictHolder(KernelBaseModel): + mapping: dict[str, "ForwardRefTarget"] = {} + + +class ForwardRefNestedHolder(KernelBaseModel): + items: list[list["ForwardRefTarget"]] = [] + + +class ForwardRefOptionalHolder(KernelBaseModel): + items: list[Optional["ForwardRefTarget"]] = [] + + +class UnresolvableForwardRefHolder(KernelBaseModel): + items: list["NotDefinedAnywhere"] = [] # noqa: F821 + + +def test_build_model_schema_resolves_forward_ref_in_list(): + """list["Target"] must produce the same schema as list[Target].""" + forward_ref = KernelJsonSchemaBuilder.build(ForwardRefListHolder)["properties"]["items"] + direct = KernelJsonSchemaBuilder.build(DirectListHolder)["properties"]["items"] + + assert forward_ref == direct + assert forward_ref["items"]["properties"] == {"value": {"type": "integer"}, "label": {"type": "string"}} + assert forward_ref["items"]["required"] == ["value", "label"] + + +def test_build_model_schema_resolves_forward_ref_in_dict(): + """dict[str, "Target"] must carry the element schema through additionalProperties.""" + schema = KernelJsonSchemaBuilder.build(ForwardRefDictHolder)["properties"]["mapping"] + + assert schema["additionalProperties"]["properties"] == { + "value": {"type": "integer"}, + "label": {"type": "string"}, + } + + +def test_build_model_schema_resolves_nested_forward_ref(): + """Resolution recurses into nested generic aliases.""" + schema = KernelJsonSchemaBuilder.build(ForwardRefNestedHolder)["properties"]["items"] + + assert schema["items"]["items"]["properties"] == { + "value": {"type": "integer"}, + "label": {"type": "string"}, + } + + +def test_build_model_schema_resolves_forward_ref_inside_optional(): + """A ForwardRef nested under Optional inside a list still resolves.""" + schema = KernelJsonSchemaBuilder.build(ForwardRefOptionalHolder)["properties"]["items"] + + assert schema["items"]["type"] == ["object", "null"] + assert schema["items"]["properties"] == {"value": {"type": "integer"}, "label": {"type": "string"}} + + +@pytest.mark.skipif( + sys.version_info >= (3, 11), + reason="On 3.11+ get_type_hints raises NameError for the unresolvable name before the builder sees it", +) +def test_build_model_schema_unresolvable_forward_ref_falls_back(): + """An unresolvable name keeps the previous placeholder rather than raising.""" + schema = KernelJsonSchemaBuilder.build(UnresolvableForwardRefHolder)["properties"]["items"] + + assert schema == {"type": "array", "items": {"type": "object"}} + + +def test_resolve_forward_refs_leaves_literal_values_and_annotated_metadata_alone(): + """Strings that are values, not names, must not be looked up even when a global matches.""" + namespace = {"ForwardRefTarget": ForwardRefTarget} + + literal = KernelJsonSchemaBuilder.resolve_forward_refs(list[Literal["ForwardRefTarget"]], namespace) + annotated = KernelJsonSchemaBuilder.resolve_forward_refs( + list[Annotated["ForwardRefTarget", "ForwardRefTarget"]], namespace + ) + + assert literal == list[Literal["ForwardRefTarget"]] + assert annotated == list[Annotated[ForwardRefTarget, "ForwardRefTarget"]] + + +def _not_a_type(): + return None + + +NOT_A_TYPE_VALUE = 3 + + +def test_resolve_forward_refs_ignores_names_that_do_not_refer_to_a_type(): + """A reference that collides with a module, function or value keeps the placeholder fallback.""" + namespace = { + "json": json, + "_not_a_type": _not_a_type, + "NOT_A_TYPE_VALUE": NOT_A_TYPE_VALUE, + "ForwardRefTarget": ForwardRefTarget, + } + + assert KernelJsonSchemaBuilder.resolve_forward_refs(list["json"], namespace) == list["json"] + assert KernelJsonSchemaBuilder.resolve_forward_refs(list["_not_a_type"], namespace) == list["_not_a_type"] + assert ( + KernelJsonSchemaBuilder.resolve_forward_refs(dict[str, "NOT_A_TYPE_VALUE"], namespace) + == (dict[str, "NOT_A_TYPE_VALUE"]) + ) + # a sibling argument that does name a type is still resolved + assert ( + KernelJsonSchemaBuilder.resolve_forward_refs(dict["json", "ForwardRefTarget"], namespace) + == (dict["json", ForwardRefTarget]) + )