From f951bedcc8798098b03b7e2689d6fc1de8a007d0 Mon Sep 17 00:00:00 2001 From: Thorsten Sommer Date: Sun, 20 Sep 2026 09:54:41 +0200 Subject: [PATCH] Stream the tool calling rounds of the Responses API --- .../Provider/BaseProvider.cs | 15 +-- .../Provider/OpenAI/ProviderOpenAI.cs | 29 +++-- .../OpenAI/ResponsesCompletedStreamLine.cs | 13 ++ .../OpenAI/ResponsesStreamAccumulator.cs | 122 ++++++++++++++++++ .../Provider/OpenAI/ResponsesStreamPart.cs | 19 +++ .../OpenAI/ResponsesToolCallingAdapter.cs | 32 +++-- .../Provider/ProviderJsonOptions.cs | 38 ++++++ 7 files changed, 227 insertions(+), 41 deletions(-) create mode 100644 app/MindWork AI Studio/Provider/OpenAI/ResponsesCompletedStreamLine.cs create mode 100644 app/MindWork AI Studio/Provider/OpenAI/ResponsesStreamAccumulator.cs create mode 100644 app/MindWork AI Studio/Provider/OpenAI/ResponsesStreamPart.cs create mode 100644 app/MindWork AI Studio/Provider/ProviderJsonOptions.cs diff --git a/app/MindWork AI Studio/Provider/BaseProvider.cs b/app/MindWork AI Studio/Provider/BaseProvider.cs index 64b895e5..9d4213a0 100644 --- a/app/MindWork AI Studio/Provider/BaseProvider.cs +++ b/app/MindWork AI Studio/Provider/BaseProvider.cs @@ -39,20 +39,7 @@ public abstract class BaseProvider : IProvider, ISecretId /// private readonly ILogger logger; - protected static readonly JsonSerializerOptions JSON_SERIALIZER_OPTIONS = new() - { - PropertyNamingPolicy = JsonNamingPolicy.SnakeCaseLower, - Converters = - { - new JsonStringEnumConverter(JsonNamingPolicy.SnakeCaseLower), - new AnnotationConverter(), - new MessageBaseConverter(), - new SubContentConverter(), - new SubContentImageSourceConverter(), - new SubContentImageUrlConverter(), - }, - AllowTrailingCommas = false - }; + protected static readonly JsonSerializerOptions JSON_SERIALIZER_OPTIONS = ProviderJsonOptions.OPTIONS; /// /// Constructor for the base provider. diff --git a/app/MindWork AI Studio/Provider/OpenAI/ProviderOpenAI.cs b/app/MindWork AI Studio/Provider/OpenAI/ProviderOpenAI.cs index 92c9d959..f6e81b2a 100644 --- a/app/MindWork AI Studio/Provider/OpenAI/ProviderOpenAI.cs +++ b/app/MindWork AI Studio/Provider/OpenAI/ProviderOpenAI.cs @@ -229,7 +229,7 @@ public sealed class ProviderOpenAI() : BaseProvider(LLMProviders.OPEN_AI, new Ur additionalApiParameters, providerTools, runnableTools, - (requestDto, requestToken) => this.ExecuteResponsesRequest(requestDto, requestedSecret, requestToken)); + (requestDto, requestToken) => this.StreamResponsesRequest(requestDto, requestedSecret, requestToken)); var loop = Program.SERVICE_PROVIDER.GetRequiredService(); var loopContext = new ToolCallingLoopContext @@ -316,22 +316,25 @@ public sealed class ProviderOpenAI() : BaseProvider(LLMProviders.OPEN_AI, new Ur yield return content; } - private async Task ExecuteResponsesRequest(ResponsesAPIRequest requestDto, RequestedSecret requestedSecret, CancellationToken token) + /// + /// Runs one round of a tool calling conversation against the Responses API. + /// + /// + /// Nothing but the HTTP request is done here. The retries, the timeouts, and the error + /// classification come from the shared stream reader, which the tool calling rounds used to + /// go without; reading the events is the adapter's business. + /// + private IAsyncEnumerable StreamResponsesRequest(ResponsesAPIRequest requestDto, RequestedSecret requestedSecret, CancellationToken token) { - using var request = new HttpRequestMessage(HttpMethod.Post, "responses"); - request.Headers.Authorization = new AuthenticationHeaderValue("Bearer", await requestedSecret.Secret.Decrypt(Program.ENCRYPTION)); - request.Content = new StringContent(JsonSerializer.Serialize(requestDto, JSON_SERIALIZER_OPTIONS), Encoding.UTF8, "application/json"); + return this.ReadServerSentEventsAsync("OpenAI", "responses call", RequestBuilder, token); - using var response = await this.HttpClient.SendAsync(request, token); - if (!response.IsSuccessStatusCode) + async Task RequestBuilder() { - var responseBody = await response.Content.ReadAsStringAsync(token); - LOGGER.LogError("Tool calling Responses API request failed with status code {ResponseStatusCode} and body: '{ResponseBody}'.", response.StatusCode, responseBody); - await ToolCallingMessages.SendToolCallingRequestFailedAsync((int)response.StatusCode); - return null; + var request = new HttpRequestMessage(HttpMethod.Post, "responses"); + request.Headers.Authorization = new AuthenticationHeaderValue("Bearer", await requestedSecret.Secret.Decrypt(Program.ENCRYPTION)); + request.Content = new StringContent(JsonSerializer.Serialize(requestDto, JSON_SERIALIZER_OPTIONS), Encoding.UTF8, "application/json"); + return request; } - - return await response.Content.ReadFromJsonAsync(JSON_SERIALIZER_OPTIONS, token); } #pragma warning disable CS1998 // Async method lacks 'await' operators and will run synchronously diff --git a/app/MindWork AI Studio/Provider/OpenAI/ResponsesCompletedStreamLine.cs b/app/MindWork AI Studio/Provider/OpenAI/ResponsesCompletedStreamLine.cs new file mode 100644 index 00000000..1591d562 --- /dev/null +++ b/app/MindWork AI Studio/Provider/OpenAI/ResponsesCompletedStreamLine.cs @@ -0,0 +1,13 @@ +namespace AIStudio.Provider.OpenAI; + +/// +/// The closing line of a streamed Responses API call, which repeats the whole response. +/// +/// +/// Everything the round produced comes back here, reasoning items included, in the same shape a +/// non-streamed call would have returned. That is why a streamed tool calling round needs no +/// reassembly: this line is the round. +/// +/// The type of the stream event. +/// The response as a non-streamed call would have returned it. +public sealed record ResponsesCompletedStreamLine(string Type, ResponsesResponse? Response); \ No newline at end of file diff --git a/app/MindWork AI Studio/Provider/OpenAI/ResponsesStreamAccumulator.cs b/app/MindWork AI Studio/Provider/OpenAI/ResponsesStreamAccumulator.cs new file mode 100644 index 00000000..19070c82 --- /dev/null +++ b/app/MindWork AI Studio/Provider/OpenAI/ResponsesStreamAccumulator.cs @@ -0,0 +1,122 @@ +using System.Text.Json; + +namespace AIStudio.Provider.OpenAI; + +/// +/// Reads a streamed Responses API call back into the response the tool calling loop works with. +/// +/// +/// The API repeats the whole response when it is done, reasoning items included, so nothing has +/// to be reassembled from fragments: that closing event is the round. What this type does beyond +/// taking it is hand out text and sources while they arrive, and keep the finished output items +/// as a fallback for gateways which never send that closing event.

