mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 03:13:41 +08:00
fix(channel): stop streaming replies from showing the same text twice
This commit is contained in:
matevip
parent
aaae3bf122
commit
ad6b0728e9
@ -36,4 +36,26 @@ public interface StreamingChannelAdapter extends ChannelAdapter {
|
||||
* @return 最终完整回复内容
|
||||
*/
|
||||
String processStream(Flux<StreamDelta> stream, ChannelMessage message, String conversationId);
|
||||
|
||||
/**
|
||||
* 判断一个 delta 的文本是否属于"最终回复内容"。
|
||||
* <p>
|
||||
* {@code segmentOnly} 的 delta 携带的是每轮 ReAct 的旁白（"我来查一下…"），
|
||||
* 共享累加器刻意不把它写进 {@code mate_message.content}。适配器如果直接
|
||||
* 累加 {@code delta.content()}，就会把每轮旁白拼进外发文本 —— 而旁白通常
|
||||
* 是对答案的复述，用户就会把同一段内容读到两三遍。被污染的文本还会回写
|
||||
* 持久化并在下一轮作为历史重放，重复量随轮次增长，而不是稳定在 2 倍。
|
||||
* <p>
|
||||
* 旁白要不要露出，由渠道的 {@code stream_progress} 开关决定：想露出就作为
|
||||
* 独立的进度消息下发，而不是混进最终答案。
|
||||
*
|
||||
* @param delta 流式片段
|
||||
* @return true 表示该片段的文本应计入最终回复
|
||||
*/
|
||||
static boolean contributesToFinalContent(StreamDelta delta) {
|
||||
return delta != null
|
||||
&& !delta.isEvent()
|
||||
&& !delta.segmentOnly()
|
||||
&& delta.content() != null;
|
||||
}
|
||||
}
|
||||
|
||||
@ -397,7 +397,10 @@ public class DingTalkChannelAdapter extends AbstractChannelAdapter implements St
|
||||
StringBuilder contentAccumulator = new StringBuilder();
|
||||
try {
|
||||
stream.doOnNext(delta -> {
|
||||
if (delta.content() != null) {
|
||||
// segmentOnly narration is skipped: appending every
|
||||
// ReAct iteration's "我来查一下…" into the card text is
|
||||
// what makes the answer read as if it were sent twice.
|
||||
if (StreamingChannelAdapter.contributesToFinalContent(delta)) {
|
||||
contentAccumulator.append(delta.content());
|
||||
aiCardManager.appendContent(outTrackId, delta.content(), false);
|
||||
}
|
||||
@ -453,7 +456,7 @@ public class DingTalkChannelAdapter extends AbstractChannelAdapter implements St
|
||||
StringBuilder contentAccumulator = new StringBuilder();
|
||||
|
||||
stream.doOnNext(delta -> {
|
||||
if (delta.content() != null) {
|
||||
if (StreamingChannelAdapter.contributesToFinalContent(delta)) {
|
||||
contentAccumulator.append(delta.content());
|
||||
}
|
||||
})
|
||||
|
||||
@ -2605,7 +2605,10 @@ public class FeishuChannelAdapter extends AbstractChannelAdapter implements Stre
|
||||
StringBuilder accumulator = new StringBuilder();
|
||||
try {
|
||||
stream.doOnNext(delta -> {
|
||||
if (delta.content() != null) {
|
||||
// segmentOnly narration is skipped: appending every
|
||||
// ReAct iteration's "我来查一下…" into the card text is
|
||||
// what makes the answer read as if it were sent twice.
|
||||
if (StreamingChannelAdapter.contributesToFinalContent(delta)) {
|
||||
accumulator.append(delta.content());
|
||||
streamingCardManager.appendContent(sessionKey, delta.content(), false);
|
||||
}
|
||||
@ -2667,7 +2670,7 @@ public class FeishuChannelAdapter extends AbstractChannelAdapter implements Stre
|
||||
private String processStreamAsText(Flux<StreamDelta> stream, ChannelMessage message) {
|
||||
StringBuilder accumulator = new StringBuilder();
|
||||
stream.doOnNext(delta -> {
|
||||
if (delta.content() != null) {
|
||||
if (StreamingChannelAdapter.contributesToFinalContent(delta)) {
|
||||
accumulator.append(delta.content());
|
||||
}
|
||||
})
|
||||
|
||||
@ -1405,37 +1405,109 @@ public class WeComChannelAdapter extends AbstractChannelAdapter implements Strea
|
||||
&& ctx.frameReqId() != null && !ctx.frameReqId().isBlank();
|
||||
|
||||
StringBuilder contentAccumulator = new StringBuilder();
|
||||
StreamOutcome outcome;
|
||||
if (!progressEnabled || !bubbleUsable) {
|
||||
// Degraded path — identical to the pre-streaming sync behavior:
|
||||
// accumulate content only (persistOnly deltas included, matching
|
||||
// the router's legacy collector) and send once at the end.
|
||||
// Degraded path — accumulate content only and send once at the end.
|
||||
// segmentOnly narration is excluded: gluing it into the reply is
|
||||
// what makes the same paragraph reach the user two or three times.
|
||||
stream.doOnNext(delta -> {
|
||||
if (!delta.isEvent() && delta.content() != null) {
|
||||
if (StreamingChannelAdapter.contributesToFinalContent(delta)) {
|
||||
contentAccumulator.append(delta.content());
|
||||
}
|
||||
})
|
||||
.blockLast(Duration.ofMinutes(10));
|
||||
outcome = new StreamOutcome(null, false);
|
||||
} else {
|
||||
consumeWithProgress(stream, message, replyTarget, ctx, contentAccumulator);
|
||||
outcome = consumeWithProgress(stream, message, replyTarget, ctx, contentAccumulator);
|
||||
}
|
||||
|
||||
String finalContent = contentAccumulator.toString();
|
||||
|
||||
// The newest narration was held back until the answer was known: when
|
||||
// a turn's closing narration and its final answer are the same text
|
||||
// (routine once the answer is short — the model restates it before the
|
||||
// last tool call), publishing both puts the identical bubble on screen
|
||||
// twice. Same text → drop the narration, the final answer covers it.
|
||||
String pendingNarration = outcome.pendingNarration();
|
||||
if (!finalContent.isBlank()) {
|
||||
if (pendingNarration != null && !sameOutboundText(pendingNarration, finalContent)) {
|
||||
publishNarrationBubble(replyTarget, pendingNarration);
|
||||
}
|
||||
renderAndSend(replyTarget, finalContent);
|
||||
} else if (pendingNarration != null) {
|
||||
// No final answer at all — the held-back narration is everything
|
||||
// the user gets, so it closes the live bubble in place instead of
|
||||
// being dropped. Not returned: narration never becomes persisted
|
||||
// content.
|
||||
renderAndSend(replyTarget, pendingNarration);
|
||||
} else {
|
||||
// Nothing to render. Without this the live bubble would sit at
|
||||
// "🤔 思考中…" with the tool trail under it forever, because only
|
||||
// renderAndSend ever finishes it.
|
||||
closeIdleProgressBubble(replyTarget, outcome.approvalPending());
|
||||
}
|
||||
return finalContent;
|
||||
}
|
||||
|
||||
/**
|
||||
* What {@link #consumeWithProgress} learned while draining the stream, but
|
||||
* which only {@link #processStream} can act on (it needs the final answer
|
||||
* first).
|
||||
*
|
||||
* @param pendingNarration the last per-stage narration, still unpublished
|
||||
* @param approvalPending a tool call is parked on human approval
|
||||
*/
|
||||
private record StreamOutcome(String pendingNarration, boolean approvalPending) {}
|
||||
|
||||
/** Compare two outbound texts the way the user sees them (post-filter, trimmed). */
|
||||
private boolean sameOutboundText(String a, String b) {
|
||||
if (a == null || b == null) {
|
||||
return false;
|
||||
}
|
||||
String left = filterOutboundContent(a).trim();
|
||||
String right = filterOutboundContent(b).trim();
|
||||
return !left.isEmpty() && left.equals(right);
|
||||
}
|
||||
|
||||
/**
|
||||
* Finish the live progress bubble when the turn produced no text at all
|
||||
* (approval park, user stop, empty answer). Leaving it open strands a
|
||||
* permanent "思考中…" bubble carrying the whole tool trail.
|
||||
*/
|
||||
private void closeIdleProgressBubble(String replyTarget, boolean approvalPending) {
|
||||
WeComReplyContext ctx = replyContexts.get(replyTarget);
|
||||
if (ctx == null || ctx.processingStreamId() == null || ctx.processingStreamId().isBlank()) {
|
||||
return;
|
||||
}
|
||||
renderAndSend(replyTarget, approvalPending
|
||||
? "⏸️ 已暂停，等待工具审批。"
|
||||
: "（本轮没有产生回复内容）");
|
||||
}
|
||||
|
||||
/**
|
||||
* Finalize the live progress bubble with a narration, or send the
|
||||
* narration as a plain message when no bubble is available (the keepalive
|
||||
* force-finish at the 180s ceiling evicts the reply context).
|
||||
*/
|
||||
private void publishNarrationBubble(String replyTarget, String narration) {
|
||||
WeComReplyContext ctx = replyContexts.get(replyTarget);
|
||||
if (ctx != null) {
|
||||
rollProgressBubble(replyTarget, ctx, narration, null);
|
||||
} else {
|
||||
sendMessage(replyTarget, narration);
|
||||
}
|
||||
}
|
||||
|
||||
/** Event-driven progress rendering into the processing-stream bubble, with per-stage bubble rolling. */
|
||||
private void consumeWithProgress(Flux<StreamDelta> stream, ChannelMessage message,
|
||||
String replyTarget, WeComReplyContext initialCtx,
|
||||
StringBuilder contentAccumulator) {
|
||||
private StreamOutcome consumeWithProgress(Flux<StreamDelta> stream, ChannelMessage message,
|
||||
String replyTarget, WeComReplyContext initialCtx,
|
||||
StringBuilder contentAccumulator) {
|
||||
boolean showThinking = !getConfigBoolean("filter_thinking", true);
|
||||
boolean standaloneToolMessages = !getConfigBoolean("filter_tool_messages", true);
|
||||
long minIntervalMs = getConfigLong("progress_interval_ms", PROGRESS_MIN_INTERVAL_MS);
|
||||
|
||||
WeComProgressRenderer progress = new WeComProgressRenderer(
|
||||
System.currentTimeMillis(), showThinking);
|
||||
System.currentTimeMillis(), showThinking, standaloneToolMessages);
|
||||
if (keepaliveScheduler != null) {
|
||||
// Silent stretches (long LLM calls with no events) keep showing a
|
||||
// fresh elapsed-time snapshot instead of the static placeholder.
|
||||
@ -1448,6 +1520,7 @@ public class WeComChannelAdapter extends AbstractChannelAdapter implements Strea
|
||||
// as the new progress bubble, so chat chronology stays intact and
|
||||
// the final answer always lands in the newest bubble.
|
||||
AtomicReference<WeComReplyContext> liveCtx = new AtomicReference<>(initialCtx);
|
||||
AtomicReference<String> pendingNarration = new AtomicReference<>();
|
||||
final long[] lastFlushAt = {0L};
|
||||
stream.doOnNext(delta -> {
|
||||
boolean flushNow = false;
|
||||
@ -1462,19 +1535,28 @@ public class WeComChannelAdapter extends AbstractChannelAdapter implements Strea
|
||||
// excluded from the final answer: glued together they read as
|
||||
// a wall of text, and persisted they pollute the next turn's
|
||||
// LLM history with unanswered chain-of-thought.
|
||||
//
|
||||
// Publishing lags one narration behind: the newest one is only
|
||||
// staged (visible live in the bubble, not yet finalized) so
|
||||
// processStream can still drop it if the final answer turns out
|
||||
// to be the same text. Without the lag the user reads the same
|
||||
// paragraph in two adjacent bubbles.
|
||||
String narration = delta.content() != null ? delta.content().trim() : "";
|
||||
if (!narration.isEmpty()) {
|
||||
WeComReplyContext ctx = liveCtx.get();
|
||||
if (replyContexts.get(replyTarget) == ctx) {
|
||||
liveCtx.set(rollProgressBubble(replyTarget, ctx, narration, progress));
|
||||
} else {
|
||||
// Bubble already force-finished (180s ceiling) — the
|
||||
// narration still goes out as a plain message.
|
||||
sendMessage(replyTarget, narration);
|
||||
String previous = pendingNarration.getAndSet(narration);
|
||||
progress.onNarration(narration);
|
||||
if (previous != null) {
|
||||
WeComReplyContext ctx = liveCtx.get();
|
||||
if (replyContexts.get(replyTarget) == ctx) {
|
||||
liveCtx.set(rollProgressBubble(replyTarget, ctx, previous, progress));
|
||||
} else {
|
||||
// Bubble already force-finished (180s ceiling) — the
|
||||
// narration still goes out as a plain message.
|
||||
sendMessage(replyTarget, previous);
|
||||
}
|
||||
}
|
||||
lastFlushAt[0] = System.currentTimeMillis();
|
||||
flushNow = true;
|
||||
}
|
||||
return;
|
||||
} else {
|
||||
if (delta.thinking() != null) {
|
||||
progress.onThinkingDelta(delta.thinking());
|
||||
@ -1506,6 +1588,8 @@ public class WeComChannelAdapter extends AbstractChannelAdapter implements Strea
|
||||
log.debug("[wecom] progress overwrite failed: {}", e.getMessage());
|
||||
}
|
||||
}).blockLast(Duration.ofMinutes(10));
|
||||
|
||||
return new StreamOutcome(pendingNarration.get(), progress.isApprovalPending());
|
||||
}
|
||||
|
||||
/**
|
||||
@ -1519,6 +1603,10 @@ public class WeComChannelAdapter extends AbstractChannelAdapter implements Strea
|
||||
* {@link #replyContexts} so {@code renderAndSend} / approval cards keep
|
||||
* working against the newest bubble, and keepalive restarts on the new
|
||||
* stream with the live progress snapshot.
|
||||
*
|
||||
* @param progress live progress renderer, or {@code null} when the stream
|
||||
* has already finished (the replacement bubble is then a
|
||||
* bare slot for {@code renderAndSend}, with no keepalive)
|
||||
*/
|
||||
private WeComReplyContext rollProgressBubble(String replyTarget, WeComReplyContext ctx,
|
||||
String narration, WeComProgressRenderer progress) {
|
||||
@ -1557,11 +1645,12 @@ public class WeComChannelAdapter extends AbstractChannelAdapter implements Strea
|
||||
WeComReplyContext next = new WeComReplyContext(ctx.frameReqId(), nextStreamId);
|
||||
replyContexts.put(replyTarget, next);
|
||||
try {
|
||||
replyStream(ctx.frameReqId(), nextStreamId, progress.snapshot(), false);
|
||||
replyStream(ctx.frameReqId(), nextStreamId,
|
||||
progress != null ? progress.snapshot() : "✍️ 正在整理…", false);
|
||||
} catch (Exception e) {
|
||||
log.debug("[wecom] next progress bubble open failed: {}", e.getMessage());
|
||||
}
|
||||
if (keepaliveScheduler != null) {
|
||||
if (progress != null && keepaliveScheduler != null) {
|
||||
try {
|
||||
keepaliveScheduler.start(this, ctx.frameReqId(), nextStreamId, replyTarget);
|
||||
keepaliveScheduler.attachTextSupplier(nextStreamId, progress::snapshot);
|
||||
|
||||
@ -33,6 +33,7 @@ final class WeComProgressRenderer {
|
||||
|
||||
private final long startedAtMillis;
|
||||
private final boolean showThinking;
|
||||
private final boolean showToolTrace;
|
||||
|
||||
private final Deque<ToolLine> toolLines = new ArrayDeque<>();
|
||||
private int collapsedToolCount;
|
||||
@ -42,10 +43,24 @@ final class WeComProgressRenderer {
|
||||
private boolean contentSeen;
|
||||
private boolean approvalPending;
|
||||
private String planStepLine;
|
||||
private String narration;
|
||||
|
||||
WeComProgressRenderer(long startedAtMillis, boolean showThinking) {
|
||||
/**
|
||||
* @param startedAtMillis turn start, for the elapsed-time counter
|
||||
* @param showThinking render the rolling reasoning window
|
||||
* ({@code filter_thinking=false})
|
||||
* @param showToolTrace render tool names and per-tool completion lines
|
||||
* ({@code filter_tool_messages=false}). When off,
|
||||
* the bubble only says that a tool is running —
|
||||
* the whole point of the toggle is that the tool
|
||||
* trail stays out of the user's view, and the
|
||||
* progress bubble is as user-visible as a
|
||||
* standalone trace message.
|
||||
*/
|
||||
WeComProgressRenderer(long startedAtMillis, boolean showThinking, boolean showToolTrace) {
|
||||
this.startedAtMillis = startedAtMillis;
|
||||
this.showThinking = showThinking;
|
||||
this.showToolTrace = showToolTrace;
|
||||
}
|
||||
|
||||
synchronized void onThinkingDelta(String delta) {
|
||||
@ -105,6 +120,20 @@ final class WeComProgressRenderer {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Stage the newest per-stage narration so it shows in the live bubble
|
||||
* straight away. Only the newest one is kept — the previous narration has
|
||||
* already been published as its own bubble by then.
|
||||
*/
|
||||
synchronized void onNarration(String text) {
|
||||
narration = (text == null || text.isBlank()) ? null : text.trim();
|
||||
}
|
||||
|
||||
/** True once a tool call has asked for human approval this turn. */
|
||||
synchronized boolean isApprovalPending() {
|
||||
return approvalPending;
|
||||
}
|
||||
|
||||
/** Render the current progress text for the stream bubble. */
|
||||
synchronized String snapshot() {
|
||||
StringBuilder sb = new StringBuilder();
|
||||
@ -116,6 +145,9 @@ final class WeComProgressRenderer {
|
||||
if (showThinking && !contentSeen && thinkingTail.length() > 0) {
|
||||
sb.append("\n\n> 💭 ").append(thinkingTail.toString().replace("\n", "\n> "));
|
||||
}
|
||||
if (narration != null) {
|
||||
sb.append("\n\n").append(narration);
|
||||
}
|
||||
if (answerTail.length() > 0) {
|
||||
sb.append("\n\n").append(answerTail);
|
||||
}
|
||||
@ -131,7 +163,9 @@ final class WeComProgressRenderer {
|
||||
}
|
||||
ToolLine running = lastRunningTool();
|
||||
if (running != null) {
|
||||
return "🔧 正在调用 " + displayName(running) + "…（" + elapsed() + "）";
|
||||
return showToolTrace
|
||||
? "🔧 正在调用 " + displayName(running) + "…（" + elapsed() + "）"
|
||||
: "🔧 正在执行工具…（" + elapsed() + "）";
|
||||
}
|
||||
if (thinkingSeen) {
|
||||
return "💭 思考中…（" + elapsed() + "）";
|
||||
@ -140,6 +174,11 @@ final class WeComProgressRenderer {
|
||||
}
|
||||
|
||||
private void appendToolLines(StringBuilder sb) {
|
||||
if (!showToolTrace) {
|
||||
// filter_tool_messages=true — the tool trail is suppressed
|
||||
// everywhere the user can see it, progress bubble included.
|
||||
return;
|
||||
}
|
||||
if (collapsedToolCount > 0) {
|
||||
sb.append("\n…等 ").append(collapsedToolCount).append(" 项已完成");
|
||||
}
|
||||
|
||||
@ -39,7 +39,8 @@ class WeComProcessStreamTest {
|
||||
@Test
|
||||
@DisplayName("progress path overwrites the bubble with tool progress, then finishes with the answer")
|
||||
void progressPathStreamsAndFinishes() throws Exception {
|
||||
TestableAdapter adapter = newAdapter("{\"progress_interval_ms\": 0}");
|
||||
TestableAdapter adapter = newAdapter(
|
||||
"{\"progress_interval_ms\": 0, \"filter_tool_messages\": false}");
|
||||
seedReplyContext(adapter, "alice", "req-1", "stream-1");
|
||||
|
||||
Flux<StreamDelta> stream = Flux.just(
|
||||
@ -68,6 +69,90 @@ class WeComProcessStreamTest {
|
||||
assertTrue(String.valueOf(finalChunk.get("content")).contains("现在是下午三点。"));
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("filter_tool_messages=true keeps the tool trail out of the progress bubble too")
|
||||
void progressBubbleHonorsToolFilter() throws Exception {
|
||||
TestableAdapter adapter = newAdapter("{\"progress_interval_ms\": 0}");
|
||||
seedReplyContext(adapter, "alice", "req-1", "stream-1");
|
||||
|
||||
Flux<StreamDelta> stream = Flux.just(
|
||||
StreamDelta.event("tool_call_started",
|
||||
Map.of("toolCallId", "c1", "toolName", "execute_code")),
|
||||
StreamDelta.event("tool_call_completed",
|
||||
Map.of("toolCallId", "c1", "toolName", "execute_code", "success", true)),
|
||||
new StreamDelta("好了。", null));
|
||||
|
||||
assertEquals("好了。", adapter.processStream(stream, inbound("alice"), "wecom:alice"));
|
||||
|
||||
List<Map<String, Object>> streamBodies = streamBodies(adapter.drainFrames());
|
||||
assertFalse(streamBodies.isEmpty(), "expected reply_stream progress frames");
|
||||
for (Map<String, Object> body : streamBodies) {
|
||||
String content = String.valueOf(body.get("content"));
|
||||
assertFalse(content.contains("execute_code"),
|
||||
"filtered run must never name the tool: " + content);
|
||||
assertFalse(content.contains("✅ execute_code"), content);
|
||||
}
|
||||
boolean sawGenericToolStatus = streamBodies.stream()
|
||||
.anyMatch(s -> String.valueOf(s.get("content")).contains("正在执行工具"));
|
||||
assertTrue(sawGenericToolStatus, "still tells the user a tool is running, just not which");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("a closing narration equal to the final answer is dropped, not shown twice")
|
||||
void narrationEqualToAnswerIsNotDuplicated() throws Exception {
|
||||
TestableAdapter adapter = newAdapter("{\"progress_interval_ms\": 0}");
|
||||
seedReplyContext(adapter, "alice", "req-1", "stream-1");
|
||||
|
||||
String answer = "中控测试会议室当前无人使用，电池还剩 1%。";
|
||||
Flux<StreamDelta> stream = Flux.just(
|
||||
StreamDelta.segmentOnly(answer, null),
|
||||
new StreamDelta(answer, null));
|
||||
|
||||
assertEquals(answer, adapter.processStream(stream, inbound("alice"), "wecom:alice"));
|
||||
|
||||
List<Map<String, Object>> finished = streamBodies(adapter.drainFrames()).stream()
|
||||
.filter(s -> Boolean.TRUE.equals(s.get("finish"))).toList();
|
||||
assertEquals(1, finished.size(),
|
||||
"narration restating the answer must not close a bubble of its own");
|
||||
assertTrue(String.valueOf(finished.get(0).get("content")).contains("无人使用"));
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("a turn with no answer closes the progress bubble instead of stranding 思考中")
|
||||
void emptyAnswerClosesProgressBubble() throws Exception {
|
||||
TestableAdapter adapter = newAdapter("{\"progress_interval_ms\": 0}");
|
||||
seedReplyContext(adapter, "alice", "req-1", "stream-1");
|
||||
|
||||
Flux<StreamDelta> stream = Flux.just(
|
||||
StreamDelta.event("tool_call_started",
|
||||
Map.of("toolCallId", "c1", "toolName", "execute_code")),
|
||||
StreamDelta.event("tool_call_completed",
|
||||
Map.of("toolCallId", "c1", "toolName", "execute_code", "success", true)));
|
||||
|
||||
assertEquals("", adapter.processStream(stream, inbound("alice"), "wecom:alice"));
|
||||
|
||||
List<Map<String, Object>> streamBodies = streamBodies(adapter.drainFrames());
|
||||
assertFalse(streamBodies.isEmpty(), "expected reply_stream frames");
|
||||
Map<String, Object> last = streamBodies.get(streamBodies.size() - 1);
|
||||
assertEquals(Boolean.TRUE, last.get("finish"),
|
||||
"the bubble must be closed, otherwise it sits at 思考中 forever");
|
||||
assertFalse(String.valueOf(last.get("content")).contains("思考中"), String.valueOf(last.get("content")));
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("degraded path keeps stage narration out of the reply text")
|
||||
void degradedPathExcludesNarration() throws Exception {
|
||||
TestableAdapter adapter = newAdapter("{\"stream_progress\": false}");
|
||||
seedReplyContext(adapter, "alice", "req-1", "stream-1");
|
||||
|
||||
Flux<StreamDelta> stream = Flux.just(
|
||||
StreamDelta.segmentOnly("我先查一下：", null),
|
||||
new StreamDelta("答案", null));
|
||||
|
||||
assertEquals("答案", adapter.processStream(stream, inbound("alice"), "wecom:alice"),
|
||||
"narration glued into the reply is what makes the answer read as sent twice");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("stage narrations roll the bubble: each stage finishes its own bubble, final answer excludes them")
|
||||
void stageNarrationsRollBubbles() throws Exception {
|
||||
|
||||
@ -12,7 +12,7 @@ class WeComProgressRendererTest {
|
||||
@Test
|
||||
@DisplayName("initial snapshot shows a thinking status with elapsed time")
|
||||
void initialSnapshot() {
|
||||
WeComProgressRenderer r = new WeComProgressRenderer(System.currentTimeMillis(), false);
|
||||
WeComProgressRenderer r = new WeComProgressRenderer(System.currentTimeMillis(), false, true);
|
||||
String s = r.snapshot();
|
||||
assertTrue(s.contains("思考中"), s);
|
||||
assertTrue(s.contains("已 "), s);
|
||||
@ -21,7 +21,7 @@ class WeComProgressRendererTest {
|
||||
@Test
|
||||
@DisplayName("tool start flips status to the running tool and requests an immediate flush")
|
||||
void toolStartUpdatesStatus() {
|
||||
WeComProgressRenderer r = new WeComProgressRenderer(System.currentTimeMillis(), false);
|
||||
WeComProgressRenderer r = new WeComProgressRenderer(System.currentTimeMillis(), false, true);
|
||||
boolean flush = r.onEvent("tool_call_started",
|
||||
Map.of("toolCallId", "c1", "toolName", "get_weather"));
|
||||
assertTrue(flush);
|
||||
@ -32,7 +32,7 @@ class WeComProgressRendererTest {
|
||||
@Test
|
||||
@DisplayName("tool completion renders a checked line with duration")
|
||||
void toolCompletionRendersLine() {
|
||||
WeComProgressRenderer r = new WeComProgressRenderer(System.currentTimeMillis(), false);
|
||||
WeComProgressRenderer r = new WeComProgressRenderer(System.currentTimeMillis(), false, true);
|
||||
r.onEvent("tool_call_started", Map.of("toolCallId", "c1", "toolName", "get_weather"));
|
||||
r.onEvent("tool_call_completed",
|
||||
Map.of("toolCallId", "c1", "toolName", "get_weather", "success", true));
|
||||
@ -44,7 +44,7 @@ class WeComProgressRendererTest {
|
||||
@Test
|
||||
@DisplayName("failed tool renders a cross line")
|
||||
void toolFailureRendersCross() {
|
||||
WeComProgressRenderer r = new WeComProgressRenderer(System.currentTimeMillis(), false);
|
||||
WeComProgressRenderer r = new WeComProgressRenderer(System.currentTimeMillis(), false, true);
|
||||
r.onEvent("tool_call_started", Map.of("toolCallId", "c1", "toolName", "search"));
|
||||
r.onEvent("tool_call_completed",
|
||||
Map.of("toolCallId", "c1", "toolName", "search", "success", false));
|
||||
@ -55,7 +55,7 @@ class WeComProgressRendererTest {
|
||||
@Test
|
||||
@DisplayName("older completed tools collapse into a counter beyond the display cap")
|
||||
void toolLinesCollapse() {
|
||||
WeComProgressRenderer r = new WeComProgressRenderer(System.currentTimeMillis(), false);
|
||||
WeComProgressRenderer r = new WeComProgressRenderer(System.currentTimeMillis(), false, true);
|
||||
for (int i = 1; i <= 5; i++) {
|
||||
String id = "c" + i;
|
||||
r.onEvent("tool_call_started", Map.of("toolCallId", id, "toolName", "tool" + i));
|
||||
@ -70,11 +70,11 @@ class WeComProgressRendererTest {
|
||||
@Test
|
||||
@DisplayName("thinking text appears as a quote block only when display is enabled")
|
||||
void thinkingDisplayGate() {
|
||||
WeComProgressRenderer shown = new WeComProgressRenderer(System.currentTimeMillis(), true);
|
||||
WeComProgressRenderer shown = new WeComProgressRenderer(System.currentTimeMillis(), true, true);
|
||||
shown.onThinkingDelta("先查当前时间");
|
||||
assertTrue(shown.snapshot().contains("> 💭 先查当前时间"), shown.snapshot());
|
||||
|
||||
WeComProgressRenderer hidden = new WeComProgressRenderer(System.currentTimeMillis(), false);
|
||||
WeComProgressRenderer hidden = new WeComProgressRenderer(System.currentTimeMillis(), false, true);
|
||||
hidden.onThinkingDelta("先查当前时间");
|
||||
String s = hidden.snapshot();
|
||||
assertFalse(s.contains("先查当前时间"), s);
|
||||
@ -84,7 +84,7 @@ class WeComProgressRendererTest {
|
||||
@Test
|
||||
@DisplayName("content deltas switch status to replying and show the answer tail")
|
||||
void contentSwitchesToReplying() {
|
||||
WeComProgressRenderer r = new WeComProgressRenderer(System.currentTimeMillis(), true);
|
||||
WeComProgressRenderer r = new WeComProgressRenderer(System.currentTimeMillis(), true, true);
|
||||
r.onThinkingDelta("想一想");
|
||||
r.onContentDelta("今天天气晴。");
|
||||
String s = r.snapshot();
|
||||
@ -97,7 +97,7 @@ class WeComProgressRendererTest {
|
||||
@Test
|
||||
@DisplayName("approval request switches status to waiting")
|
||||
void approvalSwitchesStatus() {
|
||||
WeComProgressRenderer r = new WeComProgressRenderer(System.currentTimeMillis(), false);
|
||||
WeComProgressRenderer r = new WeComProgressRenderer(System.currentTimeMillis(), false, true);
|
||||
boolean flush = r.onEvent("tool_approval_requested", Map.of("toolName", "rm"));
|
||||
assertTrue(flush);
|
||||
assertTrue(r.snapshot().contains("等待工具审批"), r.snapshot());
|
||||
@ -106,16 +106,54 @@ class WeComProgressRendererTest {
|
||||
@Test
|
||||
@DisplayName("answer tail stays bounded for very long streamed content")
|
||||
void answerTailBounded() {
|
||||
WeComProgressRenderer r = new WeComProgressRenderer(System.currentTimeMillis(), false);
|
||||
WeComProgressRenderer r = new WeComProgressRenderer(System.currentTimeMillis(), false, true);
|
||||
r.onContentDelta("x".repeat(5000));
|
||||
assertTrue(r.snapshot().length() < 2048,
|
||||
"snapshot must stay under the WeCom message limit");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("filter_tool_messages=true keeps the tool trail out of the progress bubble")
|
||||
void toolTraceGate() {
|
||||
WeComProgressRenderer hidden = new WeComProgressRenderer(System.currentTimeMillis(), false, false);
|
||||
hidden.onEvent("tool_call_started", Map.of("toolCallId", "c1", "toolName", "execute_code"));
|
||||
String running = hidden.snapshot();
|
||||
assertTrue(running.contains("正在执行工具"), running);
|
||||
assertFalse(running.contains("execute_code"), running);
|
||||
|
||||
hidden.onEvent("tool_call_completed",
|
||||
Map.of("toolCallId", "c1", "toolName", "execute_code", "success", true));
|
||||
String done = hidden.snapshot();
|
||||
assertFalse(done.contains("execute_code"), done);
|
||||
assertFalse(done.contains("✅"), done);
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("staged narration shows in the live snapshot and the newest one replaces it")
|
||||
void narrationStaging() {
|
||||
WeComProgressRenderer r = new WeComProgressRenderer(System.currentTimeMillis(), false, true);
|
||||
r.onNarration("先查一下会议室");
|
||||
assertTrue(r.snapshot().contains("先查一下会议室"), r.snapshot());
|
||||
|
||||
r.onNarration("会议室拿到了，接着改时间");
|
||||
String s = r.snapshot();
|
||||
assertTrue(s.contains("会议室拿到了"), s);
|
||||
assertFalse(s.contains("先查一下会议室"), s);
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("approval pending is observable after the stream drains")
|
||||
void approvalPendingObservable() {
|
||||
WeComProgressRenderer r = new WeComProgressRenderer(System.currentTimeMillis(), false, true);
|
||||
assertFalse(r.isApprovalPending());
|
||||
r.onEvent("tool_approval_requested", Map.of("toolName", "rm"));
|
||||
assertTrue(r.isApprovalPending());
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("unknown events are ignored without requesting a flush")
|
||||
void unknownEventsIgnored() {
|
||||
WeComProgressRenderer r = new WeComProgressRenderer(System.currentTimeMillis(), false);
|
||||
WeComProgressRenderer r = new WeComProgressRenderer(System.currentTimeMillis(), false, true);
|
||||
assertFalse(r.onEvent("_usage_final", Map.of()));
|
||||
assertFalse(r.onEvent(null, null));
|
||||
}
|
||||
|
||||
@ -3618,7 +3618,7 @@ export default {
|
||||
filterThinking: 'Filter Thinking',
|
||||
filterThinkingTooltip: 'Yes: strip <think> content from the final answer only; No: show reasoning live (WeCom streams it in the progress bubble)',
|
||||
filterToolMessages: 'Filter Tool Messages',
|
||||
filterToolMessagesTooltip: 'Yes: strip inline tool_call / Action tags from the final answer only; No: each tool call leaves a standalone trace message (supported on WeCom)',
|
||||
filterToolMessagesTooltip: 'Yes: strip inline tool_call / Action tags from the final answer, and keep tool names and the execution list out of the progress bubble too; No: show tool execution, and each tool call leaves a standalone trace message (supported on WeCom)',
|
||||
messageFormat: 'Message Format',
|
||||
formatAuto: 'Auto',
|
||||
formatMarkdown: 'Markdown',
|
||||
|
||||
@ -3718,7 +3718,7 @@ export default {
|
||||
filterThinking: '过滤思维链',
|
||||
filterThinkingTooltip: '是：仅过滤最终答案中的 <think> 内容；否：思考过程实时展示（企业微信在进度气泡中流式显示）',
|
||||
filterToolMessages: '过滤工具消息',
|
||||
filterToolMessagesTooltip: '是：仅过滤最终答案中的 tool_call / Action 等内联标签；否：每次工具调用以独立消息留痕（企业微信已支持）',
|
||||
filterToolMessagesTooltip: '是：过滤最终答案中的 tool_call / Action 等内联标签，进度气泡也不再显示工具名与执行清单；否：显示工具执行过程，并且每次工具调用以独立消息留痕（企业微信已支持）',
|
||||
messageFormat: '消息格式',
|
||||
formatAuto: '自动',
|
||||
formatMarkdown: 'Markdown',
|
||||
|
||||
Loading…
Reference in New Issue
Block a user