mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 03:13:41 +08:00
feat(channel): relay per-stage narration as standalone messages on the sync IM path
This commit is contained in:
matevip
parent
8749c915cd
commit
8a55bdd367
@ -809,6 +809,14 @@ public class ChannelMessageRouter {
|
||||
// for any Web SSE viewer of the same conversationId.
|
||||
StringBuilder replyAccumulator = new StringBuilder();
|
||||
final String channelType = adapter.getChannelType();
|
||||
// Channel-level toggle for relaying per-stage narration as
|
||||
// standalone messages mid-run. Shares the key the streaming
|
||||
// progress path uses so operators have one knob per channel.
|
||||
// Disabled → narration is dropped from the IM channel (it is
|
||||
// never part of the final reply either way; web observers
|
||||
// still see it via the live broadcast).
|
||||
final boolean relayNarration = channelConfigBoolean(
|
||||
channelEntity, "stream_progress", true);
|
||||
// Token usage + model attribution: capture _usage_final event emitted at stream end
|
||||
final int[] usage = {0, 0, 0, 0, 0}; // [prompt, completion, cacheRead, cacheWrite, reasoning]
|
||||
final String[] modelInfo = {null, null}; // [runtimeModel, runtimeProvider]
|
||||
@ -829,6 +837,26 @@ public class ChannelMessageRouter {
|
||||
if (provider != null) modelInfo[1] = provider.toString();
|
||||
}
|
||||
mirrorPlanEventToTracker(conversationId, delta, channelType);
|
||||
} else if (delta.segmentOnly()) {
|
||||
// Per-stage narration ("Let me look that up…"), emitted as
|
||||
// one complete delta per agent loop iteration. Relay it
|
||||
// immediately as its own outgoing message so the user sees
|
||||
// progress mid-run, and keep it out of the reply accumulator:
|
||||
// concatenated narrations read as a wall of text in the
|
||||
// final reply, and persisting them feeds the next turn's
|
||||
// LLM history an unanswered chain of stated intents that
|
||||
// replay mistakes for unfinished work (issue #120).
|
||||
String narration = delta.content() != null ? delta.content().trim() : "";
|
||||
if (relayNarration && !narration.isEmpty() && replyTarget != null) {
|
||||
try {
|
||||
adapter.renderAndSend(replyTarget, narration);
|
||||
} catch (Exception sendErr) {
|
||||
// A failed progress send must not abort the agent
|
||||
// run — the final reply still goes out below.
|
||||
log.warn("[{}] Narration relay failed (non-fatal): {}",
|
||||
channelType, sendErr.getMessage());
|
||||
}
|
||||
}
|
||||
} else if (delta.content() != null) {
|
||||
// Match the legacy agentService.chat() behavior: include
|
||||
// persistOnly deltas too. DirectAnswerNode-routed answers
|
||||
@ -1700,6 +1728,18 @@ public class ChannelMessageRouter {
|
||||
return "auto".equals(voiceMode) && "voice".equals(message.getInputMode());
|
||||
}
|
||||
|
||||
/**
|
||||
* Boolean lookup on the channel's configJson. Accepts Boolean or String
|
||||
* values, mirroring the adapter-side config parsing rules, so the router
|
||||
* and the adapters read the same key identically.
|
||||
*/
|
||||
private boolean channelConfigBoolean(ChannelEntity channelEntity, String key, boolean defaultValue) {
|
||||
Object value = parseChannelConfig(channelEntity.getConfigJson()).get(key);
|
||||
if (value instanceof Boolean b) return b;
|
||||
if (value instanceof String s && !s.isBlank()) return Boolean.parseBoolean(s.trim());
|
||||
return defaultValue;
|
||||
}
|
||||
|
||||
/**
|
||||
* 解析 Channel 的 configJson 为 Map
|
||||
*/
|
||||
|
||||
@ -0,0 +1,177 @@
|
||||
package vip.mate.channel;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import org.junit.jupiter.api.DisplayName;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.mockito.InOrder;
|
||||
import reactor.core.publisher.Flux;
|
||||
import vip.mate.agent.AgentService;
|
||||
import vip.mate.agent.AgentService.StreamDelta;
|
||||
import vip.mate.agent.context.ChatOrigin;
|
||||
import vip.mate.approval.ApprovalWorkflowService;
|
||||
import vip.mate.channel.model.ChannelEntity;
|
||||
import vip.mate.channel.notification.ApprovalNotificationService;
|
||||
import vip.mate.channel.service.ChannelService;
|
||||
import vip.mate.channel.web.ChatStreamTracker;
|
||||
import vip.mate.memory.event.ConversationCompletionPublisher;
|
||||
import vip.mate.tts.TtsService;
|
||||
import vip.mate.workspace.conversation.ConversationService;
|
||||
|
||||
import java.lang.reflect.Method;
|
||||
|
||||
import static org.mockito.ArgumentMatchers.*;
|
||||
import static org.mockito.Mockito.*;
|
||||
|
||||
/**
|
||||
* Sync-path narration routing for non-streaming IM adapters.
|
||||
*
|
||||
* <p>Per-iteration narration deltas ({@code segmentOnly}) must be relayed as
|
||||
* standalone messages the moment they arrive and stay out of the accumulated
|
||||
* final reply — otherwise multiple iterations glue into a wall of text and
|
||||
* the persisted assistant message pollutes the next turn's LLM history with
|
||||
* unanswered stated intents (issue #120).
|
||||
*/
|
||||
class ChannelMessageRouterNarrationTest {
|
||||
|
||||
@Test
|
||||
@DisplayName("segmentOnly narration is sent immediately and excluded from the final reply")
|
||||
void narrationRelayedImmediatelyAndExcludedFromReply() throws Exception {
|
||||
Fixture f = new Fixture();
|
||||
f.streamReturns(
|
||||
StreamDelta.segmentOnly("我先查一下天气", null),
|
||||
StreamDelta.segmentOnly("再帮你汇总结果", null),
|
||||
StreamDelta.persistOnly("今天北京晴,26 度。", null));
|
||||
|
||||
f.process("帮我查天气");
|
||||
|
||||
InOrder inOrder = inOrder(f.adapter);
|
||||
inOrder.verify(f.adapter).renderAndSend("reply-1", "我先查一下天气");
|
||||
inOrder.verify(f.adapter).renderAndSend("reply-1", "再帮你汇总结果");
|
||||
inOrder.verify(f.adapter).renderAndSend("reply-1", "今天北京晴,26 度。");
|
||||
f.verifyPersistedAssistantContent("今天北京晴,26 度。");
|
||||
f.verifyNoProcessingError();
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("blank narration deltas are skipped entirely")
|
||||
void blankNarrationSkipped() throws Exception {
|
||||
Fixture f = new Fixture();
|
||||
f.streamReturns(
|
||||
StreamDelta.segmentOnly(" ", null),
|
||||
StreamDelta.persistOnly("最终答案", null));
|
||||
|
||||
f.process("你好");
|
||||
|
||||
verify(f.adapter, times(1)).renderAndSend(anyString(), anyString());
|
||||
verify(f.adapter).renderAndSend("reply-1", "最终答案");
|
||||
f.verifyPersistedAssistantContent("最终答案");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("stream_progress=false suppresses narration relay but never re-adds it to the reply")
|
||||
void narrationSuppressedWhenProgressDisabled() throws Exception {
|
||||
Fixture f = new Fixture();
|
||||
f.channel.setConfigJson("{\"stream_progress\": false}");
|
||||
f.streamReturns(
|
||||
StreamDelta.segmentOnly("我先查一下天气", null),
|
||||
StreamDelta.persistOnly("今天晴。", null));
|
||||
|
||||
f.process("帮我查天气");
|
||||
|
||||
verify(f.adapter, never()).renderAndSend("reply-1", "我先查一下天气");
|
||||
verify(f.adapter).renderAndSend("reply-1", "今天晴。");
|
||||
f.verifyPersistedAssistantContent("今天晴。");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("a failed narration send does not abort the run — final reply still goes out")
|
||||
void narrationSendFailureIsNonFatal() throws Exception {
|
||||
Fixture f = new Fixture();
|
||||
doThrow(new RuntimeException("boom"))
|
||||
.when(f.adapter).renderAndSend("reply-1", "我先查一下天气");
|
||||
f.streamReturns(
|
||||
StreamDelta.segmentOnly("我先查一下天气", null),
|
||||
StreamDelta.persistOnly("今天晴。", null));
|
||||
|
||||
f.process("帮我查天气");
|
||||
|
||||
verify(f.adapter).renderAndSend("reply-1", "今天晴。");
|
||||
f.verifyPersistedAssistantContent("今天晴。");
|
||||
f.verifyNoProcessingError();
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("persistOnly and plain content deltas still accumulate into one reply")
|
||||
void nonNarrationDeltasStillAccumulate() throws Exception {
|
||||
Fixture f = new Fixture();
|
||||
f.streamReturns(
|
||||
new StreamDelta("答案第一段。", null),
|
||||
StreamDelta.persistOnly("答案第二段。", null));
|
||||
|
||||
f.process("你好");
|
||||
|
||||
verify(f.adapter).renderAndSend("reply-1", "答案第一段。答案第二段。");
|
||||
f.verifyPersistedAssistantContent("答案第一段。答案第二段。");
|
||||
}
|
||||
|
||||
// ==================== helpers ====================
|
||||
|
||||
/** Mocks + router wiring shared by the sync-path tests. */
|
||||
private static final class Fixture {
|
||||
final AgentService agentService = mock(AgentService.class);
|
||||
final ConversationService conversationService = mock(ConversationService.class);
|
||||
final ChatStreamTracker streamTracker = mock(ChatStreamTracker.class);
|
||||
final ChannelAdapter adapter = mock(ChannelAdapter.class);
|
||||
final ApprovalWorkflowService approvalService = mock(ApprovalWorkflowService.class);
|
||||
final ChannelErrorClassifier errorClassifier = mock(ChannelErrorClassifier.class);
|
||||
final ChannelChatOriginFactory chatOriginFactory = mock(ChannelChatOriginFactory.class);
|
||||
final ChannelEntity channel = new ChannelEntity();
|
||||
final ChannelMessageRouter router;
|
||||
|
||||
Fixture() {
|
||||
ChannelService channelService = mock(ChannelService.class);
|
||||
ChannelSessionStore channelSessionStore = mock(ChannelSessionStore.class);
|
||||
ApprovalNotificationService approvalNotificationService = mock(ApprovalNotificationService.class);
|
||||
ConversationCompletionPublisher completionPublisher = mock(ConversationCompletionPublisher.class);
|
||||
TtsService ttsService = mock(TtsService.class);
|
||||
router = new ChannelMessageRouter(agentService, conversationService,
|
||||
channelService, channelSessionStore, approvalService, approvalNotificationService,
|
||||
completionPublisher, ttsService, new ObjectMapper(), streamTracker,
|
||||
chatOriginFactory, errorClassifier);
|
||||
when(adapter.getChannelType()).thenReturn("telegram");
|
||||
when(chatOriginFactory.from(any(), any(), any(), any())).thenReturn(ChatOrigin.EMPTY);
|
||||
channel.setAgentId(100L);
|
||||
}
|
||||
|
||||
void streamReturns(StreamDelta... deltas) {
|
||||
when(agentService.chatStructuredStream(
|
||||
eq(100L), anyString(), anyString(), anyString(), any(ChatOrigin.class)))
|
||||
.thenReturn(Flux.fromArray(deltas));
|
||||
}
|
||||
|
||||
void process(String content) throws Exception {
|
||||
ChannelMessage message = ChannelMessage.builder()
|
||||
.senderId("alice")
|
||||
.replyToken("reply-1")
|
||||
.content(content)
|
||||
.build();
|
||||
Method process = ChannelMessageRouter.class.getDeclaredMethod(
|
||||
"processMessage", ChannelMessage.class, ChannelAdapter.class,
|
||||
ChannelEntity.class, String.class);
|
||||
process.setAccessible(true);
|
||||
process.invoke(router, message, adapter, channel, "telegram:alice");
|
||||
}
|
||||
|
||||
/** The persisted assistant row must hold the final-answer span only. */
|
||||
void verifyPersistedAssistantContent(String expected) {
|
||||
verify(conversationService).saveMessage(
|
||||
eq("telegram:alice"), eq("assistant"), eq(expected), isNull(), eq("completed"),
|
||||
eq(0), eq(0), eq(0), eq(0), eq(0), isNull(), isNull(), isNull());
|
||||
}
|
||||
|
||||
/** processMessage swallows failures into a generic error reply — assert none fired. */
|
||||
void verifyNoProcessingError() {
|
||||
verify(adapter, never()).sendMessage(anyString(), contains("处理消息时出现错误"));
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user