mirror of
https://github.com/jie65535/JChatGPT.git
synced 2026-09-15 02:56:10 +08:00
profile: schedule daily group updates
This commit is contained in:
@@ -34,6 +34,7 @@ import top.jie65535.mirai.data.SkillStore
|
||||
import top.jie65535.mirai.data.TokenUsageStore
|
||||
import top.jie65535.mirai.llm.LargeLanguageModels
|
||||
import top.jie65535.mirai.profile.ProfileAutoMaintenance
|
||||
import top.jie65535.mirai.profile.ProfileDailyMaintenance
|
||||
import top.jie65535.mirai.profile.UserProfileStore
|
||||
import kotlin.random.Random
|
||||
|
||||
@@ -77,6 +78,7 @@ object JChatGPT : KotlinPlugin(
|
||||
.onFailure { logger.error("初始化用户画像数据库失败，画像分析将暂时禁用", it) }
|
||||
|
||||
LargeLanguageModels.reload()
|
||||
ProfileDailyMaintenance.reload()
|
||||
PluginCommands.register()
|
||||
keyword = PluginConfig.callKeyword.takeIf(String::isNotEmpty)?.let(::Regex)
|
||||
|
||||
@@ -104,6 +106,7 @@ object JChatGPT : KotlinPlugin(
|
||||
ConversationEngine.clear()
|
||||
ConversationContext.clearAll()
|
||||
ProfileAutoMaintenance.clear()
|
||||
ProfileDailyMaintenance.clear()
|
||||
ContactSnapshotRefresher.clear()
|
||||
UserProfileStore.close()
|
||||
ContactSnapshotStore.close()
|
||||
|
||||
@@ -26,6 +26,7 @@ import top.jie65535.mirai.profile.ProfileAnalysisReport
|
||||
import top.jie65535.mirai.profile.ProfileAutoMaintenance
|
||||
import top.jie65535.mirai.profile.ProfileCategory
|
||||
import top.jie65535.mirai.profile.ProfileCompactionReport
|
||||
import top.jie65535.mirai.profile.ProfileDailyMaintenance
|
||||
import top.jie65535.mirai.profile.ProfilePersistentText
|
||||
import top.jie65535.mirai.profile.UserProfileAnalysisService
|
||||
import top.jie65535.mirai.profile.UserProfileSnapshot
|
||||
@@ -46,6 +47,7 @@ object PluginCommands : CompositeCommand(
|
||||
PluginConfig.reload()
|
||||
PluginData.reload()
|
||||
LargeLanguageModels.reload()
|
||||
ProfileDailyMaintenance.reload()
|
||||
if (!PluginConfig.profileEnabled || !PluginConfig.profileAutoUpdateEnabled) {
|
||||
ProfileAutoMaintenance.clear()
|
||||
}
|
||||
|
||||
@@ -99,6 +99,15 @@ object PluginConfig : AutoSavePluginConfig("Config") {
|
||||
@ValueDescription("全量推进群画像时，群内至少有多少条尚未处理的消息才启动；显式指定群号不受限制")
|
||||
val profileBulkGroupMinPendingMessages: Int by value(20)
|
||||
|
||||
@ValueDescription("是否每天定时推进已接近历史水位线的群画像")
|
||||
val profileDailyGroupUpdateEnabled: Boolean by value(false)
|
||||
|
||||
@ValueDescription("每日群画像推进时间，使用服务器本地时区，格式 HH:mm")
|
||||
val profileDailyGroupUpdateTime: String by value("04:30")
|
||||
|
||||
@ValueDescription("每日群画像只自动推进最早待处理消息距今不超过多少天的群，必须为正数")
|
||||
val profileDailyGroupUpdateMaxPendingAgeDays: Int by value(7)
|
||||
|
||||
@ValueDescription("每个画像分析批次最多包含多少个离散对话片段；同一秒的消息仍会一起处理")
|
||||
val profileBatchMaxEpisodes: Int by value(16)
|
||||
|
||||
|
||||
@@ -0,0 +1,202 @@
|
||||
package top.jie65535.mirai.profile
|
||||
|
||||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.coroutines.Job
|
||||
import kotlinx.coroutines.async
|
||||
import kotlinx.coroutines.awaitAll
|
||||
import kotlinx.coroutines.coroutineScope
|
||||
import kotlinx.coroutines.currentCoroutineContext
|
||||
import kotlinx.coroutines.delay
|
||||
import kotlinx.coroutines.isActive
|
||||
import kotlinx.coroutines.launch
|
||||
import top.jie65535.mirai.JChatGPT
|
||||
import top.jie65535.mirai.config.PluginConfig
|
||||
import top.jie65535.mirai.llm.LargeLanguageModels
|
||||
import java.time.Duration
|
||||
import java.time.Instant
|
||||
import java.time.LocalTime
|
||||
import java.time.ZoneId
|
||||
import java.time.ZonedDateTime
|
||||
import java.time.format.DateTimeFormatter
|
||||
import java.time.format.DateTimeFormatterBuilder
|
||||
import java.time.format.ResolverStyle
|
||||
import java.time.temporal.ChronoField
|
||||
import java.util.Locale
|
||||
|
||||
object ProfileDailyMaintenance {
|
||||
private val lock = Any()
|
||||
private var schedulerJob: Job? = null
|
||||
|
||||
fun reload() {
|
||||
synchronized(lock) {
|
||||
schedulerJob?.cancel()
|
||||
schedulerJob = null
|
||||
if (!PluginConfig.profileEnabled || !PluginConfig.profileDailyGroupUpdateEnabled) return@synchronized
|
||||
if (!UserProfileStore.isAvailable) {
|
||||
JChatGPT.logger.warning("用户画像数据库不可用，每日群画像推进未启动")
|
||||
return@synchronized
|
||||
}
|
||||
|
||||
val scheduledTime = parseProfileDailyUpdateTime(PluginConfig.profileDailyGroupUpdateTime)
|
||||
if (scheduledTime == null) {
|
||||
JChatGPT.logger.warning(
|
||||
"每日群画像推进时间无效: '${PluginConfig.profileDailyGroupUpdateTime}'，请使用 HH:mm 格式"
|
||||
)
|
||||
return@synchronized
|
||||
}
|
||||
val maxPendingAgeDays = PluginConfig.profileDailyGroupUpdateMaxPendingAgeDays
|
||||
if (maxPendingAgeDays <= 0) {
|
||||
JChatGPT.logger.warning("每日群画像最大积压天数必须为正数，当前值: $maxPendingAgeDays")
|
||||
return@synchronized
|
||||
}
|
||||
|
||||
val zoneId = ZoneId.systemDefault()
|
||||
schedulerJob = JChatGPT.launch {
|
||||
scheduleLoop(scheduledTime, zoneId, maxPendingAgeDays)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fun clear() {
|
||||
synchronized(lock) {
|
||||
schedulerJob?.cancel()
|
||||
schedulerJob = null
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun scheduleLoop(
|
||||
scheduledTime: LocalTime,
|
||||
zoneId: ZoneId,
|
||||
maxPendingAgeDays: Int,
|
||||
) {
|
||||
while (currentCoroutineContext().isActive) {
|
||||
val now = ZonedDateTime.now(zoneId)
|
||||
val nextRun = nextProfileDailyUpdateAt(now, scheduledTime)
|
||||
val delayMillis = Duration.between(now.toInstant(), nextRun.toInstant()).toMillis().coerceAtLeast(1L)
|
||||
JChatGPT.logger.info(
|
||||
"每日群画像推进已计划: next=$nextRun maxPendingAgeDays=$maxPendingAgeDays"
|
||||
)
|
||||
delay(delayMillis)
|
||||
try {
|
||||
runOnce(maxPendingAgeDays)
|
||||
} catch (cause: CancellationException) {
|
||||
throw cause
|
||||
} catch (cause: Exception) {
|
||||
JChatGPT.logger.error("每日群画像推进失败，将在下一计划时间重试", cause)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun runOnce(maxPendingAgeDays: Int) {
|
||||
if (!PluginConfig.profileEnabled || !PluginConfig.profileDailyGroupUpdateEnabled) return
|
||||
if (!UserProfileStore.isAvailable) {
|
||||
JChatGPT.logger.warning("用户画像数据库不可用，跳过本次每日群画像推进")
|
||||
return
|
||||
}
|
||||
if (LargeLanguageModels.profile == null) {
|
||||
JChatGPT.logger.warning("画像分析模型未配置，跳过本次每日群画像推进")
|
||||
return
|
||||
}
|
||||
|
||||
val oldestAllowedTime = oldestAllowedProfilePendingTime(Instant.now().epochSecond, maxPendingAgeDays)
|
||||
val groupIds = UserProfileAnalysisService.listRecentPendingHistoryGroupIds(oldestAllowedTime)
|
||||
if (groupIds.isEmpty()) {
|
||||
JChatGPT.logger.info("每日群画像推进无需运行：没有符合 $maxPendingAgeDays 天积压门槛的群")
|
||||
return
|
||||
}
|
||||
|
||||
JChatGPT.logger.info("每日群画像推进开始: groups=${groupIds.size} maxPendingAgeDays=$maxPendingAgeDays")
|
||||
val runToken = UserProfileAnalysisService.newRunToken()
|
||||
val outcomes = coroutineScope {
|
||||
groupIds.map { groupId ->
|
||||
async { analyzeGroup(groupId, runToken) }
|
||||
}.awaitAll()
|
||||
}
|
||||
|
||||
val reports = outcomes.mapNotNull(DailyGroupOutcome::report)
|
||||
JChatGPT.logger.info(
|
||||
"每日群画像推进完成: selected=${groupIds.size} " +
|
||||
"success=${outcomes.count { it.status == DailyGroupStatus.SUCCESS }} " +
|
||||
"alreadyRunning=${outcomes.count { it.status == DailyGroupStatus.ALREADY_RUNNING }} " +
|
||||
"missing=${outcomes.count { it.status == DailyGroupStatus.MISSING_HISTORY }} " +
|
||||
"stopped=${outcomes.count { it.status == DailyGroupStatus.STOPPED }} " +
|
||||
"failed=${outcomes.count { it.status == DailyGroupStatus.FAILED }} " +
|
||||
"batches=${reports.sumOf(GroupProfileAnalysisReport::processedBatches)} " +
|
||||
"messages=${reports.sumOf(GroupProfileAnalysisReport::processedMessages)} " +
|
||||
"operations=${reports.sumOf(GroupProfileAnalysisReport::appliedOperations)} " +
|
||||
"tokens=${reports.sumOf { it.usage.promptTokens }}/${reports.sumOf { it.usage.completionTokens }} " +
|
||||
"cached=${reports.sumOf { it.usage.cachedTokens }}"
|
||||
)
|
||||
}
|
||||
|
||||
private suspend fun analyzeGroup(
|
||||
groupId: Long,
|
||||
runToken: ProfileAnalysisRunToken,
|
||||
): DailyGroupOutcome {
|
||||
return try {
|
||||
val report = UserProfileAnalysisService.analyzeGroup(
|
||||
groupId = groupId,
|
||||
maxBatches = Int.MAX_VALUE,
|
||||
runToken = runToken,
|
||||
) { progress ->
|
||||
JChatGPT.logger.info(
|
||||
"PROFILE_DAILY_BATCH group=$groupId batch=${progress.batchIndex} " +
|
||||
"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 status = when {
|
||||
report.alreadyRunning -> DailyGroupStatus.ALREADY_RUNNING
|
||||
report.botId == null -> DailyGroupStatus.MISSING_HISTORY
|
||||
report.stopped -> DailyGroupStatus.STOPPED
|
||||
else -> DailyGroupStatus.SUCCESS
|
||||
}
|
||||
DailyGroupOutcome(status, report.takeUnless { report.alreadyRunning || report.botId == null })
|
||||
} catch (cause: CancellationException) {
|
||||
throw cause
|
||||
} catch (cause: Exception) {
|
||||
JChatGPT.logger.error("群 $groupId 每日画像推进失败", cause)
|
||||
DailyGroupOutcome(DailyGroupStatus.FAILED)
|
||||
}
|
||||
}
|
||||
|
||||
private data class DailyGroupOutcome(
|
||||
val status: DailyGroupStatus,
|
||||
val report: GroupProfileAnalysisReport? = null,
|
||||
)
|
||||
|
||||
private enum class DailyGroupStatus {
|
||||
SUCCESS,
|
||||
ALREADY_RUNNING,
|
||||
MISSING_HISTORY,
|
||||
STOPPED,
|
||||
FAILED,
|
||||
}
|
||||
}
|
||||
|
||||
private val PROFILE_DAILY_TIME_FORMATTER: DateTimeFormatter = DateTimeFormatterBuilder()
|
||||
.parseStrict()
|
||||
.appendValue(ChronoField.HOUR_OF_DAY, 2)
|
||||
.appendLiteral(':')
|
||||
.appendValue(ChronoField.MINUTE_OF_HOUR, 2)
|
||||
.toFormatter(Locale.ROOT)
|
||||
.withResolverStyle(ResolverStyle.STRICT)
|
||||
|
||||
internal fun parseProfileDailyUpdateTime(value: String): LocalTime? = runCatching {
|
||||
LocalTime.parse(value.trim(), PROFILE_DAILY_TIME_FORMATTER)
|
||||
}.getOrNull()
|
||||
|
||||
internal fun nextProfileDailyUpdateAt(now: ZonedDateTime, scheduledTime: LocalTime): ZonedDateTime {
|
||||
val today = now.toLocalDate().atTime(scheduledTime).atZone(now.zone)
|
||||
return if (today.isAfter(now)) today else now.toLocalDate().plusDays(1).atTime(scheduledTime).atZone(now.zone)
|
||||
}
|
||||
|
||||
internal fun oldestAllowedProfilePendingTime(nowEpochSecond: Long, maxPendingAgeDays: Int): Int {
|
||||
require(maxPendingAgeDays > 0) { "maxPendingAgeDays must be positive" }
|
||||
return (nowEpochSecond - maxPendingAgeDays.toLong() * 24L * 60L * 60L)
|
||||
.coerceIn(0L, Int.MAX_VALUE.toLong())
|
||||
.toInt()
|
||||
}
|
||||
@@ -181,6 +181,37 @@ class ProfileHistoryReader(private val databaseFile: File) {
|
||||
}
|
||||
}
|
||||
|
||||
fun filterGroupRangesByOldestPendingMessageTime(
|
||||
ranges: List<GroupTimeBounds>,
|
||||
oldestAllowedTime: Int,
|
||||
): List<GroupTimeBounds> {
|
||||
if (ranges.isEmpty()) return emptyList()
|
||||
require(oldestAllowedTime >= 0) { "oldestAllowedTime must not be negative" }
|
||||
return openReadConnection().use { connection ->
|
||||
connection.prepareStatement(
|
||||
"""
|
||||
SELECT MIN(time) AS oldest_time
|
||||
FROM message_record
|
||||
WHERE bot_id = ? AND target_id = ? AND kind = ? AND recalled = 0
|
||||
AND time >= ? AND time < ?
|
||||
""".trimIndent()
|
||||
).use { statement ->
|
||||
ranges.filter { range ->
|
||||
statement.setLong(1, range.botId)
|
||||
statement.setLong(2, range.groupId)
|
||||
statement.setInt(3, MessageSourceKind.GROUP.ordinal)
|
||||
statement.setInt(4, range.startTime)
|
||||
statement.setInt(5, range.endTime)
|
||||
statement.executeQuery().use { results ->
|
||||
check(results.next()) { "读取群 ${range.groupId} 最早待处理消息失败" }
|
||||
val oldestTime = results.getInt("oldest_time")
|
||||
results.wasNull() || oldestTime >= oldestAllowedTime
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fun loadNextBatch(
|
||||
userId: Long,
|
||||
startTime: Int,
|
||||
|
||||
@@ -54,6 +54,25 @@ object UserProfileAnalysisService {
|
||||
}
|
||||
}
|
||||
|
||||
suspend fun listRecentPendingHistoryGroupIds(oldestAllowedMessageTime: Int): List<Long> {
|
||||
require(oldestAllowedMessageTime >= 0) { "oldestAllowedMessageTime must not be negative" }
|
||||
check(PluginConfig.profileEnabled) { "历史用户画像分析未启用" }
|
||||
check(UserProfileStore.isAvailable) { "用户画像数据库不可用" }
|
||||
return withContext(Dispatchers.IO) {
|
||||
val reader = ProfileHistoryReader(resolveHistoryFile())
|
||||
val historyBounds = reader.listGroupTimeBounds()
|
||||
val cursors = UserProfileStore.loadGroupCursors()
|
||||
.associateBy { cursor -> cursor.botId to cursor.groupId }
|
||||
val pendingRanges = historyBounds.mapNotNull { bounds ->
|
||||
pendingGroupAnalysisRange(bounds, cursors[bounds.botId to bounds.groupId])
|
||||
}
|
||||
reader.filterGroupRangesByOldestPendingMessageTime(
|
||||
ranges = pendingRanges,
|
||||
oldestAllowedTime = oldestAllowedMessageTime,
|
||||
).map(ProfileHistoryReader.GroupTimeBounds::groupId)
|
||||
}
|
||||
}
|
||||
|
||||
suspend fun analyze(
|
||||
userId: Long,
|
||||
maxBatches: Int,
|
||||
|
||||
@@ -0,0 +1,46 @@
|
||||
package top.jie65535.mirai.profile
|
||||
|
||||
import java.time.LocalTime
|
||||
import java.time.ZoneId
|
||||
import java.time.ZonedDateTime
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertNull
|
||||
|
||||
class ProfileDailyMaintenanceTest {
|
||||
@Test
|
||||
fun parsesStrictDailyUpdateTime() {
|
||||
assertEquals(LocalTime.of(4, 30), parseProfileDailyUpdateTime("04:30"))
|
||||
assertEquals(LocalTime.of(23, 59), parseProfileDailyUpdateTime(" 23:59 "))
|
||||
assertNull(parseProfileDailyUpdateTime("4:30"))
|
||||
assertNull(parseProfileDailyUpdateTime("24:00"))
|
||||
assertNull(parseProfileDailyUpdateTime("04:60"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun schedulesTodayOrTomorrowInServerTimeZone() {
|
||||
val zone = ZoneId.of("Asia/Shanghai")
|
||||
val scheduledTime = LocalTime.of(4, 30)
|
||||
|
||||
assertEquals(
|
||||
ZonedDateTime.of(2026, 8, 5, 4, 30, 0, 0, zone),
|
||||
nextProfileDailyUpdateAt(
|
||||
ZonedDateTime.of(2026, 8, 5, 3, 0, 0, 0, zone),
|
||||
scheduledTime,
|
||||
),
|
||||
)
|
||||
assertEquals(
|
||||
ZonedDateTime.of(2026, 8, 6, 4, 30, 0, 0, zone),
|
||||
nextProfileDailyUpdateAt(
|
||||
ZonedDateTime.of(2026, 8, 5, 4, 30, 0, 0, zone),
|
||||
scheduledTime,
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun calculatesOldestAllowedPendingMessageTime() {
|
||||
assertEquals(395_200, oldestAllowedProfilePendingTime(1_000_000, 7))
|
||||
assertEquals(0, oldestAllowedProfilePendingTime(100, 7))
|
||||
}
|
||||
}
|
||||
@@ -62,6 +62,8 @@ class ProfileHistoryReaderTest {
|
||||
insert(1, 20, 110, kind = MessageSourceKind.FRIEND)
|
||||
insert(2, 30, 100)
|
||||
insert(2, 30, 110)
|
||||
insert(1, 40, 100)
|
||||
insert(1, 40, 1_000)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -80,6 +82,13 @@ class ProfileHistoryReaderTest {
|
||||
listOf(10L, 20L),
|
||||
reader.filterGroupRangesByMinimumMessageCount(ranges, 1).map { it.groupId },
|
||||
)
|
||||
assertEquals(
|
||||
listOf(10L, 30L),
|
||||
reader.filterGroupRangesByOldestPendingMessageTime(
|
||||
ranges = ranges + ProfileHistoryReader.GroupTimeBounds(1, 40, 90, 1_001),
|
||||
oldestAllowedTime = 115,
|
||||
).map { it.groupId },
|
||||
)
|
||||
} finally {
|
||||
directory.toFile().deleteRecursively()
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user