Skip to content

.Net: [WIP] OTel model diagnostics: streaming APIs #6242

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 12 commits into from
May 15, 2024
Merged

.Net: [WIP] OTel model diagnostics: streaming APIs #6242

Show file tree
Hide file tree
Changes from 9 commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
f79fb20
OTel model diagnostics: streaming APIs
TaoChenOSU May 14, 2024
e4876ec
Parse tool calls in chat message content
TaoChenOSU May 14, 2024
3259664
Dispose resources in catch block
TaoChenOSU May 14, 2024
9884b55
Dispose resources in a finally block
TaoChenOSU May 14, 2024
05b918f
Remove LocationPlugin
TaoChenOSU May 14, 2024
6f27752
Catch exceptions in enumerating the stream
TaoChenOSU May 14, 2024
0fc26a5
Fix: add back AddStreamingContent
TaoChenOSU May 14, 2024
da1744a
Remove unsafe dictionary in ModelDiagnostics
TaoChenOSU May 15, 2024
416600f
small fixes
TaoChenOSU May 15, 2024
5beba9f
Addressed comments
TaoChenOSU May 15, 2024
85dc0f0
Address comments 2
TaoChenOSU May 15, 2024
ec7145e
Merge branch 'main' into taochen/otel-model-diagnostics-streaming
TaoChenOSU May 15, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
127 changes: 106 additions & 21 deletions dotnet/samples/Demos/TelemetryWithAppInsights/Program.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,11 +77,26 @@ public static async Task Main()
Console.WriteLine();

Console.WriteLine("Write a poem about John Doe and translate it to Italian.");
await RunAzureOpenAIChatAsync(kernel);
using (var _ = s_activitySource.StartActivity("Chat"))
{
await RunAzureOpenAIChatAsync(kernel);
Console.WriteLine();
await RunGoogleAIChatAsync(kernel);
Console.WriteLine();
await RunHuggingFaceChatAsync(kernel);
}

Console.WriteLine();
await RunGoogleAIChatAsync(kernel);
Console.WriteLine();
await RunHuggingFaceChatAsync(kernel);

Console.WriteLine("Get weather.");
using (var _ = s_activitySource.StartActivity("ToolCalls"))
{
await RunAzureOpenAIToolCallsAsync(kernel);
Console.WriteLine();
await RunGoogleAIToolCallAsync(kernel);
// HuggingFace does not support tool calls yet.
}
}

#region Private
Expand All @@ -99,16 +114,17 @@ public static async Task Main()
/// </summary>
private static readonly ActivitySource s_activitySource = new("Telemetry.Example");

private const string AzureOpenAIChatServiceKey = "AzureOpenAIChat";
private const string GoogleAIGeminiChatServiceKey = "GoogleAIGeminiChat";
private const string HuggingFaceChatServiceKey = "HuggingFaceChat";
private const string AzureOpenAIServiceKey = "AzureOpenAI";
private const string GoogleAIGeminiServiceKey = "GoogleAIGemini";
private const string HuggingFaceServiceKey = "HuggingFace";

