.Net: New Feature: Add tool call status as a message in GetStreamingChatMessageContentsAsync() StreamingChatMessageContent responses

[kcrandall]

When a tool is called there is no way for the user to know that tool usage is in progress. All the tool usage hidden under the hood and there is no way to know that it is happening. Given the below code:

     var stream = _chatCompletionService.GetStreamingChatMessageContentsAsync(
                        prompt.ChatHistory,
                        kernel: _serviceProvider.GetRequiredService<Kernel>(),
                        executionSettings: executionSettiongs
                    );

await foreach (StreamingChatMessageContent res in stream)
{
       res.ToolCalldata <- return something either here or in metadata
}

There should be a way in the StreamingChatMessageContent messages to get the status of a tool there are log messages that show it. See below there are INFORMATION logs that say a tool is called but it isnt returned as a result as when i am printing out all the responses seriazlied there is nothing you can decode to see that the tool is invoked, finished, failed, etc.

response:{"Content":null,"Role":null,"ChoiceIndex":0,"ModelId":"gpt-4o-mini","Metadata":{"CompletionId":"chatcmpl-B0vwinhyvk4FyxnQlwGpuWxh0kECK","CreatedAt":"2025-02-14T19:47:24+00:00","SystemFingerprint":"fp_00428b782a","RefusalUpdate":null,"Usage":{"OutputTokenCount":39,"InputTokenCount":1041,"TotalTokenCount":1080,"OutputTokenDetails":{"ReasoningTokenCount":0,"AudioTokenCount":0},"InputTokenDetails":{"AudioTokenCount":0,"CachedTokenCount":0}},"FinishReason":null}}
INFORMATION | Function "GoogleSearchPlugin"-"search" invoking. {"PluginName":"GoogleSearchPlugin","FunctionName":"search","EventId":"Name: LogFunctionInvoking","SourceContext":"Microsoft.SemanticKernel.KernelFunction","ActionId":"040c98ce-1969-4a73-b56f-42f140909536","ActionName":"APIApp.Controllers.TextAssistant.GetTextAssistantJson (APIApp)","RequestId":"0HNAD61IK9L0V:00000001","RequestPath":"/TextAssistant/json","ConnectionId":"0HNAD61IK9L0V"}
INFORMATION | Function "GoogleSearchPlugin"-"search" succeeded. {"PluginName":"GoogleSearchPlugin","FunctionName":"search","EventId":"Name: LogFunctionInvokedSuccess","SourceContext":"Microsoft.SemanticKernel.KernelFunction","ActionId":"040c98ce-1969-4a73-b56f-42f140909536","ActionName":"APIApp.Controllers.TextAssistant.GetTextAssistantJson (APIApp)","RequestId":"0HNAD61IK9L0V:00000001","RequestPath":"/TextAssistant/json","ConnectionId":"0HNAD61IK9L0V"}
INFORMATION | Function "GoogleSearchPlugin"-"search" completed. Duration: 0.6508721s {"PluginName":"GoogleSearchPlugin","FunctionName":"search","Duration":"0.6508721","EventId":"Name: LogFunctionComplete","SourceContext":"Microsoft.SemanticKernel.KernelFunction","ActionId":"040c98ce-1969-4a73-b56f-42f140909536","ActionName":"APIApp.Controllers.TextAssistant.GetTextAssistantJson (APIApp)","RequestId":"0HNAD61IK9L0V:00000001","RequestPath":"/TextAssistant/json","ConnectionId":"0HNAD61IK9L0V"}
Metadata:{"CompletionId":"chatcmpl-B0vwkZsqeEzccCinlNC6thAHoIbJg","CreatedAt":"2025-02-14T19:47:26+00:00","SystemFingerprint":"fp_00428b782a","RefusalUpdate":null,"Usage":null,"FinishReason":null}
response:{"Content":"","Role":{"Label":"Assistant"},"ChoiceIndex":0,"ModelId":"gpt-4o-mini","Metadata":{"CompletionId":"chatcmpl-B0vwkZsqeEzccCinlNC6thAHoIbJg","CreatedAt":"2025-02-14T19:47:26+00:00","SystemFingerprint":"fp_00428b782a","RefusalUpdate":null,"Usage":null,"FinishReason":null}}
Metadata:{"CompletionId":"chatcmpl-B0vwkZsqeEzccCinlNC6thAHoIbJg","CreatedAt":"2025-02-14T19:47:26+00:00","SystemFingerprint":"fp_00428b782a","RefusalUpdate":null,"Usage":null,"FinishReason":null}

