Add onMessageReceived to support streaming.

This commit is contained in:
hchen 2023-07-27 10:07:39 -05:00
parent a87ec531b5
commit 842880dbbf
10 changed files with 49 additions and 32 deletions

View file

@ -46,8 +46,8 @@ public abstract class ConversationCompletionHookBase : IConversationCompletionHo
return Task.CompletedTask; return Task.CompletedTask;
} }
public virtual Task<string> AfterCompletion(string response) public virtual Task AfterCompletion(RoleDialogModel message)
{ {
return Task.FromResult(response); return Task.CompletedTask;
} }
} }

View file

@ -18,5 +18,5 @@ public interface IConversationCompletionHook
IConversationCompletionHook SetChatCompletion(IChatCompletion chatCompletion); IConversationCompletionHook SetChatCompletion(IChatCompletion chatCompletion);
Task BeforeCompletion(); Task BeforeCompletion();
Task<string> AfterCompletion(string response); Task AfterCompletion(RoleDialogModel message);
} }

View file

@ -8,8 +8,8 @@ public interface IConversationService
Task<Conversation> GetConversation(string id); Task<Conversation> GetConversation(string id);
Task<List<Conversation>> GetConversations(); Task<List<Conversation>> GetConversations();
Task DeleteConversation(string id); Task DeleteConversation(string id);
Task<string> SendMessage(string agentId, string conversationId, RoleDialogModel lastDalog); Task<bool> SendMessage(string agentId, string conversationId, RoleDialogModel lastDalog, Func<RoleDialogModel, Task> onMessageReceived);
Task<string> SendMessage(string agentId, string conversationId, List<RoleDialogModel> wholeDialogs); Task<bool> SendMessage(string agentId, string conversationId, List<RoleDialogModel> wholeDialogs, Func<RoleDialogModel, Task> onMessageReceived);
List<RoleDialogModel> GetDialogHistory(string agentId, string conversationId); List<RoleDialogModel> GetDialogHistory(string agentId, string conversationId);
Task CleanHistory(string agentId); Task CleanHistory(string agentId);
} }

View file

@ -3,11 +3,16 @@ namespace BotSharp.Abstraction.Conversations.Models;
public class RoleDialogModel public class RoleDialogModel
{ {
/// <summary> /// <summary>
/// user, system, assistant /// user, system, assistant, function
/// </summary> /// </summary>
public string Role { get; set; } public string Role { get; set; }
public string Content { get; set; } public string Content { get; set; }
/// <summary>
/// Function name if LLM response function call
/// </summary>
public string? Name { get; set; }
public RoleDialogModel(string role, string text) public RoleDialogModel(string role, string text)
{ {
Role = role; Role = role;

View file

@ -5,5 +5,5 @@ namespace BotSharp.Abstraction.MLTasks;
public interface IChatCompletion public interface IChatCompletion
{ {
string GetChatCompletions(Agent agent, List<RoleDialogModel> conversations); string GetChatCompletions(Agent agent, List<RoleDialogModel> conversations);
Task<string> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations); Task<bool> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations, Func<RoleDialogModel, Task> onMessageReceived);
} }

View file

@ -46,11 +46,13 @@ public class ConversationController : ControllerBase, IApiAdapter
{ {
var conv = _services.GetRequiredService<IConversationService>(); var conv = _services.GetRequiredService<IConversationService>();
var result = await conv.SendMessage(agentId, conversationId, new RoleDialogModel("user", input.Text)); var response = new MessageResponseModel();
return new MessageResponseModel await conv.SendMessage(agentId, conversationId, new RoleDialogModel("user", input.Text), async msg =>
{ {
Text = result response.Text += msg.Content;
}; });
return response;
} }
} }

View file

@ -67,20 +67,22 @@ public class ConversationService : IConversationService
return record.ToConversation(); return record.ToConversation();
} }
public async Task<string> SendMessage(string agentId, string conversationId, RoleDialogModel lastDalog) public async Task<bool> SendMessage(string agentId, string conversationId, RoleDialogModel lastDalog, Func<RoleDialogModel, Task> onMessageReceived)
{ {
_storage.Append(agentId, conversationId, lastDalog); _storage.Append(agentId, conversationId, lastDalog);
var wholeDialogs = GetDialogHistory(agentId, conversationId); var wholeDialogs = GetDialogHistory(agentId, conversationId);
var response = await SendMessage(agentId, conversationId, wholeDialogs); var response = await SendMessage(agentId, conversationId, wholeDialogs, async msg =>
{
_storage.Append(agentId, conversationId, new RoleDialogModel("assistant", response)); await onMessageReceived(msg);
_storage.Append(agentId, conversationId, new RoleDialogModel(msg.Role, msg.Content));
});
return response; return response;
} }
public async Task<string> SendMessage(string agentId, string conversationId, List<RoleDialogModel> wholeDialogs) public async Task<bool> SendMessage(string agentId, string conversationId, List<RoleDialogModel> wholeDialogs, Func<RoleDialogModel, Task> onMessageReceived)
{ {
var agent = await _services.GetRequiredService<IAgentService>().GetAgent(agentId); var agent = await _services.GetRequiredService<IAgentService>().GetAgent(agentId);
var converation = await GetConversation(conversationId); var converation = await GetConversation(conversationId);
@ -110,15 +112,14 @@ public class ConversationService : IConversationService
.BeforeCompletion(); .BeforeCompletion();
}); });
var response = await chatCompletion.GetChatCompletionsStreamingAsync(agent, wholeDialogs); var result = await chatCompletion.GetChatCompletionsStreamingAsync(agent, wholeDialogs, async msg =>
// After chat completion hook
hooks.ForEach(async hook =>
{ {
response = await hook.AfterCompletion(response); // After chat completion hook
hooks.ForEach(async hook => await hook.AfterCompletion(msg));
await onMessageReceived(msg);
}); });
return response; return result;
} }
public IChatCompletion GetChatCompletion() public IChatCompletion GetChatCompletion()

