89623e14c2
Streams POST {baseUrl}/chat/completions over SSE (System.Net.ServerSentEvents, no NuGet) and maps delta.content, delta.reasoning_content / delta.reasoning and delta.tool_calls onto AssistantMessageEvents via AssistantMessageBuilder. Configurable through OpenAIProviderOptions so one extension serves OpenAI, OpenRouter, Ollama and other compatible servers; ships the gpt-5 / gpt-5-mini / gpt-4.1 catalogue.
667 lines
27 KiB
C#
667 lines
27 KiB
C#
using System.Net.Http.Headers;
|
|
using System.Net.ServerSentEvents;
|
|
using System.Runtime.CompilerServices;
|
|
using System.Text;
|
|
using System.Text.Encodings.Web;
|
|
using System.Text.Json;
|
|
using System.Text.Json.Nodes;
|
|
using Volcanic.AgenticCore.Abstractions.Json;
|
|
using Volcanic.AgenticCore.Abstractions.Messages;
|
|
using Volcanic.AgenticCore.Abstractions.Providers;
|
|
|
|
namespace Volcanic.AgenticCore.Extensions.OpenAI;
|
|
|
|
/// <summary>
|
|
/// Streams <c>POST {baseUrl}/chat/completions</c> (<c>stream: true</c>) and maps the SSE chunks onto
|
|
/// <see cref="AssistantMessageEvent"/>s. Never throws: HTTP and transport failures become
|
|
/// <see cref="AssistantMessageEvent.Error"/> events.
|
|
/// </summary>
|
|
/// <param name="options">Endpoint configuration.</param>
|
|
/// <param name="http">
|
|
/// Optional client. When supplied, make sure its <see cref="HttpClient.Timeout"/> allows long streams
|
|
/// (the shared default client uses an infinite timeout and relies on <see cref="StreamOptions.Timeout"/> instead).
|
|
/// </param>
|
|
public sealed class OpenAICompletionsApi(OpenAIProviderOptions options, HttpClient? http = null)
|
|
{
|
|
private const string DoneSentinel = "[DONE]";
|
|
|
|
private static readonly HttpClient SharedHttp = new() { Timeout = Timeout.InfiniteTimeSpan };
|
|
|
|
/// <summary>Wire serialisation: relaxed escaping, nulls written verbatim (an assistant turn with only tool calls sends <c>content: null</c>).</summary>
|
|
private static readonly JsonSerializerOptions WireJson = new() { Encoder = JavaScriptEncoder.UnsafeRelaxedJsonEscaping };
|
|
|
|
private static readonly HashSet<string> ContentHeaderNames = new(StringComparer.OrdinalIgnoreCase)
|
|
{
|
|
"Allow", "Content-Disposition", "Content-Encoding", "Content-Language", "Content-Length",
|
|
"Content-Location", "Content-MD5", "Content-Range", "Content-Type", "Expires", "Last-Modified",
|
|
};
|
|
|
|
private readonly HttpClient _http = http ?? SharedHttp;
|
|
|
|
public OpenAIProviderOptions Options { get; } = options;
|
|
|
|
public async IAsyncEnumerable<AssistantMessageEvent> StreamAsync(
|
|
Model model,
|
|
LlmContext context,
|
|
StreamOptions streamOptions,
|
|
[EnumeratorCancellation] CancellationToken cancellationToken)
|
|
{
|
|
var builder = new AssistantMessageBuilder(model);
|
|
|
|
using var timeout = streamOptions.Timeout is { } limit ? new CancellationTokenSource(limit) : null;
|
|
using var linked = timeout is null ? null : CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, timeout.Token);
|
|
var ct = linked?.Token ?? cancellationToken;
|
|
|
|
HttpResponseMessage? response = null;
|
|
Stream? body = null;
|
|
string? failure = null;
|
|
var aborted = false;
|
|
|
|
try
|
|
{
|
|
using var request = await BuildHttpRequestAsync(model, context, streamOptions, ct).ConfigureAwait(false);
|
|
response = await _http.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, ct).ConfigureAwait(false);
|
|
|
|
if (streamOptions.OnResponse is { } onResponse)
|
|
await onResponse(Describe(response)).ConfigureAwait(false);
|
|
|
|
if (response.IsSuccessStatusCode)
|
|
body = await response.Content.ReadAsStreamAsync(ct).ConfigureAwait(false);
|
|
else
|
|
failure = await ReadErrorAsync(response, ct).ConfigureAwait(false);
|
|
}
|
|
catch (Exception) when (cancellationToken.IsCancellationRequested)
|
|
{
|
|
failure = "aborted";
|
|
aborted = true;
|
|
}
|
|
catch (Exception) when (timeout?.IsCancellationRequested == true)
|
|
{
|
|
failure = TimeoutMessage(streamOptions.Timeout!.Value);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
failure = $"Request failed: {ex.Message}";
|
|
}
|
|
|
|
using (response)
|
|
{
|
|
if (failure is not null || body is null)
|
|
{
|
|
yield return builder.Fail(failure ?? "Request failed: empty response body.", aborted);
|
|
yield break;
|
|
}
|
|
|
|
yield return builder.Start();
|
|
|
|
var mapper = new ChunkMapper(model, builder);
|
|
var pending = new List<AssistantMessageEvent>();
|
|
var finished = false;
|
|
|
|
await using var events = SseParser.Create(body).EnumerateAsync(ct).GetAsyncEnumerator(ct);
|
|
while (!finished)
|
|
{
|
|
pending.Clear();
|
|
try
|
|
{
|
|
if (!await events.MoveNextAsync().ConfigureAwait(false))
|
|
{
|
|
finished = true;
|
|
}
|
|
else
|
|
{
|
|
var data = events.Current.Data;
|
|
if (string.Equals(data, DoneSentinel, StringComparison.Ordinal))
|
|
finished = true;
|
|
else if (!string.IsNullOrWhiteSpace(data))
|
|
await mapper.ProcessAsync(data, streamOptions.OnProviderStreamEvent, pending).ConfigureAwait(false);
|
|
}
|
|
}
|
|
catch (Exception) when (cancellationToken.IsCancellationRequested)
|
|
{
|
|
failure = "aborted";
|
|
aborted = true;
|
|
}
|
|
catch (Exception) when (timeout?.IsCancellationRequested == true)
|
|
{
|
|
failure = TimeoutMessage(streamOptions.Timeout!.Value);
|
|
}
|
|
catch (JsonException ex)
|
|
{
|
|
failure = $"Malformed stream chunk: {ex.Message}";
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
failure = $"Stream failed: {ex.Message}";
|
|
}
|
|
|
|
foreach (var e in pending) yield return e;
|
|
|
|
failure ??= mapper.Failure;
|
|
if (failure is not null)
|
|
{
|
|
yield return builder.Fail(failure, aborted);
|
|
yield break;
|
|
}
|
|
}
|
|
|
|
pending.Clear();
|
|
mapper.Finish(pending);
|
|
foreach (var e in pending) yield return e;
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------- request
|
|
|
|
/// <summary>Builds the JSON body for one chat completions call. Exposed for tests and for <see cref="StreamOptions.OnPayload"/> consumers.</summary>
|
|
public static JsonObject BuildRequest(OpenAIProviderOptions options, Model model, LlmContext context, StreamOptions streamOptions)
|
|
{
|
|
ArgumentNullException.ThrowIfNull(options);
|
|
ArgumentNullException.ThrowIfNull(model);
|
|
ArgumentNullException.ThrowIfNull(context);
|
|
ArgumentNullException.ThrowIfNull(streamOptions);
|
|
|
|
var reasoningLevel = model.Reasoning && streamOptions.Reasoning is { } level && level != ThinkingLevel.Off ? level : (ThinkingLevel?)null;
|
|
|
|
var body = new JsonObject
|
|
{
|
|
["model"] = model.Id,
|
|
["stream"] = true,
|
|
["stream_options"] = new JsonObject { ["include_usage"] = true },
|
|
[options.MaxTokensField] = streamOptions.MaxTokens ?? model.MaxTokens,
|
|
};
|
|
|
|
// Reasoning models reject sampling parameters.
|
|
if (reasoningLevel is null && streamOptions.Temperature is { } temperature)
|
|
body["temperature"] = temperature;
|
|
|
|
body["messages"] = BuildMessages(options, context);
|
|
|
|
if (context.Tools.Count > 0)
|
|
body["tools"] = BuildTools(context.Tools);
|
|
|
|
if (reasoningLevel is { } active && options.SendReasoningEffort && ResolveReasoningEffort(model, active) is { } effort)
|
|
body["reasoning_effort"] = effort;
|
|
|
|
if (!string.IsNullOrEmpty(streamOptions.SessionId))
|
|
body["prompt_cache_key"] = streamOptions.SessionId;
|
|
|
|
return body;
|
|
}
|
|
|
|
/// <summary>Provider-neutral level to <c>reasoning_effort</c>. Uses the model's map when present (falling back to the nearest mapped level), else a generic table.</summary>
|
|
public static string? ResolveReasoningEffort(Model model, ThinkingLevel level)
|
|
{
|
|
if (level == ThinkingLevel.Off) return null;
|
|
|
|
var map = model.ThinkingLevelMap;
|
|
if (map is null) return DefaultEffort(level);
|
|
|
|
if (map.TryGetValue(level, out var mapped) && mapped is not null) return mapped;
|
|
|
|
// Mirror ThinkingLevels.Clamp: nearest supported level, preferring higher.
|
|
var order = Enum.GetValues<ThinkingLevel>();
|
|
var index = Array.IndexOf(order, level);
|
|
for (var i = index + 1; i < order.Length; i++)
|
|
if (map.TryGetValue(order[i], out var higher) && higher is not null) return higher;
|
|
for (var i = index - 1; i > 0; i--)
|
|
if (map.TryGetValue(order[i], out var lower) && lower is not null) return lower;
|
|
return null;
|
|
}
|
|
|
|
private static string DefaultEffort(ThinkingLevel level) => level switch
|
|
{
|
|
ThinkingLevel.Minimal or ThinkingLevel.Low => "low",
|
|
ThinkingLevel.Medium => "medium",
|
|
_ => "high",
|
|
};
|
|
|
|
private static JsonArray BuildMessages(OpenAIProviderOptions options, LlmContext context)
|
|
{
|
|
var messages = new JsonArray();
|
|
|
|
if (!string.IsNullOrWhiteSpace(context.SystemPrompt))
|
|
{
|
|
messages.Add(new JsonObject
|
|
{
|
|
["role"] = options.SupportsDeveloperRole ? "developer" : "system",
|
|
["content"] = context.SystemPrompt,
|
|
});
|
|
}
|
|
|
|
foreach (var message in context.Messages)
|
|
{
|
|
switch (message)
|
|
{
|
|
case UserMessage user:
|
|
messages.Add(UserNode(user.Content));
|
|
break;
|
|
case CustomMessage custom:
|
|
messages.Add(UserNode(custom.Content));
|
|
break;
|
|
case AssistantMessage assistant:
|
|
if (AssistantNode(assistant) is { } node) messages.Add(node);
|
|
break;
|
|
case ToolResultMessage result:
|
|
messages.Add(ToolResultNode(result));
|
|
break;
|
|
}
|
|
}
|
|
|
|
return messages;
|
|
}
|
|
|
|
private static JsonObject UserNode(IReadOnlyList<ContentBlock> content)
|
|
{
|
|
var hasImages = content.OfType<ImageContent>().Any();
|
|
if (!hasImages)
|
|
{
|
|
var text = string.Concat(content.OfType<TextContent>().Select(t => t.Text));
|
|
return new JsonObject { ["role"] = "user", ["content"] = text };
|
|
}
|
|
|
|
var parts = new JsonArray();
|
|
foreach (var block in content)
|
|
{
|
|
switch (block)
|
|
{
|
|
case TextContent { Text.Length: > 0 } t:
|
|
parts.Add(new JsonObject { ["type"] = "text", ["text"] = t.Text });
|
|
break;
|
|
case ImageContent image:
|
|
parts.Add(new JsonObject
|
|
{
|
|
["type"] = "image_url",
|
|
["image_url"] = new JsonObject { ["url"] = $"data:{image.MimeType};base64,{image.Data}" },
|
|
});
|
|
break;
|
|
}
|
|
}
|
|
return new JsonObject { ["role"] = "user", ["content"] = parts };
|
|
}
|
|
|
|
/// <summary>Thinking blocks are never replayed; an assistant turn with neither text nor tool calls is skipped.</summary>
|
|
private static JsonObject? AssistantNode(AssistantMessage assistant)
|
|
{
|
|
var text = string.Concat(assistant.Content.OfType<TextContent>().Select(t => t.Text));
|
|
var calls = assistant.Content.OfType<ToolCallContent>().ToList();
|
|
if (text.Length == 0 && calls.Count == 0) return null;
|
|
|
|
var node = new JsonObject
|
|
{
|
|
["role"] = "assistant",
|
|
["content"] = text.Length > 0 ? text : null,
|
|
};
|
|
|
|
if (calls.Count > 0)
|
|
{
|
|
var toolCalls = new JsonArray();
|
|
foreach (var call in calls)
|
|
{
|
|
toolCalls.Add(new JsonObject
|
|
{
|
|
["id"] = call.Id,
|
|
["type"] = "function",
|
|
["function"] = new JsonObject
|
|
{
|
|
["name"] = call.Name,
|
|
["arguments"] = call.Arguments.ValueKind is JsonValueKind.Undefined or JsonValueKind.Null ? "{}" : call.Arguments.GetRawText(),
|
|
},
|
|
});
|
|
}
|
|
node["tool_calls"] = toolCalls;
|
|
}
|
|
|
|
return node;
|
|
}
|
|
|
|
private static JsonObject ToolResultNode(ToolResultMessage result)
|
|
{
|
|
var parts = new List<string>();
|
|
foreach (var block in result.Content)
|
|
{
|
|
switch (block)
|
|
{
|
|
case TextContent t: parts.Add(t.Text); break;
|
|
case ImageContent: parts.Add("[image]"); break;
|
|
}
|
|
}
|
|
return new JsonObject
|
|
{
|
|
["role"] = "tool",
|
|
["tool_call_id"] = result.ToolCallId,
|
|
["content"] = string.Join("\n", parts),
|
|
};
|
|
}
|
|
|
|
private static JsonArray BuildTools(IReadOnlyList<ToolDescriptor> tools)
|
|
{
|
|
var array = new JsonArray();
|
|
foreach (var tool in tools)
|
|
{
|
|
array.Add(new JsonObject
|
|
{
|
|
["type"] = "function",
|
|
["function"] = new JsonObject
|
|
{
|
|
["name"] = tool.Name,
|
|
["description"] = tool.Description,
|
|
["parameters"] = tool.Parameters.ValueKind == JsonValueKind.Object
|
|
? JsonNode.Parse(tool.Parameters.GetRawText())
|
|
: new JsonObject { ["type"] = "object", ["properties"] = new JsonObject() },
|
|
},
|
|
});
|
|
}
|
|
return array;
|
|
}
|
|
|
|
private async Task<HttpRequestMessage> BuildHttpRequestAsync(Model model, LlmContext context, StreamOptions streamOptions, CancellationToken ct)
|
|
{
|
|
JsonNode payload = BuildRequest(Options, model, context, streamOptions);
|
|
if (streamOptions.OnPayload is { } onPayload)
|
|
payload = await onPayload(payload).ConfigureAwait(false) ?? payload;
|
|
|
|
var baseUrl = string.IsNullOrEmpty(model.BaseUrl) ? Options.BaseUrl : model.BaseUrl;
|
|
var request = new HttpRequestMessage(HttpMethod.Post, $"{baseUrl.TrimEnd('/')}/chat/completions")
|
|
{
|
|
Content = new StringContent(payload.ToJsonString(WireJson), Encoding.UTF8, "application/json"),
|
|
};
|
|
request.Headers.Accept.Add(new MediaTypeWithQualityHeaderValue("text/event-stream"));
|
|
if (!string.IsNullOrEmpty(streamOptions.ApiKey))
|
|
request.Headers.Authorization = new AuthenticationHeaderValue("Bearer", streamOptions.ApiKey);
|
|
|
|
foreach (var (name, value) in MergeHeaders(Options.Headers, model.Headers, streamOptions.Headers))
|
|
SetHeader(request, name, value);
|
|
|
|
ct.ThrowIfCancellationRequested();
|
|
return request;
|
|
}
|
|
|
|
/// <summary>Provider, then model, then per-request headers; a <c>null</c> request value removes the header.</summary>
|
|
private static Dictionary<string, string?> MergeHeaders(
|
|
IReadOnlyDictionary<string, string>? providerHeaders,
|
|
IReadOnlyDictionary<string, string>? modelHeaders,
|
|
IReadOnlyDictionary<string, string?>? requestHeaders)
|
|
{
|
|
var merged = new Dictionary<string, string?>(StringComparer.OrdinalIgnoreCase);
|
|
if (providerHeaders is not null) foreach (var (k, v) in providerHeaders) merged[k] = v;
|
|
if (modelHeaders is not null) foreach (var (k, v) in modelHeaders) merged[k] = v;
|
|
if (requestHeaders is not null) foreach (var (k, v) in requestHeaders) merged[k] = v;
|
|
return merged;
|
|
}
|
|
|
|
private static void SetHeader(HttpRequestMessage request, string name, string? value)
|
|
{
|
|
HttpHeaders target = ContentHeaderNames.Contains(name) && request.Content is { } content ? content.Headers : request.Headers;
|
|
target.Remove(name);
|
|
if (value is not null) target.TryAddWithoutValidation(name, value);
|
|
}
|
|
|
|
// ---------------------------------------------------------------- response
|
|
|
|
private static ProviderResponseInfo Describe(HttpResponseMessage response)
|
|
{
|
|
var headers = new Dictionary<string, string>(StringComparer.OrdinalIgnoreCase);
|
|
foreach (var (name, values) in response.Headers) headers[name] = string.Join(", ", values);
|
|
foreach (var (name, values) in response.Content.Headers) headers[name] = string.Join(", ", values);
|
|
return new ProviderResponseInfo((int)response.StatusCode, headers);
|
|
}
|
|
|
|
private static async Task<string> ReadErrorAsync(HttpResponseMessage response, CancellationToken ct)
|
|
{
|
|
var body = "";
|
|
try { body = await response.Content.ReadAsStringAsync(ct).ConfigureAwait(false); }
|
|
catch (Exception) when (!ct.IsCancellationRequested) { /* body unavailable; fall back to the status line */ }
|
|
|
|
var detail = ExtractErrorMessage(body) ?? Excerpt(body) ?? response.ReasonPhrase ?? "request failed";
|
|
return $"HTTP {(int)response.StatusCode}: {detail}";
|
|
}
|
|
|
|
/// <summary><c>{"error":{"message":...}}</c>, <c>{"error":"..."}</c> or <c>{"message":...}</c>.</summary>
|
|
private static string? ExtractErrorMessage(string body)
|
|
{
|
|
if (string.IsNullOrWhiteSpace(body)) return null;
|
|
try
|
|
{
|
|
using var doc = JsonDocument.Parse(body);
|
|
var root = doc.RootElement;
|
|
if (root.ValueKind != JsonValueKind.Object) return null;
|
|
if (root.TryGetProperty("error", out var error))
|
|
{
|
|
if (error.ValueKind == JsonValueKind.String) return error.GetString();
|
|
if (error.ValueKind == JsonValueKind.Object && TryGetString(error, "message", out var message)) return message;
|
|
}
|
|
return TryGetString(root, "message", out var top) ? top : null;
|
|
}
|
|
catch (JsonException)
|
|
{
|
|
return null;
|
|
}
|
|
}
|
|
|
|
private static string? Excerpt(string body)
|
|
{
|
|
var trimmed = body.Trim();
|
|
if (trimmed.Length == 0) return null;
|
|
return trimmed.Length <= 200 ? trimmed : trimmed[..200] + "...";
|
|
}
|
|
|
|
private static string TimeoutMessage(TimeSpan limit) => $"Request timed out after {limit.TotalSeconds:0.###}s";
|
|
|
|
private static bool TryGetString(JsonElement element, string name, out string value)
|
|
{
|
|
if (element.ValueKind == JsonValueKind.Object && element.TryGetProperty(name, out var property) && property.ValueKind == JsonValueKind.String)
|
|
{
|
|
value = property.GetString()!;
|
|
return true;
|
|
}
|
|
value = "";
|
|
return false;
|
|
}
|
|
|
|
private static int GetInt(JsonElement element, string name) =>
|
|
element.ValueKind == JsonValueKind.Object && element.TryGetProperty(name, out var property) && property.ValueKind == JsonValueKind.Number && property.TryGetInt32(out var value)
|
|
? value
|
|
: 0;
|
|
|
|
// ---------------------------------------------------------------- chunk mapping
|
|
|
|
/// <summary>Folds parsed SSE chunks into the builder, emitting explicit block boundaries so every consumer sees balanced events.</summary>
|
|
private sealed class ChunkMapper(Model model, AssistantMessageBuilder builder)
|
|
{
|
|
private enum Open { None, Text, Thinking, ToolCall }
|
|
|
|
private Open _open;
|
|
private int? _toolIndex;
|
|
private string? _toolId;
|
|
private int _nextSyntheticIndex;
|
|
private bool _idRecorded;
|
|
private bool _modelRecorded;
|
|
private StopReason? _finish;
|
|
|
|
/// <summary>Set when the stream carried a terminal error (<c>content_filter</c> or an <c>error</c> chunk).</summary>
|
|
public string? Failure { get; private set; }
|
|
|
|
public async ValueTask ProcessAsync(string data, Func<JsonElement, ValueTask>? observer, List<AssistantMessageEvent> events)
|
|
{
|
|
var chunk = VolcanicJson.Parse(data);
|
|
if (observer is not null) await observer(chunk).ConfigureAwait(false);
|
|
if (chunk.ValueKind != JsonValueKind.Object) return;
|
|
|
|
if (chunk.TryGetProperty("error", out var error) && error.ValueKind is JsonValueKind.Object or JsonValueKind.String)
|
|
{
|
|
CloseOpen(events);
|
|
Failure = StreamErrorMessage(error);
|
|
return;
|
|
}
|
|
|
|
if (!_idRecorded && TryGetString(chunk, "id", out var id))
|
|
{
|
|
builder.SetResponseId(id);
|
|
_idRecorded = true;
|
|
}
|
|
if (!_modelRecorded && TryGetString(chunk, "model", out var responseModel))
|
|
{
|
|
builder.SetResponseModel(responseModel);
|
|
_modelRecorded = true;
|
|
}
|
|
|
|
if (chunk.TryGetProperty("usage", out var usage) && usage.ValueKind == JsonValueKind.Object)
|
|
ApplyUsage(usage);
|
|
|
|
if (chunk.TryGetProperty("choices", out var choices) && choices.ValueKind == JsonValueKind.Array)
|
|
{
|
|
foreach (var choice in choices.EnumerateArray())
|
|
{
|
|
if (choice.ValueKind != JsonValueKind.Object) continue;
|
|
// Only the first completion is streamed; ignore n > 1 siblings.
|
|
if (choice.TryGetProperty("index", out var choiceIndex) && choiceIndex.ValueKind == JsonValueKind.Number && choiceIndex.GetInt32() != 0) continue;
|
|
|
|
if (choice.TryGetProperty("delta", out var delta) && delta.ValueKind == JsonValueKind.Object)
|
|
MapDelta(delta, events);
|
|
|
|
if (TryGetString(choice, "finish_reason", out var finishReason))
|
|
MapFinishReason(finishReason, events);
|
|
|
|
if (Failure is not null) return;
|
|
}
|
|
}
|
|
}
|
|
|
|
/// <summary>Closes any open block and emits <see cref="AssistantMessageEvent.Done"/>.</summary>
|
|
public void Finish(List<AssistantMessageEvent> events)
|
|
{
|
|
CloseOpen(events);
|
|
events.Add(builder.Done(_finish ?? StopReason.Stop));
|
|
}
|
|
|
|
private void MapDelta(JsonElement delta, List<AssistantMessageEvent> events)
|
|
{
|
|
// DeepSeek / vLLM use reasoning_content; OpenRouter uses reasoning.
|
|
if ((TryGetString(delta, "reasoning_content", out var thinking) || TryGetString(delta, "reasoning", out thinking)) && thinking.Length > 0)
|
|
{
|
|
EnsureOpen(Open.Thinking, events);
|
|
events.Add(builder.AppendThinking(thinking));
|
|
}
|
|
|
|
if (TryGetString(delta, "content", out var text) && text.Length > 0)
|
|
{
|
|
EnsureOpen(Open.Text, events);
|
|
events.Add(builder.AppendText(text));
|
|
}
|
|
|
|
if (delta.TryGetProperty("tool_calls", out var toolCalls) && toolCalls.ValueKind == JsonValueKind.Array)
|
|
{
|
|
foreach (var call in toolCalls.EnumerateArray())
|
|
if (call.ValueKind == JsonValueKind.Object) MapToolCall(call, events);
|
|
}
|
|
}
|
|
|
|
private void MapToolCall(JsonElement call, List<AssistantMessageEvent> events)
|
|
{
|
|
var function = call.TryGetProperty("function", out var f) && f.ValueKind == JsonValueKind.Object ? f : default;
|
|
TryGetString(call, "id", out var id);
|
|
|
|
int index;
|
|
if (call.TryGetProperty("index", out var indexProperty) && indexProperty.ValueKind == JsonValueKind.Number && indexProperty.TryGetInt32(out var explicitIndex))
|
|
{
|
|
index = explicitIndex;
|
|
}
|
|
else
|
|
{
|
|
// No index (older servers): a fresh id starts a new call, otherwise the fragment continues the open one.
|
|
var continues = _open == Open.ToolCall && (id.Length == 0 || string.Equals(id, _toolId, StringComparison.Ordinal));
|
|
index = continues ? _toolIndex!.Value : _nextSyntheticIndex++;
|
|
}
|
|
|
|
if (_open != Open.ToolCall || _toolIndex != index)
|
|
{
|
|
CloseOpen(events);
|
|
var callId = id.Length > 0 ? id : $"call_{index}";
|
|
var name = TryGetString(function, "name", out var n) ? n : "";
|
|
events.Add(builder.StartToolCall(callId, name));
|
|
_open = Open.ToolCall;
|
|
_toolIndex = index;
|
|
_toolId = callId;
|
|
_nextSyntheticIndex = Math.Max(_nextSyntheticIndex, index + 1);
|
|
}
|
|
|
|
if (function.ValueKind == JsonValueKind.Object && function.TryGetProperty("arguments", out var arguments))
|
|
{
|
|
var fragment = arguments.ValueKind switch
|
|
{
|
|
JsonValueKind.String => arguments.GetString()!,
|
|
JsonValueKind.Object or JsonValueKind.Array => arguments.GetRawText(),
|
|
_ => "",
|
|
};
|
|
if (fragment.Length > 0) events.Add(builder.AppendToolCallDelta(fragment));
|
|
}
|
|
}
|
|
|
|
private void MapFinishReason(string reason, List<AssistantMessageEvent> events)
|
|
{
|
|
switch (reason)
|
|
{
|
|
case "stop":
|
|
_finish = StopReason.Stop;
|
|
break;
|
|
case "length":
|
|
_finish = StopReason.Length;
|
|
break;
|
|
case "tool_calls":
|
|
case "function_call":
|
|
_finish = StopReason.ToolUse;
|
|
break;
|
|
case "content_filter":
|
|
CloseOpen(events);
|
|
Failure = "content_filter";
|
|
break;
|
|
default:
|
|
_finish ??= StopReason.Stop;
|
|
break;
|
|
}
|
|
}
|
|
|
|
private void ApplyUsage(JsonElement usage)
|
|
{
|
|
var prompt = GetInt(usage, "prompt_tokens");
|
|
var completion = GetInt(usage, "completion_tokens");
|
|
var cached = usage.TryGetProperty("prompt_tokens_details", out var promptDetails) ? GetInt(promptDetails, "cached_tokens") : 0;
|
|
var reasoning = usage.TryGetProperty("completion_tokens_details", out var completionDetails) ? GetInt(completionDetails, "reasoning_tokens") : 0;
|
|
|
|
var tokens = new Usage(Math.Max(0, prompt - cached), completion, cached, 0, reasoning);
|
|
builder.SetUsage(tokens with { Cost = model.CalculateCost(tokens) });
|
|
}
|
|
|
|
private void EnsureOpen(Open kind, List<AssistantMessageEvent> events)
|
|
{
|
|
if (_open == kind) return;
|
|
CloseOpen(events);
|
|
events.Add(kind == Open.Text ? builder.StartText() : builder.StartThinking());
|
|
_open = kind;
|
|
}
|
|
|
|
private void CloseOpen(List<AssistantMessageEvent> events)
|
|
{
|
|
switch (_open)
|
|
{
|
|
case Open.Text: events.Add(builder.EndText()); break;
|
|
case Open.Thinking: events.Add(builder.EndThinking()); break;
|
|
case Open.ToolCall: events.Add(builder.EndToolCall()); break;
|
|
}
|
|
_open = Open.None;
|
|
_toolIndex = null;
|
|
_toolId = null;
|
|
}
|
|
|
|
private static string StreamErrorMessage(JsonElement error)
|
|
{
|
|
if (error.ValueKind == JsonValueKind.String) return error.GetString()!;
|
|
var message = TryGetString(error, "message", out var m) ? m : error.GetRawText();
|
|
if (error.TryGetProperty("code", out var code) && code.ValueKind is JsonValueKind.Number or JsonValueKind.String)
|
|
return $"{message} (code {code.ToString()})";
|
|
return message;
|
|
}
|
|
}
|
|
}
|