mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 03:13:41 +08:00
fix(channel): sync-path IM channels stop relaying pre-tool rehearsals verbatim
This commit is contained in:
matevip
parent
8a7a04ee1f
commit
32dd6e7911
@ -884,30 +884,47 @@ public class ChannelMessageRouter {
|
||||
// and IM channels still need the text for the outgoing reply),
|
||||
// segmentOnly narration excluded (issue #120).
|
||||
AgentStreamAccumulator accumulator = newAccumulator();
|
||||
// Narration lifecycle: relayed messages on this path are
|
||||
// permanent (IM messages cannot be retracted), so per-stage
|
||||
// narration publishes one behind through the shared tracker —
|
||||
// a pre-tool rehearsal (possibly a fabricated result table)
|
||||
// is dropped once later content supersedes it instead of
|
||||
// reaching the user verbatim.
|
||||
final ProvisionalContentTracker narrationTracker =
|
||||
new ProvisionalContentTracker(channelType);
|
||||
agentService.chatStructuredStream(agentId, promptText, conversationId,
|
||||
message.getSenderId(), chatOrigin)
|
||||
.doOnNext(delta -> {
|
||||
accumulator.accept(delta, conversationId);
|
||||
if (!delta.isEvent() && delta.segmentOnly()) {
|
||||
if (delta.isEvent()) {
|
||||
if ("tool_call_completed".equals(delta.eventType())) {
|
||||
narrationTracker.onToolObservation();
|
||||
}
|
||||
return;
|
||||
}
|
||||
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.
|
||||
// one complete delta per agent loop iteration, each becoming
|
||||
// its own outgoing message so the user sees progress mid-run.
|
||||
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());
|
||||
String publishable = narrationTracker.stageNarration(narration, delta.kind());
|
||||
if (publishable != null) {
|
||||
relayNarrationSafely(adapter, replyTarget, publishable);
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
.blockLast(Duration.ofMinutes(10));
|
||||
String reply = accumulator.getContent();
|
||||
// The last narration was held back until the answer was
|
||||
// known: superseded → dropped, otherwise it still goes out
|
||||
// (before the final reply) unless it duplicates it.
|
||||
String heldNarration = narrationTracker.settle(!reply.isBlank());
|
||||
if (heldNarration != null && replyTarget != null
|
||||
&& !heldNarration.equals(reply.trim())) {
|
||||
relayNarrationSafely(adapter, replyTarget, heldNarration);
|
||||
}
|
||||
|
||||
// The IM sync path bypasses FinalAnswerNode, so hallucinated
|
||||
// /api/v1/files/generated/{id} URLs (LLM wrote a fake link
|
||||
@ -1290,6 +1307,19 @@ public class ChannelMessageRouter {
|
||||
return s == null || s.isBlank() ? null : s;
|
||||
}
|
||||
|
||||
/**
|
||||
* Send a progress narration as its own outgoing message. A failed send
|
||||
* must not abort the agent run — the final reply still goes out.
|
||||
*/
|
||||
private void relayNarrationSafely(ChannelAdapter adapter, String replyTarget, String narration) {
|
||||
try {
|
||||
adapter.renderAndSend(replyTarget, narration);
|
||||
} catch (Exception sendErr) {
|
||||
log.warn("[{}] Narration relay failed (non-fatal): {}",
|
||||
adapter.getChannelType(), sendErr.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 流式处理路径（渠道无关）
|
||||
* <p>
|
||||
|
||||
@ -48,38 +48,53 @@ public final class ProvisionalContentTracker {
|
||||
/** Where the supersede happened — metric tag, one value per surface. */
|
||||
private final String surface;
|
||||
|
||||
/** Tool observations completed so far this turn (caller-reported). */
|
||||
private int observations;
|
||||
/** Observation count at the time of the most recent staging. */
|
||||
private int lastStageMark;
|
||||
|
||||
private String pendingText;
|
||||
private boolean pendingProvisional;
|
||||
/** Observation count when the pending narration was staged. */
|
||||
private int pendingMark;
|
||||
|
||||
public ProvisionalContentTracker(String surface) {
|
||||
this.surface = surface;
|
||||
}
|
||||
|
||||
/** Report a completed tool observation (a {@code tool_call_completed} event). */
|
||||
public void onToolObservation() {
|
||||
observations++;
|
||||
}
|
||||
|
||||
/**
|
||||
* Stage a per-round narration. Returns the <em>previous</em> staged
|
||||
* narration if the new arrival makes it publishable, or {@code null} when
|
||||
* there is nothing to publish (no previous, or the previous was
|
||||
* provisional and is now superseded by this later content).
|
||||
* provisional and tool observations since its staging mean this later
|
||||
* content supersedes it).
|
||||
*
|
||||
* @param kind producer-assigned kind; {@code null} for
|
||||
* pre-tag producers
|
||||
* @param observedSinceLast structural fallback used only when {@code kind}
|
||||
* is null: whether a tool observation completed
|
||||
* since the previous narration (the pre-tag
|
||||
* online rule)
|
||||
* @param kind producer-assigned kind; {@code null} for pre-tag producers,
|
||||
* in which case a narration counts as provisional when no
|
||||
* observation completed since the previous staging (the
|
||||
* pre-tag online rule)
|
||||
*/
|
||||
public String stageNarration(String text, ContentKind kind, boolean observedSinceLast) {
|
||||
public String stageNarration(String text, ContentKind kind) {
|
||||
boolean observedSinceLast = observations > lastStageMark;
|
||||
boolean provisional = kind != null
|
||||
? kind == ContentKind.PRE_TOOL_NARRATION
|
||||
: !observedSinceLast;
|
||||
String previous = pendingText;
|
||||
boolean previousProvisional = pendingProvisional;
|
||||
int previousMark = pendingMark;
|
||||
pendingText = text;
|
||||
pendingProvisional = provisional;
|
||||
pendingMark = observations;
|
||||
lastStageMark = observations;
|
||||
if (previous == null) {
|
||||
return null;
|
||||
}
|
||||
if (previousProvisional) {
|
||||
if (previousProvisional && observations > previousMark) {
|
||||
recordSuperseded(previous);
|
||||
return null;
|
||||
}
|
||||
@ -89,7 +104,8 @@ public final class ProvisionalContentTracker {
|
||||
/**
|
||||
* Resolve the staged narration at turn end. Returns the text to publish,
|
||||
* or {@code null} when nothing remains (no staged narration, or it was
|
||||
* provisional and the turn produced final content that replaces it).
|
||||
* provisional, tools ran after it, and the turn produced final content
|
||||
* that replaces it).
|
||||
*
|
||||
* @param hasFinalContent whether the turn produced a final answer — with
|
||||
* one, a provisional narration is superseded; with
|
||||
@ -99,12 +115,14 @@ public final class ProvisionalContentTracker {
|
||||
public String settle(boolean hasFinalContent) {
|
||||
String text = pendingText;
|
||||
boolean provisional = pendingProvisional;
|
||||
int mark = pendingMark;
|
||||
pendingText = null;
|
||||
pendingProvisional = false;
|
||||
pendingMark = 0;
|
||||
if (text == null) {
|
||||
return null;
|
||||
}
|
||||
if (provisional && hasFinalContent) {
|
||||
if (provisional && hasFinalContent && observations > mark) {
|
||||
recordSuperseded(text);
|
||||
return null;
|
||||
}
|
||||
|
||||
@ -1532,14 +1532,12 @@ public class WeComChannelAdapter extends AbstractChannelAdapter implements Strea
|
||||
// the final answer always lands in the newest bubble.
|
||||
AtomicReference<WeComReplyContext> liveCtx = new AtomicReference<>(initialCtx);
|
||||
ProvisionalContentTracker tracker = new ProvisionalContentTracker("wecom");
|
||||
AtomicInteger toolResults = new AtomicInteger();
|
||||
final int[] lastNarrationToolMark = {0};
|
||||
final long[] lastFlushAt = {0L};
|
||||
stream.doOnNext(delta -> {
|
||||
boolean flushNow = false;
|
||||
if (delta.isEvent()) {
|
||||
if ("tool_call_completed".equals(delta.eventType())) {
|
||||
toolResults.incrementAndGet();
|
||||
tracker.onToolObservation();
|
||||
}
|
||||
flushNow = progress.onEvent(delta.eventType(), delta.eventData());
|
||||
if (standaloneToolMessages) {
|
||||
@ -1563,11 +1561,8 @@ public class WeComChannelAdapter extends AbstractChannelAdapter implements Strea
|
||||
// means this one was written with real output in hand).
|
||||
String narration = delta.content() != null ? delta.content().trim() : "";
|
||||
if (!narration.isEmpty()) {
|
||||
int mark = toolResults.get();
|
||||
boolean observedSinceLast = mark > lastNarrationToolMark[0];
|
||||
lastNarrationToolMark[0] = mark;
|
||||
progress.onNarration(narration);
|
||||
String publishable = tracker.stageNarration(narration, delta.kind(), observedSinceLast);
|
||||
String publishable = tracker.stageNarration(narration, delta.kind());
|
||||
if (publishable != null) {
|
||||
WeComReplyContext ctx = liveCtx.get();
|
||||
if (replyContexts.get(replyTarget) == ctx) {
|
||||
|
||||
@ -25,11 +25,14 @@ 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).
|
||||
* <p>Per-iteration narration deltas ({@code segmentOnly}) are relayed as
|
||||
* standalone messages 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). Publishing lags one narration behind through the shared
|
||||
* provisional-content tracker: messages on this path are permanent, so a
|
||||
* pre-tool rehearsal (possibly a fabricated result table) must be dropped
|
||||
* once later content supersedes it instead of reaching the user verbatim.
|
||||
*/
|
||||
class ChannelMessageRouterNarrationTest {
|
||||
|
||||
@ -100,6 +103,61 @@ class ChannelMessageRouterNarrationTest {
|
||||
f.verifyNoProcessingError();
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("kind-tagged pre-tool rehearsal is dropped once a grounded answer exists")
|
||||
void preToolRehearsalDroppedForGroundedAnswer() throws Exception {
|
||||
Fixture f = new Fixture();
|
||||
String rehearsal = "环境监测结果：温度 29.0°C，湿度 63.0%。";
|
||||
f.streamReturns(
|
||||
StreamDelta.segmentOnly(rehearsal, null, vip.mate.agent.ContentKind.PRE_TOOL_NARRATION),
|
||||
StreamDelta.event("tool_call_completed",
|
||||
java.util.Map.of("toolCallId", "t1", "toolName", "envQuery",
|
||||
"result", "data={}", "success", true)),
|
||||
StreamDelta.persistOnly("接口返回为空，所有会议室均无环境数据。", null));
|
||||
|
||||
f.process("各会议室的环境情况");
|
||||
|
||||
verify(f.adapter, never()).renderAndSend("reply-1", rehearsal);
|
||||
verify(f.adapter).renderAndSend("reply-1", "接口返回为空，所有会议室均无环境数据。");
|
||||
f.verifyPersistedAssistantContent("接口返回为空，所有会议室均无环境数据。");
|
||||
f.verifyNoProcessingError();
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("kind-tagged pre-tool narration still goes out when the turn produced no answer")
|
||||
void preToolNarrationKeptWhenNoAnswer() throws Exception {
|
||||
Fixture f = new Fixture();
|
||||
f.streamReturns(
|
||||
StreamDelta.segmentOnly("我先调用工具查询状态：", null,
|
||||
vip.mate.agent.ContentKind.PRE_TOOL_NARRATION),
|
||||
StreamDelta.event("tool_call_completed",
|
||||
java.util.Map.of("toolCallId", "t1", "toolName", "q",
|
||||
"result", "x", "success", true)));
|
||||
|
||||
f.process("查一下状态");
|
||||
|
||||
verify(f.adapter).renderAndSend("reply-1", "我先调用工具查询状态：");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("untagged narration is dropped when a tool ran after it and later content followed")
|
||||
void untaggedNarrationDroppedAfterObservation() throws Exception {
|
||||
Fixture f = new Fixture();
|
||||
f.streamReturns(
|
||||
StreamDelta.segmentOnly("我先查一下天气", null),
|
||||
StreamDelta.event("tool_call_completed",
|
||||
java.util.Map.of("toolCallId", "t1", "toolName", "weather",
|
||||
"result", "晴", "success", true)),
|
||||
StreamDelta.segmentOnly("查到了，整理结果", null),
|
||||
StreamDelta.persistOnly("今天晴。", null));
|
||||
|
||||
f.process("帮我查天气");
|
||||
|
||||
verify(f.adapter, never()).renderAndSend("reply-1", "我先查一下天气");
|
||||
verify(f.adapter).renderAndSend("reply-1", "查到了，整理结果");
|
||||
verify(f.adapter).renderAndSend("reply-1", "今天晴。");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("persistOnly and plain content deltas still accumulate into one reply")
|
||||
void nonNarrationDeltasStillAccumulate() throws Exception {
|
||||
|
||||
@ -29,9 +29,10 @@ class ProvisionalContentTrackerTest {
|
||||
void preToolNarrationSupersededByNextNarration() {
|
||||
ProvisionalContentTracker t = new ProvisionalContentTracker("test");
|
||||
|
||||
assertNull(t.stageNarration("先查询会议室数据。", ContentKind.PRE_TOOL_NARRATION, false),
|
||||
assertNull(t.stageNarration("先查询会议室数据。", ContentKind.PRE_TOOL_NARRATION),
|
||||
"first narration has no predecessor to publish");
|
||||
String publishable = t.stageNarration("第一间空闲，继续。", ContentKind.GROUNDED_NARRATION, true);
|
||||
t.onToolObservation();
|
||||
String publishable = t.stageNarration("第一间空闲，继续。", ContentKind.GROUNDED_NARRATION);
|
||||
assertNull(publishable, "provisional predecessor is superseded, not published");
|
||||
|
||||
assertEquals("第一间空闲，继续。", t.settle(true),
|
||||
@ -43,8 +44,10 @@ class ProvisionalContentTrackerTest {
|
||||
void groundedNarrationPublishesOnNext() {
|
||||
ProvisionalContentTracker t = new ProvisionalContentTracker("test");
|
||||
|
||||
t.stageNarration("时间拿到了，再查会议室：", ContentKind.GROUNDED_NARRATION, true);
|
||||
String publishable = t.stageNarration("会议室也查完了。", ContentKind.GROUNDED_NARRATION, true);
|
||||
t.onToolObservation();
|
||||
t.stageNarration("时间拿到了，再查会议室：", ContentKind.GROUNDED_NARRATION);
|
||||
t.onToolObservation();
|
||||
String publishable = t.stageNarration("会议室也查完了。", ContentKind.GROUNDED_NARRATION);
|
||||
assertEquals("时间拿到了，再查会议室：", publishable);
|
||||
}
|
||||
|
||||
@ -53,7 +56,8 @@ class ProvisionalContentTrackerTest {
|
||||
void preToolNarrationSupersededByFinalAnswer() {
|
||||
ProvisionalContentTracker t = new ProvisionalContentTracker("test");
|
||||
|
||||
t.stageNarration("环境监测结果：温度 29.0°C。", ContentKind.PRE_TOOL_NARRATION, false);
|
||||
t.stageNarration("环境监测结果：温度 29.0°C。", ContentKind.PRE_TOOL_NARRATION);
|
||||
t.onToolObservation();
|
||||
assertNull(t.settle(true), "fabricated rehearsal must not survive once a grounded answer exists");
|
||||
}
|
||||
|
||||
@ -62,16 +66,28 @@ class ProvisionalContentTrackerTest {
|
||||
void preToolNarrationCommitsWithoutAnswer() {
|
||||
ProvisionalContentTracker t = new ProvisionalContentTracker("test");
|
||||
|
||||
t.stageNarration("我先调用工具查询状态：", ContentKind.PRE_TOOL_NARRATION, false);
|
||||
t.stageNarration("我先调用工具查询状态：", ContentKind.PRE_TOOL_NARRATION);
|
||||
t.onToolObservation();
|
||||
assertEquals("我先调用工具查询状态：", t.settle(false),
|
||||
"with no replacement content the narration is everything the user gets");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("pre-tool narration with no tool run after it survives settle — nothing replaced it")
|
||||
void preToolNarrationWithoutToolsAfterSurvives() {
|
||||
ProvisionalContentTracker t = new ProvisionalContentTracker("test");
|
||||
|
||||
t.stageNarration("我先调用工具：", ContentKind.PRE_TOOL_NARRATION);
|
||||
assertEquals("我先调用工具：", t.settle(true),
|
||||
"supersede requires an observation after staging — e.g. all tools denied leaves nothing to defer to");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("grounded narration always settles publishable")
|
||||
void groundedNarrationSettles() {
|
||||
ProvisionalContentTracker t = new ProvisionalContentTracker("test");
|
||||
t.stageNarration("查完了，结果如下。", ContentKind.GROUNDED_NARRATION, true);
|
||||
t.onToolObservation();
|
||||
t.stageNarration("查完了，结果如下。", ContentKind.GROUNDED_NARRATION);
|
||||
assertEquals("查完了，结果如下。", t.settle(true));
|
||||
}
|
||||
|
||||
@ -79,7 +95,7 @@ class ProvisionalContentTrackerTest {
|
||||
@DisplayName("settle clears state — second settle finds nothing")
|
||||
void settleClearsState() {
|
||||
ProvisionalContentTracker t = new ProvisionalContentTracker("test");
|
||||
t.stageNarration("x", ContentKind.GROUNDED_NARRATION, true);
|
||||
t.stageNarration("x", ContentKind.GROUNDED_NARRATION);
|
||||
t.settle(true);
|
||||
assertNull(t.settle(true));
|
||||
}
|
||||
@ -87,26 +103,40 @@ class ProvisionalContentTrackerTest {
|
||||
@Test
|
||||
@DisplayName("null kind falls back to the observation-counter rule")
|
||||
void nullKindFallsBackToCounter() {
|
||||
// No observation before staging → provisional; a tool ran after → superseded.
|
||||
ProvisionalContentTracker t = new ProvisionalContentTracker("test");
|
||||
|
||||
// No observation since turn start → provisional (pre-tag online rule).
|
||||
t.stageNarration("我先查一下：", null, false);
|
||||
t.stageNarration("我先查一下：", null);
|
||||
t.onToolObservation();
|
||||
assertNull(t.settle(true), "untagged pre-tool narration still dropped for a grounded answer");
|
||||
|
||||
// Observation completed since last narration → grounded.
|
||||
// Observation completed before staging → grounded, publishes.
|
||||
ProvisionalContentTracker t2 = new ProvisionalContentTracker("test");
|
||||
t2.stageNarration("拿到结果了。", null, true);
|
||||
t2.onToolObservation();
|
||||
t2.stageNarration("拿到结果了。", null);
|
||||
assertEquals("拿到结果了。", t2.settle(true));
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("untagged narrations with no tool activity between them all publish — legacy relay preserved")
|
||||
void untaggedNoToolStreamKeepsRelay() {
|
||||
ProvisionalContentTracker t = new ProvisionalContentTracker("test");
|
||||
t.stageNarration("我先查一下天气", null);
|
||||
assertEquals("我先查一下天气", t.stageNarration("再帮你汇总结果", null),
|
||||
"no observation between the two — the predecessor was not replaced by anything grounded");
|
||||
assertEquals("再帮你汇总结果", t.settle(true));
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("producer kind outranks the counter signal when both are present")
|
||||
void kindOutranksCounter() {
|
||||
// The counter says an observation happened, but the producer knows the
|
||||
// text was emitted before any observation of this turn (the two can
|
||||
// disagree when multiple narrations land between two completions).
|
||||
// The counter says an observation preceded the narration, but the
|
||||
// producer knows the text was emitted before any observation of this
|
||||
// turn (the two can disagree when multiple narrations land between
|
||||
// two completions). Kind wins.
|
||||
ProvisionalContentTracker t = new ProvisionalContentTracker("test");
|
||||
t.stageNarration("预演内容", ContentKind.PRE_TOOL_NARRATION, true);
|
||||
t.onToolObservation();
|
||||
t.stageNarration("预演内容", ContentKind.PRE_TOOL_NARRATION);
|
||||
t.onToolObservation();
|
||||
assertNull(t.settle(true), "kind is authoritative — counter signal ignored when tagged");
|
||||
}
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user