Public Access
fix: support LiteLLM Responses streaming
PR and Push Build/Test / build-and-test (push) Successful in 14m17s
PR and Push Build/Test / build-and-test (push) Successful in 14m17s
This commit is contained in:
@@ -1,82 +1,223 @@
|
||||
using MeetingAssistant.Summary;
|
||||
using Microsoft.Extensions.AI;
|
||||
using System.Net;
|
||||
using System.Text;
|
||||
using System.Text.Json;
|
||||
|
||||
namespace MeetingAssistant.Tests;
|
||||
|
||||
public sealed class LiteLlmResponsesChatClientTests
|
||||
{
|
||||
[Fact]
|
||||
public void ParserIgnoresReasoningItemsWithNullStatusAndReadsText()
|
||||
public async Task ClientAssemblesStreamedTextResponseWithMetadataAndUsage()
|
||||
{
|
||||
const string json = """
|
||||
var handler = new SequencedHttpMessageHandler(
|
||||
new HttpResponseMessage(HttpStatusCode.OK)
|
||||
{
|
||||
"id": "resp_test",
|
||||
"created_at": 1779147100,
|
||||
"model": "gpt-5.5-2026-04-23",
|
||||
"output": [
|
||||
Content = new StringContent(
|
||||
"""
|
||||
data: {"type":"response.created","response":{"id":"resp_stream","created_at":1779147100,"model":"gpt-5.5","object":"response","output":[],"parallel_tool_calls":true,"status":"in_progress","store":false},"sequence_number":0}
|
||||
|
||||
data: {"type":"response.output_item.added","output_index":0,"item":{"id":"msg_stream","type":"message","status":"in_progress","content":[],"role":"assistant"},"sequence_number":1}
|
||||
|
||||
data: {"type":"response.content_part.added","item_id":"msg_stream","output_index":0,"content_index":0,"part":{"type":"output_text","annotations":[],"text":""},"sequence_number":2}
|
||||
|
||||
data: {"type":"response.output_text.delta","item_id":"msg_stream","output_index":0,"content_index":0,"delta":"Streamed OK","sequence_number":3}
|
||||
|
||||
data: {"type":"response.output_text.done","item_id":"msg_stream","output_index":0,"content_index":0,"text":"Streamed OK","sequence_number":4}
|
||||
|
||||
data: {"type":"response.content_part.done","item_id":"msg_stream","output_index":0,"content_index":0,"part":{"type":"output_text","annotations":[],"text":"Streamed OK"},"sequence_number":5}
|
||||
|
||||
data: {"type":"response.output_item.done","output_index":0,"item":{"id":"msg_stream","type":"message","status":"completed","content":[{"type":"output_text","annotations":[],"text":"Streamed OK"}],"role":"assistant"},"sequence_number":6}
|
||||
|
||||
data: {"type":"response.completed","response":{"id":"resp_stream","created_at":1779147100,"model":"gpt-5.5","object":"response","output":[],"parallel_tool_calls":true,"status":"completed","store":false,"usage":{"input_tokens":12,"output_tokens":3,"total_tokens":15}},"sequence_number":7}
|
||||
|
||||
data: [DONE]
|
||||
|
||||
""",
|
||||
Encoding.UTF8,
|
||||
"text/event-stream")
|
||||
});
|
||||
using var client = CreateClient(handler, reconnectionAttempts: 0);
|
||||
|
||||
var response = await client.GetResponseAsync([new ChatMessage(ChatRole.User, "reply")]);
|
||||
|
||||
Assert.Equal("Streamed OK", response.Text);
|
||||
Assert.Equal("resp_stream", response.ResponseId);
|
||||
Assert.Equal("gpt-5.5", response.ModelId);
|
||||
Assert.Equal(DateTimeOffset.FromUnixTimeSeconds(1779147100), response.CreatedAt);
|
||||
Assert.NotNull(response.Usage);
|
||||
Assert.Equal(12, response.Usage.InputTokenCount);
|
||||
Assert.Equal(3, response.Usage.OutputTokenCount);
|
||||
Assert.Equal(15, response.Usage.TotalTokenCount);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ClientAssemblesStreamedFunctionCall()
|
||||
{
|
||||
var handler = new SequencedHttpMessageHandler(
|
||||
new HttpResponseMessage(HttpStatusCode.OK)
|
||||
{
|
||||
Content = new StringContent(
|
||||
"""
|
||||
data: {"type":"response.output_item.done","output_index":0,"item":{"type":"function_call","call_id":"call_stream","name":"write_summary","arguments":"{\"markdown\":\"# Summary\\nDone\"}","status":"completed"}}
|
||||
|
||||
data: {"type":"response.completed","response":{"id":"resp_stream","model":"gpt-5.5","output":[]}}
|
||||
|
||||
data: [DONE]
|
||||
|
||||
""",
|
||||
Encoding.UTF8,
|
||||
"text/event-stream")
|
||||
});
|
||||
using var client = CreateClient(handler, reconnectionAttempts: 0);
|
||||
|
||||
var response = await client.GetResponseAsync([new ChatMessage(ChatRole.User, "write summary")]);
|
||||
var call = Assert.IsType<FunctionCallContent>(Assert.Single(response.Messages[0].Contents));
|
||||
|
||||
Assert.Equal("call_stream", call.CallId);
|
||||
Assert.Equal("write_summary", call.Name);
|
||||
Assert.NotNull(call.Arguments);
|
||||
Assert.Equal("# Summary\nDone", call.Arguments["markdown"]?.ToString());
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ClientUsesNonStreamingResponsesWhenConfigured()
|
||||
{
|
||||
var handler = new RecordingHttpMessageHandler(_ => new HttpResponseMessage(HttpStatusCode.OK)
|
||||
{
|
||||
Content = new StringContent(
|
||||
"""
|
||||
{
|
||||
"type": "reasoning",
|
||||
"summary": [],
|
||||
"status": null
|
||||
},
|
||||
{
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"status": "completed",
|
||||
"content": [
|
||||
"id": "resp_nonstream",
|
||||
"created_at": 1779147100,
|
||||
"model": "gpt-5.5",
|
||||
"object": "response",
|
||||
"output": [
|
||||
{
|
||||
"type": "output_text",
|
||||
"text": "OK"
|
||||
"id": "msg_nonstream",
|
||||
"type": "message",
|
||||
"status": "completed",
|
||||
"content": [
|
||||
{
|
||||
"type": "output_text",
|
||||
"annotations": [],
|
||||
"text": "Non-streamed OK"
|
||||
}
|
||||
],
|
||||
"role": "assistant"
|
||||
}
|
||||
]
|
||||
],
|
||||
"parallel_tool_calls": true,
|
||||
"status": "completed",
|
||||
"store": false,
|
||||
"usage": {
|
||||
"input_tokens": 12,
|
||||
"output_tokens": 4,
|
||||
"total_tokens": 16
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
""";
|
||||
""",
|
||||
Encoding.UTF8,
|
||||
"application/json")
|
||||
});
|
||||
using var client = CreateClient(handler, reconnectionAttempts: 0, useStreaming: false);
|
||||
|
||||
var response = LiteLlmResponsesChatClient.ParseResponseJson(json);
|
||||
var response = await client.GetResponseAsync([new ChatMessage(ChatRole.User, "reply")]);
|
||||
|
||||
Assert.Equal("OK", response.Text);
|
||||
Assert.Equal("resp_test", response.ResponseId);
|
||||
Assert.Equal("gpt-5.5-2026-04-23", response.ModelId);
|
||||
Assert.Equal("Non-streamed OK", response.Text);
|
||||
Assert.Contains("\"stream\":false", Assert.Single(handler.RequestBodies));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ClientSendsChatMessagesAsResponsesMessageItems()
|
||||
{
|
||||
var handler = new RecordingHttpMessageHandler(_ => CreateStreamedTextResponse("Done."));
|
||||
using var client = CreateClient(handler, reconnectionAttempts: 0);
|
||||
|
||||
await client.GetResponseAsync(
|
||||
[
|
||||
new ChatMessage(ChatRole.User, "write summary"),
|
||||
new ChatMessage(ChatRole.Assistant, "I will inspect the meeting.")
|
||||
]);
|
||||
|
||||
using var request = JsonDocument.Parse(Assert.Single(handler.RequestBodies));
|
||||
var inputItems = request.RootElement.GetProperty("input").EnumerateArray().ToArray();
|
||||
Assert.Equal(2, inputItems.Length);
|
||||
Assert.All(inputItems, item => Assert.Equal("message", item.GetProperty("type").GetString()));
|
||||
Assert.Equal("user", inputItems[0].GetProperty("role").GetString());
|
||||
Assert.Equal(
|
||||
"input_text",
|
||||
Assert.Single(inputItems[0].GetProperty("content").EnumerateArray())
|
||||
.GetProperty("type")
|
||||
.GetString());
|
||||
Assert.Equal("assistant", inputItems[1].GetProperty("role").GetString());
|
||||
Assert.Equal(
|
||||
"output_text",
|
||||
Assert.Single(inputItems[1].GetProperty("content").EnumerateArray())
|
||||
.GetProperty("type")
|
||||
.GetString());
|
||||
Assert.DoesNotContain("\"type\":\"unknown\"", handler.RequestBodies[0]);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task AgentLoopReturnsMalformedFunctionArgumentsWithoutInvokingTool()
|
||||
{
|
||||
var responses = new Queue<HttpResponseMessage>(
|
||||
[
|
||||
new HttpResponseMessage(HttpStatusCode.OK)
|
||||
{
|
||||
Content = new StringContent(
|
||||
"""
|
||||
data: {"type":"response.output_item.done","output_index":0,"item":{"type":"function_call","call_id":"call_invalid","name":"write_summary","arguments":"{not-json","status":"completed"}}
|
||||
|
||||
data: {"type":"response.completed","response":{"id":"resp_invalid","model":"gpt-5.5","output":[]}}
|
||||
|
||||
data: [DONE]
|
||||
|
||||
""",
|
||||
Encoding.UTF8,
|
||||
"text/event-stream")
|
||||
},
|
||||
CreateStreamedTextResponse("Recovered.")
|
||||
]);
|
||||
var handler = new RecordingHttpMessageHandler(_ => responses.Dequeue());
|
||||
var toolInvoked = false;
|
||||
var tool = AIFunctionFactory.Create(
|
||||
(string markdown) =>
|
||||
{
|
||||
toolInvoked = true;
|
||||
return markdown;
|
||||
},
|
||||
"write_summary",
|
||||
"Writes a summary.");
|
||||
using var innerClient = CreateClient(handler, reconnectionAttempts: 0);
|
||||
using var functionClient = innerClient
|
||||
.AsBuilder()
|
||||
.UseFunctionInvocation(
|
||||
loggerFactory: null,
|
||||
client => client.FunctionInvoker = FunctionInvocationGuard.InvokeAsync)
|
||||
.Build();
|
||||
|
||||
var response = await functionClient.GetResponseAsync(
|
||||
[new ChatMessage(ChatRole.User, "write summary")],
|
||||
new ChatOptions { Tools = [tool] });
|
||||
|
||||
Assert.False(toolInvoked);
|
||||
Assert.Equal("Recovered.", response.Text);
|
||||
Assert.Equal(2, handler.RequestBodies.Count);
|
||||
Assert.Contains("invalid_tool_arguments", handler.RequestBodies[1]);
|
||||
Assert.Contains("call_invalid", handler.RequestBodies[1]);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ClientReportsVisibleReasoningSummariesSeparatelyFromResponseText()
|
||||
{
|
||||
var handler = new SequencedHttpMessageHandler(
|
||||
new HttpResponseMessage(HttpStatusCode.OK)
|
||||
{
|
||||
Content = new StringContent("""
|
||||
{
|
||||
"output": [
|
||||
{
|
||||
"type": "reasoning",
|
||||
"summary": [
|
||||
{
|
||||
"type": "summary_text",
|
||||
"text": "Checked the configured rules file."
|
||||
},
|
||||
{
|
||||
"type": "summary_text",
|
||||
"text": "Prepared a targeted update."
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"type": "message",
|
||||
"content": [
|
||||
{
|
||||
"type": "output_text",
|
||||
"text": "Done."
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
""")
|
||||
});
|
||||
CreateStreamedTextResponse(
|
||||
"Done.",
|
||||
"Checked the configured rules file.",
|
||||
"Prepared a targeted update."));
|
||||
var reasoningSummaries = new List<string>();
|
||||
using var client = CreateClient(
|
||||
handler,
|
||||
@@ -98,33 +239,7 @@ public sealed class LiteLlmResponsesChatClientTests
|
||||
public async Task ClientDoesNotFailResponseWhenReasoningSummaryCallbackFails()
|
||||
{
|
||||
var handler = new SequencedHttpMessageHandler(
|
||||
new HttpResponseMessage(HttpStatusCode.OK)
|
||||
{
|
||||
Content = new StringContent("""
|
||||
{
|
||||
"output": [
|
||||
{
|
||||
"type": "reasoning",
|
||||
"summary": [
|
||||
{
|
||||
"type": "summary_text",
|
||||
"text": "Checked the configured rules file."
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"type": "message",
|
||||
"content": [
|
||||
{
|
||||
"type": "output_text",
|
||||
"text": "Done."
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
""")
|
||||
});
|
||||
CreateStreamedTextResponse("Done.", "Checked the configured rules file."));
|
||||
using var client = CreateClient(
|
||||
handler,
|
||||
reconnectionAttempts: 0,
|
||||
@@ -135,64 +250,6 @@ public sealed class LiteLlmResponsesChatClientTests
|
||||
Assert.Equal("Done.", response.Text);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ParserReadsFunctionCalls()
|
||||
{
|
||||
const string json = """
|
||||
{
|
||||
"output": [
|
||||
{
|
||||
"type": "function_call",
|
||||
"call_id": "call_1",
|
||||
"name": "write_summary",
|
||||
"arguments": "{\"markdown\":\"# Summary\\nDone\"}",
|
||||
"status": "completed"
|
||||
}
|
||||
]
|
||||
}
|
||||
""";
|
||||
|
||||
var response = LiteLlmResponsesChatClient.ParseResponseJson(json);
|
||||
var call = Assert.IsType<FunctionCallContent>(Assert.Single(response.Messages[0].Contents));
|
||||
|
||||
Assert.Equal("call_1", call.CallId);
|
||||
Assert.Equal("write_summary", call.Name);
|
||||
Assert.NotNull(call.Arguments);
|
||||
Assert.Equal("# Summary\nDone", call.Arguments["markdown"]);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ParserReadsUsage()
|
||||
{
|
||||
const string json = """
|
||||
{
|
||||
"usage": {
|
||||
"input_tokens": 123,
|
||||
"output_tokens": 45,
|
||||
"total_tokens": 168
|
||||
},
|
||||
"output": [
|
||||
{
|
||||
"type": "message",
|
||||
"content": [
|
||||
{
|
||||
"type": "output_text",
|
||||
"text": "OK"
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
""";
|
||||
|
||||
var response = LiteLlmResponsesChatClient.ParseResponseJson(json);
|
||||
|
||||
Assert.NotNull(response.Usage);
|
||||
Assert.Equal(123, response.Usage.InputTokenCount);
|
||||
Assert.Equal(45, response.Usage.OutputTokenCount);
|
||||
Assert.Equal(168, response.Usage.TotalTokenCount);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ClientRetriesTransientServerFailure()
|
||||
{
|
||||
@@ -201,24 +258,7 @@ public sealed class LiteLlmResponsesChatClientTests
|
||||
{
|
||||
Content = new StringContent("Internal Server Error")
|
||||
},
|
||||
new HttpResponseMessage(HttpStatusCode.OK)
|
||||
{
|
||||
Content = new StringContent("""
|
||||
{
|
||||
"output": [
|
||||
{
|
||||
"type": "message",
|
||||
"content": [
|
||||
{
|
||||
"type": "output_text",
|
||||
"text": "Done."
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
""")
|
||||
});
|
||||
CreateStreamedTextResponse("Done."));
|
||||
var retryCount = 0;
|
||||
using var client = CreateClient(handler, reconnectionAttempts: 1, retrying: () => retryCount++);
|
||||
|
||||
@@ -249,24 +289,7 @@ public sealed class LiteLlmResponsesChatClientTests
|
||||
[Fact]
|
||||
public async Task ClientSendsUserInitiatorOnceThenAgentInitiator()
|
||||
{
|
||||
var handler = new RecordingHttpMessageHandler(_ => new HttpResponseMessage(HttpStatusCode.OK)
|
||||
{
|
||||
Content = new StringContent("""
|
||||
{
|
||||
"output": [
|
||||
{
|
||||
"type": "message",
|
||||
"content": [
|
||||
{
|
||||
"type": "output_text",
|
||||
"text": "Done."
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
""")
|
||||
});
|
||||
var handler = new RecordingHttpMessageHandler(_ => CreateStreamedTextResponse("Done."));
|
||||
using var client = CreateClient(handler, reconnectionAttempts: 0);
|
||||
|
||||
await client.GetResponseAsync([new ChatMessage(ChatRole.User, "write summary")]);
|
||||
@@ -298,24 +321,7 @@ public sealed class LiteLlmResponsesChatClientTests
|
||||
};
|
||||
}
|
||||
|
||||
return new HttpResponseMessage(HttpStatusCode.OK)
|
||||
{
|
||||
Content = new StringContent("""
|
||||
{
|
||||
"output": [
|
||||
{
|
||||
"type": "message",
|
||||
"content": [
|
||||
{
|
||||
"type": "output_text",
|
||||
"text": "Done."
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
""")
|
||||
};
|
||||
return CreateStreamedTextResponse("Done.");
|
||||
});
|
||||
using var client = CreateClient(
|
||||
handler,
|
||||
@@ -350,24 +356,7 @@ public sealed class LiteLlmResponsesChatClientTests
|
||||
};
|
||||
}
|
||||
|
||||
return new HttpResponseMessage(HttpStatusCode.OK)
|
||||
{
|
||||
Content = new StringContent("""
|
||||
{
|
||||
"output": [
|
||||
{
|
||||
"type": "message",
|
||||
"content": [
|
||||
{
|
||||
"type": "output_text",
|
||||
"text": "Done."
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
""")
|
||||
};
|
||||
return CreateStreamedTextResponse("Done.");
|
||||
});
|
||||
using var client = CreateClient(
|
||||
handler,
|
||||
@@ -400,7 +389,8 @@ public sealed class LiteLlmResponsesChatClientTests
|
||||
int reconnectionAttempts,
|
||||
LiteLlmResponsesCompactionOptions? compactionOptions = null,
|
||||
Action? retrying = null,
|
||||
Action<string>? reasoningSummaryChanged = null)
|
||||
Action<string>? reasoningSummaryChanged = null,
|
||||
bool useStreaming = true)
|
||||
{
|
||||
return new LiteLlmResponsesChatClient(
|
||||
new HttpClient(handler)
|
||||
@@ -415,7 +405,49 @@ public sealed class LiteLlmResponsesChatClientTests
|
||||
TimeSpan.Zero,
|
||||
compactionOptions,
|
||||
retrying: retrying,
|
||||
reasoningSummaryChanged: reasoningSummaryChanged);
|
||||
reasoningSummaryChanged: reasoningSummaryChanged,
|
||||
useStreaming: useStreaming);
|
||||
}
|
||||
|
||||
private static HttpResponseMessage CreateStreamedTextResponse(
|
||||
string text,
|
||||
params string[] reasoningSummaries)
|
||||
{
|
||||
var events = new List<string>();
|
||||
for (var index = 0; index < reasoningSummaries.Length; index++)
|
||||
{
|
||||
events.Add("data: " + JsonSerializer.Serialize(new
|
||||
{
|
||||
type = "response.reasoning_summary_text.delta",
|
||||
item_id = "reasoning_stream",
|
||||
output_index = 0,
|
||||
summary_index = index,
|
||||
delta = reasoningSummaries[index]
|
||||
}));
|
||||
}
|
||||
|
||||
var outputIndex = reasoningSummaries.Length > 0 ? 1 : 0;
|
||||
events.Add("data: " + JsonSerializer.Serialize(new
|
||||
{
|
||||
type = "response.output_text.delta",
|
||||
item_id = "msg_stream",
|
||||
output_index = outputIndex,
|
||||
content_index = 0,
|
||||
delta = text
|
||||
}));
|
||||
events.Add(
|
||||
"""data: {"type":"response.completed","response":{"id":"resp_stream","created_at":1779147100,"model":"gpt-5.5","object":"response","output":[],"parallel_tool_calls":true,"status":"completed","store":false,"usage":{"input_tokens":12,"output_tokens":3,"total_tokens":15}}}""");
|
||||
events.Add("data: [DONE]");
|
||||
|
||||
return new HttpResponseMessage(HttpStatusCode.OK)
|
||||
{
|
||||
Content = new StringContent(
|
||||
string.Join($"{Environment.NewLine}{Environment.NewLine}", events)
|
||||
+ Environment.NewLine
|
||||
+ Environment.NewLine,
|
||||
Encoding.UTF8,
|
||||
"text/event-stream")
|
||||
};
|
||||
}
|
||||
|
||||
private sealed class SequencedHttpMessageHandler : HttpMessageHandler
|
||||
|
||||
Reference in New Issue
Block a user