I propose adding a an additional property on the StreamingChatMessageContent or putting it in the Metadata object that a tool has been called. Send a message when it is invoked, as well as when it succeeds or fails. Include the tool name, parameters, etc. As well as a message with the response from the tool. If multiple tools are called there should either be multiple messages returned or a list of all tools. I think multiple messages would be better as a UI could update each tool when they complete. For example there might be a widget that turns red or green and when tool 1 updates it would turn green and the other one would stay red until it completes.

[markwallace-microsoft]

@kcrandall have you considered using auto invocation filters to achieve what you need?

• https://learn.microsoft.com/en-us/semantic-kernel/concepts/enterprise-readiness/filters?pivots=programming-language-csharp#auto-function-invocation-filter
• https://github.com/microsoft/semantic-kernel/blob/main/dotnet/samples/Concepts/Filtering/AutoFunctionInvocationFiltering.cs

[kcrandall]

@markwallace-microsoft the issue is that the IAutoFunctionInvocationFilter doesn't have access to the IAsyncEnumerable<StreamingChatMessageContent> so there is no way to send a message to the output stream of the state of the function call. It's also a singleton so there isn't a way for me to inject a service that maybe can have a Channel that my request can have access to that I could combine the messages with.

If I'm understanding how this works, the only way this would work is if I made a new kernel for every single GetStreamingChatMessageContentsAsync() call.

[kcrandall]

I made a workaround but it's very very ugly. I think this is the only way that's possible to achieve this. It would be better if there was a direct way to access the output stream from the Filters or if Semantic Kernel just automatically notified the user when tools are invoked.

The other issue is I have no clue how to determine if the function completed in error or was successful with this implementation I don't see a way to get any function errors.

public class NotifyStreamingFunctionStateFilter : IAutoFunctionInvocationFilter
    {
        private static readonly ConcurrentDictionary<string, (System.Threading.Channels.Channel<StreamingChatMessageContent> Channel, DateTime LastUsed)> _channels = new();
        private static readonly TimeSpan ExpirationTime = TimeSpan.FromMinutes(5);
        private static Timer? CleanupTimer;

        public NotifyStreamingFunctionStateFilter()
        {
            if (CleanupTimer is not null)
                CleanupTimer.Dispose();
            CleanupTimer = new Timer(_ => CleanupExpiredEntries(), null, TimeSpan.Zero, TimeSpan.FromSeconds(30));
        }

        public async Task OnAutoFunctionInvocationAsync(AutoFunctionInvocationContext context,
            Func<AutoFunctionInvocationContext, Task> next)
        {
            if (context.IsStreaming)
            { ;
                object? requestId = null;
                context.ChatHistory.First()?.Metadata?.TryGetValue("requestId", out requestId);
                if (requestId is not null)
                {
                    if (_channels.TryGetValue(requestId.ToString(), out var entry))
                    {
                        await entry.Channel.Writer.WriteAsync(new StreamingChatMessageContent(role: null, content: null, innerContent: null, metadata: new Dictionary<string, object?>()
                        {
                            {"functionState", "invoked"},
                            {"pluginName", context.Function.PluginName},
                            {"functionName", context.Function.Name}
                        }));
                    }
                }
                await next(context);
                if (requestId is not null)
                {
                    if (_channels.TryGetValue(requestId.ToString(), out var entry))
                    {
                        await entry.Channel.Writer.WriteAsync(new StreamingChatMessageContent(role: null, content: null, innerContent: null, metadata: new Dictionary<string, object?>()
                        {
                            {"functionState", "completed"},
                            {"pluginName", context.Function.PluginName},
                            {"functionName", context.Function.Name}
                        }));
                    }
                }
            }
            else
            {
                await next(context);
            }
        }

        public static Channel<StreamingChatMessageContent> RegisterChannel(string id)
        {
            var channel = Channel.CreateUnbounded<StreamingChatMessageContent>();
            _channels[id] = (channel, DateTime.UtcNow);
            return channel;
        }

        public static void RemoveChannel(string chatHash)
        {
            _channels.TryRemove(chatHash, out _);
        }

        private static void CleanupExpiredEntries()
        {
            var now = DateTime.UtcNow;
            foreach (var key in _channels.Keys.ToList())
            {
                if (_channels.TryGetValue(key, out var entry) && (now - entry.LastUsed) > ExpirationTime)
                {
                    _channels.TryRemove(key, out _);
                }
            }
        }
    }

