SemianiakaVY/BotSharp
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

implement streaming chat interface

Browse source
This commit is contained in:
Jicheng Lu 2025-06-27 13:49:44 -05:00
parent 0aff924dd9
commit aac09ca3bd
21 changed files with 385 additions and 124 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
src/Infrastructure/BotSharp.Abstraction/MLTasks/IChatCompletion.cs
View file

@ -24,5 +24,5 @@ public interface IChatCompletion
Func<RoleDialogModel, Task> onFunctionExecuting);
Task<RoleDialogModel> GetChatCompletionsStreamingAsync(Agent agent,
List<RoleDialogModel> conversations) => Task.FromResult(new RoleDialogModel(AgentRole.Assistant, string.Empty));
List<RoleDialogModel> conversations);
}

3
src/Infrastructure/BotSharp.Core/Conversations/ConversationPlugin.cs
View file

@ -12,6 +12,7 @@ using BotSharp.Core.Templating;
using BotSharp.Core.Translation;
using BotSharp.Core.Observables.Queues;
using Microsoft.Extensions.Configuration;
using BotSharp.Abstraction.Observables.Models;
namespace BotSharp.Core.Conversations;
@ -42,7 +43,7 @@ public class ConversationPlugin : IBotSharpPlugin
return settingService.Bind<GoogleApiSettings>("GoogleApi");
});
services.AddSingleton<MessageHub>();
services.AddSingleton<MessageHub<HubObserveData>>();
services.AddScoped<IConversationStorage, ConversationStorage>();
services.AddScoped<IConversationService, ConversationService>();

13
src/Infrastructure/BotSharp.Core/Observables/Queues/MessageHub.cs
View file

@ -1,15 +1,14 @@
using BotSharp.Abstraction.Observables.Models;
using System.Reactive.Subjects;
namespace BotSharp.Core.Observables.Queues;
public class MessageHub
public class MessageHub<T> where T : class
{
private readonly ILogger<MessageHub> _logger;
private readonly ISubject<HubObserveData> _observable = new Subject<HubObserveData>();
public IObservable<HubObserveData> Events => _observable;
private readonly ILogger<MessageHub<T>> _logger;
private readonly ISubject<T> _observable = new Subject<T>();
public IObservable<T> Events => _observable;
public MessageHub(ILogger<MessageHub> logger)
public MessageHub(ILogger<MessageHub<T>> logger)
{
_logger = logger;
}
@ -18,7 +17,7 @@ public class MessageHub
/// Push an item to the observers.
/// </summary>
/// <param name="item"></param>
public void Push(HubObserveData item)
public void Push(T item)
{
_observable.OnNext(item);
}

3
src/Plugins/BotSharp.Plugin.AnthropicAI/Providers/ChatCompletionProvider.cs
View file

@ -96,8 +96,7 @@ public class ChatCompletionProvider : IChatCompletion
throw new NotImplementedException();
}
public Task<bool> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations,
Func<RoleDialogModel, Task> onMessageReceived)
public Task<RoleDialogModel> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations)
{
throw new NotImplementedException();
}

2
src/Plugins/BotSharp.Plugin.AzureOpenAI/BotSharp.Plugin.AzureOpenAI.csproj
View file

@ -16,7 +16,7 @@
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\Infrastructure\BotSharp.Abstraction\BotSharp.Abstraction.csproj" />
<ProjectReference Include="..\..\Infrastructure\BotSharp.Core\BotSharp.Core.csproj" />
</ItemGroup>
</Project>

135
src/Plugins/BotSharp.Plugin.AzureOpenAI/Providers/Chat/ChatCompletionProvider.cs
View file

