From f416d889b3fc1c3463ee07b7c0e771e1b3b886da Mon Sep 17 00:00:00 2001 From: jie65535 Date: Wed, 5 Aug 2026 20:24:19 +0800 Subject: [PATCH] conversation: preserve triggers and await follow-ups --- README.md | 11 +- src/main/kotlin/JChatGPT.kt | 25 +- src/main/kotlin/config/PluginConfig.kt | 2 +- .../conversation/ConversationContext.kt | 11 +- .../kotlin/conversation/ConversationEngine.kt | 219 +++++++++++++++--- .../kotlin/conversation/ConversationKey.kt | 9 + .../conversation/ConversationRuntimeState.kt | 167 +++++++++++++ .../conversation/EndConversationDirective.kt | 46 ++++ src/main/kotlin/data/ChatHistoryStore.kt | 9 +- src/main/kotlin/tools/SearchChatHistory.kt | 2 +- src/main/kotlin/tools/StopLoopAgent.kt | 59 ++++- .../ConversationRuntimeStateTest.kt | 113 +++++++++ .../EndConversationDirectiveTest.kt | 90 +++++++ 13 files changed, 712 insertions(+), 51 deletions(-) create mode 100644 src/main/kotlin/conversation/ConversationKey.kt create mode 100644 src/main/kotlin/conversation/ConversationRuntimeState.kt create mode 100644 src/main/kotlin/conversation/EndConversationDirective.kt create mode 100644 src/test/kotlin/conversation/ConversationRuntimeStateTest.kt create mode 100644 src/test/kotlin/conversation/EndConversationDirectiveTest.kt diff --git a/README.md b/README.md index 04fa411..4f7d68d 100644 --- a/README.md +++ b/README.md @@ -24,6 +24,11 @@ JChatGPT 是一个基于 Kotlin 的 Mirai Console 插件，它将大型语言模 - 通过引用群友消息 + @bot 让 Bot 识别引用消息的内容 - 回复 bot 的消息即可引用对应的上下文对话（包括这个回复的历史对话） - 使用关键字触发（默认为 "[小筱][林淋月玥]"，可在配置中修改） +- Bot 处理期间出现的新触发会合并到当前会话，并在本轮工具结算后继续处理，不再因忙碌而直接丢弃 +- Bot 可以通过 `endConversation.waitForFollowUp` 短暂等待同一会话中指定用户的下一条消息；等待超时只会关闭观察状态，不会调用模型或发送消息 + +初次触发时，Bot 按 `historyWindowMin` 和 `historyMessageLimit` 读取有限的近期历史。模型运行期间以及缓存会话再次 +激活时，会按时间水位补充自上次运行以来的全部增量消息，不受 `historyMessageLimit` 限制。 ### 工具调用 AI 可以自动调用多种工具来完成复杂任务： @@ -201,6 +206,7 @@ promptFile: 'SystemPrompt.md' # 创建Prompt时取最近多少分钟内的消息 historyWindowMin: 10 # 创建Prompt时取最多几条消息 +# 仅限制初次触发时读取的近期历史；模型运行期间的增量消息会全部补充 historyMessageLimit: 20 # 是否打印Prompt便于调试 logPrompt: false @@ -351,7 +357,7 @@ JChatGPT 使用系统提示词来定义 AI 的行为和个性。提示词文件 - sendCompositeMessage - 发送组合消息（适用于长内容或代码） 交互规则： -1. 只有当用户@你或在消息中包含你的名字时才会响应 +1. 通常只有当用户@你或在消息中包含你的名字时才会响应；你通过 endConversation 明确等待的用户回复除外 2. 回复应简洁明了，避免长篇大论 3. 对于复杂内容，使用组合消息功能发送 4. 不主动参与与你无关的对话 @@ -361,7 +367,8 @@ JChatGPT 使用系统提示词来定义 AI 的行为和个性。提示词文件 - 只在必要时使用工具 - 深度思考工具仅用于复杂问题 - 代码执行工具用于验证技术问题 -- **每次对话结束时必须调用 endConversation 工具来结束对话** +- **每次对话结束时必须调用唯一的 endConversation 工具来结束当前运行** +- 通常无参数结束；只有刚明确要求指定用户提供会影响后续处理的反馈时，才使用 waitForFollowUp 短暂等待 - **要发送消息给用户必须使用 sendSingleMessage 或 sendCompositeMessage 工具** diff --git a/src/main/kotlin/JChatGPT.kt b/src/main/kotlin/JChatGPT.kt index d544973..3797959 100644 --- a/src/main/kotlin/JChatGPT.kt +++ b/src/main/kotlin/JChatGPT.kt @@ -131,6 +131,12 @@ object JChatGPT : KotlinPlugin( private suspend fun onMessage(event: MessageEvent) { if (LargeLanguageModels.chat == null) return + + if (ConversationEngine.isExpectedUser(event)) { + if (shouldIgnoreBecauseMuted(event)) return + if (ConversationEngine.resumeObserved(event)) return + } + if (!event.toCommandSender().hasPermission(chatPermission)) { if (event is GroupMessageEvent) { if (!PluginConfig.groupOpHasChatPermission || !event.sender.isOperator()) { @@ -145,18 +151,23 @@ object JChatGPT : KotlinPlugin( event.message[QuoteReply]?.source?.fromId == event.bot.id if (!triggered) return - if (event is GroupMessageEvent && event.group.botMuteRemaining > 0) { - logger.info( - "bot 在群 ${event.group.name}(${event.group.id}) 被禁言，" + - "剩余 ${event.group.botMuteRemaining} 秒，忽略消息" - ) - return - } + if (shouldIgnoreBecauseMuted(event)) return if (PluginConfig.enableFavorabilitySystem && shouldIgnoreForFavorability(event)) return ConversationEngine.start(event) } + private fun shouldIgnoreBecauseMuted(event: MessageEvent): Boolean { + if (event !is GroupMessageEvent) return false + val remainingSeconds = event.group.botMuteRemaining + if (remainingSeconds <= 0) return false + logger.info( + "bot 在群 ${event.group.name}(${event.group.id}) 被禁言，" + + "剩余 $remainingSeconds 秒，忽略消息" + ) + return true + } + private suspend fun shouldIgnoreForFavorability(event: MessageEvent): Boolean { val info = PluginData.userFavorability[event.sender.id] ?: return false if (info.value >= 0) return false diff --git a/src/main/kotlin/config/PluginConfig.kt b/src/main/kotlin/config/PluginConfig.kt index 5277767..297945a 100644 --- a/src/main/kotlin/config/PluginConfig.kt +++ b/src/main/kotlin/config/PluginConfig.kt @@ -244,7 +244,7 @@ object PluginConfig : AutoSavePluginConfig("Config") { @ValueDescription("创建Prompt时取最近多少分钟内的消息") val historyWindowMin: Int by value(10) - @ValueDescription("创建Prompt时取最多几条消息") + @ValueDescription("初次创建Prompt时最多读取几条近期消息；模型运行期间的增量消息不受此限制") val historyMessageLimit: Int by value(20) @ValueDescription("启用对话上下文内存缓存，允许在短时间内保持上下文连续") diff --git a/src/main/kotlin/conversation/ConversationContext.kt b/src/main/kotlin/conversation/ConversationContext.kt index 7ba2b11..68bc892 100644 --- a/src/main/kotlin/conversation/ConversationContext.kt +++ b/src/main/kotlin/conversation/ConversationContext.kt @@ -72,6 +72,8 @@ internal class ReplyIndex { } internal object ConversationContext { + private val chronologicalRecordOrder = compareBy { it.time } + .thenBy { if (it.id == 0L) Long.MAX_VALUE else it.id } private val contextCache = mutableMapOf() private val replyIndexes = mutableMapOf() private val imageIndexes = mutableMapOf() @@ -160,18 +162,18 @@ internal object ConversationContext { .minusMinutes(PluginConfig.historyWindowMin.toLong()) .toEpochSecond() .toInt() - return getAfterHistory(beforeTimestamp, event) + return getAfterHistory(beforeTimestamp, event, PluginConfig.historyMessageLimit) } - fun getAfterHistory(time: Int, event: MessageEvent): String { + fun getAfterHistory(time: Int, event: MessageEvent, limit: Int? = null): String { if (!JChatGPT.includeHistory) return "" val history = try { ChatHistoryStore.query( contact = event.subject, start = time, end = OffsetDateTime.now().toEpochSecond().toInt(), - limit = PluginConfig.historyMessageLimit, - ).sortedBy { it.time }.toMutableList() + limit = limit, + ).sortedWith(chronologicalRecordOrder).toMutableList() } catch (cause: Throwable) { JChatGPT.logger.warning("查询 SQLite 消息历史失败", cause) mutableListOf() @@ -180,6 +182,7 @@ internal object ConversationContext { val messageIds = event.message.ids.joinToString(",") if (history.none { it.ids == messageIds }) { history += ChatMessageRecord.fromSuccess(event.message.source, event.message) + history.sortWith(chronologicalRecordOrder) } val result = StringBuilder() diff --git a/src/main/kotlin/conversation/ConversationEngine.kt b/src/main/kotlin/conversation/ConversationEngine.kt index e720cbe..f1d5dce 100644 --- a/src/main/kotlin/conversation/ConversationEngine.kt +++ b/src/main/kotlin/conversation/ConversationEngine.kt @@ -8,7 +8,6 @@ import com.aallam.openai.api.chat.ToolCall import com.aallam.openai.api.chat.ToolChoice 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 @@ -19,6 +18,7 @@ import kotlinx.coroutines.flow.collect import kotlinx.coroutines.launch import net.mamoe.mirai.event.events.GroupMessageEvent import net.mamoe.mirai.event.events.MessageEvent +import net.mamoe.mirai.message.data.source import top.jie65535.mirai.JChatGPT import top.jie65535.mirai.config.PluginConfig import top.jie65535.mirai.data.TokenUsageStore @@ -51,11 +51,10 @@ import top.jie65535.mirai.tools.WebSearch import top.jie65535.mirai.util.RetryBackoff import java.time.OffsetDateTime import java.time.format.DateTimeFormatter -import kotlin.time.Duration.Companion.milliseconds import kotlin.time.Duration.Companion.seconds internal object ConversationEngine { - private val activeRequests = ConcurrentSet() + private val runtimeState = ConversationRuntimeState() private val dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy年MM月dd E HH:mm:ss") private val thinkRegex = Regex("[\\s\\S]*?") private val tools: List = listOf( @@ -84,16 +83,53 @@ internal object ConversationEngine { ) fun clear() { - activeRequests.clear() + runtimeState.clear() + } + + fun isExpectedUser(event: MessageEvent): Boolean = runtimeState.isExpectedUser( + key = event.toConversationKey(), + userId = event.sender.id, + nowEpochSecond = currentEpochSecond(), + ) + + suspend fun resumeObserved(event: MessageEvent): Boolean { + val started = runtimeState.beginObserved( + key = event.toConversationKey(), + userId = event.sender.id, + nowEpochSecond = currentEpochSecond(), + ) ?: return false + runConversation(event, started.running, started.resumedWait) + return true } suspend fun start(event: MessageEvent) { - val subjectId = event.subject.id - if (!activeRequests.add(subjectId)) { - JChatGPT.logger.warning("The current Contact is busy!") - return - } + when (val result = runtimeState.beginExplicit(event.toConversationKey(), event)) { + is ConversationRuntimeState.BeginResult.Queued -> { + if (result.newlyQueued) { + JChatGPT.logger.info( + "当前会话忙碌，已暂存用户 ${event.senderName}(${event.sender.id}) 的二次触发" + ) + } else { + JChatGPT.logger.info( + "当前会话已有待处理触发，用户 ${event.senderName}(${event.sender.id}) 的消息将通过增量历史合并" + ) + } + } + is ConversationRuntimeState.BeginResult.Started -> { + runConversation(event, result.running, result.resumedWait) + } + } + } + + private suspend fun runConversation( + initialEvent: MessageEvent, + running: ConversationRuntimeState.Running, + resumedWait: FollowUpWaitDirective?, + ) { + val subjectId = initialEvent.subject.id + var currentEvent = initialEvent + var indexesReleased = false try { val cache = ConversationContext.cache(subjectId) val reuseCache = PluginConfig.enableContextCache && cache != null && @@ -112,16 +148,25 @@ internal object ConversationEngine { } else mutableListOf() if (history.isEmpty() || cache == null) { - val prompt = ConversationContext.getSystemPrompt(event) + val prompt = ConversationContext.getSystemPrompt(currentEvent) if (PluginConfig.logPrompt) JChatGPT.logger.info("Prompt: $prompt") history += ChatMessage(ChatRole.System, prompt) - val historyText = ConversationContext.getHistory(event) + val historyText = ConversationContext.getHistory(currentEvent) JChatGPT.logger.info("注入聊天记录：\n$historyText") history += ChatMessage.User(historyText) } else { - val newMessages = ConversationContext.getAfterHistory(cache.lastActivityAt, event) + val newMessages = ConversationContext.getAfterHistory(cache.lastActivityAt, currentEvent) JChatGPT.logger.info("补充聊天记录：\n$newMessages") - history += ChatMessage.User("## 以下是上次对话结束至今的新消息\n\n$newMessages") + history += ChatMessage.User( + if (resumedWait == null) { + "## 以下是上次对话结束至今的新消息\n\n$newMessages" + } else { + buildObservationResumePrompt(resumedWait, newMessages) + } + ) + } + if (resumedWait != null && !reuseCache) { + history += ChatMessage.User(buildObservationResumePrompt(resumedWait, null)) } val endpoints = LargeLanguageModels.orderedChatEndpoints() @@ -134,6 +179,7 @@ internal object ConversationEngine { var consecutiveFailures = 0 do { val endpoint = endpoints[endpointIndex] + val roundEvent = currentEvent var streamingOk = false try { val startedAt = OffsetDateTime.now().toEpochSecond().toInt() @@ -161,7 +207,7 @@ internal object ConversationEngine { if (index >= responseToolCalls.size) { responseToolCalls.lastOrNull()?.let { toolCall -> toolCallTasks += JChatGPT.async { - toolCall.toResultMessage(event) + toolCall.toResultMessage(roundEvent) } } val id = toolCallChunk.id @@ -196,22 +242,48 @@ internal object ConversationEngine { toolCalls = responseToolCalls.ifEmpty { null }, reasoningContent = if (responseToolCalls.isNotEmpty()) reasoningContent?.toString() else null, ) - recordUsage(event, lastTokenUsage, lastCacheUsage) + recordUsage(roundEvent, lastTokenUsage, lastCacheUsage) completedRounds++ if (responseToolCalls.size > toolCallTasks.size) { - val finalToolResult = responseToolCalls.last().toResultMessage(event) + val finalToolResult = responseToolCalls.last().toResultMessage(roundEvent) if (toolCallTasks.isNotEmpty()) history += toolCallTasks.awaitAll() history += finalToolResult - done = responseToolCalls.any { it.function.name == "endConversation" } - } else { - done = true } - if (!done) { + val endCalls = responseToolCalls.filter { + it.function.name == END_CONVERSATION_TOOL_NAME + } + if (endCalls.size > 1) { + JChatGPT.logger.warning("模型在同一轮调用了多次 endConversation，将采用第一次调用的参数") + } + val endCall = endCalls.firstOrNull() + val endArguments = endCall?.let { call -> + runCatching { call.function.argumentsAsJsonOrNull() } + .onFailure { + JChatGPT.logger.warning("无法解析 endConversation 参数，将按普通结束处理", it) + } + .getOrNull() + } + val waitDirective = parseFollowUpWait(endArguments) + if (endArguments?.containsKey(FOLLOW_UP_WAIT_ARGUMENT) == true && waitDirective == null) { + JChatGPT.logger.warning("endConversation.waitForFollowUp 参数无效，将按普通结束处理") + } + + val requestedEnd = responseToolCalls.isEmpty() || endCall != null + val canContinue = completedRounds < maxRounds + if (!requestedEnd && canContinue) { + val pendingEvent = runtimeState.takePending(running) + if (pendingEvent != null) currentEvent = pendingEvent history += ChatMessage.User( - buildContinuationPrompt(maxRounds - completedRounds, startedAt, event) + buildContinuationPrompt( + remainingRounds = maxRounds - completedRounds, + startedAt = startedAt, + event = currentEvent, + pendingTrigger = pendingEvent != null, + ) ) + done = false } else { if (PluginConfig.enableContextCache) { ConversationContext.saveCache( @@ -220,8 +292,48 @@ internal object ConversationEngine { ) JChatGPT.logger.debug("已保存对话上下文到缓存") } - if (event is GroupMessageEvent) { - ProfileAutoMaintenance.recordCompletedConversation(event, startedAt) + + when (val finish = runtimeState.finish( + running = running, + waitDirective = waitDirective.takeIf { requestedEnd }, + nowEpochSecond = currentEpochSecond(), + allowPendingContinuation = canContinue, + onFinished = { + ConversationContext.releaseActiveIndexes(subjectId) + indexesReleased = true + }, + )) { + is ConversationRuntimeState.FinishResult.Continue -> { + currentEvent = finish.event + history += ChatMessage.User( + buildContinuationPrompt( + remainingRounds = maxRounds - completedRounds, + startedAt = startedAt, + event = currentEvent, + pendingTrigger = true, + ) + ) + done = false + } + + is ConversationRuntimeState.FinishResult.Observing -> { + scheduleObservationTimeout(finish.observation) + (currentEvent as? GroupMessageEvent)?.let { + ProfileAutoMaintenance.recordCompletedConversation(it, startedAt) + } + JChatGPT.logger.info( + "会话已结束，等待用户 ${finish.observation.directive.fromUserIds.joinToString()} " + + "在 ${finish.observation.directive.timeoutSeconds} 秒内发言" + ) + done = true + } + + ConversationRuntimeState.FinishResult.Ended -> { + (currentEvent as? GroupMessageEvent)?.let { + ProfileAutoMaintenance.recordCompletedConversation(it, startedAt) + } + done = true + } } } } catch (cause: Exception) { @@ -264,16 +376,55 @@ internal object ConversationEngine { throw cause } catch (cause: Throwable) { JChatGPT.logger.warning(cause) - event.subject.sendMessage("很抱歉，发生异常，请稍后重试") + currentEvent.subject.sendMessage("很抱歉，发生异常，请稍后重试") } finally { - ConversationContext.releaseActiveIndexes(subjectId) - JChatGPT.launch { - delay(500.milliseconds) - activeRequests.remove(subjectId) + if (!indexesReleased) { + runtimeState.abort(running) { + ConversationContext.releaseActiveIndexes(subjectId) + indexesReleased = true + } } } } + private fun scheduleObservationTimeout(observation: ConversationRuntimeState.Observation) { + val job = JChatGPT.launch { + val remainingSeconds = observation.expiresAtEpochSecond - currentEpochSecond() + if (remainingSeconds > 0) delay(remainingSeconds.seconds) + if (runtimeState.expire(observation)) { + JChatGPT.logger.debug( + "等待用户 ${observation.directive.fromUserIds.joinToString()} 的观察窗口已超时" + ) + } + } + runtimeState.attachTimeoutJob(observation, job) + } + + private fun MessageEvent.toConversationKey(): ConversationKey = ConversationKey( + botId = bot.id, + kind = message.source.kind, + subjectId = subject.id, + ) + + private fun currentEpochSecond(): Long = OffsetDateTime.now().toEpochSecond() + + private fun buildObservationResumePrompt( + directive: FollowUpWaitDirective, + newMessages: String?, + ): String = buildString { + appendLine("## 观察状态恢复") + appendLine("你此前结束发言后，选择等待指定用户在当前会话中的下一条消息。") + append("等待用户：").appendLine(directive.fromUserIds.joinToString()) + append("等待条件：").appendLine(directive.condition) + appendLine("被观察状态唤醒不代表必须回复。请判断新消息是否满足等待条件、是否承接当前话题。") + appendLine("如果无关，不要发送任何内容，直接调用 endConversation。") + if (newMessages != null) { + appendLine() + appendLine("## 等待后出现的新消息") + append(newMessages) + } + } + private fun chatCompletions( history: List, endpoint: LargeLanguageModels.ChatEndpoint, @@ -336,13 +487,21 @@ internal object ConversationEngine { return truncated } - private fun buildContinuationPrompt(remainingRounds: Int, startedAt: Int, event: MessageEvent): String = buildString { + private fun buildContinuationPrompt( + remainingRounds: Int, + startedAt: Int, + event: MessageEvent, + pendingTrigger: Boolean, + ): String = buildString { appendLine("## 系统提示") append("本次运行最多还剩").append(remainingRounds).appendLine("轮。") appendLine("如果要多次发言，可以一次性调用多次发言工具。") appendLine("如果没有什么要做的，可以提前结束。") + if (pendingTrigger) appendLine("运行期间收到了新的显式触发，请优先处理水位后的新消息。") appendLine("当前时间：${dateTimeFormatter.format(OffsetDateTime.now())}") - val messages = ConversationContext.getAfterHistory(startedAt, event) + val messages = ConversationContext.getAfterHistory(startedAt, event).ifEmpty { + if (pendingTrigger && !JChatGPT.includeHistory) ConversationContext.getHistory(event) else "" + } if (messages.isNotEmpty()) append("## 以下是上次运行至今的新消息\n\n$messages") } diff --git a/src/main/kotlin/conversation/ConversationKey.kt b/src/main/kotlin/conversation/ConversationKey.kt new file mode 100644 index 0000000..76e72d8 --- /dev/null +++ b/src/main/kotlin/conversation/ConversationKey.kt @@ -0,0 +1,9 @@ +package top.jie65535.mirai.conversation + +import net.mamoe.mirai.message.data.MessageSourceKind + +internal data class ConversationKey( + val botId: Long, + val kind: MessageSourceKind, + val subjectId: Long, +) diff --git a/src/main/kotlin/conversation/ConversationRuntimeState.kt b/src/main/kotlin/conversation/ConversationRuntimeState.kt new file mode 100644 index 0000000..2104a0d --- /dev/null +++ b/src/main/kotlin/conversation/ConversationRuntimeState.kt @@ -0,0 +1,167 @@ +package top.jie65535.mirai.conversation + +import kotlinx.coroutines.Job + +internal class ConversationRuntimeState { + internal class Running internal constructor( + val key: ConversationKey, + ) { + internal var pendingEvent: E? = null + } + + internal class Observation internal constructor( + val key: ConversationKey, + val directive: FollowUpWaitDirective, + val expiresAtEpochSecond: Long, + ) { + internal var timeoutJob: Job? = null + } + + internal sealed interface BeginResult { + data class Started( + val running: Running, + val resumedWait: FollowUpWaitDirective? = null, + ) : BeginResult + + data class Queued(val newlyQueued: Boolean) : BeginResult + } + + internal sealed interface FinishResult { + data class Continue(val event: E) : FinishResult + data class Observing(val observation: Observation) : FinishResult + data object Ended : FinishResult + } + + private sealed interface Slot + private data class RunningSlot(val running: Running) : Slot + private data class ObservationSlot(val observation: Observation) : Slot + + private val lock = Any() + private val slots = mutableMapOf>() + + fun beginExplicit(key: ConversationKey, event: E): BeginResult = synchronized(lock) { + when (val slot = slots[key]) { + is RunningSlot -> { + val newlyQueued = slot.running.pendingEvent == null + if (newlyQueued) slot.running.pendingEvent = event + BeginResult.Queued(newlyQueued) + } + + is ObservationSlot -> { + slot.observation.timeoutJob?.cancel() + startRunning(key) + } + + null -> startRunning(key) + } + } + + fun isExpectedUser(key: ConversationKey, userId: Long, nowEpochSecond: Long): Boolean = synchronized(lock) { + val observation = (slots[key] as? ObservationSlot)?.observation ?: return@synchronized false + if (nowEpochSecond >= observation.expiresAtEpochSecond) { + slots.remove(key) + observation.timeoutJob?.cancel() + return@synchronized false + } + userId in observation.directive.fromUserIds + } + + fun beginObserved( + key: ConversationKey, + userId: Long, + nowEpochSecond: Long, + ): BeginResult.Started? = synchronized(lock) { + val observation = (slots[key] as? ObservationSlot)?.observation ?: return@synchronized null + if (nowEpochSecond >= observation.expiresAtEpochSecond) { + slots.remove(key) + observation.timeoutJob?.cancel() + return@synchronized null + } + if (userId !in observation.directive.fromUserIds) return@synchronized null + + observation.timeoutJob?.cancel() + val running = Running(key) + slots[key] = RunningSlot(running) + BeginResult.Started(running, observation.directive) + } + + fun takePending(running: Running): E? = synchronized(lock) { + val active = (slots[running.key] as? RunningSlot)?.running + if (active !== running) return@synchronized null + running.pendingEvent.also { running.pendingEvent = null } + } + + fun finish( + running: Running, + waitDirective: FollowUpWaitDirective?, + nowEpochSecond: Long, + allowPendingContinuation: Boolean, + onFinished: () -> Unit, + ): FinishResult = synchronized(lock) { + val active = (slots[running.key] as? RunningSlot)?.running + if (active !== running) { + onFinished() + return@synchronized FinishResult.Ended + } + + val pending = running.pendingEvent + running.pendingEvent = null + if (pending != null && allowPendingContinuation) { + return@synchronized FinishResult.Continue(pending) + } + + onFinished() + if (pending == null && waitDirective != null) { + val observation = Observation( + key = running.key, + directive = waitDirective, + expiresAtEpochSecond = nowEpochSecond + waitDirective.timeoutSeconds, + ) + slots[running.key] = ObservationSlot(observation) + FinishResult.Observing(observation) + } else { + slots.remove(running.key) + FinishResult.Ended + } + } + + fun abort(running: Running, onFinished: () -> Unit) = synchronized(lock) { + val active = (slots[running.key] as? RunningSlot)?.running + if (active === running) slots.remove(running.key) + onFinished() + } + + fun attachTimeoutJob(observation: Observation, job: Job) { + val attached = synchronized(lock) { + val active = (slots[observation.key] as? ObservationSlot)?.observation + if (active === observation) { + observation.timeoutJob = job + true + } else { + false + } + } + if (!attached) job.cancel() + } + + fun expire(observation: Observation): Boolean = synchronized(lock) { + val active = (slots[observation.key] as? ObservationSlot)?.observation + if (active !== observation) return@synchronized false + slots.remove(observation.key) + true + } + + fun clear() { + val jobs = synchronized(lock) { + slots.values.mapNotNull { (it as? ObservationSlot)?.observation?.timeoutJob } + .also { slots.clear() } + } + jobs.forEach { it.cancel() } + } + + private fun startRunning(key: ConversationKey): BeginResult.Started { + val running = Running(key) + slots[key] = RunningSlot(running) + return BeginResult.Started(running) + } +} diff --git a/src/main/kotlin/conversation/EndConversationDirective.kt b/src/main/kotlin/conversation/EndConversationDirective.kt new file mode 100644 index 0000000..23accb2 --- /dev/null +++ b/src/main/kotlin/conversation/EndConversationDirective.kt @@ -0,0 +1,46 @@ +package top.jie65535.mirai.conversation + +import kotlinx.serialization.json.JsonArray +import kotlinx.serialization.json.JsonObject +import kotlinx.serialization.json.JsonPrimitive +import kotlinx.serialization.json.contentOrNull +import kotlinx.serialization.json.intOrNull +import kotlinx.serialization.json.longOrNull + +internal const val END_CONVERSATION_TOOL_NAME = "endConversation" +internal const val FOLLOW_UP_WAIT_ARGUMENT = "waitForFollowUp" + +internal data class FollowUpWaitDirective( + val timeoutSeconds: Int, + val fromUserIds: Set, + val condition: String, +) + +internal fun parseFollowUpWait(arguments: JsonObject?): FollowUpWaitDirective? { + val wait = arguments?.get(FOLLOW_UP_WAIT_ARGUMENT) as? JsonObject ?: return null + val timeoutSeconds = (wait["timeoutSeconds"] as? JsonPrimitive)?.intOrNull ?: DEFAULT_WAIT_SECONDS + if (timeoutSeconds !in MIN_WAIT_SECONDS..MAX_WAIT_SECONDS) return null + + val userIdsJson = wait["fromUserIds"] as? JsonArray ?: return null + if (userIdsJson.size !in 1..MAX_WAIT_USERS) return null + val userIds = LinkedHashSet(userIdsJson.size) + for (element in userIdsJson) { + val userId = (element as? JsonPrimitive)?.longOrNull ?: return null + if (userId <= 0 || !userIds.add(userId)) return null + } + + val condition = (wait["condition"] as? JsonPrimitive)?.contentOrNull?.trim().orEmpty() + if (condition.isEmpty() || condition.length > MAX_WAIT_CONDITION_LENGTH) return null + + return FollowUpWaitDirective( + timeoutSeconds = timeoutSeconds, + fromUserIds = userIds, + condition = condition, + ) +} + +private const val DEFAULT_WAIT_SECONDS = 30 +private const val MIN_WAIT_SECONDS = 5 +private const val MAX_WAIT_SECONDS = 120 +private const val MAX_WAIT_USERS = 10 +private const val MAX_WAIT_CONDITION_LENGTH = 200 diff --git a/src/main/kotlin/data/ChatHistoryStore.kt b/src/main/kotlin/data/ChatHistoryStore.kt index 58faacf..2f42863 100644 --- a/src/main/kotlin/data/ChatHistoryStore.kt +++ b/src/main/kotlin/data/ChatHistoryStore.kt @@ -163,7 +163,7 @@ object ChatHistoryStore { contact: Contact, start: Int, end: Int, - limit: Int, + limit: Int? = null, fromId: Long? = null, ): List { check(initialized) { "聊天记录数据库尚未初始化" } @@ -227,7 +227,8 @@ object ChatHistoryStore { ) append(' ') append(conditions.joinToString(" AND ")) - append(" ORDER BY time DESC, id DESC LIMIT ?") + append(" ORDER BY time DESC, id DESC") + if (limit != null) append(" LIMIT ?") } return openReadConnection().use { connection -> @@ -239,7 +240,9 @@ object ChatHistoryStore { else -> error("不支持的查询参数类型 ${value::class}") } } - statement.setInt(parameters.size + 1, limit.coerceAtLeast(1)) + if (limit != null) { + statement.setInt(parameters.size + 1, limit.coerceAtLeast(1)) + } statement.executeQuery().use { results -> buildList { while (results.next()) { diff --git a/src/main/kotlin/tools/SearchChatHistory.kt b/src/main/kotlin/tools/SearchChatHistory.kt index f1ed14c..0dfaf7e 100644 --- a/src/main/kotlin/tools/SearchChatHistory.kt +++ b/src/main/kotlin/tools/SearchChatHistory.kt @@ -98,7 +98,7 @@ class SearchChatHistory : BaseAgent( end = endEpoch, limit = maxRecords, fromId = senderQq, - ).sortedBy { it.time } + ).sortedWith(compareBy { it.time }.thenBy { it.id }) } catch (e: Throwable) { JChatGPT.logger.warning("查询消息历史失败", e) return "查询消息历史失败: ${e.message}" diff --git a/src/main/kotlin/tools/StopLoopAgent.kt b/src/main/kotlin/tools/StopLoopAgent.kt index 8ef6a25..762e915 100644 --- a/src/main/kotlin/tools/StopLoopAgent.kt +++ b/src/main/kotlin/tools/StopLoopAgent.kt @@ -2,11 +2,64 @@ package top.jie65535.mirai.tools import com.aallam.openai.api.chat.Tool import com.aallam.openai.api.core.Parameters +import kotlinx.serialization.json.add +import kotlinx.serialization.json.put +import kotlinx.serialization.json.putJsonArray +import kotlinx.serialization.json.putJsonObject class StopLoopAgent : BaseAgent( tool = Tool.function( name = "endConversation", - description = "结束本轮对话", - parameters = Parameters.Empty + description = """ + 声明当前模型运行已经完成。主循环会先结算本轮其他工具和运行期间的新触发， + 再决定结束、继续处理或短暂等待指定用户的下一条消息。每轮完成时必须且只能调用一次。 + 通常不传waitForFollowUp并直接结束；只有刚刚明确要求他人提供会影响后续处理的反馈时才等待， + 不要仅为了看看是否有人回应、保持活跃或参与普通闲聊而等待。 + """.trimIndent(), + parameters = Parameters.buildJsonObject { + put("type", "object") + putJsonObject("properties") { + putJsonObject("waitForFollowUp") { + put("type", "object") + put( + "description", + "可选。结束当前运行后，非阻塞地等待指定用户在当前会话中的下一条消息；省略表示立即离开。" + ) + putJsonObject("properties") { + putJsonObject("timeoutSeconds") { + put("type", "integer") + put("minimum", 5) + put("maximum", 120) + put("description", "可选，默认30秒。选择满足当前具体等待所需的最短时间。") + } + putJsonObject("fromUserIds") { + put("type", "array") + put("minItems", 1) + put("maxItems", 10) + put("uniqueItems", true) + put("description", "明确等待回复的QQ用户列表。不要猜测或编造用户ID。") + putJsonObject("items") { + put("type", "integer") + } + } + putJsonObject("condition") { + put("type", "string") + put("minLength", 1) + put("maxLength", 200) + put( + "description", + "用一句话描述可由后续消息验证的具体等待条件，不能只写看看有没有人回应。" + ) + } + } + putJsonArray("required") { + add("fromUserIds") + add("condition") + } + put("additionalProperties", false) + } + } + put("additionalProperties", false) + } ) -) \ No newline at end of file +) diff --git a/src/test/kotlin/conversation/ConversationRuntimeStateTest.kt b/src/test/kotlin/conversation/ConversationRuntimeStateTest.kt new file mode 100644 index 0000000..d1f246f --- /dev/null +++ b/src/test/kotlin/conversation/ConversationRuntimeStateTest.kt @@ -0,0 +1,113 @@ +package top.jie65535.mirai.conversation + +import net.mamoe.mirai.message.data.MessageSourceKind +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFalse +import kotlin.test.assertIs +import kotlin.test.assertNull +import kotlin.test.assertTrue + +class ConversationRuntimeStateTest { + @Test + fun keepsFirstPendingTriggerUntilTheRunningLoopConsumesIt() { + val state = ConversationRuntimeState() + val running = assertIs>( + state.beginExplicit(KEY, "first") + ).running + + assertEquals( + ConversationRuntimeState.BeginResult.Queued(newlyQueued = true), + state.beginExplicit(KEY, "second"), + ) + assertEquals( + ConversationRuntimeState.BeginResult.Queued(newlyQueued = false), + state.beginExplicit(KEY, "third"), + ) + assertEquals("second", state.takePending(running)) + assertNull(state.takePending(running)) + } + + @Test + fun pendingTriggerWinsOverWaitAtTerminalSettlement() { + val state = ConversationRuntimeState() + val running = assertIs>( + state.beginExplicit(KEY, "first") + ).running + state.beginExplicit(KEY, "second") + + val finish = state.finish( + running = running, + waitDirective = WAIT, + nowEpochSecond = 100, + allowPendingContinuation = true, + onFinished = { error("continuing must keep the runtime resources active") }, + ) + + assertEquals("second", assertIs>(finish).event) + assertFalse(state.isExpectedUser(KEY, TARGET_USER, nowEpochSecond = 101)) + } + + @Test + fun expectedUserAtomicallyConsumesObservationAndRestoresWaitContext() { + val state = ConversationRuntimeState() + val running = assertIs>( + state.beginExplicit(KEY, "first") + ).running + var released = false + val observation = assertIs( + state.finish( + running = running, + waitDirective = WAIT, + nowEpochSecond = 100, + allowPendingContinuation = true, + onFinished = { released = true }, + ) + ).observation + + assertTrue(released) + assertEquals(130L, observation.expiresAtEpochSecond) + assertFalse(state.isExpectedUser(KEY, 999, nowEpochSecond = 101)) + assertTrue(state.isExpectedUser(KEY, TARGET_USER, nowEpochSecond = 101)) + + val resumed = state.beginObserved(KEY, TARGET_USER, nowEpochSecond = 101) + assertEquals(WAIT, resumed?.resumedWait) + assertFalse(state.isExpectedUser(KEY, TARGET_USER, nowEpochSecond = 101)) + assertNull(state.beginObserved(KEY, TARGET_USER, nowEpochSecond = 101)) + } + + @Test + fun observationExpiresSilentlyAndIsScopedToTheFullConversationKey() { + val state = ConversationRuntimeState() + val running = assertIs>( + state.beginExplicit(KEY, "first") + ).running + state.finish( + running = running, + waitDirective = WAIT, + nowEpochSecond = 100, + allowPendingContinuation = true, + onFinished = {}, + ) + + val otherGroup = KEY.copy(subjectId = KEY.subjectId + 1) + val otherBot = KEY.copy(botId = KEY.botId + 1) + assertFalse(state.isExpectedUser(otherGroup, TARGET_USER, nowEpochSecond = 101)) + assertFalse(state.isExpectedUser(otherBot, TARGET_USER, nowEpochSecond = 101)) + assertFalse(state.isExpectedUser(KEY, TARGET_USER, nowEpochSecond = 130)) + } + + private companion object { + const val TARGET_USER = 123L + val KEY = ConversationKey( + botId = 1, + kind = MessageSourceKind.GROUP, + subjectId = 2, + ) + val WAIT = FollowUpWaitDirective( + timeoutSeconds = 30, + fromUserIds = setOf(TARGET_USER), + condition = "等待对方补充版本信息", + ) + } +} diff --git a/src/test/kotlin/conversation/EndConversationDirectiveTest.kt b/src/test/kotlin/conversation/EndConversationDirectiveTest.kt new file mode 100644 index 0000000..548f17c --- /dev/null +++ b/src/test/kotlin/conversation/EndConversationDirectiveTest.kt @@ -0,0 +1,90 @@ +package top.jie65535.mirai.conversation + +import kotlinx.serialization.json.buildJsonObject +import kotlinx.serialization.json.add +import kotlinx.serialization.json.put +import kotlinx.serialization.json.putJsonArray +import kotlinx.serialization.json.putJsonObject +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertNull + +class EndConversationDirectiveTest { + @Test + fun emptyEndConversationLeavesImmediately() { + assertNull(parseFollowUpWait(buildJsonObject {})) + } + + @Test + fun parsesTargetedOneShotWait() { + val directive = parseFollowUpWait(arguments(timeoutSeconds = 45)) + + assertEquals( + FollowUpWaitDirective( + timeoutSeconds = 45, + fromUserIds = linkedSetOf(123L, 456L), + condition = "等待对方补充版本信息", + ), + directive, + ) + } + + @Test + fun appliesDefaultTimeout() { + assertEquals(30, parseFollowUpWait(arguments(timeoutSeconds = null))?.timeoutSeconds) + } + + @Test + fun rejectsWaitWithoutTargetsOrConcreteCondition() { + assertNull( + parseFollowUpWait( + buildJsonObject { + putJsonObject(FOLLOW_UP_WAIT_ARGUMENT) { + putJsonArray("fromUserIds") {} + put("condition", "等待回复") + } + } + ) + ) + assertNull( + parseFollowUpWait( + buildJsonObject { + putJsonObject(FOLLOW_UP_WAIT_ARGUMENT) { + putJsonArray("fromUserIds") { add(123) } + put("condition", " ") + } + } + ) + ) + } + + @Test + fun rejectsOutOfRangeTimeoutAndDuplicateTargets() { + assertNull(parseFollowUpWait(arguments(timeoutSeconds = 121))) + assertNull( + parseFollowUpWait( + buildJsonObject { + putJsonObject(FOLLOW_UP_WAIT_ARGUMENT) { + put("timeoutSeconds", 30) + putJsonArray("fromUserIds") { + add(123) + add(123) + } + put("condition", "等待对方补充版本信息") + } + } + ) + ) + } + + private fun arguments(timeoutSeconds: Int?) = buildJsonObject { + putJsonObject(FOLLOW_UP_WAIT_ARGUMENT) { + if (timeoutSeconds != null) put("timeoutSeconds", timeoutSeconds) + putJsonArray("fromUserIds") { + add(123) + add(456) + } + put("condition", "等待对方补充版本信息") + } + } +}