增加与ai大模型对话

huimv-employment/fe-api/src/main/java/com/huimv/employment/controller/mp/ConversationController.java

@@ -112,9 +112,10 @@ public class ConversationController {
     @PostMapping(value = "/{id}/chat", consumes = MediaType.APPLICATION_JSON_VALUE)
     @Operation(summary = "会话内 AI 对话（SSE 流式 + 落库）",
             description = "需先 POST /conversations 创建会话。请求体仅需 message 字段；"
-                    + "服务端组装上游 console/chat 请求，流式开始前写入 user 消息，"
-                    + "SSE complete 后由服务端根据 ready 草稿写入 content_type/card_payload/next_action（不依赖智能体返回卡片），"
-                    + "可选请求头 X-Agent-Id。")
+                    + "服务端组装上游 console/chat 请求，流式开始前写入 user 消息；"
+                    + "SSE 最后一条消息由服务端注入 content_type/next_action/card_payload"
+                    + "（会话有 pending 草稿且 missingFields 为空时，不依赖智能体），"
+                    + "并同步落库。可选请求头 X-Agent-Id。")
     public void chat(@PathVariable("id") Long conversationId,
                      @Validated @RequestBody ConversationChatRequest request,
                      HttpServletResponse response,

huimv-employment/fe-integration/src/main/java/com/huimv/employment/integration/kb/ConsoleChatSseAggregator.java

@@ -9,23 +9,19 @@ import java.util.HashSet;
 import java.util.Set;
 
 /**
- * 从 {@code /api/console/chat} SSE 字节流中增量提取 assistant 文本，并识别流结束 complete 事件。
- * <p>
- * 兼容：legacy {@code output[]}、{@code object=content} delta、
- * {@code object=response/status=completed}、{@code object=message} 完成态，以及重复 {@code data:} 前缀。
- * </p>
+ * 从 {@code /api/console/chat} SSE 字节流中增量提取 assistant 文本，
+ * 识别流结束事件，并暂存「最后一条终端消息」供业务层追加卡片字段后再转发。
  */
 public class ConsoleChatSseAggregator {
 
     private final ObjectMapper objectMapper;
     private final StringBuilder lineBuffer = new StringBuilder();
-    /** delta 格式拼接（不含 reasoning） */
     private final StringBuilder streamingText = new StringBuilder();
-    /** 完成态 assistant 回复（优先返回） */
     private final StringBuilder completedReply = new StringBuilder();
     private final Set<String> reasoningMessageIds = new HashSet<>();
-    /** 是否已收到上游 complete / DONE 结束信号 */
     private boolean completed;
+    /** 暂存的最后一条终端 SSE 行（未转发，待业务层注入卡片字段） */
+    private String heldTerminalLine;
 
     public ConsoleChatSseAggregator(ObjectMapper objectMapper) {
         this.objectMapper = objectMapper;
@@ -39,17 +35,22 @@ public class ConsoleChatSseAggregator {
         drainCompleteLines();
     }
 
-    /** 流结束时刷新缓冲区中未换行的尾部行。 */
     public void finish() {
         drainCompleteLines();
         if (lineBuffer.length() > 0) {
             consumeSseLine(lineBuffer.toString().trim());
             lineBuffer.setLength(0);
         }
-        // 流关闭本身也视为一轮对话结束
         completed = true;
     }
 
+    public void acceptLine(String line) {
+        if (!StringUtils.hasText(line)) {
+            return;
+        }
+        consumeSseLine(line.trim());
+    }
+
     public boolean isCompleted() {
         return completed;
     }
@@ -61,6 +62,39 @@ public class ConsoleChatSseAggregator {
         return streamingText.toString();
     }
 
+    /**
+     * 取出暂存的最后一条终端消息行（取出后清空）。
+     */
+    public String takeHeldTerminalLine() {
+        String line = heldTerminalLine;
+        heldTerminalLine = null;
+        return line;
+    }
+
+    /**
+     * 判断 JSON 载荷是否为对话结束/最后一条消息（需 hold 以便注入卡片）。
+     */
+    public boolean isTerminalPayload(String payload) {
+        if (!StringUtils.hasText(payload) || "[DONE]".equals(payload)) {
+            return false;
+        }
+        try {
+            JsonNode data = objectMapper.readTree(payload);
+            return isTerminalEvent(data);
+        } catch (Exception ex) {
+            return false;
+        }
+    }
+
+    /**
+     * hold 一条终端行；若已有 hold，返回旧行供立即转发。
+     */
+    public String holdTerminalLine(String line) {
+        String previous = heldTerminalLine;
+        heldTerminalLine = line;
+        return previous;
+    }
+
     private void drainCompleteLines() {
         int newlineIndex;
         while ((newlineIndex = indexOfLineBreak(lineBuffer)) >= 0) {
@@ -83,7 +117,7 @@ public class ConsoleChatSseAggregator {
             JsonNode data = objectMapper.readTree(payload);
             consumeJsonEvent(data);
         } catch (Exception ignored) {
-            // 跳过无法解析的 SSE 片段，不影响流式转发
+            // skip
         }
     }
 
@@ -102,9 +136,7 @@ public class ConsoleChatSseAggregator {
             return;
         }
 
-        // 最后一条 complete：response 完成态
-        if (("response".equals(objectType) || "chat.completion".equals(objectType))
-                && ("completed".equals(status) || "complete".equals(status))) {
+        if (isTerminalEvent(data)) {
             completed = true;
             if (data.has("output")) {
                 String reply = extractAssistantReplyFromOutput(data.get("output"));
@@ -112,17 +144,8 @@ public class ConsoleChatSseAggregator {
                     completedReply.setLength(0);
                     completedReply.append(reply);
                 }
-            }
-            return;
-        }
-
-        // 兼容 object=complete / type=complete
-        if ("complete".equals(objectType)
-                || "complete".equals(data.path("type").asText())
-                || "completed".equals(data.path("type").asText())) {
-            completed = true;
-            if (data.has("output")) {
-                String reply = extractAssistantReplyFromOutput(data.get("output"));
+            } else if ("message".equals(objectType) && "assistant".equals(data.path("role").asText())) {
+                String reply = extractTextFromContentNodes(data.get("content"));
                 if (StringUtils.hasText(reply)) {
                     completedReply.setLength(0);
                     completedReply.append(reply);
@@ -131,20 +154,6 @@ public class ConsoleChatSseAggregator {
             return;
         }
 
-        if ("message".equals(objectType)
-                && "message".equals(data.path("type").asText())
-                && "assistant".equals(data.path("role").asText())) {
-            String reply = extractTextFromContentNodes(data.get("content"));
-            if (StringUtils.hasText(reply)) {
-                completedReply.setLength(0);
-                completedReply.append(reply);
-            }
-            if ("completed".equals(status) || "complete".equals(status)) {
-                completed = true;
-            }
-            return;
-        }
-
         if ("content".equals(objectType) && "text".equals(data.path("type").asText())) {
             if (data.path("delta").asBoolean(false) && data.has("text")) {
                 String msgId = data.path("msg_id").asText(null);
@@ -165,6 +174,30 @@ public class ConsoleChatSseAggregator {
         }
     }
 
+    private static boolean isTerminalEvent(JsonNode data) {
+        if (data == null || !data.isObject()) {
+            return false;
+        }
+        String objectType = data.path("object").asText("");
+        String status = data.path("status").asText("");
+        String type = data.path("type").asText("");
+
+        if (("response".equals(objectType) || "chat.completion".equals(objectType))
+                && ("completed".equals(status) || "complete".equals(status))) {
+            return true;
+        }
+        if ("complete".equals(objectType) || "complete".equals(type) || "completed".equals(type)) {
+            return true;
+        }
+        // 完整 assistant 消息（非 reasoning、非纯 delta）
+        if ("message".equals(objectType)
+                && "message".equals(data.path("type").asText("message"))
+                && "assistant".equals(data.path("role").asText())) {
+            return true;
+        }
+        return false;
+    }
+
     private static String extractAssistantReplyFromOutput(JsonNode output) {
         if (output == null || !output.isArray()) {
             return "";
@@ -212,8 +245,7 @@ public class ConsoleChatSseAggregator {
         }
     }
 
-    /** 去掉一层或多层 {@code data:} 前缀，得到 JSON 载荷。 */
-    static String normalizeDataPayload(String line) {
+    public static String normalizeDataPayload(String line) {
         if (!StringUtils.hasText(line)) {
             return "";
         }

huimv-employment/fe-integration/src/main/java/com/huimv/employment/integration/kb/KbOpenAiProxyServiceImpl.java

@@ -67,7 +67,10 @@ public class KbOpenAiProxyServiceImpl implements KbOpenAiProxyService {
         ConsoleChatSseAggregator aggregator = new ConsoleChatSseAggregator(objectMapper);
         streamConsoleChatInternal(requestBody, agentId, output, aggregator);
         aggregator.finish();
-        return new ConsoleChatCollectResult(aggregator.getAssistantText(), aggregator.isCompleted());
+        return new ConsoleChatCollectResult(
+                aggregator.getAssistantText(),
+                aggregator.isCompleted(),
+                aggregator.takeHeldTerminalLine());
     }
 
     private void streamConsoleChatInternal(byte[] requestBody, String agentId, OutputStream output,
@@ -90,8 +93,11 @@ public class KbOpenAiProxyServiceImpl implements KbOpenAiProxyService {
             if (input != null) {
                 if (status >= 400) {
                     writeUpstreamError(output, input);
+                } else if (aggregator != null) {
+                    // 会话落库：暂缓 [DONE]，便于业务层把卡片追加到最后消息后再结束
+                    SseStreamSupport.pipeCollectingDeferDone(input, output, aggregator);
                 } else {
-                    SseStreamSupport.pipeWithFlush(input, output, aggregator);
+                    SseStreamSupport.pipeWithFlush(input, output, null);
                 }
             }
         } finally {

huimv-employment/fe-integration/src/main/java/com/huimv/employment/integration/kb/SseStreamSupport.java

@@ -34,9 +34,6 @@ public final class SseStreamSupport {
         pipeWithFlush(input, output, null);
     }
 
-    /**
-     * 转发 SSE 并在转发过程中收集字节（供 assistant 文本解析）。
-     */
     public static void pipeWithFlush(InputStream input, OutputStream output, ConsoleChatSseAggregator aggregator)
             throws IOException {
         if (input == null) {
@@ -53,6 +50,81 @@ public final class SseStreamSupport {
         }
     }
 
+    /**
+     * 会话落库：按行转发；暂缓 [DONE] 与「最后一条终端消息」，便于注入卡片字段后再写出。
+     */
+    public static void pipeCollectingDeferDone(InputStream input,
+                                               OutputStream output,
+                                               ConsoleChatSseAggregator aggregator) throws IOException {
+        if (input == null) {
+            return;
+        }
+        StringBuilder lineBuffer = new StringBuilder();
+        byte[] buffer = new byte[PIPE_BUFFER_SIZE];
+        int read;
+        while ((read = input.read(buffer)) != -1) {
+            lineBuffer.append(new String(buffer, 0, read, StandardCharsets.UTF_8));
+            drainAndForward(lineBuffer, output, aggregator);
+        }
+        if (lineBuffer.length() > 0) {
+            String leftover = lineBuffer.toString();
+            lineBuffer.setLength(0);
+            if (leftover.endsWith("\r")) {
+                leftover = leftover.substring(0, leftover.length() - 1);
+            }
+            forwardOrHoldLine(leftover, output, aggregator);
+        }
+    }
+
+    private static void drainAndForward(StringBuilder lineBuffer,
+                                        OutputStream output,
+                                        ConsoleChatSseAggregator aggregator) throws IOException {
+        int newlineIndex;
+        while ((newlineIndex = indexOfLineBreak(lineBuffer)) >= 0) {
+            String line = lineBuffer.substring(0, newlineIndex);
+            if (line.endsWith("\r")) {
+                line = line.substring(0, line.length() - 1);
+            }
+            lineBuffer.delete(0, newlineIndex + 1);
+            forwardOrHoldLine(line, output, aggregator);
+        }
+    }
+
+    private static void forwardOrHoldLine(String line,
+                                          OutputStream output,
+                                          ConsoleChatSseAggregator aggregator) throws IOException {
+        if (aggregator != null) {
+            aggregator.acceptLine(line.trim());
+        }
+        String payload = ConsoleChatSseAggregator.normalizeDataPayload(line.trim());
+        if ("[DONE]".equals(payload)) {
+            return;
+        }
+        // 终端消息：hold，旧 hold 先发出（保持流顺序）
+        if (aggregator != null && aggregator.isTerminalPayload(payload)) {
+            String previous = aggregator.holdTerminalLine(line);
+            if (previous != null) {
+                writeLine(output, previous);
+            }
+            return;
+        }
+        writeLine(output, line);
+    }
+
+    private static void writeLine(OutputStream output, String line) throws IOException {
+        output.write((line + "\n").getBytes(StandardCharsets.UTF_8));
+        output.flush();
+    }
+
+    private static int indexOfLineBreak(StringBuilder sb) {
+        for (int i = 0; i < sb.length(); i++) {
+            if (sb.charAt(i) == '\n') {
+                return i;
+            }
+        }
+        return -1;
+    }
+
     public static void writeDataEvent(OutputStream output, String dataPayload) throws IOException {
         if (dataPayload == null) {
             return;
@@ -65,4 +137,20 @@ public final class SseStreamSupport {
     public static void writeDoneEvent(OutputStream output) throws IOException {
         writeDataEvent(output, "[DONE]");
     }
+
+    /** 原样写出 hold 的 SSE 行（补齐事件分隔）。 */
+    public static void writeRawSseLine(OutputStream output, String line) throws IOException {
+        if (output == null || line == null) {
+            return;
+        }
+        String trimmed = line.trim();
+        if (trimmed.isEmpty()) {
+            return;
+        }
+        if (!trimmed.startsWith("data:")) {
+            trimmed = "data: " + trimmed;
+        }
+        output.write((trimmed + "\n\n").getBytes(StandardCharsets.UTF_8));
+        output.flush();
+    }
 }

huimv-employment/fe-service/src/main/java/com/huimv/employment/service/conversation/ConversationChatService.java

@@ -1,11 +1,14 @@
 package com.huimv.employment.service.conversation;
 
+import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.huimv.employment.dao.entity.FeConversation;
 import com.huimv.employment.integration.kb.ConsoleChatCollectResult;
+import com.huimv.employment.integration.kb.ConsoleChatSseAggregator;
 import com.huimv.employment.integration.kb.KbOpenAiProxyService;
+import com.huimv.employment.integration.kb.SseStreamSupport;
 import com.huimv.employment.service.conversation.dto.ConversationChatRequest;
 import com.huimv.employment.service.conversation.dto.ParsedAssistantReply;
 import org.slf4j.Logger;
@@ -17,10 +20,7 @@ import java.io.IOException;
 import java.io.OutputStream;
 
 /**
- * 会话内 AI 对话：组装上游请求、流式转发并在结束后落库。
- * <p>
- * 卡片字段由服务端在 SSE complete 后根据草稿状态写入，不依赖智能体返回 card JSON。
- * </p>
+ * 会话内 AI 对话：流式转发；pending 且 missingFields 为空时在 SSE 最后一条消息注入卡片字段后落库。
  */
 @Service
 public class ConversationChatService {
@@ -48,9 +48,6 @@ public class ConversationChatService {
         this.objectMapper = objectMapper;
     }
 
-    /**
-     * 在指定会话内发起一轮 AI 对话，SSE 流结束后写入 user/assistant 消息。
-     */
     public void chat(Long userId, Long conversationId, ConversationChatRequest request, String agentId,
                      OutputStream output) throws IOException {
         FeConversation conversation = conversationService.requireOwnedActive(userId, conversationId);
@@ -66,21 +63,92 @@ public class ConversationChatService {
 
         ConsoleChatCollectResult collectResult = kbOpenAiProxyService.streamConsoleChatCollecting(
                 upstreamBody, agentId, output, objectMapper);
-        // SSE complete（或流正常关闭）后：服务端按 ready 草稿写入卡片，忽略智能体卡片 JSON
+
         ParsedAssistantReply parsed = conversationDraftCardEnricher.attachCardOnComplete(
                 conversation.getId(), collectResult.getAssistantText());
 
+        // 把卡片字段写入「最后一条」SSE 消息本体后再发 [DONE]
+        writeLastSseMessageWithCard(output, collectResult.getHeldTerminalLine(), parsed);
+        SseStreamSupport.writeDoneEvent(output);
+
         if (!StringUtils.hasText(parsed.getDisplayText()) && !parsed.hasCard()) {
-            log.warn("assistant 回复为空，未写入 fe_conversation_message conversationId={} conversationNo={}",
-                    conversation.getId(), conversation.getConversationNo());
+            log.warn("assistant 回复为空，未写入 fe_conversation_message conversationId={}",
+                    conversation.getId());
         } else if (parsed.hasCard()) {
-            log.info("SSE complete 落库卡片 conversationId={} completed={} contentType={} relatedDraftId={}",
-                    conversation.getId(), collectResult.isCompleted(),
-                    parsed.getContentType(), parsed.getRelatedDraftId());
+            log.info("SSE 最后消息已注入卡片 conversationId={} relatedDraftId={}",
+                    conversation.getId(), parsed.getRelatedDraftId());
         }
         conversationMessageService.saveAssistantMessage(conversation.getId(), parsed);
     }
 
+    /**
+     * 在最后一条 SSE 消息 JSON 上追加 content_type / next_action / card_payload。
+     */
+    private void writeLastSseMessageWithCard(OutputStream output,
+                                             String heldTerminalLine,
+                                             ParsedAssistantReply parsed) throws IOException {
+        if (output == null) {
+            return;
+        }
+        if (!parsed.hasCard()) {
+            // 无卡片也要把 hold 的最后消息发出，避免客户端永远收不到完结消息
+            if (StringUtils.hasText(heldTerminalLine)) {
+                SseStreamSupport.writeRawSseLine(output, heldTerminalLine);
+            }
+            return;
+        }
+
+        ObjectNode msg = buildEnrichedLastMessage(heldTerminalLine, parsed);
+        String json = objectMapper.writeValueAsString(msg);
+        SseStreamSupport.writeDataEvent(output, json);
+        log.info("SSE 最后消息已写出 next_action={} content_type={} draftId={}",
+                msg.path("next_action").asText(),
+                msg.path("content_type").asText(),
+                parsed.getRelatedDraftId());
+    }
+
+    private ObjectNode buildEnrichedLastMessage(String heldTerminalLine, ParsedAssistantReply parsed)
+            throws IOException {
+        ObjectNode msg;
+        if (StringUtils.hasText(heldTerminalLine)) {
+            String payload = ConsoleChatSseAggregator.normalizeDataPayload(heldTerminalLine.trim());
+            JsonNode node = objectMapper.readTree(payload);
+            if (node != null && node.isObject()) {
+                msg = (ObjectNode) node;
+            } else {
+                msg = objectMapper.createObjectNode();
+                msg.put("object", "message");
+                msg.put("type", "message");
+                msg.put("role", "assistant");
+                msg.put("status", "completed");
+            }
+        } else {
+            msg = objectMapper.createObjectNode();
+            msg.put("object", "message");
+            msg.put("type", "message");
+            msg.put("role", "assistant");
+            msg.put("status", "completed");
+            ArrayNode content = msg.putArray("content");
+            ObjectNode textPart = content.addObject();
+            textPart.put("type", "text");
+            textPart.put("text", parsed.getDisplayText() != null ? parsed.getDisplayText() : "");
+        }
+
+        // 强制覆盖/追加卡片字段（不用智能体返回）
+        msg.put("content_type", MessageCardConstants.CONTENT_TYPE_CARD_DRAFT);
+        msg.put("next_action", MessageCardConstants.NEXT_ACTION_SHOW_EMPLOYMENT_DRAFT);
+        if (parsed.getRelatedDraftId() != null) {
+            msg.put("related_draft_id", parsed.getRelatedDraftId());
+        }
+        if (StringUtils.hasText(parsed.getCardPayload())) {
+            msg.set("card_payload", objectMapper.readTree(parsed.getCardPayload()));
+        }
+        if (!msg.has("status") || msg.get("status").asText("").isEmpty()) {
+            msg.put("status", "completed");
+        }
+        return msg;
+    }
+
     private byte[] buildUpstreamBody(FeConversation conversation, Long userId, String userText) throws IOException {
         ObjectNode root = objectMapper.createObjectNode();
         ArrayNode input = objectMapper.createArrayNode();

huimv-employment/fe-service/src/main/java/com/huimv/employment/service/conversation/ConversationDraftCardEnricher.java

@@ -18,7 +18,7 @@ import java.util.Map;
 import java.util.regex.Pattern;
 
 /**
- * SSE complete 后，由服务端根据会话 pending+ready 草稿写入卡片字段（不依赖智能体返回 card JSON）。
+ * SSE 最后消息时，由服务端根据会话 pending 且 missingFields 为空的草稿注入卡片字段（不依赖智能体）。
  */
 @Component
 public class ConversationDraftCardEnricher {
@@ -27,7 +27,6 @@ public class ConversationDraftCardEnricher {
 
     private static final String DB_STATUS_PENDING = "pending";
 
-    /** 剥除智能体可能附带的 fe-card / json 代码块，仅保留展示文案 */
     private static final Pattern FENCED_CARD_BLOCK = Pattern.compile(
             "```(?:fe-card|json)?\\s*[\\s\\S]*?```",
             Pattern.CASE_INSENSITIVE);
@@ -42,35 +41,37 @@ public class ConversationDraftCardEnricher {
     }
 
     /**
-     * SSE 流结束（complete）后调用：若草稿已 ready，强制写入 card_draft 字段。
+     * 会话有 pending 草稿且 missingFields 为空时追加 card_draft。
      */
     public ParsedAssistantReply attachCardOnComplete(Long conversationId, String assistantText) {
         String displayText = sanitizeDisplayText(assistantText);
         ParsedAssistantReply parsed = ParsedAssistantReply.textOnly(displayText);
         if (conversationId == null) {
+            log.warn("SSE 注入卡片跳过：conversationId 为空");
             return parsed;
         }
         FeEmploymentDraft draft = findPendingDraft(conversationId);
         if (draft == null) {
-            log.debug("会话 {} SSE complete 后无 pending 草稿，按纯文本落库", conversationId);
+            log.info("SSE 注入卡片跳过：会话 {} 无 pending 草稿", conversationId);
             return parsed;
         }
-        if (!isReady(draft)) {
-            log.debug("会话 {} 草稿 {} 尚未 ready（missing_fields={}），按纯文本落库",
-                    conversationId, draft.getId(), draft.getMissingFields());
+        List<String> missing = parseMissingFields(draft.getMissingFields());
+        if (!missing.isEmpty()) {
+            log.info("SSE 注入卡片跳过：会话 {} 草稿 {} missingFields 非空 {}",
+                    conversationId, draft.getId(), missing);
             return parsed;
         }
         try {
             Map<String, Object> payload = buildDraftCardPayload(draft);
             String cardPayload = objectMapper.writeValueAsString(payload);
-            log.info("会话 {} SSE complete，服务端写入 card_draft draftId={}", conversationId, draft.getId());
+            log.info("SSE 注入卡片 conversationId={} draftId={}", conversationId, draft.getId());
             return parsed.withCard(
                     MessageCardConstants.CONTENT_TYPE_CARD_DRAFT,
                     MessageCardConstants.NEXT_ACTION_SHOW_EMPLOYMENT_DRAFT,
                     cardPayload,
                     draft.getId());
         } catch (Exception ex) {
-            log.warn("会话 {} 草稿卡片写入失败: {}", conversationId, ex.getMessage());
+            log.warn("会话 {} 草稿卡片注入失败: {}", conversationId, ex.getMessage(), ex);
             return parsed;
         }
     }
@@ -91,10 +92,6 @@ public class ConversationDraftCardEnricher {
                 .last("LIMIT 1"));
     }
 
-    private boolean isReady(FeEmploymentDraft draft) {
-        return parseMissingFields(draft.getMissingFields()).isEmpty();
-    }
-
     private List<String> parseMissingFields(String json) {
         if (!StringUtils.hasText(json)) {
             return Collections.emptyList();
@@ -104,7 +101,9 @@ public class ConversationDraftCardEnricher {
             });
             return parsed != null ? parsed : Collections.emptyList();
         } catch (Exception ex) {
-            return Collections.emptyList();
+            log.warn("解析 missing_fields 失败，视为非空，跳过注入: {}", ex.getMessage());
+            // 解析失败时保守处理：不注入，避免误推未就绪草稿
+            return Collections.singletonList("_parse_error");
         }
     }