@ -1,6 +1,9 @@
using Azure;
using BotSharp.Abstraction.Files.Utilities;
using BotSharp.Abstraction.Hooks;
using BotSharp.Abstraction.Observables.Models;
using BotSharp.Core.Infrastructures.Streams;
using BotSharp.Core.Observables.Queues;
using OpenAI.Chat;
using System.ClientModel;
@ -203,39 +206,133 @@ public class ChatCompletionProvider : IChatCompletion
return true;
}
public async Task<bool> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations, Func<RoleDialogModel, Task> onMessageReceived)
public async Task<RoleDialogModel> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations)
{
var client = ProviderHelper.GetClient(Provider, _model, _services);
var chatClient = client.GetChatClient(_model);
var (prompt, messages, options) = PrepareOptions(agent, conversations);
var response = chatClient.CompleteChatStreamingAsync(messages, options);
var hub = _services.GetRequiredService<MessageHub<HubObserveData>>();
var messageId = conversations.LastOrDefault()?.MessageId ?? string.Empty;
await foreach (var choice in response)
var contentHooks = _services.GetHooks<IContentGeneratingHook>(agent.Id);
// Before chat completion hook
foreach (var hook in contentHooks)
{
if (choice.FinishReason == ChatFinishReason.FunctionCall || choice.FinishReason == ChatFinishReason.ToolCalls)
{
var update = choice.ToolCallUpdates?.FirstOrDefault()?.FunctionArgumentsUpdate?.ToString() ?? string.Empty;
Console.Write(update);
await hook.BeforeGenerating(agent, conversations);
}
await onMessageReceived(new RoleDialogModel(AgentRole.Assistant, update)
{
RenderedInstruction = string.Join("\r\n", renderedInstructions)
});
continue;
hub.Push(new()
{
ServiceProvider = _services,
EventName = "BeforeReceiveLlmStreamMessage",
Data = new RoleDialogModel(AgentRole.Assistant, string.Empty)
{
CurrentAgentId = agent.Id,
MessageId = messageId
}
});
using var textStream = new RealtimeTextStream();
var toolCalls = new List<StreamingChatToolCallUpdate>();
ChatTokenUsage? tokenUsage = null;
var responseMessage = new RoleDialogModel(AgentRole.Assistant, string.Empty)
{
CurrentAgentId = agent.Id,
MessageId = messageId
};
await foreach (var choice in chatClient.CompleteChatStreamingAsync(messages, options))
{
tokenUsage = choice.Usage;
if (!choice.ToolCallUpdates.IsNullOrEmpty())
{
toolCalls.AddRange(choice.ToolCallUpdates);
}
if (choice.ContentUpdate.IsNullOrEmpty()) continue;
_logger.LogInformation(choice.ContentUpdate[0]?.Text);
await onMessageReceived(new RoleDialogModel(choice.Role?.ToString() ?? ChatMessageRole.Assistant.ToString(), choice.ContentUpdate[0]?.Text ?? string.Empty)
if (!choice.ContentUpdate.IsNullOrEmpty())
{
RenderedInstruction = string.Join("\r\n", renderedInstructions)
var text = choice.ContentUpdate[0]?.Text ?? string.Empty;
textStream.Collect(text);
#if DEBUG
_logger.LogCritical($"Content update: {text}");
#endif
var content = new RoleDialogModel(AgentRole.Assistant, text)
{
CurrentAgentId = agent.Id,
MessageId = messageId
};
hub.Push(new()
{
ServiceProvider = _services,
EventName = "OnReceiveLlmStreamMessage",
Data = content
});
}
if (choice.FinishReason == ChatFinishReason.ToolCalls || choice.FinishReason == ChatFinishReason.FunctionCall)
{
var meta = toolCalls.FirstOrDefault(x => !string.IsNullOrEmpty(x.FunctionName));
var functionName = meta?.FunctionName;
var toolCallId = meta?.ToolCallId;
var args = toolCalls.Where(x => x.FunctionArgumentsUpdate != null).Select(x => x.FunctionArgumentsUpdate.ToString()).ToList();
var functionArgument = string.Join(string.Empty, args);
#if DEBUG
_logger.LogCritical($"Tool Call (id: {toolCallId}) => {functionName}({functionArgument})");
#endif
responseMessage = new RoleDialogModel(AgentRole.Function, string.Empty)
{
CurrentAgentId = agent.Id,
MessageId = messageId,
ToolCallId = toolCallId,
FunctionName = functionName,
FunctionArgs = functionArgument
};
}
else if (choice.FinishReason.HasValue)
{
var allText = textStream.GetText();
_logger.LogCritical($"Text Content: {allText}");
responseMessage = new RoleDialogModel(AgentRole.Assistant, allText)
{
CurrentAgentId = agent.Id,
MessageId = messageId,
IsStreaming = true
};
}
}
hub.Push(new()
{
ServiceProvider = _services,
EventName = "AfterReceiveLlmStreamMessage",
Data = responseMessage
});
var inputTokenDetails = tokenUsage?.InputTokenDetails;
// After chat completion hook
foreach (var hook in contentHooks)
{
await hook.AfterGenerated(responseMessage, new TokenStatsModel
{
Prompt = prompt,
Provider = Provider,
Model = _model,
TextInputTokens = (tokenUsage?.InputTokenCount ?? 0) - (inputTokenDetails?.CachedTokenCount ?? 0),
CachedTextInputTokens = inputTokenDetails?.CachedTokenCount ?? 0,
TextOutputTokens = tokenUsage?.OutputTokenCount ?? 0
});
}
return true;
return responseMessage;
}
protected (string, IEnumerable<ChatMessage>, ChatCompletionOptions) PrepareOptions(Agent agent, List<RoleDialogModel> conversations)

5
src/Plugins/BotSharp.Plugin.ChatHub/ChatHubPlugin.cs
View file

@ -1,4 +1,5 @@
using BotSharp.Abstraction.Crontab;
using BotSharp.Abstraction.Observables.Models;
using BotSharp.Core.Observables.Queues;
using BotSharp.Plugin.ChatHub.Hooks;
using BotSharp.Plugin.ChatHub.Observers;
@ -35,8 +36,8 @@ public class ChatHubPlugin : IBotSharpPlugin, IBotSharpAppPlugin
public void Configure(IApplicationBuilder app)
{
var services = app.ApplicationServices;
var queue = services.GetRequiredService<MessageHub>();
var logger = services.GetRequiredService<ILogger<MessageHub>>();
var queue = services.GetRequiredService<MessageHub<HubObserveData>>();
var logger = services.GetRequiredService<ILogger<MessageHub<HubObserveData>>>();
queue.Events.Subscribe(new ChatHubObserver(logger));
}
}

2
src/Plugins/BotSharp.Plugin.DeepSeekAI/BotSharp.Plugin.DeepSeekAI.csproj
View file

@ -15,7 +15,7 @@
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\Infrastructure\BotSharp.Abstraction\BotSharp.Abstraction.csproj" />
<ProjectReference Include="..\..\Infrastructure\BotSharp.Core\BotSharp.Core.csproj" />
</ItemGroup>
</Project>

141
src/Plugins/BotSharp.Plugin.DeepSeekAI/Providers/Chat/ChatCompletionProvider.cs
View file

@ -1,8 +1,11 @@
using BotSharp.Abstraction.Files;
using BotSharp.Abstraction.Hooks;
using BotSharp.Abstraction.Observables.Models;
using BotSharp.Core.Infrastructures.Streams;
using BotSharp.Core.Observables.Queues;
using BotSharp.Plugin.DeepSeek.Providers;
using Microsoft.Extensions.Logging;
using OpenAI.Chat;
using BotSharp.Abstraction.Files;
using BotSharp.Plugin.DeepSeek.Providers;
using BotSharp.Abstraction.Hooks;
namespace BotSharp.Plugin.DeepSeekAI.Providers.Chat;
@ -170,39 +173,133 @@ public class ChatCompletionProvider : IChatCompletion
return true;
}
public async Task<bool> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations, Func<RoleDialogModel, Task> onMessageReceived)
public async Task<RoleDialogModel> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations)
{
var client = ProviderHelper.GetClient(Provider, _model, _services);
var chatClient = client.GetChatClient(_model);
var (prompt, messages, options) = PrepareOptions(agent, conversations);
var response = chatClient.CompleteChatStreamingAsync(messages, options);
var hub = _services.GetRequiredService<MessageHub<HubObserveData>>();
var messageId = conversations.LastOrDefault()?.MessageId ?? string.Empty;
await foreach (var choice in response)
var contentHooks = _services.GetHooks<IContentGeneratingHook>(agent.Id);
// Before chat completion hook
foreach (var hook in contentHooks)
{
if (choice.FinishReason == ChatFinishReason.FunctionCall || choice.FinishReason == ChatFinishReason.ToolCalls)
{
var update = choice.ToolCallUpdates?.FirstOrDefault()?.FunctionArgumentsUpdate?.ToString() ?? string.Empty;
_logger.LogInformation(update);
await hook.BeforeGenerating(agent, conversations);
}
await onMessageReceived(new RoleDialogModel(AgentRole.Assistant, update)
{
RenderedInstruction = string.Join("\r\n", renderedInstructions)
});
continue;
hub.Push(new()
{
ServiceProvider = _services,
EventName = "BeforeReceiveLlmStreamMessage",
Data = new RoleDialogModel(AgentRole.Assistant, string.Empty)
{
CurrentAgentId = agent.Id,
MessageId = messageId
}
});
using var textStream = new RealtimeTextStream();
var toolCalls = new List<StreamingChatToolCallUpdate>();
ChatTokenUsage? tokenUsage = null;
var responseMessage = new RoleDialogModel(AgentRole.Assistant, string.Empty)
{
CurrentAgentId = agent.Id,
MessageId = messageId
};
await foreach (var choice in chatClient.CompleteChatStreamingAsync(messages, options))
{
tokenUsage = choice.Usage;
if (!choice.ToolCallUpdates.IsNullOrEmpty())
{
toolCalls.AddRange(choice.ToolCallUpdates);
}
if (choice.ContentUpdate.IsNullOrEmpty()) continue;
_logger.LogInformation(choice.ContentUpdate[0]?.Text);
await onMessageReceived(new RoleDialogModel(choice.Role?.ToString() ?? ChatMessageRole.Assistant.ToString(), choice.ContentUpdate[0]?.Text ?? string.Empty)
if (!choice.ContentUpdate.IsNullOrEmpty())
{
RenderedInstruction = string.Join("\r\n", renderedInstructions)
var text = choice.ContentUpdate[0]?.Text ?? string.Empty;
textStream.Collect(text);
#if DEBUG
_logger.LogCritical($"Content update: {text}");
#endif
var content = new RoleDialogModel(AgentRole.Assistant, text)
{
CurrentAgentId = agent.Id,
MessageId = messageId
};
hub.Push(new()
{
ServiceProvider = _services,
EventName = "OnReceiveLlmStreamMessage",
Data = content
});
}
if (choice.FinishReason == ChatFinishReason.ToolCalls || choice.FinishReason == ChatFinishReason.FunctionCall)
{
var meta = toolCalls.FirstOrDefault(x => !string.IsNullOrEmpty(x.FunctionName));
var functionName = meta?.FunctionName;
var toolCallId = meta?.ToolCallId;
var args = toolCalls.Where(x => x.FunctionArgumentsUpdate != null).Select(x => x.FunctionArgumentsUpdate.ToString()).ToList();
var functionArgument = string.Join(string.Empty, args);
#if DEBUG
_logger.LogCritical($"Tool Call (id: {toolCallId}) => {functionName}({functionArgument})");
#endif
responseMessage = new RoleDialogModel(AgentRole.Function, string.Empty)
{
CurrentAgentId = agent.Id,
MessageId = messageId,
ToolCallId = toolCallId,
FunctionName = functionName,
FunctionArgs = functionArgument
};
}
else if (choice.FinishReason.HasValue)
{
var allText = textStream.GetText();
_logger.LogCritical($"Text Content: {allText}");
responseMessage = new RoleDialogModel(AgentRole.Assistant, allText)
{
CurrentAgentId = agent.Id,
MessageId = messageId,
IsStreaming = true
};
}
}
hub.Push(new()
{
ServiceProvider = _services,
EventName = "AfterReceiveLlmStreamMessage",
Data = responseMessage
});
var inputTokenDetails = tokenUsage?.InputTokenDetails;
// After chat completion hook
foreach (var hook in contentHooks)
{
await hook.AfterGenerated(responseMessage, new TokenStatsModel
{
Prompt = prompt,
Provider = Provider,
Model = _model,
TextInputTokens = (tokenUsage?.InputTokenCount ?? 0) - (inputTokenDetails?.CachedTokenCount ?? 0),
CachedTextInputTokens = inputTokenDetails?.CachedTokenCount ?? 0,
TextOutputTokens = tokenUsage?.OutputTokenCount ?? 0
});
}
return true;
return responseMessage;
}
public void SetModelName(string model)

35
src/Plugins/BotSharp.Plugin.GoogleAI/Providers/Chat/GeminiChatCompletionProvider.cs
View file

@ -159,40 +159,9 @@ public class GeminiChatCompletionProvider : IChatCompletion
return true;
}
public async Task<bool> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations, Func<RoleDialogModel, Task> onMessageReceived)
public Task<RoleDialogModel> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations)
{
var client = ProviderHelper.GetGeminiClient(Provider, _model, _services);
var chatClient = client.CreateGenerativeModel(_model.ToModelId());
var (prompt, messages) = PrepareOptions(chatClient,agent, conversations);
var asyncEnumerable = chatClient.StreamContentAsync(messages);
await foreach (var response in asyncEnumerable)
{
if (response.GetFunction() != null)
{
var func = response.GetFunction();
var update = func?.Args?.ToJsonString().ToString() ?? string.Empty;
_logger.LogInformation(update);
await onMessageReceived(new RoleDialogModel(AgentRole.Assistant, update)
{
RenderedInstruction = string.Join("\r\n", renderedInstructions)
});
continue;
}
if (response.Text().IsNullOrEmpty()) continue;
_logger.LogInformation(response.Text());
await onMessageReceived(new RoleDialogModel(response.Candidates?.LastOrDefault()?.Content?.Role?.ToString() ?? AgentRole.Assistant.ToString(), response.Text() ?? string.Empty)
{
RenderedInstruction = string.Join("\r\n", renderedInstructions)
});
}
return true;
throw new NotImplementedException();
}
public void SetModelName(string model)

