Python: Azure AI Agent - Prevent LLM from modifying the FunctionResult

[moonbox3]

Discussed in #11043

^{Originally posted by vslepakov March 18, 2025}
Hi folks,

I have a plugin that I am using with the new Azure AI Agent (Agent) in semantic kernel (auto function invocation). One of the functions returns a string that should be displayed 100% as is without any modifications by the LLM (prompt engineering is not an option for various reasons).

I saw that something like this should be possible using a filter

I tried adding such a filter like so:

ai_agent.kernel.add_filter( FilterTypes.AUTO_FUNCTION_INVOCATION, auto_function_invocation_filter )

and the filter gets invoked. I call context.terminate = True when after this specific function was called but the string still gets modified by the LLM.

Is there a way to address this behavior?

Thanks!

[vslepakov]

on the current project we require the ability to avoid passing a function result to the LLM to avoid any modification, would this allow for it or is there another way? We still want to add the message to the thread though, so it appears in the conversation history as an assistant message (but unmodified by the LLM).

[moonbox3]

on the current project we require the ability to avoid passing a function result to the LLM to avoid any modification, would this allow for it or is there another way? We still want to add the message to the thread though, so it appears in the conversation history as an assistant message (but unmodified by the LLM).

Hi @vslepakov, I still owe you a solution on this. Apologies for the delay - this is on my radar for tomorrow. It is a big trickier with the server-side managed conversation and the need to keep the message ordering correct - a tool call requires a tool result. Should be able to get it handled soon.

[vslepakov]

Thanks @moonbox3 , sure no problem.

[moonbox3]

Hi @vslepakov, I've been looking into handling the auto function invocation filter for the AzureAIAgent and the OpenAIAssistantAgent. These server-side agents are trickier to handle regarding the filter. The reason is because once we handle the "requires action" event (an initiated tool call), and we return the tool call result, we cannot simply end the operation there -- the model needs to provide a final natural language answer. If you try to end early, and then you send another input to the model, it will throw an error that it cannot handle the input due to an ongoing run.

Previous some work I am doing now, we weren't even able to handle the auto function invocation filter for these two agents, so there is that.

As an example script, we have the following:

# Copyright (c) Microsoft. All rights reserved.

import asyncio
from typing import Annotated

from azure.identity.aio import DefaultAzureCredential

from semantic_kernel.agents import AzureAIAgent, AzureAIAgentSettings, AzureAIAgentThread
from semantic_kernel.contents import ChatMessageContent, FunctionCallContent, FunctionResultContent
from semantic_kernel.filters import (
    AutoFunctionInvocationContext,
    FilterTypes,
)
from semantic_kernel.functions import FunctionResult, kernel_function
from semantic_kernel.kernel import Kernel

"""
The following sample demonstrates how to create an Azure AI agent that answers
user questions. This sample demonstrates the basic steps to create an agent
and simulate a conversation with the agent.

This sample demonstrates how to create a filter that will be called for each
function call in the response. The filter can be used to modify the function
result or to terminate the function call. The filter can also be used to
log the function call or to perform any other action before or after the
function call.
"""


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) -> Annotated[str, "Returns the specials from the menu."]:
        return """
        Special Soup: Clam Chowder
        Special Salad: Cobb Salad
        Special Drink: Chai Tea
        """

    @kernel_function(description="Provides the price of the requested menu item.")
    def get_item_price(
        self, menu_item: Annotated[str, "The name of the menu item."]
    ) -> Annotated[str, "Returns the price of the menu item."]:
        return "$9.99"


# Define a kernel instance so we can attach the filter to it
kernel = Kernel()


# Define a list to store intermediate steps
intermediate_steps: list[ChatMessageContent] = []


# Define a callback function to handle intermediate step content messages
async def handle_intermediate_steps(message: ChatMessageContent) -> None:
    intermediate_steps.append(message)


@kernel.filter(FilterTypes.AUTO_FUNCTION_INVOCATION)
async def auto_function_invocation_filter(context: AutoFunctionInvocationContext, next):
    """A filter that will be called for each function call in the response."""
    print("\nAuto function invocation filter")
    print(f"Function: {context.function.name}")

    # if we don't call next, it will skip this function, and go to the next one
    await next(context)
    """
    Note: to simply return the unaltered function results, uncomment the `context.terminate = True` line and
    comment out the lines starting with `result = context.function_result` through `context.terminate = True`.
    context.terminate = True
    For this sample, simply setting `context.terminate = True` will return the unaltered function result:
    
    Auto function invocation filter
    Function: get_specials
    # Assistant: MenuPlugin-get_specials - 
            Special Soup: Clam Chowder
            Special Salad: Cobb Salad
            Special Drink: Chai Tea
    """
    result = context.function_result
    if "menu" in context.function.plugin_name.lower():
        print("Altering the Menu plugin function result...\n")
        context.function_result = FunctionResult(
            function=result.function,
            value="We are sold out, sorry!",
        )
        context.terminate = True


