mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 03:13:41 +08:00
feat(agent): producer-assigned content-kind tag on stream deltas
This commit is contained in:
matevip
parent
a578c8ef0e
commit
547adb30ee
@ -764,22 +764,42 @@ public class AgentService {
|
||||
// ==================== StreamDelta ====================
|
||||
|
||||
public record StreamDelta(String content, String thinking, String eventType, Map<String, Object> eventData,
|
||||
boolean persistenceOnly, boolean segmentOnly) {
|
||||
boolean persistenceOnly, boolean segmentOnly, ContentKind kind) {
|
||||
|
||||
// 兼容构造器（广播+持久化）
|
||||
public StreamDelta(String content, String thinking) {
|
||||
this(content, thinking, null, null, false, false);
|
||||
this(content, thinking, null, null, false, false, null);
|
||||
}
|
||||
|
||||
// 显式 5-参构造器：保留旧调用点对 (content, thinking, eventType, eventData, persistenceOnly) 的兼容
|
||||
public StreamDelta(String content, String thinking, String eventType,
|
||||
Map<String, Object> eventData, boolean persistenceOnly) {
|
||||
this(content, thinking, eventType, eventData, persistenceOnly, false);
|
||||
this(content, thinking, eventType, eventData, persistenceOnly, false, null);
|
||||
}
|
||||
|
||||
// 兼容构造器：kind 出现之前的 6 参 canonical 形态
|
||||
public StreamDelta(String content, String thinking, String eventType,
|
||||
Map<String, Object> eventData, boolean persistenceOnly, boolean segmentOnly) {
|
||||
this(content, thinking, eventType, eventData, persistenceOnly, segmentOnly, null);
|
||||
}
|
||||
|
||||
/** 仅用于持久化，不再广播（内容已由 NodeStreamingChatHelper 实时广播过） */
|
||||
public static StreamDelta persistOnly(String content, String thinking) {
|
||||
return new StreamDelta(content, thinking, null, null, true, false);
|
||||
return new StreamDelta(content, thinking, null, null, true, false, null);
|
||||
}
|
||||
|
||||
/** {@link #persistOnly(String, String)} 带内容语义标注的变体。 */
|
||||
public static StreamDelta persistOnly(String content, String thinking, ContentKind kind) {
|
||||
return new StreamDelta(content, thinking, null, null, true, false, kind);
|
||||
}
|
||||
|
||||
/**
|
||||
* Final-answer content of the terminal turn. {@code alreadyStreamed}
|
||||
* decides broadcast suppression exactly like the persistOnly/plain
|
||||
* split at the emission sites did before the kind tag existed.
|
||||
*/
|
||||
public static StreamDelta finalAnswer(String content, boolean alreadyStreamed) {
|
||||
return new StreamDelta(content, null, null, null, alreadyStreamed, false, ContentKind.FINAL_ANSWER);
|
||||
}
|
||||
|
||||
/**
|
||||
@ -803,15 +823,20 @@ public class AgentService {
|
||||
* persisted content field via this flavor.
|
||||
*/
|
||||
public static StreamDelta segmentOnly(String content, String thinking) {
|
||||
return new StreamDelta(content, thinking, null, null, true, true);
|
||||
return new StreamDelta(content, thinking, null, null, true, true, null);
|
||||
}
|
||||
|
||||
/** {@link #segmentOnly(String, String)} 带内容语义标注的变体。 */
|
||||
public static StreamDelta segmentOnly(String content, String thinking, ContentKind kind) {
|
||||
return new StreamDelta(content, thinking, null, null, true, true, kind);
|
||||
}
|
||||
|
||||
public static StreamDelta empty() {
|
||||
return new StreamDelta(null, null, null, null, false, false);
|
||||
return new StreamDelta(null, null, null, null, false, false, null);
|
||||
}
|
||||
|
||||
public static StreamDelta event(String type, Map<String, Object> data) {
|
||||
return new StreamDelta(null, null, type, data, false, false);
|
||||
return new StreamDelta(null, null, type, data, false, false, null);
|
||||
}
|
||||
|
||||
public boolean isEvent() {
|
||||
|
||||
@ -0,0 +1,42 @@
|
||||
package vip.mate.agent;
|
||||
|
||||
import java.util.Locale;
|
||||
|
||||
/**
|
||||
* Semantic category of a content-bearing stream delta, assigned once at the
|
||||
* producer (the agent graph) where the classification inputs — whether the
|
||||
* completion carried tool calls and whether any tool observation preceded the
|
||||
* text this turn — are definitively known.
|
||||
*
|
||||
* <p>Downstream consumers (web segment persistence, IM channel adapters, the
|
||||
* SSE client) MUST read this tag instead of re-deriving the category from
|
||||
* stream structure. Deltas from producers that predate this tag carry
|
||||
* {@code null}; consumers fall back to their legacy structural handling in
|
||||
* that case.
|
||||
*/
|
||||
public enum ContentKind {
|
||||
|
||||
/**
|
||||
* Text emitted in a completion that also carries tool calls, before any
|
||||
* tool observation this turn. Not grounded in this turn's results — it may
|
||||
* be process narration or a fully fabricated "rehearsal" of the outcome.
|
||||
* Provisional: replaced by the next content of the same turn if one
|
||||
* arrives, kept only when the turn produces no later content at all.
|
||||
*/
|
||||
PRE_TOOL_NARRATION,
|
||||
|
||||
/**
|
||||
* Intermediate narration emitted after at least one tool observation this
|
||||
* turn (even when the same completion issues further tool calls). Grounded
|
||||
* in real results; never replaced.
|
||||
*/
|
||||
GROUNDED_NARRATION,
|
||||
|
||||
/** Final-answer text of the terminal turn. */
|
||||
FINAL_ANSWER;
|
||||
|
||||
/** Stable lower-case token used in persisted segment metadata and SSE payloads. */
|
||||
public String wireName() {
|
||||
return name().toLowerCase(Locale.ROOT);
|
||||
}
|
||||
}
|
||||
@ -13,6 +13,7 @@ import reactor.core.publisher.Mono;
|
||||
import vip.mate.agent.AgentService;
|
||||
import vip.mate.agent.AgentState;
|
||||
import vip.mate.agent.BaseAgent;
|
||||
import vip.mate.agent.ContentKind;
|
||||
import vip.mate.agent.delegation.DelegatedUsageAccumulator;
|
||||
import vip.mate.agent.GraphEventPublisher;
|
||||
import vip.mate.agent.StructuredStreamCapable;
|
||||
@ -251,15 +252,16 @@ public class StateGraphReActAgent extends BaseAgent implements StructuredStreamC
|
||||
String streamed = output.state().<String>value(STREAMED_CONTENT).orElse("");
|
||||
if (!streamed.isEmpty() && !streamed.equals(lastEmittedStreamedContent.get())) {
|
||||
lastEmittedStreamedContent.set(streamed);
|
||||
deltas.add(streamedContentDelta(isFinalAnswerTurn, streamed));
|
||||
deltas.add(streamedContentDelta(isFinalAnswerTurn,
|
||||
output.state().value(NEEDS_TOOL_CALL, false),
|
||||
output.state().value(CURRENT_ITERATION, 0),
|
||||
streamed));
|
||||
}
|
||||
|
||||
if (isFinalAnswerTurn && finalAnswerEmitted.compareAndSet(false, true)) {
|
||||
String answer = extractFinalAnswer(output);
|
||||
if (answer != null && !answer.isEmpty()) {
|
||||
deltas.add(contentAlreadyStreamed
|
||||
? AgentService.StreamDelta.persistOnly(answer, null)
|
||||
: new AgentService.StreamDelta(answer, null));
|
||||
deltas.add(AgentService.StreamDelta.finalAnswer(answer, contentAlreadyStreamed));
|
||||
}
|
||||
}
|
||||
String thinking = extractFinalThinking(output);
|
||||
@ -422,15 +424,16 @@ public class StateGraphReActAgent extends BaseAgent implements StructuredStreamC
|
||||
String streamed = output.state().<String>value(STREAMED_CONTENT).orElse("");
|
||||
if (!streamed.isEmpty() && !streamed.equals(lastEmittedStreamedContent.get())) {
|
||||
lastEmittedStreamedContent.set(streamed);
|
||||
deltas.add(streamedContentDelta(isFinalAnswerTurn, streamed));
|
||||
deltas.add(streamedContentDelta(isFinalAnswerTurn,
|
||||
output.state().value(NEEDS_TOOL_CALL, false),
|
||||
output.state().value(CURRENT_ITERATION, 0),
|
||||
streamed));
|
||||
}
|
||||
|
||||
if (isFinalAnswerTurn && finalAnswerEmitted.compareAndSet(false, true)) {
|
||||
String answer = extractFinalAnswer(output);
|
||||
if (answer != null && !answer.isEmpty()) {
|
||||
deltas.add(contentAlreadyStreamed
|
||||
? AgentService.StreamDelta.persistOnly(answer, null)
|
||||
: new AgentService.StreamDelta(answer, null));
|
||||
deltas.add(AgentService.StreamDelta.finalAnswer(answer, contentAlreadyStreamed));
|
||||
}
|
||||
}
|
||||
|
||||
@ -641,15 +644,37 @@ public class StateGraphReActAgent extends BaseAgent implements StructuredStreamC
|
||||
* renderers (copy / TTS / history reload) showing the full text.</li>
|
||||
* </ul>
|
||||
*
|
||||
* <p>Beyond flavor, this is the single assignment point for the delta's
|
||||
* {@link ContentKind}: the graph is the only layer that definitively knows
|
||||
* whether the completion carried tool calls ({@code NEEDS_TOOL_CALL}) and
|
||||
* whether any tool observation preceded the text this turn
|
||||
* ({@code CURRENT_ITERATION} — ObservationNode increments it after each
|
||||
* observed round, so iteration 0 means "no observation yet"). Downstream
|
||||
* consumers read the tag instead of re-deriving it from stream structure.
|
||||
*
|
||||
* <ul>
|
||||
* <li>terminal turn → {@code FINAL_ANSWER};</li>
|
||||
* <li>completion carries tool calls and no observation happened yet this
|
||||
* turn → {@code PRE_TOOL_NARRATION} (provisional, may be replaced by
|
||||
* the turn's next content);</li>
|
||||
* <li>otherwise → {@code GROUNDED_NARRATION} (follows an observation, or
|
||||
* closed its completion without tool calls — never replaced).</li>
|
||||
* </ul>
|
||||
*
|
||||
* <p>Package-private so the unit test can pin the decision without standing
|
||||
* up a full StateGraph fixture. Returning {@code null} for blank input is the
|
||||
* caller's responsibility — this helper just decides flavor for non-blank
|
||||
* content.
|
||||
*/
|
||||
static AgentService.StreamDelta streamedContentDelta(boolean isFinalAnswerTurn, String streamed) {
|
||||
return isFinalAnswerTurn
|
||||
? AgentService.StreamDelta.persistOnly(streamed, null)
|
||||
: AgentService.StreamDelta.segmentOnly(streamed, null);
|
||||
static AgentService.StreamDelta streamedContentDelta(boolean isFinalAnswerTurn, boolean carriesToolCalls,
|
||||
int iteration, String streamed) {
|
||||
if (isFinalAnswerTurn) {
|
||||
return AgentService.StreamDelta.persistOnly(streamed, null, ContentKind.FINAL_ANSWER);
|
||||
}
|
||||
ContentKind kind = carriesToolCalls && iteration == 0
|
||||
? ContentKind.PRE_TOOL_NARRATION
|
||||
: ContentKind.GROUNDED_NARRATION;
|
||||
return AgentService.StreamDelta.segmentOnly(streamed, null, kind);
|
||||
}
|
||||
|
||||
private boolean hasFinalAnswer(NodeOutput output) {
|
||||
|
||||
@ -192,13 +192,20 @@ public final class AgentStreamAccumulator {
|
||||
}
|
||||
// segments: 追加到当前 running content segment，或创建新的
|
||||
var seg = findLastRunning("content");
|
||||
if (seg != null) {
|
||||
seg.put("text", seg.getOrDefault("text", "") + delta.content());
|
||||
} else {
|
||||
if (seg == null) {
|
||||
finalizeRunningSegments("thinking");
|
||||
var s = newSegment("content");
|
||||
s.put("text", delta.content());
|
||||
segments.add(s);
|
||||
seg = newSegment("content");
|
||||
seg.put("text", delta.content());
|
||||
segments.add(seg);
|
||||
} else {
|
||||
seg.put("text", seg.getOrDefault("text", "") + delta.content());
|
||||
}
|
||||
// Producer-assigned content semantics (first writer wins — a
|
||||
// segment never legitimately changes kind mid-flight). Absent on
|
||||
// deltas from producers that predate the tag; consumers fall back
|
||||
// to structural detection for such segments.
|
||||
if (delta.kind() != null && !seg.containsKey("kind")) {
|
||||
seg.put("kind", delta.kind().wireName());
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -3,8 +3,11 @@ package vip.mate.agent.graph;
|
||||
import org.junit.jupiter.api.DisplayName;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import vip.mate.agent.AgentService;
|
||||
import vip.mate.agent.ContentKind;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
/**
|
||||
@ -21,49 +24,94 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
* read {@code content}, not segments, so {@code segmentOnly} for that case would
|
||||
* shrink the visible message to the warning alone.
|
||||
*
|
||||
* <p>The helper under test embodies the corrected contract.
|
||||
* <p>The helper is also the single assignment point of {@link ContentKind}: the
|
||||
* graph knows definitively whether the completion carried tool calls and whether
|
||||
* any observation preceded the text this turn, so downstream consumers read the
|
||||
* tag instead of re-deriving the category from stream structure.
|
||||
*/
|
||||
class StateGraphReActAgentStreamedContentDeltaTest {
|
||||
|
||||
@Test
|
||||
@DisplayName("intermediate iteration (no FINAL_ANSWER yet) → segmentOnly — narration stays out of content")
|
||||
void intermediateIteration_routedToSegmentsOnly() {
|
||||
@DisplayName("iteration 0 with tool calls → segmentOnly + PRE_TOOL_NARRATION — provisional rehearsal")
|
||||
void preToolNarration_taggedProvisional() {
|
||||
AgentService.StreamDelta d = StateGraphReActAgent.streamedContentDelta(
|
||||
/* isFinalAnswerTurn */ false,
|
||||
"I'll search for X.");
|
||||
/* carriesToolCalls */ true,
|
||||
/* iteration */ 0,
|
||||
"先加载 skill，然后逐个查询。");
|
||||
|
||||
assertTrue(d.persistenceOnly(),
|
||||
"segmentOnly implies persistenceOnly — no re-broadcast (NodeStreamingChatHelper already pushed it)");
|
||||
assertTrue(d.segmentOnly(),
|
||||
"intermediate narration MUST set segmentOnly so content.append is skipped");
|
||||
// Sanity: the content payload survives the wrap.
|
||||
org.junit.jupiter.api.Assertions.assertEquals("I'll search for X.", d.content());
|
||||
assertEquals(ContentKind.PRE_TOOL_NARRATION, d.kind(),
|
||||
"text alongside tool calls with zero observations this turn is not grounded — provisional");
|
||||
assertEquals("先加载 skill，然后逐个查询。", d.content());
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("evidence-insufficient terminal turn (FINAL_ANSWER set) → persistOnly — answer body persists to content")
|
||||
@DisplayName("iteration ≥1 narration → segmentOnly + GROUNDED_NARRATION even when the completion issues more tool calls")
|
||||
void postObservationNarration_taggedGrounded() {
|
||||
AgentService.StreamDelta d = StateGraphReActAgent.streamedContentDelta(
|
||||
false, /* carriesToolCalls */ true, /* iteration */ 1, "第一间空闲，继续查下一间。");
|
||||
|
||||
assertTrue(d.segmentOnly());
|
||||
assertEquals(ContentKind.GROUNDED_NARRATION, d.kind(),
|
||||
"an observation already happened this turn — the narration is grounded and never replaced");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("iteration 0 without tool calls (non-terminal) → GROUNDED_NARRATION — conservative, never replaced")
|
||||
void noToolCallCompletion_taggedGrounded() {
|
||||
AgentService.StreamDelta d = StateGraphReActAgent.streamedContentDelta(
|
||||
false, /* carriesToolCalls */ false, /* iteration */ 0, "narrative without tools");
|
||||
|
||||
assertEquals(ContentKind.GROUNDED_NARRATION, d.kind(),
|
||||
"a completion that closed without tool calls has no later observation to defer to — keep it");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("evidence-insufficient terminal turn (FINAL_ANSWER set) → persistOnly + FINAL_ANSWER kind")
|
||||
void evidenceInsufficientFinalTurn_routedToPersistOnly() {
|
||||
// Regression: STREAMED_CONTENT here is the rejected answer body; FINAL_ANSWER
|
||||
// is only the "[证据不足]" warning. Persisting the streamed body keeps
|
||||
// mate_message.content readable through single-segment renderers.
|
||||
AgentService.StreamDelta d = StateGraphReActAgent.streamedContentDelta(
|
||||
/* isFinalAnswerTurn */ true,
|
||||
/* isFinalAnswerTurn */ true, false, 2,
|
||||
"The answer is 42. References: [1] [2] [3].");
|
||||
|
||||
assertTrue(d.persistenceOnly(),
|
||||
"persistOnly suppresses re-broadcast — content was already streamed live");
|
||||
assertFalse(d.segmentOnly(),
|
||||
"persistOnly variant MUST NOT set segmentOnly — content.append needs to run");
|
||||
org.junit.jupiter.api.Assertions.assertEquals(
|
||||
"The answer is 42. References: [1] [2] [3].", d.content());
|
||||
assertEquals(ContentKind.FINAL_ANSWER, d.kind());
|
||||
assertEquals("The answer is 42. References: [1] [2] [3].", d.content());
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("finalAnswer factory carries FINAL_ANSWER kind in both broadcast flavors")
|
||||
void finalAnswerFactory_taggedFinal() {
|
||||
AgentService.StreamDelta streamed = AgentService.StreamDelta.finalAnswer("done", true);
|
||||
assertTrue(streamed.persistenceOnly(), "already-streamed answer must not re-broadcast");
|
||||
assertEquals(ContentKind.FINAL_ANSWER, streamed.kind());
|
||||
|
||||
AgentService.StreamDelta fresh = AgentService.StreamDelta.finalAnswer("done", false);
|
||||
assertFalse(fresh.persistenceOnly(), "un-streamed answer still broadcasts");
|
||||
assertEquals(ContentKind.FINAL_ANSWER, fresh.kind());
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("legacy factories keep kind null — pre-tag producers stay distinguishable")
|
||||
void legacyFactories_kindNull() {
|
||||
assertNull(AgentService.StreamDelta.segmentOnly("x", null).kind());
|
||||
assertNull(AgentService.StreamDelta.persistOnly("x", null).kind());
|
||||
assertNull(new AgentService.StreamDelta("x", null).kind());
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("both flavors leave thinking null — STREAMED_CONTENT routing only carries text content")
|
||||
void thinkingFieldNeverSet() {
|
||||
org.junit.jupiter.api.Assertions.assertNull(
|
||||
StateGraphReActAgent.streamedContentDelta(false, "x").thinking());
|
||||
org.junit.jupiter.api.Assertions.assertNull(
|
||||
StateGraphReActAgent.streamedContentDelta(true, "x").thinking());
|
||||
assertNull(StateGraphReActAgent.streamedContentDelta(false, true, 0, "x").thinking());
|
||||
assertNull(StateGraphReActAgent.streamedContentDelta(true, false, 1, "x").thinking());
|
||||
}
|
||||
}
|
||||
|
||||
@ -0,0 +1,101 @@
|
||||
package vip.mate.channel.web;
|
||||
|
||||
import com.fasterxml.jackson.databind.JsonNode;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import org.junit.jupiter.api.DisplayName;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import vip.mate.agent.AgentService.StreamDelta;
|
||||
import vip.mate.agent.ContentKind;
|
||||
|
||||
import java.util.Map;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
/**
|
||||
* Pins that producer-assigned {@link ContentKind} tags survive segment
|
||||
* accumulation into persisted {@code metadata.segments}, and that untagged
|
||||
* deltas (pre-tag producers) leave the field absent so consumers can fall
|
||||
* back to structural detection.
|
||||
*/
|
||||
class AgentStreamAccumulatorKindTest {
|
||||
|
||||
private static final AgentStreamAccumulator.Sink NOOP_SINK = new AgentStreamAccumulator.Sink() {
|
||||
@Override public void broadcast(String conversationId, String eventName, Object payload) { }
|
||||
@Override public void updatePhase(String conversationId, String phase) { }
|
||||
};
|
||||
|
||||
private static JsonNode segmentsOf(AgentStreamAccumulator acc, ObjectMapper mapper) throws Exception {
|
||||
return mapper.readTree(acc.toMetadataJson()).path("segments");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("kind tags land on persisted content segments; tool segments stay untagged")
|
||||
void kindTagsPersisted() throws Exception {
|
||||
ObjectMapper mapper = new ObjectMapper();
|
||||
AgentStreamAccumulator acc = new AgentStreamAccumulator(mapper, NOOP_SINK);
|
||||
String cid = "conv-1";
|
||||
|
||||
acc.accept(StreamDelta.segmentOnly("先查询两间会议室的环境数据。", null,
|
||||
ContentKind.PRE_TOOL_NARRATION), cid);
|
||||
acc.accept(StreamDelta.event("tool_call_started",
|
||||
Map.of("toolCallId", "t1", "toolName", "envQuery", "arguments", "{}")), cid);
|
||||
acc.accept(StreamDelta.event("tool_call_completed",
|
||||
Map.of("toolCallId", "t1", "toolName", "envQuery", "result", "data={}", "success", true)), cid);
|
||||
acc.accept(StreamDelta.finalAnswer("接口返回为空，无环境数据。", true), cid);
|
||||
|
||||
JsonNode segments = segmentsOf(acc, mapper);
|
||||
assertEquals(3, segments.size(), "content + tool_call + content");
|
||||
assertEquals("pre_tool_narration", segments.get(0).path("kind").asText());
|
||||
assertEquals("tool_call", segments.get(1).path("type").asText());
|
||||
assertFalse(segments.get(1).has("kind"), "kind is content-segment semantics only");
|
||||
assertEquals("final_answer", segments.get(2).path("kind").asText());
|
||||
|
||||
assertEquals("接口返回为空，无环境数据。", acc.getContent(),
|
||||
"segmentOnly narration must stay out of the persisted top-level content");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("untagged deltas leave kind absent — legacy consumers keep structural fallback")
|
||||
void untaggedDeltaHasNoKind() throws Exception {
|
||||
ObjectMapper mapper = new ObjectMapper();
|
||||
AgentStreamAccumulator acc = new AgentStreamAccumulator(mapper, NOOP_SINK);
|
||||
|
||||
acc.accept(StreamDelta.segmentOnly("legacy narration", null), "conv-2");
|
||||
|
||||
JsonNode segments = segmentsOf(acc, mapper);
|
||||
assertEquals(1, segments.size());
|
||||
assertFalse(segments.get(0).has("kind"));
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("first writer wins — appended deltas cannot re-kind a running segment")
|
||||
void firstWriterWins() throws Exception {
|
||||
ObjectMapper mapper = new ObjectMapper();
|
||||
AgentStreamAccumulator acc = new AgentStreamAccumulator(mapper, NOOP_SINK);
|
||||
String cid = "conv-3";
|
||||
|
||||
acc.accept(StreamDelta.segmentOnly("part one ", null, ContentKind.GROUNDED_NARRATION), cid);
|
||||
acc.accept(StreamDelta.segmentOnly("part two", null, ContentKind.PRE_TOOL_NARRATION), cid);
|
||||
|
||||
JsonNode segments = segmentsOf(acc, mapper);
|
||||
assertEquals(1, segments.size(), "second delta appends into the running segment");
|
||||
assertEquals("grounded_narration", segments.get(0).path("kind").asText());
|
||||
assertTrue(segments.get(0).path("text").asText().endsWith("part two"));
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("kind fills in late when the segment opener was untagged")
|
||||
void lateKindFillsUntaggedSegment() throws Exception {
|
||||
ObjectMapper mapper = new ObjectMapper();
|
||||
AgentStreamAccumulator acc = new AgentStreamAccumulator(mapper, NOOP_SINK);
|
||||
String cid = "conv-4";
|
||||
|
||||
acc.accept(StreamDelta.segmentOnly("opener ", null), cid);
|
||||
acc.accept(StreamDelta.segmentOnly("tail", null, ContentKind.GROUNDED_NARRATION), cid);
|
||||
|
||||
JsonNode segments = segmentsOf(acc, mapper);
|
||||
assertEquals("grounded_narration", segments.get(0).path("kind").asText());
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user