2
src/Plugins/BotSharp.Plugin.GoogleAI/Providers/Chat/PalmChatCompletionProvider.cs
View file

@ -145,7 +145,7 @@ public class PalmChatCompletionProvider : IChatCompletion
throw new NotImplementedException();
}
public Task<bool> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations, Func<RoleDialogModel, Task> onMessageReceived)
public Task<RoleDialogModel> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations)
{
throw new NotImplementedException();
}

4
src/Plugins/BotSharp.Plugin.HuggingFace/Providers/ChatCompletionProvider.cs
View file

@ -76,9 +76,9 @@ public class ChatCompletionProvider : IChatCompletion
return true;
}
public async Task<bool> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations, Func<RoleDialogModel, Task> onMessageReceived)
public Task<RoleDialogModel> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations)
{
return true;
throw new NotImplementedException();
}
public void SetModelName(string model)

2
src/Plugins/BotSharp.Plugin.LLamaSharp/BotSharp.Plugin.LLamaSharp.csproj
View file

@ -15,7 +15,7 @@
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\Infrastructure\BotSharp.Abstraction\BotSharp.Abstraction.csproj" />
<ProjectReference Include="..\..\Infrastructure\BotSharp.Core\BotSharp.Core.csproj" />
</ItemGroup>
</Project>

