diff --git a/easyflow-commons/easyflow-common-chat-protocol/pom.xml b/easyflow-commons/easyflow-common-chat-protocol/pom.xml index 0bb86dbc..dd1e1bc4 100644 --- a/easyflow-commons/easyflow-common-chat-protocol/pom.xml +++ b/easyflow-commons/easyflow-common-chat-protocol/pom.xml @@ -25,6 +25,12 @@ org.springframework.boot spring-boot-starter-websocket + + junit + junit + ${junit.version} + test + - \ No newline at end of file + diff --git a/easyflow-commons/easyflow-common-chat-protocol/src/main/java/tech/easyflow/core/runtime/LegacyThinkingTagParser.java b/easyflow-commons/easyflow-common-chat-protocol/src/main/java/tech/easyflow/core/runtime/LegacyThinkingTagParser.java new file mode 100644 index 00000000..8a6ad208 --- /dev/null +++ b/easyflow-commons/easyflow-common-chat-protocol/src/main/java/tech/easyflow/core/runtime/LegacyThinkingTagParser.java @@ -0,0 +1,286 @@ +package tech.easyflow.core.runtime; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Locale; + +/** + * 将旧模型混在 {@code content} 中的 {@code } 或 {@code } 内容拆分为思考与正文增量。 + * + *

解析仅在单轮回复开头生效，避免误处理正文中的标签示例。该解析器保留少量标签前缀， + * 因而可以正确处理开始或结束标签被拆分到多个流式增量中的情况。

+ */ +public final class LegacyThinkingTagParser { + + private static final String[] OPEN_TAGS = {"", ""}; + private static final String[] CLOSE_TAGS = {"", ""}; + private static final int MAX_LEADING_WHITESPACE = 64; + + private final StringBuilder pending = new StringBuilder(); + private Mode mode = Mode.UNDECIDED; + private boolean legacyFormatDetected; + private String activeCloseTag; + + /** + * 接收普通正文增量并按需拆分旧版思考标签。 + * + * @param delta 普通正文增量 + * @return 可立即发送的思考或正文片段 + */ + public List acceptContent(String delta) { + if (delta == null || delta.isEmpty()) { + return Collections.emptyList(); + } + if (mode == Mode.CONTENT || mode == Mode.BYPASS) { + return List.of(Segment.content(delta)); + } + if (mode == Mode.THINKING) { + return consumeThinking(delta); + } + pending.append(delta); + return resolveUndecided(); + } + + /** + * 接收模型已经结构化返回的思考增量，非空时关闭旧标签自动识别。 + * + * @param delta 结构化思考增量 + * @return 待发送片段，包含必要的前置缓冲与当前思考增量 + */ + public List acceptReasoning(String delta) { + if (delta == null || delta.isBlank()) { + return Collections.emptyList(); + } + List segments = new ArrayList<>(); + flushBeforeStructuredReasoning(segments); + mode = Mode.BYPASS; + addSegment(segments, SegmentType.REASONING, delta); + return segments; + } + + /** + * 收口尚未发送的标签前缀或内容。 + * + * @return 剩余的思考或正文片段 + */ + public List finish() { + if (pending.length() == 0) { + return Collections.emptyList(); + } + SegmentType type = mode == Mode.THINKING ? SegmentType.REASONING : SegmentType.CONTENT; + String text = pending.toString(); + pending.setLength(0); + mode = Mode.CONTENT; + return List.of(new Segment(type, text)); + } + + /** + * 重置单轮解析状态。 + */ + public void reset() { + pending.setLength(0); + mode = Mode.UNDECIDED; + legacyFormatDetected = false; + activeCloseTag = null; + } + + /** + * 返回当前轮次是否识别到旧版思考标签。 + * + * @return 识别到回复开头的旧思考标签时为 {@code true} + */ + public boolean isLegacyFormatDetected() { + return legacyFormatDetected; + } + + private List resolveUndecided() { + int contentStart = firstContentIndex(pending); + if (contentStart == pending.length()) { + if (pending.length() <= MAX_LEADING_WHITESPACE) { + return Collections.emptyList(); + } + mode = Mode.CONTENT; + return drainPending(SegmentType.CONTENT); + } + + String candidate = pending.substring(contentStart); + String normalized = candidate.toLowerCase(Locale.ROOT); + int matchedTagIndex = matchingOpenTagIndex(normalized); + if (matchedTagIndex < 0 && isPossibleOpenTagPrefix(normalized)) { + return Collections.emptyList(); + } + if (matchedTagIndex < 0) { + mode = Mode.CONTENT; + return drainPending(SegmentType.CONTENT); + } + + legacyFormatDetected = true; + mode = Mode.THINKING; + activeCloseTag = CLOSE_TAGS[matchedTagIndex]; + String remainder = candidate.substring(OPEN_TAGS[matchedTagIndex].length()); + pending.setLength(0); + return consumeThinking(remainder); + } + + private List consumeThinking(String delta) { + pending.append(delta); + String normalized = pending.toString().toLowerCase(Locale.ROOT); + int closingIndex = normalized.indexOf(activeCloseTag); + List segments = new ArrayList<>(); + if (closingIndex >= 0) { + addSegment(segments, SegmentType.REASONING, pending.substring(0, closingIndex)); + String remainder = pending.substring(closingIndex + activeCloseTag.length()); + pending.setLength(0); + mode = Mode.CONTENT; + addSegment(segments, SegmentType.CONTENT, remainder); + return segments; + } + + int retainedLength = closingTagPrefixLength(normalized); + int emittedLength = pending.length() - retainedLength; + if (emittedLength > 0) { + addSegment(segments, SegmentType.REASONING, pending.substring(0, emittedLength)); + String retained = pending.substring(emittedLength); + pending.setLength(0); + pending.append(retained); + } + return segments; + } + + private void flushBeforeStructuredReasoning(List segments) { + if (pending.length() == 0) { + return; + } + if (mode == Mode.UNDECIDED && pending.toString().isBlank()) { + pending.setLength(0); + return; + } + SegmentType type = mode == Mode.THINKING ? SegmentType.REASONING : SegmentType.CONTENT; + addSegment(segments, type, pending.toString()); + pending.setLength(0); + } + + private List drainPending(SegmentType type) { + String text = pending.toString(); + pending.setLength(0); + return text.isEmpty() ? Collections.emptyList() : List.of(new Segment(type, text)); + } + + private int firstContentIndex(CharSequence value) { + int index = 0; + while (index < value.length()) { + char current = value.charAt(index); + if (!Character.isWhitespace(current) && current != '\uFEFF') { + break; + } + index++; + } + return index; + } + + private int closingTagPrefixLength(String value) { + int maxLength = Math.min(value.length(), activeCloseTag.length() - 1); + for (int length = maxLength; length > 0; length--) { + if (activeCloseTag.startsWith(value.substring(value.length() - length))) { + return length; + } + } + return 0; + } + + private int matchingOpenTagIndex(String value) { + for (int index = 0; index < OPEN_TAGS.length; index++) { + if (value.startsWith(OPEN_TAGS[index])) { + return index; + } + } + return -1; + } + + private boolean isPossibleOpenTagPrefix(String value) { + for (String openTag : OPEN_TAGS) { + if (openTag.startsWith(value)) { + return true; + } + } + return false; + } + + private void addSegment(List segments, SegmentType type, String text) { + if (text == null || text.isEmpty()) { + return; + } + if (!segments.isEmpty() && segments.get(segments.size() - 1).getType() == type) { + Segment previous = segments.remove(segments.size() - 1); + segments.add(new Segment(type, previous.getText() + text)); + return; + } + segments.add(new Segment(type, text)); + } + + private enum Mode { + UNDECIDED, + THINKING, + CONTENT, + BYPASS + } + + /** + * 兼容解析后的片段类型。 + */ + public enum SegmentType { + /** 思考增量。 */ + REASONING, + /** 最终回答增量。 */ + CONTENT + } + + /** + * 兼容解析后的不可变文本片段。 + */ + public static final class Segment { + + private final SegmentType type; + private final String text; + + /** + * 创建解析片段。 + * + * @param type 片段类型 + * @param text 片段文本 + */ + public Segment(SegmentType type, String text) { + this.type = type; + this.text = text; + } + + /** + * 创建正文片段。 + * + * @param text 正文文本 + * @return 正文片段 + */ + public static Segment content(String text) { + return new Segment(SegmentType.CONTENT, text); + } + + /** + * 获取片段类型。 + * + * @return 片段类型 + */ + public SegmentType getType() { + return type; + } + + /** + * 获取片段文本。 + * + * @return 片段文本 + */ + public String getText() { + return text; + } + } +} diff --git a/easyflow-commons/easyflow-common-chat-protocol/src/test/java/tech/easyflow/core/runtime/LegacyThinkingTagParserTest.java b/easyflow-commons/easyflow-common-chat-protocol/src/test/java/tech/easyflow/core/runtime/LegacyThinkingTagParserTest.java new file mode 100644 index 00000000..85b802f0 --- /dev/null +++ b/easyflow-commons/easyflow-common-chat-protocol/src/test/java/tech/easyflow/core/runtime/LegacyThinkingTagParserTest.java @@ -0,0 +1,123 @@ +package tech.easyflow.core.runtime; + +import org.junit.Assert; +import org.junit.Test; + +import java.util.ArrayList; +import java.util.List; + +/** + * {@link LegacyThinkingTagParser} 流式兼容测试。 + */ +public class LegacyThinkingTagParserTest { + + /** + * 验证跨增量拆分的开始与结束标签可以正确解析。 + */ + @Test + public void shouldSplitThinkingTagsAcrossChunks() { + LegacyThinkingTagParser parser = new LegacyThinkingTagParser(); + List segments = new ArrayList<>(); + + segments.addAll(parser.acceptContent(" 先分析\n最终回答")); + segments.addAll(parser.finish()); + + Assert.assertTrue(parser.isLegacyFormatDetected()); + Assert.assertEquals("先分析", join(segments, LegacyThinkingTagParser.SegmentType.REASONING)); + Assert.assertEquals("\n最终回答", join(segments, LegacyThinkingTagParser.SegmentType.CONTENT)); + } + + /** + * 验证 {@code } 别名及其跨增量结束标签可以正确解析。 + */ + @Test + public void shouldSplitThinkingAliasAcrossChunks() { + LegacyThinkingTagParser parser = new LegacyThinkingTagParser(); + List segments = new ArrayList<>(); + + segments.addAll(parser.acceptContent("先分析最终回答")); + segments.addAll(parser.finish()); + + Assert.assertTrue(parser.isLegacyFormatDetected()); + Assert.assertEquals("先分析", join(segments, LegacyThinkingTagParser.SegmentType.REASONING)); + Assert.assertEquals("最终回答", join(segments, LegacyThinkingTagParser.SegmentType.CONTENT)); + } + + /** + * 验证普通正文中的标签示例不会被错误拆分。 + */ + @Test + public void shouldKeepThinkTagWhenItIsNotAtResponseStart() { + LegacyThinkingTagParser parser = new LegacyThinkingTagParser(); + List segments = new ArrayList<>(); + + segments.addAll(parser.acceptContent("示例：内容")); + segments.addAll(parser.finish()); + + Assert.assertFalse(parser.isLegacyFormatDetected()); + Assert.assertEquals("示例：内容", + join(segments, LegacyThinkingTagParser.SegmentType.CONTENT)); + Assert.assertEquals("", join(segments, LegacyThinkingTagParser.SegmentType.REASONING)); + } + + /** + * 验证结构化思考协议会旁路旧标签识别。 + */ + @Test + public void shouldBypassLegacyParsingForStructuredReasoning() { + LegacyThinkingTagParser parser = new LegacyThinkingTagParser(); + List segments = new ArrayList<>(); + + segments.addAll(parser.acceptReasoning("结构化思考")); + segments.addAll(parser.acceptContent("正文标签示例")); + + Assert.assertFalse(parser.isLegacyFormatDetected()); + Assert.assertEquals("结构化思考", join(segments, LegacyThinkingTagParser.SegmentType.REASONING)); + Assert.assertEquals("正文标签示例", + join(segments, LegacyThinkingTagParser.SegmentType.CONTENT)); + } + + /** + * 验证空结构化思考不会阻止正文开头的旧标签识别。 + */ + @Test + public void shouldParseLegacyTagWhenStructuredReasoningIsBlank() { + LegacyThinkingTagParser parser = new LegacyThinkingTagParser(); + List segments = new ArrayList<>(); + + segments.addAll(parser.acceptReasoning(" ")); + segments.addAll(parser.acceptContent("旧版思考正文")); + + Assert.assertTrue(parser.isLegacyFormatDetected()); + Assert.assertEquals("旧版思考", join(segments, LegacyThinkingTagParser.SegmentType.REASONING)); + Assert.assertEquals("正文", join(segments, LegacyThinkingTagParser.SegmentType.CONTENT)); + } + + /** + * 验证未闭合的旧思考标签在流结束时仍作为思考内容收口。 + */ + @Test + public void shouldFlushUnclosedThinkingAsReasoning() { + LegacyThinkingTagParser parser = new LegacyThinkingTagParser(); + List segments = new ArrayList<>(); + + segments.addAll(parser.acceptContent("尚未完成 segments, + LegacyThinkingTagParser.SegmentType type) { + return segments.stream() + .filter(segment -> segment.getType() == type) + .map(LegacyThinkingTagParser.Segment::getText) + .reduce("", String::concat); + } +} diff --git a/easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/runtime/AgentRunService.java b/easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/runtime/AgentRunService.java index ee980123..6018a02b 100644 --- a/easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/runtime/AgentRunService.java +++ b/easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/runtime/AgentRunService.java @@ -355,9 +355,10 @@ public class AgentRunService { AtomicBoolean finished = new AtomicBoolean(false); StringBuilder answer = new StringBuilder(); ChatAssistantAccumulator assistantAccumulator = new ChatAssistantAccumulator(); + LegacyThinkingTagParser legacyThinkingTagParser = new LegacyThinkingTagParser(); // 注册 emit 服务 registerEmitterCancellation(requestId, chatSseEmitter, chatContext, answer, - assistantAccumulator, finished, persistChatlog); + assistantAccumulator, legacyThinkingTagParser, finished, persistChatlog); AgentRunLock.Handle lockHandle = initialLockHandle; try { if (persistChatlog) { @@ -394,10 +395,11 @@ public class AgentRunService { owner, lockHandle, event -> handleRuntimeEvent(event, requestId, chatSseEmitter, answer, - assistantAccumulator, chatContext, finished, persistChatlog), - error -> handleRuntimeError(error, requestId, chatSseEmitter, chatContext, finished, persistChatlog), - () -> finishIfNeeded(requestId, chatSseEmitter, chatContext, answer, - assistantAccumulator, finished, persistChatlog) + assistantAccumulator, legacyThinkingTagParser, chatContext, finished, persistChatlog), + error -> handleRuntimeStreamError(error, requestId, chatSseEmitter, chatContext, answer, + assistantAccumulator, legacyThinkingTagParser, finished, persistChatlog), + () -> finishRuntimeStream(requestId, chatSseEmitter, chatContext, answer, + assistantAccumulator, legacyThinkingTagParser, finished, persistChatlog) ); agentRunRegistry.register(runContext); lockHandle = null; @@ -469,10 +471,11 @@ public class AgentRunService { ChatRuntimeContext chatContext, StringBuilder answer, ChatAssistantAccumulator assistantAccumulator, + LegacyThinkingTagParser legacyThinkingTagParser, AtomicBoolean finished, boolean persistChatlog) { Runnable cancelTask = () -> cancelDisconnectedRun(requestId, chatContext, answer, - assistantAccumulator, finished, persistChatlog); + assistantAccumulator, legacyThinkingTagParser, finished, persistChatlog); SseEmitter emitter = chatSseEmitter.getEmitter(); emitter.onCompletion(cancelTask); emitter.onTimeout(cancelTask); @@ -483,6 +486,7 @@ public class AgentRunService { ChatRuntimeContext chatContext, StringBuilder answer, ChatAssistantAccumulator assistantAccumulator, + LegacyThinkingTagParser legacyThinkingTagParser, AtomicBoolean finished, boolean persistChatlog) { if (!finished.compareAndSet(false, true)) { @@ -498,6 +502,7 @@ public class AgentRunService { } agentRunRegistry.remove(requestId); cancelPending(requestId, "客户端连接已断开，Agent 运行已取消", persistChatlog); + appendAssistantSegments(legacyThinkingTagParser.finish(), answer, assistantAccumulator); if (!persistChatlog) { return; } @@ -517,32 +522,39 @@ public class AgentRunService { ChatRuntimeContext chatContext, AtomicBoolean finished, boolean persistChatlog) { + handleRuntimeEvent(event, requestId, chatSseEmitter, answer, assistantAccumulator, + new LegacyThinkingTagParser(), chatContext, finished, persistChatlog); + } + + private void handleRuntimeEvent(AgentRuntimeEvent event, + String requestId, + ChatSseEmitter chatSseEmitter, + StringBuilder answer, + ChatAssistantAccumulator assistantAccumulator, + LegacyThinkingTagParser legacyThinkingTagParser, + ChatRuntimeContext chatContext, + AtomicBoolean finished, + boolean persistChatlog) { if (event == null || event.getEventType() == null) { return; } recordRuntimeEvent(requestId, chatContext, event, persistChatlog); + if (event.getEventType() == AgentRuntimeEventType.REASONING_STARTED) { + emitAssistantSegments(legacyThinkingTagParser.finish(), requestId, chatSseEmitter, chatContext, + answer, assistantAccumulator, legacyThinkingTagParser, finished, persistChatlog); + legacyThinkingTagParser.reset(); + return; + } if (event.getEventType() == AgentRuntimeEventType.MESSAGE_DELTA) { String text = stringPayload(event, "text"); - if (text != null) { - answer.append(text); - assistantAccumulator.appendContent(text); - LOG.debug("Agent runtime message delta, requestId={}, deltaLength={}, answerLength={}, delta={}", - requestId, text.length(), answer.length(), toVisibleLogText(text)); - if (!sendEnvelope(chatSseEmitter, ChatDomain.LLM, ChatType.MESSAGE, Map.of("delta", text, "role", "assistant"))) { - cancelDisconnectedRun(requestId, chatContext, answer, assistantAccumulator, finished, persistChatlog); - } - } + emitAssistantSegments(legacyThinkingTagParser.acceptContent(text), requestId, chatSseEmitter, chatContext, + answer, assistantAccumulator, legacyThinkingTagParser, finished, persistChatlog); return; } if (event.getEventType() == AgentRuntimeEventType.REASONING_DELTA) { - Map payload = new LinkedHashMap<>(); String reasoning = firstText(stringPayload(event, "reasoning"), stringPayload(event, "text")); - assistantAccumulator.appendReasoning(reasoning); - payload.put("reasoning", reasoning); - payload.put("delta", reasoning); - if (!sendEnvelope(chatSseEmitter, ChatDomain.LLM, ChatType.THINKING, payload)) { - cancelDisconnectedRun(requestId, chatContext, answer, assistantAccumulator, finished, persistChatlog); - } + emitAssistantSegments(legacyThinkingTagParser.acceptReasoning(reasoning), requestId, chatSseEmitter, chatContext, + answer, assistantAccumulator, legacyThinkingTagParser, finished, persistChatlog); return; } if (event.getEventType() == AgentRuntimeEventType.TOOL_APPROVAL_REQUIRED) { @@ -550,17 +562,23 @@ public class AgentRunService { agentRunRegistry.registerResumeToken(requestId, resumeToken); recordApprovalRequired(requestId, chatContext, event, persistChatlog); if (!sendEnvelope(chatSseEmitter, ChatDomain.TOOL, ChatType.FORM_REQUEST, buildToolHitlPayload(requestId, event))) { - cancelDisconnectedRun(requestId, chatContext, answer, assistantAccumulator, finished, persistChatlog); + cancelDisconnectedRun(requestId, chatContext, answer, assistantAccumulator, + legacyThinkingTagParser, finished, persistChatlog); } return; } if (isAsyncToolEvent(event.getEventType())) { if (!sendEnvelope(chatSseEmitter, ChatDomain.TOOL, asyncToolChatType(event), buildAsyncToolEventPayload(event))) { - cancelDisconnectedRun(requestId, chatContext, answer, assistantAccumulator, finished, persistChatlog); + cancelDisconnectedRun(requestId, chatContext, answer, assistantAccumulator, + legacyThinkingTagParser, finished, persistChatlog); } return; } if (event.getEventType() == AgentRuntimeEventType.TOOL_CALL) { + if (!emitAssistantSegments(legacyThinkingTagParser.finish(), requestId, chatSseEmitter, chatContext, + answer, assistantAccumulator, legacyThinkingTagParser, finished, persistChatlog)) { + return; + } LOG.info("Agent runtime tool call, requestId={}, toolCallId={}, payload={}, metadata={}", requestId, event.getToolCallId(), event.getPayload(), event.getMetadata()); Map toolPayload = buildToolEventPayload(event); @@ -571,7 +589,8 @@ public class AgentRunService { firstNonNull(toolPayload.get("input"), toolPayload.get("toolInput")) ); if (!sendEnvelope(chatSseEmitter, ChatDomain.TOOL, ChatType.TOOL_CALL, toolPayload)) { - cancelDisconnectedRun(requestId, chatContext, answer, assistantAccumulator, finished, persistChatlog); + cancelDisconnectedRun(requestId, chatContext, answer, assistantAccumulator, + legacyThinkingTagParser, finished, persistChatlog); } return; } @@ -587,15 +606,19 @@ public class AgentRunService { toolPayload.get("text")) ); if (!sendEnvelope(chatSseEmitter, ChatDomain.TOOL, ChatType.TOOL_RESULT, toolPayload)) { - cancelDisconnectedRun(requestId, chatContext, answer, assistantAccumulator, finished, persistChatlog); + cancelDisconnectedRun(requestId, chatContext, answer, assistantAccumulator, + legacyThinkingTagParser, finished, persistChatlog); + return; } + legacyThinkingTagParser.reset(); return; } if (event.getEventType() == AgentRuntimeEventType.KNOWLEDGE_RETRIEVAL) { LOG.info("Agent runtime knowledge retrieval, requestId={}, payload={}, metadata={}", requestId, event.getPayload(), event.getMetadata()); if (!sendEnvelope(chatSseEmitter, ChatDomain.BUSINESS, ChatType.STATUS, buildKnowledgeRetrievalStatusPayload(event))) { - cancelDisconnectedRun(requestId, chatContext, answer, assistantAccumulator, finished, persistChatlog); + cancelDisconnectedRun(requestId, chatContext, answer, assistantAccumulator, + legacyThinkingTagParser, finished, persistChatlog); } return; } @@ -604,7 +627,8 @@ public class AgentRunService { LOG.info("Agent runtime memory compression, requestId={}, eventType={}, payload={}, metadata={}", requestId, event.getEventType(), event.getPayload(), event.getMetadata()); if (!sendEnvelope(chatSseEmitter, ChatDomain.BUSINESS, ChatType.STATUS, event.getPayload())) { - cancelDisconnectedRun(requestId, chatContext, answer, assistantAccumulator, finished, persistChatlog); + cancelDisconnectedRun(requestId, chatContext, answer, assistantAccumulator, + legacyThinkingTagParser, finished, persistChatlog); } return; } @@ -612,7 +636,8 @@ public class AgentRunService { LOG.info("Agent runtime suspended, requestId={}, payload={}, metadata={}", requestId, event.getPayload(), event.getMetadata()); if (!sendEnvelope(chatSseEmitter, ChatDomain.BUSINESS, ChatType.STATUS, buildSuspendedStatusPayload(event))) { - cancelDisconnectedRun(requestId, chatContext, answer, assistantAccumulator, finished, persistChatlog); + cancelDisconnectedRun(requestId, chatContext, answer, assistantAccumulator, + legacyThinkingTagParser, finished, persistChatlog); return; } AgentRunRegistry.AgentRunContext runContext = agentRunRegistry.get(requestId); @@ -622,15 +647,20 @@ public class AgentRunService { return; } if (event.getEventType() == AgentRuntimeEventType.COMPLETED) { + if (!emitAssistantSegments(legacyThinkingTagParser.finish(), requestId, chatSseEmitter, chatContext, + answer, assistantAccumulator, legacyThinkingTagParser, finished, persistChatlog)) { + return; + } String finalText = stringPayload(event, "text"); - if (finalText != null && !finalText.isBlank()) { + if (!legacyThinkingTagParser.isLegacyFormatDetected() && finalText != null && !finalText.isBlank()) { answer.setLength(0); answer.append(finalText); } List> citations = buildKnowledgeCitationPayload(event); if (!citations.isEmpty()) { if (!sendEnvelope(chatSseEmitter, ChatDomain.BUSINESS, ChatType.CITATIONS, Map.of("items", citations))) { - cancelDisconnectedRun(requestId, chatContext, answer, assistantAccumulator, finished, persistChatlog); + cancelDisconnectedRun(requestId, chatContext, answer, assistantAccumulator, + legacyThinkingTagParser, finished, persistChatlog); return; } } @@ -639,15 +669,150 @@ public class AgentRunService { return; } if (event.getEventType() == AgentRuntimeEventType.CANCELLED) { + if (!emitAssistantSegments(legacyThinkingTagParser.finish(), requestId, chatSseEmitter, chatContext, + answer, assistantAccumulator, legacyThinkingTagParser, finished, persistChatlog)) { + return; + } handleRuntimeCancelled(event, requestId, chatSseEmitter, chatContext, answer, assistantAccumulator, finished, persistChatlog); return; } if (event.getEventType() == AgentRuntimeEventType.FAILED) { + if (!emitAssistantSegments(legacyThinkingTagParser.finish(), requestId, chatSseEmitter, chatContext, + answer, assistantAccumulator, legacyThinkingTagParser, finished, persistChatlog)) { + return; + } handleRuntimeError(new BusinessException(errorMessage(event)), requestId, chatSseEmitter, chatContext, finished, persistChatlog); } } + /** + * 将解析后的助手片段累计、持久化并发送到前端。 + * + * @param segments 解析片段 + * @param requestId 运行请求 ID + * @param chatSseEmitter SSE 发送器 + * @param chatContext 聊天上下文 + * @param answer 最终正文缓冲 + * @param assistantAccumulator 结构化消息缓冲 + * @param legacyThinkingTagParser 旧思考标签解析器 + * @param finished 完成标记 + * @param persistChatlog 是否持久化聊天记录 + * @return 全部片段发送成功时为 {@code true} + */ + private boolean emitAssistantSegments(List segments, + String requestId, + ChatSseEmitter chatSseEmitter, + ChatRuntimeContext chatContext, + StringBuilder answer, + ChatAssistantAccumulator assistantAccumulator, + LegacyThinkingTagParser legacyThinkingTagParser, + AtomicBoolean finished, + boolean persistChatlog) { + for (LegacyThinkingTagParser.Segment segment : segments) { + String text = segment.getText(); + ChatType chatType; + Map payload = new LinkedHashMap<>(); + if (segment.getType() == LegacyThinkingTagParser.SegmentType.REASONING) { + assistantAccumulator.appendReasoning(text); + payload.put("reasoning", text); + payload.put("delta", text); + chatType = ChatType.THINKING; + } else { + answer.append(text); + assistantAccumulator.appendContent(text); + payload.put("delta", text); + payload.put("role", "assistant"); + chatType = ChatType.MESSAGE; + LOG.debug("Agent runtime message delta, requestId={}, deltaLength={}, answerLength={}, delta={}", + requestId, text.length(), answer.length(), toVisibleLogText(text)); + } + if (!sendEnvelope(chatSseEmitter, ChatDomain.LLM, chatType, payload)) { + cancelDisconnectedRun(requestId, chatContext, answer, assistantAccumulator, + legacyThinkingTagParser, finished, persistChatlog); + return false; + } + } + return true; + } + + /** + * 仅累计解析片段，用于连接已断开后的部分消息持久化。 + * + * @param segments 解析片段 + * @param answer 最终正文缓冲 + * @param assistantAccumulator 结构化消息缓冲 + */ + private void appendAssistantSegments(List segments, + StringBuilder answer, + ChatAssistantAccumulator assistantAccumulator) { + for (LegacyThinkingTagParser.Segment segment : segments) { + if (segment.getType() == LegacyThinkingTagParser.SegmentType.REASONING) { + assistantAccumulator.appendReasoning(segment.getText()); + } else { + answer.append(segment.getText()); + assistantAccumulator.appendContent(segment.getText()); + } + } + } + + /** + * 在运行时自然结束但未显式发出完成事件时收口兼容解析器。 + * + * @param requestId 运行请求 ID + * @param chatSseEmitter SSE 发送器 + * @param chatContext 聊天上下文 + * @param answer 最终正文缓冲 + * @param assistantAccumulator 结构化消息缓冲 + * @param legacyThinkingTagParser 旧思考标签解析器 + * @param finished 完成标记 + * @param persistChatlog 是否持久化聊天记录 + */ + private void finishRuntimeStream(String requestId, + ChatSseEmitter chatSseEmitter, + ChatRuntimeContext chatContext, + StringBuilder answer, + ChatAssistantAccumulator assistantAccumulator, + LegacyThinkingTagParser legacyThinkingTagParser, + AtomicBoolean finished, + boolean persistChatlog) { + if (!emitAssistantSegments(legacyThinkingTagParser.finish(), requestId, chatSseEmitter, chatContext, + answer, assistantAccumulator, legacyThinkingTagParser, finished, persistChatlog)) { + return; + } + finishIfNeeded(requestId, chatSseEmitter, chatContext, answer, + assistantAccumulator, finished, persistChatlog); + } + + /** + * 在运行时异常结束前发送兼容解析器中尚未收口的片段。 + * + * @param error 运行异常 + * @param requestId 运行请求 ID + * @param chatSseEmitter SSE 发送器 + * @param chatContext 聊天上下文 + * @param answer 最终正文缓冲 + * @param assistantAccumulator 结构化消息缓冲 + * @param legacyThinkingTagParser 旧思考标签解析器 + * @param finished 完成标记 + * @param persistChatlog 是否持久化聊天记录 + */ + private void handleRuntimeStreamError(Throwable error, + String requestId, + ChatSseEmitter chatSseEmitter, + ChatRuntimeContext chatContext, + StringBuilder answer, + ChatAssistantAccumulator assistantAccumulator, + LegacyThinkingTagParser legacyThinkingTagParser, + AtomicBoolean finished, + boolean persistChatlog) { + if (!emitAssistantSegments(legacyThinkingTagParser.finish(), requestId, chatSseEmitter, chatContext, + answer, assistantAccumulator, legacyThinkingTagParser, finished, persistChatlog)) { + return; + } + handleRuntimeError(error, requestId, chatSseEmitter, chatContext, finished, persistChatlog); + } + private void finishIfNeeded(String requestId, ChatSseEmitter chatSseEmitter, ChatRuntimeContext chatContext, diff --git a/easyflow-modules/easyflow-module-agent/src/test/java/tech/easyflow/agent/runtime/AgentRunServiceDraftAndHitlTest.java b/easyflow-modules/easyflow-module-agent/src/test/java/tech/easyflow/agent/runtime/AgentRunServiceDraftAndHitlTest.java index 77d33e05..6e6909b9 100644 --- a/easyflow-modules/easyflow-module-agent/src/test/java/tech/easyflow/agent/runtime/AgentRunServiceDraftAndHitlTest.java +++ b/easyflow-modules/easyflow-module-agent/src/test/java/tech/easyflow/agent/runtime/AgentRunServiceDraftAndHitlTest.java @@ -32,6 +32,7 @@ import tech.easyflow.core.runtime.ChatAssistantAccumulator; import tech.easyflow.core.runtime.ChatRuntimeContext; import tech.easyflow.core.runtime.ChatRuntimeManager; import tech.easyflow.core.runtime.ChatRuntimeMessage; +import tech.easyflow.core.runtime.LegacyThinkingTagParser; import java.lang.reflect.Method; import java.math.BigInteger; @@ -176,6 +177,59 @@ public class AgentRunServiceDraftAndHitlTest { Assert.assertEquals("正文增量", payload.get("delta")); } + /** + * 验证旧模型写入 content 的思考标签即使跨增量拆分，也会转换为结构化思考事件。 + * + * @throws Exception 反射调用失败时抛出 + */ + @Test + public void handleRuntimeEventShouldSplitLegacyThinkingTagsAcrossDeltas() throws Exception { + AgentRunService service = new AgentRunService(); + setField(service, "agentRunRegistry", new AgentRunRegistry()); + RecordingChatSseEmitter emitter = new RecordingChatSseEmitter(); + StringBuilder answer = new StringBuilder(); + ChatAssistantAccumulator assistantAccumulator = new ChatAssistantAccumulator(); + LegacyThinkingTagParser parser = new LegacyThinkingTagParser(); + AtomicBoolean finished = new AtomicBoolean(false); + + for (String delta : List.of("先分析\n最终回答")) { + AgentRuntimeEvent event = AgentRuntimeEvent.of(AgentRuntimeEventType.MESSAGE_DELTA); + event.getPayload().put("text", delta); + invoke(service, "handleRuntimeEvent", + legacyRuntimeEventParameterTypes(), + event, "request-legacy-thinking", emitter, answer, assistantAccumulator, + parser, chatContext(), finished, false); + } + + AgentRuntimeEvent completed = AgentRuntimeEvent.of(AgentRuntimeEventType.COMPLETED); + completed.getPayload().put("text", "先分析\n最终回答"); + invoke(service, "handleRuntimeEvent", + legacyRuntimeEventParameterTypes(), + completed, "request-legacy-thinking", emitter, answer, assistantAccumulator, + parser, chatContext(), finished, false); + + StringBuilder reasoning = new StringBuilder(); + StringBuilder content = new StringBuilder(); + for (ChatEnvelope envelope : emitter.envelopes) { + if (envelope.getDomain() != ChatDomain.LLM) { + continue; + } + @SuppressWarnings("unchecked") + Map payload = (Map) envelope.getPayload(); + if (envelope.getType() == ChatType.THINKING) { + reasoning.append(payload.get("delta")); + } else if (envelope.getType() == ChatType.MESSAGE) { + content.append(payload.get("delta")); + } + } + + Assert.assertEquals("先分析", reasoning.toString()); + Assert.assertEquals("\n最终回答", content.toString()); + Assert.assertEquals("\n最终回答", answer.toString()); + Assert.assertTrue(emitter.envelopes.stream().anyMatch(envelope -> + envelope.getDomain() == ChatDomain.SYSTEM && envelope.getType() == ChatType.DONE)); + } + /** * 验证自动上下文压缩事件会作为业务状态发送给前端。 * @@ -809,6 +863,12 @@ public class AgentRunServiceDraftAndHitlTest { ChatRuntimeContext.class, AtomicBoolean.class, boolean.class}; } + private Class[] legacyRuntimeEventParameterTypes() { + return new Class[]{AgentRuntimeEvent.class, String.class, ChatSseEmitter.class, StringBuilder.class, + ChatAssistantAccumulator.class, LegacyThinkingTagParser.class, + ChatRuntimeContext.class, AtomicBoolean.class, boolean.class}; + } + private AgentRunRegistry.AgentRunContext runContext(String requestId, String sessionId, boolean persistChatlog) { return new AgentRunRegistry.AgentRunContext( requestId, diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagents/listener/ChatStreamListener.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagents/listener/ChatStreamListener.java index 035e771f..3fde34a9 100644 --- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagents/listener/ChatStreamListener.java +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagents/listener/ChatStreamListener.java @@ -23,6 +23,7 @@ import tech.easyflow.core.runtime.ChatRuntimeContext; import tech.easyflow.core.runtime.ChatRuntimeExtKeys; import tech.easyflow.core.runtime.ChatRuntimeManager; import tech.easyflow.core.runtime.ChatRuntimeMessage; +import tech.easyflow.core.runtime.LegacyThinkingTagParser; import java.math.BigInteger; import java.util.Date; @@ -43,6 +44,7 @@ public class ChatStreamListener implements StreamResponseListener { private final ChatRuntimeManager chatRuntimeManager; private final ChatRuntimeContext runtimeContext; private final ChatAssistantAccumulator assistantAccumulator; + private final LegacyThinkingTagParser legacyThinkingTagParser = new LegacyThinkingTagParser(); // 核心标记：是否允许执行onStop业务逻辑（仅最后一次无后续工具调用时为true） private boolean canStop = true; // 辅助标记：是否进入过工具调用（避免重复递归判断） @@ -65,6 +67,7 @@ public class ChatStreamListener implements StreamResponseListener { @Override public void onStart(StreamContext context) { + legacyThinkingTagParser.reset(); StreamResponseListener.super.onStart(context); } @@ -80,6 +83,7 @@ public class ChatStreamListener implements StreamResponseListener { return; } if (aiMessage.isFinalDelta() && aiMessageResponse.hasToolCalls()) { + flushLegacyThinkingSegments(); this.canStop = false; // 工具调用期间，禁止执行onStop this.hasToolCall = true; // 标记已进入过工具调用 List toolCalls = aiMessage.getToolCalls(); @@ -103,20 +107,7 @@ public class ChatStreamListener implements StreamResponseListener { if (this.hasToolCall) { this.canStop = true; } - String reasoningContent = aiMessage.getReasoningContent(); - if (reasoningContent != null && !reasoningContent.isEmpty()) { - assistantAccumulator.appendReasoning(reasoningContent); - chatRuntimeManager.recordAssistantDelta(runtimeContext, buildAssistantDeltaMessage(reasoningContent, ChatType.THINKING)); - sendChatEnvelope(sseEmitter, reasoningContent, ChatType.THINKING); - } else { - String delta = aiMessage.getContent(); - if (delta != null && !delta.isEmpty()) { - assistantAccumulator.appendContent(delta); - chatRuntimeManager.recordAssistantDelta(runtimeContext, buildAssistantDeltaMessage(delta, ChatType.MESSAGE)); - sendChatEnvelope(sseEmitter, delta, ChatType.MESSAGE); - } - } - + handleAssistantDelta(aiMessage); } } catch (Exception e) { LOG.error("Chat stream onMessage failed, conversationId={}, message={}, exception={}", @@ -137,6 +128,7 @@ public class ChatStreamListener implements StreamResponseListener { sendSystemError(sseEmitter, context.getThrowable().getMessage(), context.getThrowable()); return; } + flushLegacyThinkingSegments(); memoryPrompt.addMessage(context.getFullMessage()); chatRuntimeManager.recordAssistantCompleted(runtimeContext, buildAssistantCompletedMessage(context)); chatRuntimeManager.recordCompleted(runtimeContext); @@ -186,6 +178,49 @@ public class ChatStreamListener implements StreamResponseListener { } } + /** + * 归一化并发送单个模型增量。 + * + * @param aiMessage 模型增量消息 + */ + private void handleAssistantDelta(AiMessage aiMessage) { + String reasoningContent = aiMessage.getReasoningContent(); + if (StringUtil.hasText(reasoningContent)) { + emitAssistantSegments(legacyThinkingTagParser.acceptReasoning(reasoningContent)); + return; + } + emitAssistantSegments(legacyThinkingTagParser.acceptContent(aiMessage.getContent())); + } + + /** + * 发送兼容解析后的思考与正文片段。 + * + * @param segments 解析片段 + */ + private void emitAssistantSegments(List segments) { + for (LegacyThinkingTagParser.Segment segment : segments) { + String text = segment.getText(); + if (segment.getType() == LegacyThinkingTagParser.SegmentType.REASONING) { + assistantAccumulator.appendReasoning(text); + chatRuntimeManager.recordAssistantDelta(runtimeContext, + buildAssistantDeltaMessage(text, ChatType.THINKING)); + sendChatEnvelope(sseEmitter, text, ChatType.THINKING); + continue; + } + assistantAccumulator.appendContent(text); + chatRuntimeManager.recordAssistantDelta(runtimeContext, + buildAssistantDeltaMessage(text, ChatType.MESSAGE)); + sendChatEnvelope(sseEmitter, text, ChatType.MESSAGE); + } + } + + /** + * 收口并发送旧标签解析器仍保留的少量前缀。 + */ + private void flushLegacyThinkingSegments() { + emitAssistantSegments(legacyThinkingTagParser.finish()); + } + private void sendToolCallEnvelope(ToolCall toolCall) { if (toolCall == null) { return; @@ -330,7 +365,10 @@ public class ChatStreamListener implements StreamResponseListener { message.setRole("assistant"); message.setContentType("TEXT"); String fullContent = context != null && context.getFullMessage() != null ? context.getFullMessage().getContent() : null; - message.setContentText(StringUtil.hasText(fullContent) ? fullContent : assistantAccumulator.getContent()); + String normalizedContent = assistantAccumulator.getContent(); + message.setContentText(legacyThinkingTagParser.isLegacyFormatDetected() + ? normalizedContent + : (StringUtil.hasText(fullContent) ? fullContent : normalizedContent)); message.setContentPayload(assistantAccumulator.buildPayload(message.getContentText())); message.setCreatedAt(new Date()); message.setSenderId(runtimeContext.getAssistantId());