Preliminary support for Assistants streaming

.dotnet/scripts/Add-Customizations.ps1

@@ -34,11 +34,15 @@ function Set-LangVersionToLatest {
     $xml.Save($filePath)
 }
 
-function Edit-RunObjectSerialization {
+function Edit-DateTimeOffsetSerialization {
+    param(
+        [string]$filename
+    )
+
     $root = Split-Path $PSScriptRoot -Parent
     $directory = Join-Path -Path $root -ChildPath "src\Generated\Models"
 
-    $file = Get-ChildItem -Path $directory -Filter "RunObject.Serialization.cs"
+    $file = Get-ChildItem -Path $directory -Filter $filename
     $content = Get-Content -Path $file -Raw
 
     Write-Output "Editing $($file.FullName)"
@@ -48,11 +52,13 @@ function Edit-RunObjectSerialization {
     $content = $content -creplace "cancelledAt = property\.Value\.GetDateTimeOffset\(`"O`"\);", "// BUG: https://github.com/Azure/autorest.csharp/issues/4296`r`n                    // cancelledAt = property.Value.GetDateTimeOffset(`"O`");`r`n                    cancelledAt = DateTimeOffset.FromUnixTimeSeconds(property.Value.GetInt64());"
     $content = $content -creplace "failedAt = property\.Value\.GetDateTimeOffset\(`"O`"\);", "// BUG: https://github.com/Azure/autorest.csharp/issues/4296`r`n                    // failedAt = property.Value.GetDateTimeOffset(`"O`");`r`n                    failedAt = DateTimeOffset.FromUnixTimeSeconds(property.Value.GetInt64());"
     $content = $content -creplace "completedAt = property\.Value\.GetDateTimeOffset\(`"O`"\);", "// BUG: https://github.com/Azure/autorest.csharp/issues/4296`r`n                    // completedAt = property.Value.GetDateTimeOffset(`"O`");`r`n                    completedAt = DateTimeOffset.FromUnixTimeSeconds(property.Value.GetInt64());"
+    $content = $content -creplace "incompleteAt = property\.Value\.GetDateTimeOffset\(`"O`"\);", "// BUG: https://github.com/Azure/autorest.csharp/issues/4296`r`n                    // completedAt = property.Value.GetDateTimeOffset(`"O`");`r`n                    completedAt = DateTimeOffset.FromUnixTimeSeconds(property.Value.GetInt64());"
 
     $content | Set-Content -Path $file.FullName -NoNewline
 }
 
 Update-SystemTextJsonPackage
 Update-MicrosoftBclAsyncInterfacesPackage
 Set-LangVersionToLatest
-Edit-RunObjectSerialization
+Edit-DateTimeOffsetSerialization -filename "RunObject.Serialization.cs"
+Edit-DateTimeOffsetSerialization -filename "MessageObject.Serialization.cs"

.dotnet/src/Custom/Assistants/AssistantClient.Protocol.cs

@@ -1,6 +1,8 @@
+using System;
 using System.ClientModel;
 using System.ClientModel.Primitives;
 using System.ComponentModel;
+using System.Text;
 using System.Threading.Tasks;
 
 namespace OpenAI.Assistants;
@@ -318,6 +320,15 @@ public virtual ClientResult CreateRun(
         RequestOptions options = null)
         => RunShim.CreateRun(threadId, content, options);
 