64
src/Plugins/BotSharp.Plugin.LLamaSharp/Providers/ChatCompletionProvider.cs
View file

@ -1,5 +1,12 @@
using BotSharp.Abstraction.Agents;
using BotSharp.Abstraction.Hooks;
using BotSharp.Abstraction.Loggers;
using BotSharp.Abstraction.Observables.Models;
using BotSharp.Core.Infrastructures.Streams;
using BotSharp.Core.Observables.Queues;
using Microsoft.AspNetCore.SignalR;
using static LLama.Common.ChatHistory;
using static System.Net.Mime.MediaTypeNames;
namespace BotSharp.Plugin.LLamaSharp.Providers;
@ -159,12 +166,8 @@ public class ChatCompletionProvider : IChatCompletion
return true;
}
public async Task<bool> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations, Func<RoleDialogModel, Task> onMessageReceived)
public async Task<RoleDialogModel> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations)
{
string totalResponse = "";
var content = string.Join("\r\n", conversations.Select(x => $"{x.Role}: {x.Content}")).Trim();
content += $"\r\n{AgentRole.Assistant}: ";
var state = _services.GetRequiredService<IConversationStateService>();
var model = state.GetState("model", "llama-2-7b-chat.Q8_0");
@ -180,13 +183,60 @@ public class ChatCompletionProvider : IChatCompletion
_logger.LogInformation(agent.Instruction);
}
var hub = _services.GetRequiredService<MessageHub<HubObserveData>>();
var messageId = conversations.LastOrDefault()?.MessageId ?? string.Empty;
hub.Push(new()
{
ServiceProvider = _services,
EventName = "BeforeReceiveLlmStreamMessage",
Data = new RoleDialogModel(AgentRole.Assistant, string.Empty)
{
CurrentAgentId = agent.Id,
MessageId = messageId
}
});
using var textStream = new RealtimeTextStream();
var responseMessage = new RoleDialogModel(AgentRole.Assistant, string.Empty)
{
CurrentAgentId = agent.Id,
MessageId = messageId
};
await foreach (var response in executor.InferAsync(agent.Instruction, inferenceParams))
{
Console.Write(response);
totalResponse += response;
textStream.Collect(response);
var content = new RoleDialogModel(AgentRole.Assistant, response)
{
CurrentAgentId = agent.Id,
MessageId = messageId
};
hub.Push(new()
{
ServiceProvider = _services,
EventName = "OnReceiveLlmStreamMessage",
Data = content
});
}
return true;
responseMessage = new RoleDialogModel(AgentRole.Assistant, textStream.GetText())
{
CurrentAgentId = agent.Id,
MessageId = messageId,
IsStreaming = true
};
hub.Push(new()
{
ServiceProvider = _services,
EventName = "AfterReceiveLlmStreamMessage",
Data = responseMessage
});
return responseMessage;
}
public void SetModelName(string model)

