兼容 SuperAgent SSE 结果解析

2026-07-11 12:35:09 +08:00
parent 1046e4ef74
commit c5e72078e5
2 changed files with 247 additions and 17 deletions
--- a/server/src/main/java/cn/nianxx/thhotel/integrations/ai/superagent/service/impl/SuperAgentOpenApiSseParser.java
+++ b/server/src/main/java/cn/nianxx/thhotel/integrations/ai/superagent/service/impl/SuperAgentOpenApiSseParser.java
@@ -10,7 +10,7 @@ import java.util.Set;
 import org.springframework.stereotype.Component;

 /**
- * SuperAgent SSE 响应解析器。优先从 values.messages 中提取 finish_reason=stop 的最终 AI 回答。
+ * SuperAgent SSE 响应解析器。优先从 values/messages 中提取 finish_reason=stop 的最终 AI 回答。
 */
@Component
 public class SuperAgentOpenApiSseParser {
@@ -43,6 +43,19 @@ public class SuperAgentOpenApiSseParser {
        for (SseEvent event : splitEvents(sseBody)) {
            eventTypes.add(event.eventType());
            consumeEvent(event, state);
+            if (state.endSeen) {
+                break;
+            }
+        }
+        if (!state.endSeen) {
+            throw new SuperAgentOpenApiException("SuperAgent SSE 未收到结束事件。");
+        }
+        if ((state.rawAnswer == null || state.rawAnswer.isBlank()) && state.fallbackRawAnswer != null) {
+            state.rawAnswer = state.fallbackRawAnswer;
+            state.modelName = state.fallbackModelName;
+            state.inputTokens = state.fallbackInputTokens;
+            state.outputTokens = state.fallbackOutputTokens;
+            state.totalTokens = state.fallbackTotalTokens;
        }
        if (state.rawAnswer == null || state.rawAnswer.isBlank()) {
            throw new SuperAgentOpenApiException("SuperAgent SSE 未找到最终 AI 回答。");
@@ -65,12 +78,19 @@ public class SuperAgentOpenApiSseParser {
     */
    private void consumeEvent(SseEvent event, ParsedState state) {
        try {
+            if ("end".equals(event.eventType())) {
+                state.endSeen = true;
+                return;
+            }
            JsonNode data = objectMapper.readTree(event.data());
            if ("metadata".equals(event.eventType())) {
                state.runId = text(data, "run_id", state.runId);
                state.profileId = text(data, "resolved_profile_id", state.profileId);
                state.profileVersionId = text(data, "resolved_profile_version_id", state.profileVersionId);
            }
+            if ("messages".equals(event.eventType())) {
+                consumeMessages(data, state);
+            }
            if ("values".equals(event.eventType())) {
                consumeValuesEvent(data, state);
            }
@@ -84,24 +104,58 @@ public class SuperAgentOpenApiSseParser {
     */
    private void consumeValuesEvent(JsonNode data, ParsedState state) {
        JsonNode messages = data.path("messages");
-        if (!messages.isArray()) {
+        consumeMessages(messages, state);
+    }
+
+    /**
+     * 兼容单条 message、messages 数组以及包装在 messages 字段中的聚合结构。
+     */
+    private void consumeMessages(JsonNode data, ParsedState state) {
+        if (data == null || data.isMissingNode() || data.isNull()) {
            return;
        }
-        for (JsonNode message : messages) {
-            if (!"ai".equals(message.path("type").asText())) {
-                continue;
-            }
-            String finishReason = message.path("response_metadata").path("finish_reason").asText();
-            String content = message.path("content").asText(null);
-            if ("stop".equals(finishReason) && content != null && !content.isBlank()) {
-                state.rawAnswer = content;
-                state.modelName = text(message.path("response_metadata"), "model_name", state.modelName);
-                JsonNode usage = message.path("usage_metadata");
-                state.inputTokens = intValue(usage, "input_tokens", state.inputTokens);
-                state.outputTokens = intValue(usage, "output_tokens", state.outputTokens);
-                state.totalTokens = intValue(usage, "total_tokens", state.totalTokens);
+        if (data.isArray()) {
+            for (JsonNode message : data) {
+                consumeAiMessage(message, state);
            }
+            return;
        }
+        JsonNode messages = data.path("messages");
+        if (messages.isArray()) {
+            consumeMessages(messages, state);
+            return;
+        }
+        consumeAiMessage(data, state);
+    }
+
+    /**
+     * 提取 AI 消息内容。finish_reason=stop 的内容优先，非 stop 内容只作为流正常结束后的兜底。
+     */
+    private void consumeAiMessage(JsonNode message, ParsedState state) {
+        if (!"ai".equals(message.path("type").asText())) {
+            return;
+        }
+        String content = contentText(message.path("content"));
+        if (content == null || content.isBlank()) {
+            return;
+        }
+        JsonNode responseMetadata = message.path("response_metadata");
+        JsonNode usage = message.path("usage_metadata");
+        String modelName = text(responseMetadata, "model_name", state.modelName);
+        String finishReason = responseMetadata.path("finish_reason").asText();
+        if ("stop".equals(finishReason)) {
+            state.rawAnswer = content;
+            state.modelName = modelName;
+            state.inputTokens = intValue(usage, "input_tokens", state.inputTokens);
+            state.outputTokens = intValue(usage, "output_tokens", state.outputTokens);
+            state.totalTokens = intValue(usage, "total_tokens", state.totalTokens);
+            return;
+        }
+        state.fallbackRawAnswer = content;
+        state.fallbackModelName = modelName;
+        state.fallbackInputTokens = intValue(usage, "input_tokens", state.fallbackInputTokens);
+        state.fallbackOutputTokens = intValue(usage, "output_tokens", state.fallbackOutputTokens);
+        state.fallbackTotalTokens = intValue(usage, "total_tokens", state.fallbackTotalTokens);
    }

    /**
@@ -125,7 +179,7 @@ public class SuperAgentOpenApiSseParser {
                    data.append(line.substring("data:".length()).trim());
                }
            }
-            if (!data.isEmpty()) {
+            if (!data.isEmpty() || "end".equals(eventType)) {
                events.add(new SseEvent(eventType, data.toString()));
            }
        }
@@ -145,7 +199,51 @@ public class SuperAgentOpenApiSseParser {
     */
    private Integer intValue(JsonNode node, String fieldName, Integer fallback) {
        JsonNode value = node.path(fieldName);
-        return value.isMissingNode() || value.isNull() ? fallback : value.asInt();
+        if (value.isMissingNode() || value.isNull()) {
+            return fallback;
+        }
+        return value.asInt();
+    }
+
+    /**
+     * 读取 AI content，兼容字符串、文本片段数组和简单文本对象。
+     */
+    private String contentText(JsonNode content) {
+        if (content == null || content.isMissingNode() || content.isNull()) {
+            return null;
+        }
+        if (content.isTextual()) {
+            return content.asText();
+        }
+        if (content.isArray()) {
+            StringBuilder builder = new StringBuilder();
+            for (JsonNode item : content) {
+                appendContentText(builder, contentText(item));
+            }
+            return builder.isEmpty() ? null : builder.toString();
+        }
+        if (content.isObject()) {
+            String text = contentText(content.path("text"));
+            if (text != null && !text.isBlank()) {
+                return text;
+            }
+            text = contentText(content.path("content"));
+            return text == null || text.isBlank() ? content.toString() : text;
+        }
+        return null;
+    }
+
+    /**
+     * 拼接数组型 content 中的非空文本片段。
+     */
+    private void appendContentText(StringBuilder builder, String text) {
+        if (text == null || text.isBlank()) {
+            return;
+        }
+        if (!builder.isEmpty()) {
+            builder.append('\n');
+        }
+        builder.append(text);
    }

    /**
@@ -158,6 +256,7 @@ public class SuperAgentOpenApiSseParser {
     * SSE 解析过程中的可变状态。
     */
    private static final class ParsedState {
+        private boolean endSeen;
        private String runId;
        private String profileId;
        private String profileVersionId;
@@ -166,5 +265,10 @@ public class SuperAgentOpenApiSseParser {
        private Integer inputTokens;
        private Integer outputTokens;
        private Integer totalTokens;
+        private String fallbackRawAnswer;
+        private String fallbackModelName;
+        private Integer fallbackInputTokens;
+        private Integer fallbackOutputTokens;
+        private Integer fallbackTotalTokens;
    }
 }
--- a/server/src/test/java/cn/nianxx/thhotel/integrations/ai/superagent/service/impl/SuperAgentOpenApiSseParserTest.java
+++ b/server/src/test/java/cn/nianxx/thhotel/integrations/ai/superagent/service/impl/SuperAgentOpenApiSseParserTest.java
@@ -1,6 +1,7 @@
 package cn.nianxx.thhotel.integrations.ai.superagent.service.impl;

 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;

 import cn.nianxx.thhotel.integrations.ai.superagent.common.result.SuperAgentOpenApiResult;
 import org.junit.jupiter.api.Test;
@@ -38,4 +39,129 @@ class SuperAgentOpenApiSseParserTest {
        assertThat(result.totalTokens()).isEqualTo(18);
        assertThat(result.eventTypes()).containsExactly("metadata", "messages", "values", "end");
    }
+
+    @Test
+    void shouldExtractFinalAiAnswerFromMessagesEventWhenStopReasonPresent() {
+        String sse = """
+                event: metadata
+                data: {"run_id":"run-debug-002","resolved_profile_id":"profile-debug"}
+
+                event: messages
+                data: {"type":"ai","content":"{\\"ai_task_results\\":[{\\"task_type\\":\\"Update Booking\\"}]}","response_metadata":{"finish_reason":"stop","model_name":"debug-model-from-message"},"usage_metadata":{"input_tokens":21,"output_tokens":13,"total_tokens":34}}
+
+                event: end
+                data: {}
+
+                """;
+
+        SuperAgentOpenApiResult result = parser.parse("session-debug-002", sse);
+
+        assertThat(result.runId()).isEqualTo("run-debug-002");
+        assertThat(result.rawAnswer()).isEqualTo("{\"ai_task_results\":[{\"task_type\":\"Update Booking\"}]}");
+        assertThat(result.modelName()).isEqualTo("debug-model-from-message");
+        assertThat(result.inputTokens()).isEqualTo(21);
+        assertThat(result.outputTokens()).isEqualTo(13);
+        assertThat(result.totalTokens()).isEqualTo(34);
+        assertThat(result.eventTypes()).containsExactly("metadata", "messages", "end");
+    }
+
+    @Test
+    void shouldFallbackToLastAiMessageWhenStreamEndedWithoutStopReason() {
+        String sse = """
+                event: messages
+                data: {"type":"ai","content":"partial answer"}
+
+                event: messages
+                data: {"type":"ai","content":"{\\"ai_task_results\\":[{\\"task_type\\":\\"Rooming List\\"}]}","response_metadata":{"model_name":"debug-model-fallback"}}
+
+                event: end
+                data: {}
+
+                """;
+
+        SuperAgentOpenApiResult result = parser.parse("session-debug-003", sse);
+
+        assertThat(result.rawAnswer()).isEqualTo("{\"ai_task_results\":[{\"task_type\":\"Rooming List\"}]}");
+        assertThat(result.modelName()).isEqualTo("debug-model-fallback");
+    }
+
+    @Test
+    void shouldFailWhenEndEventMissingEvenIfAiContentExists() {
+        String sse = """
+                event: messages
+                data: {"type":"ai","content":"{\\"ai_task_results\\":[]}","response_metadata":{"finish_reason":"stop"}}
+
+                """;
+
+        assertThatThrownBy(() -> parser.parse("session-debug-004", sse))
+                .isInstanceOf(SuperAgentOpenApiException.class)
+                .hasMessageContaining("SuperAgent SSE 未收到结束事件。");
+    }
+
+    @Test
+    void shouldFailWhenStreamEndedWithoutAiContent() {
+        String sse = """
+                event: metadata
+                data: {"run_id":"run-debug-005"}
+
+                event: values
+                data: {"messages":[{"type":"human","content":"input"}]}
+
+                event: end
+                data: {}
+
+                """;
+
+        assertThatThrownBy(() -> parser.parse("session-debug-005", sse))
+                .isInstanceOf(SuperAgentOpenApiException.class)
+                .hasMessageContaining("SuperAgent SSE 未找到最终 AI 回答。");
+    }
+
+    @Test
+    void shouldNotUseMessagesAfterEndAsFinalAnswer() {
+        String sse = """
+                event: end
+                data: {}
+
+                event: messages
+                data: {"type":"ai","content":"{\\"ai_task_results\\":[]}","response_metadata":{"finish_reason":"stop"}}
+
+                """;
+
+        assertThatThrownBy(() -> parser.parse("session-debug-006", sse))
+                .isInstanceOf(SuperAgentOpenApiException.class)
+                .hasMessageContaining("SuperAgent SSE 未找到最终 AI 回答。");
+    }
+
+    @Test
+    void shouldAcceptEndEventWithoutData() {
+        String sse = """
+                event: messages
+                data: {"type":"ai","content":"{\\"ai_task_results\\":[{\\"task_type\\":\\"Voucher Payment\\"}]}","response_metadata":{"finish_reason":"stop"}}
+
+                event: end
+
+                """;
+
+        SuperAgentOpenApiResult result = parser.parse("session-debug-007", sse);
+
+        assertThat(result.rawAnswer()).isEqualTo("{\"ai_task_results\":[{\"task_type\":\"Voucher Payment\"}]}");
+        assertThat(result.eventTypes()).containsExactly("messages", "end");
+    }
+
+    @Test
+    void shouldSerializeObjectContentAsFinalAnswer() {
+        String sse = """
+                event: messages
+                data: {"type":"ai","content":{"ai_task_results":[{"task_type":"Cancel Booking"}]},"response_metadata":{"finish_reason":"stop"}}
+
+                event: end
+                data: {}
+
+                """;
+
+        SuperAgentOpenApiResult result = parser.parse("session-debug-008", sse);
+
+        assertThat(result.rawAnswer()).isEqualTo("{\"ai_task_results\":[{\"task_type\":\"Cancel Booking\"}]}");
+    }
 }