From fa93d48002001cdd8e51e2aa249ec17e7f9f39cf Mon Sep 17 00:00:00 2001 From: jie65535 Date: Mon, 3 Aug 2026 11:55:21 +0800 Subject: [PATCH] profile: parallelize and control analysis jobs --- README.md | 11 +- src/main/kotlin/command/PluginCommands.kt | 162 +++++++---- .../kotlin/command/ProfileCommandArguments.kt | 15 ++ .../kotlin/profile/ProfileAnalysisRunGate.kt | 28 ++ .../kotlin/profile/ProfileCompactionModels.kt | 2 + .../kotlin/profile/ProfileOperationLogger.kt | 31 +++ .../kotlin/profile/ProfileUserLockManager.kt | 47 ++++ .../profile/UserProfileAnalysisService.kt | 251 +++++++++++------- src/main/kotlin/profile/UserProfileModels.kt | 2 + src/main/kotlin/profile/UserProfileStore.kt | 13 + .../command/ProfileCommandArgumentsTest.kt | 30 +++ .../profile/ProfileAnalysisRunGateTest.kt | 18 ++ .../profile/ProfileOperationLoggerTest.kt | 59 ++++ .../profile/ProfileUserLockManagerTest.kt | 43 +++ .../profile/UserProfileAnalysisServiceTest.kt | 99 ++++++- .../kotlin/profile/UserProfileStoreTest.kt | 1 + 16 files changed, 650 insertions(+), 162 deletions(-) create mode 100644 src/main/kotlin/command/ProfileCommandArguments.kt create mode 100644 src/main/kotlin/profile/ProfileAnalysisRunGate.kt create mode 100644 src/main/kotlin/profile/ProfileOperationLogger.kt create mode 100644 src/main/kotlin/profile/ProfileUserLockManager.kt create mode 100644 src/test/kotlin/command/ProfileCommandArgumentsTest.kt create mode 100644 src/test/kotlin/profile/ProfileAnalysisRunGateTest.kt create mode 100644 src/test/kotlin/profile/ProfileOperationLoggerTest.kt create mode 100644 src/test/kotlin/profile/ProfileUserLockManagerTest.kt diff --git a/README.md b/README.md index 456263d..a7f278d 100644 --- a/README.md +++ b/README.md @@ -57,10 +57,11 @@ AI 可以自动调用多种工具来完成复杂任务： ### 渐进式历史画像（实验） - 日常使用无需画像命令：群聊缓存会话闭合后，一次模型调用会静默归纳其中所有有实质发言的参与者 -- `/jgpt profileAnalyze [batches]` - 手动推进指定用户画像，默认1批 -- `/jgpt profileAnalyzeGroup [batches]` - 按群历史批量推进参与者画像，默认1批 -- `/jgpt profileShow ` - 诊断或验收时查看已经提交的完整画像和覆盖时间 -- `/jgpt profileCompact ` - 独立反思并压缩指定用户的重复、过细或低价值画像条目 +- `/jgpt profileAnalyze [batches]` - 分析用户画像，多个 ID 用逗号分隔 +- `/jgpt profileAnalyzeGroup [batches]` - 分析群画像，多个 ID 用逗号分隔 +- `/jgpt profileShow ` - 查看用户画像 +- `/jgpt profileCompact [userIds]` - 压缩画像，不传 ID 时处理全部用户 +- `/jgpt profileStop` - 当前批次完成后停止画像任务 ## 配置文件 @@ -239,6 +240,8 @@ searchHistoryMaxRecords: 5000 好友昵称、群名片、群角色、签名、年龄等联系人公开资料只作为人物识别提示，不直接视为画像证据。画像模型 仍必须依据对应用户本人在聊天历史中的发言，才能新增或确认长期画像条目。 +画像发生变更时，日志会输出 `PROFILE_OPERATIONS` 供抽查。 + 画像结果保存在插件数据目录的 `user-profile.sqlite`，不会覆盖现有好感度或旧印象数据。实验时可把历史 备份配置为只读来源，例如： diff --git a/src/main/kotlin/command/PluginCommands.kt b/src/main/kotlin/command/PluginCommands.kt index 6667c0b..0b396de 100644 --- a/src/main/kotlin/command/PluginCommands.kt +++ b/src/main/kotlin/command/PluginCommands.kt @@ -1,7 +1,9 @@ package top.jie65535.mirai.command import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.launch +import kotlinx.coroutines.withContext import net.mamoe.mirai.console.command.CommandSender import net.mamoe.mirai.console.command.CompositeCommand import net.mamoe.mirai.console.permission.PermissionService.Companion.cancel @@ -49,61 +51,69 @@ object PluginCommands : CompositeCommand( } @SubCommand - suspend fun CommandSender.profileAnalyze(userId: Long, batches: Int = 1) { + suspend fun CommandSender.profileAnalyze(userIds: String, batches: Int = 1) { require(batches > 0) { "batches 必须是正数" } - sendMessage("已启动用户 $userId 的画像分析，本次最多推进 $batches 个批次。") - JChatGPT.launch { - try { - val report = UserProfileAnalysisService.analyze(userId, batches) { progress -> - JChatGPT.logger.info( - "PROFILE_BATCH user=$userId batch=${progress.batchIndex}/$batches " + - "range=${progress.startTime}-${progress.endTime} " + - "messages=${progress.messageCount} operations=${progress.operationCount} " + - "skipped=${progress.skippedOperationCount} " + - "tokens=${progress.usage.promptTokens}/${progress.usage.completionTokens} " + - "cached=${progress.usage.cachedTokens}" - ) + val parsedUserIds = parseProfileUserIds(userIds) + val runToken = UserProfileAnalysisService.newRunToken() + sendMessage("已启动 ${parsedUserIds.size} 个用户的画像分析，每人最多推进 $batches 个批次。") + parsedUserIds.forEach { userId -> + JChatGPT.launch { + try { + val report = UserProfileAnalysisService.analyze(userId, batches, runToken) { progress -> + JChatGPT.logger.info( + "PROFILE_BATCH user=$userId batch=${progress.batchIndex}/$batches " + + "range=${progress.startTime}-${progress.endTime} " + + "messages=${progress.messageCount} operations=${progress.operationCount} " + + "skipped=${progress.skippedOperationCount} " + + "tokens=${progress.usage.promptTokens}/${progress.usage.completionTokens} " + + "cached=${progress.usage.cachedTokens}" + ) + } + when { + report.alreadyRunning -> sendMessage("用户 $userId 已有画像分析任务在运行。") + report.profile == null -> sendMessage("聊天记录中没有找到用户 $userId 的群聊发言。") + else -> sendMessage(formatProfileReport(report)) + } + } catch (cause: CancellationException) { + throw cause + } catch (cause: Exception) { + JChatGPT.logger.error("用户 $userId 画像分析失败", cause) + sendMessage("用户 $userId 画像分析失败：${cause.message ?: cause::class.simpleName}") } - when { - report.alreadyRunning -> sendMessage("用户 $userId 已有画像分析任务在运行。") - report.profile == null -> sendMessage("聊天记录中没有找到用户 $userId 的群聊发言。") - else -> sendMessage(formatProfileReport(report)) - } - } catch (cause: CancellationException) { - throw cause - } catch (cause: Exception) { - JChatGPT.logger.error("用户 $userId 画像分析失败", cause) - sendMessage("用户 $userId 画像分析失败：${cause.message ?: cause::class.simpleName}") } } } @SubCommand - suspend fun CommandSender.profileAnalyzeGroup(groupId: Long, batches: Int = 1) { + suspend fun CommandSender.profileAnalyzeGroup(groupIds: String, batches: Int = 1) { require(batches > 0) { "batches 必须是正数" } - sendMessage("已启动群 $groupId 的批量画像分析，本次最多推进 $batches 个批次。") - JChatGPT.launch { - try { - val report = UserProfileAnalysisService.analyzeGroup(groupId, batches) { progress -> - JChatGPT.logger.info( - "PROFILE_GROUP_BATCH group=$groupId batch=${progress.batchIndex}/$batches " + - "range=${progress.startTime}-${progress.endTime} " + - "messages=${progress.messageCount} users=${progress.analyzedUsers} " + - "operations=${progress.appliedOperations} skipped=${progress.skippedOperations} " + - "tokens=${progress.usage.promptTokens}/${progress.usage.completionTokens} " + - "cached=${progress.usage.cachedTokens}" - ) + val parsedGroupIds = parseProfileGroupIds(groupIds) + val runToken = UserProfileAnalysisService.newRunToken() + sendMessage("已启动 ${parsedGroupIds.size} 个群的画像分析，每群最多推进 $batches 个批次。") + parsedGroupIds.forEach { groupId -> + JChatGPT.launch { + try { + val report = UserProfileAnalysisService.analyzeGroup(groupId, batches, runToken) { progress -> + JChatGPT.logger.info( + "PROFILE_GROUP_BATCH group=$groupId batch=${progress.batchIndex}/$batches " + + "range=${progress.startTime}-${progress.endTime} " + + "messages=${progress.messageCount} users=${progress.analyzedUsers} " + + "operations=${progress.appliedOperations} skipped=${progress.skippedOperations} " + + "tokens=${progress.usage.promptTokens}/${progress.usage.completionTokens} " + + "cached=${progress.usage.cachedTokens}" + ) + } + when { + report.alreadyRunning -> sendMessage("群 $groupId 已有画像分析任务在运行。") + report.botId == null -> sendMessage("聊天记录中没有找到群 $groupId 的消息。") + else -> sendMessage(formatGroupProfileReport(report)) + } + } catch (cause: CancellationException) { + throw cause + } catch (cause: Exception) { + JChatGPT.logger.error("群 $groupId 批量画像分析失败", cause) + sendMessage("群 $groupId 批量画像分析失败：${cause.message ?: cause::class.simpleName}") } - when { - report.alreadyRunning -> sendMessage("群 $groupId 已有画像分析任务在运行。") - report.botId == null -> sendMessage("聊天记录中没有找到群 $groupId 的消息。") - else -> sendMessage(formatGroupProfileReport(report)) - } - } catch (cause: CancellationException) { - throw cause - } catch (cause: Exception) { - JChatGPT.logger.error("群 $groupId 批量画像分析失败", cause) - sendMessage("群 $groupId 批量画像分析失败：${cause.message ?: cause::class.simpleName}") } } } @@ -119,21 +129,49 @@ object PluginCommands : CompositeCommand( } @SubCommand - suspend fun CommandSender.profileCompact(userId: Long) { - sendMessage("已启动用户 $userId 的画像压缩反思。") - JChatGPT.launch { - try { - val report = UserProfileAnalysisService.compact(userId) - sendMessage(formatProfileCompactionReport(report)) - } catch (cause: CancellationException) { - throw cause - } catch (cause: Exception) { - JChatGPT.logger.error("用户 $userId 画像压缩失败", cause) - sendMessage("用户 $userId 画像压缩失败：${cause.message ?: cause::class.simpleName}") + suspend fun CommandSender.profileCompact(userIds: String = "") { + check(PluginConfig.profileEnabled) { "历史用户画像分析未启用" } + check(UserProfileStore.isAvailable) { "用户画像数据库不可用" } + val parsedUserIds = if (userIds.isBlank()) { + withContext(Dispatchers.IO) { UserProfileStore.listUserIds() } + } else { + parseProfileUserIds(userIds) + } + if (parsedUserIds.isEmpty()) { + sendMessage("当前没有可压缩的用户画像。") + return + } + val runToken = UserProfileAnalysisService.newRunToken() + sendMessage("已启动 ${parsedUserIds.size} 个用户的画像压缩反思。") + parsedUserIds.forEach { userId -> + JChatGPT.launch { + try { + val report = UserProfileAnalysisService.compact(userId, runToken) + when { + report.alreadyRunning -> sendMessage("用户 $userId 已有画像压缩任务在运行。") + report.stopped -> sendMessage("用户 $userId 的画像压缩已按请求停止，未开始新的压缩轮次。") + else -> sendMessage(formatProfileCompactionReport(report)) + } + } catch (cause: CancellationException) { + throw cause + } catch (cause: Exception) { + JChatGPT.logger.error("用户 $userId 画像压缩失败", cause) + sendMessage("用户 $userId 画像压缩失败：${cause.message ?: cause::class.simpleName}") + } } } } + @SubCommand + suspend fun CommandSender.profileStop() { + val report = UserProfileAnalysisService.stopAll() + sendMessage( + "已发出画像任务停止请求；当前检测到 ${report.totalTasks} 个任务" + + "（用户分析 ${report.userTasks}，群分析 ${report.groupTasks}，压缩 ${report.compactionTasks}）。" + + "当前轮次会正常完成，之后不再开始新一轮。" + ) + } + @SubCommand suspend fun CommandSender.skills() { val all = SkillStore.all @@ -325,7 +363,7 @@ object PluginCommands : CompositeCommand( "${formatNumber(report.usage.completionTokens)}，缓存命中 " + formatNumber(report.usage.cachedTokens) ) - appendLine("状态：${if (report.caughtUp) "已追平当前快照" else "可继续推进"}") + appendLine("状态：${profileAnalysisStatus(report.caughtUp, report.stopped)}") append(formatProfile(checkNotNull(report.profile))) }.trim() @@ -341,9 +379,15 @@ object PluginCommands : CompositeCommand( formatNumber(report.usage.cachedTokens) ) appendLine("群历史覆盖至 ${formatProfileTime(report.cursorTime)}") - append("状态：${if (report.caughtUp) "已追平当前快照" else "可继续推进"}") + append("状态：${profileAnalysisStatus(report.caughtUp, report.stopped)}") }.trim() + private fun profileAnalysisStatus(caughtUp: Boolean, stopped: Boolean): String = when { + caughtUp -> "已追平当前快照" + stopped -> "已按请求停止，可继续推进" + else -> "可继续推进" + } + private fun formatProfileCompactionReport(report: ProfileCompactionReport): String = buildString { appendLine( "画像压缩完成：${report.beforeItems} -> ${report.afterItems} 条，" + diff --git a/src/main/kotlin/command/ProfileCommandArguments.kt b/src/main/kotlin/command/ProfileCommandArguments.kt new file mode 100644 index 0000000..5ea899a --- /dev/null +++ b/src/main/kotlin/command/ProfileCommandArguments.kt @@ -0,0 +1,15 @@ +package top.jie65535.mirai.command + +internal fun parseProfileGroupIds(raw: String): List = parseProfileIds(raw, "群号") + +internal fun parseProfileUserIds(raw: String): List = parseProfileIds(raw, "用户号") + +private fun parseProfileIds(raw: String, label: String): List { + val ids = raw.split(',', '，', ';', '；') + .map(String::trim) + .filter(String::isNotEmpty) + .map { value -> value.toLongOrNull()?.takeIf { it > 0 } ?: error("无效$label: $value") } + .distinct() + require(ids.isNotEmpty()) { "至少需要一个$label" } + return ids +} diff --git a/src/main/kotlin/profile/ProfileAnalysisRunGate.kt b/src/main/kotlin/profile/ProfileAnalysisRunGate.kt new file mode 100644 index 0000000..efcf8a7 --- /dev/null +++ b/src/main/kotlin/profile/ProfileAnalysisRunGate.kt @@ -0,0 +1,28 @@ +package top.jie65535.mirai.profile + +import java.util.concurrent.atomic.AtomicLong + +class ProfileAnalysisRunToken internal constructor( + internal val generation: Long, +) + +data class ProfileAnalysisStopReport( + val userTasks: Int, + val groupTasks: Int, + val compactionTasks: Int, +) { + val totalTasks: Int + get() = userTasks + groupTasks + compactionTasks +} + +internal class ProfileAnalysisRunGate { + private val generation = AtomicLong() + + fun newToken(): ProfileAnalysisRunToken = ProfileAnalysisRunToken(generation.get()) + + fun stopCurrentRuns() { + generation.incrementAndGet() + } + + fun canContinue(token: ProfileAnalysisRunToken): Boolean = token.generation == generation.get() +} diff --git a/src/main/kotlin/profile/ProfileCompactionModels.kt b/src/main/kotlin/profile/ProfileCompactionModels.kt index 5058817..9eb5724 100644 --- a/src/main/kotlin/profile/ProfileCompactionModels.kt +++ b/src/main/kotlin/profile/ProfileCompactionModels.kt @@ -79,4 +79,6 @@ data class ProfileCompactionReport( val skippedOperations: Int, val usage: ProfileTokenUsage, val profile: UserProfileSnapshot, + val alreadyRunning: Boolean = false, + val stopped: Boolean = false, ) diff --git a/src/main/kotlin/profile/ProfileOperationLogger.kt b/src/main/kotlin/profile/ProfileOperationLogger.kt new file mode 100644 index 0000000..13fe4b3 --- /dev/null +++ b/src/main/kotlin/profile/ProfileOperationLogger.kt @@ -0,0 +1,31 @@ +package top.jie65535.mirai.profile + +import top.jie65535.mirai.JChatGPT + +internal object ProfileOperationLogger { + fun log(context: String, reductions: Collection) { + format(context, reductions)?.let(JChatGPT.logger::info) + } + + internal fun format(context: String, reductions: Collection): String? { + val operations = reductions.flatMap { reduction -> + reduction.operations.map { operation -> reduction.profile.userId to operation } + } + if (operations.isEmpty()) return null + + return buildString { + append("PROFILE_OPERATIONS ").append(context) + .append(" operations=").appendLine(operations.size) + operations.forEach { (userId, operation) -> + append("- user=").append(userId) + .append(" action=").append(operation.action) + .append(" category=").append(operation.category.name.lowercase()) + .append(" confidence=").append(operation.confidence.name.lowercase()) + operation.relatedUserId?.let { append(" related=").append(it) } + append(" content=").appendLine(operation.content.normalized()) + } + }.trimEnd() + } + + private fun String.normalized(): String = trim().replace(Regex("\\s+"), " ") +} diff --git a/src/main/kotlin/profile/ProfileUserLockManager.kt b/src/main/kotlin/profile/ProfileUserLockManager.kt new file mode 100644 index 0000000..9daafbe --- /dev/null +++ b/src/main/kotlin/profile/ProfileUserLockManager.kt @@ -0,0 +1,47 @@ +package top.jie65535.mirai.profile + +import kotlinx.coroutines.sync.Mutex +import java.util.concurrent.ConcurrentHashMap + +internal class ProfileUserLockManager { + private val entries = ConcurrentHashMap() + + suspend fun withUserLocks(userIds: Collection, block: suspend () -> T): T { + val reserved = userIds.asSequence() + .distinct() + .sorted() + .map { userId -> userId to reserve(userId) } + .toList() + val acquired = mutableListOf() + return try { + reserved.forEach { (_, entry) -> + entry.mutex.lock() + acquired += entry + } + block() + } finally { + acquired.asReversed().forEach { entry -> entry.mutex.unlock() } + reserved.forEach { (userId, entry) -> release(userId, entry) } + } + } + + internal val activeLockCount: Int + get() = entries.size + + private fun reserve(userId: Long): LockEntry = entries.compute(userId) { _, current -> + (current ?: LockEntry()).also { it.references++ } + } ?: error("无法创建用户画像锁: $userId") + + private fun release(userId: Long, expected: LockEntry) { + entries.computeIfPresent(userId) { _, current -> + check(current === expected) { "用户画像锁状态不一致: $userId" } + current.references-- + current.takeIf { it.references > 0 } + } + } + + private class LockEntry( + val mutex: Mutex = Mutex(), + var references: Int = 0, + ) +} diff --git a/src/main/kotlin/profile/UserProfileAnalysisService.kt b/src/main/kotlin/profile/UserProfileAnalysisService.kt index 051f364..5342b0c 100644 --- a/src/main/kotlin/profile/UserProfileAnalysisService.kt +++ b/src/main/kotlin/profile/UserProfileAnalysisService.kt @@ -2,8 +2,6 @@ package top.jie65535.mirai.profile import kotlinx.coroutines.CancellationException import kotlinx.coroutines.Dispatchers -import kotlinx.coroutines.sync.Semaphore -import kotlinx.coroutines.sync.withPermit import kotlinx.coroutines.withContext import top.jie65535.mirai.JChatGPT import top.jie65535.mirai.config.PluginConfig @@ -16,11 +14,26 @@ import java.util.concurrent.ConcurrentHashMap object UserProfileAnalysisService { private val runningUsers = ConcurrentHashMap.newKeySet() private val runningGroups = ConcurrentHashMap.newKeySet() - private val concurrencyLimiter = Semaphore(1) + private val runningCompactions = ConcurrentHashMap.newKeySet() + private val userLocks = ProfileUserLockManager() + private val runGate = ProfileAnalysisRunGate() + + fun newRunToken(): ProfileAnalysisRunToken = runGate.newToken() + + fun stopAll(): ProfileAnalysisStopReport { + val report = ProfileAnalysisStopReport( + userTasks = runningUsers.size, + groupTasks = runningGroups.size, + compactionTasks = runningCompactions.size, + ) + runGate.stopCurrentRuns() + return report + } suspend fun analyze( userId: Long, maxBatches: Int, + runToken: ProfileAnalysisRunToken = newRunToken(), onProgress: suspend (ProfileAnalysisProgress) -> Unit = {}, ): ProfileAnalysisReport { require(userId > 0) { "userId 必须是正数" } @@ -43,65 +56,75 @@ object UserProfileAnalysisService { } try { - return concurrencyLimiter.withPermit { - analyzeExclusive(userId, maxBatches, onProgress) + return userLocks.withUserLocks(listOf(userId)) { + analyzeExclusive(userId, maxBatches, runToken, onProgress) } } finally { runningUsers.remove(userId) } } - suspend fun compact(userId: Long): ProfileCompactionReport { + suspend fun compact( + userId: Long, + runToken: ProfileAnalysisRunToken = newRunToken(), + ): ProfileCompactionReport { require(userId > 0) { "userId 必须是正数" } check(PluginConfig.profileEnabled) { "历史用户画像分析未启用" } check(UserProfileStore.isAvailable) { "用户画像数据库不可用" } - return concurrencyLimiter.withPermit { + if (!runningCompactions.add(userId)) { val profile = withContext(Dispatchers.IO) { UserProfileStore.load(userId) } ?: throw IllegalArgumentException("用户 $userId 尚无画像") - if (profile.items.isEmpty()) { - return@withPermit ProfileCompactionReport( + return unchangedCompactionReport(profile, alreadyRunning = true) + } + + try { + return userLocks.withUserLocks(listOf(userId)) locked@{ + val profile = withContext(Dispatchers.IO) { UserProfileStore.load(userId) } + ?: throw IllegalArgumentException("用户 $userId 尚无画像") + if (!runGate.canContinue(runToken)) { + return@locked unchangedCompactionReport(profile, stopped = true) + } + if (profile.items.isEmpty()) { + return@locked unchangedCompactionReport(profile) + } + + val endpoint = checkNotNull(LargeLanguageModels.profile) { "画像分析模型未配置" } + val model: ProfileCompactionModel = ProfileModelClient(endpoint) + val supportStats = withContext(Dispatchers.IO) { UserProfileStore.loadSupportStats(userId) } + val (result, plan) = compactWithRetry(model, profile, supportStats) + if (plan.reduction.profile.version != profile.version) { + val batch = compactionBatch(profile, result.rawResponse) + withContext(Dispatchers.IO) { + UserProfileStore.commitCompaction(plan, batch, result.usage) + } + ProfileOperationLogger.log( + context = "source=COMPACTION user=$userId", + reductions = listOf(plan.reduction), + ) + } + ProfileCompactionReport( userId = userId, beforeItems = profile.items.size, - afterItems = profile.items.size, - mergedGroups = 0, - rewrittenItems = 0, - deletedItems = 0, - summaryChanged = false, - skippedOperations = 0, - usage = ProfileTokenUsage(), - profile = profile, + afterItems = plan.reduction.profile.items.size, + mergedGroups = plan.mergedGroups, + rewrittenItems = plan.rewrittenItems, + deletedItems = plan.deletedItems, + summaryChanged = plan.reduction.profile.summary != profile.summary, + skippedOperations = plan.skippedOperations.size, + usage = result.usage, + profile = plan.reduction.profile, ) } - - val endpoint = checkNotNull(LargeLanguageModels.profile) { "画像分析模型未配置" } - val model: ProfileCompactionModel = ProfileModelClient(endpoint) - val supportStats = withContext(Dispatchers.IO) { UserProfileStore.loadSupportStats(userId) } - val (result, plan) = compactWithRetry(model, profile, supportStats) - if (plan.reduction.profile.version != profile.version) { - val batch = compactionBatch(profile, result.rawResponse) - withContext(Dispatchers.IO) { - UserProfileStore.commitCompaction(plan, batch, result.usage) - } - } - ProfileCompactionReport( - userId = userId, - beforeItems = profile.items.size, - afterItems = plan.reduction.profile.items.size, - mergedGroups = plan.mergedGroups, - rewrittenItems = plan.rewrittenItems, - deletedItems = plan.deletedItems, - summaryChanged = plan.reduction.profile.summary != profile.summary, - skippedOperations = plan.skippedOperations.size, - usage = result.usage, - profile = plan.reduction.profile, - ) + } finally { + runningCompactions.remove(userId) } } suspend fun analyzeGroup( groupId: Long, maxBatches: Int, + runToken: ProfileAnalysisRunToken = newRunToken(), onProgress: suspend (GroupProfileAnalysisProgress) -> Unit = {}, ): GroupProfileAnalysisReport { require(groupId > 0) { "groupId 必须是正数" } @@ -127,9 +150,7 @@ object UserProfileAnalysisService { } try { - return concurrencyLimiter.withPermit { - analyzeGroupExclusive(groupId, maxBatches, onProgress) - } + return analyzeGroupExclusive(groupId, maxBatches, runToken, onProgress) } finally { runningGroups.remove(groupId) } @@ -138,6 +159,7 @@ object UserProfileAnalysisService { private suspend fun analyzeExclusive( userId: Long, maxBatches: Int, + runToken: ProfileAnalysisRunToken, onProgress: suspend (ProfileAnalysisProgress) -> Unit, ): ProfileAnalysisReport { val endpoint = checkNotNull(LargeLanguageModels.profile) { @@ -166,7 +188,7 @@ object UserProfileAnalysisService { var totalUsage = ProfileTokenUsage() var caughtUp = false - while (processedBatches < maxBatches) { + while (processedBatches < maxBatches && runGate.canContinue(runToken)) { val batch = withContext(Dispatchers.IO) { reader.loadNextBatch( userId = userId, @@ -195,6 +217,10 @@ object UserProfileAnalysisService { source = ProfileRevisionSource.BACKFILL, ) } + ProfileOperationLogger.log( + context = "source=BACKFILL batch=[${batch.startTime},${batch.endTime})", + reductions = listOf(reduction), + ) profile = reduction.profile processedBatches++ processedMessages += batch.messages.size @@ -215,6 +241,7 @@ object UserProfileAnalysisService { } if (!caughtUp && profile.cursorTime >= profile.snapshotEndTime) caughtUp = true + val stopped = processedBatches < maxBatches && !caughtUp && !runGate.canContinue(runToken) return ProfileAnalysisReport( userId = userId, processedBatches = processedBatches, @@ -224,12 +251,14 @@ object UserProfileAnalysisService { usage = totalUsage, profile = withContext(Dispatchers.IO) { UserProfileStore.load(userId) } ?: profile, caughtUp = caughtUp, + stopped = stopped, ) } private suspend fun analyzeGroupExclusive( groupId: Long, maxBatches: Int, + runToken: ProfileAnalysisRunToken, onProgress: suspend (GroupProfileAnalysisProgress) -> Unit, ): GroupProfileAnalysisReport { val reader = withContext(Dispatchers.IO) { ProfileHistoryReader(resolveHistoryFile()) } @@ -259,7 +288,7 @@ object UserProfileAnalysisService { var totalUsage = ProfileTokenUsage() var caughtUp = cursor.cursorTime >= cursor.snapshotEndTime - while (processedBatches < maxBatches && !caughtUp) { + while (processedBatches < maxBatches && !caughtUp && runGate.canContinue(runToken)) { val batch = withContext(Dispatchers.IO) { reader.loadNextConversationBatch( botId = cursor.botId, @@ -287,6 +316,7 @@ object UserProfileAnalysisService { retryMax = PluginConfig.profileRetryMax, summaryMaxLength = PluginConfig.profileSummaryMaxLength, onRetryFailure = { message, cause -> JChatGPT.logger.warning(message, cause) }, + onCommittedOperations = ProfileOperationLogger::log, ) cursor = cursor.copy( cursorTime = batch.endTime, @@ -328,6 +358,7 @@ object UserProfileAnalysisService { cursorTime = cursor.cursorTime, snapshotEndTime = cursor.snapshotEndTime, caughtUp = caughtUp, + stopped = processedBatches < maxBatches && !caughtUp && !runGate.canContinue(runToken), ) } @@ -342,29 +373,28 @@ object UserProfileAnalysisService { check(PluginConfig.profileEnabled) { "历史用户画像分析未启用" } check(UserProfileStore.isAvailable) { "用户画像数据库不可用" } - return concurrencyLimiter.withPermit { - val endpoint = checkNotNull(LargeLanguageModels.profile) { "画像分析模型未配置" } - val model: ConversationProfileModel = ProfileModelClient(endpoint) - val reader = withContext(Dispatchers.IO) { ProfileHistoryReader(resolveLiveHistoryFile()) } - val batch = withContext(Dispatchers.IO) { - reader.loadConversationBatch( - botId = botId, - groupId = groupId, - startTime = startTime, - endTime = endTime, - messageLimit = PluginConfig.profileAutoConversationMessageLimit.coerceIn(20, 500), - maxMessageChars = PluginConfig.profileMaxMessageChars.coerceAtLeast(80), - ) - } ?: return@withPermit null - analyzeConversationBatch( - batch = batch, - minAuthoredTextChars = minAuthoredTextChars, - model = model, - retryMax = PluginConfig.profileRetryMax, - summaryMaxLength = PluginConfig.profileSummaryMaxLength, - onRetryFailure = { message, cause -> JChatGPT.logger.warning(message, cause) }, + val endpoint = checkNotNull(LargeLanguageModels.profile) { "画像分析模型未配置" } + val model: ConversationProfileModel = ProfileModelClient(endpoint) + val reader = withContext(Dispatchers.IO) { ProfileHistoryReader(resolveLiveHistoryFile()) } + val batch = withContext(Dispatchers.IO) { + reader.loadConversationBatch( + botId = botId, + groupId = groupId, + startTime = startTime, + endTime = endTime, + messageLimit = PluginConfig.profileAutoConversationMessageLimit.coerceIn(20, 500), + maxMessageChars = PluginConfig.profileMaxMessageChars.coerceAtLeast(80), ) - } + } ?: return null + return analyzeConversationBatch( + batch = batch, + minAuthoredTextChars = minAuthoredTextChars, + model = model, + retryMax = PluginConfig.profileRetryMax, + summaryMaxLength = PluginConfig.profileSummaryMaxLength, + onRetryFailure = { message, cause -> JChatGPT.logger.warning(message, cause) }, + onCommittedOperations = ProfileOperationLogger::log, + ) } internal suspend fun analyzeConversationBatch( @@ -374,47 +404,55 @@ object UserProfileAnalysisService { retryMax: Int, summaryMaxLength: Int, onRetryFailure: (String, Throwable) -> Unit = { _, _ -> }, + onCommittedOperations: (String, Collection) -> Unit = { _, _ -> }, ): ConversationProfileAnalysisReport? { check(UserProfileStore.isAvailable) { "用户画像数据库不可用" } val eligibleUserIds = batch.authoredTextCharsByUser .filterValues { it >= minAuthoredTextChars.coerceAtLeast(1) } .keys if (eligibleUserIds.isEmpty()) return null - if (withContext(Dispatchers.IO) { UserProfileStore.isConversationProcessed(batch.inputHash) }) { - return null - } - val profiles = withContext(Dispatchers.IO) { - eligibleUserIds.associateWith { userId -> - UserProfileStore.load(userId) ?: UserProfileSnapshot( - userId = userId, - cursorTime = 0, - snapshotEndTime = 0, + return userLocks.withUserLocks(eligibleUserIds) locked@{ + if (withContext(Dispatchers.IO) { UserProfileStore.isConversationProcessed(batch.inputHash) }) { + return@locked null + } + val profiles = withContext(Dispatchers.IO) { + eligibleUserIds.associateWith { userId -> + UserProfileStore.load(userId) ?: UserProfileSnapshot( + userId = userId, + cursorTime = 0, + snapshotEndTime = 0, + ) + } + } + + val (result, reductions) = analyzeConversationWithRetry( + model = model, + profiles = profiles, + batch = batch, + eligibleUserIds = eligibleUserIds, + retryMax = retryMax, + summaryMaxLength = summaryMaxLength, + onRetryFailure = onRetryFailure, + ) + withContext(Dispatchers.IO) { + UserProfileStore.commitConversation( + reductions = reductions.map { reduction -> reduction to batch.forUser(reduction.profile.userId) }, + usage = result.usage, ) } - } - - val (result, reductions) = analyzeConversationWithRetry( - model = model, - profiles = profiles, - batch = batch, - eligibleUserIds = eligibleUserIds, - retryMax = retryMax, - summaryMaxLength = summaryMaxLength, - onRetryFailure = onRetryFailure, - ) - withContext(Dispatchers.IO) { - UserProfileStore.commitConversation( - reductions = reductions.map { reduction -> reduction to batch.forUser(reduction.profile.userId) }, + onCommittedOperations( + "source=CONVERSATION bot=${batch.botId} group=${batch.groupId} " + + "batch=[${batch.startTime},${batch.endTime})", + reductions, + ) + ConversationProfileAnalysisReport( + analyzedUsers = eligibleUserIds.size, + processedMessages = batch.messages.size, + appliedOperations = reductions.sumOf { it.operations.size }, + skippedOperations = reductions.sumOf { it.skippedOperations.size }, usage = result.usage, ) } - return ConversationProfileAnalysisReport( - analyzedUsers = eligibleUserIds.size, - processedMessages = batch.messages.size, - appliedOperations = reductions.sumOf { it.operations.size }, - skippedOperations = reductions.sumOf { it.skippedOperations.size }, - usage = result.usage, - ) } private suspend fun analyzeConversationWithRetry( @@ -586,6 +624,25 @@ object UserProfileAnalysisService { caughtUp = true, ) + private fun unchangedCompactionReport( + profile: UserProfileSnapshot, + alreadyRunning: Boolean = false, + stopped: Boolean = false, + ) = ProfileCompactionReport( + userId = profile.userId, + beforeItems = profile.items.size, + afterItems = profile.items.size, + mergedGroups = 0, + rewrittenItems = 0, + deletedItems = 0, + summaryChanged = false, + skippedOperations = 0, + usage = ProfileTokenUsage(), + profile = profile, + alreadyRunning = alreadyRunning, + stopped = stopped, + ) + private operator fun ProfileTokenUsage.plus(other: ProfileTokenUsage) = ProfileTokenUsage( promptTokens = promptTokens + other.promptTokens, completionTokens = completionTokens + other.completionTokens, diff --git a/src/main/kotlin/profile/UserProfileModels.kt b/src/main/kotlin/profile/UserProfileModels.kt index 7a0bdaa..7e789ad 100644 --- a/src/main/kotlin/profile/UserProfileModels.kt +++ b/src/main/kotlin/profile/UserProfileModels.kt @@ -261,6 +261,7 @@ data class GroupProfileAnalysisReport( val snapshotEndTime: Int, val caughtUp: Boolean, val alreadyRunning: Boolean = false, + val stopped: Boolean = false, ) data class ProfileAnalysisReport( @@ -273,4 +274,5 @@ data class ProfileAnalysisReport( val profile: UserProfileSnapshot?, val caughtUp: Boolean, val alreadyRunning: Boolean = false, + val stopped: Boolean = false, ) diff --git a/src/main/kotlin/profile/UserProfileStore.kt b/src/main/kotlin/profile/UserProfileStore.kt index 2c00508..63de85b 100644 --- a/src/main/kotlin/profile/UserProfileStore.kt +++ b/src/main/kotlin/profile/UserProfileStore.kt @@ -98,6 +98,19 @@ object UserProfileStore { } } + fun listUserIds(): List { + check(initialized) { "用户画像数据库尚未初始化" } + return openReadConnection().use { connection -> + connection.prepareStatement("SELECT user_id FROM user_profile ORDER BY user_id").use { statement -> + statement.executeQuery().use { results -> + buildList { + while (results.next()) add(results.getLong("user_id")) + } + } + } + } + } + fun loadSupportStats(userId: Long): Map { check(initialized) { "用户画像数据库尚未初始化" } return openReadConnection().use { connection -> diff --git a/src/test/kotlin/command/ProfileCommandArgumentsTest.kt b/src/test/kotlin/command/ProfileCommandArgumentsTest.kt new file mode 100644 index 0000000..d7600d7 --- /dev/null +++ b/src/test/kotlin/command/ProfileCommandArgumentsTest.kt @@ -0,0 +1,30 @@ +package top.jie65535.mirai.command + +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFailsWith + +class ProfileCommandArgumentsTest { + @Test + fun parsesAndDeduplicatesMultipleGroupIds() { + assertEquals( + listOf(111L, 222L, 333L), + parseProfileGroupIds("111, 222，333;111"), + ) + } + + @Test + fun keepsSingleGroupCommandCompatible() { + assertEquals(listOf(818800431L), parseProfileGroupIds("818800431")) + } + + @Test + fun parsesMultipleUserIds() { + assertEquals(listOf(100L, 200L), parseProfileUserIds("100,200,100")) + } + + @Test + fun rejectsInvalidGroupId() { + assertFailsWith { parseProfileGroupIds("111,abc") } + } +} diff --git a/src/test/kotlin/profile/ProfileAnalysisRunGateTest.kt b/src/test/kotlin/profile/ProfileAnalysisRunGateTest.kt new file mode 100644 index 0000000..9546452 --- /dev/null +++ b/src/test/kotlin/profile/ProfileAnalysisRunGateTest.kt @@ -0,0 +1,18 @@ +package top.jie65535.mirai.profile + +import kotlin.test.Test +import kotlin.test.assertFalse +import kotlin.test.assertTrue + +class ProfileAnalysisRunGateTest { + @Test + fun stopInvalidatesExistingTokensButNotFutureRuns() { + val gate = ProfileAnalysisRunGate() + val existing = gate.newToken() + + gate.stopCurrentRuns() + + assertFalse(gate.canContinue(existing)) + assertTrue(gate.canContinue(gate.newToken())) + } +} diff --git a/src/test/kotlin/profile/ProfileOperationLoggerTest.kt b/src/test/kotlin/profile/ProfileOperationLoggerTest.kt new file mode 100644 index 0000000..3cc5a20 --- /dev/null +++ b/src/test/kotlin/profile/ProfileOperationLoggerTest.kt @@ -0,0 +1,59 @@ +package top.jie65535.mirai.profile + +import kotlin.test.Test +import kotlin.test.assertContains +import kotlin.test.assertFalse +import kotlin.test.assertNull + +class ProfileOperationLoggerTest { + @Test + fun formatsOnlyAppliedOperations() { + val reduction = ProfileReduction( + profile = UserProfileSnapshot( + userId = 100, + summary = "不应打印的摘要", + cursorTime = 0, + snapshotEndTime = 0, + ), + operations = listOf( + AppliedProfileOperation( + action = ProfileOperationAction.ADD, + itemId = "item-1", + category = ProfileCategory.INTEREST, + content = "关注 Kotlin\n开发", + confidence = ProfileConfidence.MEDIUM, + relatedUserId = null, + evidenceRefs = listOf(1), + ), + AppliedProfileOperation( + action = ProfileOperationAction.DELETE, + itemId = "item-2", + category = ProfileCategory.NOTABLE_FACT, + content = "已经过期的信息", + confidence = ProfileConfidence.LOW, + relatedUserId = null, + evidenceRefs = listOf(2), + ), + ), + ) + + val output = ProfileOperationLogger.format("source=CONVERSATION group=10", listOf(reduction)) + + assertContains(checkNotNull(output), "operations=2") + assertContains(output, "user=100 action=ADD category=interest confidence=medium content=关注 Kotlin 开发") + assertContains(output, "action=DELETE") + assertFalse(output.contains("不应打印的摘要")) + assertFalse(output.contains("item-1")) + assertFalse(output.contains("evidence")) + } + + @Test + fun omitsBatchWhenThereAreNoOperations() { + val reduction = ProfileReduction( + profile = UserProfileSnapshot(userId = 100, cursorTime = 0, snapshotEndTime = 0), + operations = emptyList(), + ) + + assertNull(ProfileOperationLogger.format("source=BACKFILL", listOf(reduction))) + } +} diff --git a/src/test/kotlin/profile/ProfileUserLockManagerTest.kt b/src/test/kotlin/profile/ProfileUserLockManagerTest.kt new file mode 100644 index 0000000..ed37a58 --- /dev/null +++ b/src/test/kotlin/profile/ProfileUserLockManagerTest.kt @@ -0,0 +1,43 @@ +package top.jie65535.mirai.profile + +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.async +import kotlinx.coroutines.cancelAndJoin +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.delay +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeout +import kotlin.test.Test +import kotlin.test.assertEquals + +class ProfileUserLockManagerTest { + @Test + fun cancellationWhileWaitingReleasesAlreadyAcquiredUserLocks() = runBlocking { + val locks = ProfileUserLockManager() + val secondUserLocked = CompletableDeferred() + val releaseSecondUser = CompletableDeferred() + + coroutineScope { + val holder = async { + locks.withUserLocks(listOf(2)) { + secondUserLocked.complete(Unit) + releaseSecondUser.await() + } + } + secondUserLocked.await() + val waiter = async { + locks.withUserLocks(listOf(1, 2)) { error("已取消任务不应进入临界区") } + } + delay(50) + waiter.cancelAndJoin() + + withTimeout(1_000) { + locks.withUserLocks(listOf(1)) { } + } + releaseSecondUser.complete(Unit) + holder.await() + } + + assertEquals(0, locks.activeLockCount) + } +} diff --git a/src/test/kotlin/profile/UserProfileAnalysisServiceTest.kt b/src/test/kotlin/profile/UserProfileAnalysisServiceTest.kt index 7f4314c..d2a2ef4 100644 --- a/src/test/kotlin/profile/UserProfileAnalysisServiceTest.kt +++ b/src/test/kotlin/profile/UserProfileAnalysisServiceTest.kt @@ -1,10 +1,16 @@ package top.jie65535.mirai.profile +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.async +import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeout +import kotlinx.coroutines.withTimeoutOrNull import net.mamoe.mirai.message.data.MessageSourceKind import top.jie65535.mirai.data.ChatMessageRecord import java.io.IOException import java.nio.file.Files +import java.util.concurrent.atomic.AtomicInteger import kotlin.test.Test import kotlin.test.assertEquals import kotlin.test.assertFailsWith @@ -88,6 +94,69 @@ class UserProfileAnalysisServiceTest { assertFalse(UserProfileStore.isConversationProcessed(INPUT_HASH)) } + @Test + fun analyzesDisjointGroupsConcurrently() = withProfileStore { + val entered = AtomicInteger() + val bothEntered = CompletableDeferred() + val release = CompletableDeferred() + fun concurrentModel(content: String) = InspectingConversationProfileModel { + if (entered.incrementAndGet() == 2) bothEntered.complete(Unit) + release.await() + result(responseFor("U1", content, evidenceRef = 1)) + } + + coroutineScope { + val first = async { + analyze(singleUserBatch(USER_A, 10, "parallel-a"), concurrentModel("用户 A 的信息")) + } + val second = async { + analyze(singleUserBatch(USER_B, 20, "parallel-b"), concurrentModel("用户 B 的信息")) + } + withTimeout(1_000) { bothEntered.await() } + release.complete(Unit) + assertNotNull(first.await()) + assertNotNull(second.await()) + } + } + + @Test + fun sharedUserBatchWaitsAndLoadsLatestCommittedProfile() = withProfileStore { + val firstEntered = CompletableDeferred() + val releaseFirst = CompletableDeferred() + val secondStarted = CompletableDeferred() + val secondEntered = CompletableDeferred() + val firstModel = InspectingConversationProfileModel { profiles -> + assertEquals(0, profiles.getValue(USER_A).version) + firstEntered.complete(Unit) + releaseFirst.await() + result(responseFor("U1", "第一群归纳的信息", evidenceRef = 1)) + } + val secondModel = InspectingConversationProfileModel { profiles -> + assertEquals(1, profiles.getValue(USER_A).version) + secondEntered.complete(Unit) + result(responseFor("U1", "第二群归纳的信息", evidenceRef = 1)) + } + + coroutineScope { + val first = async { analyze(singleUserBatch(USER_A, 10, "shared-a"), firstModel) } + firstEntered.await() + val second = async { + secondStarted.complete(Unit) + analyze(singleUserBatch(USER_A, 20, "shared-b"), secondModel) + } + secondStarted.await() + assertNull(withTimeoutOrNull(100) { secondEntered.await() }) + releaseFirst.complete(Unit) + assertNotNull(first.await()) + withTimeout(1_000) { secondEntered.await() } + assertNotNull(second.await()) + } + + val profile = assertNotNull(UserProfileStore.load(USER_A)) + assertEquals(2, profile.version) + assertEquals(2, profile.items.size) + } + private suspend fun analyze( batch: ConversationProfileBatch, model: ConversationProfileModel, @@ -145,11 +214,25 @@ class UserProfileAnalysisServiceTest { inputHash = INPUT_HASH, ) - private fun message(ref: Int, fromId: Long, text: String) = ProfilePromptMessage( + private fun singleUserBatch( + userId: Long, + groupId: Long, + inputHash: String, + ) = ConversationProfileBatch( + botId = BOT, + groupId = groupId, + startTime = 100, + endTime = 200, + messages = listOf(message(1, userId, "用于并发画像测试的信息", groupId)), + aliases = mapOf(BOT to "BOT", userId to "U1"), + inputHash = inputHash, + ) + + private fun message(ref: Int, fromId: Long, text: String, groupId: Long = GROUP) = ProfilePromptMessage( record = ChatMessageRecord( botId = BOT, fromId = fromId, - targetId = GROUP, + targetId = groupId, ids = null, internalIds = null, time = 120 + ref, @@ -190,6 +273,18 @@ class UserProfileAnalysisServiceTest { } } + private class InspectingConversationProfileModel( + private val behavior: suspend (Map) -> ConversationProfileModelResult, + ) : ConversationProfileModel { + override val modelName: String = "inspecting-profile-model" + + override suspend fun analyzeConversation( + profiles: Map, + batch: ConversationProfileBatch, + eligibleUserIds: Set, + ): ConversationProfileModelResult = behavior(profiles) + } + companion object { private const val BOT = 1L private const val GROUP = 10L diff --git a/src/test/kotlin/profile/UserProfileStoreTest.kt b/src/test/kotlin/profile/UserProfileStoreTest.kt index e9782a8..db35845 100644 --- a/src/test/kotlin/profile/UserProfileStoreTest.kt +++ b/src/test/kotlin/profile/UserProfileStoreTest.kt @@ -210,6 +210,7 @@ class UserProfileStoreTest { assertNotNull(UserProfileStore.load(100)) assertNotNull(UserProfileStore.load(200)) + assertEquals(listOf(100L, 200L), UserProfileStore.listUserIds()) assertTrue(UserProfileStore.isConversationProcessed("shared-conversation-hash")) val database = directory.resolve("user-profile.sqlite") DriverManager.getConnection("jdbc:sqlite:${database.toAbsolutePath()}").use { connection ->