2
src/Plugins/BotSharp.Plugin.LangChain/Providers/ChatCompletionProvider.cs
View file

@ -65,7 +65,7 @@ namespace BotSharp.Plugin.VertexAI.Providers
throw new NotImplementedException();
}
public Task<bool> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations, Func<RoleDialogModel, Task> onMessageReceived)
public Task<RoleDialogModel> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations)
{
throw new NotImplementedException();
}

2
src/Plugins/BotSharp.Plugin.MetaGLM/Providers/ChatCompletionProvider.cs
View file

@ -235,7 +235,7 @@ public class ChatCompletionProvider : IChatCompletion
throw new NotImplementedException();
}
public Task<bool> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations, Func<RoleDialogModel, Task> onMessageReceived)
public Task<RoleDialogModel> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations)
{
throw new NotImplementedException();
}

4
src/Plugins/BotSharp.Plugin.MicrosoftExtensionsAI/MicrosoftExtensionsAIChatCompletionProvider.cs
View file

@ -169,8 +169,10 @@ public sealed class MicrosoftExtensionsAIChatCompletionProvider : IChatCompletio
throw new NotImplementedException();
/// <inheritdoc/>
public Task<bool> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations, Func<RoleDialogModel, Task> onMessageReceived) =>
public Task<RoleDialogModel> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations)
{
throw new NotImplementedException();
}
private sealed class NopAIFunction(string name, string description, JsonElement schema) : AIFunction
{

5
src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.cs
View file

@ -1,5 +1,6 @@
using Azure;
using BotSharp.Abstraction.Hooks;
using BotSharp.Abstraction.Observables.Models;
using BotSharp.Core.Infrastructures.Streams;
using BotSharp.Core.Observables.Queues;
using BotSharp.Plugin.OpenAI.Models.Realtime;
@ -190,7 +191,7 @@ public class ChatCompletionProvider : IChatCompletion
var chatClient = client.GetChatClient(_model);
var (prompt, messages, options) = PrepareOptions(agent, conversations);
var hub = _services.GetRequiredService<MessageHub>();
var hub = _services.GetRequiredService<MessageHub<HubObserveData>>();
var messageId = conversations.LastOrDefault()?.MessageId ?? string.Empty;
var contentHooks = _services.GetHooks<IContentGeneratingHook>(agent.Id);
@ -210,7 +211,6 @@ public class ChatCompletionProvider : IChatCompletion
MessageId = messageId
}
});
using var textStream = new RealtimeTextStream();
var toolCalls = new List<StreamingChatToolCallUpdate>();
@ -273,7 +273,6 @@ public class ChatCompletionProvider : IChatCompletion
FunctionName = functionName,
FunctionArgs = functionArgument
};
}
else if (choice.FinishReason.HasValue)
{

2
src/Plugins/BotSharp.Plugin.SemanticKernel/SemanticKernelChatCompletionProvider.cs
View file

@ -94,7 +94,7 @@ namespace BotSharp.Plugin.SemanticKernel
throw new NotImplementedException();
}
/// <inheritdoc/>
public Task<bool> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations, Func<RoleDialogModel, Task> onMessageReceived)
public Task<RoleDialogModel> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations)
{
throw new NotImplementedException();
}

