opensource/mateclaw
2
0
Fork 0
mirror of https://gitee.com/mateos/mateclaw.git synced 2026-09-15 03:55:09 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix(webchat): persist assistant reply and publish memory event on stream end

Browse Source
This commit is contained in:
matevip 2026-04-23 08:09:48 +08:00
parent 869e0c47e6
commit 9632edb008
1 changed files with 26 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
mateclaw-server/src/main/java/vip/mate/channel/webchat/WebChatController.java
View File

@ -14,6 +14,7 @@ import vip.mate.channel.model.ChannelEntity;
import vip.mate.channel.service.ChannelService; import vip.mate.channel.service.ChannelService;
import vip.mate.channel.web.ChatStreamTracker; import vip.mate.channel.web.ChatStreamTracker;
import vip.mate.common.result.R; import vip.mate.common.result.R;
import vip.mate.memory.event.ConversationCompletionPublisher;
import vip.mate.workspace.conversation.ConversationService; import vip.mate.workspace.conversation.ConversationService;
import vip.mate.workspace.conversation.model.MessageContentPart; import vip.mate.workspace.conversation.model.MessageContentPart;
@ -47,6 +48,7 @@ public class WebChatController {
private final ConversationService conversationService; private final ConversationService conversationService;
private final ChatStreamTracker streamTracker; private final ChatStreamTracker streamTracker;
private final ObjectMapper objectMapper; private final ObjectMapper objectMapper;
private final ConversationCompletionPublisher completionPublisher;
private final ExecutorService sseExecutor = Executors.newCachedThreadPool(); private final ExecutorService sseExecutor = Executors.newCachedThreadPool();
@ -110,19 +112,39 @@ public class WebChatController {
streamTracker.register(conversationId); streamTracker.register(conversationId);
streamTracker.attach(conversationId, emitter); streamTracker.attach(conversationId, emitter);
// 调用 Agent 流式对话 // Accumulate the assistant reply so it can be persisted on stream completion.
// Pattern mirrors ChatController: always accumulate, only broadcast when the
// delta is not a persistence-only echo of content already streamed by inner nodes.
StringBuilder assistantReply = new StringBuilder();
agentService.chatStructuredStream(agentId, message, conversationId, visitorId) agentService.chatStructuredStream(agentId, message, conversationId, visitorId)
.doOnNext(delta -> { .doOnNext(delta -> {
if (delta.content() != null && !delta.content().isEmpty()) { if (delta.content() != null && !delta.content().isEmpty()) {
streamTracker.broadcast(conversationId, "content_delta", assistantReply.append(delta.content());
"{\"text\":" + escapeJson(delta.content()) + "}"); if (!delta.persistenceOnly()) {
streamTracker.broadcast(conversationId, "content_delta",
"{\"text\":" + escapeJson(delta.content()) + "}");
}
} }
if (delta.thinking() != null && !delta.thinking().isEmpty()) { if (delta.thinking() != null && !delta.thinking().isEmpty()
&& !delta.persistenceOnly()) {
streamTracker.broadcast(conversationId, "thinking_delta", streamTracker.broadcast(conversationId, "thinking_delta",
"{\"text\":" + escapeJson(delta.thinking()) + "}"); "{\"text\":" + escapeJson(delta.thinking()) + "}");
} }
}) })
.doOnComplete(() -> { .doOnComplete(() -> {
String reply = assistantReply.toString();
try {
if (!reply.isBlank()) {
conversationService.saveMessage(
conversationId, "assistant", reply, List.of());
}
completionPublisher.publish(
agentId, conversationId, message, reply, "webchat");
} catch (Exception persistErr) {
log.warn("[WebChat] Failed to persist assistant reply / publish event: {}",
persistErr.getMessage());
}
streamTracker.broadcast(conversationId, "done", "{\"status\":\"completed\"}"); streamTracker.broadcast(conversationId, "done", "{\"status\":\"completed\"}");
streamTracker.complete(conversationId); streamTracker.complete(conversationId);
}) })