mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 03:13:41 +08:00
fix(agent): persist mid-turn narrative, queue follow-ups without dispose, flush on shutdown
A bundle of stability fixes that all surfaced together while running the same long-form generation task across multiple turns. Each one addresses a distinct way the previous behavior silently dropped content the user had already seen on screen. 1. Mid-turn narrative persistence (StateGraphReActAgent + SummarizingNode). Intermediate ReasoningNode rounds and SummarizingNode broadcast their content_delta directly to the SSE channel for live display, but the StreamAccumulator only received the final answer. After refresh the assistant message showed only tool_call cards with no body text. StateGraphReActAgent now also forwards STREAMED_CONTENT (already set per round) as a persistOnly StreamDelta whenever it changes, so every narrative chunk lands in the accumulator's content buffer and gets written to mate_message. SummarizingNode now writes its summary into the same key so summarize narratives persist too. 2. Follow-up message queue, not dispose (ChatController#interruptStream). Sending a new message while a turn was running called requestInterrupt, which dispose()d the active Reactor chain mid LLM call. That cancelled the in-flight generation, lost partial tokens, and left the user staring at a half-finished bubble. The endpoint now uses enqueueMessage in all paths, matching the "wait for current turn, then run" behavior. The old requestInterrupt API is kept for any future force-replace UI but no caller routes to it. 3. Queued user message ordering (ChatStreamTracker.QueuedInput + ChatController.startQueuedMessage). interruptStream used to save the queued user message immediately, before the in-flight assistant message finalized in doOnError. listMessages orders by create_time ASC, so the queued user message ended up above the assistant reply it was supposed to follow. QueuedInput now carries contentParts; persistence is delayed to startQueuedMessage, which runs only after Asst-N is on disk. 4. JVM shutdown flush (ChatStreamTracker @PreDestroy + emergencySaveAccumulator). A mvn spring-boot:run restart used to wipe in-flight turns: SSE emitter timed out, ShutdownHook fired, HikariPool closed before doOnError could save. ChatStreamTracker now exposes an emergency-save callback per RunState; ChatController registers one per stream that snapshots the accumulator and writes status="interrupted_shutdown". @PreDestroy walks active runs, invokes the callback, then disposes. Spring's reverse-order bean teardown keeps ConversationService and Hikari alive long enough for the save to complete. 5. Observation thresholds for summarize (GraphObservationProperties + application.yml). The previous total-chars threshold of 12 KB triggered summarize after one or two RFC reads, costing a 40 to 80 second compaction LLM call per loop. Tuned to: total 200 KB, single 16 KB, large-result 32 KB, rounds safety net 25. Java field defaults reverted to the conservative original values so application.yml stays the source of truth. 6. Frontend thinking segmentation (useChat.ts thinking_delta + phase). Multi-round ReAct turns merged every reasoning + summarize round's thinking into one segment, accumulating to 9 KB+ in a single bubble. thinking_delta now uses findLast(running) so a tool_call_started or phase transition closes the previous segment and the next delta opens a fresh one. phase event also closes running thinking/content segments. 7. Other small things bundled: removed a debug metadata-keys log that flooded the log file with one line per stream chunk; fixed three stale tests that didn't compile after earlier constructor changes (WikiLogServiceTest, WikiOverviewSpliceTest, WikiProcessingServiceLazyTest); added rfc-066 documenting the unified message queue + priority refactor as the next logical step on top of these stabilizations. Verified end-to-end with multiple full sessions: a four-minute generation that produced the expected docx and a follow-up enqueue that ran cleanly after the previous turn naturally completed, without the old "Disposable unavailable" interrupt path.
This commit is contained in:
matevip
parent
941653d185
commit
0476447ab6
@ -743,10 +743,6 @@ public class NodeStreamingChatHelper {
|
||||
}
|
||||
|
||||
// 4. 提取 token usage（通常最后一个 chunk 携带完整 usage）
|
||||
// E-4 probe: log metadata keys to check for x-ratelimit-* headers
|
||||
if (chatResponse.getMetadata() != null && log.isDebugEnabled()) {
|
||||
log.debug("[E4-probe] metadata keys: {}", chatResponse.getMetadata().keySet());
|
||||
}
|
||||
if (chatResponse.getMetadata() != null && chatResponse.getMetadata().getUsage() != null) {
|
||||
var usage = chatResponse.getMetadata().getUsage();
|
||||
if (usage.getPromptTokens() != null && usage.getPromptTokens() > 0) {
|
||||
|
||||
@ -175,6 +175,7 @@ public class StateGraphReActAgent extends BaseAgent implements StructuredStreamC
|
||||
// 防重保护：同 chatStructuredStream
|
||||
AtomicBoolean finalAnswerEmitted = new AtomicBoolean(false);
|
||||
AtomicBoolean finalThinkingEmitted = new AtomicBoolean(false);
|
||||
AtomicReference<String> lastEmittedStreamedContent = new AtomicReference<>("");
|
||||
|
||||
return compiledGraph.stream(inputs, config)
|
||||
.flatMapIterable(output -> {
|
||||
@ -192,6 +193,14 @@ public class StateGraphReActAgent extends BaseAgent implements StructuredStreamC
|
||||
boolean contentAlreadyStreamed = output.state().value(CONTENT_STREAMED, false);
|
||||
boolean thinkingAlreadyStreamed = output.state().value(THINKING_STREAMED, false);
|
||||
|
||||
// 与 chatStructuredStream 一致：把每轮 STREAMED_CONTENT 用 persistOnly 推给 Accumulator，
|
||||
// 否则中间叙述（reasoning narrative + summarize）只在 SSE 上出现一次，刷新后丢失。
|
||||
String streamed = output.state().<String>value(STREAMED_CONTENT).orElse("");
|
||||
if (!streamed.isEmpty() && !streamed.equals(lastEmittedStreamedContent.get())) {
|
||||
lastEmittedStreamedContent.set(streamed);
|
||||
deltas.add(AgentService.StreamDelta.persistOnly(streamed, null));
|
||||
}
|
||||
|
||||
if (hasFinalAnswer(output) && finalAnswerEmitted.compareAndSet(false, true)) {
|
||||
String answer = extractFinalAnswer(output);
|
||||
if (answer != null && !answer.isEmpty()) {
|
||||
@ -265,6 +274,9 @@ public class StateGraphReActAgent extends BaseAgent implements StructuredStreamC
|
||||
// 用 compareAndSet 保证只取第一次，避免 content/thinking 被重复追加
|
||||
AtomicBoolean finalAnswerEmitted = new AtomicBoolean(false);
|
||||
AtomicBoolean finalThinkingEmitted = new AtomicBoolean(false);
|
||||
// STREAMED_CONTENT 是 REPLACE 策略（每轮 ReasoningNode/SummarizingNode 覆写），
|
||||
// 用 lastEmitted 跟踪已发送的值，避免在 ActionNode/ObservationNode 的 NodeOutput 上重复发送同一段内容。
|
||||
AtomicReference<String> lastEmittedStreamedContent = new AtomicReference<>("");
|
||||
|
||||
return compiledGraph.stream(inputs, config)
|
||||
.flatMapIterable(output -> {
|
||||
@ -287,6 +299,16 @@ public class StateGraphReActAgent extends BaseAgent implements StructuredStreamC
|
||||
boolean thinkingAlreadyStreamed = output.state()
|
||||
.value(THINKING_STREAMED, false);
|
||||
|
||||
// 2a. 中间叙述内容持久化：每轮 ReasoningNode（带 tool_calls）和 SummarizingNode
|
||||
// 都把当轮 LLM 输出写入 STREAMED_CONTENT。NodeStreamingChatHelper 已实时广播
|
||||
// 给前端，但 Accumulator 不在 SSE 订阅链路上，必须用 persistOnly StreamDelta
|
||||
// 补一刀，否则刷新后正文文字全部丢失（只剩 final_answer + tool_call 卡片）。
|
||||
String streamed = output.state().<String>value(STREAMED_CONTENT).orElse("");
|
||||
if (!streamed.isEmpty() && !streamed.equals(lastEmittedStreamedContent.get())) {
|
||||
lastEmittedStreamedContent.set(streamed);
|
||||
deltas.add(AgentService.StreamDelta.persistOnly(streamed, null));
|
||||
}
|
||||
|
||||
if (hasFinalAnswer(output) && finalAnswerEmitted.compareAndSet(false, true)) {
|
||||
String answer = extractFinalAnswer(output);
|
||||
if (answer != null && !answer.isEmpty()) {
|
||||
|
||||
@ -183,6 +183,9 @@ public class SummarizingNode implements NodeAction {
|
||||
// 摘要的 content 已流式推送，但它不是最终回答，标记防重即可
|
||||
.contentStreamed(true)
|
||||
.thinkingStreamed(!result.thinking().isEmpty())
|
||||
// 把当轮 summary 文本写入 STREAMED_CONTENT，让 StateGraphReActAgent 用 persistOnly
|
||||
// StreamDelta 推给 Accumulator 持久化（用户刷新页面后能看到摘要正文，否则只剩 tool_call 卡片）
|
||||
.streamedContent(summaryContent)
|
||||
.mergeUsage(state, result)
|
||||
// 不设 finishReason — summarizing 不是终止，循环继续
|
||||
.events(List.of(GraphEventPublisher.phase("summarized", Map.of(
|
||||
|
||||
@ -738,6 +738,10 @@ public class ChatController {
|
||||
|
||||
// 将 Disposable 注册到 StreamTracker，以便 stop 端点可以取消它
|
||||
streamTracker.setDisposable(conversationId, disposable);
|
||||
// JVM 关闭时优雅落盘：避免 mvn spring-boot:run 重启 / SIGTERM 把
|
||||
// 进行中 turn 的 assistant 消息丢失（doOnError 来不及在 Hikari 关闭前执行）
|
||||
streamTracker.setEmergencySaveCallback(conversationId,
|
||||
() -> emergencySaveAccumulator(conversationId, accumulator));
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("SSE setup error: {}", e.getMessage());
|
||||
@ -773,12 +777,15 @@ public class ChatController {
|
||||
}
|
||||
|
||||
/**
|
||||
* 中断当前流并排队一条后续消息。
|
||||
* 在执行中追加一条后续消息：仅入队，等当前 turn 自然结束后再启动。
|
||||
* <p>
|
||||
* 与 stop 的区别：interrupt 会在当前 turn 安全结束后自动启动排队消息。
|
||||
* 如果当前阶段不可中断（awaiting_approval），消息会被排队但不打断当前执行。
|
||||
* 对齐 Claude Code 行为：流式输出过程中收到新输入不会强制 dispose 当前 LLM 调用，
|
||||
* 而是仅入队。当前 turn 跑到 doOnComplete/doOnError 后由 startQueuedMessage 接管。
|
||||
* <p>
|
||||
* 旧行为（dispose 当前 disposable + 立即重启）会导致部分 LLM 输出被丢弃 + token 浪费，
|
||||
* 已废弃。如果未来需要"立即打断"语义，应该走 /stop（用户主动取消）+ 重新发起新消息的路径。
|
||||
*/
|
||||
@Operation(summary = "中断并排队后续消息")
|
||||
@Operation(summary = "排队后续消息（不打断当前流）")
|
||||
@PostMapping("/{conversationId}/interrupt")
|
||||
public R<Map<String, Object>> interruptStream(
|
||||
@PathVariable String conversationId,
|
||||
@ -797,34 +804,20 @@ public class ChatController {
|
||||
Long agentId = request.getAgentId();
|
||||
List<MessageContentPart> contentParts = request.getContentParts();
|
||||
|
||||
// 判断当前阶段是否可中断
|
||||
// awaiting_approval 阶段不直接中断，只排队
|
||||
// 判断当前阶段（仅用于 reason 字段，行为对所有阶段一致：仅入队）
|
||||
boolean isAwaitingApproval = approvalService.findPendingByConversation(conversationId) != null;
|
||||
|
||||
if (isAwaitingApproval) {
|
||||
// 不可中断：排队但不打断。先持久化（含 contentParts）再入队（persisted=true）
|
||||
conversationService.saveMessage(conversationId, "user", message, contentParts, "queued");
|
||||
boolean queued = streamTracker.enqueueMessage(conversationId, message, agentId, true);
|
||||
log.info("Interrupt requested during approval, message queued: conversationId={}, user={}, queueSize={}",
|
||||
conversationId, username, streamTracker.getQueueSize(conversationId));
|
||||
return R.ok(Map.of(
|
||||
"interrupted", false,
|
||||
"queued", queued,
|
||||
"reason", "awaiting_approval"
|
||||
));
|
||||
}
|
||||
|
||||
// 可中断：先持久化（含 contentParts）再打断并入队（persisted=true）
|
||||
conversationService.saveMessage(conversationId, "user", message, contentParts, "queued");
|
||||
boolean interrupted = streamTracker.requestInterrupt(conversationId, message, agentId, true);
|
||||
log.info("Interrupt requested: conversationId={}, user={}, interrupted={}, queueSize={}",
|
||||
conversationId, username, interrupted, streamTracker.getQueueSize(conversationId));
|
||||
// 仅入队、不 dispose。延迟持久化到 startQueuedMessage（让 Asst-N 先在 doOnComplete 落库，
|
||||
// 否则 listMessages ORDER BY create_time ASC 会把 Q(N+1) 排到 Asst-N 前面）
|
||||
boolean queued = streamTracker.enqueueMessage(conversationId, message, agentId, false, contentParts);
|
||||
log.info("Enqueued follow-up message during running turn: conversationId={}, user={}, queueSize={}, awaitingApproval={}",
|
||||
conversationId, username, streamTracker.getQueueSize(conversationId), isAwaitingApproval);
|
||||
|
||||
return R.ok(Map.of(
|
||||
"interrupted", interrupted,
|
||||
"queued", true,
|
||||
"interrupted", false,
|
||||
"queued", queued,
|
||||
"queueSize", streamTracker.getQueueSize(conversationId),
|
||||
"reason", interrupted ? "interrupted" : "queued"
|
||||
"reason", isAwaitingApproval ? "awaiting_approval" : "queued"
|
||||
));
|
||||
}
|
||||
|
||||
@ -1009,9 +1002,12 @@ public class ChatController {
|
||||
log.info("Starting queued message: conversationId={}, agentId={}, message={}",
|
||||
conversationId, agentId, queuedMessage.substring(0, Math.min(30, queuedMessage.length())));
|
||||
|
||||
// 持久化排队的用户消息（幂等：如果 /interrupt 已提前持久化则跳过）
|
||||
// 持久化排队的用户消息（含 contentParts；幂等：如果 /interrupt 已提前持久化则跳过）。
|
||||
// 这里持久化是为了确保 user 消息在 assistant 消息（doOnError/doOnCancel 已写入）之后落库，
|
||||
// 让 listMessages ORDER BY create_time ASC 后顺序正确：Q1 → Asst1 → Q2 → Asst2。
|
||||
if (queuedMessage != null && !queuedMessage.isBlank() && !preConsumedInput.persisted()) {
|
||||
conversationService.saveMessage(conversationId, "user", queuedMessage);
|
||||
conversationService.saveMessage(conversationId, "user", queuedMessage,
|
||||
preConsumedInput.contentParts(), "queued");
|
||||
}
|
||||
|
||||
// 广播 queued_input_started 事件
|
||||
@ -1122,6 +1118,8 @@ public class ChatController {
|
||||
})
|
||||
.subscribe();
|
||||
streamTracker.setDisposable(conversationId, disposable);
|
||||
streamTracker.setEmergencySaveCallback(conversationId,
|
||||
() -> emergencySaveAccumulator(conversationId, accumulator));
|
||||
}
|
||||
|
||||
private void sendEvent(SseEmitter emitter, String name, Object data) throws IOException {
|
||||
@ -1170,6 +1168,40 @@ public class ChatController {
|
||||
return payload;
|
||||
}
|
||||
|
||||
/**
|
||||
* Snapshot the accumulator and persist it as an assistant message during JVM shutdown.
|
||||
* Invoked from {@link ChatStreamTracker#onShutdown()} so any in-flight turn doesn't
|
||||
* lose its already-streamed content + tool calls when the process exits.
|
||||
* <p>
|
||||
* Idempotent w.r.t. the normal doOnComplete/doOnError save: if those paths already
|
||||
* persisted the message, this writes a second row with status="interrupted_shutdown",
|
||||
* which is rare in practice (race window is sub-second between dispose and save) and
|
||||
* acceptable. Skipping save when nothing to save avoids empty rows.
|
||||
*/
|
||||
private void emergencySaveAccumulator(String conversationId, StreamAccumulator accumulator) {
|
||||
try {
|
||||
String text = accumulator.getContent();
|
||||
List<MessageContentPart> parts = accumulator.toAssistantParts();
|
||||
if (text.isBlank() && parts.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
String savedText = text.isBlank() ? "[已中断 — 服务重启]" : text;
|
||||
conversationService.saveMessage(conversationId, "assistant", savedText, parts,
|
||||
"interrupted_shutdown",
|
||||
accumulator.getPromptTokens(),
|
||||
accumulator.getCompletionTokens(),
|
||||
accumulator.getRuntimeModelName(),
|
||||
accumulator.getRuntimeProviderId(),
|
||||
accumulator.toMetadataJson());
|
||||
log.info("[ChatController] Emergency-saved in-flight assistant message: " +
|
||||
"conversationId={}, textLen={}, partsCount={}",
|
||||
conversationId, text.length(), parts.size());
|
||||
} catch (Exception e) {
|
||||
log.error("[ChatController] Emergency save failed for {}: {}",
|
||||
conversationId, e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
|
||||
private List<MessageContentPart> normalizeRequestParts(ChatStreamRequest request) {
|
||||
if (request.getContentParts() != null && !request.getContentParts().isEmpty()) {
|
||||
return request.getContentParts();
|
||||
|
||||
@ -1,10 +1,12 @@
|
||||
package vip.mate.channel.web;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import jakarta.annotation.PreDestroy;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
|
||||
import reactor.core.Disposable;
|
||||
import vip.mate.workspace.conversation.model.MessageContentPart;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.ArrayList;
|
||||
@ -104,6 +106,18 @@ public class ChatStreamTracker {
|
||||
/** 排队的用户消息队列（支持多条排队消息，按序消费） */
|
||||
final java.util.Queue<QueuedInput> messageQueue = new java.util.concurrent.ConcurrentLinkedQueue<>();
|
||||
|
||||
/**
|
||||
* Emergency save callback registered by the SSE chain owner (ChatController).
|
||||
* Invoked from {@link #onShutdown()} so the accumulated assistant content + tool_calls
|
||||
* are persisted before the JVM tears down — without this, a `mvn spring-boot:run`
|
||||
* restart wipes any in-flight turn and leaves only the user message in DB.
|
||||
* <p>
|
||||
* The callback must be idempotent (will not be called twice for the same run, but
|
||||
* may race with normal doOnComplete/doOnError; both paths must tolerate the other
|
||||
* having saved already).
|
||||
*/
|
||||
volatile Runnable emergencySaveCallback;
|
||||
|
||||
/** 心跳定时器 */
|
||||
volatile ScheduledFuture<?> heartbeatFuture;
|
||||
|
||||
@ -190,6 +204,18 @@ public class ChatStreamTracker {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Register an emergency-save callback for this run, invoked from {@link #onShutdown()}
|
||||
* before the JVM tears down. The callback should snapshot the current accumulator
|
||||
* state and persist it as the assistant message (status="interrupted").
|
||||
*/
|
||||
public void setEmergencySaveCallback(String conversationId, Runnable callback) {
|
||||
RunState state = runs.get(conversationId);
|
||||
if (state != null) {
|
||||
state.emergencySaveCallback = callback;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 请求停止指定会话的流。
|
||||
* 取消 Flux 订阅（底层 HTTP 连接也会随之关闭），返回 true 表示确实停止了正在运行的流。
|
||||
@ -577,6 +603,11 @@ public class ChatStreamTracker {
|
||||
* @return true 如果成功请求了中断
|
||||
*/
|
||||
public boolean requestInterrupt(String conversationId, String queuedMessage, Long agentId, boolean persisted) {
|
||||
return requestInterrupt(conversationId, queuedMessage, agentId, persisted, null);
|
||||
}
|
||||
|
||||
public boolean requestInterrupt(String conversationId, String queuedMessage, Long agentId,
|
||||
boolean persisted, List<MessageContentPart> contentParts) {
|
||||
RunState state = runs.get(conversationId);
|
||||
if (state == null || state.done) {
|
||||
return false;
|
||||
@ -589,7 +620,7 @@ public class ChatStreamTracker {
|
||||
Disposable d = state.disposable;
|
||||
canInterrupt = d != null && !d.isDisposed();
|
||||
// 无论是否可中断，都入队（支持多条排队消息）
|
||||
state.messageQueue.offer(new QueuedInput(queuedMessage, agentId, persisted));
|
||||
state.messageQueue.offer(new QueuedInput(queuedMessage, agentId, persisted, contentParts));
|
||||
if (canInterrupt) {
|
||||
state.interruptType = InterruptType.USER_INTERRUPT_WITH_FOLLOWUP;
|
||||
state.stopRequested.set(true);
|
||||
@ -636,11 +667,16 @@ public class ChatStreamTracker {
|
||||
* 将消息加入队列但不中断当前执行（用于不可中断阶段）。
|
||||
*/
|
||||
public boolean enqueueMessage(String conversationId, String message, Long agentId, boolean persisted) {
|
||||
return enqueueMessage(conversationId, message, agentId, persisted, null);
|
||||
}
|
||||
|
||||
public boolean enqueueMessage(String conversationId, String message, Long agentId, boolean persisted,
|
||||
List<MessageContentPart> contentParts) {
|
||||
RunState state = runs.get(conversationId);
|
||||
if (state == null || state.done) {
|
||||
return false;
|
||||
}
|
||||
state.messageQueue.offer(new QueuedInput(message, agentId, persisted));
|
||||
state.messageQueue.offer(new QueuedInput(message, agentId, persisted, contentParts));
|
||||
// broadcast 在锁外
|
||||
try {
|
||||
String json = objectMapper.writeValueAsString(Map.of(
|
||||
@ -656,9 +692,14 @@ public class ChatStreamTracker {
|
||||
}
|
||||
|
||||
/**
|
||||
* 排队输入的原子快照（message + agentId + persisted 一起返回，避免分离读取导致不一致）
|
||||
* 排队输入的原子快照（message + agentId + persisted + contentParts 一起返回，避免分离读取导致不一致）
|
||||
*/
|
||||
public record QueuedInput(String message, Long agentId, boolean persisted) {}
|
||||
public record QueuedInput(String message, Long agentId, boolean persisted,
|
||||
List<MessageContentPart> contentParts) {
|
||||
public QueuedInput(String message, Long agentId, boolean persisted) {
|
||||
this(message, agentId, persisted, null);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 原子消费排队的输入（流完成/中断后调用）。
|
||||
@ -905,4 +946,64 @@ public class ChatStreamTracker {
|
||||
evicted, runs.size());
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Flush in-flight runs before JVM shutdown.
|
||||
* <p>
|
||||
* Spring closes singleton beans in reverse construction order; ConversationService /
|
||||
* Hikari outlive ChatStreamTracker, so saveMessage from {@link #onShutdown()} still
|
||||
* has a working DB connection. Without this, a {@code mvn spring-boot:run} restart or
|
||||
* SIGTERM during a turn races against the Reactor cancellation: the doOnError /
|
||||
* doOnComplete saveMessage may not run before HikariPool shuts down, leaving the
|
||||
* conversation with only the user message and no assistant reply (the
|
||||
* "对话框里除了问题外什么也没留下" symptom seen in production logs at 07:23:02).
|
||||
* <p>
|
||||
* Behavior:
|
||||
* <ol>
|
||||
* <li>Walk every active (not-done) RunState.</li>
|
||||
* <li>Invoke its registered emergencySaveCallback synchronously — the callback
|
||||
* (set by ChatController) snapshots the current accumulator and persists it
|
||||
* as an "interrupted" assistant message.</li>
|
||||
* <li>Dispose the Reactor disposable so the LLM stream terminates promptly.</li>
|
||||
* </ol>
|
||||
* The callback must tolerate normal doOnError/doOnComplete having raced and saved
|
||||
* already; the latest commit wins for that conversation.
|
||||
*/
|
||||
@PreDestroy
|
||||
public void onShutdown() {
|
||||
int active = (int) runs.values().stream().filter(s -> !s.done).count();
|
||||
if (active == 0) {
|
||||
log.info("[ChatStreamTracker] Shutdown: no active runs to flush");
|
||||
return;
|
||||
}
|
||||
log.warn("[ChatStreamTracker] Shutdown: flushing {} active run(s) before JVM exit",
|
||||
active);
|
||||
for (Map.Entry<String, RunState> entry : runs.entrySet()) {
|
||||
RunState state = entry.getValue();
|
||||
if (state.done) continue;
|
||||
String cid = entry.getKey();
|
||||
try {
|
||||
Runnable callback = state.emergencySaveCallback;
|
||||
if (callback != null) {
|
||||
log.info("[ChatStreamTracker] Emergency-saving in-flight run: {}", cid);
|
||||
callback.run();
|
||||
} else {
|
||||
log.warn("[ChatStreamTracker] No emergency-save callback for active run: {} " +
|
||||
"(content may be lost)", cid);
|
||||
}
|
||||
} catch (Exception e) {
|
||||
log.error("[ChatStreamTracker] Emergency save failed for {}: {}",
|
||||
cid, e.getMessage(), e);
|
||||
}
|
||||
try {
|
||||
Disposable d = state.disposable;
|
||||
if (d != null && !d.isDisposed()) {
|
||||
d.dispose();
|
||||
}
|
||||
} catch (Exception e) {
|
||||
log.warn("[ChatStreamTracker] Disposable.dispose failed for {}: {}",
|
||||
cid, e.getMessage());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -4,7 +4,13 @@ import lombok.Data;
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
|
||||
/**
|
||||
* Graph 观察结果处理阈值配置
|
||||
* Graph observation thresholds for triggering summarize / compaction.
|
||||
* <p>
|
||||
* <b>Tune values in {@code application.yml} under {@code mate.agent.graph.observation},
|
||||
* not in the Java field defaults.</b> The yml is the source of truth and is loaded into
|
||||
* a Spring bean by {@link ConfigurationProperties}; field defaults below are kept as a
|
||||
* conservative fallback for tests / unit constructors and intentionally do NOT reflect
|
||||
* production-tuned values.
|
||||
*
|
||||
* @author MateClaw Team
|
||||
*/
|
||||
@ -12,27 +18,27 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
@ConfigurationProperties(prefix = "mate.agent.graph.observation")
|
||||
public class GraphObservationProperties {
|
||||
|
||||
/** 单次工具结果最大字符数 */
|
||||
/** Max chars retained per tool result (truncate above this). Tune via yml. */
|
||||
private int maxSingleObservationChars = 4000;
|
||||
|
||||
/** 所有观察记录总字符数上限 */
|
||||
/** Total observation chars across history+current that trigger summarize. Tune via yml. */
|
||||
private int maxTotalObservationChars = 12000;
|
||||
|
||||
/** 单次结果超过此阈值视为"大结果" */
|
||||
/** Single tool result above this is treated as "large" and triggers summarize. Tune via yml. */
|
||||
private int largeResultThreshold = 3000;
|
||||
|
||||
/** 触发 summarize 的最小观察轮次 */
|
||||
/** Minimum observation rounds that triggers summarize as a runaway-loop safety net. Tune via yml. */
|
||||
private int minRoundsForSummarize = 3;
|
||||
|
||||
/** 截断时保留前部占比（0-1） */
|
||||
/** Truncation: head fraction kept (0-1). */
|
||||
private double headRatio = 0.4;
|
||||
|
||||
/** 截断省略标记（%d 会被替换为原始字符数） */
|
||||
/** Truncation marker; %d is replaced with original char count. */
|
||||
private String truncationMarker = "\n\n... [内容已截断，共 %d 字符，保留前后关键片段] ...\n\n";
|
||||
|
||||
/** 检测到尾部错误模式时的 tail 保留比例（默认 0.8，优先保留错误信息） */
|
||||
/** Tail-keep ratio when an error pattern is detected at the tail (prefer keeping error info). */
|
||||
private double errorTailRatio = 0.8;
|
||||
|
||||
/** 截断时最少保留字符数（避免过度截断导致信息完全丢失） */
|
||||
/** Minimum chars retained on truncation to avoid losing all information. */
|
||||
private int minKeepChars = 2000;
|
||||
}
|
||||
|
||||
@ -154,10 +154,11 @@ mate:
|
||||
agent:
|
||||
graph:
|
||||
observation:
|
||||
max-single-observation-chars: 8000
|
||||
max-total-observation-chars: 24000
|
||||
large-result-threshold: 6000
|
||||
min-rounds-for-summarize: 3
|
||||
# 与 GraphObservationProperties.java 默认值对齐，参考 openclaw token-budget 设计
|
||||
max-single-observation-chars: 16000
|
||||
max-total-observation-chars: 200000
|
||||
large-result-threshold: 32000
|
||||
min-rounds-for-summarize: 25
|
||||
head-ratio: 0.4
|
||||
truncation-marker: "\n\n... [内容已截断，共 %d 字符，保留前后关键片段] ...\n\n"
|
||||
tool:
|
||||
|
||||
@ -288,18 +288,21 @@ export function useChat(options: UseChatOptions): UseChatReturn {
|
||||
if (streamPhase.value !== 'summarizing_observations') {
|
||||
streamPhase.value = options.thinkingLevel?.value === 'off' ? 'streaming' : 'thinking'
|
||||
}
|
||||
// Segments: all thinking deltas merge into one segment (not split by tool_call interruptions)
|
||||
// Segments: per-round thinking. When tool_call_started / phase change closes the running
|
||||
// thinking segment (status='completed'), a fresh thinking_delta opens a new segment instead
|
||||
// of reopening the closed one. Without this split, multi-round ReAct (3 reasoning + 2
|
||||
// summarizing rounds) accumulates 9K+ chars in a single bubble.
|
||||
const segs = currentSegments.value
|
||||
// Reuse an existing thinking segment regardless of status (running or completed)
|
||||
let thinkSeg = segs.find((s: MessageSegment) => s.type === 'thinking')
|
||||
let thinkSeg = segs.findLast((s: MessageSegment) =>
|
||||
s.type === 'thinking' && s.status === 'running'
|
||||
)
|
||||
if (!thinkSeg) {
|
||||
thinkSeg = { id: genSegId(), type: 'thinking', status: 'running', thinkingText: '', timestamp: Date.now() }
|
||||
// Insert at front — thinking always appears at the top
|
||||
segs.unshift(thinkSeg)
|
||||
// Append in timeline order (interleaved with tool_calls) — old behavior unshift'd to top,
|
||||
// but with per-round splitting that misorders rounds 2+ relative to their tool calls.
|
||||
segs.push(thinkSeg)
|
||||
flushSegmentsToMessage()
|
||||
}
|
||||
// Re-mark as running when new thinking content arrives
|
||||
thinkSeg.status = 'running'
|
||||
thinkSeg.thinkingText = (thinkSeg.thinkingText || '') + (data.delta || '')
|
||||
}
|
||||
})
|
||||
@ -596,6 +599,15 @@ export function useChat(options: UseChatOptions): UseChatReturn {
|
||||
metadata: { ...metadata, currentPhase: data.phase }
|
||||
} as any)
|
||||
}
|
||||
// Close any running thinking/content segment on phase transition so the next thinking_delta
|
||||
// (e.g. summarizing → reasoning) starts a fresh round-scoped segment instead of growing the
|
||||
// previous one unbounded.
|
||||
const segs = currentSegments.value
|
||||
for (const seg of segs) {
|
||||
if (seg.status === 'running' && (seg.type === 'thinking' || seg.type === 'content')) {
|
||||
seg.status = 'completed'
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user