From 1275053d660c61c27f8be54af724d6724da4e2b0 Mon Sep 17 00:00:00 2001 From: Jicheng Lu <103353@smsassist.com> Date: Thu, 6 Mar 2025 16:46:38 -0600 Subject: [PATCH] migrate conv latest states --- .../Conversations/IConversationService.cs | 2 + .../Repositories/IBotSharpRepository.cs | 4 + .../Services/ConversationService.Migration.cs | 98 +++++++++++++++++++ .../Services/LoggerService.Instruction.cs | 2 +- .../FileRepository.Conversation.cs | 55 +++++++++++ .../Controllers/ConversationController.cs | 10 ++ .../MigrateLatestStateRequest.cs | 7 ++ .../MongoRepository.Conversation.cs | 38 +++++++ 8 files changed, 215 insertions(+), 1 deletion(-) create mode 100644 src/Infrastructure/BotSharp.Core/Conversations/Services/ConversationService.Migration.cs create mode 100644 src/Infrastructure/BotSharp.OpenAPI/ViewModels/Conversations/MigrateLatestStateRequest.cs diff --git a/src/Infrastructure/BotSharp.Abstraction/Conversations/IConversationService.cs b/src/Infrastructure/BotSharp.Abstraction/Conversations/IConversationService.cs index 613940bf..6e558b1b 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Conversations/IConversationService.cs +++ b/src/Infrastructure/BotSharp.Abstraction/Conversations/IConversationService.cs @@ -70,4 +70,6 @@ public interface IConversationService /// if pre-loading, then keys are not filter by the search query /// Task> GetConversationStateSearhKeys(string query, int convLimit = 100, int keyLimit = 10, bool preload = false); + + Task MigrateLatestStates(int batchSize = 100, int errorLimit = 10); } diff --git a/src/Infrastructure/BotSharp.Abstraction/Repositories/IBotSharpRepository.cs b/src/Infrastructure/BotSharp.Abstraction/Repositories/IBotSharpRepository.cs index 6f8de07d..d523dd09 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Repositories/IBotSharpRepository.cs +++ b/src/Infrastructure/BotSharp.Abstraction/Repositories/IBotSharpRepository.cs @@ -151,6 +151,10 @@ public interface IBotSharpRepository : IHaveServiceProvider => throw new NotImplementedException(); List GetConversationStateSearchKeys(int messageLowerLimit = 2, int convUpperLimit = 100) => throw new NotImplementedException(); + List GetConversationsToMigrate(int batchSize = 100) + => throw new NotImplementedException(); + bool MigrateConvsersationLatestStates(string conversationId) + => throw new NotImplementedException(); #endregion #region LLM Completion Log diff --git a/src/Infrastructure/BotSharp.Core/Conversations/Services/ConversationService.Migration.cs b/src/Infrastructure/BotSharp.Core/Conversations/Services/ConversationService.Migration.cs new file mode 100644 index 00000000..23012706 --- /dev/null +++ b/src/Infrastructure/BotSharp.Core/Conversations/Services/ConversationService.Migration.cs @@ -0,0 +1,98 @@ +using NetTopologySuite.Algorithm; +using System.Diagnostics; + +namespace BotSharp.Core.Conversations.Services; + +public partial class ConversationService +{ + public async Task MigrateLatestStates(int batchSize = 100, int errorLimit = 10) + { + var db = _services.GetRequiredService(); + var isSuccess = true; + var errorCount = 0; + var batchNum = 0; + var info = string.Empty; + var error = string.Empty; + +#if DEBUG + Console.WriteLine($"\r\n#Start migrating Conversation Latest States...\r\n"); +#else + _logger.LogInformation($"#Start migrating Conversation Latest States..."); +#endif + var sw = Stopwatch.StartNew(); + + var convIds = db.GetConversationsToMigrate(batchSize); + + while (!convIds.IsNullOrEmpty()) + { + batchNum++; + var innerSw = Stopwatch.StartNew(); +#if DEBUG + Console.WriteLine($"\r\n#Start migrating Conversation Latest States (batch number: {batchNum})\r\n"); +#else + _logger.LogInformation($"#Start migrating Conversation Latest States (batch number: {batchNum})"); +#endif + + for (int i = 0; i < convIds.Count; i++) + { + var convId = convIds.ElementAt(i); + try + { + var done = db.MigrateConvsersationLatestStates(convId); + info = $"Conversation {convId} latest states have been migrated ({i + 1}/{convIds.Count})!"; +#if DEBUG + Console.WriteLine($"\r\n{info}\r\n"); +#else + _logger.LogInformation($"{info}"); +#endif + } + catch (Exception ex) + { + errorCount++; + error = $"Conversation {convId} latest states fail to be migrated! ({i + 1}/{convIds.Count})\r\n{ex.Message}\r\n{ex.InnerException}"; +#if DEBUG + Console.WriteLine($"\r\n{error}\r\n"); +#else + _logger.LogError($"{error}"); +#endif + } + } + + if (errorCount >= errorLimit) + { + error = $"\r\nErrors exceed limit => stop the migration!\r\n"; +#if DEBUG + Console.WriteLine($"{error}"); +#else + _logger.LogError($"{error}"); +#endif + innerSw.Stop(); + isSuccess = false; + break; + } + + innerSw.Stop(); + info = $"#Done migrating Conversation Latest States (batch number: {batchNum}) " + + $"(Total time: {innerSw.Elapsed.Hours} hrs, {innerSw.Elapsed.Minutes} mins, {innerSw.Elapsed.Seconds} seconds)"; +#if DEBUG + Console.WriteLine($"\r\n{info}\r\n"); +#else + _logger.LogInformation($"{info}"); +#endif + + await Task.Delay(100); + convIds = db.GetConversationsToMigrate(batchSize); + } + + sw.Stop(); + info = $"#Done with migrating Conversation Latest States! " + + $"(Total time: {sw.Elapsed.Days} days, {sw.Elapsed.Hours} hrs, {sw.Elapsed.Minutes} mins, {sw.Elapsed.Seconds} seconds)"; +#if DEBUG + Console.WriteLine($"\r\n{info}\r\n"); +#else + _logger.LogInformation($"{info}"); +#endif + + return isSuccess; + } +} diff --git a/src/Infrastructure/BotSharp.Core/Loggers/Services/LoggerService.Instruction.cs b/src/Infrastructure/BotSharp.Core/Loggers/Services/LoggerService.Instruction.cs index 18a0564b..84efae99 100644 --- a/src/Infrastructure/BotSharp.Core/Loggers/Services/LoggerService.Instruction.cs +++ b/src/Infrastructure/BotSharp.Core/Loggers/Services/LoggerService.Instruction.cs @@ -41,7 +41,7 @@ public partial class LoggerService var items = logs.Items.Select(x => { - x.AgentId = !string.IsNullOrEmpty(x.AgentId) ? agents.FirstOrDefault(a => a.Id == x.AgentId)?.Name : null; + x.AgentName = !string.IsNullOrEmpty(x.AgentId) ? agents.FirstOrDefault(a => a.Id == x.AgentId)?.Name : null; if (!isAdmin) { diff --git a/src/Infrastructure/BotSharp.Core/Repository/FileRepository/FileRepository.Conversation.cs b/src/Infrastructure/BotSharp.Core/Repository/FileRepository/FileRepository.Conversation.cs index 08a534d4..8880a4e0 100644 --- a/src/Infrastructure/BotSharp.Core/Repository/FileRepository/FileRepository.Conversation.cs +++ b/src/Infrastructure/BotSharp.Core/Repository/FileRepository/FileRepository.Conversation.cs @@ -687,6 +687,56 @@ public partial class FileRepository } + + public List GetConversationsToMigrate(int batchSize = 100) + { + var baseDir = Path.Combine(_dbSettings.FileRepository, _conversationSettings.DataDir); + if (!Directory.Exists(baseDir)) return []; + + var convIds = new List(); + var dirs = Directory.GetDirectories(baseDir); + + foreach (var dir in dirs) + { + var latestStateFile = Path.Combine(dir, CONV_LATEST_STATE_FILE); + if (File.Exists(latestStateFile)) continue; + + var convId = dir.Split(Path.DirectorySeparatorChar).Last(); + if (string.IsNullOrEmpty(convId)) continue; + + convIds.Add(convId); + if (convIds.Count >= batchSize) + { + break; + } + } + + return convIds; + } + + + public bool MigrateConvsersationLatestStates(string conversationId) + { + if (string.IsNullOrEmpty(conversationId)) return false; + + var convDir = FindConversationDirectory(conversationId); + if (string.IsNullOrEmpty(convDir)) + { + return false; + } + + var stateFile = Path.Combine(convDir, STATE_FILE); + var states = CollectConversationStates(stateFile); + var latestStates = BuildLatestStates(states); + + var latestStateFile = Path.Combine(convDir, CONV_LATEST_STATE_FILE); + var stateStr = JsonSerializer.Serialize(latestStates, _options); + File.WriteAllText(latestStateFile, stateStr); + + return true; + } + + #region Private methods private string? FindConversationDirectory(string conversationId) { @@ -883,6 +933,11 @@ public partial class FileRepository private Dictionary BuildLatestStates(List states) { var endNodes = new Dictionary(); + if (states.IsNullOrEmpty()) + { + return endNodes; + } + foreach (var pair in states) { var value = pair.Values?.LastOrDefault(); diff --git a/src/Infrastructure/BotSharp.OpenAPI/Controllers/ConversationController.cs b/src/Infrastructure/BotSharp.OpenAPI/Controllers/ConversationController.cs index 75dc1113..10317815 100644 --- a/src/Infrastructure/BotSharp.OpenAPI/Controllers/ConversationController.cs +++ b/src/Infrastructure/BotSharp.OpenAPI/Controllers/ConversationController.cs @@ -563,6 +563,16 @@ public class ConversationController : ControllerBase } #endregion + #region Migrate Latest States + [HttpPost("/conversation/latest-state/migrate")] + public async Task MigrateConversationLatestStates([FromBody] MigrateLatestStateRequest request) + { + var convService = _services.GetRequiredService(); + var res = await convService.MigrateLatestStates(request.BatchSize, request.ErrorLimit); + return res; + } + #endregion + #region Private methods private void SetStates(IConversationService conv, NewMessageModel input) { diff --git a/src/Infrastructure/BotSharp.OpenAPI/ViewModels/Conversations/MigrateLatestStateRequest.cs b/src/Infrastructure/BotSharp.OpenAPI/ViewModels/Conversations/MigrateLatestStateRequest.cs new file mode 100644 index 00000000..99085b0a --- /dev/null +++ b/src/Infrastructure/BotSharp.OpenAPI/ViewModels/Conversations/MigrateLatestStateRequest.cs @@ -0,0 +1,7 @@ +namespace BotSharp.OpenAPI.ViewModels.Conversations; + +public class MigrateLatestStateRequest +{ + public int BatchSize { get; set; } = 1000; + public int ErrorLimit { get; set; } = 10; +} diff --git a/src/Plugins/BotSharp.Plugin.MongoStorage/Repository/MongoRepository.Conversation.cs b/src/Plugins/BotSharp.Plugin.MongoStorage/Repository/MongoRepository.Conversation.cs index 1c0eeb2f..b9931bd4 100644 --- a/src/Plugins/BotSharp.Plugin.MongoStorage/Repository/MongoRepository.Conversation.cs +++ b/src/Plugins/BotSharp.Plugin.MongoStorage/Repository/MongoRepository.Conversation.cs @@ -1,5 +1,6 @@ using BotSharp.Abstraction.Conversations.Models; using BotSharp.Abstraction.Repositories.Filters; +using MongoDB.Driver; using System.Text.Json; namespace BotSharp.Plugin.MongoStorage.Repository; @@ -661,6 +662,37 @@ public partial class MongoRepository return keys; } + + + public List GetConversationsToMigrate(int batchSize = 100) + { + var convFilter = Builders.Filter.Exists(x => x.LatestStates, false); + var sortDef = Builders.Sort.Ascending(x => x.CreatedTime); + var convIds = _dc.Conversations.Find(convFilter).Sort(sortDef) + .Limit(batchSize).ToEnumerable() + .Select(x => x.Id).ToList(); + return convIds ?? []; + } + + public bool MigrateConvsersationLatestStates(string conversationId) + { + if (string.IsNullOrEmpty(conversationId)) return false; + + var stateFilter = Builders.Filter.Eq(x => x.ConversationId, conversationId); + var foundStates = _dc.ConversationStates.Find(stateFilter).FirstOrDefault(); + if (foundStates?.States == null) return false; + + var states = foundStates.States.ToList(); + var latestStates = BuildLatestStates(states); + + var convFilter = Builders.Filter.Eq(x => x.Id, conversationId); + var convUpdate = Builders.Update.Set(x => x.LatestStates, latestStates); + _dc.Conversations.UpdateOne(convFilter, convUpdate); + + return true; + } + + #region Private methods private string ConvertSnakeCaseToPascalCase(string snakeCase) { string[] words = snakeCase.Split('_'); @@ -682,6 +714,11 @@ public partial class MongoRepository private Dictionary BuildLatestStates(List states) { var endNodes = new Dictionary(); + if (states.IsNullOrEmpty()) + { + return endNodes; + } + foreach (var pair in states) { var value = pair.Values?.LastOrDefault(); @@ -703,4 +740,5 @@ public partial class MongoRepository return endNodes; } + #endregion }