mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-15 11:58:34 +08:00
feat(chat): live pre-tool narration collapse via segment_kind SSE event
This commit is contained in:
parent
32dd6e7911
commit
fefb3b25b8
@ -252,7 +252,7 @@ public class StateGraphReActAgent extends BaseAgent implements StructuredStreamC
|
|||||||
String streamed = output.state().<String>value(STREAMED_CONTENT).orElse("");
|
String streamed = output.state().<String>value(STREAMED_CONTENT).orElse("");
|
||||||
if (!streamed.isEmpty() && !streamed.equals(lastEmittedStreamedContent.get())) {
|
if (!streamed.isEmpty() && !streamed.equals(lastEmittedStreamedContent.get())) {
|
||||||
lastEmittedStreamedContent.set(streamed);
|
lastEmittedStreamedContent.set(streamed);
|
||||||
deltas.add(streamedContentDelta(isFinalAnswerTurn,
|
addWithKindEvent(deltas, streamedContentDelta(isFinalAnswerTurn,
|
||||||
output.state().value(NEEDS_TOOL_CALL, false),
|
output.state().value(NEEDS_TOOL_CALL, false),
|
||||||
output.state().value(CURRENT_ITERATION, 0),
|
output.state().value(CURRENT_ITERATION, 0),
|
||||||
streamed));
|
streamed));
|
||||||
@ -261,7 +261,7 @@ public class StateGraphReActAgent extends BaseAgent implements StructuredStreamC
|
|||||||
if (isFinalAnswerTurn && finalAnswerEmitted.compareAndSet(false, true)) {
|
if (isFinalAnswerTurn && finalAnswerEmitted.compareAndSet(false, true)) {
|
||||||
String answer = extractFinalAnswer(output);
|
String answer = extractFinalAnswer(output);
|
||||||
if (answer != null && !answer.isEmpty()) {
|
if (answer != null && !answer.isEmpty()) {
|
||||||
deltas.add(AgentService.StreamDelta.finalAnswer(answer, contentAlreadyStreamed));
|
addWithKindEvent(deltas, AgentService.StreamDelta.finalAnswer(answer, contentAlreadyStreamed));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
String thinking = extractFinalThinking(output);
|
String thinking = extractFinalThinking(output);
|
||||||
@ -424,7 +424,7 @@ public class StateGraphReActAgent extends BaseAgent implements StructuredStreamC
|
|||||||
String streamed = output.state().<String>value(STREAMED_CONTENT).orElse("");
|
String streamed = output.state().<String>value(STREAMED_CONTENT).orElse("");
|
||||||
if (!streamed.isEmpty() && !streamed.equals(lastEmittedStreamedContent.get())) {
|
if (!streamed.isEmpty() && !streamed.equals(lastEmittedStreamedContent.get())) {
|
||||||
lastEmittedStreamedContent.set(streamed);
|
lastEmittedStreamedContent.set(streamed);
|
||||||
deltas.add(streamedContentDelta(isFinalAnswerTurn,
|
addWithKindEvent(deltas, streamedContentDelta(isFinalAnswerTurn,
|
||||||
output.state().value(NEEDS_TOOL_CALL, false),
|
output.state().value(NEEDS_TOOL_CALL, false),
|
||||||
output.state().value(CURRENT_ITERATION, 0),
|
output.state().value(CURRENT_ITERATION, 0),
|
||||||
streamed));
|
streamed));
|
||||||
@ -433,7 +433,7 @@ public class StateGraphReActAgent extends BaseAgent implements StructuredStreamC
|
|||||||
if (isFinalAnswerTurn && finalAnswerEmitted.compareAndSet(false, true)) {
|
if (isFinalAnswerTurn && finalAnswerEmitted.compareAndSet(false, true)) {
|
||||||
String answer = extractFinalAnswer(output);
|
String answer = extractFinalAnswer(output);
|
||||||
if (answer != null && !answer.isEmpty()) {
|
if (answer != null && !answer.isEmpty()) {
|
||||||
deltas.add(AgentService.StreamDelta.finalAnswer(answer, contentAlreadyStreamed));
|
addWithKindEvent(deltas, AgentService.StreamDelta.finalAnswer(answer, contentAlreadyStreamed));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -666,6 +666,24 @@ public class StateGraphReActAgent extends BaseAgent implements StructuredStreamC
|
|||||||
* caller's responsibility — this helper just decides flavor for non-blank
|
* caller's responsibility — this helper just decides flavor for non-blank
|
||||||
* content.
|
* content.
|
||||||
*/
|
*/
|
||||||
|
/**
|
||||||
|
* Append a content-bearing delta plus, when it carries a producer-assigned
|
||||||
|
* kind, a {@code segment_kind} broadcast event tagging the just-emitted
|
||||||
|
* content span. The kind cannot ride on the live {@code content_delta}
|
||||||
|
* broadcasts — text streams before the producer knows whether the
|
||||||
|
* completion carries tool calls — so it is delivered as a follow-up event
|
||||||
|
* once the completion resolves, letting the client tag its running content
|
||||||
|
* segment and collapse a provisional narration the moment later content
|
||||||
|
* arrives, without waiting for the persisted-metadata round-trip.
|
||||||
|
*/
|
||||||
|
static void addWithKindEvent(List<AgentService.StreamDelta> deltas, AgentService.StreamDelta delta) {
|
||||||
|
deltas.add(delta);
|
||||||
|
if (delta.kind() != null) {
|
||||||
|
deltas.add(AgentService.StreamDelta.event("segment_kind",
|
||||||
|
Map.of("kind", delta.kind().wireName())));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
static AgentService.StreamDelta streamedContentDelta(boolean isFinalAnswerTurn, boolean carriesToolCalls,
|
static AgentService.StreamDelta streamedContentDelta(boolean isFinalAnswerTurn, boolean carriesToolCalls,
|
||||||
int iteration, String streamed) {
|
int iteration, String streamed) {
|
||||||
if (isFinalAnswerTurn) {
|
if (isFinalAnswerTurn) {
|
||||||
|
|||||||
@ -114,4 +114,28 @@ class StateGraphReActAgentStreamedContentDeltaTest {
|
|||||||
assertNull(StateGraphReActAgent.streamedContentDelta(false, true, 0, "x").thinking());
|
assertNull(StateGraphReActAgent.streamedContentDelta(false, true, 0, "x").thinking());
|
||||||
assertNull(StateGraphReActAgent.streamedContentDelta(true, false, 1, "x").thinking());
|
assertNull(StateGraphReActAgent.streamedContentDelta(true, false, 1, "x").thinking());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
@DisplayName("kind-carrying delta is followed by a segment_kind broadcast event")
|
||||||
|
void kindDeltaEmitsSegmentKindEvent() {
|
||||||
|
java.util.List<AgentService.StreamDelta> deltas = new java.util.ArrayList<>();
|
||||||
|
StateGraphReActAgent.addWithKindEvent(deltas,
|
||||||
|
StateGraphReActAgent.streamedContentDelta(false, true, 0, "先查询。"));
|
||||||
|
|
||||||
|
assertEquals(2, deltas.size());
|
||||||
|
AgentService.StreamDelta event = deltas.get(1);
|
||||||
|
assertTrue(event.isEvent());
|
||||||
|
assertEquals("segment_kind", event.eventType());
|
||||||
|
assertEquals("pre_tool_narration", event.eventData().get("kind"));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
@DisplayName("untagged delta emits no segment_kind event")
|
||||||
|
void untaggedDeltaEmitsNoEvent() {
|
||||||
|
java.util.List<AgentService.StreamDelta> deltas = new java.util.ArrayList<>();
|
||||||
|
StateGraphReActAgent.addWithKindEvent(deltas, AgentService.StreamDelta.segmentOnly("x", null));
|
||||||
|
|
||||||
|
assertEquals(1, deltas.size());
|
||||||
|
assertFalse(deltas.get(0).isEvent());
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@ -497,6 +497,17 @@ export function useChat(options: UseChatOptions): UseChatReturn {
|
|||||||
if (thinkingSeg) thinkingSeg.status = 'completed'
|
if (thinkingSeg) thinkingSeg.status = 'completed'
|
||||||
contentSeg = { id: genSegId(), type: 'content', status: 'running', text: '', timestamp: Date.now() }
|
contentSeg = { id: genSegId(), type: 'content', status: 'running', text: '', timestamp: Date.now() }
|
||||||
applyIterationTags(contentSeg)
|
applyIterationTags(contentSeg)
|
||||||
|
// Later content supersedes any earlier provisional narration of the
|
||||||
|
// same turn — collapse it in place, live, instead of waiting for the
|
||||||
|
// persisted-metadata annotations in the done payload. Mirrors the
|
||||||
|
// backend tracker's rule for kind-tagged segments.
|
||||||
|
for (const s of segs) {
|
||||||
|
if (s.type === 'content' && s.kind === 'pre_tool_narration' && !s.superseded) {
|
||||||
|
s.superseded = true
|
||||||
|
s.supersededBySegmentId = String(contentSeg.id)
|
||||||
|
s.supersededReason = 'pre_tool_content_replaced_by_post_tool_answer'
|
||||||
|
}
|
||||||
|
}
|
||||||
segs.push(contentSeg)
|
segs.push(contentSeg)
|
||||||
flushSegmentsToMessage() // sync once when a new content segment is created
|
flushSegmentsToMessage() // sync once when a new content segment is created
|
||||||
}
|
}
|
||||||
@ -504,6 +515,23 @@ export function useChat(options: UseChatOptions): UseChatReturn {
|
|||||||
}
|
}
|
||||||
})
|
})
|
||||||
|
|
||||||
|
stream.on('segment_kind', (data) => {
|
||||||
|
if (isStaleEvent(data)) return
|
||||||
|
// Producer-assigned semantics of the content span that just closed its
|
||||||
|
// completion. Text streams live before the backend knows whether the
|
||||||
|
// completion carries tool calls, so the kind arrives as this follow-up
|
||||||
|
// event; tag the newest content segment (still running at this point —
|
||||||
|
// tool_call_started closes it afterwards). First writer wins, matching
|
||||||
|
// the persistence side.
|
||||||
|
const kind = typeof data?.kind === 'string' ? data.kind : ''
|
||||||
|
if (!kind) return
|
||||||
|
const seg = currentSegments.value.findLast((s: MessageSegment) => s.type === 'content')
|
||||||
|
if (seg && !seg.kind) {
|
||||||
|
seg.kind = kind
|
||||||
|
flushSegmentsToMessage()
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
stream.on('thinking_delta', (data) => {
|
stream.on('thinking_delta', (data) => {
|
||||||
if (isStaleEvent(data)) return
|
if (isStaleEvent(data)) return
|
||||||
// Suppress thinking display when thinkingLevel=off
|
// Suppress thinking display when thinkingLevel=off
|
||||||
@ -731,6 +759,9 @@ export function useChat(options: UseChatOptions): UseChatReturn {
|
|||||||
if (remote.supersededReason !== undefined) {
|
if (remote.supersededReason !== undefined) {
|
||||||
next.supersededReason = remote.supersededReason
|
next.supersededReason = remote.supersededReason
|
||||||
}
|
}
|
||||||
|
if (next.kind == null && remote.kind != null) {
|
||||||
|
next.kind = remote.kind
|
||||||
|
}
|
||||||
// Server wall-clock bounds are authoritative for durations
|
// Server wall-clock bounds are authoritative for durations
|
||||||
// ("thought for Ns"). Local segments often miss endTimestamp:
|
// ("thought for Ns"). Local segments often miss endTimestamp:
|
||||||
// round-boundary closes flip status without stamping an end,
|
// round-boundary closes flip status without stamping an end,
|
||||||
|
|||||||
@ -27,6 +27,9 @@ export type SSEEventType =
|
|||||||
| 'tool_approval_resolved'
|
| 'tool_approval_resolved'
|
||||||
// 恢复/警告事件
|
// 恢复/警告事件
|
||||||
| 'warning'
|
| 'warning'
|
||||||
|
// Producer-assigned content semantics of the span that just closed its
|
||||||
|
// completion (pre_tool_narration / grounded_narration / final_answer)
|
||||||
|
| 'segment_kind'
|
||||||
// Interrupt + Queue 事件
|
// Interrupt + Queue 事件
|
||||||
| 'heartbeat'
|
| 'heartbeat'
|
||||||
| 'turn_interrupt_requested'
|
| 'turn_interrupt_requested'
|
||||||
|
|||||||
@ -265,6 +265,14 @@ export interface MessageSegment {
|
|||||||
repetitionWarning?: 'char_pattern' | 'sentence_repetition'
|
repetitionWarning?: 'char_pattern' | 'sentence_repetition'
|
||||||
/** Number of trailing characters dropped when the repetition guard fired. */
|
/** Number of trailing characters dropped when the repetition guard fired. */
|
||||||
truncatedChars?: number
|
truncatedChars?: number
|
||||||
|
/**
|
||||||
|
* Producer-assigned content semantics from the backend agent graph:
|
||||||
|
* 'pre_tool_narration' (provisional — text emitted alongside tool calls
|
||||||
|
* before any observation this turn), 'grounded_narration', or
|
||||||
|
* 'final_answer'. Delivered live via the segment_kind SSE event and
|
||||||
|
* persisted in metadata.segments; absent on legacy messages.
|
||||||
|
*/
|
||||||
|
kind?: string
|
||||||
/** Backend marked this model-predicted tool result as replaced by a later actual tool result. */
|
/** Backend marked this model-predicted tool result as replaced by a later actual tool result. */
|
||||||
superseded?: boolean
|
superseded?: boolean
|
||||||
/** Segment ID that replaced this pre-tool prediction. */
|
/** Segment ID that replaced this pre-tool prediction. */
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user