2
src/Plugins/BotSharp.Plugin.SparkDesk/BotSharp.Plugin.SparkDesk.csproj
View file

@ -15,7 +15,7 @@
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\Infrastructure\BotSharp.Abstraction\BotSharp.Abstraction.csproj" />
<ProjectReference Include="..\..\Infrastructure\BotSharp.Core\BotSharp.Core.csproj" />
</ItemGroup>
</Project>

79
src/Plugins/BotSharp.Plugin.SparkDesk/Providers/ChatCompletionProvider.cs
View file

@ -1,6 +1,10 @@
using BotSharp.Abstraction.Agents;
using BotSharp.Abstraction.Agents.Enums;
using BotSharp.Abstraction.Loggers;
using BotSharp.Abstraction.Observables.Models;
using BotSharp.Core.Infrastructures.Streams;
using BotSharp.Core.Observables.Queues;
using Microsoft.AspNetCore.SignalR;
namespace BotSharp.Plugin.SparkDesk.Providers;
@ -143,34 +147,77 @@ public class ChatCompletionProvider : IChatCompletion
return true;
}
public async Task<bool> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations, Func<RoleDialogModel, Task> onMessageReceived)
public async Task<RoleDialogModel> GetChatCompletionsStreamingAsync(Agent agent, List<RoleDialogModel> conversations)
{
var client = new SparkDeskClient(appId: _settings.AppId, apiKey: _settings.ApiKey, apiSecret: _settings.ApiSecret);
var (prompt, messages, funcall) = PrepareOptions(agent, conversations);
var messageId = conversations.LastOrDefault()?.MessageId ?? string.Empty;
var hub = _services.GetRequiredService<MessageHub<HubObserveData>>();
hub.Push(new()
{
ServiceProvider = _services,
EventName = "BeforeReceiveLlmStreamMessage",
Data = new RoleDialogModel(AgentRole.Assistant, string.Empty)
{
CurrentAgentId = agent.Id,
MessageId = messageId
}
});
var responseMessage = new RoleDialogModel(AgentRole.Assistant, string.Empty)
{
CurrentAgentId = agent.Id,
MessageId = messageId
};
using var textStream = new RealtimeTextStream();
await foreach (StreamedChatResponse response in client.ChatAsStreamAsync(modelVersion: _settings.ModelVersion, messages, functions: funcall.Length == 0 ? null : funcall))
{
if (response.FunctionCall !=null)
if (response.FunctionCall != null)
{
await onMessageReceived(new RoleDialogModel(AgentRole.Function, response.Text)
{
responseMessage = new RoleDialogModel(AgentRole.Function, string.Empty)
{
CurrentAgentId = agent.Id,
MessageId = messageId,
ToolCallId = response.FunctionCall.Name,
FunctionName = response.FunctionCall.Name,
FunctionArgs = response.FunctionCall.Arguments,
RenderedInstruction = string.Join("\r\n", renderedInstructions)
});
continue;
FunctionArgs = response.FunctionCall.Arguments
};
}
await onMessageReceived(new RoleDialogModel(AgentRole.Assistant, response.Text)
else
{
CurrentAgentId = agent.Id,
RenderedInstruction = string.Join("\r\n", renderedInstructions)
});
}
textStream.Collect(response.Text);
responseMessage = new RoleDialogModel(AgentRole.Assistant, response.Text)
{
CurrentAgentId = agent.Id,
MessageId = messageId
};
return true;
hub.Push(new()
{
ServiceProvider = _services,
EventName = "OnReceiveLlmStreamMessage",
Data = responseMessage
});
}
}
if (responseMessage.Role == AgentRole.Assistant)
{
responseMessage.Content = textStream.GetText();
responseMessage.IsStreaming = true;
}
hub.Push(new()
{
ServiceProvider = _services,
EventName = "AfterReceiveLlmStreamMessage",
Data = responseMessage
});
return responseMessage;
}
public void SetModelName(string model)