|
@@ -0,0 +1,621 @@
|
|
|
|
|
+package com.zsjz.ai.module.agent.service.impl;
|
|
|
|
|
+
|
|
|
|
|
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
|
|
|
|
+import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
|
|
|
|
|
+import com.baomidou.mybatisplus.spring.service.impl.ServiceImpl;
|
|
|
|
|
+import com.fasterxml.jackson.core.JsonProcessingException;
|
|
|
|
|
+import com.fasterxml.jackson.databind.ObjectMapper;
|
|
|
|
|
+import com.zsjz.ai.common.exception.ServerException;
|
|
|
|
|
+import com.zsjz.ai.module.agent.dto.ChatRequestDTO;
|
|
|
|
|
+import com.zsjz.ai.module.agent.dto.CreateChatSessionDTO;
|
|
|
|
|
+import com.zsjz.ai.module.agent.dto.SendMessageDTO;
|
|
|
|
|
+import com.zsjz.ai.module.agent.entity.AgentChatSession;
|
|
|
|
|
+import com.zsjz.ai.module.agent.entity.AgentMessage;
|
|
|
|
|
+import com.zsjz.ai.module.agent.mapper.AgentChatSessionMapper;
|
|
|
|
|
+import com.zsjz.ai.module.agent.mapper.AgentMessageMapper;
|
|
|
|
|
+import com.zsjz.ai.module.agent.service.AgentChatService;
|
|
|
|
|
+import com.zsjz.ai.module.agent.service.AgentModelService;
|
|
|
|
|
+import com.zsjz.ai.module.agent.service.AgentService;
|
|
|
|
|
+import com.zsjz.ai.module.agent.usage.UsageStore;
|
|
|
|
|
+import com.zsjz.ai.module.agent.vo.ChatSessionVO;
|
|
|
|
|
+import com.zsjz.ai.module.agent.vo.MessageVO;
|
|
|
|
|
+import io.agentscope.core.agent.RuntimeContext;
|
|
|
|
|
+import io.agentscope.core.event.*;
|
|
|
|
|
+import io.agentscope.core.message.Msg;
|
|
|
|
|
+import io.agentscope.core.message.MsgRole;
|
|
|
|
|
+import io.agentscope.harness.agent.HarnessAgent;
|
|
|
|
|
+import lombok.extern.slf4j.Slf4j;
|
|
|
|
|
+import org.springframework.http.codec.ServerSentEvent;
|
|
|
|
|
+import org.springframework.stereotype.Service;
|
|
|
|
|
+import org.springframework.transaction.annotation.Transactional;
|
|
|
|
|
+import org.springframework.util.StringUtils;
|
|
|
|
|
+import reactor.core.publisher.Flux;
|
|
|
|
|
+import reactor.core.publisher.Mono;
|
|
|
|
|
+import reactor.core.scheduler.Schedulers;
|
|
|
|
|
+
|
|
|
|
|
+import java.time.LocalDateTime;
|
|
|
|
|
+import java.util.*;
|
|
|
|
|
+import java.util.concurrent.ConcurrentHashMap;
|
|
|
|
|
+import java.util.concurrent.atomic.AtomicBoolean;
|
|
|
|
|
+import java.util.concurrent.atomic.AtomicReference;
|
|
|
|
|
+import java.util.stream.Collectors;
|
|
|
|
|
+
|
|
|
|
|
+/**
|
|
|
|
|
+ * 聊天服务实现
|
|
|
|
|
+ *
|
|
|
|
|
+ * <p>流式对话走官方 API 直连:{@link HarnessAgent#streamEvents(List, RuntimeContext)}。
|
|
|
|
|
+ * 会话状态(对话历史)由官方 {@code AgentStateStore} 按 {@code (userId, sessionId)}
|
|
|
|
|
+ * 自动加载/保存;会话转录与 sessions.json 索引由官方 {@code TranscriptMiddleware}
|
|
|
|
|
+ * 在每轮对话结束后自动写入 agent 工作目录。
|
|
|
|
|
+ */
|
|
|
|
|
+@Slf4j
|
|
|
|
|
+@Service
|
|
|
|
|
+public class AgentChatServiceImpl extends ServiceImpl<AgentChatSessionMapper, AgentChatSession> implements AgentChatService {
|
|
|
|
|
+
|
|
|
|
|
+ private static final Long DEFAULT_AGENT_ROW_ID = 1L;
|
|
|
|
|
+
|
|
|
|
|
+ private final AgentChatSessionMapper chatSessionMapper;
|
|
|
|
|
+ private final AgentMessageMapper messageMapper;
|
|
|
|
|
+ private final AgentService agentService;
|
|
|
|
|
+ private final AgentModelService agentModelService;
|
|
|
|
|
+ private final UsageStore usageStore;
|
|
|
|
|
+ private static final ObjectMapper MAPPER = new ObjectMapper();
|
|
|
|
|
+
|
|
|
|
|
+ public AgentChatServiceImpl(UsageStore usageStore,
|
|
|
|
|
+ AgentService agentService,
|
|
|
|
|
+ AgentModelService agentModelService,
|
|
|
|
|
+ AgentMessageMapper messageMapper,
|
|
|
|
|
+ AgentChatSessionMapper chatSessionMapper) {
|
|
|
|
|
+ this.usageStore = usageStore;
|
|
|
|
|
+ this.agentService = agentService;
|
|
|
|
|
+ this.agentModelService = agentModelService;
|
|
|
|
|
+ this.messageMapper = messageMapper;
|
|
|
|
|
+ this.chatSessionMapper = chatSessionMapper;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ @Override
|
|
|
|
|
+ @Transactional(rollbackFor = Exception.class)
|
|
|
|
|
+ public ChatSessionVO createSession(CreateChatSessionDTO dto, Long userId) {
|
|
|
|
|
+ AgentChatSession entity = new AgentChatSession();
|
|
|
|
|
+ entity.setId(generateId());
|
|
|
|
|
+ entity.setCaseId(dto.getWorkspaceId());
|
|
|
|
|
+ entity.setAgentId(dto.getAgentId() != null ? dto.getAgentId() : DEFAULT_AGENT_ROW_ID);
|
|
|
|
|
+
|
|
|
|
|
+ entity.setTitle(dto.getTitle() != null ? dto.getTitle() : "新对话");
|
|
|
|
|
+ entity.setContext(dto.getContext() != null ? dto.getContext() : "{}");
|
|
|
|
|
+ // 模型绑定:请求指定 > 默认模型
|
|
|
|
|
+ entity.setModelId(dto.getModelId() != null ? dto.getModelId() : agentModelService.getDefaultModelId());
|
|
|
|
|
+ entity.setMessageCount(0);
|
|
|
|
|
+ entity.setTotalTokens(0L);
|
|
|
|
|
+ entity.setPinned(false);
|
|
|
|
|
+ entity.setCreateAt(LocalDateTime.now());
|
|
|
|
|
+ entity.setUpdateAt(LocalDateTime.now());
|
|
|
|
|
+
|
|
|
|
|
+ chatSessionMapper.insert(entity);
|
|
|
|
|
+ log.info("创建会话成功: {}, 标题: {}, 模型: {}", entity.getId(), entity.getTitle(), entity.getModelId());
|
|
|
|
|
+
|
|
|
|
|
+ return convertSessionToVO(entity);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ @Override
|
|
|
|
|
+ public List<ChatSessionVO> listSessions(Long workspaceId, Long userId) {
|
|
|
|
|
+ List<AgentChatSession> list = chatSessionMapper.selectList(
|
|
|
|
|
+ new LambdaQueryWrapper<AgentChatSession>()
|
|
|
|
|
+ .eq(AgentChatSession::getCaseId, workspaceId)
|
|
|
|
|
+ .orderByDesc(AgentChatSession::getPinned)
|
|
|
|
|
+ .orderByDesc(AgentChatSession::getLastMessageAt)
|
|
|
|
|
+ .orderByDesc(AgentChatSession::getCreateAt)
|
|
|
|
|
+ );
|
|
|
|
|
+ return list.stream()
|
|
|
|
|
+ .map(this::convertSessionToVO)
|
|
|
|
|
+ .collect(Collectors.toList());
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ @Override
|
|
|
|
|
+ public ChatSessionVO getSession(Long sessionId, Long userId) {
|
|
|
|
|
+ AgentChatSession entity = assertOwner(sessionId, userId);
|
|
|
|
|
+ return convertSessionToVO(entity);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ @Override
|
|
|
|
|
+ @Transactional(rollbackFor = Exception.class)
|
|
|
|
|
+ public void updateSessionTitle(Long sessionId, String title, Long userId) {
|
|
|
|
|
+ AgentChatSession entity = assertOwner(sessionId, userId);
|
|
|
|
|
+
|
|
|
|
|
+ chatSessionMapper.update(null,
|
|
|
|
|
+ new LambdaUpdateWrapper<AgentChatSession>()
|
|
|
|
|
+ .eq(AgentChatSession::getId, sessionId)
|
|
|
|
|
+ .set(AgentChatSession::getTitle, title)
|
|
|
|
|
+ .set(AgentChatSession::getUpdateAt, LocalDateTime.now())
|
|
|
|
|
+ );
|
|
|
|
|
+ log.info("更新会话标题成功: {}, 标题: {}", sessionId, title);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ @Override
|
|
|
|
|
+ @Transactional(rollbackFor = Exception.class)
|
|
|
|
|
+ public void deleteSession(Long sessionId, Long userId) {
|
|
|
|
|
+ assertOwner(sessionId, userId);
|
|
|
|
|
+
|
|
|
|
|
+ // 删除会话下的所有消息
|
|
|
|
|
+ messageMapper.delete(
|
|
|
|
|
+ new LambdaQueryWrapper<AgentMessage>()
|
|
|
|
|
+ .eq(AgentMessage::getSessionId, sessionId)
|
|
|
|
|
+ );
|
|
|
|
|
+
|
|
|
|
|
+ // 删除会话
|
|
|
|
|
+ chatSessionMapper.deleteById(sessionId);
|
|
|
|
|
+ log.info("删除会话成功: {}", sessionId);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ @Override
|
|
|
|
|
+ @Transactional(rollbackFor = Exception.class)
|
|
|
|
|
+ public void togglePinSession(Long sessionId, Long userId) {
|
|
|
|
|
+ AgentChatSession entity = assertOwner(sessionId, userId);
|
|
|
|
|
+
|
|
|
|
|
+ chatSessionMapper.update(null,
|
|
|
|
|
+ new LambdaUpdateWrapper<AgentChatSession>()
|
|
|
|
|
+ .eq(AgentChatSession::getId, sessionId)
|
|
|
|
|
+ .set(AgentChatSession::getPinned, !entity.getPinned())
|
|
|
|
|
+ .set(AgentChatSession::getUpdateAt, LocalDateTime.now())
|
|
|
|
|
+ );
|
|
|
|
|
+ log.info("切换会话置顶状态成功: {}, 置顶: {}", sessionId, !entity.getPinned());
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ @Override
|
|
|
|
|
+ public List<MessageVO> listMessages(Long sessionId, Long userId) {
|
|
|
|
|
+ assertOwner(sessionId, userId);
|
|
|
|
|
+ List<AgentMessage> list = messageMapper.selectList(
|
|
|
|
|
+ new LambdaQueryWrapper<AgentMessage>()
|
|
|
|
|
+ .eq(AgentMessage::getSessionId, sessionId)
|
|
|
|
|
+ .orderByAsc(AgentMessage::getCreateAt)
|
|
|
|
|
+ );
|
|
|
|
|
+ return list.stream()
|
|
|
|
|
+ .map(this::convertMessageToVO)
|
|
|
|
|
+ .collect(Collectors.toList());
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ @Override
|
|
|
|
|
+ @Transactional(rollbackFor = Exception.class)
|
|
|
|
|
+ public MessageVO sendMessage(SendMessageDTO dto, Long userId) {
|
|
|
|
|
+ AgentChatSession session = assertOwner(dto.getSessionId(), userId);
|
|
|
|
|
+
|
|
|
|
|
+ // 保存用户消息
|
|
|
|
|
+ AgentMessage userMessage = new AgentMessage();
|
|
|
|
|
+ userMessage.setId(generateId());
|
|
|
|
|
+ userMessage.setSessionId(dto.getSessionId());
|
|
|
|
|
+ userMessage.setRole("user");
|
|
|
|
|
+ userMessage.setContent(dto.getContent());
|
|
|
|
|
+ userMessage.setContentType(dto.getContentType() != null ? dto.getContentType() : "text");
|
|
|
|
|
+ userMessage.setMessageType(dto.getMessageType() != null ? dto.getMessageType() : "text");
|
|
|
|
|
+ userMessage.setReplyToMessageId(dto.getReplyToMessageId());
|
|
|
|
|
+ userMessage.setStarred(false);
|
|
|
|
|
+ userMessage.setCreateAt(LocalDateTime.now());
|
|
|
|
|
+
|
|
|
|
|
+ messageMapper.insert(userMessage);
|
|
|
|
|
+
|
|
|
|
|
+ // 更新会话信息
|
|
|
|
|
+ chatSessionMapper.update(null,
|
|
|
|
|
+ new LambdaUpdateWrapper<AgentChatSession>()
|
|
|
|
|
+ .eq(AgentChatSession::getId, dto.getSessionId())
|
|
|
|
|
+ .set(AgentChatSession::getLastMessageAt, LocalDateTime.now())
|
|
|
|
|
+ .set(AgentChatSession::getMessageCount, session.getMessageCount() + 1)
|
|
|
|
|
+ .set(AgentChatSession::getUpdateAt, LocalDateTime.now())
|
|
|
|
|
+ );
|
|
|
|
|
+
|
|
|
|
|
+ // 如果是第一条消息,更新会话标题
|
|
|
|
|
+ if (session.getMessageCount() == 0) {
|
|
|
|
|
+ String title = dto.getContent().length() > 50
|
|
|
|
|
+ ? dto.getContent().substring(0, 50) + "..."
|
|
|
|
|
+ : dto.getContent();
|
|
|
|
|
+ chatSessionMapper.update(null,
|
|
|
|
|
+ new LambdaUpdateWrapper<AgentChatSession>()
|
|
|
|
|
+ .eq(AgentChatSession::getId, dto.getSessionId())
|
|
|
|
|
+ .set(AgentChatSession::getTitle, title)
|
|
|
|
|
+ );
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ log.info("发送消息成功: {}, 会话: {}", userMessage.getId(), dto.getSessionId());
|
|
|
|
|
+ return convertMessageToVO(userMessage);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ @Override
|
|
|
|
|
+ @Transactional(rollbackFor = Exception.class)
|
|
|
|
|
+ public void deleteMessage(Long messageId, Long userId) {
|
|
|
|
|
+ AgentMessage entity = messageMapper.selectById(messageId);
|
|
|
|
|
+ if (entity == null) {
|
|
|
|
|
+ throw new ServerException(404, "消息不存在: " + messageId);
|
|
|
|
|
+ }
|
|
|
|
|
+ assertOwner(entity.getSessionId(), userId);
|
|
|
|
|
+
|
|
|
|
|
+ messageMapper.deleteById(messageId);
|
|
|
|
|
+
|
|
|
|
|
+ // 更新会话消息数量
|
|
|
|
|
+ AgentChatSession session = chatSessionMapper.selectById(entity.getSessionId());
|
|
|
|
|
+ if (session != null) {
|
|
|
|
|
+ chatSessionMapper.update(null,
|
|
|
|
|
+ new LambdaUpdateWrapper<AgentChatSession>()
|
|
|
|
|
+ .eq(AgentChatSession::getId, entity.getSessionId())
|
|
|
|
|
+ .set(AgentChatSession::getMessageCount, Math.max(0, session.getMessageCount() - 1))
|
|
|
|
|
+ .set(AgentChatSession::getUpdateAt, LocalDateTime.now())
|
|
|
|
|
+ );
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ log.info("删除消息成功: {}", messageId);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ @Override
|
|
|
|
|
+ @Transactional(rollbackFor = Exception.class)
|
|
|
|
|
+ public void toggleStarMessage(Long messageId, Long userId) {
|
|
|
|
|
+ AgentMessage entity = messageMapper.selectById(messageId);
|
|
|
|
|
+ if (entity == null) {
|
|
|
|
|
+ throw new ServerException(404, "消息不存在: " + messageId);
|
|
|
|
|
+ }
|
|
|
|
|
+ assertOwner(entity.getSessionId(), userId);
|
|
|
|
|
+
|
|
|
|
|
+ messageMapper.update(null,
|
|
|
|
|
+ new LambdaUpdateWrapper<AgentMessage>()
|
|
|
|
|
+ .eq(AgentMessage::getId, messageId)
|
|
|
|
|
+ .set(AgentMessage::getStarred, !entity.getStarred())
|
|
|
|
|
+ );
|
|
|
|
|
+ log.info("切换消息收藏状态成功: {}, 收藏: {}", messageId, !entity.getStarred());
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ @Override
|
|
|
|
|
+ public Flux<ServerSentEvent<String>> stream(ChatRequestDTO req, Long userId) {
|
|
|
|
|
+ if (!StringUtils.hasText(req.getMessage())) {
|
|
|
|
|
+ return Flux.just(sse("error", Map.of("type", "error", "error", "消息内容不能为空")));
|
|
|
|
|
+ }
|
|
|
|
|
+ // defer:订阅时才解析会话/构建 agent,异常统一转 error 帧
|
|
|
|
|
+ return Flux.defer(() -> doStream(req, userId))
|
|
|
|
|
+ .onErrorResume(ex -> {
|
|
|
|
|
+ String errorMsg = ex.getMessage() != null ? ex.getMessage() : "未知错误";
|
|
|
|
|
+ log.warn("Chat stream error: sessionKey={}, error={}", req.getSessionKey(), errorMsg);
|
|
|
|
|
+ // 错误信息持久化:确保刷新后仍可见
|
|
|
|
|
+ Long sessionId = parseSessionId(req.getSessionKey());
|
|
|
|
|
+ if (sessionId != null) {
|
|
|
|
|
+ Mono.fromRunnable(() -> persistErrorMessage(sessionId, errorMsg))
|
|
|
|
|
+ .subscribeOn(Schedulers.boundedElastic())
|
|
|
|
|
+ .subscribe();
|
|
|
|
|
+ }
|
|
|
|
|
+ return Flux.just(sse("error",
|
|
|
|
|
+ Map.of("type", "error", "error", errorMsg)));
|
|
|
|
|
+ });
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ /**
|
|
|
|
|
+ * 流式对话主链路(官方 API 直连)。
|
|
|
|
|
+ *
|
|
|
|
|
+ * <p>从会话记录解析三维隔离参数(userId / workspaceId / agentRowId)与模型;
|
|
|
|
|
+ * 请求体携带 modelId 且与会话绑定不同时,视为动态切换模型并回写会话。
|
|
|
|
|
+ * 记忆槽位 = (userId, sessionId),与 agent 实例/模型无关,切换模型后记忆自动延续。
|
|
|
|
|
+ */
|
|
|
|
|
+ private Flux<ServerSentEvent<String>> doStream(ChatRequestDTO req, Long userId) {
|
|
|
|
|
+ Long sessionId = parseSessionId(req.getSessionKey());
|
|
|
|
|
+ if (sessionId == null) {
|
|
|
|
|
+ throw new ServerException(400, "缺少有效会话ID(sessionKey)");
|
|
|
|
|
+ }
|
|
|
|
|
+ AgentChatSession session = assertOwner(sessionId, userId);
|
|
|
|
|
+
|
|
|
|
|
+ String userIdStr = String.valueOf(userId);
|
|
|
|
|
+ Long workspaceId = session.getCaseId();
|
|
|
|
|
+ Long agentRowId = session.getAgentId() != null ? session.getAgentId() : DEFAULT_AGENT_ROW_ID;
|
|
|
|
|
+
|
|
|
|
|
+ // 模型优先级:请求体 modelId(动态切换)> session.modelId > 默认模型
|
|
|
|
|
+ Long modelId = req.getModelId() != null ? req.getModelId() : session.getModelId();
|
|
|
|
|
+ if (modelId == null) {
|
|
|
|
|
+ modelId = agentModelService.getDefaultModelId();
|
|
|
|
|
+ }
|
|
|
|
|
+ if (modelId == null) {
|
|
|
|
|
+ throw new ServerException(400, "未指定模型且无默认模型,请先选择模型");
|
|
|
|
|
+ }
|
|
|
|
|
+ // 动态切换:回写会话,下一轮默认使用新模型
|
|
|
|
|
+ if (!Objects.equals(modelId, session.getModelId())) {
|
|
|
|
|
+ chatSessionMapper.update(null,
|
|
|
|
|
+ new LambdaUpdateWrapper<AgentChatSession>()
|
|
|
|
|
+ .eq(AgentChatSession::getId, sessionId)
|
|
|
|
|
+ .set(AgentChatSession::getModelId, modelId)
|
|
|
|
|
+ .set(AgentChatSession::getUpdateAt, LocalDateTime.now()));
|
|
|
|
|
+ session.setModelId(modelId);
|
|
|
|
|
+ log.info("会话动态切换模型: sessionId={}, modelId={}", sessionId, modelId);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ HarnessAgent agent = agentService.getOrCreateAgent(userIdStr, workspaceId, agentRowId, modelId);
|
|
|
|
|
+ RuntimeContext ctx = RuntimeContext.builder()
|
|
|
|
|
+ .userId(userIdStr)
|
|
|
|
|
+ .sessionId(String.valueOf(sessionId))
|
|
|
|
|
+ .build();
|
|
|
|
|
+
|
|
|
|
|
+ // 意图识别由 IntentMiddleware 在 onAgent 钩子内完成(消息增强 + intent 事件),
|
|
|
|
|
+ // 这里原文直发,ChatServiceImpl 只负责事件转换
|
|
|
|
|
+ List<Msg> msgs = List.of(
|
|
|
|
|
+ Msg.builder().role(MsgRole.USER).textContent(req.getMessage()).build());
|
|
|
|
|
+ Flux<AgentEvent> events = agent.streamEvents(msgs, ctx);
|
|
|
|
|
+
|
|
|
|
|
+ long startMs = System.currentTimeMillis();
|
|
|
|
|
+ String sessionKey = String.valueOf(sessionId);
|
|
|
|
|
+
|
|
|
|
|
+ // 工具调用状态聚合(流内累积,随流结束释放)
|
|
|
|
|
+ Map<String, StringBuilder> toolInputs = new ConcurrentHashMap<>();
|
|
|
|
|
+ Map<String, StringBuilder> toolResults = new ConcurrentHashMap<>();
|
|
|
|
|
+ Map<String, String> toolNames = new ConcurrentHashMap<>();
|
|
|
|
|
+
|
|
|
|
|
+ // 流中断追踪:用户主动中止时,若 AgentResultEvent 尚未到达,将已累积的文本保存
|
|
|
|
|
+ AtomicReference<StringBuilder> accumulatedText = new AtomicReference<>(new StringBuilder());
|
|
|
|
|
+ AtomicBoolean resultPersisted = new AtomicBoolean(false);
|
|
|
|
|
+
|
|
|
|
|
+ return events.concatMap(event -> {
|
|
|
|
|
+ List<ServerSentEvent<String>> frames = new ArrayList<>();
|
|
|
|
|
+ if (event instanceof CustomEvent e && "intent".equals(e.getName())) {
|
|
|
|
|
+ // IntentMiddleware 发出的意图识别结果 → SSE intent 帧
|
|
|
|
|
+ frames.add(sse("intent", e.getValue()));
|
|
|
|
|
+ } else if (event instanceof CustomEvent e && "suggestions".equals(e.getName())) {
|
|
|
|
|
+ // FollowupMiddleware 在对话结束后追加的下一步引导建议 → SSE suggestions 帧
|
|
|
|
|
+ frames.add(sse("suggestions", e.getValue()));
|
|
|
|
|
+ } else if (event instanceof ToolCallStartEvent e) {
|
|
|
|
|
+ Map<String, Object> data = new LinkedHashMap<>();
|
|
|
|
|
+ data.put("type", "tool_call");
|
|
|
|
|
+ data.put("toolCallId", e.getToolCallId());
|
|
|
|
|
+ data.put("toolName", e.getToolCallName());
|
|
|
|
|
+ toolNames.put(e.getToolCallId(), e.getToolCallName());
|
|
|
|
|
+ frames.add(sse("tool_call", data));
|
|
|
|
|
+ } else if (event instanceof ToolCallDeltaEvent e) {
|
|
|
|
|
+ toolInputs
|
|
|
|
|
+ .computeIfAbsent(e.getToolCallId(), k -> new StringBuilder())
|
|
|
|
|
+ .append(e.getDelta());
|
|
|
|
|
+ } else if (event instanceof ToolCallEndEvent e) {
|
|
|
|
|
+ Map<String, Object> data = new LinkedHashMap<>();
|
|
|
|
|
+ data.put("type", "tool_input");
|
|
|
|
|
+ data.put("toolCallId", e.getToolCallId());
|
|
|
|
|
+ StringBuilder input = toolInputs.get(e.getToolCallId());
|
|
|
|
|
+ data.put("toolInput", input != null ? input.toString() : "");
|
|
|
|
|
+ frames.add(sse("tool_input", data));
|
|
|
|
|
+ } else if (event instanceof ToolResultTextDeltaEvent e) {
|
|
|
|
|
+ toolResults
|
|
|
|
|
+ .computeIfAbsent(e.getToolCallId(), k -> new StringBuilder())
|
|
|
|
|
+ .append(e.getDelta());
|
|
|
|
|
+ } else if (event instanceof ToolResultEndEvent e) {
|
|
|
|
|
+ Map<String, Object> data = new LinkedHashMap<>();
|
|
|
|
|
+ data.put("type", "tool_result");
|
|
|
|
|
+ data.put("toolCallId", e.getToolCallId());
|
|
|
|
|
+ data.put("toolName", e.getToolCallName());
|
|
|
|
|
+ StringBuilder result = toolResults.get(e.getToolCallId());
|
|
|
|
|
+ if (!result.isEmpty()) {
|
|
|
|
|
+ String text = result.toString();
|
|
|
|
|
+ // 截断超长结果,避免单帧 SSE 过大(表格首页 JSON 可达数十 k)
|
|
|
|
|
+ data.put("toolResult", text.length() > 100000 ? text.substring(0, 100000) : text);
|
|
|
|
|
+ }
|
|
|
|
|
+ frames.add(sse("tool_result", data));
|
|
|
|
|
+ } else if (event instanceof TextBlockDeltaEvent e) {
|
|
|
|
|
+ accumulatedText.get().append(e.getDelta());
|
|
|
|
|
+ frames.add(sse("token", Map.of("type", "token", "data", e.getDelta())));
|
|
|
|
|
+ } else if (event instanceof AgentResultEvent e) {
|
|
|
|
|
+ // AI 回复落库(异步,不阻塞事件流);工具调用事件一并持久化成 tool_events
|
|
|
|
|
+ resultPersisted.set(true);
|
|
|
|
|
+ String toolEventsJson = buildToolEventsJson(toolInputs, toolResults, toolNames);
|
|
|
|
|
+ Mono.fromRunnable(() -> persistAssistantMessage(sessionId, e.getResult(), toolEventsJson))
|
|
|
|
|
+ .subscribeOn(Schedulers.boundedElastic())
|
|
|
|
|
+ .subscribe();
|
|
|
|
|
+ } else if (event instanceof AgentEndEvent) {
|
|
|
|
|
+ usageStore.record(userIdStr, String.valueOf(agentRowId), System.currentTimeMillis() - startMs);
|
|
|
|
|
+ frames.add(sse("done", Map.of("type", "done", "sessionKey", sessionKey)));
|
|
|
|
|
+ }
|
|
|
|
|
+ return Flux.fromIterable(frames);
|
|
|
|
|
+ }).doOnCancel(() -> {
|
|
|
|
|
+ // 用户主动中止:若 AgentResultEvent 尚未到达,保存已流式输出的部分内容
|
|
|
|
|
+ if (!resultPersisted.get()) {
|
|
|
|
|
+ String partialText = accumulatedText.get().toString().trim();
|
|
|
|
|
+ String toolEventsJson = buildToolEventsJson(toolInputs, toolResults, toolNames);
|
|
|
|
|
+ if (!partialText.isEmpty() || !toolEventsJson.isEmpty()) {
|
|
|
|
|
+ Mono.fromRunnable(() -> persistPartialMessage(sessionId, partialText, toolEventsJson))
|
|
|
|
|
+ .subscribeOn(Schedulers.boundedElastic())
|
|
|
|
|
+ .subscribe();
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ });
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ /**
|
|
|
|
|
+ * 校验会话归属:会话不存在返回 404,非本人会话返回 403。
|
|
|
|
|
+ *
|
|
|
|
|
+ * @return 归属校验通过的会话实体
|
|
|
|
|
+ */
|
|
|
|
|
+ private AgentChatSession assertOwner(Long sessionId, Long caseId) {
|
|
|
|
|
+ AgentChatSession session = chatSessionMapper.selectById(sessionId);
|
|
|
|
|
+ if (session == null) {
|
|
|
|
|
+ throw new ServerException(404, "会话不存在: " + sessionId);
|
|
|
|
|
+ }
|
|
|
|
|
+ if (!Objects.equals(session.getCaseId(), caseId)) {
|
|
|
|
|
+ throw new ServerException(403, "无权访问该会话");
|
|
|
|
|
+ }
|
|
|
|
|
+ return session;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ /**
|
|
|
|
|
+ * 将 AI 回复持久化为 assistant 消息。失败不影响流式输出。
|
|
|
|
|
+ *
|
|
|
|
|
+ * @param toolEventsJson 工具调用事件 JSON(execute_sql 入参/结果等),可为 null
|
|
|
|
|
+ */
|
|
|
|
|
+ private void persistAssistantMessage(Long sessionId, Msg result, String toolEventsJson) {
|
|
|
|
|
+ try {
|
|
|
|
|
+ AgentChatSession session = chatSessionMapper.selectById(sessionId);
|
|
|
|
|
+ if (session == null) {
|
|
|
|
|
+ return;
|
|
|
|
|
+ }
|
|
|
|
|
+ String text = result.getTextContent() != null ? result.getTextContent() : "";
|
|
|
|
|
+ if (text.isBlank()) {
|
|
|
|
|
+ return;
|
|
|
|
|
+ }
|
|
|
|
|
+ AgentMessage assistantMessage = new AgentMessage();
|
|
|
|
|
+ assistantMessage.setSessionId(sessionId);
|
|
|
|
|
+ assistantMessage.setRole("assistant");
|
|
|
|
|
+ assistantMessage.setContent(text);
|
|
|
|
|
+ assistantMessage.setContentType("text");
|
|
|
|
|
+ assistantMessage.setMessageType("text");
|
|
|
|
|
+ assistantMessage.setStarred(false);
|
|
|
|
|
+ assistantMessage.setToolEvents(toolEventsJson);
|
|
|
|
|
+ assistantMessage.setCreateAt(LocalDateTime.now());
|
|
|
|
|
+ messageMapper.insert(assistantMessage);
|
|
|
|
|
+ chatSessionMapper.update(
|
|
|
|
|
+ null,
|
|
|
|
|
+ new LambdaUpdateWrapper<AgentChatSession>()
|
|
|
|
|
+ .eq(AgentChatSession::getId, sessionId)
|
|
|
|
|
+ .set(AgentChatSession::getLastMessageAt, LocalDateTime.now())
|
|
|
|
|
+ .set(AgentChatSession::getMessageCount, session.getMessageCount() + 1)
|
|
|
|
|
+ .set(AgentChatSession::getUpdateAt, LocalDateTime.now()));
|
|
|
|
|
+ log.info("AI回复落库成功: sessionId={}, messageId={}", sessionId, assistantMessage.getId());
|
|
|
|
|
+ } catch (Exception e) {
|
|
|
|
|
+ log.warn("AI回复落库失败: sessionId={}, error={}", sessionId, e.getMessage());
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ /**
|
|
|
|
|
+ * 持久化错误信息为 assistant 消息(messageType=error),确保刷新后仍可见。
|
|
|
|
|
+ */
|
|
|
|
|
+ private void persistErrorMessage(Long sessionId, String errorMsg) {
|
|
|
|
|
+ try {
|
|
|
|
|
+ AgentMessage msg = new AgentMessage();
|
|
|
|
|
+ msg.setSessionId(sessionId);
|
|
|
|
|
+ msg.setRole("assistant");
|
|
|
|
|
+ msg.setContent("⚠️ " + errorMsg);
|
|
|
|
|
+ msg.setContentType("text");
|
|
|
|
|
+ msg.setMessageType("error");
|
|
|
|
|
+ msg.setMetadata("{\"error\": true}");
|
|
|
|
|
+ msg.setStarred(false);
|
|
|
|
|
+ msg.setCreateAt(LocalDateTime.now());
|
|
|
|
|
+ messageMapper.insert(msg);
|
|
|
|
|
+ chatSessionMapper.update(null,
|
|
|
|
|
+ new LambdaUpdateWrapper<AgentChatSession>()
|
|
|
|
|
+ .eq(AgentChatSession::getId, sessionId)
|
|
|
|
|
+ .setSql("message_count = message_count + 1")
|
|
|
|
|
+ .set(AgentChatSession::getLastMessageAt, LocalDateTime.now())
|
|
|
|
|
+ .set(AgentChatSession::getUpdateAt, LocalDateTime.now()));
|
|
|
|
|
+ log.info("错误消息落库成功: sessionId={}, messageId={}", sessionId, msg.getId());
|
|
|
|
|
+ } catch (Exception e) {
|
|
|
|
|
+ log.warn("错误消息落库失败: sessionId={}, error={}", sessionId, e.getMessage());
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ /**
|
|
|
|
|
+ * 持久化用户中断时的部分内容(messageType=interrupted),标记为不完整回复。
|
|
|
|
|
+ */
|
|
|
|
|
+ private void persistPartialMessage(Long sessionId, String partialText, String toolEventsJson) {
|
|
|
|
|
+ try {
|
|
|
|
|
+ AgentMessage msg = new AgentMessage();
|
|
|
|
|
+ msg.setSessionId(sessionId);
|
|
|
|
|
+ msg.setRole("assistant");
|
|
|
|
|
+ msg.setContent(partialText.isEmpty() ? "(回复已中断)" : partialText + "\n\n*(回复已中断)*");
|
|
|
|
|
+ msg.setContentType("text");
|
|
|
|
|
+ msg.setMessageType("interrupted");
|
|
|
|
|
+ msg.setMetadata("{\"interrupted\": true}");
|
|
|
|
|
+ msg.setToolEvents(toolEventsJson);
|
|
|
|
|
+ msg.setStarred(false);
|
|
|
|
|
+ msg.setCreateAt(LocalDateTime.now());
|
|
|
|
|
+ messageMapper.insert(msg);
|
|
|
|
|
+ chatSessionMapper.update(null,
|
|
|
|
|
+ new LambdaUpdateWrapper<AgentChatSession>()
|
|
|
|
|
+ .eq(AgentChatSession::getId, sessionId)
|
|
|
|
|
+ .setSql("message_count = message_count + 1")
|
|
|
|
|
+ .set(AgentChatSession::getLastMessageAt, LocalDateTime.now())
|
|
|
|
|
+ .set(AgentChatSession::getUpdateAt, LocalDateTime.now()));
|
|
|
|
|
+ log.info("中断回复落库成功: sessionId={}, messageId={}, textLen={}", sessionId, msg.getId(), partialText.length());
|
|
|
|
|
+ } catch (Exception e) {
|
|
|
|
|
+ log.warn("中断回复落库失败: sessionId={}, error={}", sessionId, e.getMessage());
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ /**
|
|
|
|
|
+ * sessionKey → 会话ID(当前约定:sessionKey 即业务会话 ID 字符串)
|
|
|
|
|
+ */
|
|
|
|
|
+ private static Long parseSessionId(String sessionKey) {
|
|
|
|
|
+ if (!StringUtils.hasText(sessionKey)) {
|
|
|
|
|
+ return null;
|
|
|
|
|
+ }
|
|
|
|
|
+ try {
|
|
|
|
|
+ return Long.parseLong(sessionKey.trim());
|
|
|
|
|
+ } catch (NumberFormatException e) {
|
|
|
|
|
+ return null;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ private Long generateId() {
|
|
|
|
|
+ return UUID.randomUUID().getMostSignificantBits() & Long.MAX_VALUE;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ /**
|
|
|
|
|
+ * 将流内累积的工具调用聚合为 JSON 字符串(tool_events 列存储形态,与前端 ToolStep 对齐)。
|
|
|
|
|
+ *
|
|
|
|
|
+ * <p>每条工具调用产出 {@code {id, toolName, input, result, status}}。input/result 为字符串原样保留:
|
|
|
|
|
+ * execute_sql 的 result 是完整 JSON,前端据此渲染分页表格。
|
|
|
|
|
+ */
|
|
|
|
|
+ private static String buildToolEventsJson(Map<String, StringBuilder> toolInputs,
|
|
|
|
|
+ Map<String, StringBuilder> toolResults,
|
|
|
|
|
+ Map<String, String> toolNames) {
|
|
|
|
|
+ if (toolNames.isEmpty()) {
|
|
|
|
|
+ return null;
|
|
|
|
|
+ }
|
|
|
|
|
+ List<Map<String, Object>> events = new ArrayList<>();
|
|
|
|
|
+ for (String toolCallId : toolNames.keySet()) {
|
|
|
|
|
+ Map<String, Object> ev = new LinkedHashMap<>();
|
|
|
|
|
+ ev.put("id", toolCallId);
|
|
|
|
|
+ ev.put("toolName", toolNames.get(toolCallId));
|
|
|
|
|
+ StringBuilder input = toolInputs.get(toolCallId);
|
|
|
|
|
+ ev.put("input", input != null ? input.toString() : "");
|
|
|
|
|
+ StringBuilder result = toolResults.get(toolCallId);
|
|
|
|
|
+ ev.put("result", result != null ? result.toString() : "");
|
|
|
|
|
+ ev.put("status", "done");
|
|
|
|
|
+ events.add(ev);
|
|
|
|
|
+ }
|
|
|
|
|
+ try {
|
|
|
|
|
+ return MAPPER.writeValueAsString(events);
|
|
|
|
|
+ } catch (JsonProcessingException e) {
|
|
|
|
|
+ return null;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ private ChatSessionVO convertSessionToVO(AgentChatSession entity) {
|
|
|
|
|
+ ChatSessionVO vo = new ChatSessionVO();
|
|
|
|
|
+ vo.setId(entity.getId());
|
|
|
|
|
+ vo.setAgentId(entity.getAgentId());
|
|
|
|
|
+ vo.setModelId(entity.getModelId());
|
|
|
|
|
+ vo.setCaseId(entity.getCaseId());
|
|
|
|
|
+ vo.setTitle(entity.getTitle());
|
|
|
|
|
+ vo.setContext(entity.getContext());
|
|
|
|
|
+ vo.setMessageCount(entity.getMessageCount());
|
|
|
|
|
+ vo.setTotalTokens(entity.getTotalTokens());
|
|
|
|
|
+ vo.setPinned(entity.getPinned());
|
|
|
|
|
+ vo.setAttachments(entity.getAttachments());
|
|
|
|
|
+ vo.setLastMessageAt(entity.getLastMessageAt());
|
|
|
|
|
+ vo.setCreateAt(entity.getCreateAt());
|
|
|
|
|
+ vo.setUpdateAt(entity.getUpdateAt());
|
|
|
|
|
+ return vo;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ private MessageVO convertMessageToVO(AgentMessage entity) {
|
|
|
|
|
+ MessageVO vo = new MessageVO();
|
|
|
|
|
+ vo.setId(entity.getId());
|
|
|
|
|
+ vo.setSessionId(entity.getSessionId());
|
|
|
|
|
+ vo.setRole(entity.getRole());
|
|
|
|
|
+ vo.setContent(entity.getContent());
|
|
|
|
|
+ vo.setContentType(entity.getContentType());
|
|
|
|
|
+ vo.setTokenCount(entity.getTokenCount());
|
|
|
|
|
+ vo.setMetadata(entity.getMetadata());
|
|
|
|
|
+ vo.setToolEvents(entity.getToolEvents());
|
|
|
|
|
+ vo.setParentId(entity.getParentId());
|
|
|
|
|
+ vo.setMessageType(entity.getMessageType());
|
|
|
|
|
+ vo.setReplyToMessageId(entity.getReplyToMessageId());
|
|
|
|
|
+ vo.setStarred(entity.getStarred());
|
|
|
|
|
+ vo.setCreateAt(entity.getCreateAt());
|
|
|
|
|
+ return vo;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ private ServerSentEvent<String> sse(String eventType, Object data) {
|
|
|
|
|
+ String json;
|
|
|
|
|
+ try {
|
|
|
|
|
+ json = MAPPER.writeValueAsString(data);
|
|
|
|
|
+ } catch (JsonProcessingException e) {
|
|
|
|
|
+ json = "{\"type\":\"" + eventType + "\"}";
|
|
|
|
|
+ }
|
|
|
|
|
+ return ServerSentEvent.<String>builder().event(eventType).data(json).build();
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+}
|