Actual usage of the notifier requires this.

var requestId = Guid.NewGuid().ToString();
 if (chatHistory.First() is not null)
            {
                var metadata = new Dictionary<string, object?>();
                if (chatHistory.First().Metadata is not null)
                {
                    foreach (var item in chatHistory.First().Metadata)
                    {
                        metadata.Add(item.Key, item.Value);
                    }
                }
                metadata["requestId"] = requestId;
                chatHistory.First().Metadata = metadata;
            }
Channel<StreamingChatMessageContent> notificationChannel = NotifyStreamingFunctionStateFilter.RegisterChannel(requestId);
var stream = _chatCompletionService.GetStreamingChatMessageContentsAsync(
                    prompt.ChatHistory,
                    kernel: _serviceProvider.GetRequiredService<Kernel>(),
                    executionSettings: semanticLayerExecutionSettiongs
                );
var combinedStream = Utils.IAsyncEnumerable.Helpers.MergeAsync(stream, notificationChannel.Reader.ReadAllAsync(), Helpers.MergeCompletionMode.Any);
await foreach (var res in combinedStream)
                {
                    var token = res?.Content;
                    // This will have the functionState and functionName keys in some of the messages
                    Console.WriteLine("Metadata:" + JsonSerializer.Serialize(res.Metadata));
                  }
NotifyStreamingFunctionStateFilter.RemoveChannel(requestId);

Helper method to merge two IAsyncEnumerable.

 public enum MergeCompletionMode
    {
        All,
        Any
    }
    
    public static async IAsyncEnumerable<T> MergeAsync<T>(
        IAsyncEnumerable<T> first,
        IAsyncEnumerable<T> second,
        MergeCompletionMode mode = MergeCompletionMode.All,
        [EnumeratorCancellation] CancellationToken cancellationToken = default)
    {
        // Create an unbounded channel to merge the items.
        var channel = Channel.CreateUnbounded<T>();
        
        // Create a CTS that lets us cancel the reading loops
        using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
        var linkedToken = linkedCts.Token;

        // Local function that reads from one source and writes items into the channel.
        async Task ReadAllAsync(IAsyncEnumerable<T> source)
        {
            try
            {
                await foreach (var item in source.WithCancellation(linkedToken))
                {
                    await channel.Writer.WriteAsync(item, cancellationToken);
                }
            }
            catch (OperationCanceledException) when (linkedToken.IsCancellationRequested)
            {
                // Ignore cancellation.
            }
            catch (Exception ex)
            {
                // If one source fails, complete the channel with the error.
                channel.Writer.TryComplete(ex);
                return;
            }
        }

        // Start reading from both sources concurrently.
        var task1 = ReadAllAsync(first);
        var task2 = ReadAllAsync(second);

        if (mode == MergeCompletionMode.Any)
        {
            // In "Any" mode, as soon as one source finishes, cancel the other.
            _ = Task.WhenAny(task1, task2)
                .ContinueWith(_ => channel.Writer.Complete(), cancellationToken);
        }
        else // MergeCompletionMode.All
        {
            // In "All" mode, complete only after both tasks finish.
            _ = Task.WhenAll(task1, task2)
                .ContinueWith(_ => channel.Writer.Complete(), cancellationToken);
        }

        // Yield items from the channel as they arrive.
        await foreach (var item in channel.Reader.ReadAllAsync(cancellationToken))
        {
            yield return item;
        }
    }