AsyncStreamingResponseUpdateCollectionResult.cs

// Copyright (c) Microsoft. All rights reserved.

using System.ClientModel;
using System.Diagnostics.CodeAnalysis;
using Microsoft.Shared.DiagnosticIds;
using OpenAI.Responses;

namespace Microsoft.Agents.AI.OpenAI;

[Experimental(DiagnosticIds.Experiments.AIOpenAIResponses)]
internal sealed class AsyncStreamingResponseUpdateCollectionResult : AsyncCollectionResult<StreamingResponseUpdate>
{
    private readonly IAsyncEnumerable<AgentResponseUpdate> _updates;

    internal AsyncStreamingResponseUpdateCollectionResult(IAsyncEnumerable<AgentResponseUpdate> updates)
    {
        this._updates = updates;
    }

    public override ContinuationToken? GetContinuationToken(ClientResult page) => null;

    public override async IAsyncEnumerable<ClientResult> GetRawPagesAsync()
    {
        yield return ClientResult.FromValue(this._updates, new StreamingUpdatePipelineResponse(this._updates));
    }

    protected async override IAsyncEnumerable<StreamingResponseUpdate> GetValuesFromPageAsync(ClientResult page)
    {
        var updates = ((ClientResult<IAsyncEnumerable<AgentResponseUpdate>>)page).Value;

        await foreach (var update in updates.ConfigureAwait(false))
        {
            switch (update.RawRepresentation)
            {
                case StreamingResponseUpdate rawUpdate:
                    yield return rawUpdate;
                    break;

                case Extensions.AI.ChatResponseUpdate { RawRepresentation: StreamingResponseUpdate rawUpdate }:
                    yield return rawUpdate;
                    break;

                default:
                    // TODO: The OpenAI library does not currently expose model factory methods for creating
                    // StreamingResponseUpdates. We are thus unable to manufacture such instances when there isn't
                    // already one in the update and instead skip them.
                    break;
            }
        }
    }
}