Stevan/anthropic-client
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: yield events for streaming including custom message complete event

Browse Source
This commit is contained in:
Stevan Freeborn
2024-06-28 16:30:09 -05:00
parent a4784fd5db
commit 6ed99a6a23
12 changed files with 367 additions and 173 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/AnthropicClient/AnthropicApiClient.cs
+233
View File
@@ -0,0 +1,233 @@
using System.Net.Http.Headers;
using System.Text;
using System.Text.Json;
using AnthropicClient.Json;
using AnthropicClient.Utils;
using AnthropicClient.Models;
namespace AnthropicClient;
/// <summary>
/// Represents a client for interacting with the Anthropic API.
/// </summary>
public interface IAnthropicApiClient
{
/// <summary>
/// Creates a chat message asynchronously.
/// </summary>
/// <param name="request">The chat message request to create.</param>
/// <returns>A task that represents the asynchronous operation. The task result contains the chat response as an <see cref="AnthropicResult{T}"/>.</returns>
Task<AnthropicResult<ChatResponse>> CreateChatMessageAsync(ChatMessageRequest request);
/// <summary>
/// Creates a chat message asynchronously and streams the response.
/// </summary>
/// <param name="request">The chat message request to create.</param>
/// <returns>An asynchronous enumerable that yields the chat response line by line.</returns>
IAsyncEnumerable<AnthropicEvent> CreateChatMessageAsync(StreamChatMessageRequest request);
}
/// <inheritdoc cref="IAnthropicApiClient"/>
public class AnthropicApiClient : IAnthropicApiClient
{
private const string BaseUrl = "https://api.anthropic.com/v1/";
private const string ApiKeyHeader = "x-api-key";
private const string MessagesEndpoint = "messages";
private const string JsonContentType = "application/json";
private const string RequestIdHeader = "request-id";
private const string EventPrefix = "event:";
private const string DataPrefix = "data:";
private readonly Dictionary<string, string> _defaultHeaders = new()
{
{ "anthropic-version", "2023-06-01" },
};
private readonly HttpClient _httpClient;
/// <summary>
/// Initializes a new instance of the <see cref="AnthropicApiClient"/> class.
/// </summary>
/// <param name="apiKey">The API key to use for the client.</param>
/// <param name="httpClient">The HTTP client to use for the client.</param>
/// <exception cref="ArgumentNullException">Thrown when the API key or HTTP client is null.</exception>
/// <returns>A new instance of the <see cref="AnthropicApiClient"/> class.</returns>
public AnthropicApiClient(string apiKey, HttpClient httpClient)
{
ArgumentValidator.ThrowIfNull(apiKey, nameof(apiKey));
ArgumentValidator.ThrowIfNull(httpClient, nameof(httpClient));
_httpClient = httpClient;
_httpClient.BaseAddress = new Uri(BaseUrl);
_httpClient.DefaultRequestHeaders.Add(ApiKeyHeader, apiKey);
_httpClient.DefaultRequestHeaders
.Accept
.Add(new MediaTypeWithQualityHeaderValue(JsonContentType));
foreach (var pair in _defaultHeaders)
{
_httpClient.DefaultRequestHeaders.Add(pair.Key, pair.Value);
}
}
/// <inheritdoc />
public async Task<AnthropicResult<ChatResponse>> CreateChatMessageAsync(ChatMessageRequest request)
{
var response = await SendRequestAsync(request);
var anthropicHeaders = new AnthropicHeaders(response.Headers);
var responseContent = await response.Content.ReadAsStringAsync();
if (response.IsSuccessStatusCode is false)
{
var error = Deserialize<AnthropicError>(responseContent) ?? new AnthropicError();
return AnthropicResult<ChatResponse>.Failure(error, anthropicHeaders);
}
var chatResponse = Deserialize<ChatResponse>(responseContent) ?? new ChatResponse();
return AnthropicResult<ChatResponse>.Success(chatResponse, anthropicHeaders);
}
/// <inheritdoc />
public async IAsyncEnumerable<AnthropicEvent> CreateChatMessageAsync(StreamChatMessageRequest request)
{
var response = await SendRequestAsync(request);
var anthropicHeaders = new AnthropicHeaders(response.Headers);
using var responseContent = await response.Content.ReadAsStreamAsync();
using var streamReader = new StreamReader(responseContent);
ChatResponse? chatResponse = null;
Content? content = null;
var toolInputJsonStringBuilder = new StringBuilder();
var currentEvent = new AnthropicEvent();
do
{
var line = await streamReader.ReadLineAsync();
// I know...this is not pretty, but here is why...
// as events are being yielded I want to also
// build up the complete chat response
// so I can yield it as a special event to make tool
// calling easier to handle
// initialize chat response on message start
if (currentEvent.Type is EventType.MessageStart && currentEvent.Data is MessageStartEventData msgStartData)
{
chatResponse = msgStartData.Message;
}
// initialize content block on content block start
if (currentEvent.Type is EventType.ContentBlockStart && currentEvent.Data is ContentStartEventData contentStartData)
{
content = contentStartData.ContentBlock;
}
// update content block with deltas based on
// current content type and delta type
if (currentEvent.Type is EventType.ContentBlockDelta && currentEvent.Data is ContentDeltaEventData contentDeltaData)
{
if (content is TextContent textContent && contentDeltaData.Delta is TextDelta textDelta)
{
var newText = textContent.Text + textDelta.Text;
content = new TextContent(newText);
}
if (content is ToolUseContent toolUseContent && contentDeltaData.Delta is JsonDelta jsonDelta)
{
toolInputJsonStringBuilder.Append(jsonDelta.PartialJson);
}
}
// finalize content block on content block stop
// and add it to the chat response
if (currentEvent.Type is EventType.ContentBlockStop)
{
if (content is not null && chatResponse is not null)
{
if (content is TextContent textContent)
{
chatResponse.Content.Add(textContent);
}
if (content is ToolUseContent toolUseContent)
{
var input = Deserialize<Dictionary<string, object?>>(toolInputJsonStringBuilder.ToString());
var newToolUseContent = new ToolUseContent()
{
Id = toolUseContent.Id,
Name = toolUseContent.Name,
Input = input!,
};
chatResponse.Content.Add(newToolUseContent);
}
content = null;
}
}
// update chat response with message delta data
if (
currentEvent.Type is EventType.MessageDelta &&
currentEvent.Data is MessageDeltaEventData msgDeltaData &&
chatResponse is not null
)
{
chatResponse = new ChatResponse()
{
Id = chatResponse.Id,
Model = chatResponse.Model,
Role = chatResponse.Role,
StopReason = msgDeltaData.Delta.StopReason,
StopSequence = msgDeltaData.Delta.StopSequence,
Type = chatResponse.Type,
Usage = msgDeltaData.Usage,
Content = chatResponse.Content,
};
}
// yield chat response on message stop
if (currentEvent.Type is EventType.MessageStop && chatResponse is not null)
{
var eventData = new MessageCompleteEventData(chatResponse, anthropicHeaders);
yield return new AnthropicEvent(EventType.MessageComplete, eventData);
chatResponse = null;
}
if (line is null)
{
break;
}
if (line == string.Empty)
{
yield return currentEvent;
currentEvent = new AnthropicEvent();
}
if (line.StartsWith(EventPrefix))
{
var eventType = line.Substring(EventPrefix.Length).Trim();
currentEvent = new AnthropicEvent(eventType, currentEvent.Data);
continue;
}
if (line.StartsWith(DataPrefix))
{
var eventData = line.Substring(DataPrefix.Length).Trim();
var eventDataJson = Deserialize<EventData>(eventData);
currentEvent = new AnthropicEvent(currentEvent.Type, eventDataJson!);
continue;
}
} while (true);
}
private async Task<HttpResponseMessage> SendRequestAsync(MessageRequest request)
{
var requestContent = new StringContent(Serialize(request), Encoding.UTF8, JsonContentType);
return await _httpClient.PostAsync(MessagesEndpoint, requestContent);
}
private string Serialize<T>(T obj) => JsonSerializer.Serialize(obj, JsonSerializationOptions.DefaultOptions);
private T? Deserialize<T>(string json) => JsonSerializer.Deserialize<T>(json, JsonSerializationOptions.DefaultOptions);
}
src/AnthropicClient/Client.cs
-154
View File
@@ -1,154 +0,0 @@
using System.Net.Http.Headers;
using System.Text;
using System.Text.Json;
using AnthropicClient.Json;
using AnthropicClient.Utils;
using AnthropicClient.Models;
namespace AnthropicClient;
/// <summary>
/// Represents a client for interacting with the Anthropic API.
/// </summary>
public interface IClient
{
/// <summary>
/// Creates a chat message asynchronously.
/// </summary>
/// <param name="request">The chat message request to create.</param>
/// <returns>A task that represents the asynchronous operation. The task result contains the chat response as an <see cref="AnthropicResult{T}"/>.</returns>
Task<AnthropicResult<ChatResponse>> CreateChatMessageAsync(ChatMessageRequest request);
/// <summary>
/// Creates a chat message asynchronously and streams the response.
/// </summary>
/// <param name="request">The chat message request to create.</param>
/// <returns>An asynchronous enumerable that yields the chat response line by line.</returns>
IAsyncEnumerable<AnthropicEvent> CreateChatMessageAsync(StreamChatMessageRequest request);
}
/// <inheritdoc cref="IClient"/>
public class Client : IClient
{
private const string BaseUrl = "https://api.anthropic.com/v1/";
private const string ApiKeyHeader = "x-api-key";
private const string MessagesEndpoint = "messages";
private const string JsonContentType = "application/json";
private const string RequestIdHeader = "request-id";
private readonly Dictionary<string, string> _defaultHeaders = new()
{
{ "anthropic-version", "2023-06-01" },
};
private readonly HttpClient _httpClient;
/// <summary>
/// Initializes a new instance of the <see cref="Client"/> class.
/// </summary>
/// <param name="apiKey">The API key to use for the client.</param>
/// <param name="httpClient">The HTTP client to use for the client.</param>
/// <exception cref="ArgumentNullException">Thrown when the API key or HTTP client is null.</exception>
/// <returns>A new instance of the <see cref="Client"/> class.</returns>
public Client(string apiKey, HttpClient httpClient)
{
ArgumentValidator.ThrowIfNull(apiKey, nameof(apiKey));
ArgumentValidator.ThrowIfNull(httpClient, nameof(httpClient));
_httpClient = httpClient;
_httpClient.BaseAddress = new Uri(BaseUrl);
_httpClient.DefaultRequestHeaders.Add(ApiKeyHeader, apiKey);
_httpClient.DefaultRequestHeaders
.Accept
.Add(new MediaTypeWithQualityHeaderValue(JsonContentType));
foreach (var pair in _defaultHeaders)
{
_httpClient.DefaultRequestHeaders.Add(pair.Key, pair.Value);
}
}
/// <inheritdoc />
public async Task<AnthropicResult<ChatResponse>> CreateChatMessageAsync(ChatMessageRequest request)
{
var response = await SendRequestAsync(request);
var requestId = GetRequestId(response);
var responseContent = await response.Content.ReadAsStringAsync();
if (response.IsSuccessStatusCode is false)
{
var error = Deserialize<AnthropicError>(responseContent) ?? new AnthropicError();
return AnthropicResult<ChatResponse>.Failure(error, requestId);
}
var chatResponse = Deserialize<ChatResponse>(responseContent) ?? new ChatResponse();
return AnthropicResult<ChatResponse>.Success(chatResponse, requestId);
}
/// <inheritdoc />
public async IAsyncEnumerable<AnthropicEvent> CreateChatMessageAsync(StreamChatMessageRequest request)
{
var response = await SendRequestAsync(request);
var requestId = GetRequestId(response);
using var responseContent = await response.Content.ReadAsStreamAsync();
using var streamReader = new StreamReader(responseContent);
var currentEvent = new AnthropicEvent();
// TODO: I'd like to emit custom events unique to this client
// - "content_block_complete"
// - provides the complete content block that was streamed
// - useful for streaming in text, but then handling tool calls with their entire input
// will need to keep track of current content so that we can build it up as relevant
// deltas are streamed in
// will want to capture the content block start event
// then continue to build this up until we get to the content block stop event
// at that point we should have the complete block and we should be able to yield
// return the complete block and reset the current content block
// event: content_block_start
// data: {"type":"content_block_start","index":1,"content_block":{"type":"tool_use","id":"toolu_01T1x1fJ34qAmk2tNTrN7Up6","name":"get_weather","input":{}}}
do
{
var line = await streamReader.ReadLineAsync();
if (line is null)
{
break;
}
if (line.StartsWith("event:"))
{
var eventType = line.Substring("event:".Length).Trim();
currentEvent = currentEvent with { Type = eventType };
}
if (line.StartsWith("data:"))
{
var eventData = line.Substring("data:".Length).Trim();
var eventDataJson = JsonSerializer.Deserialize<EventData>(eventData, JsonSerializationOptions.DefaultOptions);
currentEvent = currentEvent with { Data = eventDataJson! };
}
if (line == string.Empty)
{
yield return currentEvent;
currentEvent = new AnthropicEvent();
}
} while (true);
}
private async Task<HttpResponseMessage> SendRequestAsync(MessageRequest request)
{
var requestContent = new StringContent(Serialize(request), Encoding.UTF8, JsonContentType);
return await _httpClient.PostAsync(MessagesEndpoint, requestContent);
}
private string Serialize<T>(T obj) => JsonSerializer.Serialize(obj, JsonSerializationOptions.DefaultOptions);
private T? Deserialize<T>(string json) => JsonSerializer.Deserialize<T>(json, JsonSerializationOptions.DefaultOptions);
private string GetRequestId(HttpResponseMessage response) => response.Headers.GetValues(RequestIdHeader).FirstOrDefault() ?? string.Empty;
}
src/AnthropicClient/Models/AnthropicEvent.cs
+1 -1
View File
@@ -2,7 +2,7 @@ using System.Text.Json.Serialization;
namespace AnthropicClient.Models;
public record AnthropicEvent
public class AnthropicEvent
{
public string Type { get; init; } = string.Empty;
public EventData Data { get; init; } = default!;
src/AnthropicClient/Models/AnthropicHeaders.cs
+79
View File
@@ -0,0 +1,79 @@
using System.Net.Http.Headers;
namespace AnthropicClient.Models;
/// <summary>
/// Represents headers included in Anthropic API responses.
/// </summary>
public class AnthropicHeaders
{
private const string RequestIdHeaderKey = "request-id";
private const string RateLimitRequestsLimitHeaderKey = "anthropic-ratelimit-requests-limit";
private const string RateLimitRequestsRemainingHeaderKey = "anthropic-ratelimit-requests-remaining";
private const string RateLimitRequestsResetHeaderKey = "anthropic-ratelimit-requests-reset";
private const string RateLimitTokensLimitHeaderKey = "anthropic-ratelimit-tokens-limit";
private const string RateLimitTokensRemainingHeaderKey = "anthropic-ratelimit-tokens-remaining";
private const string RateLimitTokensResetHeaderKey = "anthropic-ratelimit-tokens-reset";
private const string RetryAfterHeaderKey = "retry-after";
/// <summary>
/// Gets the request ID.
/// </summary>
public string RequestId { get; init; } = string.Empty;
/// <summary>
/// Gets the rate limit requests limit.
/// </summary>
public int RateLimitRequestsLimit { get; init; }
/// <summary>
/// Gets the rate limit requests remaining.
/// </summary>
public int RateLimitRequestsRemaining { get; init; }
/// <summary>
/// Gets the time of the rate limit requests reset.
/// </summary>
public DateTimeOffset RateLimitRequestsReset { get; init; }
/// <summary>
/// Gets the rate limit tokens limit.
/// </summary>
public int RateLimitTokensLimit { get; init; }
/// <summary>
/// Gets the rate limit tokens remaining.
/// </summary>
public int RateLimitTokensRemaining { get; init; }
/// <summary>
/// Gets the time of the rate limit tokens reset.
/// </summary>
public DateTimeOffset RateLimitTokensReset { get; init; }
/// <summary>
/// Gets the retry after value.
/// </summary>
public int RetryAfter { get; init; }
internal AnthropicHeaders()
{
}
/// <summary>
/// Initializes a new instance of the <see cref="AnthropicHeaders"/> class.
/// </summary>
/// <param name="headers">The HTTP response headers.</param>
/// <returns>A new instance of the <see cref="AnthropicHeaders"/> class.</returns>
public AnthropicHeaders(HttpResponseHeaders headers)
{
RequestId = headers.GetValues(RequestIdHeaderKey).FirstOrDefault() ?? string.Empty;
RateLimitRequestsLimit = int.Parse(headers.GetValues(RateLimitRequestsLimitHeaderKey).FirstOrDefault() ?? "0");
RateLimitRequestsRemaining = int.Parse(headers.GetValues(RateLimitRequestsRemainingHeaderKey).FirstOrDefault() ?? "0");
RateLimitRequestsReset = DateTimeOffset.Parse(headers.GetValues(RateLimitRequestsResetHeaderKey).FirstOrDefault() ?? DateTimeOffset.MinValue.ToString());
RateLimitTokensLimit = int.Parse(headers.GetValues(RateLimitTokensLimitHeaderKey).FirstOrDefault() ?? "0");
RateLimitTokensRemaining = int.Parse(headers.GetValues(RateLimitTokensRemainingHeaderKey).FirstOrDefault() ?? "0");
RateLimitTokensReset = DateTimeOffset.Parse(headers.GetValues(RateLimitTokensResetHeaderKey).FirstOrDefault() ?? DateTimeOffset.MinValue.ToString());
RetryAfter = int.Parse(headers.GetValues(RetryAfterHeaderKey).FirstOrDefault() ?? "0");
}
}
src/AnthropicClient/Models/AnthropicResult.cs
+10 -10
View File
@@ -24,7 +24,7 @@ public class AnthropicResult<T>
/// <summary>
/// The request ID of the operation.
/// </summary>
public string RequestId { get; }
public AnthropicHeaders Headers { get; }
/// <summary>
/// Initializes a new instance of the <see cref="AnthropicResult{T}"/> class.
@@ -32,35 +32,35 @@ public class AnthropicResult<T>
/// <param name="value">The value of the result.</param>
/// <param name="error">The error of the result.</param>
/// <param name="isSuccess">Indicates whether the operation was successful.</param>
/// <param name="requestId">The request ID of the operation.</param>
/// <param name="headers">The Anthropic headers for the result.</param>
/// <returns>A new instance of the <see cref="AnthropicResult{T}"/> class.</returns>
protected AnthropicResult(T value, AnthropicError error, bool isSuccess, string requestId)
protected AnthropicResult(T value, AnthropicError error, bool isSuccess, AnthropicHeaders headers)
{
Value = value;
Error = error;
IsSuccess = isSuccess;
RequestId = requestId;
Headers = headers;
}
/// <summary>
/// Creates a successful result.
/// </summary>
/// <param name="value">The value of the result.</param>
/// <param name="requestId">The request ID of the operation.</param>
/// <param name="headers">The Anthropic headers for the result.</param>
/// <returns>A new instance of the <see cref="AnthropicResult{T}"/> class.</returns>
public static AnthropicResult<T> Success(T value, string requestId)
public static AnthropicResult<T> Success(T value, AnthropicHeaders headers)
{
return new AnthropicResult<T>(value, null!, true, requestId);
return new AnthropicResult<T>(value, null!, true, headers);
}
/// <summary>
/// Creates a failed result.
/// </summary>
/// <param name="error">The error of the result.</param>
/// <param name="requestId">The request ID of the operation.</param>
/// <param name="headers">The Anthropic headers for the result.</param>
/// <returns>A new instance of the <see cref="AnthropicResult{T}"/> class.</returns>
public static AnthropicResult<T> Failure(AnthropicError error, string requestId)
public static AnthropicResult<T> Failure(AnthropicError error, AnthropicHeaders headers)
{
return new AnthropicResult<T>(default!, error, false, requestId);
return new AnthropicResult<T>(default!, error, false, headers);
}
}
src/AnthropicClient/Models/ContentStartEventData.cs
+1 -1
View File
@@ -7,7 +7,7 @@ public class ContentStartEventData : EventData
public int Index { get; init; }
[JsonPropertyName("content_block")]
public Content ContentBlock { get; init; }
public Content ContentBlock { get; init; } = default!;
[JsonConstructor]
internal ContentStartEventData() : base(EventType.ContentBlockStart)
src/AnthropicClient/Models/EventType.cs
+1
View File
@@ -7,6 +7,7 @@ public static class EventType
public const string MessageStart = "message_start";
public const string MessageDelta = "message_delta";
public const string MessageStop = "message_stop";
public const string MessageComplete = "message_complete";
public const string ContentBlockStart = "content_block_start";
public const string ContentBlockDelta = "content_block_delta";
public const string ContentBlockStop = "content_block_stop";
src/AnthropicClient/Models/MessageCompleteEventData.cs
+15
View File
@@ -0,0 +1,15 @@
using System.Text.Json.Serialization;
namespace AnthropicClient.Models;
public class MessageCompleteEventData : EventData
{
public AnthropicHeaders Headers { get; init; } = new();
public ChatResponse Message { get; init; } = new();
public MessageCompleteEventData(ChatResponse message, AnthropicHeaders headers) : base(EventType.MessageComplete)
{
Message = message;
Headers = headers;
}
}
src/AnthropicClient/Models/MessageDeltaEventData.cs
+2 -2
View File
@@ -4,8 +4,8 @@ namespace AnthropicClient.Models;
public class MessageDeltaEventData : EventData
{
public MessageDelta Delta { get; init; }
public ChatUsage Usage { get; init; }
public MessageDelta Delta { get; init; } = new();
public ChatUsage Usage { get; init; } = new();
[JsonConstructor]
internal MessageDeltaEventData() : base(EventType.MessageDelta)
src/AnthropicClient/Models/MessageStartEventData.cs
+2 -2
View File
@@ -4,14 +4,14 @@ namespace AnthropicClient.Models;
public class MessageStartEventData : EventData
{
public ChatMessage Message { get; init; } = new();
public ChatResponse Message { get; init; } = new();
[JsonConstructor]
internal MessageStartEventData() : base(EventType.MessageStart)
{
}
public MessageStartEventData(ChatMessage message) : base(EventType.MessageStart)
public MessageStartEventData(ChatResponse message) : base(EventType.MessageStart)
{
Message = message;
}
tests/AnthropicClient.Tests/EndToEnd/ClientTests.cs → tests/AnthropicClient.Tests/EndToEnd/AnthropicApiClientTests.cs
+22 -2
View File
@@ -6,7 +6,7 @@ public class ClientTests(
) : EndToEndTest(httpClientFixture, configFixture)
{
[Fact]
public async Task CreateChatMessage_WhenCalled_ShouldReturnChatResponse()
public async Task CreateChatMessage_WhenCalled_ItShouldReturnChatResponse()
{
var request = new ChatMessageRequest(
model: AnthropicModels.Claude3Haiku,
@@ -20,7 +20,7 @@ public class ClientTests(
}
[Fact]
public async Task CreateChatMessage_WhenCalledWithStreamRequest_IteratesOverChatResponse()
public async Task CreateChatMessage_WhenCalledWithStreamRequest_ItShouldReturnEvents()
{
var request = new StreamChatMessageRequest(
model: AnthropicModels.Claude3Haiku,
@@ -38,4 +38,24 @@ public class ClientTests(
events.Should().NotBeEmpty();
}
[Fact]
public async Task CreateChatMessage_WhenCalledWithStreamRequest_ItShouldYieldAMessageCompleteEvent()
{
var request = new StreamChatMessageRequest(
model: AnthropicModels.Claude3Haiku,
messages: [new(MessageRole.User, [new TextContent("Hello!")])]
);
var response = _client.CreateChatMessageAsync(request);
await foreach (var e in response)
{
if (e.Data is MessageCompleteEventData messageCompleteData)
{
messageCompleteData.Message.Should().NotBeNull();
break;
}
}
}
}
tests/AnthropicClient.Tests/EndToEnd/EndToEndTest.cs
+1 -1
View File
@@ -5,5 +5,5 @@ public class EndToEndTest(
ConfigurationFixture configFixture
) : IClassFixture<HttpClientFixture>, IClassFixture<ConfigurationFixture>
{
protected readonly Client _client = new(configFixture.AnthropicApiKey, httpClientFixture.HttpClient);
protected readonly AnthropicApiClient _client = new(configFixture.AnthropicApiKey, httpClientFixture.HttpClient);
}