|
@@ -4,7 +4,9 @@ import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
|
|
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
|
|
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
|
|
|
import com.baomidou.mybatisplus.spring.service.impl.ServiceImpl;
|
|
import com.baomidou.mybatisplus.spring.service.impl.ServiceImpl;
|
|
|
import com.fasterxml.jackson.core.JsonProcessingException;
|
|
import com.fasterxml.jackson.core.JsonProcessingException;
|
|
|
|
|
+import com.fasterxml.jackson.databind.JsonNode;
|
|
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
|
|
|
|
+import com.fasterxml.jackson.databind.node.ObjectNode;
|
|
|
import com.zsjz.ai.common.exception.ServerException;
|
|
import com.zsjz.ai.common.exception.ServerException;
|
|
|
import com.zsjz.ai.module.agent.dto.ChatRequestDTO;
|
|
import com.zsjz.ai.module.agent.dto.ChatRequestDTO;
|
|
|
import com.zsjz.ai.module.agent.dto.CreateChatSessionDTO;
|
|
import com.zsjz.ai.module.agent.dto.CreateChatSessionDTO;
|
|
@@ -284,7 +286,7 @@ public class AgentChatServiceImpl extends ServiceImpl<AgentChatSessionMapper, Ag
|
|
|
return Flux.defer(() -> doStream(req))
|
|
return Flux.defer(() -> doStream(req))
|
|
|
.onErrorResume(ex -> {
|
|
.onErrorResume(ex -> {
|
|
|
String errorMsg = ex.getMessage() != null ? ex.getMessage() : "未知错误";
|
|
String errorMsg = ex.getMessage() != null ? ex.getMessage() : "未知错误";
|
|
|
- log.warn("Chat stream error: sessionKey={}, error={}", req.getSessionKey(), errorMsg);
|
|
|
|
|
|
|
+ log.warn("Chat stream error: sessionKey={}, ", req.getSessionKey(), ex);
|
|
|
// 错误信息持久化:确保刷新后仍可见
|
|
// 错误信息持久化:确保刷新后仍可见
|
|
|
Long sessionId = parseSessionId(req.getSessionKey());
|
|
Long sessionId = parseSessionId(req.getSessionKey());
|
|
|
if (sessionId != null) {
|
|
if (sessionId != null) {
|
|
@@ -364,6 +366,8 @@ public class AgentChatServiceImpl extends ServiceImpl<AgentChatSessionMapper, Ag
|
|
|
|
|
|
|
|
// 流中断追踪:用户主动中止时,若 AgentResultEvent 尚未到达,将已累积的文本保存
|
|
// 流中断追踪:用户主动中止时,若 AgentResultEvent 尚未到达,将已累积的文本保存
|
|
|
AtomicReference<StringBuilder> accumulatedText = new AtomicReference<>(new StringBuilder());
|
|
AtomicReference<StringBuilder> accumulatedText = new AtomicReference<>(new StringBuilder());
|
|
|
|
|
+ // 模型思考过程累积(reasoning_content / thinking):随流结束后写入消息 metadata
|
|
|
|
|
+ AtomicReference<StringBuilder> accumulatedThinking = new AtomicReference<>(new StringBuilder());
|
|
|
AtomicBoolean resultPersisted = new AtomicBoolean(false);
|
|
AtomicBoolean resultPersisted = new AtomicBoolean(false);
|
|
|
|
|
|
|
|
return events.concatMap(event -> {
|
|
return events.concatMap(event -> {
|
|
@@ -411,11 +415,30 @@ public class AgentChatServiceImpl extends ServiceImpl<AgentChatSessionMapper, Ag
|
|
|
} else if (event instanceof TextBlockDeltaEvent e) {
|
|
} else if (event instanceof TextBlockDeltaEvent e) {
|
|
|
accumulatedText.get().append(e.getDelta());
|
|
accumulatedText.get().append(e.getDelta());
|
|
|
frames.add(sse("token", Map.of("type", "token", "data", e.getDelta())));
|
|
frames.add(sse("token", Map.of("type", "token", "data", e.getDelta())));
|
|
|
|
|
+ } else if (event instanceof ThinkingBlockStartEvent) {
|
|
|
|
|
+ // 模型思考过程(reasoning_content / thinking)开始:一轮 ReAct 推理可能有多段思考,
|
|
|
|
|
+ // 段间补空行分隔,保证前端拼接后可读
|
|
|
|
|
+ StringBuilder thinking = accumulatedThinking.get();
|
|
|
|
|
+ if (thinking.length() > 0) {
|
|
|
|
|
+ thinking.append("\n\n");
|
|
|
|
|
+ }
|
|
|
|
|
+ frames.add(sse("thinking", Map.of("type", "thinking", "phase", "start")));
|
|
|
|
|
+ } else if (event instanceof ThinkingBlockDeltaEvent e) {
|
|
|
|
|
+ String delta = e.getDelta() != null ? e.getDelta() : "";
|
|
|
|
|
+ accumulatedThinking.get().append(delta);
|
|
|
|
|
+ Map<String, Object> data = new LinkedHashMap<>();
|
|
|
|
|
+ data.put("type", "thinking");
|
|
|
|
|
+ data.put("phase", "delta");
|
|
|
|
|
+ data.put("data", delta);
|
|
|
|
|
+ frames.add(sse("thinking", data));
|
|
|
|
|
+ } else if (event instanceof ThinkingBlockEndEvent) {
|
|
|
|
|
+ frames.add(sse("thinking", Map.of("type", "thinking", "phase", "end")));
|
|
|
} else if (event instanceof AgentResultEvent e) {
|
|
} else if (event instanceof AgentResultEvent e) {
|
|
|
- // AI 回复落库(异步,不阻塞事件流);工具调用事件一并持久化成 tool_events
|
|
|
|
|
|
|
+ // AI 回复落库(异步,不阻塞事件流);工具调用事件与思考过程一并持久化
|
|
|
resultPersisted.set(true);
|
|
resultPersisted.set(true);
|
|
|
String toolEventsJson = buildToolEventsJson(toolInputs, toolResults, toolNames);
|
|
String toolEventsJson = buildToolEventsJson(toolInputs, toolResults, toolNames);
|
|
|
- Mono.fromRunnable(() -> persistAssistantMessage(sessionId, e.getResult(), toolEventsJson))
|
|
|
|
|
|
|
+ String thinkingJson = buildThinkingMetadata(accumulatedThinking.get().toString());
|
|
|
|
|
+ Mono.fromRunnable(() -> persistAssistantMessage(sessionId, e.getResult(), toolEventsJson, thinkingJson))
|
|
|
.subscribeOn(Schedulers.boundedElastic())
|
|
.subscribeOn(Schedulers.boundedElastic())
|
|
|
.subscribe();
|
|
.subscribe();
|
|
|
} else if (event instanceof AgentEndEvent) {
|
|
} else if (event instanceof AgentEndEvent) {
|
|
@@ -428,8 +451,9 @@ public class AgentChatServiceImpl extends ServiceImpl<AgentChatSessionMapper, Ag
|
|
|
if (!resultPersisted.get()) {
|
|
if (!resultPersisted.get()) {
|
|
|
String partialText = accumulatedText.get().toString().trim();
|
|
String partialText = accumulatedText.get().toString().trim();
|
|
|
String toolEventsJson = buildToolEventsJson(toolInputs, toolResults, toolNames);
|
|
String toolEventsJson = buildToolEventsJson(toolInputs, toolResults, toolNames);
|
|
|
|
|
+ String thinkingJson = buildThinkingMetadata(accumulatedThinking.get().toString());
|
|
|
if (!partialText.isEmpty() || !toolEventsJson.isEmpty()) {
|
|
if (!partialText.isEmpty() || !toolEventsJson.isEmpty()) {
|
|
|
- Mono.fromRunnable(() -> persistPartialMessage(sessionId, partialText, toolEventsJson))
|
|
|
|
|
|
|
+ Mono.fromRunnable(() -> persistPartialMessage(sessionId, partialText, toolEventsJson, thinkingJson))
|
|
|
.subscribeOn(Schedulers.boundedElastic())
|
|
.subscribeOn(Schedulers.boundedElastic())
|
|
|
.subscribe();
|
|
.subscribe();
|
|
|
}
|
|
}
|
|
@@ -456,8 +480,9 @@ public class AgentChatServiceImpl extends ServiceImpl<AgentChatSessionMapper, Ag
|
|
|
* 将 AI 回复持久化为 assistant 消息。失败不影响流式输出。
|
|
* 将 AI 回复持久化为 assistant 消息。失败不影响流式输出。
|
|
|
*
|
|
*
|
|
|
* @param toolEventsJson 工具调用事件 JSON(execute_sql 入参/结果等),可为 null
|
|
* @param toolEventsJson 工具调用事件 JSON(execute_sql 入参/结果等),可为 null
|
|
|
|
|
+ * @param thinkingJson 思考过程 metadata JSON({@code {"thinking": "..."}}),可为 null
|
|
|
*/
|
|
*/
|
|
|
- private void persistAssistantMessage(Long sessionId, Msg result, String toolEventsJson) {
|
|
|
|
|
|
|
+ private void persistAssistantMessage(Long sessionId, Msg result, String toolEventsJson, String thinkingJson) {
|
|
|
try {
|
|
try {
|
|
|
AgentChatSession session = chatSessionMapper.selectById(sessionId);
|
|
AgentChatSession session = chatSessionMapper.selectById(sessionId);
|
|
|
if (session == null) {
|
|
if (session == null) {
|
|
@@ -475,6 +500,7 @@ public class AgentChatServiceImpl extends ServiceImpl<AgentChatSessionMapper, Ag
|
|
|
assistantMessage.setMessageType("text");
|
|
assistantMessage.setMessageType("text");
|
|
|
assistantMessage.setStarred(false);
|
|
assistantMessage.setStarred(false);
|
|
|
assistantMessage.setToolEvents(toolEventsJson);
|
|
assistantMessage.setToolEvents(toolEventsJson);
|
|
|
|
|
+ assistantMessage.setMetadata(thinkingJson);
|
|
|
assistantMessage.setCreateAt(LocalDateTime.now());
|
|
assistantMessage.setCreateAt(LocalDateTime.now());
|
|
|
messageMapper.insert(assistantMessage);
|
|
messageMapper.insert(assistantMessage);
|
|
|
chatSessionMapper.update(
|
|
chatSessionMapper.update(
|
|
@@ -556,7 +582,7 @@ public class AgentChatServiceImpl extends ServiceImpl<AgentChatSessionMapper, Ag
|
|
|
/**
|
|
/**
|
|
|
* 持久化用户中断时的部分内容(messageType=interrupted),标记为不完整回复。
|
|
* 持久化用户中断时的部分内容(messageType=interrupted),标记为不完整回复。
|
|
|
*/
|
|
*/
|
|
|
- private void persistPartialMessage(Long sessionId, String partialText, String toolEventsJson) {
|
|
|
|
|
|
|
+ private void persistPartialMessage(Long sessionId, String partialText, String toolEventsJson, String thinkingJson) {
|
|
|
try {
|
|
try {
|
|
|
AgentMessage msg = new AgentMessage();
|
|
AgentMessage msg = new AgentMessage();
|
|
|
msg.setSessionId(sessionId);
|
|
msg.setSessionId(sessionId);
|
|
@@ -564,7 +590,7 @@ public class AgentChatServiceImpl extends ServiceImpl<AgentChatSessionMapper, Ag
|
|
|
msg.setContent(partialText.isEmpty() ? "(回复已中断)" : partialText + "\n\n*(回复已中断)*");
|
|
msg.setContent(partialText.isEmpty() ? "(回复已中断)" : partialText + "\n\n*(回复已中断)*");
|
|
|
msg.setContentType("text");
|
|
msg.setContentType("text");
|
|
|
msg.setMessageType("interrupted");
|
|
msg.setMessageType("interrupted");
|
|
|
- msg.setMetadata("{\"interrupted\": true}");
|
|
|
|
|
|
|
+ msg.setMetadata(mergeThinkingMetadata("{\"interrupted\": true}", thinkingJson));
|
|
|
msg.setToolEvents(toolEventsJson);
|
|
msg.setToolEvents(toolEventsJson);
|
|
|
msg.setStarred(false);
|
|
msg.setStarred(false);
|
|
|
msg.setCreateAt(LocalDateTime.now());
|
|
msg.setCreateAt(LocalDateTime.now());
|
|
@@ -599,6 +625,49 @@ public class AgentChatServiceImpl extends ServiceImpl<AgentChatSessionMapper, Ag
|
|
|
return UUID.randomUUID().getMostSignificantBits() & Long.MAX_VALUE;
|
|
return UUID.randomUUID().getMostSignificantBits() & Long.MAX_VALUE;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+ /**
|
|
|
|
|
+ * 把流内累积的思考过程包成 metadata JSON({@code {"thinking": "..."}})。
|
|
|
|
|
+ *
|
|
|
|
|
+ * <p>思考过程存 agent_message.metadata(jsonb)而不是单独建列:刷新会话后前端仍能还原
|
|
|
|
|
+ * 「思考过程」面板;无思考内容时返回 null,保持 metadata 原样。
|
|
|
|
|
+ */
|
|
|
|
|
+ private static String buildThinkingMetadata(String thinking) {
|
|
|
|
|
+ String text = thinking == null ? "" : thinking.trim();
|
|
|
|
|
+ if (text.isEmpty()) {
|
|
|
|
|
+ return null;
|
|
|
|
|
+ }
|
|
|
|
|
+ try {
|
|
|
|
|
+ ObjectNode node = MAPPER.createObjectNode();
|
|
|
|
|
+ node.put("thinking", text);
|
|
|
|
|
+ return MAPPER.writeValueAsString(node);
|
|
|
|
|
+ } catch (JsonProcessingException e) {
|
|
|
|
|
+ return null;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ /**
|
|
|
|
|
+ * 把思考过程合并进已有 metadata JSON(用于中断消息等场景,保留原有标志位)。
|
|
|
|
|
+ *
|
|
|
|
|
+ * @param baseJson 原有 metadata JSON(如 {@code {"interrupted": true}})
|
|
|
|
|
+ * @param thinkingJson {@link #buildThinkingMetadata(String)} 的产物,可为 null
|
|
|
|
|
+ */
|
|
|
|
|
+ private static String mergeThinkingMetadata(String baseJson, String thinkingJson) {
|
|
|
|
|
+ if (thinkingJson == null) {
|
|
|
|
|
+ return baseJson;
|
|
|
|
|
+ }
|
|
|
|
|
+ try {
|
|
|
|
|
+ JsonNode base = MAPPER.readTree(baseJson);
|
|
|
|
|
+ JsonNode thinking = MAPPER.readTree(thinkingJson).get("thinking");
|
|
|
|
|
+ if (!(base instanceof ObjectNode obj) || thinking == null) {
|
|
|
|
|
+ return baseJson;
|
|
|
|
|
+ }
|
|
|
|
|
+ obj.set("thinking", thinking);
|
|
|
|
|
+ return MAPPER.writeValueAsString(obj);
|
|
|
|
|
+ } catch (Exception e) {
|
|
|
|
|
+ return baseJson;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
/**
|
|
/**
|
|
|
* 将流内累积的工具调用聚合为 JSON 字符串(tool_events 列存储形态,与前端 ToolStep 对齐)。
|
|
* 将流内累积的工具调用聚合为 JSON 字符串(tool_events 列存储形态,与前端 ToolStep 对齐)。
|
|
|
*
|
|
*
|