#region chat completion
private static async Task RunAzureOpenAIChatAsync(Kernel kernel)
{
Console.WriteLine("============= Azure OpenAI Chat Completion =============");

using var activity = s_activitySource.StartActivity(AzureOpenAIChatServiceKey);
SetTargetService(kernel, AzureOpenAIChatServiceKey);
using var activity = s_activitySource.StartActivity(AzureOpenAIServiceKey);
SetTargetService(kernel, AzureOpenAIServiceKey);
try
{
await RunChatAsync(kernel);
Expand All @@ -124,8 +140,8 @@ private static async Task RunGoogleAIChatAsync(Kernel kernel)
{
Console.WriteLine("============= Google Gemini Chat Completion =============");

using var activity = s_activitySource.StartActivity(GoogleAIGeminiChatServiceKey);
SetTargetService(kernel, GoogleAIGeminiChatServiceKey);
using var activity = s_activitySource.StartActivity(GoogleAIGeminiServiceKey);
SetTargetService(kernel, GoogleAIGeminiServiceKey);

try
{
Expand All @@ -142,8 +158,8 @@ private static async Task RunHuggingFaceChatAsync(Kernel kernel)
{
Console.WriteLine("============= HuggingFace Chat Completion =============");

using var activity = s_activitySource.StartActivity(HuggingFaceChatServiceKey);
SetTargetService(kernel, HuggingFaceChatServiceKey);
using var activity = s_activitySource.StartActivity(HuggingFaceServiceKey);
SetTargetService(kernel, HuggingFaceServiceKey);

try
{
Expand All @@ -158,22 +174,72 @@ private static async Task RunHuggingFaceChatAsync(Kernel kernel)

private static async Task RunChatAsync(Kernel kernel)
{
// Using non-streaming to get the poem.
var poem = await kernel.InvokeAsync<string>(
"WriterPlugin",
"ShortPoem",
new KernelArguments { ["input"] = "Write a poem about John Doe." });
var translatedPoem = await kernel.InvokeAsync<string>(
Console.WriteLine($"Poem:\n{poem}\n");

// Use streaming to translate the poem.
Console.WriteLine("Translated Poem:");
await foreach (var update in kernel.InvokeStreamingAsync<string>(
"WriterPlugin",
"Translate",
new KernelArguments
{
["input"] = poem,
["language"] = "Italian"
});
}))
{
Console.Write(update);
}
}
#endregion

Console.WriteLine($"Poem:\n{poem}\n\nTranslated Poem:\n{translatedPoem}");
#region tool calls
private static async Task RunAzureOpenAIToolCallsAsync(Kernel kernel)
{
Console.WriteLine("============= Azure OpenAI ToolCalls =============");

using var activity = s_activitySource.StartActivity(AzureOpenAIServiceKey);
SetTargetService(kernel, AzureOpenAIServiceKey);
try
{
await RunAutoToolCallAsync(kernel);
}
catch (Exception ex)
{
activity?.SetStatus(ActivityStatusCode.Error, ex.Message);
Console.WriteLine($"Error: {ex.Message}");
}
}

private static async Task RunGoogleAIToolCallAsync(Kernel kernel)
{
Console.WriteLine("============= Google Gemini ToolCalls =============");

using var activity = s_activitySource.StartActivity(GoogleAIGeminiServiceKey);
SetTargetService(kernel, GoogleAIGeminiServiceKey);
try
{
await RunAutoToolCallAsync(kernel);
}
catch (Exception ex)
{
activity?.SetStatus(ActivityStatusCode.Error, ex.Message);
Console.WriteLine($"Error: {ex.Message}");
}
}

private static async Task RunAutoToolCallAsync(Kernel kernel)
{
var result = await kernel.InvokePromptAsync("What is the weather like in Seattle?");

Console.WriteLine(result);
}
#endregion

private static Kernel GetKernel(ILoggerFactory loggerFactory)
{
var folder = RepoFiles.SamplePluginsPath();
Expand All @@ -187,19 +253,20 @@ private static Kernel GetKernel(ILoggerFactory loggerFactory)
modelId: TestConfiguration.AzureOpenAI.ChatModelId,
endpoint: TestConfiguration.AzureOpenAI.Endpoint,
apiKey: TestConfiguration.AzureOpenAI.ApiKey,
serviceId: AzureOpenAIChatServiceKey)
serviceId: AzureOpenAIServiceKey)
.AddGoogleAIGeminiChatCompletion(
modelId: TestConfiguration.GoogleAI.Gemini.ModelId,
apiKey: TestConfiguration.GoogleAI.ApiKey,
serviceId: GoogleAIGeminiChatServiceKey)
serviceId: GoogleAIGeminiServiceKey)
.AddHuggingFaceChatCompletion(
model: TestConfiguration.HuggingFace.ModelId,
endpoint: new Uri("https://api-inference.huggingface.co"),
apiKey: TestConfiguration.HuggingFace.ApiKey,
serviceId: HuggingFaceChatServiceKey);
serviceId: HuggingFaceServiceKey);

builder.Services.AddSingleton<IAIServiceSelector>(new AIServiceSelector());
builder.Plugins.AddFromPromptDirectory(Path.Combine(folder, "WriterPlugin"));
builder.Plugins.AddFromType<WeatherPlugin>();

return builder.Build();
}
Expand Down Expand Up @@ -240,9 +307,17 @@ public bool TrySelectAIService<T>(
service = targetService;
serviceSettings = targetServiceKey switch
{
AzureOpenAIChatServiceKey => new OpenAIPromptExecutionSettings(),
GoogleAIGeminiChatServiceKey => new GeminiPromptExecutionSettings(),
HuggingFaceChatServiceKey => new HuggingFacePromptExecutionSettings(),
AzureOpenAIServiceKey => new OpenAIPromptExecutionSettings()
{
Temperature = 0,
ToolCallBehavior = ToolCallBehavior.AutoInvokeKernelFunctions
},
GoogleAIGeminiServiceKey => new GeminiPromptExecutionSettings()
{
Temperature = 0,
ToolCallBehavior = GeminiToolCallBehavior.AutoInvokeKernelFunctions
},
HuggingFaceServiceKey => new HuggingFacePromptExecutionSettings(),
_ => null,
};

Expand All @@ -256,4 +331,14 @@ public bool TrySelectAIService<T>(
}
}
#endregion

#region Plugins

public sealed class WeatherPlugin
{
[KernelFunction]
public string GetWeather(string location) => $"Weather in {location} is 70°F.";
}

#endregion
}
58 changes: 50 additions & 8 deletions dotnet/src/Connectors/Connectors.Google/Core/Gemini/Clients/GeminiChatCompletionClient.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -226,15 +226,57 @@ public async IAsyncEnumerable<StreamingChatMessageContent> StreamGenerateChatMes