View file

@ -19,7 +19,7 @@ public class ChatCompletionProvider : IChatCompletion
throw new NotImplementedException(); throw new NotImplementedException();
} }
public Task<string> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations) public async Task<bool> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations, Func<RoleDialogModel, Task> onMessageReceived)
{ {
string totalResponse = ""; string totalResponse = "";
var content = string.Join("\n", conversations.Select(x => $"{x.Role}: {x.Content.Replace("user:", "")}")).Trim(); var content = string.Join("\n", conversations.Select(x => $"{x.Role}: {x.Content.Replace("user:", "")}")).Trim();
@ -36,6 +36,6 @@ public class ChatCompletionProvider : IChatCompletion
totalResponse += response; totalResponse += response;
} }
return Task.FromResult(totalResponse.Trim()); return true;
} }
} }

View file

@ -86,7 +86,7 @@ public class ChatCompletionProvider : IChatCompletion
return functions; return functions;
} }
public async Task<string> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations) public async Task<bool> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations, Func<RoleDialogModel, Task> onMessageReceived)
{ {
var client = new OpenAIClient(new Uri(_settings.Endpoint), new AzureKeyCredential(_settings.ApiKey)); var client = new OpenAIClient(new Uri(_settings.Endpoint), new AzureKeyCredential(_settings.ApiKey));
var chatCompletionsOptions = PrepareOptions(agent, conversations); var chatCompletionsOptions = PrepareOptions(agent, conversations);
@ -97,16 +97,23 @@ public class ChatCompletionProvider : IChatCompletion
string output = ""; string output = "";
await foreach (var choice in streaming.GetChoicesStreaming()) await foreach (var choice in streaming.GetChoicesStreaming())
{ {
if (choice.FinishReason == CompletionsFinishReason.FunctionCall)
{
}
await foreach (var message in choice.GetMessageStreaming()) await foreach (var message in choice.GetMessageStreaming())
{ {
if (message.Content == null) if (message.Content == null)
continue; continue;
Console.Write(message.Content); Console.Write(message.Content);
output += message.Content; output += message.Content;
await onMessageReceived(new RoleDialogModel(message.Role.ToString(), message.Content));
} }
output = "";
} }
return output.Trim(); return true;
} }
private ChatCompletionsOptions PrepareOptions(Agent agent, List<RoleDialogModel> conversations) private ChatCompletionsOptions PrepareOptions(Agent agent, List<RoleDialogModel> conversations)

View file

@ -78,22 +78,24 @@ public class ChatbotUiController : ControllerBase, IApiAdapter
converation = await conversationService.NewConversation(sess); converation = await conversationService.NewConversation(sess);
} }
var result = await conversationService.SendMessage(input.AgentId, input.ConversationId, conversations); var result = await conversationService.SendMessage(input.AgentId,
input.ConversationId,
conversations,
async msg =>
await OnChunkReceived(outputStream, msg));
await OnChunkReceived(outputStream, result);
await OnEventCompleted(outputStream); await OnEventCompleted(outputStream);
} }
private async Task OnChunkReceived(Stream outputStream, string content) private async Task OnChunkReceived(Stream outputStream, RoleDialogModel message)
{ {
var response = new OpenAiChatOutput var response = new OpenAiChatOutput
{ {
Choices = new List<OpenAiChoice> Choices = new List<OpenAiChoice>
{ {
new OpenAiChoice new OpenAiChoice
{ {
Delta = new RoleDialogModel("assistant", content) Delta = new RoleDialogModel(message.Role, message.Content)
} }
} }
}; };
@ -105,7 +107,7 @@ public class ChatbotUiController : ControllerBase, IApiAdapter
var buffer = Encoding.UTF8.GetBytes($"data:{json}\n"); var buffer = Encoding.UTF8.GetBytes($"data:{json}\n");
await outputStream.WriteAsync(buffer, 0, buffer.Length); await outputStream.WriteAsync(buffer, 0, buffer.Length);
await Task.Delay(100); await Task.Delay(10);
buffer = Encoding.UTF8.GetBytes("\n"); buffer = Encoding.UTF8.GetBytes("\n");
await outputStream.WriteAsync(buffer, 0, buffer.Length); await outputStream.WriteAsync(buffer, 0, buffer.Length);