Python: ChatCompletionAgent with Auto Function Calling - Pass sensitive information to functions

[vslepakov]

Hi,

I’m working with the ChatCompletionAgent and have a use case where I need to provide sensitive data (like a user access token) to one of the plugin functions. I tried using KernelArguments with get_response, but it seems that these arguments are passed to the LLM. Is there a way to supply this kind of information without exposing it to the LLM?

Thanks!

[moonbox3]

Hi @vslepakov, we only broadcast the kernel function's arguments to the model, so it knows what is required/not required. We are not broadcasting kernel argument values.

As a note, can you please have a look at the follow doc: https://learn.microsoft.com/en-us/semantic-kernel/concepts/ai-services/chat-completion/function-calling/?pivots=programming-language-python#custom-reserved-parameter-names-for-auto-function-calling

In a code sample there, we show special_arg: Annotated[str, {"include_in_function_choices": False} - where you can now provide the Annotated metadata. This tells the model that the argument is not required, and thus can be handled as a kernel argument and not have the model try and populate an "access token."

Separately, you could have an auth callback in your kernel function that gets the required access token. This would remove the input param to the kernel function related to auth. See an example here:

semantic-kernel/python/semantic_kernel/core_plugins/sessions_python_tool/sessions_python_plugin.py

Line 174 in 2a5fa44

auth_token = await self._ensure_auth_token()

[vslepakov]

Hi @moonbox3, thank for the quick response!
I am using this code:

...
thread = ChatHistoryAgentThread = ChatHistoryAgentThread(chat_history=history)
result = await self._agent.get_response(
        messages=chat_messages,
        thread=thread
    )
...

with ChatHistory being ChatHistoryInCosmosDB from this sample and I can see all KernelArguments values persisted in CosmosDB including the token.

it seems that the auth callback option would not work for us as the token is passed to our web api in a header on each HTTP request and in the sample it looks like it is being retrieved from somewhere. We do not persist it nor have we the ability to fetch the token from an endpoint.

I didn't know about the include_in_function_choices annotation. Thanks for the pointer! In case I add that special_arg to the KerbelArguments, do I still need to have that sepecial_arg explicitly as function parameter or do i mark the whole KernelArguments as include_in_function_choices = False ?

[moonbox3]

Hi @vslepakov, let's take the following kernel function example:

class MenuPlugin:
    """A sample Menu Plugin used for the concept sample."""

    @kernel_function(description="Provides a list of specials from the menu.")
    def get_specials(
        self, api_key: Annotated[str, {"include_in_function_choices": False}]
    ) -> Annotated[str, "Returns the specials from the menu."]:
        print(f"API Key: {api_key}")
        return """
        Special Soup: Clam Chowder
        Special Salad: Cobb Salad
        Special Drink: Chai Tea
        """

You can see I added api_key: Annotated[str, {"include_in_function_choices": False}] to our typical example that has no arg for that method, making note to include {"include_in_function_choices": False}.

When we broadcast the json to the model, it sees:

{
    "type": "function",
    "function":
    {
        "name": "MenuPlugin-get_specials",
        "description": "Provides a list of specials from the menu.",
        "parameters":
        {
            "type": "object",
            "properties":
            {},
            "required":
            []
        }
    }
}

No mention of api_key as a parameter.

And when we actually call the function with the following script:

USER_INPUTS = [
    "Hello",
    "What is the special soup?",
    "What does that cost?",
    "Thank you",
]


async def main():
    # 1. Create the agent
    agent = ChatCompletionAgent(
        service=AzureChatCompletion(),
        name="Host",
        instructions="Answer questions about the menu.",
        plugins=[MenuPlugin()],
    )

    # 2. Create a thread to hold the conversation
    # If no thread is provided, a new thread will be
    # created and returned with the initial response
    thread: ChatHistoryAgentThread = None

    for user_input in USER_INPUTS:
        print(f"# User: {user_input}")
        # 4. Invoke the agent for a response
        arguments = KernelArguments(api_key="12345")
        response = await agent.get_response(messages=user_input, thread=thread, arguments=arguments)
        print(f"# {response.name}: {response} ")
        thread = response.thread

    # 4. Cleanup: Clear the thread
    await thread.delete() if thread else None

I see:

# User: Hello
# Host: Hello! How can I assist you today? 
# User: What is the special soup?
API Key: 12345 <-- printed from my sample kernel function
# Host: The special soup is Clam Chowder. Is there anything else you'd like to know about the specials? 
# User: What does that cost?
# Host: The Clam Chowder costs $9.99. If you have any more questions, feel free to ask! 
# User: Thank you
# Host: You're welcome! If you have any other questions in the future, don't hesitate to reach out. Have a great day! 

[vslepakov]

ok, that would be fine of course. Thanks for looking into this!

[moonbox3]

@vslepakov please let me know what your exact flow is here, so that I am investigating the same path. For example, what SK content types are handled when, what is being serialized, deserialized, etc. As I mentioned, the metadata of a content type is never sent to the model.

If you can give me a quick flow, or even a small script to reproduce anything, that would help. Thanks.

[vslepakov]

Hey @moonbox3, sure. The flow is quite simple:

await history.read_messages(thread_id=thread_id)
...

thread: ChatHistoryAgentThread = ChatHistoryAgentThread(
        chat_history=history, thread_id=thread_id
    )

argument_overrides = KernelArguments(request_context=request_context)

result = await self._agent.get_response(
        messages=chat_messages,
        thread=thread,
        arguments=argument_overrides
    )
...

await history.store_messages(thread.id)

request_context above contains the token. When the messages are stored in CosmosDB, the token is part of the metadata.
I do not specify any content type explicitly, it is all strings.
If the token is not passed to the LLM, this is great, so probably no need to investigate.

The only question is: how to make sure the token is not stored in CosmosDB?

I do need to store the function result, as it contains the API response with data needed to continue the conversation BUT I'd like to remove the metadata with the sensitive data. Is there an easy way to do this or is this metadata required to continue the conversation when I rehydrate the chat history?

P.S. storing and rehydrating instances of ChatMessageContent

[moonbox3]

I'd recommend just popping the metadata out then before storing in CosmosDB. The metadata isn't required when rehydrating the chat history.

[vslepakov]

ok, got it. I'll do that. Thanks!