From d1a820728fd766bc5e2027e17d4b0d21f5a9a742 Mon Sep 17 00:00:00 2001 From: ageerle Date: Thu, 23 Jul 2026 15:35:10 +0800 Subject: [PATCH] =?UTF-8?q?fix(sse):=20SSE=20=E6=94=B9=E4=B8=BA=E6=8C=89?= =?UTF-8?q?=E4=BC=9A=E8=AF=9D=E9=9A=94=E7=A6=BB=20+=20=E4=BF=AE=20trace=20?= =?UTF-8?q?=E7=A7=9F=E6=88=B7=20bug=20+=20=E6=B8=85=E7=90=86=E5=86=97?= =?UTF-8?q?=E4=BD=99?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SSE 串台修复： - SseEmitterManager 新增按 sessionId 维度的 connect/sendEvent/disconnect， 每个会话一个 SSE 连接，替代原 userId+token 维度（同用户多会话串台） - SseMessageDto 增加 sessionId + eventDto 字段，跨实例按会话路由 - SseTopicListener 优先按 sessionId 路由，回退原 userId/群发逻辑 - SseMessageUtils 增加 sessionId 重载，对话链路全部切换 - ChatServiceFacade / MyMcpClientListener / WorkflowStarter 切到 sessionId - 通知/全局 /sse 端点保留 userId 模式（通知按用户） trace 租户 bug 修复： - trace_run / trace_node 加入 tenant.excludes，监控表跨租户全局可见， 绕过异步线程租户上下文不传播导致 node 查不到的问题 冗余清理： - 删除零使用的 @TraceNode 注解 + TraceNodeAspect 切面 + 对应单测 - TraceProperties 删除 recordDetail/maxInputLength/maxOutputLength （仅对已删切面生效，编程式埋点用 RagTracePayloadBuilder 不受影响） - TracePayloadUtils 删除 input/output/asString 方法 - TraceConstants 删除无引用的 NODE_METHOD/HTTP/DB/CACHE/TASK/STREAM - TraceContext 删除预留的 copyNodeStack - TraceNodeVo 删除未用的 children 字段 - TraceRecordServiceImpl.getDetail 删除重复 enrich - application.yml 删除 trace.payload 无效配置项 - RagTraceNodeTypes 删除无引用的 NODE_STREAM Co-Authored-By: Claude --- .../src/main/resources/application.yml | 12 +- .../common/sse/core/SseEmitterManager.java | 125 +++++++++++++++- .../ruoyi/common/sse/dto/SseMessageDto.java | 10 ++ .../common/sse/listener/SseTopicListener.java | 18 ++- .../common/sse/utils/SseMessageUtils.java | 64 +++++++++ .../common/trace/annotation/TraceNode.java | 31 ---- .../common/trace/aspect/TraceNodeAspect.java | 89 ------------ .../trace/config/TraceAutoConfiguration.java | 14 +- .../common/trace/config/TraceProperties.java | 15 -- .../common/trace/constant/TraceConstants.java | 7 - .../ruoyi/common/trace/core/TraceContext.java | 8 -- .../common/trace/domain/vo/TraceNodeVo.java | 9 +- .../service/impl/TraceRecordServiceImpl.java | 3 - .../common/trace/util/TracePayloadUtils.java | 18 --- .../trace/aspect/TraceNodeAspectTest.java | 134 ------------------ .../common/trace/core/TraceContextTest.java | 12 -- .../trace/util/TracePayloadUtilsTest.java | 12 -- .../workflow/workflow/WorkflowStarter.java | 6 +- .../observability/MyMcpClientListener.java | 14 +- .../service/chat/impl/ChatServiceFacade.java | 62 ++++---- .../org/ruoyi/trace/RagTraceNodeTypes.java | 1 - 21 files changed, 259 insertions(+), 405 deletions(-) delete mode 100644 ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/annotation/TraceNode.java delete mode 100644 ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/aspect/TraceNodeAspect.java delete mode 100644 ruoyi-common/ruoyi-common-trace/src/test/java/org/ruoyi/common/trace/aspect/TraceNodeAspectTest.java diff --git a/ruoyi-admin/src/main/resources/application.yml b/ruoyi-admin/src/main/resources/application.yml index c93019e0..aba5aa1a 100644 --- a/ruoyi-admin/src/main/resources/application.yml +++ b/ruoyi-admin/src/main/resources/application.yml @@ -146,6 +146,9 @@ tenant: - sys_client - sys_oss_config - flow_spel + # 链路追踪监控表：运维需跨租户全局查看，且 trace_node 在异步线程写入、租户上下文不传播，故排除租户过滤 + - trace_run + - trace_node # MyBatisPlus配置 # https://baomidou.com/config/ @@ -238,15 +241,6 @@ trace: # 关闭后所有埋点代码会直接透传业务逻辑，不写库、不创建上下文，零性能开销 enabled: true payload: - # 是否记录方法真实入参和返回值，默认 false（仅存摘要） - # 仅对 @TraceNode 注解方式生效，编程式埋点使用 TracePayloadBuilder 手动构建的摘要不受此开关影响 - # true → 自动序列化原始入参/返回值，截断后写入 input_payload / output_payload - # false → 只存 nodeName / 耗时 / 状态等元信息，input_payload / output_payload 为空 - record-detail: false - # input payload 最大字符长度，超过部分会被截断丢弃 - max-input-length: 1000 - # output payload 最大字符长度，超过部分会被截断丢弃 - max-output-length: 2000 # 错误信息最大字符长度（格式: "异常类名: 异常消息"），超过部分会被截断丢弃 max-error-length: 1000 diff --git a/ruoyi-common/ruoyi-common-sse/src/main/java/org/ruoyi/common/sse/core/SseEmitterManager.java b/ruoyi-common/ruoyi-common-sse/src/main/java/org/ruoyi/common/sse/core/SseEmitterManager.java index fe4835db..121c16b5 100644 --- a/ruoyi-common/ruoyi-common-sse/src/main/java/org/ruoyi/common/sse/core/SseEmitterManager.java +++ b/ruoyi-common/ruoyi-common-sse/src/main/java/org/ruoyi/common/sse/core/SseEmitterManager.java @@ -32,6 +32,16 @@ public class SseEmitterManager { */ private final static String SSE_TOPIC = "global:sse"; + /** + * 按会话维度管理：每个会话一个 SSE 连接，用于对话流式响应 + * Key: sessionId + */ + private final static Map SESSION_EMITTERS = new ConcurrentHashMap<>(); + + /** + * 按用户维度管理：全局通知、站内信等场景，一个用户可有多个连接（按 token 区分） + * Key: userId, Value: token -> SseEmitter + */ private final static Map> USER_TOKEN_EMITTERS = new ConcurrentHashMap<>(); public SseEmitterManager() { @@ -40,6 +50,98 @@ public class SseEmitterManager { .scheduleWithFixedDelay(this::sseMonitor, 60L, 60L, TimeUnit.SECONDS); } + // ======================== 会话维度（对话流式响应） ======================== + + /** + * 建立与指定会话的 SSE 连接，每个会话仅保留一个连接，重复建连会替换旧连接。 + * + * @param sessionId 会话ID + * @return SseEmitter 实例 + */ + public SseEmitter connect(String sessionId) { + if (sessionId == null) { + throw new IllegalArgumentException("sessionId 不能为空"); + } + // 关闭已存在的 SseEmitter，保证每个会话只有一个活跃连接 + SseEmitter oldEmitter = SESSION_EMITTERS.remove(sessionId); + if (oldEmitter != null) { + oldEmitter.complete(); + } + + SseEmitter emitter = new SseEmitter(86400000L); + SESSION_EMITTERS.put(sessionId, emitter); + + emitter.onCompletion(() -> removeSessionEmitter(sessionId, emitter)); + emitter.onTimeout(() -> removeSessionEmitter(sessionId, emitter)); + emitter.onError(e -> removeSessionEmitter(sessionId, emitter)); + + try { + emitter.send(SseEmitter.event().comment("connected")); + } catch (IOException e) { + SESSION_EMITTERS.remove(sessionId); + } + return emitter; + } + + /** + * 断开指定会话的 SSE 连接 + * + * @param sessionId 会话ID + */ + public void disconnect(String sessionId) { + if (sessionId == null) { + return; + } + SseEmitter emitter = SESSION_EMITTERS.remove(sessionId); + if (emitter != null) { + try { + emitter.send(SseEmitter.event().comment("disconnected")); + } catch (Exception exception) { + log.error(exception.getMessage()); + } + emitter.complete(); + } + } + + /** + * 向指定会话发送结构化事件 + * + * @param sessionId 会话ID + * @param eventDto SSE事件对象 + */ + public void sendEvent(String sessionId, SseEventDto eventDto) { + if (sessionId == null || eventDto == null) { + return; + } + SseEmitter emitter = SESSION_EMITTERS.get(sessionId); + if (emitter == null) { + log.warn("【SSE发送失败】sessionId: {} 没有活跃的SSE连接", sessionId); + return; + } + try { + log.debug("【SSE发送】sessionId: {}, event: {}", sessionId, eventDto.getEvent()); + emitter.send(SseEmitter.event() + .name(eventDto.getEvent()) + .data(JSONUtil.toJsonStr(eventDto))); + } catch (Exception e) { + log.error("【SSE发送失败】sessionId: {}, error: {}", sessionId, e.getMessage()); + removeSessionEmitter(sessionId, emitter); + } + } + + private void removeSessionEmitter(String sessionId, SseEmitter emitter) { + boolean removed = SESSION_EMITTERS.remove(sessionId, emitter); + if (removed) { + try { + emitter.complete(); + } catch (Exception ignore) { + // 忽略重复关闭异常 + } + } + } + + // ======================== 用户维度（全局通知） ======================== + /** * 建立与指定用户的 SSE 连接 * @@ -154,6 +256,23 @@ public class SseEmitterManager { // 循环结束后统一清理空用户，避免并发修改异常 toRemoveUsers.forEach(USER_TOKEN_EMITTERS::remove); + + // 会话维度心跳：发送失败的连接移除 + if (!SESSION_EMITTERS.isEmpty()) { + SESSION_EMITTERS.entrySet().removeIf(entry -> { + try { + entry.getValue().send(heartbeat); + return false; + } catch (Exception ex) { + try { + entry.getValue().complete(); + } catch (Exception ignore) { + // 忽略重复关闭异常 + } + return true; + } + }); + } } /** @@ -243,9 +362,11 @@ public class SseEmitterManager { SseMessageDto broadcastMessage = new SseMessageDto(); broadcastMessage.setMessage(sseMessageDto.getMessage()); broadcastMessage.setUserIds(sseMessageDto.getUserIds()); + broadcastMessage.setSessionId(sseMessageDto.getSessionId()); + broadcastMessage.setEventDto(sseMessageDto.getEventDto()); RedisUtils.publish(SSE_TOPIC, broadcastMessage, consumer -> { - log.info("SSE发送主题订阅消息topic:{} session keys:{} message:{}", - SSE_TOPIC, sseMessageDto.getUserIds(), sseMessageDto.getMessage()); + log.info("SSE发送主题订阅消息topic:{} session:{} session keys:{} message:{}", + SSE_TOPIC, sseMessageDto.getSessionId(), sseMessageDto.getUserIds(), sseMessageDto.getMessage()); }); } diff --git a/ruoyi-common/ruoyi-common-sse/src/main/java/org/ruoyi/common/sse/dto/SseMessageDto.java b/ruoyi-common/ruoyi-common-sse/src/main/java/org/ruoyi/common/sse/dto/SseMessageDto.java index 5f914e70..f15a705f 100644 --- a/ruoyi-common/ruoyi-common-sse/src/main/java/org/ruoyi/common/sse/dto/SseMessageDto.java +++ b/ruoyi-common/ruoyi-common-sse/src/main/java/org/ruoyi/common/sse/dto/SseMessageDto.java @@ -26,4 +26,14 @@ public class SseMessageDto implements Serializable { * 需要发送的消息 */ private String message; + + /** + * 按会话定向推送的会话ID（非空时优先按会话路由，忽略 userIds） + */ + private String sessionId; + + /** + * 结构化事件（按会话定向推送时使用，message 为兼容旧逻辑保留） + */ + private SseEventDto eventDto; } diff --git a/ruoyi-common/ruoyi-common-sse/src/main/java/org/ruoyi/common/sse/listener/SseTopicListener.java b/ruoyi-common/ruoyi-common-sse/src/main/java/org/ruoyi/common/sse/listener/SseTopicListener.java index 43e87d81..bd688235 100644 --- a/ruoyi-common/ruoyi-common-sse/src/main/java/org/ruoyi/common/sse/listener/SseTopicListener.java +++ b/ruoyi-common/ruoyi-common-sse/src/main/java/org/ruoyi/common/sse/listener/SseTopicListener.java @@ -1,8 +1,10 @@ package org.ruoyi.common.sse.listener; import cn.hutool.core.collection.CollUtil; +import cn.hutool.core.util.StrUtil; import lombok.extern.slf4j.Slf4j; import org.ruoyi.common.sse.core.SseEmitterManager; +import org.ruoyi.common.sse.dto.SseMessageDto; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.ApplicationArguments; import org.springframework.boot.ApplicationRunner; @@ -28,8 +30,20 @@ public class SseTopicListener implements ApplicationRunner, Ordered { @Override public void run(ApplicationArguments args) throws Exception { sseEmitterManager.subscribeMessage((message) -> { - log.info("SSE主题订阅收到消息session keys={} message={}", message.getUserIds(), message.getMessage()); - // 如果key不为空就按照key发消息 如果为空就群发 + log.info("SSE主题订阅收到消息session:{} session keys={} message={}", + message.getSessionId(), message.getUserIds(), message.getMessage()); + // 优先按会话路由（对话流式响应） + if (StrUtil.isNotBlank(message.getSessionId())) { + if (message.getEventDto() != null) { + sseEmitterManager.sendEvent(message.getSessionId(), message.getEventDto()); + } else if (message.getMessage() != null) { + // 兼容按会话发纯文本的场景 + sseEmitterManager.sendEvent(message.getSessionId(), + org.ruoyi.common.sse.dto.SseEventDto.content(message.getMessage())); + } + return; + } + // 否则按用户/群发路由（全局通知） if (CollUtil.isNotEmpty(message.getUserIds())) { message.getUserIds().forEach(key -> { sseEmitterManager.sendMessage(key, message.getMessage()); diff --git a/ruoyi-common/ruoyi-common-sse/src/main/java/org/ruoyi/common/sse/utils/SseMessageUtils.java b/ruoyi-common/ruoyi-common-sse/src/main/java/org/ruoyi/common/sse/utils/SseMessageUtils.java index 49173343..2da4c626 100644 --- a/ruoyi-common/ruoyi-common-sse/src/main/java/org/ruoyi/common/sse/utils/SseMessageUtils.java +++ b/ruoyi-common/ruoyi-common-sse/src/main/java/org/ruoyi/common/sse/utils/SseMessageUtils.java @@ -93,6 +93,15 @@ public class SseMessageUtils { MANAGER.disconnect(userId, tokenValue); } + /** + * 完成指定会话的SSE连接 + * + * @param sessionId 会话ID + */ + public static void completeConnection(String sessionId) { + MANAGER.disconnect(sessionId); + } + /** * 向指定的SSE会话发送结构化事件 * @@ -106,6 +115,22 @@ public class SseMessageUtils { MANAGER.sendEvent(userId, eventDto); } + /** + * 向指定会话发送结构化事件（通过 Redis 广播，跨实例可达） + * + * @param sessionId 会话ID + * @param eventDto SSE事件对象 + */ + public static void sendEvent(String sessionId, SseEventDto eventDto) { + if (!isEnable() || sessionId == null) { + return; + } + SseMessageDto dto = new SseMessageDto(); + dto.setSessionId(sessionId); + dto.setEventDto(eventDto); + MANAGER.publishMessage(dto); + } + /** * 发送内容事件 * @@ -116,6 +141,16 @@ public class SseMessageUtils { sendEvent(userId, SseEventDto.content(content)); } + /** + * 向指定会话发送内容事件 + * + * @param sessionId 会话ID + * @param content 内容 + */ + public static void sendContent(String sessionId, String content) { + sendEvent(sessionId, SseEventDto.content(content)); + } + /** * 发送推理内容事件 * @@ -126,6 +161,16 @@ public class SseMessageUtils { sendEvent(userId, SseEventDto.reasoning(reasoningContent)); } + /** + * 向指定会话发送推理内容事件 + * + * @param sessionId 会话ID + * @param reasoningContent 推理内容 + */ + public static void sendReasoning(String sessionId, String reasoningContent) { + sendEvent(sessionId, SseEventDto.reasoning(reasoningContent)); + } + /** * 发送完成事件 * @@ -135,6 +180,15 @@ public class SseMessageUtils { sendEvent(userId, SseEventDto.done()); } + /** + * 向指定会话发送完成事件 + * + * @param sessionId 会话ID + */ + public static void sendDone(String sessionId) { + sendEvent(sessionId, SseEventDto.done()); + } + /** * 发送错误事件 * @@ -145,6 +199,16 @@ public class SseMessageUtils { sendEvent(userId, SseEventDto.error(error)); } + /** + * 向指定会话发送错误事件 + * + * @param sessionId 会话ID + * @param error 错误信息 + */ + public static void sendError(String sessionId, String error) { + sendEvent(sessionId, SseEventDto.error(error)); + } + /** * 是否开启 */ diff --git a/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/annotation/TraceNode.java b/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/annotation/TraceNode.java deleted file mode 100644 index 7e8a8761..00000000 --- a/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/annotation/TraceNode.java +++ /dev/null @@ -1,31 +0,0 @@ -package org.ruoyi.common.trace.annotation; - -import org.ruoyi.common.trace.constant.TraceConstants; - -import java.lang.annotation.ElementType; -import java.lang.annotation.Retention; -import java.lang.annotation.RetentionPolicy; -import java.lang.annotation.Target; - -/** - * Trace 方法节点标记。 - */ -@Target(ElementType.METHOD) -@Retention(RetentionPolicy.RUNTIME) -public @interface TraceNode { - - /** - * 节点名称。 - */ - String name() default ""; - - /** - * 节点类型。 - */ - String type() default TraceConstants.NODE_METHOD; - - /** - * 安全输入摘要。 - */ - String input() default ""; -} diff --git a/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/aspect/TraceNodeAspect.java b/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/aspect/TraceNodeAspect.java deleted file mode 100644 index 606afaf1..00000000 --- a/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/aspect/TraceNodeAspect.java +++ /dev/null @@ -1,89 +0,0 @@ -package org.ruoyi.common.trace.aspect; - -import cn.hutool.core.util.IdUtil; -import cn.hutool.core.util.StrUtil; -import lombok.RequiredArgsConstructor; -import lombok.extern.slf4j.Slf4j; -import org.aspectj.lang.ProceedingJoinPoint; -import org.aspectj.lang.annotation.Around; -import org.aspectj.lang.annotation.Aspect; -import org.aspectj.lang.reflect.MethodSignature; -import org.ruoyi.common.trace.annotation.TraceNode; -import org.ruoyi.common.trace.config.TraceProperties; -import org.ruoyi.common.trace.constant.TraceConstants; -import org.ruoyi.common.trace.core.TraceContext; -import org.ruoyi.common.trace.service.TraceRecordService; -import org.ruoyi.common.trace.util.TracePayloadUtils; -import org.springframework.core.Ordered; -import org.springframework.core.annotation.Order; - -import java.lang.reflect.Method; -import java.util.Date; - -/** - * Trace 方法节点采集切面。 - */ -@Slf4j -@Aspect -@Order(Ordered.HIGHEST_PRECEDENCE + 10) -@RequiredArgsConstructor -public class TraceNodeAspect { - - private final TraceRecordService traceRecordService; - private final TraceProperties traceProperties; - - @Around("@annotation(traceNode)") - public Object aroundNode(ProceedingJoinPoint joinPoint, TraceNode traceNode) throws Throwable { - if (!traceProperties.isEnabled() || StrUtil.isBlank(TraceContext.getTraceId())) { - return joinPoint.proceed(); - } - Method method = ((MethodSignature) joinPoint.getSignature()).getMethod(); - String nodeId = IdUtil.getSnowflakeNextIdStr(); - String traceId = TraceContext.getTraceId(); - long startMillis = System.currentTimeMillis(); - Date startTime = new Date(); - - org.ruoyi.common.trace.domain.TraceNode node = new org.ruoyi.common.trace.domain.TraceNode(); - node.setTraceId(traceId); - node.setNodeId(nodeId); - node.setParentNodeId(TraceContext.currentNodeId()); - node.setDepth(TraceContext.depth()); - node.setNodeName(StrUtil.blankToDefault(traceNode.name(), method.getName())); - node.setNodeType(StrUtil.blankToDefault(traceNode.type(), TraceConstants.NODE_METHOD)); - node.setClassName(method.getDeclaringClass().getName()); - node.setMethodName(method.getName()); - node.setStatus(TraceConstants.STATUS_RUNNING); - node.setStartTime(startTime); - node.setInputPayload(StrUtil.isBlank(traceNode.input()) ? null : traceNode.input()); - - try { - traceRecordService.startNode(node); - } catch (Exception ex) { - log.warn("写入 trace 节点失败，traceId={}", traceId, ex); - return joinPoint.proceed(); - } - - TraceContext.pushNode(nodeId); - try { - Object result = joinPoint.proceed(); - safeFinishNode(traceId, nodeId, TraceConstants.STATUS_SUCCESS, null, - TracePayloadUtils.output(result, traceProperties), startMillis); - return result; - } catch (Throwable ex) { - safeFinishNode(traceId, nodeId, TraceConstants.STATUS_ERROR, - TracePayloadUtils.error(ex, traceProperties), null, startMillis); - throw ex; - } finally { - TraceContext.popNode(); - } - } - - private void safeFinishNode(String traceId, String nodeId, String status, String errorMessage, String outputPayload, long startMillis) { - try { - traceRecordService.finishNode(traceId, nodeId, status, errorMessage, outputPayload, - new Date(), System.currentTimeMillis() - startMillis); - } catch (Exception ex) { - log.warn("更新 trace 节点失败，traceId={}, nodeId={}", traceId, nodeId, ex); - } - } -} diff --git a/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/config/TraceAutoConfiguration.java b/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/config/TraceAutoConfiguration.java index d0475311..17c0c481 100644 --- a/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/config/TraceAutoConfiguration.java +++ b/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/config/TraceAutoConfiguration.java @@ -1,24 +1,14 @@ package org.ruoyi.common.trace.config; -import org.ruoyi.common.trace.aspect.TraceNodeAspect; -import org.ruoyi.common.trace.service.TraceRecordService; import org.springframework.boot.autoconfigure.AutoConfiguration; -import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; -import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; -import org.springframework.context.annotation.Bean; /** * 通用链路追踪自动配置。 + *

