From 6ecd51b49cf4f45ac02f5ed87ae388f29b22f913 Mon Sep 17 00:00:00 2001 From: "jason.wang" Date: Fri, 8 Nov 2024 14:51:56 +0800 Subject: [PATCH 1/6] fix affiliate claim name --- .../BotSharp.Core/Users/Services/UserService.cs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/Infrastructure/BotSharp.Core/Users/Services/UserService.cs b/src/Infrastructure/BotSharp.Core/Users/Services/UserService.cs index daf8a340..441908f3 100644 --- a/src/Infrastructure/BotSharp.Core/Users/Services/UserService.cs +++ b/src/Infrastructure/BotSharp.Core/Users/Services/UserService.cs @@ -317,8 +317,8 @@ public class UserService : IUserService new Claim("role", user.Role ?? UserRole.User), new Claim(JwtRegisteredClaimNames.Jti, Guid.NewGuid().ToString()), new Claim("phone", user.Phone ?? string.Empty), - new Claim("affiliateId", user.AffiliateId ?? string.Empty), - new Claim("employeeId", user.EmployeeId ?? string.Empty), + new Claim("affiliate_id", user.AffiliateId ?? string.Empty), + new Claim("employee_id", user.EmployeeId ?? string.Empty), new Claim("regionCode", user.RegionCode ?? "CN") }; From cb6f788295d41a7c5a6ec57aa5685a446fac71bd Mon Sep 17 00:00:00 2001 From: Haiping Chen Date: Fri, 8 Nov 2024 18:10:15 +0000 Subject: [PATCH 2/6] Add eventbus in Redis --- .../Infrastructures/Events/IEventPublisher.cs | 14 ++++ .../Events/IEventSubscriber.cs | 8 ++ .../BotSharp.Core/BotSharpCoreExtensions.cs | 19 ++++- .../Infrastructures/DistributedLocker.cs | 31 ++------ .../Infrastructures/Events/RedisPublisher.cs | 32 ++++++++ .../Infrastructures/Events/RedisSubscriber.cs | 73 +++++++++++++++++++ src/Infrastructure/BotSharp.Core/Using.cs | 3 +- .../BotSharpOpenApiExtensions.cs | 2 +- 8 files changed, 152 insertions(+), 30 deletions(-) create mode 100644 src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventPublisher.cs create mode 100644 src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventSubscriber.cs create mode 100644 src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs create mode 100644 src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs diff --git a/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventPublisher.cs b/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventPublisher.cs new file mode 100644 index 00000000..471d810e --- /dev/null +++ b/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventPublisher.cs @@ -0,0 +1,14 @@ +namespace BotSharp.Abstraction.Infrastructures.Events; + +public interface IEventPublisher +{ + /// + /// Boardcast message to all subscribers + /// + /// + /// + /// + Task BroadcastAsync(string channel, string message); + + Task PublishAsync(string channel, string message); +} diff --git a/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventSubscriber.cs b/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventSubscriber.cs new file mode 100644 index 00000000..f295e54e --- /dev/null +++ b/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventSubscriber.cs @@ -0,0 +1,8 @@ +namespace BotSharp.Abstraction.Infrastructures.Events; + +public interface IEventSubscriber +{ + Task SubscribeAsync(string channel, Func received); + + Task SubscribeAsync(string channel, string group, Func received); +} diff --git a/src/Infrastructure/BotSharp.Core/BotSharpCoreExtensions.cs b/src/Infrastructure/BotSharp.Core/BotSharpCoreExtensions.cs index 4bdfe0d7..812d89a8 100644 --- a/src/Infrastructure/BotSharp.Core/BotSharpCoreExtensions.cs +++ b/src/Infrastructure/BotSharp.Core/BotSharpCoreExtensions.cs @@ -9,6 +9,8 @@ using BotSharp.Abstraction.Users.Settings; using BotSharp.Abstraction.Interpreters.Settings; using BotSharp.Abstraction.Infrastructures; using BotSharp.Core.Processors; +using StackExchange.Redis; +using BotSharp.Core.Infrastructures.Events; namespace BotSharp.Core; @@ -34,10 +36,12 @@ public static class BotSharpCoreExtensions services.AddSingleton(x => cacheSettings); services.AddSingleton(); + AddRedisEvents(services, config); + services.AddMemoryCache(); RegisterPlugins(services, config); - ConfigureBotSharpOptions(services, configOptions); + AddBotSharpOptions(services, configOptions); return services; } @@ -79,7 +83,7 @@ public static class BotSharpCoreExtensions return app; } - private static void ConfigureBotSharpOptions(IServiceCollection services, Action? configure) + private static void AddBotSharpOptions(IServiceCollection services, Action? configure) { var options = new BotSharpOptions(); if (configure != null) @@ -91,6 +95,17 @@ public static class BotSharpCoreExtensions services.AddSingleton(options); } + private static void AddRedisEvents(IServiceCollection services, IConfiguration config) + { + // Add Redis connection as a singleton + var dbSettings = new BotSharpDatabaseSettings(); + config.Bind("Database", dbSettings); + + services.AddSingleton(ConnectionMultiplexer.Connect(dbSettings.Redis)); + services.AddSingleton(); + services.AddSingleton(); + } + private static void AddDefaultJsonConverters(BotSharpOptions options) { options.JsonSerializerOptions.Converters.Add(new RichContentJsonConverter()); diff --git a/src/Infrastructure/BotSharp.Core/Infrastructures/DistributedLocker.cs b/src/Infrastructure/BotSharp.Core/Infrastructures/DistributedLocker.cs index 25c3a047..e37145ff 100644 --- a/src/Infrastructure/BotSharp.Core/Infrastructures/DistributedLocker.cs +++ b/src/Infrastructure/BotSharp.Core/Infrastructures/DistributedLocker.cs @@ -5,21 +5,18 @@ namespace BotSharp.Core.Infrastructures; public class DistributedLocker { - private readonly BotSharpDatabaseSettings _settings; - private static ConnectionMultiplexer connection; + private readonly IConnectionMultiplexer _redis; - public DistributedLocker(BotSharpDatabaseSettings settings) + public DistributedLocker(IConnectionMultiplexer redis) { - _settings = settings; + _redis = redis; } public async Task Lock(string resource, Func> action, int timeoutInSeconds = 30) { - await ConnectToRedisAsync(); - var timeout = TimeSpan.FromSeconds(timeoutInSeconds); - var @lock = new RedisDistributedLock(resource, connection.GetDatabase()); + var @lock = new RedisDistributedLock(resource, _redis.GetDatabase()); await using (var handle = await @lock.TryAcquireAsync(timeout)) { if (handle == null) @@ -33,11 +30,9 @@ public class DistributedLocker public void Lock(string resource, Action action, int timeoutInSeconds = 30) { - ConnectToRedis(); - var timeout = TimeSpan.FromSeconds(timeoutInSeconds); - var @lock = new RedisDistributedLock(resource, connection.GetDatabase()); + var @lock = new RedisDistributedLock(resource, _redis.GetDatabase()); using (var handle = @lock.TryAcquire(timeout)) { if (handle == null) @@ -50,20 +45,4 @@ public class DistributedLocker } } } - - private void ConnectToRedis() - { - if (connection == null) - { - connection = ConnectionMultiplexer.Connect(_settings.Redis); - } - } - - private async Task ConnectToRedisAsync() - { - if (connection == null) - { - connection = await ConnectionMultiplexer.ConnectAsync(_settings.Redis); - } - } } diff --git a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs new file mode 100644 index 00000000..4f3ad405 --- /dev/null +++ b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs @@ -0,0 +1,32 @@ +using StackExchange.Redis; + +namespace BotSharp.Core.Infrastructures.Events; + +public class RedisPublisher : IEventPublisher +{ + private readonly IConnectionMultiplexer _redis; + private readonly ISubscriber _subscriber; + private readonly ILogger _logger; + + public RedisPublisher(IConnectionMultiplexer redis, ILogger logger) + { + _redis = redis; + _logger = logger; + _subscriber = _redis.GetSubscriber(); + } + + public async Task BroadcastAsync(string channel, string message) + { + await _subscriber.PublishAsync(channel, message); + } + + public async Task PublishAsync(string channel, string message) + { + var db = _redis.GetDatabase(); + // Add a message to the stream, keeping only the latest 1 million messages + await db.StreamAddAsync(channel, "message", message, + maxLength: 1000 * 10000); + + _logger.LogInformation($"Published message {channel} {message}"); + } +} diff --git a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs new file mode 100644 index 00000000..39bb1f91 --- /dev/null +++ b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs @@ -0,0 +1,73 @@ +using StackExchange.Redis; +using System.Threading.Channels; + +namespace BotSharp.Core.Infrastructures.Events; + +public class RedisSubscriber : IEventSubscriber +{ + private readonly IConnectionMultiplexer _redis; + private readonly ISubscriber _subscriber; + private readonly ILogger _logger; + + public RedisSubscriber(IConnectionMultiplexer redis, ILogger logger) + { + _redis = redis; + _logger = logger; + _subscriber = _redis.GetSubscriber(); + } + + public async Task SubscribeAsync(string channel, Func received) + { + await _subscriber.SubscribeAsync(channel, async (ch, message) => + { + _logger.LogInformation($"Received event from channel: {ch} message: {message}"); + await received(ch, message); + }); + } + + public async Task SubscribeAsync(string channel, string group, Func received) + { + var db = _redis.GetDatabase(); + + // Create the consumer group if it doesn't exist + try + { + await db.StreamCreateConsumerGroupAsync(channel, group, StreamPosition.NewMessages, createStream: true); + } + catch (RedisServerException ex) when (ex.Message.Contains("BUSYGROUP")) + { + // Group already exists, ignore the error + _logger.LogWarning($"Consumer group '{group}' already exists (caught exception)."); + } + catch (Exception ex) + { + _logger.LogError($"Error creating consumer group: '{group}' {ex.Message}"); + throw; + } + + while (true) + { + var entries = await db.StreamReadGroupAsync(channel, group, Environment.MachineName, count: 1); + foreach (var entry in entries) + { + _logger.LogInformation($"Consumer {Environment.MachineName} received: {channel} {entry.Values[0].Value}"); + await db.StreamAcknowledgeAsync(channel, group, entry.Id); + + try + { + await received(channel, entry.Values[0].Value); + + // Optionally delete the message to save space + await db.StreamDeleteAsync(channel, [entry.Id]); + } + catch (Exception ex) + { + _logger.LogError($"Error processing message: {ex.Message}, event id: {channel} {entry.Id}"); + } + } + + await Task.Delay(Random.Shared.Next(1, 11) * 100); + } + + } +} diff --git a/src/Infrastructure/BotSharp.Core/Using.cs b/src/Infrastructure/BotSharp.Core/Using.cs index 8a0ca2af..9e9177ba 100644 --- a/src/Infrastructure/BotSharp.Core/Using.cs +++ b/src/Infrastructure/BotSharp.Core/Using.cs @@ -40,4 +40,5 @@ global using BotSharp.Core.Agents.Services; global using BotSharp.Core.Conversations.Services; global using BotSharp.Core.Infrastructures; global using BotSharp.Core.Users.Services; -global using Aspects.Cache; \ No newline at end of file +global using Aspects.Cache; +global using BotSharp.Abstraction.Infrastructures.Events; \ No newline at end of file diff --git a/src/Infrastructure/BotSharp.OpenAPI/BotSharpOpenApiExtensions.cs b/src/Infrastructure/BotSharp.OpenAPI/BotSharpOpenApiExtensions.cs index efad6e6c..4d85167f 100644 --- a/src/Infrastructure/BotSharp.OpenAPI/BotSharpOpenApiExtensions.cs +++ b/src/Infrastructure/BotSharp.OpenAPI/BotSharpOpenApiExtensions.cs @@ -209,7 +209,7 @@ public static class BotSharpOpenApiExtensions app.UseSwagger(); - // if (env.IsDevelopment()) + if (env.IsDevelopment()) { IdentityModelEventSource.ShowPII = true; app.UseSwaggerUI(); From afe5fb9bc940029c8fc4c27b7a9d37fd40597c92 Mon Sep 17 00:00:00 2001 From: Jicheng Lu <103353@smsassist.com> Date: Fri, 8 Nov 2024 13:36:43 -0600 Subject: [PATCH 3/6] refine prompt --- .../Evaluations/Models/EvaluationRequest.cs | 12 ++++++++++ .../Services/EvaluatingService.Evaluate.cs | 22 +++++++++++++++-- .../templates/instruction.simulator.liquid | 24 +++++++++++++++++++ 3 files changed, 56 insertions(+), 2 deletions(-) diff --git a/src/Infrastructure/BotSharp.Abstraction/Evaluations/Models/EvaluationRequest.cs b/src/Infrastructure/BotSharp.Abstraction/Evaluations/Models/EvaluationRequest.cs index 9dcad586..538b0609 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Evaluations/Models/EvaluationRequest.cs +++ b/src/Infrastructure/BotSharp.Abstraction/Evaluations/Models/EvaluationRequest.cs @@ -10,6 +10,18 @@ public class EvaluationRequest : LlmBaseRequest [JsonPropertyName("states")] public IEnumerable States { get; set; } = []; + [JsonPropertyName("duplicate_limit")] + public int DuplicateLimit { get; set; } = 2; + [JsonPropertyName("max_rounds")] public int MaxRounds { get; set; } = 20; + + + [JsonPropertyName("additional_instruction")] + [JsonIgnore(Condition = JsonIgnoreCondition.WhenWritingNull)] + public string? AdditionalInstruction { get; set; } + + [JsonPropertyName("stop_criteria")] + [JsonIgnore(Condition = JsonIgnoreCondition.WhenWritingNull)] + public string? StopCriteria { get; set; } } diff --git a/src/Infrastructure/BotSharp.Core/Evaluations/Services/EvaluatingService.Evaluate.cs b/src/Infrastructure/BotSharp.Core/Evaluations/Services/EvaluatingService.Evaluate.cs index a75450af..0442c857 100644 --- a/src/Infrastructure/BotSharp.Core/Evaluations/Services/EvaluatingService.Evaluate.cs +++ b/src/Infrastructure/BotSharp.Core/Evaluations/Services/EvaluatingService.Evaluate.cs @@ -42,6 +42,7 @@ public partial class EvaluatingService private async Task SimulateConversation(string initMessage, IEnumerable refDialogs, EvaluationRequest request) { var count = 0; + var duplicateCount = 0; var convId = Guid.NewGuid().ToString(); var curDialogs = new List(); var curUserMsg = initMessage; @@ -79,13 +80,30 @@ public partial class EvaluatingService { { "ref_conversation", refDialogs }, { "cur_conversation", curDialogs }, + { "additional_instruction", request.AdditionalInstruction }, + { "stop_criteria", request.StopCriteria } } }); _logger.LogInformation($"Generated message: {result?.GeneratedMessage}, stop: {result?.Stop}, reason: {result?.Reason}"); - if (curUserMsg.IsEqualTo(prevUserMsg) || curBotMsg.IsEqualTo(prevBotMsg) - || count > request.MaxRounds || (result != null && result.Stop)) + if (count > request.MaxRounds || (result != null && result.Stop)) + { + break; + } + + + if (curUserMsg.IsEqualTo(prevUserMsg) || curBotMsg.IsEqualTo(prevBotMsg)) + { + duplicateCount++; + } + else + { + duplicateCount = 0; + } + + + if (duplicateCount >= request.DuplicateLimit) { break; } diff --git a/src/Infrastructure/BotSharp.Core/data/agents/dfd9b46d-d00c-40af-8a75-3fbdc2b89869/templates/instruction.simulator.liquid b/src/Infrastructure/BotSharp.Core/data/agents/dfd9b46d-d00c-40af-8a75-3fbdc2b89869/templates/instruction.simulator.liquid index 7569f20c..a2af5fe2 100644 --- a/src/Infrastructure/BotSharp.Core/data/agents/dfd9b46d-d00c-40af-8a75-3fbdc2b89869/templates/instruction.simulator.liquid +++ b/src/Infrastructure/BotSharp.Core/data/agents/dfd9b46d-d00c-40af-8a75-3fbdc2b89869/templates/instruction.simulator.liquid @@ -4,7 +4,31 @@ Please take the content in the [REFERENCE CONVERSATION] section as a reference, ** You need to take a close look at the content in both [REFERENCE CONVERSATION] and [ONGOING CONVERSATION], and determine whether to generate a text message or stop the ongoing conversation. ** When you generate a message, please assume you are the user and reply in the user perceptive. ** Please do not generate or append a message with similar meaning that you have already mentioned in the [ONGOING CONVERSATION]. + + +================= +[ADDITIONAL INSTRUCTION] +{{ "\r\n" }} +{%- if additional_instruction != empty -%} +{{ additional_instruction }} +{%- endif -%} +{{ "\r\n" }} + + +================= +[CONVERSATION STOP CRITERIA] +{{ "\r\n" }} +{%- if stop_criteria != empty -%} +{{ stop_criteria }} +{%- else -%} ** If you see the assistant replies two or more than two similar messages in the [ONGOING CONVERSATION], please stop the conversation immediately. +{%- endif -%} +{{ "\r\n" }} + + +================= +[OUTPUT JSON FORMAT] + ** The output must be in JSON format: { "generated_message": the generated text message using the user tone, From 4f9cd09b404523641954feb75911a3b4c464b3df Mon Sep 17 00:00:00 2001 From: Haiping Chen Date: Fri, 8 Nov 2024 20:19:52 +0000 Subject: [PATCH 4/6] remove duplicate DistributedLocker --- src/Infrastructure/BotSharp.Core/BotSharpCoreExtensions.cs | 2 -- 1 file changed, 2 deletions(-) diff --git a/src/Infrastructure/BotSharp.Core/BotSharpCoreExtensions.cs b/src/Infrastructure/BotSharp.Core/BotSharpCoreExtensions.cs index 812d89a8..862ef899 100644 --- a/src/Infrastructure/BotSharp.Core/BotSharpCoreExtensions.cs +++ b/src/Infrastructure/BotSharp.Core/BotSharpCoreExtensions.cs @@ -28,8 +28,6 @@ public static class BotSharpCoreExtensions services.AddScoped(); services.AddScoped(); - services.AddSingleton(); - // Register cache service var cacheSettings = new SharpCacheSettings(); config.Bind("SharpCache", cacheSettings); From 0d89528562ca2b69ecd3649a5942700dea07a25c Mon Sep 17 00:00:00 2001 From: Haiping Chen Date: Fri, 8 Nov 2024 20:53:04 +0000 Subject: [PATCH 5/6] _logger in DistributedLocker --- .../Infrastructures/DistributedLocker.cs | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/src/Infrastructure/BotSharp.Core/Infrastructures/DistributedLocker.cs b/src/Infrastructure/BotSharp.Core/Infrastructures/DistributedLocker.cs index e37145ff..32c7af52 100644 --- a/src/Infrastructure/BotSharp.Core/Infrastructures/DistributedLocker.cs +++ b/src/Infrastructure/BotSharp.Core/Infrastructures/DistributedLocker.cs @@ -6,10 +6,12 @@ namespace BotSharp.Core.Infrastructures; public class DistributedLocker { private readonly IConnectionMultiplexer _redis; + private readonly ILogger _logger; - public DistributedLocker(IConnectionMultiplexer redis) + public DistributedLocker(IConnectionMultiplexer redis, ILogger logger) { _redis = redis; + _logger = logger; } public async Task Lock(string resource, Func> action, int timeoutInSeconds = 30) @@ -21,14 +23,14 @@ public class DistributedLocker { if (handle == null) { - Serilog.Log.Logger.Error($"Acquire lock for {resource} failed due to after {timeout}s timeout."); + _logger.LogWarning($"Acquire lock for {resource} failed due to after {timeout}s timeout."); } return await action(); } } - public void Lock(string resource, Action action, int timeoutInSeconds = 30) + public bool Lock(string resource, Action action, int timeoutInSeconds = 30) { var timeout = TimeSpan.FromSeconds(timeoutInSeconds); @@ -37,11 +39,13 @@ public class DistributedLocker { if (handle == null) { - Serilog.Log.Logger.Error($"Acquire lock for {resource} failed due to after {timeout}s timeout."); + _logger.LogWarning($"Acquire lock for {resource} failed due to after {timeout}s timeout."); + return false; } else { action(); + return true; } } } From b4e25f309670313b5de32fd6478bb1bb0926f6e9 Mon Sep 17 00:00:00 2001 From: Haiping Chen Date: Mon, 11 Nov 2024 04:13:50 +0000 Subject: [PATCH 6/6] ReDispatchAsync --- .../Infrastructures/Events/IEventPublisher.cs | 2 ++ .../Infrastructures/Events/RedisPublisher.cs | 25 +++++++++++++++++++ 2 files changed, 27 insertions(+) diff --git a/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventPublisher.cs b/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventPublisher.cs index 471d810e..e242937e 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventPublisher.cs +++ b/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventPublisher.cs @@ -11,4 +11,6 @@ public interface IEventPublisher Task BroadcastAsync(string channel, string message); Task PublishAsync(string channel, string message); + + Task ReDispatchAsync(string channel, int count = 10, string order = "asc"); } diff --git a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs index 4f3ad405..f6daa59d 100644 --- a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs +++ b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs @@ -29,4 +29,29 @@ public class RedisPublisher : IEventPublisher _logger.LogInformation($"Published message {channel} {message}"); } + + public async Task ReDispatchAsync(string channel, int count = 10, string order = "asc") + { + var db = _redis.GetDatabase(); + + var entries = await db.StreamRangeAsync(channel, "-", "+", count: count, messageOrder: order == "asc" ? Order.Ascending : Order.Descending); + foreach (var entry in entries) + { + _logger.LogInformation($"Fetched message: {channel} {entry.Values[0].Value} ({entry.Id})"); + + try + { + var messageId = await db.StreamAddAsync(channel, "message", entry.Values[0].Value); + + _logger.LogWarning($"ReDispatched message: {channel} {entry.Values[0].Value} ({messageId})"); + + // Optionally delete the message to save space + await db.StreamDeleteAsync(channel, [entry.Id]); + } + catch (Exception ex) + { + _logger.LogError($"Error processing message: {ex.Message}, event id: {channel} {entry.Id}"); + } + } + } }