+    /// <inheritdoc cref="Internal.Runs.CreateRun(string, BinaryContent, RequestOptions)"/>
+    [EditorBrowsable(EditorBrowsableState.Never)]
+    public virtual ClientResult CreateRunStreaming(string threadId, BinaryContent content, RequestOptions options = null)
+    {
+        PipelineMessage message = CreateCreateRunRequest(threadId, content, stream: true, options);
+        RunShim.Pipeline.Send(message);
+        return ClientResult.FromResponse(message.ExtractResponse());
+    }
+
     /// <inheritdoc cref="Internal.Runs.CreateRunAsync(string, BinaryContent, RequestOptions)"/>
     [EditorBrowsable(EditorBrowsableState.Never)]
     public virtual async Task<ClientResult> CreateRunAsync(
@@ -326,20 +337,47 @@ public virtual async Task<ClientResult> CreateRunAsync(
         RequestOptions options = null)
         => await RunShim.CreateRunAsync(threadId, content, options).ConfigureAwait(false);
 
+    /// <inheritdoc cref="Internal.Runs.CreateRunAsync(string, BinaryContent, RequestOptions)"/>
+    [EditorBrowsable(EditorBrowsableState.Never)]
+    public virtual async Task<ClientResult> CreateRunStreamingAsync(string threadId, BinaryContent content, RequestOptions options = null)
+    {
+        PipelineMessage message = CreateCreateRunRequest(threadId, content, stream: true, options);
+        await RunShim.Pipeline.SendAsync(message);
+        return ClientResult.FromResponse(message.ExtractResponse());
+    }
+
     /// <inheritdoc cref="Internal.Runs.CreateThreadAndRun(BinaryContent, RequestOptions)"/>
     [EditorBrowsable(EditorBrowsableState.Never)]
     public virtual ClientResult CreateThreadAndRun(
         BinaryContent content,
         RequestOptions options = null)
         => RunShim.CreateThreadAndRun(content, options);
 
+    /// <inheritdoc cref="Internal.Runs.CreateThreadAndRun(BinaryContent, RequestOptions)"/>
+    [EditorBrowsable(EditorBrowsableState.Never)]
+    public virtual ClientResult CreateThreadAndRunStreaming(BinaryContent content, RequestOptions options = null)
+    {
+        PipelineMessage message = CreateCreateThreadAndRunRequest(content, stream: true, options);
+        RunShim.Pipeline.Send(message);
+        return ClientResult.FromResponse(message.ExtractResponse());
+    }
+
     /// <inheritdoc cref="Internal.Runs.CreateThreadAndRunAsync(BinaryContent, RequestOptions)"/>
     [EditorBrowsable(EditorBrowsableState.Never)]
     public virtual async Task<ClientResult> CreateThreadAndRunAsync(
         BinaryContent content,
         RequestOptions options = null)
         => await RunShim.CreateThreadAndRunAsync(content, options).ConfigureAwait(false);
 
+    /// <inheritdoc cref="Internal.Runs.CreateThreadAndRun(BinaryContent, RequestOptions)"/>
+    [EditorBrowsable(EditorBrowsableState.Never)]
+    public virtual async Task<ClientResult> CreateThreadAndRunStreamingAsync(BinaryContent content, RequestOptions options = null)
+    {
+        PipelineMessage message = CreateCreateThreadAndRunRequest(content, stream: true, options);
+        await RunShim.Pipeline.SendAsync(message);
+        return ClientResult.FromResponse(message.ExtractResponse());
+    }
+
     /// <inheritdoc cref="Internal.Runs.GetRun(string, string, RequestOptions)"/>
     [EditorBrowsable(EditorBrowsableState.Never)]
     public virtual ClientResult GetRun(
@@ -421,6 +459,19 @@ public virtual ClientResult SubmitToolOutputs(
         RequestOptions options = null)
         => RunShim.SubmitToolOuputsToRun(threadId, runId, content, options);
 
+    /// <inheritdoc cref="Internal.Runs.SubmitToolOuputsToRun(string, string, BinaryContent, RequestOptions)"/>
+    [EditorBrowsable(EditorBrowsableState.Never)]
+    public virtual ClientResult SubmitToolOutputsStreaming(
+        string threadId,
+        string runId,
+        BinaryContent content,
+        RequestOptions options = null)
+    {
+        PipelineMessage message = CreateSubmitToolOutputsRequest(threadId, runId, content, stream: true, options);
+        RunShim.Pipeline.Send(message);
+        return ClientResult.FromResponse(message.ExtractResponse());
+    }
+
     /// <inheritdoc cref="Internal.Runs.SubmitToolOuputsToRunAsync(string, string, BinaryContent, RequestOptions)"/>
     [EditorBrowsable(EditorBrowsableState.Never)]
     public virtual async Task<ClientResult> SubmitToolOutputsAsync(
@@ -430,6 +481,19 @@ public virtual async Task<ClientResult> SubmitToolOutputsAsync(
         RequestOptions options = null)
         => await RunShim.SubmitToolOuputsToRunAsync(threadId, runId, content, options).ConfigureAwait(false);
 
+    /// <inheritdoc cref="Internal.Runs.SubmitToolOuputsToRunAsync(string, string, BinaryContent, RequestOptions)"/>
+    [EditorBrowsable(EditorBrowsableState.Never)]
+    public virtual async Task<ClientResult> SubmitToolOutputsStreamingAsync(
+        string threadId,
+        string runId,
+        BinaryContent content,
+        RequestOptions options = null)
+    {
+        PipelineMessage message = CreateSubmitToolOutputsRequest(threadId, runId, content, stream: true, options);
+        await RunShim.Pipeline.SendAsync(message);
+        return ClientResult.FromResponse(message.ExtractResponse());
+    }
+
     /// <inheritdoc cref="Internal.Runs.GetRunStep(string, string, string, RequestOptions)"/>
     public virtual ClientResult GetRunStep(
         string threadId,
@@ -467,4 +531,33 @@ public virtual async Task<ClientResult> GetRunStepsAsync(
         string subsequentStepId,
         RequestOptions options)
         => await RunShim.GetRunStepsAsync(threadId, runId, maxResults, createdSortOrder, previousStepId, subsequentStepId, options).ConfigureAwait(false);
+
+    internal PipelineMessage CreateCreateRunRequest(string threadId, BinaryContent content, bool? stream = null, RequestOptions options = null)
+        => CreateAssistantProtocolRequest($"/threads/{threadId}/runs", content, stream, options);
+
+    internal PipelineMessage CreateCreateThreadAndRunRequest(BinaryContent content, bool? stream = null, RequestOptions options = null)
+        => CreateAssistantProtocolRequest($"/threads/runs", content, stream, options);
+
+    internal PipelineMessage CreateSubmitToolOutputsRequest(string threadId, string runId, BinaryContent content, bool? stream = null, RequestOptions options = null)
+        => CreateAssistantProtocolRequest($"/threads/{threadId}/runs/{runId}/submit_tool_outputs", content, stream, options);
+
+    internal PipelineMessage CreateAssistantProtocolRequest(string path, BinaryContent content, bool? stream = null, RequestOptions options = null)
+    {
+        PipelineMessage message = Shim.Pipeline.CreateMessage();
+        message.ResponseClassifier = ResponseErrorClassifier200;
+        if (stream == true)
+        {
+            message.BufferResponse = false;
+        }
+        PipelineRequest request = message.Request;
+        request.Method = "POST";
+        UriBuilder uriBuilder = new(_clientConnector.Endpoint.AbsoluteUri);
+        uriBuilder.Path += path;
+        request.Uri = uriBuilder.Uri;
+        request.Headers.Set("Content-Type", "application/json");
+        request.Headers.Set("Accept", stream == true ? "text/event-stream" : "application/json");
+        request.Content = content;
+        message.Apply(options ?? new());
+        return message;
+    }
 }

.dotnet/src/Custom/Assistants/AssistantClient.cs

@@ -1,8 +1,11 @@
+using OpenAI.Chat;
 using OpenAI.ClientShared.Internal;
+using OpenAI.Internal;
 using System;
 using System.ClientModel;
 using System.ClientModel.Primitives;
 using System.Collections.Generic;
+using System.Text;
 using System.Threading.Tasks;
 
 namespace OpenAI.Assistants;
@@ -506,6 +509,26 @@ public virtual async Task<ClientResult<ThreadRun>> CreateRunAsync(
         return ClientResult.FromValue(new ThreadRun(internalResult.Value), internalResult.GetRawResponse());
     }
 
+    public virtual StreamingClientResult<StreamingUpdate> CreateRunStreaming(
+        string threadId,
+        string assistantId,
+        RunCreationOptions options = null)
+    {
+        PipelineMessage message = CreateCreateRunRequest(threadId, assistantId, options, stream: true);
+        RunShim.Pipeline.Send(message);
+        return CreateStreamingRunResult(message);
+    }
+
+    public virtual async Task<StreamingClientResult<StreamingUpdate>> CreateRunStreamingAsync(
+        string threadId,
+        string assistantId,
+        RunCreationOptions options = null)
+    {
+        PipelineMessage message = CreateCreateRunRequest(threadId, assistantId, options, stream: true);
+        await RunShim.Pipeline.SendAsync(message);
+        return CreateStreamingRunResult(message);
+    }
+
     public virtual ClientResult<ThreadRun> CreateThreadAndRun(
         string assistantId,
         ThreadCreationOptions threadOptions = null,
@@ -529,6 +552,26 @@ Internal.Models.CreateThreadAndRunRequest request
         return ClientResult.FromValue(new ThreadRun(internalResult.Value), internalResult.GetRawResponse());
     }
 
+    public virtual StreamingClientResult<StreamingUpdate> CreateThreadAndRunStreaming(
+        string assistantId,
+        ThreadCreationOptions threadOptions = null,
+        RunCreationOptions runOptions = null)
+    {
+        PipelineMessage message = CreateCreateThreadAndRunRequest(assistantId, threadOptions, runOptions, stream: true);
+        Shim.Pipeline.Send(message);
+        return CreateStreamingRunResult(message);
+    }
+
+    public virtual async Task<StreamingClientResult<StreamingUpdate>> CreateThreadAndRunStreamingAsync(
+        string assistantId,
+        ThreadCreationOptions threadOptions = null,
+        RunCreationOptions runOptions = null)
+    {
+        PipelineMessage message = CreateCreateThreadAndRunRequest(assistantId, threadOptions, runOptions, stream: true);
+        await Shim.Pipeline.SendAsync(message);
+        return CreateStreamingRunResult(message);
+    }
+
     public virtual ClientResult<ThreadRun> GetRun(string threadId, string runId)
     {
         ClientResult<Internal.Models.RunObject> internalResult = RunShim.GetRun(threadId, runId);
@@ -611,7 +654,7 @@ public virtual ClientResult<ThreadRun> SubmitToolOutputs(string threadId, string
             requestToolOutputs.Add(new(toolOutput.Id, toolOutput.Output, null));
         }
 
-        Internal.Models.SubmitToolOutputsRunRequest request = new(requestToolOutputs, null);
+        Internal.Models.SubmitToolOutputsRunRequest request = new(requestToolOutputs, null, serializedAdditionalRawData: null);
         ClientResult<Internal.Models.RunObject> internalResult = RunShim.SubmitToolOuputsToRun(threadId, runId, request);
         return ClientResult.FromValue(new ThreadRun(internalResult.Value), internalResult.GetRawResponse());
     }
@@ -625,11 +668,81 @@ public virtual async Task<ClientResult<ThreadRun>> SubmitToolOutputsAsync(string
             requestToolOutputs.Add(new(toolOutput.Id, toolOutput.Output, null));
         }
 
-        Internal.Models.SubmitToolOutputsRunRequest request = new(requestToolOutputs, null);
+        Internal.Models.SubmitToolOutputsRunRequest request = new(requestToolOutputs, null, serializedAdditionalRawData: null);
         ClientResult<Internal.Models.RunObject> internalResult = await RunShim.SubmitToolOuputsToRunAsync(threadId, runId, request).ConfigureAwait(false);
         return ClientResult.FromValue(new ThreadRun(internalResult.Value), internalResult.GetRawResponse());
     }
 
+    public virtual StreamingClientResult<StreamingUpdate> SubmitToolOutputsStreaming(string threadId, string runId, IEnumerable<ToolOutput> toolOutputs)
+    {
+        PipelineMessage message = CreateSubmitToolOutputsRequest(threadId, runId, toolOutputs, stream: true);
+        Shim.Pipeline.SendAsync(message);
+        return CreateStreamingRunResult(message);
+    }
+
+    public virtual async Task<StreamingClientResult<StreamingUpdate>> SubmitToolOutputsStreamingAsync(string threadId, string runId, IEnumerable<ToolOutput> toolOutputs)
+    {
+        PipelineMessage message = CreateSubmitToolOutputsRequest(threadId, runId, toolOutputs, stream: true);
+        await Shim.Pipeline.SendAsync(message);
+        return CreateStreamingRunResult(message);
+    }
+
+    internal PipelineMessage CreateCreateRunRequest(string threadId, string assistantId, RunCreationOptions runOptions, bool? stream = null)
+    {
+        Internal.Models.CreateRunRequest internalCreateRunRequest = CreateInternalCreateRunRequest(assistantId, runOptions, stream);
+        BinaryContent requestBody = BinaryContent.Create(internalCreateRunRequest);
+        return CreateCreateRunRequest(threadId, requestBody, stream: true);
+    }
+
+    internal PipelineMessage CreateCreateThreadAndRunRequest(
+        string assistantId,
+        ThreadCreationOptions threadOptions,
+        RunCreationOptions runOptions,
+        bool? stream = null)
+    {
+        Internal.Models.CreateThreadAndRunRequest internalRequest
+            = CreateInternalCreateThreadAndRunRequest(assistantId, threadOptions, runOptions, stream: true);
+        BinaryContent content = BinaryContent.Create(internalRequest);
+        return CreateCreateThreadAndRunRequest(content, stream: true);
+    }
+
+    internal PipelineMessage CreateSubmitToolOutputsRequest(string threadId, string runId, IEnumerable<ToolOutput> toolOutputs, bool? stream)
+    {
+        List<Internal.Models.SubmitToolOutputsRunRequestToolOutput> requestToolOutputs = [];
+        foreach (ToolOutput toolOutput in toolOutputs)
+        {
+            requestToolOutputs.Add(new(toolOutput.Id, toolOutput.Output, null));
+        }
+        Internal.Models.SubmitToolOutputsRunRequest internalRequest = new(requestToolOutputs, stream, serializedAdditionalRawData: null);
+        BinaryContent content = BinaryContent.Create(internalRequest);
+        return CreateSubmitToolOutputsRequest(threadId, runId, content, stream: true);
+    }
+
+    internal static StreamingClientResult<StreamingUpdate> CreateStreamingRunResult(PipelineMessage message)
+    {
+        if (message.Response.IsError)
+        {
+            throw new ClientResultException(message.Response);
+        }
+        PipelineResponse response = null;
+        try
+        {
+            response = message.ExtractResponse();
+            ClientResult genericResult = ClientResult.FromResponse(response);
+            StreamingClientResult<StreamingUpdate> streamingResult = StreamingClientResult<StreamingUpdate>.CreateFromResponse(
+                genericResult,
+                (responseForEnumeration) => SseAsyncEnumerator<StreamingUpdate>.EnumerateFromSseJsonStream(
+                    responseForEnumeration.GetRawResponse().ContentStream,
+                    StreamingUpdate.DeserializeSseRunUpdates));
+            response = null;
+            return streamingResult;
+        }
+        finally
+        {
+            response?.Dispose();
+        }
+    }
+
     internal static Internal.Models.CreateAssistantRequest CreateInternalCreateAssistantRequest(
         string modelName,
         AssistantCreationOptions options)
@@ -670,7 +783,8 @@ internal static Internal.Models.CreateThreadRequest CreateInternalCreateThreadRe
 
     internal static Internal.Models.CreateRunRequest CreateInternalCreateRunRequest(
         string assistantId,
-        RunCreationOptions options = null)
+        RunCreationOptions options = null,
+        bool? stream = null)
     {
         options ??= new();
         return new(
@@ -680,13 +794,15 @@ internal static Internal.Models.CreateRunRequest CreateInternalCreateRunRequest(
             options.AdditionalInstructions,
             ToInternalBinaryDataList(options.OverrideTools),
             options.Metadata,
+            stream,
             serializedAdditionalRawData: null);
     }
 
     internal static Internal.Models.CreateThreadAndRunRequest CreateInternalCreateThreadAndRunRequest(
         string assistantId,
         ThreadCreationOptions threadOptions,
-        RunCreationOptions runOptions)
+        RunCreationOptions runOptions,
+        bool? stream = null)
     {
         threadOptions ??= new();
         runOptions ??= new();
@@ -698,6 +814,7 @@ internal static Internal.Models.CreateThreadAndRunRequest CreateInternalCreateTh
             runOptions.OverrideInstructions,
             ToInternalBinaryDataList(runOptions?.OverrideTools),
             runOptions?.Metadata,
+            stream,
             serializedAdditionalRawData: null);
     }
 
@@ -766,4 +883,7 @@ internal virtual async Task<ClientResult<ListQueryPage<T>>> GetListQueryPageAsyn
         ListQueryPage<T> convertedValue = ListQueryPage.Create(internalResult.Value) as ListQueryPage<T>;
         return ClientResult.FromValue(convertedValue, internalResult.GetRawResponse());
     }
+
+    private static PipelineMessageClassifier _responseErrorClassifier200;
+    private static PipelineMessageClassifier ResponseErrorClassifier200 => _responseErrorClassifier200 ??= PipelineMessageClassifier.Create(stackalloc ushort[] { 200 });
 }

.dotnet/src/Custom/Assistants/FunctionToolDefinition.cs

@@ -82,7 +82,8 @@ internal override void WriteDerived(Utf8JsonWriter writer, ModelReaderWriterOpti
         if (Optional.IsDefined(Parameters))
         {
             writer.WritePropertyName("parameters"u8);
-            writer.WriteRawValue(Parameters.ToString());
+            using JsonDocument parametersJson = JsonDocument.Parse(Parameters);
+            parametersJson.WriteTo(writer);
         }
         writer.WriteEndObject();
     }