Python: Fix: Implement __deepcopy__ on KernelFunction to handle non-serializable OTEL metrics (#12803)

### Description

This PR fixes TypeError: cannot pickle '_thread.RLock' object which
occurs when kernel.clone() is called while OpenTelemetry metrics are
enabled.

-   Fixes: #12802

### Motivation and Context

When using features like `HandoffOrchestration`, the agent's `Kernel` is
cloned. This process uses `copy.deepcopy`, which fails on the
`KernelFunction`'s OpenTelemetry `Histogram` fields
(`invocation_duration_histogram` and `streaming_duration_histogram`).
These histogram objects contain thread locks, making them
non-serializable.

Previous solutions, like the one in **PR #9292**, used
`Field(exclude=True, default_factory=...)`. While that was a necessary
step, it was insufficient because `deepcopy` does not respect the
`exclude` flag in the same way that Pydantic's serialization does. This
bug blocks the use of certain agentic features in any application that
has metrics-based observability enabled.

### Description of the change

This solution introduces a custom __deepcopy__ method to the
KernelFunction Pydantic model. This method correctly handles the
deep-copying process for KernelFunction instances:

1.  It intercepts the deepcopy call for KernelFunction objects.
2. It uses self.model_copy(deep=False) to create a new instance of the
Pydantic model. Using deep=False allows the default_factory for the
histogram fields to create new, clean instances on the copied object,
while other fields are shallow-copied.
3. It then manually deep-copies all other attributes from the original
object to the new one, ensuring that any other mutable objects (like
metadata`) are correctly duplicated. The histogram fields are explicitly
skipped in this step, as they have already been recreated by
`model_copy.
4. Standard deepcopy memoization is used to prevent infinite recursion
in the case of cyclic object graphs.

This approach ensures that the non-serializable fields are
re-initialized instead of copied, resolving the TypeError while
maintaining the integrity of the cloned object.

### How has this been tested?

The fix was validated using the reproduction steps outlined in the
associated issue. Running an agent orchestration that triggers
kernel.clone() with OpenTelemetry metrics enabled now completes
successfully without any errors.

### Contribution Checklist

<!-- Before submitting this PR, please make sure: -->

- [X] The code builds clean without any errors or warnings
- [X] The PR follows the [SK Contribution
Guidelines](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md)
and the [pre-submission formatting
script](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md#development-scripts)
raises no violations
- [X] All unit tests pass, and I have added new tests where possible
- [X] I didn't break anyone 😄

---------

Co-authored-by: Evan Mattson <35585003+moonbox3@users.noreply.github.com>

Author: MMoshtaghi

python/semantic_kernel/connectors/azure_ai_search.py

@@ -675,7 +675,18 @@ def _parse_attribute_chain(attr_node: ast.Attribute) -> str:
             case ast.Name():
                 raise NotImplementedError("Constants are not supported, make sure to use a value or a attribute.")
             case ast.Constant():
-                return str(node.value) if isinstance(node.value, float | int) else f"'{node.value}'"
+                value = node.value
+                if isinstance(value, str):
+                    return "'" + value.replace("'", "''") + "'"
+                if isinstance(value, bytes):
+                    return "'" + value.decode("utf-8").replace("'", "''") + "'"
+                if isinstance(value, bool):
+                    return str(value).lower()
+                if value is None:
+                    return "null"
+                if isinstance(value, (int, float)):
+                    return str(value)
+                raise VectorStoreOperationException(f"Unsupported constant type: {type(value)}")
         raise NotImplementedError(f"Unsupported AST node: {type(node)}")
 
     @override

python/semantic_kernel/connectors/chroma.py

@@ -422,7 +422,14 @@ def _lambda_parser(self, node: ast.AST) -> dict[str, Any] | str | int | float |
                     )
                 return node.id
             case ast.Constant():
-                return node.value
+                value = node.value
+                if isinstance(value, str):
+                    return value.replace("'", "''")
+                if isinstance(value, bytes):
+                    return value.decode("utf-8").replace("'", "''")
+                if isinstance(value, (int, float, bool)) or value is None:
+                    return value
+                raise VectorStoreOperationException(f"Unsupported constant type: {type(value)}")
         raise NotImplementedError(f"Unsupported AST node: {type(node)}")
 
 

python/semantic_kernel/connectors/sql_server.py

@@ -637,9 +637,11 @@ def parse(node: ast.AST) -> str:
                         )
                     return f"[{node.id}]"
                 case ast.Constant():
-                    # Always use parameterization for constants
-                    command.add_parameter(node.value)
-                    return "?"
+                    value = node.value
+                    if isinstance(value, (str, int, float, bool, bytes)) or value is None:
+                        command.add_parameter(str(value))
+                        return "?"
+                    raise VectorStoreOperationException(f"Unsupported constant type: {type(value)}")
                 case ast.List():
                     # For IN/NOT IN lists, parameterize each element
                     placeholders = []

python/semantic_kernel/connectors/weaviate.py

@@ -469,7 +469,14 @@ def _lambda_parser(self, node: ast.AST) -> "_Filters | FilterValues":
                     )
                 return node.id
             case ast.Constant():
