Skip to content

OpenAI: investigation for StreamingClientResult<T> #39

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
annelo-msft wants to merge 2 commits into
base: main
Choose a base branch
from
Draft

OpenAI: investigation for StreamingClientResult<T> #39

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e1d8bfa
Initial implementation
annelo-msft Mar 21, 2024
4760585
make subtype internal
annelo-msft Mar 21, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 10 additions & 24 deletions .dotnet/OpenAI.sln
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,37 +1,19 @@
Microsoft Visual Studio Solution File, Format Version 12.00
# Visual Studio Version 16
VisualStudioVersion = 16.0.29709.97
# Visual Studio Version 17
VisualStudioVersion = 17.9.34701.34
MinimumVisualStudioVersion = 10.0.40219.1
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "OpenAI", "src\OpenAI.csproj", "{28FF4005-4467-4E36-92E7-DEA27DEB1519}"
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "OpenAI", "src\OpenAI.csproj", "{28FF4005-4467-4E36-92E7-DEA27DEB1519}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "OpenAI.Tests", "tests\OpenAI.Tests.csproj", "{1F1CD1D4-9932-4B73-99D8-C252A67D4B46}"
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "OpenAI.Tests", "tests\OpenAI.Tests.csproj", "{1F1CD1D4-9932-4B73-99D8-C252A67D4B46}"
EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "System.ClientModel", "..\..\azure-sdk-for-net\sdk\core\System.ClientModel\src\System.ClientModel.csproj", "{297DA5FF-1CD7-4183-8C13-45987286D33F}"
EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU
Release|Any CPU = Release|Any CPU
EndGlobalSection
GlobalSection(ProjectConfigurationPlatforms) = postSolution
{B0C276D1-2930-4887-B29A-D1A33E7009A2}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{B0C276D1-2930-4887-B29A-D1A33E7009A2}.Debug|Any CPU.Build.0 = Debug|Any CPU
{B0C276D1-2930-4887-B29A-D1A33E7009A2}.Release|Any CPU.ActiveCfg = Release|Any CPU
{B0C276D1-2930-4887-B29A-D1A33E7009A2}.Release|Any CPU.Build.0 = Release|Any CPU
{8E9A77AC-792A-4432-8320-ACFD46730401}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{8E9A77AC-792A-4432-8320-ACFD46730401}.Debug|Any CPU.Build.0 = Debug|Any CPU
{8E9A77AC-792A-4432-8320-ACFD46730401}.Release|Any CPU.ActiveCfg = Release|Any CPU
{8E9A77AC-792A-4432-8320-ACFD46730401}.Release|Any CPU.Build.0 = Release|Any CPU
{A4241C1F-A53D-474C-9E4E-075054407E74}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{A4241C1F-A53D-474C-9E4E-075054407E74}.Debug|Any CPU.Build.0 = Debug|Any CPU
{A4241C1F-A53D-474C-9E4E-075054407E74}.Release|Any CPU.ActiveCfg = Release|Any CPU
{A4241C1F-A53D-474C-9E4E-075054407E74}.Release|Any CPU.Build.0 = Release|Any CPU
{FA8BD3F1-8616-47B6-974C-7576CDF4717E}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{FA8BD3F1-8616-47B6-974C-7576CDF4717E}.Debug|Any CPU.Build.0 = Debug|Any CPU
{FA8BD3F1-8616-47B6-974C-7576CDF4717E}.Release|Any CPU.ActiveCfg = Release|Any CPU
{FA8BD3F1-8616-47B6-974C-7576CDF4717E}.Release|Any CPU.Build.0 = Release|Any CPU
{85677AD3-C214-42FA-AE6E-49B956CAC8DC}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{85677AD3-C214-42FA-AE6E-49B956CAC8DC}.Debug|Any CPU.Build.0 = Debug|Any CPU
{85677AD3-C214-42FA-AE6E-49B956CAC8DC}.Release|Any CPU.ActiveCfg = Release|Any CPU
{85677AD3-C214-42FA-AE6E-49B956CAC8DC}.Release|Any CPU.Build.0 = Release|Any CPU
{28FF4005-4467-4E36-92E7-DEA27DEB1519}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{28FF4005-4467-4E36-92E7-DEA27DEB1519}.Debug|Any CPU.Build.0 = Debug|Any CPU
{28FF4005-4467-4E36-92E7-DEA27DEB1519}.Release|Any CPU.ActiveCfg = Release|Any CPU
Expand All @@ -40,6 +22,10 @@ Global
{1F1CD1D4-9932-4B73-99D8-C252A67D4B46}.Debug|Any CPU.Build.0 = Debug|Any CPU
{1F1CD1D4-9932-4B73-99D8-C252A67D4B46}.Release|Any CPU.ActiveCfg = Release|Any CPU
{1F1CD1D4-9932-4B73-99D8-C252A67D4B46}.Release|Any CPU.Build.0 = Release|Any CPU
{297DA5FF-1CD7-4183-8C13-45987286D33F}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{297DA5FF-1CD7-4183-8C13-45987286D33F}.Debug|Any CPU.Build.0 = Debug|Any CPU
{297DA5FF-1CD7-4183-8C13-45987286D33F}.Release|Any CPU.ActiveCfg = Release|Any CPU
{297DA5FF-1CD7-4183-8C13-45987286D33F}.Release|Any CPU.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE
Expand Down
15 changes: 6 additions & 9 deletions .dotnet/src/Custom/Chat/ChatClient.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -208,11 +208,10 @@ public virtual StreamingClientResult<StreamingChatUpdate> CompleteChatStreaming(
throw new ClientResultException(response);
}