+ * 仅注册配置属性；节点采集通过 {@code TraceNodeTemplate} / {@code DefaultTraceStreamSpan} 编程式埋点完成。 */ @AutoConfiguration @EnableConfigurationProperties(TraceProperties.class) public class TraceAutoConfiguration { - - @Bean - @ConditionalOnMissingBean - @ConditionalOnProperty(prefix = "trace", name = "enabled", havingValue = "true", matchIfMissing = true) - public TraceNodeAspect traceNodeAspect(TraceRecordService traceRecordService, TraceProperties traceProperties) { - return new TraceNodeAspect(traceRecordService, traceProperties); - } } diff --git a/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/config/TraceProperties.java b/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/config/TraceProperties.java index 4e3c5247..4f6349cf 100644 --- a/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/config/TraceProperties.java +++ b/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/config/TraceProperties.java @@ -23,21 +23,6 @@ public class TraceProperties { @Data public static class Payload { - /** - * 是否记录截断后的详情。默认 false，只记录摘要。 - */ - private boolean recordDetail = false; - - /** - * input payload 最大长度。 - */ - private int maxInputLength = 1000; - - /** - * output payload 最大长度。 - */ - private int maxOutputLength = 2000; - /** * 错误信息最大长度。 */ diff --git a/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/constant/TraceConstants.java b/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/constant/TraceConstants.java index 846d1589..96556721 100644 --- a/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/constant/TraceConstants.java +++ b/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/constant/TraceConstants.java @@ -12,11 +12,4 @@ public final class TraceConstants { public static final String STATUS_SUCCESS = "SUCCESS"; public static final String STATUS_ERROR = "ERROR"; public static final String STATUS_CANCELLED = "CANCELLED"; - - public static final String NODE_METHOD = "METHOD"; - public static final String NODE_HTTP = "HTTP"; - public static final String NODE_DB = "DB"; - public static final String NODE_CACHE = "CACHE"; - public static final String NODE_TASK = "TASK"; - public static final String NODE_STREAM = "STREAM"; } diff --git a/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/core/TraceContext.java b/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/core/TraceContext.java index 85db1dc0..11df5f90 100644 --- a/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/core/TraceContext.java +++ b/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/core/TraceContext.java @@ -96,14 +96,6 @@ public final class TraceContext { } } - /** - * 为未来的跨线程上下文传播保留的拷贝入口，避免共享可变栈。 - */ - public static Deque copyNodeStack() { - Deque stack = NODE_STACK.get(); - return stack == null ? new ArrayDeque<>() : new ArrayDeque<>(stack); - } - public static void clear() { TRACE_ID.remove(); BUSINESS_TYPE.remove(); diff --git a/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/domain/vo/TraceNodeVo.java b/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/domain/vo/TraceNodeVo.java index 1ac81555..d7b49811 100644 --- a/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/domain/vo/TraceNodeVo.java +++ b/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/domain/vo/TraceNodeVo.java @@ -8,7 +8,6 @@ import org.ruoyi.common.trace.domain.TraceNode; import java.io.Serial; import java.io.Serializable; import java.util.Date; -import java.util.List; import java.util.Map; /** @@ -39,17 +38,15 @@ public class TraceNodeVo implements Serializable { private Long durationMs; private String errorMessage; - /** 原始 payload 字符串（兼容旧版），新代码请使用 parsedInput */ + /** 原始 input payload 字符串，parsedInput 解析失败时回退使用 */ private String inputPayload; - /** 原始 payload 字符串（兼容旧版），新代码请使用 parsedOutput */ + /** 原始 output payload 字符串，parsedOutput 解析失败时回退使用 */ private String outputPayload; - /** 原始 metadata 字符串（兼容旧版），新代码请使用 parsedMetadata */ + /** 原始 metadata 字符串，parsedMetadata 解析失败时回退使用 */ private String metadata; - private List children; - // ======================== 展示用计算字段 ======================== /** 节点类型中文标签，如 "知识检索"、"LLM 调用" */ diff --git a/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/service/impl/TraceRecordServiceImpl.java b/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/service/impl/TraceRecordServiceImpl.java index 1b47173e..1c27f994 100644 --- a/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/service/impl/TraceRecordServiceImpl.java +++ b/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/service/impl/TraceRecordServiceImpl.java @@ -134,9 +134,6 @@ public class TraceRecordServiceImpl implements TraceRecordService { detail.setRun(run); List nodes = listNodes(traceId); - if (nodes != null) { - nodes.forEach(this::enrichNodeVo); - } // 返回扁平列表，前端自行按 parentNodeId 建树 detail.setNodes(nodes != null ? nodes : new ArrayList<>()); diff --git a/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/util/TracePayloadUtils.java b/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/util/TracePayloadUtils.java index 84497bbe..811b2711 100644 --- a/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/util/TracePayloadUtils.java +++ b/ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/util/TracePayloadUtils.java @@ -33,20 +33,6 @@ public final class TracePayloadUtils { private TracePayloadUtils() { } - public static String input(Object value, TraceProperties properties) { - if (value == null || properties == null || !properties.getPayload().isRecordDetail()) { - return null; - } - return truncate(asString(value), properties.getPayload().getMaxInputLength()); - } - - public static String output(Object value, TraceProperties properties) { - if (value == null || properties == null || !properties.getPayload().isRecordDetail()) { - return null; - } - return truncate(asString(value), properties.getPayload().getMaxOutputLength()); - } - public static String error(Throwable throwable, TraceProperties properties) { if (throwable == null) { return null; @@ -93,8 +79,4 @@ public final class TracePayloadUtils { return Collections.singletonMap("_raw", json); } } - - private static String asString(Object value) { - return value instanceof String str ? str : toJson(value); - } } diff --git a/ruoyi-common/ruoyi-common-trace/src/test/java/org/ruoyi/common/trace/aspect/TraceNodeAspectTest.java b/ruoyi-common/ruoyi-common-trace/src/test/java/org/ruoyi/common/trace/aspect/TraceNodeAspectTest.java deleted file mode 100644 index 15413b3f..00000000 --- a/ruoyi-common/ruoyi-common-trace/src/test/java/org/ruoyi/common/trace/aspect/TraceNodeAspectTest.java +++ /dev/null @@ -1,134 +0,0 @@ -package org.ruoyi.common.trace.aspect; - -import org.aspectj.lang.ProceedingJoinPoint; -import org.aspectj.lang.reflect.MethodSignature; -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.extension.ExtendWith; -import org.mockito.ArgumentCaptor; -import org.mockito.Mock; -import org.mockito.junit.jupiter.MockitoExtension; -import org.ruoyi.common.trace.annotation.TraceNode; -import org.ruoyi.common.trace.config.TraceProperties; -import org.ruoyi.common.trace.constant.TraceConstants; -import org.ruoyi.common.trace.core.TraceContext; -import org.ruoyi.common.trace.service.TraceRecordService; - -import java.lang.reflect.Method; -import java.util.Date; - -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertNull; -import static org.junit.jupiter.api.Assertions.assertThrows; -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.anyLong; -import static org.mockito.ArgumentMatchers.eq; -import static org.mockito.ArgumentMatchers.isNull; -import static org.mockito.Mockito.doThrow; -import static org.mockito.Mockito.never; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; - -@ExtendWith(MockitoExtension.class) -class TraceNodeAspectTest { - - @Mock - private TraceRecordService traceRecordService; - - @Mock - private ProceedingJoinPoint joinPoint; - - @Mock - private MethodSignature methodSignature; - - @AfterEach - void tearDown() { - TraceContext.clear(); - } - - @Test - void shouldProceedWithoutRecordingWhenNoTraceContext() throws Throwable { - TraceNodeAspect aspect = new TraceNodeAspect(traceRecordService, new TraceProperties()); - Method method = SampleService.class.getDeclaredMethod("sample"); - when(joinPoint.proceed()).thenReturn("ok"); - - Object result = aspect.aroundNode(joinPoint, method.getAnnotation(TraceNode.class)); - - assertEquals("ok", result); - verify(traceRecordService, never()).startNode(any()); - } - - @Test - void shouldRecordSuccessNode() throws Throwable { - TraceProperties properties = new TraceProperties(); - properties.getPayload().setRecordDetail(true); - TraceNodeAspect aspect = new TraceNodeAspect(traceRecordService, properties); - Method method = SampleService.class.getDeclaredMethod("sample"); - prepareSignature(method); - when(joinPoint.proceed()).thenReturn("ok"); - - try (var ignored = TraceContext.begin("trace-1", "TEST", "biz-1", 1L, "000000")) { - Object result = aspect.aroundNode(joinPoint, method.getAnnotation(TraceNode.class)); - - assertEquals("ok", result); - assertNull(TraceContext.currentNodeId()); - } - - ArgumentCaptor nodeCaptor = - ArgumentCaptor.forClass(org.ruoyi.common.trace.domain.TraceNode.class); - verify(traceRecordService).startNode(nodeCaptor.capture()); - org.ruoyi.common.trace.domain.TraceNode node = nodeCaptor.getValue(); - assertEquals("trace-1", node.getTraceId()); - assertEquals("sample-node", node.getNodeName()); - assertEquals(TraceConstants.NODE_METHOD, node.getNodeType()); - assertEquals("safe-input", node.getInputPayload()); - verify(traceRecordService).finishNode(eq("trace-1"), eq(node.getNodeId()), eq(TraceConstants.STATUS_SUCCESS), - isNull(), eq("ok"), any(Date.class), anyLong()); - } - - @Test - void shouldRethrowBusinessExceptionAfterErrorRecord() throws Throwable { - TraceNodeAspect aspect = new TraceNodeAspect(traceRecordService, new TraceProperties()); - Method method = SampleService.class.getDeclaredMethod("sample"); - prepareSignature(method); - IllegalArgumentException error = new IllegalArgumentException("bad"); - when(joinPoint.proceed()).thenThrow(error); - - try (var ignored = TraceContext.begin("trace-1", "TEST", "biz-1", 1L, "000000")) { - assertThrows(IllegalArgumentException.class, - () -> aspect.aroundNode(joinPoint, method.getAnnotation(TraceNode.class))); - } - - ArgumentCaptor nodeCaptor = - ArgumentCaptor.forClass(org.ruoyi.common.trace.domain.TraceNode.class); - verify(traceRecordService).startNode(nodeCaptor.capture()); - verify(traceRecordService).finishNode(eq("trace-1"), eq(nodeCaptor.getValue().getNodeId()), eq(TraceConstants.STATUS_ERROR), - eq("IllegalArgumentException: bad"), isNull(), any(Date.class), anyLong()); - } - - @Test - void writeFailureShouldNotAffectBusinessReturn() throws Throwable { - TraceNodeAspect aspect = new TraceNodeAspect(traceRecordService, new TraceProperties()); - Method method = SampleService.class.getDeclaredMethod("sample"); - prepareSignature(method); - when(joinPoint.proceed()).thenReturn("ok"); - doThrow(new IllegalStateException("db down")).when(traceRecordService).startNode(any()); - - try (var ignored = TraceContext.begin("trace-1", "TEST", "biz-1", 1L, "000000")) { - assertEquals("ok", aspect.aroundNode(joinPoint, method.getAnnotation(TraceNode.class))); - } - } - - private void prepareSignature(Method method) { - when(joinPoint.getSignature()).thenReturn(methodSignature); - when(methodSignature.getMethod()).thenReturn(method); - } - - private static class SampleService { - - @TraceNode(name = "sample-node", input = "safe-input") - String sample() { - return "ok"; - } - } -} diff --git a/ruoyi-common/ruoyi-common-trace/src/test/java/org/ruoyi/common/trace/core/TraceContextTest.java b/ruoyi-common/ruoyi-common-trace/src/test/java/org/ruoyi/common/trace/core/TraceContextTest.java index 1347e6ec..5dad6fd7 100644 --- a/ruoyi-common/ruoyi-common-trace/src/test/java/org/ruoyi/common/trace/core/TraceContextTest.java +++ b/ruoyi-common/ruoyi-common-trace/src/test/java/org/ruoyi/common/trace/core/TraceContextTest.java @@ -3,8 +3,6 @@ package org.ruoyi.common.trace.core; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; -import java.util.Deque; - import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNull; @@ -44,14 +42,4 @@ class TraceContextTest { assertNull(TraceContext.currentNodeId()); assertEquals(0, TraceContext.depth()); } - - @Test - void copyNodeStackShouldReturnIndependentDeque() { - TraceContext.pushNode("root"); - Deque copy = TraceContext.copyNodeStack(); - copy.push("child"); - - assertEquals("root", TraceContext.currentNodeId()); - assertEquals("child", copy.peek()); - } } diff --git a/ruoyi-common/ruoyi-common-trace/src/test/java/org/ruoyi/common/trace/util/TracePayloadUtilsTest.java b/ruoyi-common/ruoyi-common-trace/src/test/java/org/ruoyi/common/trace/util/TracePayloadUtilsTest.java index 7c787c43..5b797ddb 100644 --- a/ruoyi-common/ruoyi-common-trace/src/test/java/org/ruoyi/common/trace/util/TracePayloadUtilsTest.java +++ b/ruoyi-common/ruoyi-common-trace/src/test/java/org/ruoyi/common/trace/util/TracePayloadUtilsTest.java @@ -1,7 +1,6 @@ package org.ruoyi.common.trace.util; import org.junit.jupiter.api.Test; -import org.ruoyi.common.trace.config.TraceProperties; import java.util.Map; @@ -26,15 +25,4 @@ class TracePayloadUtilsTest { String json = TracePayloadUtils.toJson(Map.of("count", 2)); assertTrue(json.contains("\"count\":2")); } - - @Test - void detailEnabledShouldTruncateInputAndOutput() { - TraceProperties properties = new TraceProperties(); - properties.getPayload().setRecordDetail(true); - properties.getPayload().setMaxInputLength(4); - properties.getPayload().setMaxOutputLength(5); - - assertEquals("1234", TracePayloadUtils.input("123456", properties)); - assertEquals("12345", TracePayloadUtils.output("1234567", properties)); - } } diff --git a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WorkflowStarter.java b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WorkflowStarter.java index bb99f201..1a83e95c 100644 --- a/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WorkflowStarter.java +++ b/ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WorkflowStarter.java @@ -56,10 +56,10 @@ public class WorkflowStarter implements IWorkFlowStarterService { public SseEmitter streaming(User user, String workflowUuid, List userInputs, Long sessionId) { // 获取用户ID Long userId = LoginHelper.getUserId(); - // 获取登录Token + // 获取登录Token（仅透传给 WfState，工作流 SSE 通过 emitter 直发，不串台） String tokenValue = StpUtil.getTokenValue(); - // 根据用户ID和Token连接SSE对象 - SseEmitter sseEmitter = sseEmitterManager.connect(userId, tokenValue); + // 根据会话ID连接SSE对象（每会话一个连接，避免同用户多会话串台） + SseEmitter sseEmitter = sseEmitterManager.connect(String.valueOf(sessionId)); if (!sseEmitterHelper.checkOrComplete(user, sseEmitter)) { return sseEmitter; } diff --git a/ruoyi-modules/ruoyi-chat/src/main/java/org/ruoyi/observability/MyMcpClientListener.java b/ruoyi-modules/ruoyi-chat/src/main/java/org/ruoyi/observability/MyMcpClientListener.java index c1457100..b0730dcb 100644 --- a/ruoyi-modules/ruoyi-chat/src/main/java/org/ruoyi/observability/MyMcpClientListener.java +++ b/ruoyi-modules/ruoyi-chat/src/main/java/org/ruoyi/observability/MyMcpClientListener.java @@ -41,14 +41,14 @@ public class MyMcpClientListener implements McpClientListener { private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); - private final Long userId; + private final String sessionId; - public MyMcpClientListener(Long userId) { - this.userId = userId; + public MyMcpClientListener(String sessionId) { + this.sessionId = sessionId; } public MyMcpClientListener() { - this.userId = null; + this.sessionId = null; } // ==================== 工具执行 ==================== @@ -146,8 +146,8 @@ public class MyMcpClientListener implements McpClientListener { * 推送 MCP 事件到前端 */ private void pushMcpEvent(String name, String status, String result) { - if (userId == null) { - log.warn("userId 为空，无法推送 MCP 事件"); + if (sessionId == null) { + log.warn("sessionId 为空，无法推送 MCP 事件"); return; } try { @@ -157,7 +157,7 @@ public class MyMcpClientListener implements McpClientListener { content.put("result", result); String json = OBJECT_MAPPER.writeValueAsString(content); - SseMessageUtils.sendEvent(userId, SseEventDto.builder() + SseMessageUtils.sendEvent(sessionId, SseEventDto.builder() .event("mcp") .content(json) .build()); diff --git a/ruoyi-modules/ruoyi-chat/src/main/java/org/ruoyi/service/chat/impl/ChatServiceFacade.java b/ruoyi-modules/ruoyi-chat/src/main/java/org/ruoyi/service/chat/impl/ChatServiceFacade.java index 2f04c806..32c86960 100644 --- a/ruoyi-modules/ruoyi-chat/src/main/java/org/ruoyi/service/chat/impl/ChatServiceFacade.java +++ b/ruoyi-modules/ruoyi-chat/src/main/java/org/ruoyi/service/chat/impl/ChatServiceFacade.java @@ -72,7 +72,6 @@ import org.ruoyi.domain.vo.agent.AgentVo; import org.ruoyi.domain.vo.knowledge.KnowledgeInfoVo; import org.ruoyi.factory.ChatServiceFactory; import org.ruoyi.mcp.service.core.LangChain4jMcpToolProviderService; -import org.ruoyi.mcp.service.core.ToolProviderFactory; import org.ruoyi.observability.*; import org.ruoyi.service.agent.IAgentService; import org.ruoyi.service.chat.AbstractChatService; @@ -81,7 +80,6 @@ import org.ruoyi.service.chat.impl.memory.PersistentChatMemoryStore; import org.ruoyi.service.knowledge.IKnowledgeInfoService; import org.ruoyi.service.retrieval.KnowledgeRetrievalService; import org.ruoyi.service.knowledge.retriever.CustomVectorRetriever; -import org.ruoyi.service.vector.VectorStoreService; import org.ruoyi.trace.RagTraceNodeTypes; import org.ruoyi.trace.RagTracePayloadBuilder; import org.springframework.stereotype.Service; @@ -120,8 +118,6 @@ public class ChatServiceFacade implements IChatService { private final IKnowledgeInfoService knowledgeInfoService; - private final VectorStoreService vectorStoreService; - private final KnowledgeRetrievalService knowledgeRetrievalService; private final SseEmitterManager sseEmitterManager; @@ -130,7 +126,6 @@ public class ChatServiceFacade implements IChatService { private final IWorkFlowStarterService workFlowStarterService; - private final ToolProviderFactory toolProviderFactory; private final IAgentService agentService; @@ -156,13 +151,13 @@ public class ChatServiceFacade implements IChatService { */ public SseEmitter sseChat(ChatRequest chatRequest) { - // 4. 具体的服务实现 + // 具体的服务实现 Long userId = LoginHelper.getUserId(); String tokenValue = StpUtil.getTokenValue(); - SseEmitter emitter = sseEmitterManager.connect(userId, tokenValue); + // 每个会话一个 SSE 连接，避免同用户多会话串台 + SseEmitter emitter = sseEmitterManager.connect(String.valueOf(chatRequest.getSessionId())); - // 0. 智能体解析：传入 agentId 时按智能体绑定的模型覆盖 model 字段 - // （前端默认走智能体；enableThinking 不再作为对话模式开关，Supervisor 多 Agent 编排成为默认智能体路径） + // 智能体解析：传入 agentId 时按智能体绑定的模型覆盖 model 字段 AgentVo agentVo = null; if (chatRequest.getAgentId() != null) { agentVo = agentService.queryById(chatRequest.getAgentId()); @@ -176,14 +171,14 @@ public class ChatServiceFacade implements IChatService { } } - // 1. 根据模型名称查询完整配置 + // 根据模型名称查询完整配置 ChatModelVo chatModelVo = chatModelService.selectModelByName(chatRequest.getModel()); if (chatModelVo == null) { throw new IllegalArgumentException("模型不存在: " + chatRequest.getModel()); } - // 2. 构建上下文消息列表（系统提示词 + 历史消息 + 当前用户消息） - // 注意：RAG 检索增强统一在 handleAgentChat 中执行一次，此处不再重复检索 + // 构建上下文消息列表（系统提示词 + 历史消息 + 当前用户消息） + // 注意：RAG 检索增强统一在 handleAgentChat 中执行一次，此处不再重复检索 List contextMessages = buildContextMessages(chatRequest, agentVo); chatRequest.setEmitter(emitter); @@ -242,13 +237,14 @@ public class ChatServiceFacade implements IChatService { ChatModel plannerModel = chatService.buildChatModel(chatModelVo); Long userId = chatRequest.getUserId(); + String sessionId = String.valueOf(chatRequest.getSessionId()); // 工具装配：智能体有关联工具ID时按ID装配，否则回退到原有硬编码 MCP 客户端 ToolProvider toolProvider; if (agentVo != null && agentVo.getMcpToolIds() != null && !agentVo.getMcpToolIds().isEmpty()) { toolProvider = langChain4jMcpToolProviderService.getToolProvider(agentVo.getMcpToolIds()); } else { - toolProvider = buildDefaultMcpToolProvider(userId); + toolProvider = buildDefaultMcpToolProvider(sessionId); } // Skills 装配：智能体有勾选技能名时按名过滤磁盘 skills，否则加载全部 @@ -321,16 +317,14 @@ public class ChatServiceFacade implements IChatService { promptBuilder.append(augmentedInput); String prompt = promptBuilder.toString(); - String tokenValue = chatRequest.getTokenValue(); - // 异步执行 supervisor，避免阻塞 HTTP 请求线程导致 SSE 事件被缓冲 CompletableFuture.runAsync(() -> { TraceStreamSpan llmSpan = null; try (TraceScope ignored = openTraceScope(traceRun, userId)) { llmSpan = startLlmCallSpan(traceRun, chatRequest); String result = supervisor.invoke(prompt); - SseMessageUtils.sendContent(userId, result); - SseMessageUtils.sendDone(userId); + SseMessageUtils.sendContent(sessionId, result); + SseMessageUtils.sendDone(sessionId); // 保存助手回复到数据库（智能体对话为默认路径后，需在此落库以保留历史） if (StringUtils.isNotBlank(result)) { chatMessageService.saveChatMessage(userId, chatRequest.getSessionId(), @@ -347,12 +341,12 @@ public class ChatServiceFacade implements IChatService { } finishTraceRun(traceRun, TraceConstants.STATUS_ERROR, e); log.error("Supervisor 执行失败", e); - SseMessageUtils.sendError(userId, e.getMessage()); + SseMessageUtils.sendError(sessionId, e.getMessage()); } finally { if (llmSpan != null) { llmSpan.detach(); } - SseMessageUtils.completeConnection(userId, tokenValue); + SseMessageUtils.completeConnection(sessionId); } }); return chatRequest.getEmitter(); @@ -461,8 +455,10 @@ public class ChatServiceFacade implements IChatService { /** * 兜底 MCP 工具装配（无智能体时使用，保留原有 3 个硬编码客户端逻辑） + * + * @param sessionId 会话ID，用于 MCP 工具事件按会话推送 SSE */ - private ToolProvider buildDefaultMcpToolProvider(Long userId) { + private ToolProvider buildDefaultMcpToolProvider(String sessionId) { String npxCommand = resolveNpxCommand(); McpTransport playwrightTransport = new StdioMcpTransport.Builder() .command(List.of(npxCommand, "-y", "@playwright/mcp@latest")) @@ -470,7 +466,7 @@ public class ChatServiceFacade implements IChatService { .build(); McpClient playwrightMcpClient = new DefaultMcpClient.Builder() .transport(playwrightTransport) - .listener(new MyMcpClientListener(userId)) + .listener(new MyMcpClientListener(sessionId)) .build(); String userDir = System.getProperty("user.dir"); @@ -481,7 +477,7 @@ public class ChatServiceFacade implements IChatService { .build(); McpClient filesystemMcpClient = new DefaultMcpClient.Builder() .transport(filesystemTransport) - .listener(new MyMcpClientListener(userId)) + .listener(new MyMcpClientListener(sessionId)) .build(); return McpToolProvider.builder() @@ -566,16 +562,15 @@ public class ChatServiceFacade implements IChatService { // 4. 获取用户信息 Long userId = LoginHelper.getUserId(); - String tokenValue = StpUtil.getTokenValue(); - // 5. 建立 SSE 连接（用于前端监听） - sseEmitterManager.connect(userId, tokenValue); + // 5. 建立 SSE 连接（用于前端监听，按会话隔离） + sseEmitterManager.connect(String.valueOf(chatRequest.getSessionId())); // 保存用户消息 chatMessageService.saveChatMessage(userId, chatRequest.getSessionId(), chatRequest.getContent(), RoleType.USER.getName(), chatRequest.getModel()); // 6. 创建组合 handler：同时发送到 SSE 和外部 handler - StreamingChatResponseHandler combinedHandler = createCombinedHandler(userId, tokenValue, externalHandler); + StreamingChatResponseHandler combinedHandler = createCombinedHandler(String.valueOf(chatRequest.getSessionId()), externalHandler); // 7. 发起对话 StreamingChatModel streamingChatModel = chatService.buildStreamingChatModel(chatModelVo, chatRequest); @@ -799,12 +794,11 @@ public class ChatServiceFacade implements IChatService { /** * 创建组合响应处理器 - 同时发送到 SSE 和外部 handler * - * @param userId 用户ID - * @param tokenValue 会话令牌 + * @param sessionId 会话ID（SSE 按会话隔离推送） * @param externalHandler 外部响应处理器（可为 null） * @return 组合的流式响应处理器 */ - protected StreamingChatResponseHandler createCombinedHandler(Long userId, String tokenValue, + protected StreamingChatResponseHandler createCombinedHandler(String sessionId, StreamingChatResponseHandler externalHandler) { return new StreamingChatResponseHandler() { @@ -817,7 +811,7 @@ public class ChatServiceFacade implements IChatService { messageBuffer.append(partialResponse); // 2. 发送内容事件到 SSE（前端可通过 SSE 监听） - SseMessageUtils.sendContent(userId, partialResponse); + SseMessageUtils.sendContent(sessionId, partialResponse); // 3. 转发给外部 handler（Workflow 等模块可处理） if (externalHandler != null) { @@ -828,7 +822,7 @@ public class ChatServiceFacade implements IChatService { @Override public void onPartialThinking(PartialThinking partialThinking) { // 发送推理内容到 SSE（前端通过 reasoning 事件监听） - SseMessageUtils.sendReasoning(userId, partialThinking.text()); + SseMessageUtils.sendReasoning(sessionId, partialThinking.text()); // 转发给外部 handler if (externalHandler != null) { @@ -840,10 +834,10 @@ public class ChatServiceFacade implements IChatService { public void onCompleteResponse(ChatResponse completeResponse) { try { // 1. 发送完成事件 - SseMessageUtils.sendDone(userId); + SseMessageUtils.sendDone(sessionId); // 2. 关闭 SSE 连接 - SseMessageUtils.completeConnection(userId, tokenValue); + SseMessageUtils.completeConnection(sessionId); // 3. 转发给外部 handler if (externalHandler != null) { @@ -857,7 +851,7 @@ public class ChatServiceFacade implements IChatService { @Override public void onError(Throwable error) { // 发送错误事件 - SseMessageUtils.sendError(userId, error.getMessage()); + SseMessageUtils.sendError(sessionId, error.getMessage()); log.error("流式响应错误: {}", error.getMessage(), error); // 转发给外部 handler diff --git a/ruoyi-modules/ruoyi-chat/src/main/java/org/ruoyi/trace/RagTraceNodeTypes.java b/ruoyi-modules/ruoyi-chat/src/main/java/org/ruoyi/trace/RagTraceNodeTypes.java index ab751e03..edc04bb4 100644 --- a/ruoyi-modules/ruoyi-chat/src/main/java/org/ruoyi/trace/RagTraceNodeTypes.java +++ b/ruoyi-modules/ruoyi-chat/src/main/java/org/ruoyi/trace/RagTraceNodeTypes.java @@ -14,5 +14,4 @@ public final class RagTraceNodeTypes { public static final String NODE_RETRIEVAL = "RETRIEVAL"; public static final String NODE_RERANK = "RERANK"; public static final String NODE_LLM_CALL = "LLM_CALL"; - public static final String NODE_STREAM = "STREAM"; }