easyflow-commons/easyflow-common-chat-protocol/pom.xml

@@ -25,6 +25,12 @@
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-websocket</artifactId>
         </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>${junit.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
 </project>

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 <think>} 或 {@code <thinking>} 内容拆分为思考与正文增量。
+ *
+ * <p>解析仅在单轮回复开头生效，避免误处理正文中的标签示例。该解析器保留少量标签前缀，
+ * 因而可以正确处理开始或结束标签被拆分到多个流式增量中的情况。</p>
+ */
+public final class LegacyThinkingTagParser {
+
+    private static final String[] OPEN_TAGS = {"<think>", "<thinking>"};
+    private static final String[] CLOSE_TAGS = {"</think>", "</thinking>"};
+    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<Segment> 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<Segment> acceptReasoning(String delta) {
+        if (delta == null || delta.isBlank()) {
+            return Collections.emptyList();
+        }
+        List<Segment> segments = new ArrayList<>();
+        flushBeforeStructuredReasoning(segments);
+        mode = Mode.BYPASS;
+        addSegment(segments, SegmentType.REASONING, delta);
+        return segments;
+    }
+
+    /**
+     * 收口尚未发送的标签前缀或内容。
+     *
+     * @return 剩余的思考或正文片段
+     */
+    public List<Segment> 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<Segment> 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<Segment> consumeThinking(String delta) {
+        pending.append(delta);
+        String normalized = pending.toString().toLowerCase(Locale.ROOT);
+        int closingIndex = normalized.indexOf(activeCloseTag);
+        List<Segment> 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<Segment> 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<Segment> 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<Segment> 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;
+        }
+    }
+}

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<LegacyThinkingTagParser.Segment> segments = new ArrayList<>();
+
+        segments.addAll(parser.acceptContent("  <thi"));
+        segments.addAll(parser.acceptContent("nk>先分析</thi"));
+        segments.addAll(parser.acceptContent("nk>\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 <thinking>} 别名及其跨增量结束标签可以正确解析。
+     */
+    @Test
+    public void shouldSplitThinkingAliasAcrossChunks() {
+        LegacyThinkingTagParser parser = new LegacyThinkingTagParser();
+        List<LegacyThinkingTagParser.Segment> segments = new ArrayList<>();
+
+        segments.addAll(parser.acceptContent("<thinking>先分析</think"));
+        segments.addAll(parser.acceptContent("ing>最终回答"));
+        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<LegacyThinkingTagParser.Segment> segments = new ArrayList<>();
+
+        segments.addAll(parser.acceptContent("示例：<think>内容</think>"));
+        segments.addAll(parser.finish());
+
+        Assert.assertFalse(parser.isLegacyFormatDetected());
+        Assert.assertEquals("示例：<think>内容</think>",
+                join(segments, LegacyThinkingTagParser.SegmentType.CONTENT));
+        Assert.assertEquals("", join(segments, LegacyThinkingTagParser.SegmentType.REASONING));
+    }
+
+    /**
+     * 验证结构化思考协议会旁路旧标签识别。
+     */
+    @Test
+    public void shouldBypassLegacyParsingForStructuredReasoning() {
+        LegacyThinkingTagParser parser = new LegacyThinkingTagParser();
+        List<LegacyThinkingTagParser.Segment> segments = new ArrayList<>();
+
+        segments.addAll(parser.acceptReasoning("结构化思考"));
+        segments.addAll(parser.acceptContent("<think>正文标签示例</think>"));
+
+        Assert.assertFalse(parser.isLegacyFormatDetected());
+        Assert.assertEquals("结构化思考", join(segments, LegacyThinkingTagParser.SegmentType.REASONING));
+        Assert.assertEquals("<think>正文标签示例</think>",
+                join(segments, LegacyThinkingTagParser.SegmentType.CONTENT));
+    }
+
+    /**
+     * 验证空结构化思考不会阻止正文开头的旧标签识别。
+     */
+    @Test
+    public void shouldParseLegacyTagWhenStructuredReasoningIsBlank() {
+        LegacyThinkingTagParser parser = new LegacyThinkingTagParser();
+        List<LegacyThinkingTagParser.Segment> segments = new ArrayList<>();
+
+        segments.addAll(parser.acceptReasoning("  "));
+        segments.addAll(parser.acceptContent("<think>旧版思考</think>正文"));
+
+        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<LegacyThinkingTagParser.Segment> segments = new ArrayList<>();
+
+        segments.addAll(parser.acceptContent("<think>尚未完成</thi"));
+        segments.addAll(parser.finish());
+
+        Assert.assertTrue(parser.isLegacyFormatDetected());
+        Assert.assertEquals("尚未完成</thi",
+                join(segments, LegacyThinkingTagParser.SegmentType.REASONING));
+        Assert.assertEquals("", join(segments, LegacyThinkingTagParser.SegmentType.CONTENT));
+    }
+
+    private String join(List<LegacyThinkingTagParser.Segment> segments,
+                        LegacyThinkingTagParser.SegmentType type) {
+        return segments.stream()
+                .filter(segment -> segment.getType() == type)
+                .map(LegacyThinkingTagParser.Segment::getText)
+                .reduce("", String::concat);
+    }
+}

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<String, Object> 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<String, Object> 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<Map<String, Object>> 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<LegacyThinkingTagParser.Segment> 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<String, Object> 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<LegacyThinkingTagParser.Segment> 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,

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("<thi", "nk>先分析</thi", "nk>\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", "<think>先分析</think>\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<String, Object> payload = (Map<String, Object>) 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,

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<ToolCall> 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<LegacyThinkingTagParser.Segment> 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());