ClientResult genericResult = ClientResult.FromResponse(response);
return StreamingClientResult<StreamingChatUpdate>.CreateFromResponse(
genericResult,
return StreamingEventResult<StreamingChatUpdate>.CreateFromResponse(
response,
(responseForEnumeration) => SseAsyncEnumerator<StreamingChatUpdate>.EnumerateFromSseStream(
responseForEnumeration.GetRawResponse().ContentStream,
responseForEnumeration.ContentStream,
e => StreamingChatUpdate.DeserializeStreamingChatUpdates(e)));
}

Expand All @@ -229,7 +228,6 @@ public virtual StreamingClientResult<StreamingChatUpdate> CompleteChatStreaming(
/// The number of independent, alternative choices that the chat completion request should generate.
/// </param>
/// <param name="options"> Additional options for the chat completion request. </param>
/// <param name="cancellationToken"> The cancellation token for the operation. </param>
/// <returns> A streaming result with incremental chat completion updates. </returns>
public virtual async Task<StreamingClientResult<StreamingChatUpdate>> CompleteChatStreamingAsync(
IEnumerable<ChatRequestMessage> messages,
Expand All @@ -246,11 +244,10 @@ public virtual async Task<StreamingClientResult<StreamingChatUpdate>> CompleteCh
throw new ClientResultException(response);
}

ClientResult genericResult = ClientResult.FromResponse(response);
return StreamingClientResult<StreamingChatUpdate>.CreateFromResponse(
genericResult,
return StreamingEventResult<StreamingChatUpdate>.CreateFromResponse(
response,
(responseForEnumeration) => SseAsyncEnumerator<StreamingChatUpdate>.EnumerateFromSseStream(
responseForEnumeration.GetRawResponse().ContentStream,
responseForEnumeration.ContentStream,
e => StreamingChatUpdate.DeserializeStreamingChatUpdates(e)));
}

Expand Down
5 changes: 4 additions & 1 deletion .dotnet/src/OpenAI.csproj
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,10 @@
</PropertyGroup>

<ItemGroup>
<PackageReference Include="System.ClientModel" Version="1.1.0-alpha.20240319.1" />
<PackageReference Include="System.Text.Json" Version="8.0.2" />
</ItemGroup>

<ItemGroup>
<ProjectReference Include="..\..\..\azure-sdk-for-net\sdk\core\System.ClientModel\src\System.ClientModel.csproj" />
</ItemGroup>
</Project>
91 changes: 17 additions & 74 deletions .dotnet/src/Utility/StreamingResult.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,9 +1,8 @@
using System.ClientModel;
using System;
using System.ClientModel;
using System.ClientModel.Primitives;
using System.Threading;
using System.Collections.Generic;
using System;
using System.Threading;

namespace OpenAI;

Expand All @@ -12,84 +11,28 @@ namespace OpenAI;
/// is still being received.
/// </summary>
/// <typeparam name="T"> The data type representative of distinct, streamable items. </typeparam>
public class StreamingClientResult<T>
: IDisposable
, IAsyncEnumerable<T>
internal class StreamingEventResult<T> : StreamingClientResult<T>
Copy link
Copy Markdown
Collaborator Author

@annelo-msft annelo-msft Mar 21, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This type becomes internal.

{
private ClientResult _rawResult { get; }
private IAsyncEnumerable<T> _asyncEnumerableSource { get; }
private bool _disposedValue { get; set; }

private StreamingClientResult() { }

private StreamingClientResult(
ClientResult rawResult,
Func<ClientResult, IAsyncEnumerable<T>> asyncEnumerableProcessor)

private StreamingEventResult(PipelineResponse response,
Func<PipelineResponse, IAsyncEnumerable<T>> asyncEnumerableProcessor)
: base(response)
{
_rawResult = rawResult;
_asyncEnumerableSource = asyncEnumerableProcessor.Invoke(rawResult);
_asyncEnumerableSource = asyncEnumerableProcessor.Invoke(response);
}

/// <summary>
/// Creates a new instance of <see cref="StreamingClientResult{T}"/> using the provided underlying HTTP response. The
/// provided function will be used to resolve the response into an asynchronous enumeration of streamed response
/// items.
/// </summary>
/// <param name="result">The HTTP response.</param>
/// <param name="asyncEnumerableProcessor">
/// The function that will resolve the provided response into an IAsyncEnumerable.
/// </param>
/// <returns>
/// A new instance of <see cref="StreamingClientResult{T}"/> that will be capable of asynchronous enumeration of
/// <typeparamref name="T"/> items from the HTTP response.
/// </returns>
internal static StreamingClientResult<T> CreateFromResponse(
ClientResult result,
Func<ClientResult, IAsyncEnumerable<T>> asyncEnumerableProcessor)
internal static StreamingEventResult<T> CreateFromResponse(
PipelineResponse response,
Func<PipelineResponse, IAsyncEnumerable<T>> asyncEnumerableProcessor)
{
return new(result, asyncEnumerableProcessor);
return new(response, asyncEnumerableProcessor);
}

/// <summary>
/// Gets the underlying <see cref="PipelineResponse"/> instance that this <see cref="StreamingClientResult{T}"/> may enumerate
/// over.
/// </summary>
/// <returns> The <see cref="PipelineResponse"/> instance attached to this <see cref="StreamingClientResult{T}"/>. </returns>
public PipelineResponse GetRawResponse() => _rawResult.GetRawResponse();

/// <summary>
/// Gets the asynchronously enumerable collection of distinct, streamable items in the response.
/// </summary>
/// <remarks>
/// <para> The return value of this method may be used with the "await foreach" statement. </para>
/// <para>
/// As <see cref="StreamingClientResult{T}"/> explicitly implements <see cref="IAsyncEnumerable{T}"/>, callers may
/// enumerate a <see cref="StreamingClientResult{T}"/> instance directly instead of calling this method.
/// </para>
/// </remarks>
/// <returns></returns>
public IAsyncEnumerable<T> EnumerateValues() => this;

/// <inheritdoc/>
public void Dispose()
{
Dispose(disposing: true);
GC.SuppressFinalize(this);
}

/// <inheritdoc/>
protected virtual void Dispose(bool disposing)
{
if (!_disposedValue)
{
if (disposing)
{
_rawResult?.GetRawResponse()?.Dispose();
}
_disposedValue = true;
}
}
// TODO: Handle disposal via Enumerator? Validate that this will work.
Copy link
Copy Markdown
Collaborator Author

@annelo-msft annelo-msft Mar 21, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is an open question we'll still need to resolve - whether or not to make StreamingClientResult implement IDisposable.

Copy link
Copy Markdown
Collaborator

@trrwilson trrwilson Mar 21, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

From the earlier-era Toub discussions, I think we'll end up needing IDisposable (or have a variant, equivalent mechanism, or whatever else). We looked at the option of implicitly using the end of enumeration as the sole disposal mechanism (and the existing implementation does opportunistically do it at the end of enumeration), but the ergonomics around making that an explicit lifetime requirement were unfavorably received. I think Krzysztof also mentioned that leaving an unbuffered response floating was a substantial issue.

// If it doesn't, we likely need to implement IDisposable or IAsyncDisposable
// on StreamingClientResult<T>.

IAsyncEnumerator<T> IAsyncEnumerable<T>.GetAsyncEnumerator(CancellationToken cancellationToken)
public override IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = default)
=> _asyncEnumerableSource.GetAsyncEnumerator(cancellationToken);
}
}
2 changes: 1 addition & 1 deletion .dotnet/tests/OpenAI.Tests.csproj
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
<LangVersion>latest</LangVersion>
</PropertyGroup>
<ItemGroup>
<ProjectReference Include="..\..\..\azure-sdk-for-net\sdk\core\System.ClientModel\src\System.ClientModel.csproj" />
<ProjectReference Include="..\src\OpenAI.csproj" />
</ItemGroup>
<ItemGroup>
Expand All @@ -14,6 +15,5 @@
<PackageReference Include="NUnit3TestAdapter" Version="4.4.2" />
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.0.0" />
<PackageReference Include="Moq" Version="[4.18.2]" />
<PackageReference Include="System.ClientModel" Version="1.1.0-alpha.20240319.1" />
</ItemGroup>
</Project>
1 change: 1 addition & 0 deletions .dotnet/tests/Samples/Chat/Sample02_StreamingChatAsync.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
using NUnit.Framework;
using OpenAI.Chat;
using System;
using System.ClientModel;
using System.Threading.Tasks;

namespace OpenAI.Samples
Expand Down