-                return node.value
+                value = node.value
+                if isinstance(value, str):
+                    return value.replace("'", "''")
+                if isinstance(value, bytes):
+                    return value.decode("utf-8").replace("'", "''")
+                if isinstance(value, (int, float, bool)) or value is None:
+                    return value
+                raise VectorStoreOperationException(f"Unsupported constant type: {type(value)}")
         raise NotImplementedError(f"Unsupported AST node: {type(node)}")
 
     async def _inner_vectorized_search(

python/semantic_kernel/functions/kernel_function.py

@@ -100,6 +100,25 @@ class KernelFunction(KernelBaseModel):
         default_factory=_create_function_streaming_duration_histogram, exclude=True
     )
 
+    def __deepcopy__(self, memo: dict[int, Any] | None = None) -> "KernelFunction":
+        """Create a deep copy of the kernel function, recreating uncopyable fields."""
+        if memo is None:
+            memo = {}
+        if id(self) in memo:
+            return memo[id(self)]
+
+        # Use model_copy to create a shallow copy of the pydantic model
+        # this is the recommended way to copy pydantic models
+        new_obj = self.model_copy(deep=False)
+        memo[id(self)] = new_obj
+
+        # now deepcopy the fields that are not the histograms
+        for key, value in self.__dict__.items():
+            if key not in ("invocation_duration_histogram", "streaming_duration_histogram"):
+                setattr(new_obj, key, deepcopy(value, memo))
+
+        return new_obj
+
     @classmethod
     def from_prompt(
         cls,

python/tests/unit/functions/test_kernel_function_from_prompt.py

@@ -2,6 +2,7 @@
 
 import os
 import tempfile
+from copy import deepcopy
 from unittest.mock import patch
 
 import pytest
@@ -626,3 +627,22 @@ def test_from_directory_backward_compatibility():
         function = KernelFunctionFromPrompt.from_directory(temp_dir)
         assert function.description == "Basic function"
         assert function.prompt_template.prompt_template_config.template == "Basic ASCII content: {{$input}}"
+
+
+def test_kernel_function_from_prompt_deepcopy():
+    """Test deepcopying a KernelFunctionFromPrompt."""
+    function = KernelFunctionFromPrompt(
+        function_name="test_function",
+        plugin_name="test_plugin",
+        prompt="Hello, world!",
+        description="A test function.",
+    )
+    copied_function = deepcopy(function)
+    assert copied_function is not function
+    assert copied_function.name == function.name
+    assert copied_function.plugin_name == function.plugin_name
+    assert copied_function.description == function.description
+    assert copied_function.prompt_template.prompt_template_config.template == (
+        function.prompt_template.prompt_template_config.template
+    )
+    assert copied_function.prompt_template is not function.prompt_template