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
66 changes: 56 additions & 10 deletions python/semantic_kernel/schema/kernel_json_schema_builder.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,18 +36,28 @@ class KernelJsonSchemaBuilder:

@classmethod
def build(
cls, parameter_type: type | str | Any, description: str | None = None, structured_output: bool = False
cls,
parameter_type: type | str | Any,
description: str | None = None,
structured_output: bool = False,
globalns: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Builds the JSON schema for a given parameter type and description.

Args:
parameter_type: The parameter type.
description: The description of the parameter. Defaults to None.
structured_output: Whether the outputs are structured. Defaults to False.
globalns: The module globals to resolve string forward references
against. Defaults to None.

Returns:
dict[str, Any]: The JSON schema for the parameter type.
"""
if globalns is None:
# No owning model to provide globals (direct call): resolve string
# forward references against the caller's module globals instead.
globalns = sys._getframe(1).f_globals
if isinstance(parameter_type, str):
return cls.build_from_type_name(parameter_type, description)
if isinstance(parameter_type, KernelBaseModel):
Expand All @@ -57,7 +67,7 @@ def build(
if hasattr(parameter_type, "__annotations__"):
return cls.build_model_schema(parameter_type, description, structured_output)
if hasattr(parameter_type, "__args__"):
return cls.handle_complex_type(parameter_type, description, structured_output)
return cls.handle_complex_type(parameter_type, description, structured_output, globalns)
schema = cls.get_json_schema(parameter_type)
if description:
schema["description"] = description
Expand Down Expand Up @@ -97,7 +107,7 @@ def build_model_schema(
field_description = field_info.description
if not cls._is_optional(field_type):
required.append(field_name)
properties[field_name] = cls.build(field_type, field_description, structured_output)
properties[field_name] = cls.build(field_type, field_description, structured_output, model_module_globals)

schema = {"type": "object", "properties": properties}
if required:
Expand Down Expand Up @@ -150,16 +160,37 @@ def get_json_schema(cls, parameter_type: type) -> dict[str, Any]:
type_name = TYPE_MAPPING.get(parameter_type, "object")
return {"type": type_name}

@classmethod
def _resolve_forward_ref(cls, arg: Any, globalns: dict[str, Any] | None) -> Any:
"""Resolve a string argument against the module globals if it names a type.

Generic aliases like ``list["Inner"]`` store the raw string in
``__args__`` without a ``ForwardRef`` wrapper, so ``get_type_hints``
never evaluates it. Resolve it here against the owning model's module
globals so ``list["Inner"]`` produces the same schema as ``list[Inner]``.
"""
if isinstance(arg, str) and globalns is not None:
resolved = globalns.get(arg)
if isinstance(resolved, type):
return resolved
return arg

@classmethod
def handle_complex_type(
cls, parameter_type: type, description: str | None = None, structured_output: bool = False
cls,
parameter_type: type,
description: str | None = None,
structured_output: bool = False,
globalns: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Handles building the JSON schema for complex types.

Args:
parameter_type: The parameter type.
description: The description of the parameter. Defaults to None.
structured_output: Whether the outputs are structured. Defaults to False.
globalns: The module globals to resolve string forward references
against. Defaults to None.

Returns:
dict[str, Any]: The JSON schema for the parameter type.
Expand All @@ -169,17 +200,18 @@ def handle_complex_type(

schema: dict[str, Any] = {}
if origin is list or origin is set:
item_type = args[0]
item_type = cls._resolve_forward_ref(args[0], globalns)
schema = {
"type": "array",
"items": cls.build(item_type, structured_output=structured_output),
"items": cls.build(item_type, structured_output=structured_output, globalns=globalns),
}
if description:
schema["description"] = description
return schema
if origin is dict:
_, value_type = args
additional_properties = cls.build(value_type, structured_output=structured_output)
value_type = cls._resolve_forward_ref(value_type, globalns)
additional_properties = cls.build(value_type, structured_output=structured_output, globalns=globalns)
if additional_properties == {"type": "object"}:
additional_properties["properties"] = {} # Account for differences in Python 3.10 dict
schema = {"type": "object", "additionalProperties": additional_properties}
Expand All @@ -189,7 +221,12 @@ def handle_complex_type(
schema["additionalProperties"] = False
return schema
if origin is tuple:
items = [cls.build(arg, structured_output=structured_output) for arg in args]
items = [
cls.build(
cls._resolve_forward_ref(arg, globalns), structured_output=structured_output, globalns=globalns
)
Comment on lines 221 to +227

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The tuple branch does not handle the homogeneous-tuple marker (tuple["Inner", ...]). _resolve_forward_ref only transforms strings, so Ellipsis passes through unchanged and is treated as a real tuple element. The resulting schema is a two-item tuple rather than an array of repeated ForwardRefInnerModels. The Ellipsis sentinel needs to be detected before iterating over args.

Suggested change
schema["additionalProperties"] = False
return schema
if origin is tuple:
items = [cls.build(arg, structured_output=structured_output) for arg in args]
items = [
cls.build(
cls._resolve_forward_ref(arg, globalns), structured_output=structured_output, globalns=globalns
)
if origin is tuple:
if len(args) == 2 and args[1] is Ellipsis:
item_type = cls._resolve_forward_ref(args[0], globalns)
schema = {
"type": "array",
"items": cls.build(item_type, structured_output=structured_output, globalns=globalns),
}
else:
items = [
cls.build(
cls._resolve_forward_ref(arg, globalns), structured_output=structured_output, globalns=globalns
)
for arg in args
]
schema = {"type": "array", "items": items}

for arg in args
]
schema = {"type": "array", "items": items}
if description:
schema["description"] = description
Expand All @@ -200,14 +237,23 @@ def handle_complex_type(
# Handle Optional[T] (Union[T, None]) by making schema nullable
if len(args) == 2 and type(None) in args:
non_none_type = args[0] if args[1] is type(None) else args[1]
schema = cls.build(non_none_type, structured_output=structured_output)
non_none_type = cls._resolve_forward_ref(non_none_type, globalns)
schema = cls.build(non_none_type, structured_output=structured_output, globalns=globalns)
schema["type"] = [schema["type"], "null"]
if description:
schema["description"] = description
if structured_output:
schema["additionalProperties"] = False
return schema
schemas = [cls.build(arg, description, structured_output=structured_output) for arg in args]
schemas = [
cls.build(
cls._resolve_forward_ref(arg, globalns),
description,
structured_output=structured_output,
globalns=globalns,
)
for arg in args
]
return {"anyOf": schemas}
schema = cls.get_json_schema(parameter_type)
if description:
Expand Down
106 changes: 106 additions & 0 deletions python/tests/unit/schema/test_schema_builder.py
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,27 @@ class NonPydanticReasoning:
final_answer: str


class ForwardRefInnerModel(KernelBaseModel):
value: int
label: str


class ModelWithListForwardRef(KernelBaseModel):
items: list["ForwardRefInnerModel"] = []


class ModelWithNestedListForwardRef(KernelBaseModel):
matrix: list[list["ForwardRefInnerModel"]]


class ModelWithDictForwardRef(KernelBaseModel):
mapping: dict[str, "ForwardRefInnerModel"]


class ModelWithTopLevelForwardRef(KernelBaseModel):
one: "ForwardRefInnerModel"


def test_build_with_kernel_base_model():
expected_schema = {
"type": "object",
Expand Down Expand Up @@ -455,3 +476,88 @@ def test_build_schema_with_nonpydantic_structured_output():
}

assert structured_output_schema == expected_schema


def test_build_with_list_string_forward_reference():
schema = KernelJsonSchemaBuilder.build(ModelWithListForwardRef)
items = schema["properties"]["items"]
assert items == {
"type": "array",
"items": {
"type": "object",
"properties": {"value": {"type": "integer"}, "label": {"type": "string"}},
"required": ["value", "label"],
},
}


def test_build_with_nested_list_string_forward_reference():
schema = KernelJsonSchemaBuilder.build(ModelWithNestedListForwardRef)
matrix = schema["properties"]["matrix"]
inner = matrix["items"]["items"]
assert matrix["type"] == "array"
assert inner["type"] == "object"
assert inner["properties"]["value"]["type"] == "integer"
assert inner["properties"]["label"]["type"] == "string"
assert inner["required"] == ["value", "label"]


def test_build_with_dict_string_forward_reference():
schema = KernelJsonSchemaBuilder.build(ModelWithDictForwardRef)
additional_properties = schema["properties"]["mapping"]["additionalProperties"]
assert additional_properties["type"] == "object"
assert additional_properties["properties"]["value"]["type"] == "integer"
assert additional_properties["properties"]["label"]["type"] == "string"
assert additional_properties["required"] == ["value", "label"]


def test_build_with_top_level_string_forward_reference():
schema = KernelJsonSchemaBuilder.build(ModelWithTopLevelForwardRef)
one = schema["properties"]["one"]
assert one["type"] == "object"
assert one["properties"]["value"]["type"] == "integer"
assert one["required"] == ["value", "label"]


def test_build_list_alias_with_string_forward_reference():
# A generic alias built with a raw string arg keeps the string in __args__
# (no ForwardRef wrapper), so resolution has to happen against the module
# globals of the model that owns the annotation.
schema = KernelJsonSchemaBuilder.build(list["ForwardRefInnerModel"])
assert schema == {
"type": "array",
"items": {
"type": "object",
"properties": {"value": {"type": "integer"}, "label": {"type": "string"}},
"required": ["value", "label"],
},
}


def test_build_nested_list_alias_with_string_forward_reference():
schema = KernelJsonSchemaBuilder.build(list[list["ForwardRefInnerModel"]])
inner = schema["items"]["items"]
assert schema["type"] == "array"
assert inner["type"] == "object"
assert inner["properties"]["value"]["type"] == "integer"
assert inner["properties"]["label"]["type"] == "string"
assert inner["required"] == ["value", "label"]


def test_build_dict_alias_with_string_forward_reference():
schema = KernelJsonSchemaBuilder.build(dict[str, "ForwardRefInnerModel"])
additional_properties = schema["additionalProperties"]
assert additional_properties["type"] == "object"
assert additional_properties["properties"]["value"]["type"] == "integer"
assert additional_properties["properties"]["label"]["type"] == "string"
assert additional_properties["required"] == ["value", "label"]


def test_build_tuple_alias_with_string_forward_reference():
schema = KernelJsonSchemaBuilder.build(tuple["ForwardRefInnerModel", "ForwardRefInnerModel"])
assert schema["type"] == "array"
for item in schema["items"]:
assert item["type"] == "object"
assert item["properties"]["value"]["type"] == "integer"
assert item["properties"]["label"]["type"] == "string"
assert item["required"] == ["value", "label"]
Loading