增加与ai大模型对话

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

@@ -113,8 +113,7 @@ public class ConversationController {
     @Operation(summary = "会话内 AI 对话（SSE 流式 + 落库）",
             description = "需先 POST /conversations 创建会话。请求体仅需 message 字段；"
                     + "服务端组装上游 console/chat 请求，流式开始前写入 user 消息，"
-                    + "流式结束后解析 assistant 卡片（content_type/card_payload/next_action）并落库；"
-                    + "若 Agent 未返回卡片且会话草稿 status=ready，服务端自动补全 card_draft。"
+                    + "SSE complete 后由服务端根据 ready 草稿写入 content_type/card_payload/next_action（不依赖智能体返回卡片），"
                     + "可选请求头 X-Agent-Id。")
     public void chat(@PathVariable("id") Long conversationId,
                      @Validated @RequestBody ConversationChatRequest request,

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

@@ -2,17 +2,30 @@ package com.huimv.employment.integration.kb;
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import org.springframework.util.StringUtils;
 
 import java.nio.charset.StandardCharsets;
+import java.util.HashSet;
+import java.util.Set;
 
 /**
- * 从 {@code /api/console/chat} SSE 字节流中增量提取 assistant 文本。
+ * 从 {@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>
  */
 public class ConsoleChatSseAggregator {
 
     private final ObjectMapper objectMapper;
     private final StringBuilder lineBuffer = new StringBuilder();
-    private final StringBuilder assistantText = 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;
 
     public ConsoleChatSseAggregator(ObjectMapper objectMapper) {
         this.objectMapper = objectMapper;
@@ -26,8 +39,26 @@ public class ConsoleChatSseAggregator {
         drainCompleteLines();
     }
 
+    /** 流结束时刷新缓冲区中未换行的尾部行。 */
+    public void finish() {
+        drainCompleteLines();
+        if (lineBuffer.length() > 0) {
+            consumeSseLine(lineBuffer.toString().trim());
+            lineBuffer.setLength(0);
+        }
+        // 流关闭本身也视为一轮对话结束
+        completed = true;
+    }
+
+    public boolean isCompleted() {
+        return completed;
+    }
+
     public String getAssistantText() {
-        return assistantText.toString();
+        if (completedReply.length() > 0) {
+            return completedReply.toString();
+        }
+        return streamingText.toString();
     }
 
     private void drainCompleteLines() {
@@ -40,36 +71,157 @@ public class ConsoleChatSseAggregator {
     }
 
     private void consumeSseLine(String line) {
-        if (!line.startsWith("data:")) {
+        String payload = normalizeDataPayload(line);
+        if (!StringUtils.hasText(payload)) {
             return;
         }
-        String payload = line.substring(5).trim();
-        if (payload.isEmpty() || "[DONE]".equals(payload)) {
+        if ("[DONE]".equals(payload)) {
+            completed = true;
             return;
         }
         try {
             JsonNode data = objectMapper.readTree(payload);
-            JsonNode output = data.get("output");
-            if (output == null || !output.isArray()) {
-                return;
+            consumeJsonEvent(data);
+        } catch (Exception ignored) {
+            // 跳过无法解析的 SSE 片段，不影响流式转发
+        }
+    }
+
+    private void consumeJsonEvent(JsonNode data) {
+        if (data == null || data.isNull()) {
+            return;
+        }
+        String objectType = data.path("object").asText("");
+        String status = data.path("status").asText("");
+
+        if ("message".equals(objectType) && "reasoning".equals(data.path("type").asText())) {
+            String messageId = data.path("id").asText(null);
+            if (StringUtils.hasText(messageId)) {
+                reasoningMessageIds.add(messageId);
             }
-            for (JsonNode item : output) {
-                if (!"assistant".equals(item.path("role").asText())) {
-                    continue;
+            return;
+        }
+
+        // 最后一条 complete：response 完成态
+        if (("response".equals(objectType) || "chat.completion".equals(objectType))
+                && ("completed".equals(status) || "complete".equals(status))) {
+            completed = true;
+            if (data.has("output")) {
+                String reply = extractAssistantReplyFromOutput(data.get("output"));
+                if (StringUtils.hasText(reply)) {
+                    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"));
+                if (StringUtils.hasText(reply)) {
+                    completedReply.setLength(0);
+                    completedReply.append(reply);
                 }
-                JsonNode contents = item.get("content");
-                if (contents == null || !contents.isArray()) {
-                    continue;
+            }
+            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);
+                if (StringUtils.hasText(msgId) && reasoningMessageIds.contains(msgId)) {
+                    return;
                 }
-                for (JsonNode content : contents) {
-                    if ("text".equals(content.path("type").asText()) && content.has("text")) {
-                        assistantText.append(content.get("text").asText(""));
-                    }
+                streamingText.append(data.get("text").asText(""));
+            }
+            return;
+        }
+
+        JsonNode output = data.get("output");
+        if (output != null && output.isArray()) {
+            String reply = extractAssistantReplyFromOutput(output);
+            if (StringUtils.hasText(reply) && completedReply.length() == 0 && streamingText.length() == 0) {
+                streamingText.append(reply);
+            }
+        }
+    }
+
+    private static String extractAssistantReplyFromOutput(JsonNode output) {
+        if (output == null || !output.isArray()) {
+            return "";
+        }
+        StringBuilder reply = new StringBuilder();
+        for (JsonNode item : output) {
+            if (!"assistant".equals(item.path("role").asText())) {
+                continue;
+            }
+            if (!"message".equals(item.path("type").asText("message"))) {
+                continue;
+            }
+            String text = extractTextFromContentNodes(item.get("content"));
+            if (StringUtils.hasText(text)) {
+                if (reply.length() > 0) {
+                    reply.append('\n');
                 }
+                reply.append(text);
             }
-        } catch (Exception ignored) {
-            // 跳过无法解析的 SSE 片段，不影响流式转发
         }
+        return reply.toString();
+    }
+
+    private static String extractTextFromContentNodes(JsonNode contents) {
+        if (contents == null || !contents.isArray()) {
+            return "";
+        }
+        StringBuilder text = new StringBuilder();
+        for (JsonNode content : contents) {
+            appendTextPart(text, content);
+        }
+        return text.toString();
+    }
+
+    private static void appendTextPart(StringBuilder target, JsonNode content) {
+        if (content == null || content.isNull()) {
+            return;
+        }
+        if (content.has("text") && "text".equals(content.path("type").asText("text"))) {
+            target.append(content.get("text").asText(""));
+            return;
+        }
+        if (content.has("text") && content.has("object") && "content".equals(content.path("object").asText())) {
+            target.append(content.get("text").asText(""));
+        }
+    }
+
+    /** 去掉一层或多层 {@code data:} 前缀，得到 JSON 载荷。 */
+    static String normalizeDataPayload(String line) {
+        if (!StringUtils.hasText(line)) {
+            return "";
+        }
+        String payload = line.trim();
+        while (payload.startsWith("data:")) {
+            payload = payload.substring(5).trim();
+        }
+        return payload;
     }
 
     private static int indexOfLineBreak(StringBuilder sb) {

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

@@ -18,7 +18,9 @@ public interface KbOpenAiProxyService {
 
     /**
      * 流式转发 SSE 并聚合 assistant 文本（供会话落库场景使用，不影响纯透传接口）。
+     *
+     * @return 助手正文 + 是否收到 complete / 流结束
      */
-    String streamConsoleChatCollecting(byte[] requestBody, String agentId, OutputStream output,
-                                       ObjectMapper objectMapper) throws IOException;
+    ConsoleChatCollectResult streamConsoleChatCollecting(byte[] requestBody, String agentId, OutputStream output,
+                                                         ObjectMapper objectMapper) throws IOException;
 }

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

@@ -62,11 +62,12 @@ public class KbOpenAiProxyServiceImpl implements KbOpenAiProxyService {
     }
 
     @Override
-    public String streamConsoleChatCollecting(byte[] requestBody, String agentId, OutputStream output,
-                                              ObjectMapper objectMapper) throws IOException {
+    public ConsoleChatCollectResult streamConsoleChatCollecting(byte[] requestBody, String agentId, OutputStream output,
+                                                                ObjectMapper objectMapper) throws IOException {
         ConsoleChatSseAggregator aggregator = new ConsoleChatSseAggregator(objectMapper);
         streamConsoleChatInternal(requestBody, agentId, output, aggregator);
-        return aggregator.getAssistantText();
+        aggregator.finish();
+        return new ConsoleChatCollectResult(aggregator.getAssistantText(), aggregator.isCompleted());
     }
 
     private void streamConsoleChatInternal(byte[] requestBody, String agentId, OutputStream output,

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

@@ -4,6 +4,7 @@ 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.KbOpenAiProxyService;
 import com.huimv.employment.service.conversation.dto.ConversationChatRequest;
 import com.huimv.employment.service.conversation.dto.ParsedAssistantReply;
@@ -17,6 +18,9 @@ import java.io.OutputStream;
 
 /**
  * 会话内 AI 对话：组装上游请求、流式转发并在结束后落库。
+ * <p>
+ * 卡片字段由服务端在 SSE complete 后根据草稿状态写入，不依赖智能体返回 card JSON。
+ * </p>
  */
 @Service
 public class ConversationChatService {
@@ -59,13 +63,20 @@ public class ConversationChatService {
             conversationIntentService.classifyAndUpdateTitleAsync(
                     conversation.getConversationNo(), userText, agentId);
         }
-        String assistantText = kbOpenAiProxyService.streamConsoleChatCollecting(
+
+        ConsoleChatCollectResult collectResult = kbOpenAiProxyService.streamConsoleChatCollecting(
                 upstreamBody, agentId, output, objectMapper);
-        ParsedAssistantReply parsed = AssistantReplyParser.parse(assistantText, objectMapper);
-        parsed = conversationDraftCardEnricher.enrichIfReady(conversation.getId(), parsed);
+        // SSE complete（或流正常关闭）后：服务端按 ready 草稿写入卡片，忽略智能体卡片 JSON
+        ParsedAssistantReply parsed = conversationDraftCardEnricher.attachCardOnComplete(
+                conversation.getId(), collectResult.getAssistantText());
+
         if (!StringUtils.hasText(parsed.getDisplayText()) && !parsed.hasCard()) {
             log.warn("assistant 回复为空，未写入 fe_conversation_message conversationId={} conversationNo={}",
                     conversation.getId(), conversation.getConversationNo());
+        } else if (parsed.hasCard()) {
+            log.info("SSE complete 落库卡片 conversationId={} completed={} contentType={} relatedDraftId={}",
+                    conversation.getId(), collectResult.isCompleted(),
+                    parsed.getContentType(), parsed.getRelatedDraftId());
         }
         conversationMessageService.saveAssistantMessage(conversation.getId(), parsed);
     }

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

@@ -15,9 +15,10 @@ import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.regex.Pattern;
 
 /**
- * 当 Agent 未返回卡片 JSON 时，根据会话 pending 且 ready 的草稿自动补全 card_draft 消息。
+ * SSE complete 后，由服务端根据会话 pending+ready 草稿写入卡片字段（不依赖智能体返回 card JSON）。
  */
 @Component
 public class ConversationDraftCardEnricher {
@@ -26,6 +27,11 @@ 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);
+
     private final FeEmploymentDraftMapper feEmploymentDraftMapper;
     private final ObjectMapper objectMapper;
 
@@ -35,29 +41,48 @@ public class ConversationDraftCardEnricher {
         this.objectMapper = objectMapper;
     }
 
-    public ParsedAssistantReply enrichIfReady(Long conversationId, ParsedAssistantReply parsed) {
-        if (parsed == null || parsed.hasCard() || conversationId == null) {
+    /**
+     * SSE 流结束（complete）后调用：若草稿已 ready，强制写入 card_draft 字段。
+     */
+    public ParsedAssistantReply attachCardOnComplete(Long conversationId, String assistantText) {
+        String displayText = sanitizeDisplayText(assistantText);
+        ParsedAssistantReply parsed = ParsedAssistantReply.textOnly(displayText);
+        if (conversationId == null) {
             return parsed;
         }
         FeEmploymentDraft draft = findPendingDraft(conversationId);
-        if (draft == null || !isReady(draft)) {
+        if (draft == null) {
+            log.debug("会话 {} SSE complete 后无 pending 草稿，按纯文本落库", conversationId);
+            return parsed;
+        }
+        if (!isReady(draft)) {
+            log.debug("会话 {} 草稿 {} 尚未 ready（missing_fields={}），按纯文本落库",
+                    conversationId, draft.getId(), draft.getMissingFields());
             return parsed;
         }
         try {
             Map<String, Object> payload = buildDraftCardPayload(draft);
             String cardPayload = objectMapper.writeValueAsString(payload);
-            log.debug("会话 {} 草稿 {} 已 ready，自动补全 card_draft 消息", conversationId, draft.getId());
+            log.info("会话 {} SSE complete，服务端写入 card_draft 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());
             return parsed;
         }
     }
 
+    private String sanitizeDisplayText(String raw) {
+        if (!StringUtils.hasText(raw)) {
+            return "";
+        }
+        String cleaned = FENCED_CARD_BLOCK.matcher(raw).replaceAll("").trim();
+        return StringUtils.hasText(cleaned) ? cleaned : raw.trim();
+    }
+
     private FeEmploymentDraft findPendingDraft(Long conversationId) {
         return feEmploymentDraftMapper.selectOne(new LambdaQueryWrapper<FeEmploymentDraft>()
                 .eq(FeEmploymentDraft::getConversationId, conversationId)