Add support for streaming text IAsyncEnumerable<string> results

[alexminza]

Is there an existing issue for this?

• I have searched the existing issues

Is your feature request related to a problem? Please describe the problem.

I am trying to return a streaming IAsyncEnumerable<string> SematicKernel chat completion from GetStreamingChatCompletionsAsync, GetStreamingChatMessageAsync methods.

Currently simply returning IAsyncEnumerable<string> produces a streaming JSON array of strings result. The desired behavior is simple streaming text strings result.

This would effectively produce a streaming ChatGTP-like completion response generated by the method as the results become available from the OpenAI endpoints.

Describe the solution you'd like

public class AsyncEnumerableStringsResult : IResult, IContentTypeHttpResult, IStatusCodeHttpResult
{
    protected readonly IAsyncEnumerable<string> chunks;

    public string? ContentType => "text/plain; charset=utf-8";

    public int StatusCode => StatusCodes.Status200OK;

    int? IStatusCodeHttpResult.StatusCode => StatusCode;

    public AsyncEnumerableStringsResult(IAsyncEnumerable<string> chunks) => this.chunks = chunks ?? throw new ArgumentNullException(nameof(chunks));

    public async Task ExecuteAsync(HttpContext httpContext)
    {
        if (httpContext == null)
            throw new ArgumentNullException(nameof(httpContext));

        httpContext.Response.ContentType = this.ContentType;
        httpContext.Response.StatusCode = this.StatusCode;

        await foreach (var chunk in this.chunks)
            if (!string.IsNullOrEmpty(chunk))
                await httpContext.Response.WriteAsync(chunk, cancellationToken: httpContext.RequestAborted);
    }
}

Usage example:

app.MapPost("/ChatAsyncStream", ([FromBody] ChatRequest chatRequest, ChatPlugin plugin, ILogger logger, CancellationToken cancellationToken) =>
    {
        if (string.IsNullOrWhiteSpace(chatRequest.Question))
            throw new ArgumentNullException(nameof(chatRequest.Question));

        var result = plugin.ChatAsyncStream(
            question: chatRequest.Question,
            chatHistory: chatRequest.ChatHistory,
            logger: logger,
            cancellationToken: cancellationToken
        );

        return new AsyncEnumerableStringsResult(result);
    })
    .WithName("ChatAsyncStream")
    .WithOpenApi()
    .Produces<IAsyncEnumerable<string>>();

Additional context

No response

[davidfowl]

How is the client consuming this response? Do you have an example?

[alexminza]

@davidfowl here's a simple example built by me, based on the great tutorial from Streamlit https://docs.streamlit.io/knowledge-base/tutorials/build-conversational-apps#write-the-app

Streamlit ChatBot app

#!/usr/bin/env python3

import os, logging
import streamlit as st
from azureapi import AzureAPI

from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv()) # read local .env file

azure_api_endpoint = os.getenv('AZUREAPI_ENDPOINT')
azure_api = AzureAPI(endpoint=azure_api_endpoint)

#https://docs.streamlit.io/knowledge-base/tutorials/build-conversational-apps
st.set_page_config(
    page_title="ChatBot",
    page_icon=":robot:"
)

st.title("ChatBot")

# Initialize chat history
if "messages" not in st.session_state:
    st.session_state.messages = []
if "chat_history" not in st.session_state:
    st.session_state.chat_history = ""

# Display chat messages from history on app rerun
for message in st.session_state.messages:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

# React to user input
if prompt := st.chat_input("Enter message here"):
    # Add user message to chat history
    st.session_state.messages.append({"role": "user", "content": prompt})

    # Display user message in chat message container
    with st.chat_message("user"):
        st.markdown(prompt)

    with st.spinner(text="In progress..."):
        with st.chat_message("assistant"):
            message_placeholder = st.empty()
            response_text = ''

            response_stream = azure_api.ChatStream(question=prompt, chatHistory=st.session_state.chat_history)
            for response_chunk in response_stream:
                if response_chunk:
                    response_text += response_chunk
                    message_placeholder.markdown(response_text + "▌")

            message_placeholder.markdown(response_text)

    # Display assistant response in chat message container
    if response_text:
        # Add assistant response to chat history
        st.session_state.messages.append({"role": "assistant", "content": response_text})
    else:
        st.error("ERROR")

API Client

class AzureAPI:
    SESSION = requests.Session()
    DEFAULT_TIMEOUT = 180
    API_ENDPOINT = None

    def __init__(self, endpoint: str) -> None:
        self.API_ENDPOINT = endpoint

    def ChatStream(self, question: str, chatHistory: str = None):
        url = f'{self.API_ENDPOINT}/ChatAsyncStream'
        json = {
            "question": question,
            "chatHistory": chatHistory
        }

        with AzureAPI.SESSION.request(method='POST', url=url, json=json, stream=True, timeout=AzureAPI.DEFAULT_TIMEOUT) as response:
            yield from response.iter_content(chunk_size=None, decode_unicode=True)

[flq]

A more generic solution might be to have a Server-Sent-Events result object that accepts an IAsyncEnumerable - a client can then consume it via javascript's EventSource class.

[dotnet-policy-service]

Looks like this PR hasn't been active for some time and the codebase could have been changed in the meantime.
To make sure no conflicting changes have occurred, please rerun validation before merging. You can do this by leaving an /azp run comment here (requires commit rights), or by simply closing and reopening.

[dotnet-policy-service]

Looks like this PR hasn't been active for some time and the codebase could have been changed in the meantime.
To make sure no conflicting changes have occurred, please rerun validation before merging. You can do this by leaving an /azp run comment here (requires commit rights), or by simply closing and reopening.

[davidfowl]

Related to dotnet/runtime#98105

[captainsafia]

Closing since we support a built-in SSE results type as of .NET 10.