From c5e72078e50d2d08f234b46868193b939c2e50d3 Mon Sep 17 00:00:00 2001 From: andy Date: Sat, 11 Jul 2026 12:35:09 +0800 Subject: [PATCH] =?UTF-8?q?=E5=85=BC=E5=AE=B9=20SuperAgent=20SSE=20?= =?UTF-8?q?=E7=BB=93=E6=9E=9C=E8=A7=A3=E6=9E=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../impl/SuperAgentOpenApiSseParser.java | 138 +++++++++++++++--- .../impl/SuperAgentOpenApiSseParserTest.java | 126 ++++++++++++++++ 2 files changed, 247 insertions(+), 17 deletions(-) diff --git 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 index c192097..edcd91e 100644 --- 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; } } diff --git 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 index 705a7eb..1403839 100644 --- 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\"}]}"); + } }