diff --git a/README.md b/README.md index a7f278d..9e6f19b 100644 --- a/README.md +++ b/README.md @@ -135,6 +135,10 @@ profileAutoInjectMaxUsers: 4 profileAutoInjectSummaryMaxChars: 300 # 备用接入点冷却时间（分钟）。某接入点失败后在此时间内会被排到重试队尾，避免每条消息都先卡在故障接入点上。0为禁用 fallbackCooldownMinutes: 5 +# 对话、画像和视觉模型失败后的统一退避：首轮基础毫秒数、单次最大毫秒数；设为0可禁用 +# 实际等待会按失败次数指数增长，并加入20%以内的随机抖动，两个值最大均为60000 +retryBackoffBaseMillis: 1000 +retryBackoffMaxMillis: 10000 # 推理模型额外请求体JSON，会合并到请求体中。例如DeepSeek启用思维: {"thinking": {"type": "enabled"}} reasoningModelExtraBody: '' # 视觉模型额外请求体JSON，会合并到请求体中。 @@ -428,10 +432,10 @@ fallbackCooldownMinutes: 5 ### 工作机制 - 主接入点为列表首位，备用接入点按配置顺序排在其后。 -- **单次对话内**：某接入点流式调用失败时，立即切换到下一个接入点重试，而非反复重试同一个故障点。 +- **单次对话内**：某接入点流式调用失败时，经短暂指数退避后切换到下一个接入点重试，而非反复重试同一个故障点。 - **跨对话冷却**：失败的接入点进入冷却期（`fallbackCooldownMinutes` 分钟），冷却期内会被排到重试队尾。这样主接入点 key 到期后，后续消息会直接走健康的备用接入点，不必每条都先卡一次超时。调用成功或冷却到期后自动恢复。 - 仅在**LLM 调用本身失败**时才切换接入点，后续工具执行异常不会误判正常接入点为故障。 -- 推理模型、视觉模型不受影响，仍各自独立配置。 +- 备用接入点切换只作用于聊天模型；推理模型和视觉模型仍使用各自接入点，其中视觉模型自身的失败重试也使用统一退避。 - `/jgpt reload` 会重建接入点列表并清空冷却状态。 ## 工具系统 diff --git a/src/main/kotlin/config/PluginConfig.kt b/src/main/kotlin/config/PluginConfig.kt index a169687..fdc6105 100644 --- a/src/main/kotlin/config/PluginConfig.kt +++ b/src/main/kotlin/config/PluginConfig.kt @@ -138,6 +138,12 @@ object PluginConfig : AutoSavePluginConfig("Config") { @ValueDescription("备用接入点冷却时间（分钟）。某接入点失败后在此时间内会被排到重试队尾，避免每条消息都先卡在故障接入点上。设为0禁用，默认5分钟") val fallbackCooldownMinutes: Long by value(5L) + @ValueDescription("失败后首次重试的基础退避时间（毫秒），后续按指数增长并加入抖动。设为0禁用退避，最大60000") + val retryBackoffBaseMillis: Long by value(1000L) + + @ValueDescription("失败重试的最大退避时间（毫秒）。设为0禁用退避，最大60000") + val retryBackoffMaxMillis: Long by value(10000L) + @ValueDescription("推理模型额外请求体JSON，会合并到请求体中。例如DeepSeek启用思维: {\"thinking\": {\"type\": \"enabled\"}}") val reasoningModelExtraBody: String by value("") diff --git a/src/main/kotlin/conversation/ConversationEngine.kt b/src/main/kotlin/conversation/ConversationEngine.kt index 711b7d5..a0e6ac6 100644 --- a/src/main/kotlin/conversation/ConversationEngine.kt +++ b/src/main/kotlin/conversation/ConversationEngine.kt @@ -8,6 +8,7 @@ import com.aallam.openai.api.chat.ToolCall import com.aallam.openai.api.core.Usage import com.aallam.openai.api.model.ModelId import io.ktor.util.collections.ConcurrentSet +import kotlinx.coroutines.CancellationException import kotlinx.coroutines.Deferred import kotlinx.coroutines.async import kotlinx.coroutines.awaitAll @@ -46,6 +47,7 @@ import top.jie65535.mirai.tools.VisitWeb import top.jie65535.mirai.tools.VisualAgent import top.jie65535.mirai.tools.WeatherService import top.jie65535.mirai.tools.WebSearch +import top.jie65535.mirai.util.RetryBackoff import java.time.OffsetDateTime import java.time.format.DateTimeFormatter import kotlin.math.max @@ -128,6 +130,8 @@ internal object ConversationEngine { var endpointIndex = 0 var done: Boolean var retry = max(PluginConfig.retryMax, 3) + val retryBackoff = RetryBackoff.fromConfig() + var consecutiveFailures = 0 do { val endpoint = endpoints[min(endpointIndex, endpoints.lastIndex)] var streamingOk = false @@ -183,6 +187,7 @@ internal object ConversationEngine { streamingOk = true LargeLanguageModels.reportSuccess(endpoint) + consecutiveFailures = 0 val answer = responseContent?.replace(thinkRegex, "")?.trim() JChatGPT.logger.info("LLM Response: $answer") history += ChatMessage( @@ -217,24 +222,34 @@ internal object ConversationEngine { } } } catch (cause: Exception) { - if (!streamingOk) { + if (cause is CancellationException) throw cause + val failureMessage = if (!streamingOk) { LargeLanguageModels.reportFailure(endpoint) if (endpointIndex < endpoints.lastIndex) { endpointIndex++ - JChatGPT.logger.warning( - "接入点[${endpoint.label}]调用失败，切换备用接入点[${endpoints[endpointIndex].label}]重试", - cause, - ) + "接入点[${endpoint.label}]调用失败，将切换备用接入点[${endpoints[endpointIndex].label}]" } else { - JChatGPT.logger.warning("接入点[${endpoint.label}]调用失败，无更多备用接入点，重试中", cause) + "接入点[${endpoint.label}]调用失败，无更多备用接入点" } } else { - JChatGPT.logger.warning("调用llm后处理时发生异常，重试中", cause) + "调用llm后处理时发生异常" } - if (retry <= 1) throw cause + if (retry <= 1) { + JChatGPT.logger.warning("$failureMessage，已无剩余尝试", cause) + throw cause + } + consecutiveFailures++ + val retryDelayMillis = retryBackoff.delayMillis(consecutiveFailures) + JChatGPT.logger.warning( + "$failureMessage，将在 ${retryDelayMillis}ms 后重试", + cause, + ) + if (retryDelayMillis > 0) delay(retryDelayMillis) done = false } } while (!done && 0 < --retry) + } catch (cause: CancellationException) { + throw cause } catch (cause: Throwable) { JChatGPT.logger.warning(cause) event.subject.sendMessage("很抱歉，发生异常，请稍后重试") diff --git a/src/main/kotlin/profile/UserProfileAnalysisService.kt b/src/main/kotlin/profile/UserProfileAnalysisService.kt index 5342b0c..090b26f 100644 --- a/src/main/kotlin/profile/UserProfileAnalysisService.kt +++ b/src/main/kotlin/profile/UserProfileAnalysisService.kt @@ -2,11 +2,13 @@ package top.jie65535.mirai.profile import kotlinx.coroutines.CancellationException import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.delay import kotlinx.coroutines.withContext import top.jie65535.mirai.JChatGPT import top.jie65535.mirai.config.PluginConfig import top.jie65535.mirai.data.ChatHistoryStore import top.jie65535.mirai.llm.LargeLanguageModels +import top.jie65535.mirai.util.RetryBackoff import java.io.File import java.security.MessageDigest import java.util.concurrent.ConcurrentHashMap @@ -465,6 +467,7 @@ object UserProfileAnalysisService { onRetryFailure: (String, Throwable) -> Unit, ): Pair> { val attempts = retryMax.coerceIn(0, 3) + 1 + val retryBackoff = RetryBackoff.fromConfig() var lastFailure: Throwable? = null repeat(attempts) { attempt -> try { @@ -482,10 +485,14 @@ object UserProfileAnalysisService { } catch (cause: Exception) { if (cause is CancellationException) throw cause lastFailure = cause - onRetryFailure( - "群 ${batch.groupId} 会话画像 [${batch.startTime}, ${batch.endTime}) " + + handleRetryFailure( + attempt = attempt, + attempts = attempts, + message = "群 ${batch.groupId} 会话画像 [${batch.startTime}, ${batch.endTime}) " + "第 ${attempt + 1}/$attempts 次分析失败", - cause, + cause = cause, + retryBackoff = retryBackoff, + logFailure = onRetryFailure, ) } } @@ -502,6 +509,7 @@ object UserProfileAnalysisService { advanceBackfillCursor: Boolean = true, ): Pair { val attempts = PluginConfig.profileRetryMax.coerceIn(0, 3) + 1 + val retryBackoff = RetryBackoff.fromConfig() var lastFailure: Throwable? = null repeat(attempts) { attempt -> try { @@ -523,10 +531,14 @@ object UserProfileAnalysisService { } catch (cause: Exception) { if (cause is CancellationException) throw cause lastFailure = cause - JChatGPT.logger.warning( - "用户 ${batch.userId} 画像批次 [${batch.startTime}, ${batch.endTime}) " + + handleRetryFailure( + attempt = attempt, + attempts = attempts, + message = "用户 ${batch.userId} 画像批次 [${batch.startTime}, ${batch.endTime}) " + "第 ${attempt + 1}/$attempts 次分析失败", - cause, + cause = cause, + retryBackoff = retryBackoff, + logFailure = JChatGPT.logger::warning, ) } } @@ -542,6 +554,7 @@ object UserProfileAnalysisService { supportStats: Map, ): Pair { val attempts = PluginConfig.profileRetryMax.coerceIn(0, 3) + 1 + val retryBackoff = RetryBackoff.fromConfig() var lastFailure: Throwable? = null repeat(attempts) { attempt -> try { @@ -558,9 +571,13 @@ object UserProfileAnalysisService { } catch (cause: Exception) { if (cause is CancellationException) throw cause lastFailure = cause - JChatGPT.logger.warning( - "用户 ${profile.userId} 画像压缩第 ${attempt + 1}/$attempts 次失败", - cause, + handleRetryFailure( + attempt = attempt, + attempts = attempts, + message = "用户 ${profile.userId} 画像压缩第 ${attempt + 1}/$attempts 次失败", + cause = cause, + retryBackoff = retryBackoff, + logFailure = JChatGPT.logger::warning, ) } } @@ -570,6 +587,23 @@ object UserProfileAnalysisService { ) } + private suspend fun handleRetryFailure( + attempt: Int, + attempts: Int, + message: String, + cause: Throwable, + retryBackoff: RetryBackoff, + logFailure: (String, Throwable) -> Unit, + ) { + if (attempt + 1 >= attempts) { + logFailure("$message，已无剩余尝试", cause) + return + } + val retryDelayMillis = retryBackoff.delayMillis(attempt + 1) + logFailure("$message，将在 ${retryDelayMillis}ms 后重试", cause) + if (retryDelayMillis > 0) delay(retryDelayMillis) + } + private fun compactionBatch(profile: UserProfileSnapshot, rawResponse: String): ProfileHistoryBatch { val digest = MessageDigest.getInstance("SHA-256") .digest("${profile.userId}|${profile.version}|$rawResponse".toByteArray(Charsets.UTF_8)) diff --git a/src/main/kotlin/tools/VisualAgent.kt b/src/main/kotlin/tools/VisualAgent.kt index 8e984dd..77f45f2 100644 --- a/src/main/kotlin/tools/VisualAgent.kt +++ b/src/main/kotlin/tools/VisualAgent.kt @@ -25,6 +25,7 @@ import net.mamoe.mirai.event.events.MessageEvent import top.jie65535.mirai.JChatGPT import top.jie65535.mirai.config.PluginConfig import top.jie65535.mirai.llm.LargeLanguageModels +import top.jie65535.mirai.util.RetryBackoff import java.net.URI class VisualAgent : BaseAgent( @@ -118,6 +119,7 @@ class VisualAgent : BaseAgent( val messageContent = buildMessageContent(imageGroups, prompt) val maxAttempts = PluginConfig.visualRetryMax.coerceIn(1, 3) + val retryBackoff = RetryBackoff.fromConfig() var lastError: Throwable? = null repeat(maxAttempts) { attempt -> try { @@ -150,11 +152,12 @@ class VisualAgent : BaseAgent( if (!isRetryable(e)) throw e lastError = e if (attempt + 1 < maxAttempts) { + val retryDelayMillis = retryBackoff.delayMillis(attempt + 1) JChatGPT.logger.warning( - "视觉模型调用失败，将进行第 ${attempt + 2}/$maxAttempts 次尝试", + "视觉模型调用失败，将在 ${retryDelayMillis}ms 后进行第 ${attempt + 2}/$maxAttempts 次尝试", e ) - delay(RETRY_BASE_DELAY_MILLIS * (attempt + 1L)) + if (retryDelayMillis > 0) delay(retryDelayMillis) } } } @@ -165,7 +168,6 @@ class VisualAgent : BaseAgent( companion object { private const val VISUAL_MAX_CONCURRENCY = 2 - private const val RETRY_BASE_DELAY_MILLIS = 800L private const val MAX_SOURCE_IMAGES = 16 private const val MAX_MODEL_IMAGES = 32 private const val MAX_TOTAL_PAYLOAD_CHARS = 48_000_000 diff --git a/src/main/kotlin/util/RetryBackoff.kt b/src/main/kotlin/util/RetryBackoff.kt new file mode 100644 index 0000000..4949a40 --- /dev/null +++ b/src/main/kotlin/util/RetryBackoff.kt @@ -0,0 +1,46 @@ +package top.jie65535.mirai.util + +import top.jie65535.mirai.config.PluginConfig +import kotlin.random.Random + +internal class RetryBackoff( + initialDelayMillis: Long, + maxDelayMillis: Long, + private val randomFraction: () -> Double = { Random.nextDouble() }, +) { + private val initialDelayMillis = initialDelayMillis.coerceIn(0L, MAX_DELAY_MILLIS) + private val maxDelayMillis = maxDelayMillis.coerceIn(0L, MAX_DELAY_MILLIS) + + /** + * Returns the delay before the given retry. Retry numbers start at 1. + * A 20% downward jitter prevents concurrent failures from retrying in lockstep. + */ + fun delayMillis(retryNumber: Int): Long { + if (retryNumber <= 0 || initialDelayMillis == 0L || maxDelayMillis == 0L) return 0L + + var nominalDelay = minOf(initialDelayMillis, maxDelayMillis) + repeat((retryNumber - 1).coerceAtMost(MAX_EXPONENT)) { + nominalDelay = when { + nominalDelay >= maxDelayMillis -> maxDelayMillis + nominalDelay > maxDelayMillis / 2 -> maxDelayMillis + else -> nominalDelay * 2 + } + } + + val jitterWindow = nominalDelay / JITTER_DIVISOR + if (jitterWindow == 0L) return nominalDelay + val fraction = randomFraction().coerceIn(0.0, 1.0) + return nominalDelay - jitterWindow + (jitterWindow * fraction).toLong() + } + + companion object { + private const val JITTER_DIVISOR = 5L + private const val MAX_EXPONENT = 62 + const val MAX_DELAY_MILLIS = 60_000L + + fun fromConfig(): RetryBackoff = RetryBackoff( + initialDelayMillis = PluginConfig.retryBackoffBaseMillis, + maxDelayMillis = PluginConfig.retryBackoffMaxMillis, + ) + } +} diff --git a/src/test/kotlin/util/RetryBackoffTest.kt b/src/test/kotlin/util/RetryBackoffTest.kt new file mode 100644 index 0000000..8605be7 --- /dev/null +++ b/src/test/kotlin/util/RetryBackoffTest.kt @@ -0,0 +1,48 @@ +package top.jie65535.mirai.util + +import kotlin.test.Test +import kotlin.test.assertEquals + +class RetryBackoffTest { + @Test + fun growsExponentiallyAndStopsAtMaximum() { + val backoff = RetryBackoff( + initialDelayMillis = 1_000, + maxDelayMillis = 10_000, + randomFraction = { 1.0 }, + ) + + assertEquals(1_000, backoff.delayMillis(1)) + assertEquals(2_000, backoff.delayMillis(2)) + assertEquals(4_000, backoff.delayMillis(3)) + assertEquals(8_000, backoff.delayMillis(4)) + assertEquals(10_000, backoff.delayMillis(5)) + assertEquals(10_000, backoff.delayMillis(Int.MAX_VALUE)) + } + + @Test + fun appliesBoundedDownwardJitter() { + val minimum = RetryBackoff(1_000, 10_000) { 0.0 } + val maximum = RetryBackoff(1_000, 10_000) { 1.0 } + + assertEquals(800, minimum.delayMillis(1)) + assertEquals(1_000, maximum.delayMillis(1)) + assertEquals(3_200, minimum.delayMillis(3)) + assertEquals(4_000, maximum.delayMillis(3)) + } + + @Test + fun zeroOrInvalidRetryDisablesDelay() { + assertEquals(0, RetryBackoff(1_000, 10_000).delayMillis(0)) + assertEquals(0, RetryBackoff(0, 10_000).delayMillis(1)) + assertEquals(0, RetryBackoff(1_000, 0).delayMillis(1)) + } + + @Test + fun clampsMisconfiguredValuesToSafetyLimit() { + val backoff = RetryBackoff(Long.MAX_VALUE, Long.MAX_VALUE) { 1.0 } + + assertEquals(RetryBackoff.MAX_DELAY_MILLIS, backoff.delayMillis(1)) + assertEquals(RetryBackoff.MAX_DELAY_MILLIS, backoff.delayMillis(Int.MAX_VALUE)) + } +}