profile: require pending messages for bulk runs

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