for (state.Iteration = 1; ; state.Iteration++)
{
using var httpRequestMessage = await this.CreateHttpRequestAsync(state.GeminiRequest, this._chatStreamingEndpoint).ConfigureAwait(false);
using var response = await this.SendRequestAndGetResponseImmediatelyAfterHeadersReadAsync(httpRequestMessage, cancellationToken)
.ConfigureAwait(false);
using var responseStream = await response.Content.ReadAsStreamAndTranslateExceptionAsync()
.ConfigureAwait(false);

await foreach (var messageContent in this.GetStreamingChatMessageContentsOrPopulateStateForToolCallingAsync(state, responseStream, cancellationToken).ConfigureAwait(false))
using (var activity = ModelDiagnostics.StartCompletionActivity(
this._chatGenerationEndpoint, this._modelId, ModelProvider, chatHistory, executionSettings))
{
yield return messageContent;
HttpResponseMessage? httpResponseMessage = null;
Stream? responseStream = null;
try
{
using var httpRequestMessage = await this.CreateHttpRequestAsync(state.GeminiRequest, this._chatStreamingEndpoint).ConfigureAwait(false);
// We cannot dispose these two objects leaving the try-catch block because we need them to read the response stream
httpResponseMessage = await this.SendRequestAndGetResponseImmediatelyAfterHeadersReadAsync(httpRequestMessage, cancellationToken).ConfigureAwait(false);
responseStream = await httpResponseMessage.Content.ReadAsStreamAndTranslateExceptionAsync().ConfigureAwait(false);
}
catch (Exception ex)
{
activity?.SetError(ex);
httpResponseMessage?.Dispose();
responseStream?.Dispose();
throw;
}

var responseEnumerator = this.GetStreamingChatMessageContentsOrPopulateStateForToolCallingAsync(state, responseStream, cancellationToken)
.ConfigureAwait(false)
.GetAsyncEnumerator();
List<StreamingChatMessageContent> streamedContents = [];
try
{
while (true)
{
try
{
if (!await responseEnumerator.MoveNextAsync())
{
break;
}
}
catch (Exception ex)
{
activity?.SetError(ex);
throw;
}

streamedContents.Add(responseEnumerator.Current);
yield return responseEnumerator.Current;
}
}
finally
{
activity?.EndStreaming(streamedContents);
httpResponseMessage?.Dispose();
responseStream?.Dispose();
}
}

if (!state.AutoInvoke)
Expand Down
55 changes: 46 additions & 9 deletions dotnet/src/Connectors/Connectors.HuggingFace/Core/HuggingFaceClient.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -169,17 +169,54 @@ public async IAsyncEnumerable<StreamingTextContent> StreamGenerateTextAsync(
var request = this.CreateTextRequest(prompt, executionSettings);
request.Stream = true;

using var httpRequestMessage = this.CreatePost(request, endpoint, this.ApiKey);

using var response = await this.SendRequestAndGetResponseImmediatelyAfterHeadersReadAsync(httpRequestMessage, cancellationToken)
.ConfigureAwait(false);

using var responseStream = await response.Content.ReadAsStreamAndTranslateExceptionAsync()
.ConfigureAwait(false);
using var activity = ModelDiagnostics.StartCompletionActivity(endpoint, modelId, this.ModelProvider, prompt, executionSettings);
HttpResponseMessage? httpResponseMessage = null;
Stream? responseStream = null;
try
{
using var httpRequestMessage = this.CreatePost(request, endpoint, this.ApiKey);
// We cannot dispose these two objects leaving the try-catch block because we need them to read the response stream
httpResponseMessage = await this.SendRequestAndGetResponseImmediatelyAfterHeadersReadAsync(httpRequestMessage, cancellationToken).ConfigureAwait(false);
responseStream = await httpResponseMessage.Content.ReadAsStreamAndTranslateExceptionAsync().ConfigureAwait(false);
}
catch (Exception ex)
{
activity?.SetError(ex);
httpResponseMessage?.Dispose();
responseStream?.Dispose();
throw;
}

await foreach (var streamingTextContent in this.ProcessTextResponseStreamAsync(responseStream, modelId, cancellationToken).ConfigureAwait(false))
var responseEnumerator = this.ProcessTextResponseStreamAsync(responseStream, modelId, cancellationToken)
.ConfigureAwait(false)
.GetAsyncEnumerator();
List<StreamingTextContent> streamedContents = [];
try
{
while (true)
{
try
{
if (!await responseEnumerator.MoveNextAsync())
{
break;
}
}
catch (Exception ex)
{
activity?.SetError(ex);
throw;
}

streamedContents.Add(responseEnumerator.Current);
yield return responseEnumerator.Current;
}
}
finally
{
yield return streamingTextContent;
activity?.EndStreaming(streamedContents);
httpResponseMessage?.Dispose();
responseStream?.Dispose();
}
}

Expand Down
Loading
Loading