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
41 changes: 37 additions & 4 deletions python/semantic_kernel/functions/kernel_arguments.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
# Copyright (c) Microsoft. All rights reserved.

import json
import logging
from typing import TYPE_CHECKING, Any

from pydantic import BaseModel
Expand All @@ -14,6 +15,8 @@

from semantic_kernel.connectors.ai.prompt_execution_settings import PromptExecutionSettings

logger: logging.Logger = logging.getLogger(__name__)


class KernelArguments(dict):
"""The arguments sent to the KernelFunction."""
Expand Down Expand Up @@ -108,15 +111,45 @@ def __ior__(self, value: "SupportsKeysAndGetItem[Any, Any] | Iterable[tuple[Any,
return self

def dumps(self, include_execution_settings: bool = False) -> str:
"""Serializes the KernelArguments to a JSON string."""
"""Serializes the KernelArguments to a JSON string.

Handles arguments that contain objects with circular references (e.g.,
KernelProcessStepContext) by falling back to their string representation.
"""
data = dict(self)
if include_execution_settings and self.execution_settings:
data["execution_settings"] = self.execution_settings

def default(obj):
seen: set[int] = set()

def default(obj: Any) -> Any:
obj_id = id(obj)
if obj_id in seen:
return f"<circular ref: {type(obj).__name__}>"
seen.add(obj_id)

if isinstance(obj, BaseModel):
return obj.model_dump()
try:
return obj.model_dump()
except Exception:
return f"<{type(obj).__name__}>"

return str(obj)

return json.dumps(data, default=default)
try:
return json.dumps(data, default=default)
except ValueError:
# Catch circular reference errors that json.dumps raises when
# model_dump() produces dicts with internal circular references.
# This can happen with complex runtime objects like
# KernelProcessStepContext whose step_message_channel holds
# back-references to the process graph.
logger.debug("Circular reference detected while serializing KernelArguments, using string fallback.")
safe_data: dict[str, Any] = {}
for key, value in data.items():
try:
json.dumps(value, default=default)
safe_data[key] = value
except (ValueError, TypeError):
safe_data[key] = f"<{type(value).__name__}>"
return json.dumps(safe_data, default=default)
44 changes: 44 additions & 0 deletions python/tests/unit/functions/test_kernel_arguments.py
Original file line number Diff line number Diff line change
Expand Up @@ -179,3 +179,47 @@ def test_kernel_arguments_ror_operator_with_invalid_type(lhs):
"""Test the __ror__ operator with an invalid type raises TypeError."""
with pytest.raises(TypeError):
lhs | KernelArguments()


def test_kernel_arguments_dumps_basic():
"""Test basic dumps serialization."""
kargs = KernelArguments(name="test", value=42)
result = kargs.dumps()
import json

parsed = json.loads(result)
assert parsed == {"name": "test", "value": 42}


def test_kernel_arguments_dumps_with_pydantic_model():
"""Test dumps serialization with a Pydantic model argument."""
from pydantic import BaseModel

class SimpleModel(BaseModel):
field: str = "hello"

kargs = KernelArguments(model=SimpleModel())
result = kargs.dumps()
import json

parsed = json.loads(result)
assert parsed == {"model": {"field": "hello"}}


def test_kernel_arguments_dumps_with_circular_reference():
"""Test dumps handles arguments with circular references gracefully.

This reproduces the bug from issue #13393 where KernelProcessStepContext
(which contains a step_message_channel that references back to the process
graph) caused 'Circular reference detected' errors during OTel diagnostics.
"""
# Create a dict with a circular reference to simulate what happens
# when model_dump() produces circular structures
circular: dict = {"key": "value"}
circular["self"] = circular

kargs = KernelArguments(data=circular)
# This should not raise ValueError: Circular reference detected
result = kargs.dumps()
assert isinstance(result, str)
assert "data" in result
Loading