Python: Avoid logging function argument values

python/semantic_kernel/functions/kernel_function.py

@@ -401,11 +401,15 @@ def _handle_exception(self, current_span: trace.Span, exception: Exception, attr
             exception (Exception): The exception.
             attributes (Attributes): The attributes to be modified.
         """
-        attributes[ERROR_TYPE] = type(exception).__name__
-
-        current_span.record_exception(exception)
-        current_span.set_attribute(ERROR_TYPE, type(exception).__name__)
-        current_span.set_status(trace.StatusCode.ERROR, description=str(exception))
+        error_type = type(exception).__name__
+        attributes[ERROR_TYPE] = error_type
+
+        current_span.set_attribute(ERROR_TYPE, error_type)
+        if function_tracer.are_sensitive_events_enabled():
+            current_span.record_exception(exception)
+            current_span.set_status(trace.StatusCode.ERROR, description=str(exception))
+        else:
+            current_span.set_status(trace.StatusCode.ERROR)
 
         KernelFunctionLogMessages.log_function_error(logger, exception)
 

python/semantic_kernel/functions/kernel_function_log_messages.py

@@ -20,8 +20,8 @@ def log_function_invoking(logger: Logger, kernel_function_name: str):
 
     @staticmethod
     def log_function_arguments(logger: Logger, arguments: KernelArguments):
-        """Log message when a kernel function is invoked."""
-        logger.debug("Function arguments: %s", arguments)
+        """Log the number of arguments provided to a kernel function."""
+        logger.debug("Function arguments provided: %d.", len(arguments))
 
     @staticmethod
     def log_function_invoked_success(logger: Logger, kernel_function_name: str):
@@ -30,22 +30,19 @@ def log_function_invoked_success(logger: Logger, kernel_function_name: str):
 
     @staticmethod
     def log_function_result_value(logger: Logger, function_result: FunctionResult | None):
-        """Log message when a kernel function result is returned."""
+        """Log whether a kernel function returned a result."""
         if not logger.isEnabledFor(logging.DEBUG):
             return
 
         if function_result is not None:
-            try:
-                logger.debug("Function result: %s", function_result)
-            except Exception:
-                logger.error("Function result: Failed to convert result value to string")
+            logger.debug("Function result available.")
         else:
             logger.debug("Function result: None")
 
     @staticmethod
     def log_function_error(logger: Logger, error: Exception):
         """Log message when a kernel function fails."""
-        logger.error("Function failed. Error: %s", error)
+        logger.error("Function failed. Error type: %s.", type(error).__name__)
 
     @staticmethod
     def log_function_completed(logger: Logger, duration: float):

python/semantic_kernel/kernel.py

@@ -423,7 +423,7 @@ async def invoke_function_call(
             chat_history.add_message(message=frc.to_chat_message_content())
             return None
 
-        logger.info(f"Calling {function_call.name} function with args: {function_call.arguments}")
+        logger.info("Calling %s function.", function_call.name)
 
         _rebuild_auto_function_invocation_context()
         invocation_context = AutoFunctionInvocationContext(
@@ -475,7 +475,11 @@ async def _inner_auto_function_invoke_handler(self, context: AutoFunctionInvocat
             if result:
                 context.function_result = result
         except Exception as exc:
-            logger.exception(f"Error invoking function {context.function.fully_qualified_name}: {exc}.")
+            logger.error(
+                "Error invoking function %s. Error type: %s.",
+                context.function.fully_qualified_name,
+                type(exc).__name__,
+            )
             value = f"An error occurred while invoking the function {context.function.fully_qualified_name}: {exc}"
             if context.function_result is not None:
                 context.function_result.value = value

python/semantic_kernel/prompt_template/utils/template_function_helpers.py

@@ -96,13 +96,15 @@ def func(*args, **kwargs):
             actual_args = args
 
         if this is not None:
-            logger.debug(f"Handlebars context with `this`: {this}")
+            logger.debug("Handlebars context with `this`.")
         else:
             logger.debug("Jinja2 context or Handlebars context without `this`")
 
         logger.debug(
-            f"Invoking function {function.metadata.fully_qualified_name} "
-            f"with args: {actual_args} and kwargs: {kwargs} and this: {this}."
+            "Invoking function %s with positional argument count %d and keyword argument count %d.",
+            function.metadata.fully_qualified_name,
+            len(actual_args),
+            len(kwargs),
         )
 
         result = asyncio.run(function.invoke(kernel=kernel, arguments=arguments))
@@ -131,7 +133,10 @@ async def func(*args, **kwargs):
         arguments.update(base_arguments)
         arguments.update(kwargs)
         logger.debug(
-            f"Invoking function {function.metadata.fully_qualified_name} with args: {arguments} and kwargs: {kwargs}."
+            "Invoking function %s with positional argument count %d and keyword argument count %d.",
+            function.metadata.fully_qualified_name,
+            len(args),
+            len(kwargs),
         )
         result = await function.invoke(kernel=kernel, arguments=arguments)
         if allow_dangerously_set_content:

python/semantic_kernel/utils/telemetry/model_diagnostics/function_tracer.py

@@ -62,4 +62,9 @@ def start_as_current_span(
     if function.description:
         attributes[TOOL_DESCRIPTION] = function.description
 
-    return tracer.start_as_current_span(f"{OPERATION_NAME} {function.fully_qualified_name}", attributes=attributes)
+    return tracer.start_as_current_span(
+        f"{OPERATION_NAME} {function.fully_qualified_name}",
+        attributes=attributes,
+        record_exception=False,
+        set_status_on_exception=False,
+    )

python/tests/unit/functions/test_kernel_function_from_method.py

@@ -1,9 +1,13 @@
 # Copyright (c) Microsoft. All rights reserved.
+import logging
 from collections.abc import AsyncGenerator, Iterable
+from contextlib import nullcontext
+from importlib import import_module
 from typing import Annotated, Any
-from unittest.mock import Mock
+from unittest.mock import Mock, patch
 
 import pytest
+from opentelemetry.trace import StatusCode
 
 from semantic_kernel.connectors.ai.open_ai.services.open_ai_chat_completion import OpenAIChatCompletion
 from semantic_kernel.exceptions import FunctionExecutionException, FunctionInitializationError
@@ -18,6 +22,8 @@
 from semantic_kernel.kernel import Kernel
 from semantic_kernel.kernel_pydantic import KernelBaseModel
 
+kernel_function_module = import_module("semantic_kernel.functions.kernel_function")
+
 
 class CustomType(KernelBaseModel):
     id: str
@@ -177,6 +183,115 @@ def non_async_function() -> str:
             pass
 
 
+async def test_invoke_logs_argument_count_without_values(kernel: Kernel, caplog):
+    secret = "SYNTHETIC_SECRET_VALUE"
+
+    @kernel_function
+    def echo(value: str) -> str:
+        return value
+
+    native_function = KernelFunction.from_method(method=echo, plugin_name="MockPlugin")
+
+    with caplog.at_level(logging.DEBUG, logger="semantic_kernel.functions.kernel_function"):
+        result = await native_function.invoke(kernel=kernel, arguments=KernelArguments(value=secret))
+
+    messages = [
+        record.getMessage() for record in caplog.records if record.name == "semantic_kernel.functions.kernel_function"
+    ]
+    assert result.value == secret
+    assert secret not in "\n".join(messages)
+    assert "Function arguments provided: 1." in messages
+    assert "Function result available." in messages
+
+
+@pytest.mark.parametrize("streaming", [False, True])
+async def test_invoke_failure_omits_exception_value_without_sensitive_diagnostics(
+    kernel: Kernel, caplog, streaming: bool
+):
+    secret = "SYNTHETIC_SECRET_VALUE"
+
+    if streaming:
+
+        @kernel_function
+        def fail(value: str) -> Iterable[str]:
+            if value:
+                raise ValueError(value)
+            yield ""
+
+    else:
+
+        @kernel_function
+        def fail(value: str) -> str:
+            raise ValueError(value)
+
+    native_function = KernelFunction.from_method(method=fail, plugin_name="MockPlugin")
+    span = Mock()
+
+    with (
+        patch.object(
+            kernel_function_module.tracer,
+            "start_as_current_span",
+            return_value=nullcontext(span),
+        ) as start_span,
+        patch.object(
+            kernel_function_module.function_tracer,
+            "are_sensitive_events_enabled",
+            return_value=False,
+        ),
+        caplog.at_level(logging.ERROR, logger="semantic_kernel.functions.kernel_function"),
+        pytest.raises(ValueError, match=secret),
+    ):
+        if streaming:
+            async for _ in native_function.invoke_stream(kernel=kernel, arguments=KernelArguments(value=secret)):
+                pass
+        else:
+            await native_function.invoke(kernel=kernel, arguments=KernelArguments(value=secret))
+
+    messages = [
+        record.getMessage() for record in caplog.records if record.name == "semantic_kernel.functions.kernel_function"
+    ]
+    assert secret not in "\n".join(messages)
+    assert "Function failed. Error type: ValueError." in messages
+    assert start_span.call_args.kwargs["record_exception"] is False
+    assert start_span.call_args.kwargs["set_status_on_exception"] is False
+    span.record_exception.assert_not_called()
+    span.set_status.assert_called_once_with(StatusCode.ERROR)
+
+
+async def test_invoke_failure_records_exception_with_sensitive_diagnostics(kernel: Kernel, caplog):
+    secret = "SYNTHETIC_SECRET_VALUE"
+
+    @kernel_function
+    def fail(value: str) -> str:
+        raise ValueError(value)
+
+    native_function = KernelFunction.from_method(method=fail, plugin_name="MockPlugin")
+    span = Mock()
+
+    with (
+        patch.object(
+            kernel_function_module.tracer,
+            "start_as_current_span",
+            return_value=nullcontext(span),
+        ),
+        patch.object(
+            kernel_function_module.function_tracer,
+            "are_sensitive_events_enabled",
+            return_value=True,
+        ),
+        caplog.at_level(logging.ERROR, logger="semantic_kernel.functions.kernel_function"),
+        pytest.raises(ValueError, match=secret) as exc_info,
+    ):
+        await native_function.invoke(kernel=kernel, arguments=KernelArguments(value=secret))
+
+    messages = [
+        record.getMessage() for record in caplog.records if record.name == "semantic_kernel.functions.kernel_function"
+    ]
+    assert secret not in "\n".join(messages)
+    span.record_exception.assert_called_once_with(exc_info.value)
+    span.set_status.assert_called_once_with(StatusCode.ERROR, description=secret)
+
+
 async def test_invoke_async(kernel: Kernel):
     @kernel_function()
     async def async_function() -> str:

python/tests/unit/kernel/test_kernel.py

@@ -1,5 +1,6 @@
 # Copyright (c) Microsoft. All rights reserved.
 
+import logging
 import os
 import tempfile
 from collections.abc import Callable
@@ -298,6 +299,61 @@ async def test_invoke_function_call(kernel: Kernel, get_tool_call_mock):
         )
 
 
+async def test_invoke_function_call_does_not_log_argument_values(kernel: Kernel, caplog):
+    secret = "SYNTHETIC_SECRET_VALUE"
+
+    class TestPlugin:
+        @kernel_function
+        def echo(self, value: str) -> str:
+            return value
+
+    kernel.add_plugin(TestPlugin(), plugin_name="test")
+
+    caplog.set_level(logging.INFO, logger="semantic_kernel.kernel")
+    caplog.set_level(logging.DEBUG, logger="semantic_kernel.functions.kernel_function")
+    await kernel.invoke_function_call(
+        function_call=FunctionCallContent(name="test-echo", arguments={"value": secret}),
+        chat_history=ChatHistory(),
+        function_behavior=FunctionChoiceBehavior.Auto(),
+    )
+
+    messages = [
+        record.getMessage()
+        for record in caplog.records
+        if record.name in {"semantic_kernel.kernel", "semantic_kernel.functions.kernel_function"}
+    ]
+    assert secret not in "\n".join(messages)
+    assert "Calling test-echo function." in messages
+
+
+async def test_invoke_function_call_does_not_log_exception_values(kernel: Kernel, caplog):
+    secret = "SYNTHETIC_SECRET_VALUE"
+
+    class TestPlugin:
+        @kernel_function
+        def fail(self, value: str) -> str:
+            raise ValueError(value)
+
+    kernel.add_plugin(TestPlugin(), plugin_name="test")
+
+    caplog.set_level(logging.ERROR, logger="semantic_kernel.kernel")
+    caplog.set_level(logging.ERROR, logger="semantic_kernel.functions.kernel_function")
+    await kernel.invoke_function_call(
+        function_call=FunctionCallContent(name="test-fail", arguments={"value": secret}),
+        chat_history=ChatHistory(),
+        function_behavior=FunctionChoiceBehavior.Auto(),
+    )
+
+    messages = [
+        record.getMessage()
+        for record in caplog.records
+        if record.name in {"semantic_kernel.kernel", "semantic_kernel.functions.kernel_function"}
+    ]
+    assert secret not in "\n".join(messages)
+    assert "Function failed. Error type: ValueError." in messages
+    assert "Error invoking function test-fail. Error type: ValueError." in messages
+
+
 class LightsPlugin:
     lights = [
         {"id": 1, "name": "Table Lamp", "is_on": False},

python/tests/unit/prompt_template/test_template_helper.py

@@ -1,11 +1,13 @@
 # Copyright (c) Microsoft. All rights reserved.
 
+import logging
+
 import pytest
 
 from semantic_kernel.functions import kernel_function
 from semantic_kernel.functions.kernel_function_from_method import KernelFunctionFromMethod
 from semantic_kernel.kernel import Kernel
-from semantic_kernel.prompt_template.const import JINJA2_TEMPLATE_FORMAT_NAME
+from semantic_kernel.prompt_template.const import HANDLEBARS_TEMPLATE_FORMAT_NAME, JINJA2_TEMPLATE_FORMAT_NAME
 from semantic_kernel.prompt_template.utils.template_function_helpers import create_template_helper_from_function
 
 
@@ -26,6 +28,48 @@ def test_create_helpers(kernel: Kernel):
     assert int(str(result(x=1))) == 2
 
 
+async def test_template_helpers_do_not_log_argument_values(kernel: Kernel, caplog):
+    secret = "SYNTHETIC_SECRET_VALUE"
+    function = KernelFunctionFromMethod(kernel_function(lambda value: value, name="test"), plugin_name="test")
+    sync_jinja_helper = create_template_helper_from_function(
+        function, kernel, {}, JINJA2_TEMPLATE_FORMAT_NAME, enable_async=False
+    )
+    async_jinja_helper = create_template_helper_from_function(
+        function, kernel, {}, JINJA2_TEMPLATE_FORMAT_NAME, enable_async=True
+    )
+    handlebars_helper = create_template_helper_from_function(
+        function, kernel, {}, HANDLEBARS_TEMPLATE_FORMAT_NAME, enable_async=False
+    )
+
+    caplog.set_level(logging.DEBUG, logger="semantic_kernel.prompt_template.utils.template_function_helpers")
+    caplog.set_level(logging.DEBUG, logger="semantic_kernel.functions.kernel_function")
+    sync_jinja_helper(value=secret)
+    await async_jinja_helper(value=secret)
+    handlebars_helper(secret, secret, value=secret)
+
+    all_messages = [
+        record.getMessage()
+        for record in caplog.records
+        if record.name
+        in {
+            "semantic_kernel.prompt_template.utils.template_function_helpers",
+            "semantic_kernel.functions.kernel_function",
+        }
+    ]
+    messages = [
+        record.getMessage()
+        for record in caplog.records
+        if record.name == "semantic_kernel.prompt_template.utils.template_function_helpers"
+    ]
+    assert secret not in "\n".join(all_messages)
+    assert (
+        messages.count("Invoking function test-test with positional argument count 0 and keyword argument count 1.")
+        == 2
+    )
+    assert "Invoking function test-test with positional argument count 1 and keyword argument count 1." in messages
+    assert "Handlebars context with `this`." in messages
+
+
 @pytest.mark.parametrize(
     "template_format, enable_async, exception",
     [