From 74bcf0b8d68e9f3c6eb7ef037fc22e1c74e6c0b5 Mon Sep 17 00:00:00 2001 From: jie65535 Date: Mon, 3 Aug 2026 17:58:58 +0800 Subject: [PATCH] profile: rebase concurrent conversation updates --- README.md | 5 +- .../profile/ConversationProfileReducer.kt | 53 ++++++++ .../profile/UserProfileAnalysisService.kt | 118 ++++++++---------- .../profile/ConversationProfileReducerTest.kt | 81 ++++++++++++ .../profile/UserProfileAnalysisServiceTest.kt | 24 +--- 5 files changed, 198 insertions(+), 83 deletions(-) diff --git a/README.md b/README.md index 6b4c2f1..b57b6ed 100644 --- a/README.md +++ b/README.md @@ -237,8 +237,9 @@ searchHistoryMaxRecords: 5000 插件会从画像历史库枚举所有含有效群消息的群，并在启动任务前排除游标已覆盖最新历史快照的群，只并发推进仍有 历史待处理的群。画像模型通过应用层信号量按 `profileMaxConcurrentRequests` 限制同时执行的请求数，默认 128； 等待信号量的时间不计入首块响应超时，OkHttp 使用相同上限兜底。群批次只在读取画像快照和提交结果时短暂 -持有用户锁，模型请求在锁外执行；提交前若发现画像已变化，会加载最新版并重新分析，避免共享群友把网络请求 -串行化。显式传入群号时仍只分析指定群。全量模式仅发送启动和最终汇总，逐群进度与结果写入日志，避免回执刷屏。 +持有用户锁，模型请求在锁外执行；提交前若发现画像已变化，会按稳定条目 ID 将同一响应的操作重放到最新版， +已不存在的操作目标会被跳过，不会重新请求模型。显式传入群号时仍只分析指定群。全量模式仅发送启动和最终汇总， +逐群进度与结果写入日志，避免回执刷屏。 下一次正常群聊会自动携带触发者和最近发言者的认识。现有好感度、Bot 代号、标签和主观印象会与证据驱动的 长期画像按同一个人合并渲染，并明确给出长期画像条目数；私聊也会携带对方的可靠画像摘要和条目数。 diff --git a/src/main/kotlin/profile/ConversationProfileReducer.kt b/src/main/kotlin/profile/ConversationProfileReducer.kt index 52e52a5..bcbd56b 100644 --- a/src/main/kotlin/profile/ConversationProfileReducer.kt +++ b/src/main/kotlin/profile/ConversationProfileReducer.kt @@ -1,6 +1,8 @@ package top.jie65535.mirai.profile object ConversationProfileReducer { + private const val MISSING_ITEM_REFERENCE = "P2147483647" + fun reduce( profiles: Map, batch: ConversationProfileBatch, @@ -33,4 +35,55 @@ object ConversationProfileReducer { ) } } + + fun reduceRebased( + expectedProfiles: Map, + latestProfiles: Map, + batch: ConversationProfileBatch, + eligibleUserIds: Set, + response: ConversationProfileModelResponse, + model: String, + promptVersion: String, + summaryMaxLength: Int, + ): List { + val rebasedResponse = response.copy( + users = response.users.map { userResponse -> + val userId = batch.aliasToUserId[userResponse.userAlias] + val expected = userId?.let(expectedProfiles::get) + val latest = userId?.let(latestProfiles::get) + if (userId == null || userId !in eligibleUserIds || expected == null || latest == null) { + userResponse + } else { + userResponse.copy( + operations = userResponse.operations.map { operation -> + operation.rebaseItemReference(expected, latest) + } + ) + } + } + ) + return reduce( + profiles = latestProfiles, + batch = batch, + eligibleUserIds = eligibleUserIds, + response = rebasedResponse, + model = model, + promptVersion = promptVersion, + summaryMaxLength = summaryMaxLength, + ) + } + + private fun ProfileModelOperation.rebaseItemReference( + expected: UserProfileSnapshot, + latest: UserProfileSnapshot, + ): ProfileModelOperation { + if (action == ProfileOperationAction.ADD) return this + val expectedItem = ProfileItemReferences.resolve(expected, itemRef) + ?: return copy(itemRef = MISSING_ITEM_REFERENCE) + val latestReference = ProfileItemReferences.entries(latest) + .firstOrNull { entry -> entry.item.id == expectedItem.id } + ?.reference + ?: MISSING_ITEM_REFERENCE + return copy(itemRef = latestReference) + } } diff --git a/src/main/kotlin/profile/UserProfileAnalysisService.kt b/src/main/kotlin/profile/UserProfileAnalysisService.kt index 40b8d0d..362d9d4 100644 --- a/src/main/kotlin/profile/UserProfileAnalysisService.kt +++ b/src/main/kotlin/profile/UserProfileAnalysisService.kt @@ -14,8 +14,6 @@ import java.security.MessageDigest import java.util.concurrent.ConcurrentHashMap object UserProfileAnalysisService { - private const val MAX_CONVERSATION_CONFLICT_RETRIES = 3 - private val runningUsers = ConcurrentHashMap.newKeySet() private val runningGroups = ConcurrentHashMap.newKeySet() private val runningCompactions = ConcurrentHashMap.newKeySet() @@ -433,75 +431,70 @@ object UserProfileAnalysisService { .filterValues { it >= minAuthoredTextChars.coerceAtLeast(1) } .keys if (eligibleUserIds.isEmpty()) return null - var profiles = userLocks.withUserLocks(eligibleUserIds) { + val profiles = userLocks.withUserLocks(eligibleUserIds) { withContext(Dispatchers.IO) { if (UserProfileStore.isConversationProcessed(batch.inputHash)) null else loadConversationProfiles(eligibleUserIds) } } ?: return null - var conflictRetries = 0 - var totalUsage = ProfileTokenUsage() - - while (true) { - val (result, reductions) = analyzeConversationWithRetry( - model = model, - profiles = profiles, - batch = batch, - eligibleUserIds = eligibleUserIds, - retryMax = retryMax, - summaryMaxLength = summaryMaxLength, - onRetryFailure = onRetryFailure, - ) - totalUsage += result.usage - val commitOutcome = userLocks.withUserLocks(eligibleUserIds) { - withContext(Dispatchers.IO) { - if (UserProfileStore.isConversationProcessed(batch.inputHash)) { - ConversationCommitOutcome.AlreadyProcessed + val (result, reductions) = analyzeConversationWithRetry( + model = model, + profiles = profiles, + batch = batch, + eligibleUserIds = eligibleUserIds, + retryMax = retryMax, + summaryMaxLength = summaryMaxLength, + onRetryFailure = onRetryFailure, + ) + val totalUsage = result.usage + val commitOutcome = userLocks.withUserLocks(eligibleUserIds) { + withContext(Dispatchers.IO) { + if (UserProfileStore.isConversationProcessed(batch.inputHash)) { + ConversationCommitOutcome.AlreadyProcessed + } else { + val latestProfiles = loadConversationProfiles(eligibleUserIds) + val committedReductions = if (hasProfileVersionConflict(profiles, latestProfiles)) { + ConversationProfileReducer.reduceRebased( + expectedProfiles = profiles, + latestProfiles = latestProfiles, + batch = batch, + eligibleUserIds = eligibleUserIds, + response = result.response, + model = model.modelName, + promptVersion = ProfilePromptStore.PROMPT_VERSION, + summaryMaxLength = summaryMaxLength.coerceAtLeast(100), + ) } else { - val latestProfiles = loadConversationProfiles(eligibleUserIds) - if (hasProfileVersionConflict(profiles, latestProfiles)) { - ConversationCommitOutcome.Conflict(latestProfiles) - } else { - UserProfileStore.commitConversation( - reductions = reductions.map { reduction -> - reduction to batch.forUser(reduction.profile.userId) - }, - usage = totalUsage, - ) - ConversationCommitOutcome.Committed - } + reductions } - } - } - - when (commitOutcome) { - ConversationCommitOutcome.AlreadyProcessed -> return null - ConversationCommitOutcome.Committed -> { - onCommittedOperations( - "source=CONVERSATION bot=${batch.botId} group=${batch.groupId} " + - "batch=[${batch.startTime},${batch.endTime})", - reductions, - ) - return ConversationProfileAnalysisReport( - analyzedUsers = eligibleUserIds.size, - processedMessages = batch.messages.size, - appliedOperations = reductions.sumOf { it.operations.size }, - skippedOperations = reductions.sumOf { it.skippedOperations.size }, + UserProfileStore.commitConversation( + reductions = committedReductions.map { reduction -> + reduction to batch.forUser(reduction.profile.userId) + }, usage = totalUsage, ) - } - - is ConversationCommitOutcome.Conflict -> { - conflictRetries++ - if (conflictRetries > MAX_CONVERSATION_CONFLICT_RETRIES) { - throw IllegalStateException( - "群 ${batch.groupId} 会话画像提交连续冲突 $conflictRetries 次，未提交结果" - ) - } - profiles = commitOutcome.latestProfiles + ConversationCommitOutcome.Committed(committedReductions) } } } + + val committedReductions = when (commitOutcome) { + ConversationCommitOutcome.AlreadyProcessed -> return null + is ConversationCommitOutcome.Committed -> commitOutcome.reductions + } + + onCommittedOperations( + "source=CONVERSATION bot=${batch.botId} group=${batch.groupId} " + + "batch=[${batch.startTime},${batch.endTime})", + committedReductions, + ) + return ConversationProfileAnalysisReport( + analyzedUsers = eligibleUserIds.size, + processedMessages = batch.messages.size, + appliedOperations = committedReductions.sumOf { it.operations.size }, + skippedOperations = committedReductions.sumOf { it.skippedOperations.size }, + usage = totalUsage, + ) } private fun loadConversationProfiles(userIds: Set): Map = @@ -755,11 +748,10 @@ object UserProfileAnalysisService { } private sealed class ConversationCommitOutcome { - object Committed : ConversationCommitOutcome() - object AlreadyProcessed : ConversationCommitOutcome() - data class Conflict( - val latestProfiles: Map, + data class Committed( + val reductions: List, ) : ConversationCommitOutcome() + object AlreadyProcessed : ConversationCommitOutcome() } } diff --git a/src/test/kotlin/profile/ConversationProfileReducerTest.kt b/src/test/kotlin/profile/ConversationProfileReducerTest.kt index 22b37bf..7899220 100644 --- a/src/test/kotlin/profile/ConversationProfileReducerTest.kt +++ b/src/test/kotlin/profile/ConversationProfileReducerTest.kt @@ -144,6 +144,78 @@ class ConversationProfileReducerTest { assertFalse(prompt.contains(oldItem.id)) } + @Test + fun rebasesItemReferenceByStableIdWhenConcurrentUpdateChangesItemOrder() { + val target = item("target", "原始目标条目", 100) + val expected = emptyProfile(USER_A).copy(version = 1, items = listOf(target)) + val inserted = item("inserted", "并发新增条目", 50) + val latest = expected.copy(version = 2, items = listOf(inserted, target)) + + val reduction = ConversationProfileReducer.reduceRebased( + expectedProfiles = mapOf(USER_A to expected, USER_B to emptyProfile(USER_B)), + latestProfiles = mapOf(USER_A to latest, USER_B to emptyProfile(USER_B)), + batch = batch(), + eligibleUserIds = USERS, + response = ConversationProfileModelResponse( + users = listOf( + ConversationProfileUserResponse( + userAlias = "U1", + operations = listOf( + ProfileModelOperation( + action = ProfileOperationAction.UPDATE, + itemRef = "P1", + content = "更新后的目标条目", + evidenceRefs = listOf(1), + ) + ), + ) + ) + ), + model = "test-model", + promptVersion = "test-prompt", + summaryMaxLength = 500, + ).single { it.profile.userId == USER_A } + + assertEquals("并发新增条目", reduction.profile.items.single { it.id == "inserted" }.content) + assertEquals("更新后的目标条目", reduction.profile.items.single { it.id == "target" }.content) + assertEquals(1, reduction.operations.size) + } + + @Test + fun skipsRebasedOperationWhenConcurrentUpdateRemovedTargetItem() { + val target = item("target", "稍后被删除的条目", 100) + val expected = emptyProfile(USER_A).copy(version = 1, items = listOf(target)) + val latest = expected.copy(version = 2, items = emptyList()) + + val reduction = ConversationProfileReducer.reduceRebased( + expectedProfiles = mapOf(USER_A to expected, USER_B to emptyProfile(USER_B)), + latestProfiles = mapOf(USER_A to latest, USER_B to emptyProfile(USER_B)), + batch = batch(), + eligibleUserIds = USERS, + response = ConversationProfileModelResponse( + users = listOf( + ConversationProfileUserResponse( + userAlias = "U1", + operations = listOf( + ProfileModelOperation( + action = ProfileOperationAction.DELETE, + itemRef = "P1", + evidenceRefs = listOf(1), + ) + ), + ) + ) + ), + model = "test-model", + promptVersion = "test-prompt", + summaryMaxLength = 500, + ).single { it.profile.userId == USER_A } + + assertTrue(reduction.operations.isEmpty()) + assertTrue(reduction.skippedOperations.single().contains("不存在")) + assertTrue(reduction.profile.items.isEmpty()) + } + @Test fun acceptsMoreThanFourOperationsForOneUser() { val operations = (1..5).map { index -> @@ -297,6 +369,15 @@ class ConversationProfileReducerTest { snapshotEndTime = 0, ) + private fun item(id: String, content: String, firstSeenAt: Int) = UserProfileItem( + id = id, + category = ProfileCategory.NOTABLE_FACT, + content = content, + confidence = ProfileConfidence.MEDIUM, + firstSeenAt = firstSeenAt, + lastConfirmedAt = firstSeenAt, + ) + private fun message(ref: Int, fromId: Long, text: String) = ProfilePromptMessage( record = ChatMessageRecord( botId = BOT, diff --git a/src/test/kotlin/profile/UserProfileAnalysisServiceTest.kt b/src/test/kotlin/profile/UserProfileAnalysisServiceTest.kt index a4262d9..b5173e5 100644 --- a/src/test/kotlin/profile/UserProfileAnalysisServiceTest.kt +++ b/src/test/kotlin/profile/UserProfileAnalysisServiceTest.kt @@ -149,12 +149,11 @@ class UserProfileAnalysisServiceTest { } @Test - fun sharedUserRequestsRunConcurrentlyAndConflictReloadsLatestProfile() = withProfileStore { + fun sharedUserRequestsRunConcurrentlyAndConflictRebasesWithoutAnotherModelCall() = withProfileStore { val firstEntered = CompletableDeferred() val releaseFirst = CompletableDeferred() val secondFirstEntered = CompletableDeferred() val releaseSecondFirst = CompletableDeferred() - val secondRetried = CompletableDeferred() val secondCalls = AtomicInteger() val firstModel = InspectingConversationProfileModel { profiles -> assertEquals(0, profiles.getValue(USER_A).version) @@ -163,20 +162,10 @@ class UserProfileAnalysisServiceTest { result(responseFor("U1", "第一群归纳的信息", evidenceRef = 1)) } val secondModel = InspectingConversationProfileModel { profiles -> - when (secondCalls.incrementAndGet()) { - 1 -> { - assertEquals(0, profiles.getValue(USER_A).version) - secondFirstEntered.complete(Unit) - releaseSecondFirst.await() - } - - 2 -> { - assertEquals(1, profiles.getValue(USER_A).version) - secondRetried.complete(Unit) - } - - else -> error("共享用户冲突只应触发一次重算") - } + secondCalls.incrementAndGet() + assertEquals(0, profiles.getValue(USER_A).version) + secondFirstEntered.complete(Unit) + releaseSecondFirst.await() result(responseFor("U1", "第二群归纳的信息", evidenceRef = 1)) } @@ -188,11 +177,10 @@ class UserProfileAnalysisServiceTest { releaseFirst.complete(Unit) assertNotNull(first.await()) releaseSecondFirst.complete(Unit) - withTimeout(1_000) { secondRetried.await() } assertNotNull(second.await()) } - assertEquals(2, secondCalls.get()) + assertEquals(1, secondCalls.get()) val profile = assertNotNull(UserProfileStore.load(USER_A)) assertEquals(2, profile.version) assertEquals(2, profile.items.size)