# Simulate a conversation with the agent
USER_INPUTS = ["What's the special food on the menu?", "What should I do then?"]


async def main() -> None:
    ai_agent_settings = AzureAIAgentSettings.create()

    async with (
        DefaultAzureCredential() as creds,
        AzureAIAgent.create_client(credential=creds) as client,
    ):
        # 1. Create an agent on the Azure AI agent service
        agent_definition = await client.agents.create_agent(
            model=ai_agent_settings.model_deployment_name,
            name="Host",
            instructions="Answer the user's questions about the menu.",
        )

        # 2. Create a Semantic Kernel agent for the Azure AI agent
        agent = AzureAIAgent(
            kernel=kernel,
            client=client,
            definition=agent_definition,
            plugins=[MenuPlugin()],  # Add the plugin to the agent
        )

        # 3. Create a thread for the agent
        # If no thread is provided, a new thread will be
        # created and returned with the initial response
        thread: AzureAIAgentThread = None

        try:
            for user_input in USER_INPUTS:
                print(f"# User: {user_input}")
                # 4. Invoke the agent with the specified message for response
                async for response in agent.invoke(
                    messages=user_input, thread=thread, on_intermediate_message=handle_intermediate_steps
                ):
                    # 5. Print the response
                    print(f"# {response.name}: {response}")
                    thread = response.thread
        finally:
            # 6. Cleanup: Delete the thread and agent
            await thread.delete() if thread else None
            await client.agents.delete_agent(agent.id)

        # Print the intermediate steps
        print("\nIntermediate Steps:")
        for msg in intermediate_steps:
            if any(isinstance(item, FunctionResultContent) for item in msg.items):
                for fr in msg.items:
                    if isinstance(fr, FunctionResultContent):
                        print(f"Function Result:> {fr.result} for function: {fr.name}")
            elif any(isinstance(item, FunctionCallContent) for item in msg.items):
                for fcc in msg.items:
                    if isinstance(fcc, FunctionCallContent):
                        print(f"Function Call:> {fcc.name} with arguments: {fcc.arguments}")
            else:
                print(f"{msg.role}: {msg.content}")


if __name__ == "__main__":
    asyncio.run(main())

This would give the following output:

"""
Sample Output:

# User: What's the special food on the menu?

Auto function invocation filter
Function: get_specials
Altering the Menu plugin function result...

# Host: I'm sorry, but all the specials on the menu are currently sold out. If there's anything else you're 
    looking for, please let me know!
# User: What should I do then?
# Host: You might consider ordering from the regular menu items instead. If you need any recommendations or 
    information about specific items, such as prices or ingredients, feel free to ask!

Intermediate Steps:
Function Call:> MenuPlugin-get_specials with arguments: {}
Function Result:> We are sold out, sorry! for function: MenuPlugin-get_specials
AuthorRole.ASSISTANT: I'm sorry, but all the specials on the menu are currently sold out. If there's anything 
    else you're looking for, please let me know!
AuthorRole.ASSISTANT: You might consider ordering from the regular menu items instead. If you need any 
    recommendations or information about specific items, such as prices or ingredients, feel free to ask!
"""

As a caller for these agent types, you will have the ability to configure the filter and receive the FunctionResultContent that contains the result of running the filter. Per my explanation above, I don't see a current way to "exit early" from the current operation with the model where it doesn't put it's final "natural language spin" on the response.

Note that the way to receive the function result content, as part of the filter, and/or function calling in general is to provide the on_intermediate_message callback.

[vslepakov]

Thanks @moonbox3. This aligns with my observations as well.
My workaround so far has been:

• Pass in KernelArguments with a context object in get_response
• The invoked function modifies the context object by setting the value of a property
• After the run completes I add the value of this property from the context object to the thread with AuthorRole.ASSISTANT without triggering a run.

the LLM does add its "natural language spin" but I still have my unmodified string as the last message.

I noticed that agent.add_chat_message is being deprecated, so will probably need to add that message on the thread object directly if this is even possible (don't see an add_message method).

Not sure if anything speaks agains this approach. Having that filter implementation might help to optimize the code though.

[moonbox3]

Thanks for the added context, @vslepakov. You should still have access to the same information if you decide to go the route shown above.

Regarding the deprecation for agent.add_chat_message(): yes, we're guiding users to provide the input message(s) directly on the agent.get_response(messages=<input messages>, ...), ... agent.invoke(messages=<input messages>, ...) or agent.invoke_stream(messages=<input messages>, ...).