From 31803396a8866e9d45623906ffece0905855c30d Mon Sep 17 00:00:00 2001 From: jie65535 Date: Tue, 4 Aug 2026 23:57:10 +0800 Subject: [PATCH] profile: require pending messages for bulk runs --- README.md | 7 +++-- src/main/kotlin/command/PluginCommands.kt | 16 +++++++++-- src/main/kotlin/config/PluginConfig.kt | 3 ++ .../profile/UserProfileAnalysisService.kt | 28 ++++++++++++++----- .../profile/UserProfileAnalysisServiceTest.kt | 22 +++++++++++++++ 5 files changed, 64 insertions(+), 12 deletions(-) diff --git a/README.md b/README.md index b57b6ed..46f3c8f 100644 --- a/README.md +++ b/README.md @@ -116,6 +116,8 @@ contactSnapshotRefreshIntervalMinutes: 1440 contactSnapshotGroupDelayMillis: 200 # 每批目标用户消息数、片段软间隔及上下文限制 profileBatchTargetMessages: 120 +# 全量推进群画像时，至少累计多少条尚未处理的有效群消息才启动 +profileBulkGroupMinPendingMessages: 20 profileBatchMaxEpisodes: 16 profileEpisodeGapMinutes: 60 profileContextBeforeMessages: 30 @@ -234,8 +236,9 @@ searchHistoryMaxRecords: 5000 模型只使用单次请求内有效的临时编号，不接触画像条目的内部 UUID；不合规建议会被跳过，不阻断其他有效更新。 临时用户别名不会写入最终画像。`profileCompact` 可独立清理重复或低价值条目，且不推进历史水位线。 旧历史可通过 `profileAnalyze` 或 `profileAnalyzeGroup` 手动分批推进。执行不带参数的 `profileAnalyzeGroup` 时， -插件会从画像历史库枚举所有含有效群消息的群，并在启动任务前排除游标已覆盖最新历史快照的群，只并发推进仍有 -历史待处理的群。画像模型通过应用层信号量按 `profileMaxConcurrentRequests` 限制同时执行的请求数，默认 128； +插件会从画像历史库枚举所有含有效群消息的群，并在启动任务前排除游标已覆盖最新历史快照的群。默认只有尚未 +处理的有效消息累计达到 `profileBulkGroupMinPendingMessages`（默认 20）才启动，避免低活跃群反复触发；显式指定 +群号时不受此门槛限制。画像模型通过应用层信号量按 `profileMaxConcurrentRequests` 限制同时执行的请求数，默认 128； 等待信号量的时间不计入首块响应超时，OkHttp 使用相同上限兜底。群批次只在读取画像快照和提交结果时短暂 持有用户锁，模型请求在锁外执行；提交前若发现画像已变化，会按稳定条目 ID 将同一响应的操作重放到最新版， 已不存在的操作目标会被跳过，不会重新请求模型。显式传入群号时仍只分析指定群。全量模式仅发送启动和最终汇总， diff --git a/src/main/kotlin/command/PluginCommands.kt b/src/main/kotlin/command/PluginCommands.kt index 2c46f86..32f597f 100644 --- a/src/main/kotlin/command/PluginCommands.kt +++ b/src/main/kotlin/command/PluginCommands.kt @@ -90,20 +90,30 @@ object PluginCommands : CompositeCommand( suspend fun CommandSender.profileAnalyzeGroup(groupIds: String = "", batches: Int = 1) { require(batches > 0) { "batches 必须是正数" } val analyzeAllGroups = groupIds.isBlank() + val minimumMessages = PluginConfig.profileBulkGroupMinPendingMessages.coerceAtLeast(1) val parsedGroupIds = if (analyzeAllGroups) { - UserProfileAnalysisService.listPendingHistoryGroupIds() + UserProfileAnalysisService.listPendingHistoryGroupIds(minimumMessages) } else { parseProfileGroupIds(groupIds) } if (parsedGroupIds.isEmpty()) { - sendMessage("没有待推进的群画像：历史库中无有效群消息，或所有群均已追平当前快照。") + sendMessage( + if (analyzeAllGroups) { + "没有达到启动条件的群画像：历史库中无有效群消息、群已追平，" + + "或尚未处理的消息少于 $minimumMessages 条。" + } else { + "没有可推进的指定群画像。" + } + ) return } val runToken = UserProfileAnalysisService.newRunToken() sendMessage( "已启动 ${parsedGroupIds.size} 个群的画像分析，每群最多推进 $batches 个批次。" + if (analyzeAllGroups) { - " 画像请求并发上限 ${normalizeMaxConcurrentRequests(PluginConfig.profileMaxConcurrentRequests)}。" + " 全量启动门槛 $minimumMessages 条待处理消息，" + + "画像请求并发上限 " + + "${normalizeMaxConcurrentRequests(PluginConfig.profileMaxConcurrentRequests)}。" } else { "" } diff --git a/src/main/kotlin/config/PluginConfig.kt b/src/main/kotlin/config/PluginConfig.kt index 0ce8e4a..db909d5 100644 --- a/src/main/kotlin/config/PluginConfig.kt +++ b/src/main/kotlin/config/PluginConfig.kt @@ -96,6 +96,9 @@ object PluginConfig : AutoSavePluginConfig("Config") { @ValueDescription("每个画像分析批次最多读取目标用户多少条消息") val profileBatchTargetMessages: Int by value(120) + @ValueDescription("全量推进群画像时，群内至少有多少条尚未处理的消息才启动；显式指定群号不受限制") + val profileBulkGroupMinPendingMessages: Int by value(20) + @ValueDescription("每个画像分析批次最多包含多少个离散对话片段；同一秒的消息仍会一起处理") val profileBatchMaxEpisodes: Int by value(16) diff --git a/src/main/kotlin/profile/UserProfileAnalysisService.kt b/src/main/kotlin/profile/UserProfileAnalysisService.kt index 362d9d4..d8521a4 100644 --- a/src/main/kotlin/profile/UserProfileAnalysisService.kt +++ b/src/main/kotlin/profile/UserProfileAnalysisService.kt @@ -32,17 +32,21 @@ object UserProfileAnalysisService { return report } - suspend fun listPendingHistoryGroupIds(): List { + suspend fun listPendingHistoryGroupIds( + minimumMessages: Int = PluginConfig.profileBulkGroupMinPendingMessages, + ): List { check(PluginConfig.profileEnabled) { "历史用户画像分析未启用" } check(UserProfileStore.isAvailable) { "用户画像数据库不可用" } return withContext(Dispatchers.IO) { - val historyBounds = ProfileHistoryReader(resolveHistoryFile()).listGroupTimeBounds() + val reader = ProfileHistoryReader(resolveHistoryFile()) + val historyBounds = reader.listGroupTimeBounds() val cursors = UserProfileStore.loadGroupCursors() .associateBy { cursor -> cursor.botId to cursor.groupId } - historyBounds.asSequence() - .filter { bounds -> - isGroupAnalysisPending(bounds, cursors[bounds.botId to bounds.groupId]) - } + val pendingRanges = historyBounds.mapNotNull { bounds -> + pendingGroupAnalysisRange(bounds, cursors[bounds.botId to bounds.groupId]) + } + reader.filterGroupRangesByMinimumMessageCount(pendingRanges, minimumMessages) + .asSequence() .map(ProfileHistoryReader.GroupTimeBounds::groupId) .toList() } @@ -758,4 +762,14 @@ object UserProfileAnalysisService { internal fun isGroupAnalysisPending( bounds: ProfileHistoryReader.GroupTimeBounds, cursor: GroupProfileCursor?, -): Boolean = cursor == null || cursor.cursorTime < maxOf(cursor.snapshotEndTime, bounds.endTime) +): Boolean = pendingGroupAnalysisRange(bounds, cursor) != null + +internal fun pendingGroupAnalysisRange( + bounds: ProfileHistoryReader.GroupTimeBounds, + cursor: GroupProfileCursor?, +): ProfileHistoryReader.GroupTimeBounds? { + val startTime = cursor?.cursorTime ?: bounds.startTime + val endTime = maxOf(cursor?.snapshotEndTime ?: bounds.endTime, bounds.endTime) + return bounds.copy(startTime = startTime, endTime = endTime) + .takeIf { startTime < endTime } +} diff --git a/src/test/kotlin/profile/UserProfileAnalysisServiceTest.kt b/src/test/kotlin/profile/UserProfileAnalysisServiceTest.kt index b5173e5..2f661bb 100644 --- a/src/test/kotlin/profile/UserProfileAnalysisServiceTest.kt +++ b/src/test/kotlin/profile/UserProfileAnalysisServiceTest.kt @@ -47,6 +47,28 @@ class UserProfileAnalysisServiceTest { GroupProfileCursor(BOT, GROUP, cursorTime = 500, snapshotEndTime = 500), ) ) + + assertEquals(bounds, pendingGroupAnalysisRange(bounds, null)) + assertEquals( + bounds.copy(startTime = 300), + pendingGroupAnalysisRange( + bounds, + GroupProfileCursor(BOT, GROUP, cursorTime = 300, snapshotEndTime = 500), + ) + ) + assertEquals( + bounds.copy(startTime = 400, endTime = 600), + pendingGroupAnalysisRange( + bounds, + GroupProfileCursor(BOT, GROUP, cursorTime = 400, snapshotEndTime = 600), + ) + ) + assertNull( + pendingGroupAnalysisRange( + bounds, + GroupProfileCursor(BOT, GROUP, cursorTime = 500, snapshotEndTime = 500), + ) + ) } @Test