Files
meeting-assistant/MeetingAssistant/Summary/LiteLlmResponsesChatClient.cs
T
codex e192ae7cd8
PR and Push Build/Test / build-and-test (push) Successful in 14m17s
fix: support LiteLLM Responses streaming
2026-07-27 14:44:32 +02:00

836 lines
29 KiB
C#

using System.Net.Http.Headers;
using System.ClientModel;
using System.ClientModel.Primitives;
using System.Text;
using System.Text.Json;
using System.Text.Json.Nodes;
using Microsoft.Agents.AI.Compaction;
using Microsoft.Extensions.AI;
using Microsoft.Extensions.Logging;
using OpenAI;
using OpenAI.Responses;
namespace MeetingAssistant.Summary;
#pragma warning disable MAAI001
#pragma warning disable OPENAI001
public sealed class LiteLlmResponsesChatClient : IChatClient
{
private static readonly JsonSerializerOptions JsonOptions = new(JsonSerializerDefaults.Web);
private readonly HttpClient httpClient;
private readonly ResponsesClient responsesClient;
private readonly AsyncLocal<string?> requestInitiator = new();
private readonly string model;
private readonly bool useStreaming;
private readonly bool enableThinking;
private readonly string reasoningEffort;
private readonly int reconnectionAttempts;
private readonly TimeSpan reconnectionDelay;
private readonly LiteLlmResponsesCompactionOptions? compactionOptions;
private readonly ILogger? logger;
private readonly Action? retrying;
private readonly Action<string>? reasoningSummaryChanged;
private int responsesRequestCount;
public LiteLlmResponsesChatClient(
Uri endpoint,
string apiKey,
string model,
bool enableThinking,
string reasoningEffort,
int reconnectionAttempts,
TimeSpan reconnectionDelay,
LiteLlmResponsesCompactionOptions? compactionOptions = null,
ILogger? logger = null,
bool firstRequestIsUser = true,
Action? retrying = null,
Action<string>? reasoningSummaryChanged = null,
bool useStreaming = true)
: this(
new HttpClient { BaseAddress = NormalizeEndpoint(endpoint) },
apiKey,
model,
enableThinking,
reasoningEffort,
reconnectionAttempts,
reconnectionDelay,
compactionOptions,
logger,
firstRequestIsUser,
retrying,
reasoningSummaryChanged,
useStreaming)
{
}
internal LiteLlmResponsesChatClient(
HttpClient httpClient,
string apiKey,
string model,
bool enableThinking,
string reasoningEffort,
int reconnectionAttempts,
TimeSpan reconnectionDelay,
LiteLlmResponsesCompactionOptions? compactionOptions = null,
ILogger? logger = null,
bool firstRequestIsUser = true,
Action? retrying = null,
Action<string>? reasoningSummaryChanged = null,
bool useStreaming = true)
{
this.httpClient = httpClient;
this.httpClient.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", apiKey);
responsesClient = CreateResponsesClient(httpClient, apiKey, requestInitiator);
this.model = model;
this.useStreaming = useStreaming;
this.enableThinking = enableThinking;
this.reasoningEffort = reasoningEffort;
this.reconnectionAttempts = Math.Max(0, reconnectionAttempts);
this.reconnectionDelay = reconnectionDelay > TimeSpan.Zero ? reconnectionDelay : TimeSpan.Zero;
this.compactionOptions = compactionOptions;
this.logger = logger;
this.retrying = retrying;
this.reasoningSummaryChanged = reasoningSummaryChanged;
responsesRequestCount = firstRequestIsUser ? 0 : 1;
}
public void Dispose()
{
httpClient.Dispose();
}
internal static LiteLlmResponsesChatClient Create(
AgentOptions options,
string apiKey,
LiteLlmResponsesCompactionOptions? compactionOptions,
ILogger? logger,
bool firstRequestIsUser,
Action? retrying = null,
Action<string>? reasoningSummaryChanged = null)
{
return new LiteLlmResponsesChatClient(
new Uri(options.Endpoint),
apiKey,
options.Model,
options.EnableThinking,
options.ReasoningEffort switch
{
ReasoningEffortOption.None => "none",
ReasoningEffortOption.Low => "low",
ReasoningEffortOption.High => "high",
ReasoningEffortOption.ExtraHigh => "xhigh",
_ => "medium"
},
options.ReconnectionAttempts,
options.ReconnectionDelay,
compactionOptions,
logger,
firstRequestIsUser,
retrying,
reasoningSummaryChanged,
options.UseStreaming);
}
public async Task<ChatResponse> GetResponseAsync(
IEnumerable<ChatMessage> messages,
ChatOptions? options = null,
CancellationToken cancellationToken = default)
{
var payload = await CreateCompactedPayloadAsync(messages.ToList(), options, cancellationToken).ConfigureAwait(false);
var result = await PostWithRetryAsync(payload, cancellationToken).ConfigureAwait(false);
ReportVisibleReasoningSummaries(result.ReasoningSummaries);
LogResponseDiagnostics(result.Response);
ThrowIfResponseHasNoContent(result.Response);
LogResponseUsage(result.Response.Usage);
return result.Response;
}
public async IAsyncEnumerable<ChatResponseUpdate> GetStreamingResponseAsync(
IEnumerable<ChatMessage> messages,
ChatOptions? options = null,
[System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
{
var response = await GetResponseAsync(messages, options, cancellationToken).ConfigureAwait(false);
foreach (var message in response.Messages)
{
yield return new ChatResponseUpdate(message.Role, message.Contents);
}
}
public object? GetService(Type serviceType, object? serviceKey = null)
{
if (serviceKey is not null)
{
return null;
}
return serviceType == typeof(ChatClientMetadata)
? new ChatClientMetadata("litellm-responses", httpClient.BaseAddress, model)
: null;
}
private static IReadOnlyList<string> ExtractVisibleReasoningSummaries(ChatResponse response)
{
return response.Messages
.SelectMany(message => message.Contents)
.OfType<TextReasoningContent>()
.Select(content => content.Text)
.Where(text => !string.IsNullOrWhiteSpace(text))
.ToArray();
}
private void ReportVisibleReasoningSummaries(IReadOnlyList<string> summaries)
{
if (reasoningSummaryChanged is null)
{
return;
}
foreach (var summary in summaries)
{
try
{
reasoningSummaryChanged(summary);
}
catch (Exception exception)
{
logger?.LogWarning(
exception,
"Reasoning summary callback failed; continuing with the LiteLLM response.");
}
}
}
private async Task<JsonObject> CreateCompactedPayloadAsync(
IReadOnlyList<ChatMessage> messages,
ChatOptions? options,
CancellationToken cancellationToken)
{
var payload = CreatePayload(messages, options);
var estimate = EstimateTokens(payload);
var triggerTokens = compactionOptions?.TriggerTokens ?? int.MaxValue;
if (compactionOptions?.Enabled != true || estimate < triggerTokens)
{
LogContextWindow(estimate, compacted: false, source: "none");
return payload;
}
LogContextWindow(estimate, compacted: false, source: "threshold");
var remotePayload = await TryCompactWithResponsesEndpointAsync(payload, estimate, cancellationToken)
.ConfigureAwait(false);
if (remotePayload is not null)
{
LogContextWindow(EstimateTokens(remotePayload), compacted: true, source: "responses_compact");
return remotePayload;
}
if (compactionOptions.FallbackStrategy is null)
{
return payload;
}
try
{
var compactedMessages = await CompactionProvider.CompactAsync(
compactionOptions.FallbackStrategy,
messages,
logger ?? Microsoft.Extensions.Logging.Abstractions.NullLogger.Instance,
cancellationToken)
.ConfigureAwait(false);
var fallbackPayload = CreatePayload(compactedMessages, options);
LogContextWindow(EstimateTokens(fallbackPayload), compacted: true, source: "agent_framework");
return fallbackPayload;
}
catch (Exception exception) when (exception is not OperationCanceledException)
{
logger?.LogWarning(
exception,
"Agent Framework compaction fallback failed; sending un-compacted summary request");
return payload;
}
}
private async Task<JsonObject?> TryCompactWithResponsesEndpointAsync(
JsonObject payload,
int estimatedTokens,
CancellationToken cancellationToken)
{
try
{
var request = new JsonObject
{
["model"] = payload["model"]?.DeepClone(),
["input"] = payload["input"]?.DeepClone(),
["instructions"] = payload["instructions"]?.DeepClone(),
["context_window_tokens"] = compactionOptions!.ContextWindowTokens,
["max_output_tokens"] = compactionOptions.MaxOutputTokens,
["estimated_input_tokens"] = estimatedTokens,
["target_input_tokens"] = compactionOptions.TriggerTokens,
["strategy"] = new JsonArray
{
"collapse_tool_results",
"summarize_older_spans",
"keep_last_4_user_turns",
"drop_oldest_groups"
}
};
using var requestMessage = CreateJsonRequest(
NormalizeRelativePath(compactionOptions.CompactPath),
request.ToJsonString(JsonOptions),
"agent");
using var response = await httpClient.SendAsync(requestMessage, cancellationToken).ConfigureAwait(false);
var responseJson = await response.Content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false);
if (!response.IsSuccessStatusCode)
{
logger?.LogWarning(
"Responses compaction endpoint failed with {StatusCode} {ReasonPhrase}: {Body}",
(int)response.StatusCode,
response.ReasonPhrase,
responseJson);
return null;
}
return ApplyCompactedInput(payload, responseJson);
}
catch (Exception exception) when (exception is not OperationCanceledException)
{
logger?.LogWarning(exception, "Responses compaction endpoint failed; falling back to Agent Framework compaction");
return null;
}
}
private static JsonObject? ApplyCompactedInput(JsonObject payload, string responseJson)
{
using var document = JsonDocument.Parse(responseJson);
var root = document.RootElement;
var input = FindCompactedInput(root);
if (input is null)
{
return null;
}
var compactedPayload = (JsonObject)payload.DeepClone();
compactedPayload["input"] = JsonNode.Parse(input.Value.GetRawText());
if (root.TryGetProperty("instructions", out var instructions) && instructions.ValueKind == JsonValueKind.String)
{
compactedPayload["instructions"] = instructions.GetString();
}
return compactedPayload;
}
private static JsonElement? FindCompactedInput(JsonElement root)
{
foreach (var propertyName in new[] { "input", "compacted_input", "messages", "output" })
{
if (root.TryGetProperty(propertyName, out var value) && value.ValueKind == JsonValueKind.Array)
{
return value;
}
}
return null;
}
private JsonObject CreatePayload(IEnumerable<ChatMessage> messages, ChatOptions? options)
{
var instructions = new StringBuilder();
if (!string.IsNullOrWhiteSpace(options?.Instructions))
{
instructions.AppendLine(options.Instructions);
}
var input = new JsonArray();
foreach (var message in messages)
{
AddInputItem(input, instructions, message);
}
var payload = new JsonObject
{
["model"] = options?.ModelId ?? model,
["input"] = input,
["store"] = false,
["tool_choice"] = "auto"
};
if (instructions.Length > 0)
{
payload["instructions"] = instructions.ToString().Trim();
}
if (enableThinking)
{
payload["reasoning"] = new JsonObject
{
["effort"] = reasoningEffort
};
}
if (options?.MaxOutputTokens is > 0)
{
payload["max_output_tokens"] = options.MaxOutputTokens.Value;
}
var tools = CreateTools(options?.Tools);
if (tools.Count > 0)
{
payload["tools"] = tools;
payload["parallel_tool_calls"] = true;
}
return payload;
}
private async Task<ResponsesChatResult> PostWithRetryAsync(
JsonObject payload,
CancellationToken cancellationToken)
{
var payloadJson = payload.ToJsonString(JsonOptions);
Exception? lastException = null;
for (var attempt = 0; attempt <= reconnectionAttempts; attempt++)
{
try
{
var initiator = NextInitiator();
LogRequestDiagnostics(payload, initiator, attempt);
var createOptions = ModelReaderWriter.Read<CreateResponseOptions>(
BinaryData.FromString(payloadJson),
ModelReaderWriterOptions.Json)
?? throw new InvalidOperationException("Unable to create OpenAI Responses request options.");
createOptions.StreamingEnabled = useStreaming;
var previousInitiator = requestInitiator.Value;
requestInitiator.Value = initiator;
try
{
return useStreaming
? await GetStreamingResponseAsync(createOptions, cancellationToken).ConfigureAwait(false)
: CreateNonStreamingResult((await responsesClient
.CreateResponseAsync(createOptions, cancellationToken)
.ConfigureAwait(false))
.Value
.AsChatResponse(createOptions));
}
finally
{
requestInitiator.Value = previousInitiator;
}
}
catch (ClientResultException exception)
when (!IsRetryableStatusCode(exception.Status) || attempt == reconnectionAttempts)
{
throw new InvalidOperationException(
$"LiteLLM Responses request failed with {exception.Status}: {exception.Message}",
exception);
}
catch (Exception exception) when (IsRetryableException(exception) && attempt < reconnectionAttempts)
{
lastException = exception;
}
retrying?.Invoke();
await DelayBeforeRetryAsync(cancellationToken).ConfigureAwait(false);
}
throw lastException ?? new InvalidOperationException("LiteLLM Responses request failed.");
}
private async Task<ResponsesChatResult> GetStreamingResponseAsync(
CreateResponseOptions createOptions,
CancellationToken cancellationToken)
{
var updates = new List<ChatResponseUpdate>();
var reasoningSummaries = new Dictionary<(string ItemId, int SummaryIndex), StringBuilder>();
await foreach (var update in responsesClient
.CreateResponseStreamingAsync(createOptions, cancellationToken)
.AsChatResponseUpdatesAsync(createOptions, cancellationToken)
.ConfigureAwait(false))
{
updates.Add(update);
var reasoningUpdate = update.RawRepresentation
as StreamingResponseReasoningSummaryTextDeltaUpdate;
if (reasoningUpdate is null)
{
continue;
}
var key = (reasoningUpdate.ItemId, reasoningUpdate.SummaryIndex);
if (!reasoningSummaries.TryGetValue(key, out var summary))
{
summary = new StringBuilder();
reasoningSummaries[key] = summary;
}
summary.Append(reasoningUpdate.Delta);
}
return new ResponsesChatResult(
updates.ToChatResponse(),
reasoningSummaries.Values
.Select(summary => summary.ToString())
.Where(summary => !string.IsNullOrWhiteSpace(summary))
.ToArray());
}
private static ResponsesChatResult CreateNonStreamingResult(ChatResponse response)
{
return new ResponsesChatResult(response, ExtractVisibleReasoningSummaries(response));
}
private HttpRequestMessage CreateJsonRequest(
string requestUri,
string payloadJson,
string? initiator = null)
{
var request = new HttpRequestMessage(HttpMethod.Post, requestUri)
{
Content = new StringContent(payloadJson, Encoding.UTF8, "application/json")
};
if (!string.IsNullOrWhiteSpace(initiator))
{
request.Headers.TryAddWithoutValidation("X-Initiator", initiator);
}
return request;
}
private string NextInitiator()
{
return Interlocked.Increment(ref responsesRequestCount) == 1 ? "user" : "agent";
}
private async Task DelayBeforeRetryAsync(CancellationToken cancellationToken)
{
if (reconnectionDelay > TimeSpan.Zero)
{
await Task.Delay(reconnectionDelay, cancellationToken).ConfigureAwait(false);
}
}
private static bool IsRetryableStatusCode(int statusCode)
{
return statusCode is 408 or 409 or 425 or 429
|| statusCode >= 500;
}
private static bool IsRetryableException(Exception exception)
{
return exception is ClientResultException clientException
&& IsRetryableStatusCode(clientException.Status)
|| exception is HttpRequestException or TaskCanceledException;
}
private void LogContextWindow(int estimatedTokens, bool compacted, string source)
{
if (compactionOptions is null || logger is null)
{
return;
}
var contextWindow = Math.Max(1, compactionOptions.ContextWindowTokens);
var remainingTokens = Math.Max(0, contextWindow - estimatedTokens);
logger.LogInformation(
"Summary context estimate: {EstimatedTokens}/{ContextWindowTokens} tokens, {RemainingTokens} remaining, compacted={Compacted}, source={Source}",
estimatedTokens,
contextWindow,
remainingTokens,
compacted,
source);
}
private void LogResponseUsage(UsageDetails? usage)
{
if (usage is null || compactionOptions is null || logger is null)
{
return;
}
var contextWindow = Math.Max(1, compactionOptions.ContextWindowTokens);
var inputTokens = usage.InputTokenCount ?? 0;
var outputTokens = usage.OutputTokenCount ?? 0;
var totalTokens = usage.TotalTokenCount ?? inputTokens + outputTokens;
logger.LogInformation(
"Summary response usage: input={InputTokens}, output={OutputTokens}, total={TotalTokens}, contextWindow={ContextWindowTokens}",
inputTokens,
outputTokens,
totalTokens,
contextWindow);
}
private void LogRequestDiagnostics(JsonObject payload, string initiator, int attempt)
{
if (logger is null)
{
return;
}
var inputCount = payload.TryGetPropertyValue("input", out var input) && input is JsonArray inputArray
? inputArray.Count
: 0;
var tools = payload.TryGetPropertyValue("tools", out var toolsNode) && toolsNode is JsonArray toolsArray
? toolsArray
: [];
var toolNames = tools
.OfType<JsonObject>()
.Select(tool => tool.TryGetPropertyValue("name", out var name) ? name?.GetValue<string>() : null)
.Where(name => !string.IsNullOrWhiteSpace(name))
.ToArray();
var instructionsPreview = payload.TryGetPropertyValue("instructions", out var instructionsNode)
? Truncate(instructionsNode?.GetValue<string>() ?? "", maxLength: 500)
: "";
logger.LogInformation(
"LiteLLM Responses request: initiator={Initiator}, attempt={Attempt}, inputItems={InputItems}, tools={ToolCount}, toolNames={ToolNames}, instructionsPreview={InstructionsPreview}",
initiator,
attempt + 1,
inputCount,
toolNames.Length,
string.Join(", ", toolNames),
instructionsPreview);
}
private void LogResponseDiagnostics(ChatResponse response)
{
if (logger is null)
{
return;
}
var parsedTypes = response.Messages
.SelectMany(message => message.Contents)
.Select(content => content.GetType().Name)
.ToArray();
var outputTypes = GetOutputItemTypes(response);
var functionCalls = response.Messages
.SelectMany(message => message.Contents)
.OfType<FunctionCallContent>()
.Select(call => call.Name)
.ToArray();
logger.LogInformation(
"LiteLLM Responses response: responseId={ResponseId}, outputTypes={OutputTypes}, parsedContents={ParsedContents}, functionCalls={FunctionCalls}, textLength={TextLength}",
response.ResponseId,
string.Join(", ", outputTypes),
string.Join(", ", parsedTypes),
string.Join(", ", functionCalls),
response.Text?.Length ?? 0);
if (parsedTypes.Length == 0)
{
logger.LogWarning(
"LiteLLM Responses response contained no parseable message text or function calls.");
}
}
private static void ThrowIfResponseHasNoContent(ChatResponse response)
{
if (response.Messages.SelectMany(message => message.Contents).Any())
{
return;
}
var outputTypes = GetOutputItemTypes(response);
throw new InvalidOperationException(
"LiteLLM Responses returned no parseable message text or function calls. " +
$"ResponseId={response.ResponseId ?? "<none>"}, outputTypes=[{string.Join(", ", outputTypes)}].");
}
private static string[] GetOutputItemTypes(ChatResponse response)
{
return response.Messages
.SelectMany(message => message.Contents)
.Select(content => content.RawRepresentation?.GetType().Name ?? content.GetType().Name)
.ToArray();
}
private static string Truncate(string value, int maxLength)
{
return value.Length <= maxLength
? value
: value[..maxLength] + "...";
}
private static void AddInputItem(JsonArray input, StringBuilder instructions, ChatMessage message)
{
var role = message.Role.Value;
var text = message.Text;
if (role == ChatRole.System.Value)
{
if (!string.IsNullOrWhiteSpace(text))
{
instructions.AppendLine(text);
}
return;
}
if (!string.IsNullOrWhiteSpace(text))
{
var isAssistant = role == ChatRole.Assistant.Value;
input.Add(new JsonObject
{
["type"] = "message",
["role"] = isAssistant ? "assistant" : "user",
["content"] = new JsonArray
{
new JsonObject
{
["type"] = isAssistant ? "output_text" : "input_text",
["text"] = text
}
}
});
}
foreach (var content in message.Contents)
{
if (content is FunctionCallContent call)
{
input.Add(new JsonObject
{
["type"] = "function_call",
["call_id"] = call.CallId,
["name"] = call.Name,
["arguments"] = SerializeFunctionArguments(call)
});
}
else if (content is FunctionResultContent result)
{
input.Add(new JsonObject
{
["type"] = "function_call_output",
["call_id"] = result.CallId,
["output"] = ResultToString(result.Result)
});
}
}
}
private static JsonArray CreateTools(IList<AITool>? tools)
{
var result = new JsonArray();
if (tools is null)
{
return result;
}
foreach (var tool in tools.OfType<AIFunctionDeclaration>())
{
result.Add(new JsonObject
{
["type"] = "function",
["name"] = tool.Name,
["description"] = tool.Description ?? string.Empty,
["parameters"] = JsonNode.Parse(tool.JsonSchema.GetRawText())
});
}
return result;
}
private static string SerializeFunctionArguments(FunctionCallContent call)
{
if (call.RawRepresentation is FunctionCallResponseItem responseItem
&& responseItem.FunctionArguments is not null)
{
return responseItem.FunctionArguments.ToString();
}
return JsonSerializer.Serialize(call.Arguments, JsonOptions);
}
private static string ResultToString(object? result)
{
return result switch
{
null => string.Empty,
string text => text,
_ => JsonSerializer.Serialize(result, JsonOptions)
};
}
private static int EstimateTokens(JsonObject payload)
{
var json = payload.ToJsonString(JsonOptions);
return Math.Max(1, (int)Math.Ceiling(json.Length / 4.0));
}
private static Uri NormalizeEndpoint(Uri endpoint)
{
var value = endpoint.ToString().TrimEnd('/');
if (!value.EndsWith("/v1", StringComparison.OrdinalIgnoreCase))
{
value += "/v1";
}
return new Uri(value + "/");
}
private static string NormalizeRelativePath(string value)
{
return value.TrimStart('/');
}
private static ResponsesClient CreateResponsesClient(
HttpClient httpClient,
string apiKey,
AsyncLocal<string?> requestInitiator)
{
var options = new OpenAIClientOptions
{
Endpoint = httpClient.BaseAddress
?? throw new InvalidOperationException("LiteLLM HTTP client requires a base address."),
Transport = new HttpClientPipelineTransport(httpClient),
RetryPolicy = new ClientRetryPolicy(0)
};
options.AddPolicy(
new InitiatorHeaderPolicy(requestInitiator),
PipelinePosition.PerCall);
return new ResponsesClient(
new ApiKeyCredential(apiKey),
options);
}
private sealed record ResponsesChatResult(
ChatResponse Response,
IReadOnlyList<string> ReasoningSummaries);
private sealed class InitiatorHeaderPolicy(AsyncLocal<string?> requestInitiator) : PipelinePolicy
{
public override void Process(
PipelineMessage message,
IReadOnlyList<PipelinePolicy> pipeline,
int currentIndex)
{
SetHeader(message);
ProcessNext(message, pipeline, currentIndex);
}
public override ValueTask ProcessAsync(
PipelineMessage message,
IReadOnlyList<PipelinePolicy> pipeline,
int currentIndex)
{
SetHeader(message);
return ProcessNextAsync(message, pipeline, currentIndex);
}
private void SetHeader(PipelineMessage message)
{
if (!string.IsNullOrWhiteSpace(requestInitiator.Value))
{
message.Request.Headers.Set("X-Initiator", requestInitiator.Value);
}
}
}
}
#pragma warning restore OPENAI001
#pragma warning restore MAAI001