Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 14 additions & 3 deletions python/semantic_kernel/functions/kernel_function_decorator.py
Original file line number Diff line number Diff line change
Expand Up @@ -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__)

Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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():
Expand All @@ -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)

Expand Down
85 changes: 84 additions & 1 deletion python/semantic_kernel/schema/kernel_json_schema_builder.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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]
Expand All @@ -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)
Expand Down
66 changes: 66 additions & 0 deletions python/tests/unit/functions/test_kernel_function_decorators.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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"]
121 changes: 121 additions & 0 deletions python/tests/unit/schema/test_schema_builder.py
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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])
)
Loading