Merge branch 'master' into lida_Dev

src/Infrastructure/BotSharp.Abstraction/Evaluations/Models/EvaluationRequest.cs

@@ -10,6 +10,18 @@ public class EvaluationRequest : LlmBaseRequest
     [JsonPropertyName("states")]
     public IEnumerable<MessageState> 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; }
 }

src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventPublisher.cs

@@ -0,0 +1,16 @@
+namespace BotSharp.Abstraction.Infrastructures.Events;
+
+public interface IEventPublisher
+{
+    /// <summary>
+    /// Boardcast message to all subscribers
+    /// </summary>
+    /// <param name="channel"></param>
+    /// <param name="message"></param>
+    /// <returns></returns>
+    Task BroadcastAsync(string channel, string message);
+
+    Task PublishAsync(string channel, string message);
+
+    Task ReDispatchAsync(string channel, int count = 10, string order = "asc");
+}

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<string, string, Task> received);
+
+    Task SubscribeAsync(string channel, string group, Func<string, string, Task> received);
+}

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;
 
@@ -26,18 +28,18 @@ public static class BotSharpCoreExtensions
         services.AddScoped<IUserService, UserService>();
         services.AddScoped<ProcessorFactory>();
 
-        services.AddSingleton<DistributedLocker>();
-
         // Register cache service
         var cacheSettings = new SharpCacheSettings();
         config.Bind("SharpCache", cacheSettings);
         services.AddSingleton(x => cacheSettings);
         services.AddSingleton<ICacheService, RedisCacheService>();
 
+        AddRedisEvents(services, config);
+
         services.AddMemoryCache();
 
         RegisterPlugins(services, config);
-        ConfigureBotSharpOptions(services, configOptions);
+        AddBotSharpOptions(services, configOptions);
 
         return services;
     }
@@ -79,7 +81,7 @@ public static class BotSharpCoreExtensions
         return app;
     }
 
-    private static void ConfigureBotSharpOptions(IServiceCollection services, Action<BotSharpOptions>? configure)
+    private static void AddBotSharpOptions(IServiceCollection services, Action<BotSharpOptions>? configure)
     {
         var options = new BotSharpOptions();
         if (configure != null)
@@ -91,6 +93,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<IConnectionMultiplexer>(ConnectionMultiplexer.Connect(dbSettings.Redis));
+        services.AddSingleton<IEventPublisher, RedisPublisher>();
+        services.AddSingleton<IEventSubscriber, RedisSubscriber>();
+    }
+
     private static void AddDefaultJsonConverters(BotSharpOptions options)
     {
         options.JsonSerializerOptions.Converters.Add(new RichContentJsonConverter());

src/Infrastructure/BotSharp.Core/Evaluations/Services/EvaluatingService.Evaluate.cs

@@ -42,6 +42,7 @@ public partial class EvaluatingService
     private async Task<string> SimulateConversation(string initMessage, IEnumerable<string> refDialogs, EvaluationRequest request)
     {
         var count = 0;
+        var duplicateCount = 0;
         var convId = Guid.NewGuid().ToString();
         var curDialogs = new List<string>();
         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;
             }

src/Infrastructure/BotSharp.Core/Infrastructures/DistributedLocker.cs

@@ -5,65 +5,48 @@ namespace BotSharp.Core.Infrastructures;
 
 public class DistributedLocker
 {
-    private readonly BotSharpDatabaseSettings _settings;
-    private static ConnectionMultiplexer connection;
+    private readonly IConnectionMultiplexer _redis;
+    private readonly ILogger _logger;
 
-    public DistributedLocker(BotSharpDatabaseSettings settings)
+    public DistributedLocker(IConnectionMultiplexer redis, ILogger<DistributedLocker> logger)
     {
-        _settings = settings;
+        _redis = redis;
+        _logger = logger;
     }
 
     public async Task<T> Lock<T>(string resource, Func<Task<T>> 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) 
             {
-                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)
     {
-        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)
             {
-                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;
             }
         }
     }
-
-    private void ConnectToRedis()
-    {
-        if (connection == null)
-        {
-            connection = ConnectionMultiplexer.Connect(_settings.Redis);
-        }
-    }
-
-    private async Task ConnectToRedisAsync()
-    {
-        if (connection == null)
-        {
-            connection = await ConnectionMultiplexer.ConnectAsync(_settings.Redis);
-        }
-    }
 }

src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs

@@ -0,0 +1,57 @@
+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<RedisPublisher> 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}");
+    }
+
+    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}");
+            }
+        }
+    }
+}

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<RedisSubscriber> logger)
+    {
+        _redis = redis;
+        _logger = logger;
+        _subscriber = _redis.GetSubscriber();
+    }
+
+    public async Task SubscribeAsync(string channel, Func<string, string, Task> 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<string, string, Task> 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);
+        }
+
+    }
+}

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")
         };
 

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;
+global using Aspects.Cache;
+global using BotSharp.Abstraction.Infrastructures.Events;

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,

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();