+/// No HTTP, no dependency injection, no provider: everything here is a decision about bytes, and +/// those are the decisions worth having a test for. +/// +public sealed class ResponsesStreamAccumulator +{ + private const string EVENT_COMPLETED = "response.completed"; + private const string EVENT_TEXT_DELTA = "response.output_text.delta"; + private const string EVENT_ANNOTATION_ADDED = "response.output_text.annotation.added"; + private const string EVENT_OUTPUT_ITEM_DONE = "response.output_item.done"; + + private readonly List completedOutputItems = []; + private ResponsesResponse? completedResponse; + + /// + /// Takes the next event of the stream and returns what it has to show. + /// + /// The event to read. + /// The text and sources of this event, both empty when it carried neither. + public ResponsesStreamPart Process(ServerSentEvent serverSentEvent) + { + if (serverSentEvent.Data.Length is 0) + return ResponsesStreamPart.Nothing; + + string eventType; + try + { + using var document = JsonDocument.Parse(serverSentEvent.Data); + var root = document.RootElement; + if (root.ValueKind is not JsonValueKind.Object || + !root.TryGetProperty("type", out var typeProperty) || + typeProperty.ValueKind is not JsonValueKind.String) + return ResponsesStreamPart.Nothing; + + eventType = typeProperty.GetString() ?? string.Empty; + + // + // The item is cloned because its document is disposed at the end of this block, and + // an element which outlives its document reads memory that is no longer there. + // + if (eventType is EVENT_OUTPUT_ITEM_DONE && root.TryGetProperty("item", out var outputItem)) + this.completedOutputItems.Add(outputItem.Clone()); + } + catch (JsonException) + { + // A line we cannot read is a line we skip, exactly as the plain text path does: + return ResponsesStreamPart.Nothing; + } + + switch (eventType) + { + case EVENT_COMPLETED: + this.completedResponse = TryDeserialize(serverSentEvent.Data)?.Response ?? this.completedResponse; + return ResponsesStreamPart.Nothing; + + case EVENT_TEXT_DELTA: + var deltaLine = TryDeserialize(serverSentEvent.Data); + if (deltaLine is null || !deltaLine.ContainsContent()) + return ResponsesStreamPart.Nothing; + + return new ResponsesStreamPart(deltaLine.GetContent().Content, []); + + case EVENT_ANNOTATION_ADDED: + var annotationLine = TryDeserialize(serverSentEvent.Data); + if (annotationLine is null || !annotationLine.ContainsSources()) + return ResponsesStreamPart.Nothing; + + return new ResponsesStreamPart(string.Empty, annotationLine.GetSources()); + + default: + return ResponsesStreamPart.Nothing; + } + } + + /// + /// Builds the response of the round from everything the stream said. + /// + /// + /// The response, or null when the stream ended before it said anything usable. Null is how a + /// failed request and a truncated stream look from here, and both end the round. + /// + public ResponsesResponse? Build() + { + if (this.completedResponse is not null) + return this.completedResponse; + + if (this.completedOutputItems.Count is 0) + return null; + + // + // No closing event came, so the round is put back together from the items which did. + // Reasoning items are among them, which is what the next request needs to continue. + // + return new ResponsesResponse + { + Output = [..this.completedOutputItems], + }; + } + + private static T? TryDeserialize(string json) where T : class + { + try + { + return JsonSerializer.Deserialize(json, ProviderJsonOptions.OPTIONS); + } + catch (JsonException) + { + return null; + } + } +} \ No newline at end of file diff --git a/app/MindWork AI Studio/Provider/OpenAI/ResponsesStreamPart.cs b/app/MindWork AI Studio/Provider/OpenAI/ResponsesStreamPart.cs new file mode 100644 index 00000000..fc40da97 --- /dev/null +++ b/app/MindWork AI Studio/Provider/OpenAI/ResponsesStreamPart.cs @@ -0,0 +1,19 @@ +namespace AIStudio.Provider.OpenAI; + +/// +/// What one line of a streamed Responses API call has to show to the user. +/// +/// The text this line carried, empty when it carried none. +/// The sources this line announced, empty when it announced none. +public readonly record struct ResponsesStreamPart(string TextDelta, IList Sources) +{ + /// + /// The part of a line which says nothing to the user, such as a bookkeeping event. + /// + public static ResponsesStreamPart Nothing => new(string.Empty, []); + + /// + /// Whether this part has anything to show at all. + /// + public bool HasContent => this.TextDelta.Length > 0 || this.Sources.Count > 0; +} \ No newline at end of file diff --git a/app/MindWork AI Studio/Provider/OpenAI/ResponsesToolCallingAdapter.cs b/app/MindWork AI Studio/Provider/OpenAI/ResponsesToolCallingAdapter.cs index 132dd5b6..bdd16ff7 100644 --- a/app/MindWork AI Studio/Provider/OpenAI/ResponsesToolCallingAdapter.cs +++ b/app/MindWork AI Studio/Provider/OpenAI/ResponsesToolCallingAdapter.cs @@ -15,7 +15,7 @@ namespace AIStudio.Provider.OpenAI; /// public sealed class ResponsesToolCallingAdapter(Model chatModel, IList baseInput, IDictionary apiParameters, IList providerTools, IReadOnlyList<(ToolDefinition Definition, IToolImplementation Implementation)> runnableTools, - Func> executeRequestAsync) : IToolCallingProviderAdapter + Func> streamRequestAsync) : IToolCallingProviderAdapter { private readonly List internalItems = []; private readonly List recordedRequestTexts = []; @@ -47,31 +47,35 @@ public sealed class ResponsesToolCallingAdapter(Model chatModel, IList b requestInput.AddRange(this.internalItems); - var response = await executeRequestAsync(new ResponsesAPIRequest + var request = new ResponsesAPIRequest { Model = chatModel.Id, Input = requestInput, - Stream = false, + Stream = true, Store = false, Tools = includeTools ? this.effectiveProviderTools : [], AdditionalApiParameters = apiParameters, - }, token); + }; + // + // The text goes out while it is being written, the round only once the stream closed it. + // Sources travel with the text because the API announces them as it cites them. + // + var accumulator = new ResponsesStreamAccumulator(); + await foreach (var serverSentEvent in streamRequestAsync(request, token)) + { + var part = accumulator.Process(serverSentEvent); + if (part.HasContent) + yield return ToolCallingStreamEvent.TextDelta(new ContentStreamChunk(part.TextDelta, part.Sources)); + } + + var response = accumulator.Build(); if (response is null) yield break; this.lastResponse = response; - - // - // The whole round arrives at once for now, so its text goes out as one delta. What the - // loop and the UI see is already the streaming shape; only the pieces are still large. - // - var textOutput = response.GetTextOutput(); - if (!string.IsNullOrEmpty(textOutput)) - yield return ToolCallingStreamEvent.TextDelta(textOutput); - yield return ToolCallingStreamEvent.RoundCompleted(new ToolCallingRound( - textOutput, + response.GetTextOutput(), response.GetFunctionCalls() .Select(call => new ToolCallingRequestedCall( call.CallId ?? string.Empty, diff --git a/app/MindWork AI Studio/Provider/ProviderJsonOptions.cs b/app/MindWork AI Studio/Provider/ProviderJsonOptions.cs new file mode 100644 index 00000000..213a29ab --- /dev/null +++ b/app/MindWork AI Studio/Provider/ProviderJsonOptions.cs @@ -0,0 +1,38 @@ +using System.Text.Json; +using System.Text.Json.Serialization; + +using AIStudio.Provider.Anthropic; +using AIStudio.Provider.OpenAI; + +namespace AIStudio.Provider; + +/// +/// The JSON options every provider request and response is read and written with. +/// +/// +/// They sit outside the provider base class so that the types which interpret a stream can share +/// them without being a provider themselves. Those types are the ones worth testing, and a +/// provider cannot be constructed in a test at all -- it reaches for the service provider in its +/// constructor. Options rebuilt inside a test would be a second set of rules drifting away from +/// the one that actually reads the wire. +/// +public static class ProviderJsonOptions +{ + /// + /// The shared options. + /// + public static readonly JsonSerializerOptions OPTIONS = new() + { + PropertyNamingPolicy = JsonNamingPolicy.SnakeCaseLower, + Converters = + { + new JsonStringEnumConverter(JsonNamingPolicy.SnakeCaseLower), + new AnnotationConverter(), + new MessageBaseConverter(), + new SubContentConverter(), + new SubContentImageSourceConverter(), + new SubContentImageUrlConverter(), + }, + AllowTrailingCommas = false + }; +} \ No newline at end of file