mirror of
https://gitcode.com/ageerle/ruoyi-ai.git
synced 2026-09-13 00:14:59 +00:00
Compare commits
7 Commits
9adeee6c83
...
afaa86ef6e
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
afaa86ef6e | ||
|
|
d1a820728f | ||
|
|
11bb1dba0f | ||
|
|
345d2819d8 | ||
|
|
36e3cf8c52 | ||
|
|
06ff95cdc3 | ||
|
|
e925050207 |
128
docs/script/sql/update/update-0615-trace.sql
Normal file
128
docs/script/sql/update/update-0615-trace.sql
Normal file
@@ -0,0 +1,128 @@
|
||||
-- 链路追踪运行记录表
|
||||
CREATE TABLE IF NOT EXISTS `trace_run` (
|
||||
`id` bigint NOT NULL COMMENT '主键',
|
||||
`trace_id` varchar(64) NOT NULL COMMENT '链路ID',
|
||||
`trace_name` varchar(128) NOT NULL COMMENT '链路名称',
|
||||
`business_type` varchar(64) NOT NULL COMMENT '业务类型',
|
||||
`business_id` varchar(128) DEFAULT NULL COMMENT '业务ID',
|
||||
`user_id` bigint DEFAULT NULL COMMENT '用户ID',
|
||||
`tenant_id` varchar(20) DEFAULT '000000' COMMENT '租户编号',
|
||||
`status` varchar(32) NOT NULL COMMENT '状态',
|
||||
`start_time` datetime NOT NULL COMMENT '开始时间',
|
||||
`end_time` datetime DEFAULT NULL COMMENT '结束时间',
|
||||
`duration_ms` bigint DEFAULT NULL COMMENT '耗时毫秒',
|
||||
`error_message` varchar(1000) DEFAULT NULL COMMENT '错误摘要',
|
||||
`metadata` text DEFAULT NULL COMMENT '元数据JSON',
|
||||
`create_dept` bigint DEFAULT NULL COMMENT '创建部门',
|
||||
`create_by` bigint DEFAULT NULL COMMENT '创建者',
|
||||
`create_time` datetime DEFAULT NULL COMMENT '创建时间',
|
||||
`update_by` bigint DEFAULT NULL COMMENT '更新者',
|
||||
`update_time` datetime DEFAULT NULL COMMENT '更新时间',
|
||||
`del_flag` char(1) DEFAULT '0' COMMENT '删除标志(0代表存在 2代表删除)',
|
||||
PRIMARY KEY (`id`) USING BTREE,
|
||||
UNIQUE KEY `uk_trace_run_trace_id` (`trace_id`) USING BTREE,
|
||||
KEY `idx_trace_run_business` (`business_type`, `business_id`) USING BTREE,
|
||||
KEY `idx_trace_run_status_time` (`status`, `start_time`) USING BTREE,
|
||||
KEY `idx_trace_run_tenant_time` (`tenant_id`, `start_time`) USING BTREE
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='链路追踪运行记录表' ROW_FORMAT=DYNAMIC;
|
||||
|
||||
-- 链路追踪节点记录表
|
||||
CREATE TABLE IF NOT EXISTS `trace_node` (
|
||||
`id` bigint NOT NULL COMMENT '主键',
|
||||
`trace_id` varchar(64) NOT NULL COMMENT '链路ID',
|
||||
`node_id` varchar(64) NOT NULL COMMENT '节点ID',
|
||||
`tenant_id` varchar(20) DEFAULT '000000' COMMENT '租户编号',
|
||||
`parent_node_id` varchar(64) DEFAULT NULL COMMENT '父节点ID',
|
||||
`node_name` varchar(128) NOT NULL COMMENT '节点名称',
|
||||
`node_type` varchar(64) NOT NULL COMMENT '节点类型',
|
||||
`depth` int DEFAULT 0 COMMENT '节点深度',
|
||||
`sort_order` int DEFAULT 0 COMMENT '排序',
|
||||
`class_name` varchar(255) DEFAULT NULL COMMENT '类名',
|
||||
`method_name` varchar(128) DEFAULT NULL COMMENT '方法名',
|
||||
`status` varchar(32) NOT NULL COMMENT '状态',
|
||||
`start_time` datetime NOT NULL COMMENT '开始时间',
|
||||
`end_time` datetime DEFAULT NULL COMMENT '结束时间',
|
||||
`duration_ms` bigint DEFAULT NULL COMMENT '耗时毫秒',
|
||||
`error_message` varchar(1000) DEFAULT NULL COMMENT '错误摘要',
|
||||
`input_payload` text DEFAULT NULL COMMENT '输入JSON',
|
||||
`output_payload` text DEFAULT NULL COMMENT '输出JSON',
|
||||
`metadata` text DEFAULT NULL COMMENT '元数据JSON',
|
||||
`create_dept` bigint DEFAULT NULL COMMENT '创建部门',
|
||||
`create_by` bigint DEFAULT NULL COMMENT '创建者',
|
||||
`create_time` datetime DEFAULT NULL COMMENT '创建时间',
|
||||
`update_by` bigint DEFAULT NULL COMMENT '更新者',
|
||||
`update_time` datetime DEFAULT NULL COMMENT '更新时间',
|
||||
`del_flag` char(1) DEFAULT '0' COMMENT '删除标志(0代表存在 2代表删除)',
|
||||
PRIMARY KEY (`id`) USING BTREE,
|
||||
KEY `idx_trace_node_trace_id` (`trace_id`) USING BTREE,
|
||||
KEY `idx_trace_node_parent` (`trace_id`, `parent_node_id`) USING BTREE,
|
||||
KEY `idx_trace_node_time` (`trace_id`, `start_time`) USING BTREE,
|
||||
KEY `idx_trace_node_tenant_time` (`tenant_id`, `start_time`) USING BTREE,
|
||||
KEY `idx_trace_node_type_status` (`node_type`, `status`) USING BTREE
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='链路追踪节点记录表' ROW_FORMAT=DYNAMIC;
|
||||
|
||||
-- 兼容已存在的旧 trace 表:CREATE TABLE IF NOT EXISTS 不会给旧表补新字段
|
||||
SET @trace_run_add_tenant_sql = (
|
||||
SELECT IF(COUNT(*) = 0,
|
||||
'ALTER TABLE `trace_run` ADD COLUMN `tenant_id` varchar(20) DEFAULT ''000000'' COMMENT ''租户编号'' AFTER `user_id`',
|
||||
'SELECT 1'
|
||||
)
|
||||
FROM information_schema.COLUMNS
|
||||
WHERE TABLE_SCHEMA = DATABASE()
|
||||
AND TABLE_NAME = 'trace_run'
|
||||
AND COLUMN_NAME = 'tenant_id'
|
||||
);
|
||||
PREPARE trace_run_add_tenant_stmt FROM @trace_run_add_tenant_sql;
|
||||
EXECUTE trace_run_add_tenant_stmt;
|
||||
DEALLOCATE PREPARE trace_run_add_tenant_stmt;
|
||||
|
||||
SET @trace_run_add_tenant_idx_sql = (
|
||||
SELECT IF(COUNT(*) = 0,
|
||||
'ALTER TABLE `trace_run` ADD INDEX `idx_trace_run_tenant_time` (`tenant_id`, `start_time`) USING BTREE',
|
||||
'SELECT 1'
|
||||
)
|
||||
FROM information_schema.STATISTICS
|
||||
WHERE TABLE_SCHEMA = DATABASE()
|
||||
AND TABLE_NAME = 'trace_run'
|
||||
AND INDEX_NAME = 'idx_trace_run_tenant_time'
|
||||
);
|
||||
PREPARE trace_run_add_tenant_idx_stmt FROM @trace_run_add_tenant_idx_sql;
|
||||
EXECUTE trace_run_add_tenant_idx_stmt;
|
||||
DEALLOCATE PREPARE trace_run_add_tenant_idx_stmt;
|
||||
|
||||
SET @trace_node_add_tenant_sql = (
|
||||
SELECT IF(COUNT(*) = 0,
|
||||
'ALTER TABLE `trace_node` ADD COLUMN `tenant_id` varchar(20) DEFAULT ''000000'' COMMENT ''租户编号'' AFTER `node_id`',
|
||||
'SELECT 1'
|
||||
)
|
||||
FROM information_schema.COLUMNS
|
||||
WHERE TABLE_SCHEMA = DATABASE()
|
||||
AND TABLE_NAME = 'trace_node'
|
||||
AND COLUMN_NAME = 'tenant_id'
|
||||
);
|
||||
PREPARE trace_node_add_tenant_stmt FROM @trace_node_add_tenant_sql;
|
||||
EXECUTE trace_node_add_tenant_stmt;
|
||||
DEALLOCATE PREPARE trace_node_add_tenant_stmt;
|
||||
|
||||
SET @trace_node_add_tenant_idx_sql = (
|
||||
SELECT IF(COUNT(*) = 0,
|
||||
'ALTER TABLE `trace_node` ADD INDEX `idx_trace_node_tenant_time` (`tenant_id`, `start_time`) USING BTREE',
|
||||
'SELECT 1'
|
||||
)
|
||||
FROM information_schema.STATISTICS
|
||||
WHERE TABLE_SCHEMA = DATABASE()
|
||||
AND TABLE_NAME = 'trace_node'
|
||||
AND INDEX_NAME = 'idx_trace_node_tenant_time'
|
||||
);
|
||||
PREPARE trace_node_add_tenant_idx_stmt FROM @trace_node_add_tenant_idx_sql;
|
||||
EXECUTE trace_node_add_tenant_idx_stmt;
|
||||
DEALLOCATE PREPARE trace_node_add_tenant_idx_stmt;
|
||||
|
||||
-- 链路追踪监控菜单 & 按钮权限
|
||||
INSERT INTO `sys_menu`
|
||||
(`menu_id`, `menu_name`, `parent_id`, `order_num`, `path`, `component`, `query_param`, `is_frame`, `is_cache`, `menu_type`, `visible`, `status`, `perms`, `icon`, `create_dept`, `create_by`, `create_time`, `remark`)
|
||||
SELECT (SELECT COALESCE(MAX(`menu_id`), 0) + 1 FROM (SELECT `menu_id` FROM `sys_menu`) t), '链路追踪', 2, 7, 'trace', 'monitor/trace/index', '', 1, 0, 'C', '0', '0', 'monitor:trace:list', 'tabler:route', 103, 1, NOW(), '链路追踪监控菜单';
|
||||
|
||||
INSERT INTO `sys_menu`
|
||||
(`menu_id`, `menu_name`, `parent_id`, `order_num`, `path`, `component`, `query_param`, `is_frame`, `is_cache`, `menu_type`, `visible`, `status`, `perms`, `icon`, `create_dept`, `create_by`, `create_time`, `remark`)
|
||||
SELECT (SELECT COALESCE(MAX(`menu_id`), 0) + 1 FROM (SELECT `menu_id` FROM `sys_menu`) t), '链路追踪查询', (SELECT `menu_id` FROM `sys_menu` WHERE `perms` = 'monitor:trace:list' AND `menu_type` = 'C' LIMIT 1), 1, '#', '', '', 1, 0, 'F', '0', '0', 'monitor:trace:query', '#', 103, 1, NOW(), '';
|
||||
@@ -146,6 +146,9 @@ tenant:
|
||||
- sys_client
|
||||
- sys_oss_config
|
||||
- flow_spel
|
||||
# 链路追踪监控表:运维需跨租户全局查看,且 trace_node 在异步线程写入、租户上下文不传播,故排除租户过滤
|
||||
- trace_run
|
||||
- trace_node
|
||||
|
||||
# MyBatisPlus配置
|
||||
# https://baomidou.com/config/
|
||||
@@ -232,6 +235,15 @@ xss:
|
||||
excludeUrls:
|
||||
- /system/notice
|
||||
|
||||
--- # 链路追踪配置
|
||||
trace:
|
||||
# 是否启用链路追踪,默认 true
|
||||
# 关闭后所有埋点代码会直接透传业务逻辑,不写库、不创建上下文,零性能开销
|
||||
enabled: true
|
||||
payload:
|
||||
# 错误信息最大字符长度(格式: "异常类名: 异常消息"),超过部分会被截断丢弃
|
||||
max-error-length: 1000
|
||||
|
||||
--- # 分布式锁 lock4j 全局配置
|
||||
lock4j:
|
||||
# 获取分布式锁超时时间,默认为 3000 毫秒
|
||||
|
||||
@@ -35,6 +35,7 @@
|
||||
<module>ruoyi-common-tenant</module>
|
||||
<module>ruoyi-common-websocket</module>
|
||||
<module>ruoyi-common-sse</module>
|
||||
<module>ruoyi-common-trace</module>
|
||||
</modules>
|
||||
|
||||
<artifactId>ruoyi-common</artifactId>
|
||||
|
||||
@@ -186,6 +186,13 @@
|
||||
<version>${revision}</version>
|
||||
</dependency>
|
||||
|
||||
<!-- 链路追踪模块 -->
|
||||
<dependency>
|
||||
<groupId>org.ruoyi</groupId>
|
||||
<artifactId>ruoyi-common-trace</artifactId>
|
||||
<version>${revision}</version>
|
||||
</dependency>
|
||||
|
||||
</dependencies>
|
||||
</dependencyManagement>
|
||||
|
||||
|
||||
@@ -32,6 +32,16 @@ public class SseEmitterManager {
|
||||
*/
|
||||
private final static String SSE_TOPIC = "global:sse";
|
||||
|
||||
/**
|
||||
* 按会话维度管理:每个会话一个 SSE 连接,用于对话流式响应
|
||||
* Key: sessionId
|
||||
*/
|
||||
private final static Map<String, SseEmitter> SESSION_EMITTERS = new ConcurrentHashMap<>();
|
||||
|
||||
/**
|
||||
* 按用户维度管理:全局通知、站内信等场景,一个用户可有多个连接(按 token 区分)
|
||||
* Key: userId, Value: token -> SseEmitter
|
||||
*/
|
||||
private final static Map<Long, Map<String, SseEmitter>> 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());
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@@ -26,4 +26,14 @@ public class SseMessageDto implements Serializable {
|
||||
* 需要发送的消息
|
||||
*/
|
||||
private String message;
|
||||
|
||||
/**
|
||||
* 按会话定向推送的会话ID(非空时优先按会话路由,忽略 userIds)
|
||||
*/
|
||||
private String sessionId;
|
||||
|
||||
/**
|
||||
* 结构化事件(按会话定向推送时使用,message 为兼容旧逻辑保留)
|
||||
*/
|
||||
private SseEventDto eventDto;
|
||||
}
|
||||
|
||||
@@ -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());
|
||||
|
||||
@@ -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));
|
||||
}
|
||||
|
||||
/**
|
||||
* 是否开启
|
||||
*/
|
||||
|
||||
42
ruoyi-common/ruoyi-common-trace/pom.xml
Normal file
42
ruoyi-common/ruoyi-common-trace/pom.xml
Normal file
@@ -0,0 +1,42 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<parent>
|
||||
<groupId>org.ruoyi</groupId>
|
||||
<artifactId>ruoyi-common</artifactId>
|
||||
<version>${revision}</version>
|
||||
</parent>
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<artifactId>ruoyi-common-trace</artifactId>
|
||||
|
||||
<description>
|
||||
ruoyi-common-trace 通用链路追踪
|
||||
</description>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.ruoyi</groupId>
|
||||
<artifactId>ruoyi-common-core</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.ruoyi</groupId>
|
||||
<artifactId>ruoyi-common-mybatis</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.ruoyi</groupId>
|
||||
<artifactId>ruoyi-common-json</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.mockito</groupId>
|
||||
<artifactId>mockito-junit-jupiter</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</project>
|
||||
@@ -0,0 +1,14 @@
|
||||
package org.ruoyi.common.trace.config;
|
||||
|
||||
import org.springframework.boot.autoconfigure.AutoConfiguration;
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
|
||||
/**
|
||||
* 通用链路追踪自动配置。
|
||||
* <p>
|
||||
* 仅注册配置属性;节点采集通过 {@code TraceNodeTemplate} / {@code DefaultTraceStreamSpan} 编程式埋点完成。
|
||||
*/
|
||||
@AutoConfiguration
|
||||
@EnableConfigurationProperties(TraceProperties.class)
|
||||
public class TraceAutoConfiguration {
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
package org.ruoyi.common.trace.config;
|
||||
|
||||
import lombok.Data;
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
|
||||
/**
|
||||
* 通用链路追踪配置。
|
||||
*/
|
||||
@Data
|
||||
@ConfigurationProperties(prefix = "trace")
|
||||
public class TraceProperties {
|
||||
|
||||
/**
|
||||
* 是否启用链路追踪。
|
||||
*/
|
||||
private boolean enabled = true;
|
||||
|
||||
/**
|
||||
* payload 记录策略。
|
||||
*/
|
||||
private Payload payload = new Payload();
|
||||
|
||||
@Data
|
||||
public static class Payload {
|
||||
|
||||
/**
|
||||
* 错误信息最大长度。
|
||||
*/
|
||||
private int maxErrorLength = 1000;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,15 @@
|
||||
package org.ruoyi.common.trace.constant;
|
||||
|
||||
/**
|
||||
* 通用链路追踪常量。
|
||||
*/
|
||||
public final class TraceConstants {
|
||||
|
||||
private TraceConstants() {
|
||||
}
|
||||
|
||||
public static final String STATUS_RUNNING = "RUNNING";
|
||||
public static final String STATUS_SUCCESS = "SUCCESS";
|
||||
public static final String STATUS_ERROR = "ERROR";
|
||||
public static final String STATUS_CANCELLED = "CANCELLED";
|
||||
}
|
||||
@@ -0,0 +1,163 @@
|
||||
package org.ruoyi.common.trace.constant;
|
||||
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* 链路追踪中文展示常量。
|
||||
* <p>
|
||||
* 将技术标识映射为用户可读的中文标签,供前端直接展示。
|
||||
*/
|
||||
public final class TraceDisplayConstants {
|
||||
|
||||
private TraceDisplayConstants() {
|
||||
}
|
||||
|
||||
// ======================== 节点类型中文映射 ========================
|
||||
|
||||
/**
|
||||
* 通用节点类型中文标签。
|
||||
*/
|
||||
public static final Map<String, String> NODE_TYPE_LABELS = Map.ofEntries(
|
||||
Map.entry("RETRIEVAL", "知识检索"),
|
||||
Map.entry("RERANK", "重排序"),
|
||||
Map.entry("LLM_CALL", "LLM 调用"),
|
||||
Map.entry("STREAM", "流式输出"),
|
||||
Map.entry("METHOD", "方法调用"),
|
||||
Map.entry("HTTP", "HTTP 请求"),
|
||||
Map.entry("DB", "数据库"),
|
||||
Map.entry("CACHE", "缓存操作"),
|
||||
Map.entry("TASK", "异步任务"),
|
||||
Map.entry("ROOT", "根节点")
|
||||
);
|
||||
|
||||
/**
|
||||
* 业务类型中文标签。
|
||||
*/
|
||||
public static final Map<String, String> BUSINESS_TYPE_LABELS = Map.ofEntries(
|
||||
Map.entry("RAG_CHAT", "RAG 对话"),
|
||||
Map.entry("API", "API 调用"),
|
||||
Map.entry("SCHEDULED", "定时任务")
|
||||
);
|
||||
|
||||
/**
|
||||
* 状态中文标签。
|
||||
*/
|
||||
public static final Map<String, String> STATUS_LABELS = Map.ofEntries(
|
||||
Map.entry("RUNNING", "运行中"),
|
||||
Map.entry("SUCCESS", "成功"),
|
||||
Map.entry("ERROR", "失败"),
|
||||
Map.entry("CANCELLED", "已取消"),
|
||||
Map.entry("TIMEOUT", "超时")
|
||||
);
|
||||
|
||||
// ======================== 工具方法 ========================
|
||||
|
||||
/**
|
||||
* 获取节点类型中文标签,未匹配时返回原始值。
|
||||
*/
|
||||
public static String nodeTypeLabel(String nodeType) {
|
||||
if (nodeType == null) {
|
||||
return "-";
|
||||
}
|
||||
return NODE_TYPE_LABELS.getOrDefault(nodeType.toUpperCase(), nodeType);
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取业务类型中文标签,未匹配时返回原始值。
|
||||
*/
|
||||
public static String businessTypeLabel(String businessType) {
|
||||
if (businessType == null) {
|
||||
return "-";
|
||||
}
|
||||
return BUSINESS_TYPE_LABELS.getOrDefault(businessType.toUpperCase(), businessType);
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取状态中文标签,未匹配时返回原始值。
|
||||
*/
|
||||
public static String statusLabel(String status) {
|
||||
if (status == null) {
|
||||
return "未知";
|
||||
}
|
||||
return STATUS_LABELS.getOrDefault(status.toUpperCase(), status);
|
||||
}
|
||||
|
||||
/**
|
||||
* 将技术节点名称转为可读展示名。
|
||||
* <p>
|
||||
* 支持 kebab-case / snake_case / camelCase → 首字母大写空格分隔。
|
||||
*/
|
||||
public static String prettifyNodeName(String raw) {
|
||||
if (raw == null || raw.isBlank()) {
|
||||
return "-";
|
||||
}
|
||||
String trimmed = raw.trim();
|
||||
// 已知映射优先
|
||||
Map<String, String> known = Map.ofEntries(
|
||||
Map.entry("rag-chat", "RAG 流式对话"),
|
||||
Map.entry("rag-stream-chat", "RAG 流式对话"),
|
||||
Map.entry("retrieval-engine", "知识库检索"),
|
||||
Map.entry("multi-channel-retrieval", "多路召回"),
|
||||
Map.entry("context-build", "上下文组装"),
|
||||
Map.entry("prompt-render", "Prompt 渲染"),
|
||||
Map.entry("query-rewrite-and-split", "问题改写与拆分"),
|
||||
Map.entry("intent-resolve", "意图识别"),
|
||||
Map.entry("guidance-detect", "歧义引导"),
|
||||
Map.entry("conversation-title-gen", "会话标题生成"),
|
||||
Map.entry("user-first-packet", "用户感知首包"),
|
||||
Map.entry("llm-first-packet", "LLM 首包"),
|
||||
Map.entry("llm-chat-routing", "LLM 路由调度"),
|
||||
Map.entry("llm-stream-routing", "LLM 流式路由")
|
||||
);
|
||||
if (known.containsKey(trimmed)) {
|
||||
return known.get(trimmed);
|
||||
}
|
||||
// 通用格式化: 按 [-_] 分割,每段首字母大写
|
||||
String[] parts = trimmed.split("[-_\\s]+");
|
||||
StringBuilder sb = new StringBuilder();
|
||||
for (String part : parts) {
|
||||
if (part.isEmpty()) {
|
||||
continue;
|
||||
}
|
||||
if (sb.length() > 0) {
|
||||
sb.append(' ');
|
||||
}
|
||||
sb.append(Character.toUpperCase(part.charAt(0)));
|
||||
if (part.length() > 1) {
|
||||
sb.append(part.substring(1));
|
||||
}
|
||||
}
|
||||
return sb.length() > 0 ? sb.toString() : trimmed;
|
||||
}
|
||||
|
||||
/**
|
||||
* 判断状态是否为失败。
|
||||
*/
|
||||
public static boolean isFailed(String status) {
|
||||
if (status == null) {
|
||||
return false;
|
||||
}
|
||||
String upper = status.toUpperCase();
|
||||
return "ERROR".equals(upper) || "FAILED".equals(upper) || "TIMEOUT".equals(upper);
|
||||
}
|
||||
|
||||
/**
|
||||
* 判断状态是否为成功。
|
||||
*/
|
||||
public static boolean isSuccess(String status) {
|
||||
if (status == null) {
|
||||
return false;
|
||||
}
|
||||
return "SUCCESS".equalsIgnoreCase(status);
|
||||
}
|
||||
|
||||
/**
|
||||
* 判断状态是否为运行中。
|
||||
*/
|
||||
public static boolean isRunning(String status) {
|
||||
if (status == null) {
|
||||
return false;
|
||||
}
|
||||
return "RUNNING".equalsIgnoreCase(status);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,71 @@
|
||||
package org.ruoyi.common.trace.core;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.ruoyi.common.trace.config.TraceProperties;
|
||||
import org.ruoyi.common.trace.constant.TraceConstants;
|
||||
import org.ruoyi.common.trace.service.TraceRecordService;
|
||||
import org.ruoyi.common.trace.util.TracePayloadUtils;
|
||||
|
||||
import java.util.Date;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
/**
|
||||
* 默认流式 trace 节点实现。
|
||||
*/
|
||||
@Slf4j
|
||||
public class DefaultTraceStreamSpan implements TraceStreamSpan {
|
||||
|
||||
private final TraceRecordService traceRecordService;
|
||||
private final TraceProperties traceProperties;
|
||||
private final String traceId;
|
||||
private final String nodeId;
|
||||
private final long startMillis;
|
||||
private final AtomicBoolean finished = new AtomicBoolean(false);
|
||||
private final AtomicBoolean detached = new AtomicBoolean(false);
|
||||
|
||||
public DefaultTraceStreamSpan(TraceRecordService traceRecordService,
|
||||
TraceProperties traceProperties,
|
||||
String traceId,
|
||||
String nodeId,
|
||||
long startMillis) {
|
||||
this.traceRecordService = traceRecordService;
|
||||
this.traceProperties = traceProperties;
|
||||
this.traceId = traceId;
|
||||
this.nodeId = nodeId;
|
||||
this.startMillis = startMillis;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void detach() {
|
||||
if (detached.compareAndSet(false, true)) {
|
||||
TraceContext.popNode();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void finishSuccess(String outputPayload) {
|
||||
finish(TraceConstants.STATUS_SUCCESS, null, outputPayload);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void finishError(Throwable throwable) {
|
||||
finish(TraceConstants.STATUS_ERROR, TracePayloadUtils.error(throwable, traceProperties), null);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void finishCancelledIfRunning() {
|
||||
finish(TraceConstants.STATUS_CANCELLED, null, null);
|
||||
}
|
||||
|
||||
private void finish(String status, String errorMessage, String outputPayload) {
|
||||
if (!finished.compareAndSet(false, true)) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
traceRecordService.finishNode(traceId, nodeId, status, errorMessage, outputPayload,
|
||||
new Date(), System.currentTimeMillis() - startMillis);
|
||||
} catch (Exception e) {
|
||||
log.warn("结束 trace stream span 失败,traceId={}, nodeId={}", traceId, nodeId, e);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,107 @@
|
||||
package org.ruoyi.common.trace.core;
|
||||
|
||||
import java.util.ArrayDeque;
|
||||
import java.util.Deque;
|
||||
|
||||
/**
|
||||
* 通用链路追踪上下文。
|
||||
*/
|
||||
public final class TraceContext {
|
||||
|
||||
private static final ThreadLocal<String> TRACE_ID = new ThreadLocal<>();
|
||||
private static final ThreadLocal<String> BUSINESS_TYPE = new ThreadLocal<>();
|
||||
private static final ThreadLocal<String> BUSINESS_ID = new ThreadLocal<>();
|
||||
private static final ThreadLocal<Long> USER_ID = new ThreadLocal<>();
|
||||
private static final ThreadLocal<String> TENANT_ID = new ThreadLocal<>();
|
||||
private static final ThreadLocal<Deque<String>> NODE_STACK = new ThreadLocal<>();
|
||||
|
||||
private TraceContext() {
|
||||
}
|
||||
|
||||
public static TraceScope begin(String traceId, String businessType, String businessId, Long userId, String tenantId) {
|
||||
TraceScope scope = new TraceScope(getTraceId(), getBusinessType(), getBusinessId(), getUserId(), getTenantId());
|
||||
TRACE_ID.set(traceId);
|
||||
BUSINESS_TYPE.set(businessType);
|
||||
BUSINESS_ID.set(businessId);
|
||||
USER_ID.set(userId);
|
||||
TENANT_ID.set(tenantId);
|
||||
NODE_STACK.remove();
|
||||
return scope;
|
||||
}
|
||||
|
||||
static void restore(String traceId, String businessType, String businessId, Long userId, String tenantId) {
|
||||
setOrRemove(TRACE_ID, traceId);
|
||||
setOrRemove(BUSINESS_TYPE, businessType);
|
||||
setOrRemove(BUSINESS_ID, businessId);
|
||||
setOrRemove(USER_ID, userId);
|
||||
setOrRemove(TENANT_ID, tenantId);
|
||||
NODE_STACK.remove();
|
||||
}
|
||||
|
||||
private static <T> void setOrRemove(ThreadLocal<T> holder, T value) {
|
||||
if (value == null) {
|
||||
holder.remove();
|
||||
} else {
|
||||
holder.set(value);
|
||||
}
|
||||
}
|
||||
|
||||
public static String getTraceId() {
|
||||
return TRACE_ID.get();
|
||||
}
|
||||
|
||||
public static String getBusinessType() {
|
||||
return BUSINESS_TYPE.get();
|
||||
}
|
||||
|
||||
public static String getBusinessId() {
|
||||
return BUSINESS_ID.get();
|
||||
}
|
||||
|
||||
public static Long getUserId() {
|
||||
return USER_ID.get();
|
||||
}
|
||||
|
||||
public static String getTenantId() {
|
||||
return TENANT_ID.get();
|
||||
}
|
||||
|
||||
public static String currentNodeId() {
|
||||
Deque<String> stack = NODE_STACK.get();
|
||||
return stack == null ? null : stack.peek();
|
||||
}
|
||||
|
||||
public static int depth() {
|
||||
Deque<String> stack = NODE_STACK.get();
|
||||
return stack == null ? 0 : stack.size();
|
||||
}
|
||||
|
||||
public static void pushNode(String nodeId) {
|
||||
Deque<String> stack = NODE_STACK.get();
|
||||
if (stack == null) {
|
||||
stack = new ArrayDeque<>();
|
||||
NODE_STACK.set(stack);
|
||||
}
|
||||
stack.push(nodeId);
|
||||
}
|
||||
|
||||
public static void popNode() {
|
||||
Deque<String> stack = NODE_STACK.get();
|
||||
if (stack == null || stack.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
stack.pop();
|
||||
if (stack.isEmpty()) {
|
||||
NODE_STACK.remove();
|
||||
}
|
||||
}
|
||||
|
||||
public static void clear() {
|
||||
TRACE_ID.remove();
|
||||
BUSINESS_TYPE.remove();
|
||||
BUSINESS_ID.remove();
|
||||
USER_ID.remove();
|
||||
TENANT_ID.remove();
|
||||
NODE_STACK.remove();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,159 @@
|
||||
package org.ruoyi.common.trace.core;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.ruoyi.common.core.utils.StringUtils;
|
||||
import org.ruoyi.common.trace.config.TraceProperties;
|
||||
import org.ruoyi.common.trace.constant.TraceConstants;
|
||||
import org.ruoyi.common.trace.domain.TraceNode;
|
||||
import org.ruoyi.common.trace.service.TraceRecordService;
|
||||
import org.ruoyi.common.trace.util.TracePayloadUtils;
|
||||
|
||||
import java.util.Date;
|
||||
import java.util.UUID;
|
||||
import java.util.function.Function;
|
||||
|
||||
/**
|
||||
* 链路追踪节点模板,封装节点创建、上下文压栈、执行、结束、出栈的标准生命周期。
|
||||
* <p>
|
||||
* 用于同步方法内的 trace 埋点,消除手写 start/finish/pop 的重复代码。
|
||||
* 对于需要异步结束的场景(如流式响应),请使用 {@link DefaultTraceStreamSpan}。
|
||||
*
|
||||
* @see DefaultTraceStreamSpan
|
||||
*/
|
||||
@Slf4j
|
||||
public final class TraceNodeTemplate {
|
||||
|
||||
private TraceNodeTemplate() {
|
||||
}
|
||||
|
||||
/**
|
||||
* 在 trace 节点上下文中执行业务逻辑,成功后使用 outputBuilder 生成输出摘要。
|
||||
*
|
||||
* @param traceRecordService 记录服务
|
||||
* @param traceProperties 配置
|
||||
* @param nodeName 节点名称
|
||||
* @param nodeType 节点类型
|
||||
* @param className 类名
|
||||
* @param methodName 方法名
|
||||
* @param inputPayload 输入摘要
|
||||
* @param action 业务逻辑
|
||||
* @param successOutput 成功时从结果构建输出摘要
|
||||
* @param <T> 业务返回值类型
|
||||
* @return 业务执行结果
|
||||
*/
|
||||
public static <T> T withNode(
|
||||
TraceRecordService traceRecordService,
|
||||
TraceProperties traceProperties,
|
||||
String nodeName,
|
||||
String nodeType,
|
||||
String className,
|
||||
String methodName,
|
||||
String inputPayload,
|
||||
NodeAction<T> action,
|
||||
Function<T, String> successOutput) {
|
||||
|
||||
if (!traceProperties.isEnabled() || StringUtils.isBlank(TraceContext.getTraceId())) {
|
||||
return unwrap(action);
|
||||
}
|
||||
|
||||
String traceId = TraceContext.getTraceId();
|
||||
String nodeId = UUID.randomUUID().toString().replace("-", "");
|
||||
long startMillis = System.currentTimeMillis();
|
||||
|
||||
TraceNode node = buildNode(traceId, nodeId, nodeName, nodeType,
|
||||
className, methodName, inputPayload, startMillis);
|
||||
|
||||
try {
|
||||
traceRecordService.startNode(node);
|
||||
} catch (Exception e) {
|
||||
log.warn("写入 trace 节点失败,traceId={}, nodeId={}", traceId, nodeId, e);
|
||||
return unwrap(action);
|
||||
}
|
||||
|
||||
TraceContext.pushNode(nodeId);
|
||||
try {
|
||||
T result = action.execute();
|
||||
String output = successOutput != null && result != null ? successOutput.apply(result) : null;
|
||||
finishNode(traceRecordService, traceProperties, traceId, nodeId,
|
||||
TraceConstants.STATUS_SUCCESS, null, output, startMillis);
|
||||
return result;
|
||||
} catch (Throwable ex) {
|
||||
finishNode(traceRecordService, traceProperties, traceId, nodeId,
|
||||
TraceConstants.STATUS_ERROR, ex, null, startMillis);
|
||||
throw rethrow(ex);
|
||||
} finally {
|
||||
TraceContext.popNode();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 在 trace 节点上下文中执行业务逻辑(无输出摘要)。
|
||||
*/
|
||||
public static <T> T withNode(
|
||||
TraceRecordService traceRecordService,
|
||||
TraceProperties traceProperties,
|
||||
String nodeName,
|
||||
String nodeType,
|
||||
String className,
|
||||
String methodName,
|
||||
String inputPayload,
|
||||
NodeAction<T> action) {
|
||||
return withNode(traceRecordService, traceProperties, nodeName, nodeType,
|
||||
className, methodName, inputPayload, action, null);
|
||||
}
|
||||
|
||||
// ======================== 内部工具方法 ========================
|
||||
|
||||
private static TraceNode buildNode(String traceId, String nodeId, String nodeName,
|
||||
String nodeType, String className, String methodName,
|
||||
String inputPayload, long startMillis) {
|
||||
TraceNode node = new TraceNode();
|
||||
node.setTraceId(traceId);
|
||||
node.setNodeId(nodeId);
|
||||
node.setParentNodeId(TraceContext.currentNodeId());
|
||||
node.setDepth(TraceContext.depth());
|
||||
node.setNodeName(nodeName);
|
||||
node.setNodeType(nodeType);
|
||||
node.setClassName(className);
|
||||
node.setMethodName(methodName);
|
||||
node.setStatus(TraceConstants.STATUS_RUNNING);
|
||||
node.setStartTime(new Date(startMillis));
|
||||
node.setInputPayload(inputPayload);
|
||||
return node;
|
||||
}
|
||||
|
||||
private static void finishNode(TraceRecordService service, TraceProperties props,
|
||||
String traceId, String nodeId, String status,
|
||||
Throwable error, String outputPayload, long startMillis) {
|
||||
try {
|
||||
service.finishNode(traceId, nodeId, status,
|
||||
TracePayloadUtils.error(error, props), outputPayload,
|
||||
new Date(), System.currentTimeMillis() - startMillis);
|
||||
} catch (Exception e) {
|
||||
log.warn("结束 trace 节点失败,traceId={}, nodeId={}", traceId, nodeId, e);
|
||||
}
|
||||
}
|
||||
|
||||
private static <T> T unwrap(NodeAction<T> action) {
|
||||
try {
|
||||
return action.execute();
|
||||
} catch (RuntimeException | Error e) {
|
||||
throw e;
|
||||
} catch (Throwable e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private static <T extends Throwable> T rethrow(Throwable t) throws T {
|
||||
throw (T) t;
|
||||
}
|
||||
|
||||
/**
|
||||
* 可抛出 Throwable 的业务动作。
|
||||
*/
|
||||
@FunctionalInterface
|
||||
public interface NodeAction<T> {
|
||||
T execute() throws Throwable;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,30 @@
|
||||
package org.ruoyi.common.trace.core;
|
||||
|
||||
/**
|
||||
* Trace 上下文作用域。
|
||||
*/
|
||||
public final class TraceScope implements AutoCloseable {
|
||||
|
||||
private final String previousTraceId;
|
||||
private final String previousBusinessType;
|
||||
private final String previousBusinessId;
|
||||
private final Long previousUserId;
|
||||
private final String previousTenantId;
|
||||
|
||||
TraceScope(String previousTraceId,
|
||||
String previousBusinessType,
|
||||
String previousBusinessId,
|
||||
Long previousUserId,
|
||||
String previousTenantId) {
|
||||
this.previousTraceId = previousTraceId;
|
||||
this.previousBusinessType = previousBusinessType;
|
||||
this.previousBusinessId = previousBusinessId;
|
||||
this.previousUserId = previousUserId;
|
||||
this.previousTenantId = previousTenantId;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
TraceContext.restore(previousTraceId, previousBusinessType, previousBusinessId, previousUserId, previousTenantId);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,22 @@
|
||||
package org.ruoyi.common.trace.core;
|
||||
|
||||
/**
|
||||
* 可跨回调结束的流式 trace 节点。
|
||||
*/
|
||||
public interface TraceStreamSpan {
|
||||
|
||||
void detach();
|
||||
|
||||
/**
|
||||
* 结束成功的流式节点,并可写入输出摘要。
|
||||
*/
|
||||
default void finishSuccess() {
|
||||
finishSuccess(null);
|
||||
}
|
||||
|
||||
void finishSuccess(String outputPayload);
|
||||
|
||||
void finishError(Throwable throwable);
|
||||
|
||||
void finishCancelledIfRunning();
|
||||
}
|
||||
@@ -0,0 +1,45 @@
|
||||
package org.ruoyi.common.trace.domain;
|
||||
|
||||
import com.baomidou.mybatisplus.annotation.TableId;
|
||||
import com.baomidou.mybatisplus.annotation.TableLogic;
|
||||
import com.baomidou.mybatisplus.annotation.TableName;
|
||||
import lombok.Data;
|
||||
import lombok.EqualsAndHashCode;
|
||||
import org.ruoyi.common.mybatis.core.domain.BaseEntity;
|
||||
|
||||
import java.io.Serial;
|
||||
import java.util.Date;
|
||||
|
||||
/**
|
||||
* 链路追踪节点记录 trace_node。
|
||||
*/
|
||||
@Data
|
||||
@EqualsAndHashCode(callSuper = true)
|
||||
@TableName("trace_node")
|
||||
public class TraceNode extends BaseEntity {
|
||||
|
||||
@Serial
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
@TableId(value = "id")
|
||||
private Long id;
|
||||
private String traceId;
|
||||
private String nodeId;
|
||||
private String parentNodeId;
|
||||
private String nodeName;
|
||||
private String nodeType;
|
||||
private Integer depth;
|
||||
private Integer sortOrder;
|
||||
private String className;
|
||||
private String methodName;
|
||||
private String status;
|
||||
private Date startTime;
|
||||
private Date endTime;
|
||||
private Long durationMs;
|
||||
private String errorMessage;
|
||||
private String inputPayload;
|
||||
private String outputPayload;
|
||||
private String metadata;
|
||||
@TableLogic
|
||||
private String delFlag;
|
||||
}
|
||||
@@ -0,0 +1,40 @@
|
||||
package org.ruoyi.common.trace.domain;
|
||||
|
||||
import com.baomidou.mybatisplus.annotation.TableId;
|
||||
import com.baomidou.mybatisplus.annotation.TableLogic;
|
||||
import com.baomidou.mybatisplus.annotation.TableName;
|
||||
import lombok.Data;
|
||||
import lombok.EqualsAndHashCode;
|
||||
import org.ruoyi.common.mybatis.core.domain.BaseEntity;
|
||||
|
||||
import java.io.Serial;
|
||||
import java.util.Date;
|
||||
|
||||
/**
|
||||
* 链路追踪运行记录 trace_run。
|
||||
*/
|
||||
@Data
|
||||
@EqualsAndHashCode(callSuper = true)
|
||||
@TableName("trace_run")
|
||||
public class TraceRun extends BaseEntity {
|
||||
|
||||
@Serial
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
@TableId(value = "id")
|
||||
private Long id;
|
||||
private String traceId;
|
||||
private String traceName;
|
||||
private String businessType;
|
||||
private String businessId;
|
||||
private Long userId;
|
||||
private String tenantId;
|
||||
private String status;
|
||||
private Date startTime;
|
||||
private Date endTime;
|
||||
private Long durationMs;
|
||||
private String errorMessage;
|
||||
private String metadata;
|
||||
@TableLogic
|
||||
private String delFlag;
|
||||
}
|
||||
@@ -0,0 +1,32 @@
|
||||
package org.ruoyi.common.trace.domain.bo;
|
||||
|
||||
import io.github.linpeilie.annotations.AutoMapper;
|
||||
import lombok.Data;
|
||||
import lombok.EqualsAndHashCode;
|
||||
import org.ruoyi.common.mybatis.core.domain.BaseEntity;
|
||||
import org.ruoyi.common.trace.domain.TraceRun;
|
||||
|
||||
import java.util.Date;
|
||||
|
||||
/**
|
||||
* 链路追踪运行记录查询对象。
|
||||
*/
|
||||
@Data
|
||||
@EqualsAndHashCode(callSuper = true)
|
||||
@AutoMapper(target = TraceRun.class, reverseConvertGenerate = false)
|
||||
public class TraceRunBo extends BaseEntity {
|
||||
|
||||
private Long id;
|
||||
private String traceId;
|
||||
private String traceName;
|
||||
private String businessType;
|
||||
private String businessId;
|
||||
private Long userId;
|
||||
private String tenantId;
|
||||
private String status;
|
||||
private Date startTime;
|
||||
private Date endTime;
|
||||
private Long durationMs;
|
||||
private String errorMessage;
|
||||
private String metadata;
|
||||
}
|
||||
@@ -0,0 +1,83 @@
|
||||
package org.ruoyi.common.trace.domain.vo;
|
||||
|
||||
import lombok.Data;
|
||||
|
||||
import java.io.Serial;
|
||||
import java.io.Serializable;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* 链路追踪详情视图对象。
|
||||
* <p>
|
||||
* 包含运行信息、节点树以及统计摘要。
|
||||
*/
|
||||
@Data
|
||||
public class TraceDetailVo implements Serializable {
|
||||
|
||||
@Serial
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
private TraceRunVo run;
|
||||
private List<TraceNodeVo> nodes;
|
||||
private TraceStatistics statistics;
|
||||
|
||||
/**
|
||||
* 链路追踪统计摘要,帮助快速了解整体执行情况。
|
||||
*/
|
||||
@Data
|
||||
public static class TraceStatistics implements Serializable {
|
||||
|
||||
@Serial
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
/** 总节点数 */
|
||||
private int totalNodes;
|
||||
|
||||
/** 成功节点数 */
|
||||
private int successCount;
|
||||
|
||||
/** 失败节点数 */
|
||||
private int failedCount;
|
||||
|
||||
/** 运行中节点数 */
|
||||
private int runningCount;
|
||||
|
||||
/** 最大调用深度 */
|
||||
private int maxDepth;
|
||||
|
||||
/** 平均耗时 (ms) */
|
||||
private long avgDurationMs;
|
||||
|
||||
/** 总链路耗时 (ms) */
|
||||
private long totalDurationMs;
|
||||
|
||||
/** 慢节点 Top N */
|
||||
private List<SlowNodeInfo> topSlowNodes = new ArrayList<>();
|
||||
}
|
||||
|
||||
/**
|
||||
* 慢节点简要信息。
|
||||
*/
|
||||
@Data
|
||||
public static class SlowNodeInfo implements Serializable {
|
||||
|
||||
@Serial
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
/** 节点 ID */
|
||||
private String nodeId;
|
||||
|
||||
/** 节点展示名称 */
|
||||
private String nodeDisplayName;
|
||||
|
||||
/** 节点类型中文标签 */
|
||||
private String nodeTypeLabel;
|
||||
|
||||
/** 耗时 (ms) */
|
||||
private long durationMs;
|
||||
|
||||
/** 占总耗时百分比 */
|
||||
private double percentOfTotal;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,77 @@
|
||||
package org.ruoyi.common.trace.domain.vo;
|
||||
|
||||
import com.fasterxml.jackson.annotation.JsonProperty;
|
||||
import io.github.linpeilie.annotations.AutoMapper;
|
||||
import lombok.Data;
|
||||
import org.ruoyi.common.trace.domain.TraceNode;
|
||||
|
||||
import java.io.Serial;
|
||||
import java.io.Serializable;
|
||||
import java.util.Date;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* 链路追踪节点记录视图对象。
|
||||
* <p>
|
||||
* 除实体映射字段外,还提供前端可直接展示的显示标签和解析后的 payload 对象。
|
||||
*/
|
||||
@Data
|
||||
@AutoMapper(target = TraceNode.class)
|
||||
public class TraceNodeVo implements Serializable {
|
||||
|
||||
@Serial
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
private Long id;
|
||||
private String traceId;
|
||||
private String nodeId;
|
||||
private String parentNodeId;
|
||||
private String nodeName;
|
||||
private String nodeType;
|
||||
private Integer depth;
|
||||
private Integer sortOrder;
|
||||
private String className;
|
||||
private String methodName;
|
||||
private String status;
|
||||
private Date startTime;
|
||||
private Date endTime;
|
||||
private Long durationMs;
|
||||
private String errorMessage;
|
||||
|
||||
/** 原始 input payload 字符串,parsedInput 解析失败时回退使用 */
|
||||
private String inputPayload;
|
||||
|
||||
/** 原始 output payload 字符串,parsedOutput 解析失败时回退使用 */
|
||||
private String outputPayload;
|
||||
|
||||
/** 原始 metadata 字符串,parsedMetadata 解析失败时回退使用 */
|
||||
private String metadata;
|
||||
|
||||
// ======================== 展示用计算字段 ========================
|
||||
|
||||
/** 节点类型中文标签,如 "知识检索"、"LLM 调用" */
|
||||
@JsonProperty("nodeTypeLabel")
|
||||
private String nodeTypeLabel;
|
||||
|
||||
/** 状态中文标签,如 "成功"、"失败"、"运行中" */
|
||||
@JsonProperty("statusLabel")
|
||||
private String statusLabel;
|
||||
|
||||
/** 节点展示名称(中文友好),从 nodeName 转换 */
|
||||
@JsonProperty("nodeDisplayName")
|
||||
private String nodeDisplayName;
|
||||
|
||||
// ======================== 解析后的 payload ========================
|
||||
|
||||
/** input payload 解析为 Map,前端可直接读取结构化字段 */
|
||||
@JsonProperty("parsedInput")
|
||||
private Map<String, Object> parsedInput;
|
||||
|
||||
/** output payload 解析为 Map,前端可直接读取结构化字段 */
|
||||
@JsonProperty("parsedOutput")
|
||||
private Map<String, Object> parsedOutput;
|
||||
|
||||
/** metadata 解析为 Map */
|
||||
@JsonProperty("parsedMetadata")
|
||||
private Map<String, Object> parsedMetadata;
|
||||
}
|
||||
@@ -0,0 +1,54 @@
|
||||
package org.ruoyi.common.trace.domain.vo;
|
||||
|
||||
import com.fasterxml.jackson.annotation.JsonProperty;
|
||||
import io.github.linpeilie.annotations.AutoMapper;
|
||||
import lombok.Data;
|
||||
import org.ruoyi.common.trace.domain.TraceRun;
|
||||
|
||||
import java.io.Serial;
|
||||
import java.io.Serializable;
|
||||
import java.util.Date;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* 链路追踪运行记录视图对象。
|
||||
* <p>
|
||||
* 除实体映射字段外,还提供前端可直接展示的显示标签和解析后的 metadata 对象。
|
||||
*/
|
||||
@Data
|
||||
@AutoMapper(target = TraceRun.class)
|
||||
public class TraceRunVo implements Serializable {
|
||||
|
||||
@Serial
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
private Long id;
|
||||
private String traceId;
|
||||
private String traceName;
|
||||
private String businessType;
|
||||
private String businessId;
|
||||
private Long userId;
|
||||
private String tenantId;
|
||||
private String status;
|
||||
private Date startTime;
|
||||
private Date endTime;
|
||||
private Long durationMs;
|
||||
private String errorMessage;
|
||||
|
||||
/** 原始 metadata 字符串(兼容旧版),新代码请使用 parsedMetadata */
|
||||
private String metadata;
|
||||
|
||||
// ======================== 展示用计算字段 ========================
|
||||
|
||||
/** 状态中文标签,如 "成功"、"失败"、"运行中" */
|
||||
@JsonProperty("statusLabel")
|
||||
private String statusLabel;
|
||||
|
||||
/** 业务类型中文标签,如 "RAG 对话" */
|
||||
@JsonProperty("businessTypeLabel")
|
||||
private String businessTypeLabel;
|
||||
|
||||
/** metadata 解析为 Map */
|
||||
@JsonProperty("parsedMetadata")
|
||||
private Map<String, Object> parsedMetadata;
|
||||
}
|
||||
@@ -0,0 +1,11 @@
|
||||
package org.ruoyi.common.trace.mapper;
|
||||
|
||||
import org.ruoyi.common.mybatis.core.mapper.BaseMapperPlus;
|
||||
import org.ruoyi.common.trace.domain.TraceNode;
|
||||
import org.ruoyi.common.trace.domain.vo.TraceNodeVo;
|
||||
|
||||
/**
|
||||
* 链路追踪节点记录 Mapper。
|
||||
*/
|
||||
public interface TraceNodeMapper extends BaseMapperPlus<TraceNode, TraceNodeVo> {
|
||||
}
|
||||
@@ -0,0 +1,11 @@
|
||||
package org.ruoyi.common.trace.mapper;
|
||||
|
||||
import org.ruoyi.common.mybatis.core.mapper.BaseMapperPlus;
|
||||
import org.ruoyi.common.trace.domain.TraceRun;
|
||||
import org.ruoyi.common.trace.domain.vo.TraceRunVo;
|
||||
|
||||
/**
|
||||
* 链路追踪运行记录 Mapper。
|
||||
*/
|
||||
public interface TraceRunMapper extends BaseMapperPlus<TraceRun, TraceRunVo> {
|
||||
}
|
||||
@@ -0,0 +1,35 @@
|
||||
package org.ruoyi.common.trace.service;
|
||||
|
||||
import org.ruoyi.common.mybatis.core.page.PageQuery;
|
||||
import org.ruoyi.common.mybatis.core.page.TableDataInfo;
|
||||
import org.ruoyi.common.trace.domain.TraceNode;
|
||||
import org.ruoyi.common.trace.domain.TraceRun;
|
||||
import org.ruoyi.common.trace.domain.bo.TraceRunBo;
|
||||
import org.ruoyi.common.trace.domain.vo.TraceDetailVo;
|
||||
import org.ruoyi.common.trace.domain.vo.TraceNodeVo;
|
||||
import org.ruoyi.common.trace.domain.vo.TraceRunVo;
|
||||
|
||||
import java.util.Date;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* 链路追踪记录服务。
|
||||
*/
|
||||
public interface TraceRecordService {
|
||||
|
||||
void startRun(TraceRun run);
|
||||
|
||||
void finishRun(String traceId, String status, String errorMessage, Date endTime, long durationMs);
|
||||
|
||||
void startNode(TraceNode node);
|
||||
|
||||
void finishNode(String traceId, String nodeId, String status, String errorMessage, String outputPayload, Date endTime, long durationMs);
|
||||
|
||||
TableDataInfo<TraceRunVo> pageRuns(TraceRunBo bo, PageQuery pageQuery);
|
||||
|
||||
TraceRunVo getRun(String traceId);
|
||||
|
||||
List<TraceNodeVo> listNodes(String traceId);
|
||||
|
||||
TraceDetailVo getDetail(String traceId);
|
||||
}
|
||||
@@ -0,0 +1,253 @@
|
||||
package org.ruoyi.common.trace.service.impl;
|
||||
|
||||
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||||
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
|
||||
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.ruoyi.common.core.utils.StringUtils;
|
||||
import org.ruoyi.common.mybatis.core.page.PageQuery;
|
||||
import org.ruoyi.common.mybatis.core.page.TableDataInfo;
|
||||
import org.ruoyi.common.trace.constant.TraceDisplayConstants;
|
||||
import org.ruoyi.common.trace.domain.TraceNode;
|
||||
import org.ruoyi.common.trace.domain.TraceRun;
|
||||
import org.ruoyi.common.trace.domain.bo.TraceRunBo;
|
||||
import org.ruoyi.common.trace.domain.vo.TraceDetailVo;
|
||||
import org.ruoyi.common.trace.domain.vo.TraceNodeVo;
|
||||
import org.ruoyi.common.trace.domain.vo.TraceRunVo;
|
||||
import org.ruoyi.common.trace.mapper.TraceNodeMapper;
|
||||
import org.ruoyi.common.trace.mapper.TraceRunMapper;
|
||||
import org.ruoyi.common.trace.service.TraceRecordService;
|
||||
import org.ruoyi.common.trace.util.TracePayloadUtils;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Comparator;
|
||||
import java.util.Date;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
/**
|
||||
* 链路追踪记录服务实现。
|
||||
* <p>
|
||||
* 在自动映射的基础上,对 VO 进行二次加工:补充展示标签、解析 payload 为结构化对象、计算统计信息。
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
public class TraceRecordServiceImpl implements TraceRecordService {
|
||||
|
||||
private static final int TOP_SLOW_NODES_LIMIT = 5;
|
||||
|
||||
private final TraceRunMapper traceRunMapper;
|
||||
private final TraceNodeMapper traceNodeMapper;
|
||||
|
||||
@Override
|
||||
public void startRun(TraceRun run) {
|
||||
try {
|
||||
traceRunMapper.insert(run);
|
||||
} catch (Exception e) {
|
||||
log.warn("写入 trace run 失败,traceId={}", run == null ? null : run.getTraceId(), e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void finishRun(String traceId, String status, String errorMessage, Date endTime, long durationMs) {
|
||||
try {
|
||||
TraceRun update = new TraceRun();
|
||||
update.setStatus(status);
|
||||
update.setErrorMessage(errorMessage);
|
||||
update.setEndTime(endTime);
|
||||
update.setDurationMs(durationMs);
|
||||
traceRunMapper.update(update, Wrappers.lambdaUpdate(TraceRun.class).eq(TraceRun::getTraceId, traceId));
|
||||
} catch (Exception e) {
|
||||
log.warn("更新 trace run 失败,traceId={}", traceId, e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void startNode(TraceNode node) {
|
||||
try {
|
||||
traceNodeMapper.insert(node);
|
||||
} catch (Exception e) {
|
||||
log.warn("写入 trace node 失败,traceId={}, nodeId={}",
|
||||
node == null ? null : node.getTraceId(), node == null ? null : node.getNodeId(), e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void finishNode(String traceId, String nodeId, String status, String errorMessage, String outputPayload, Date endTime, long durationMs) {
|
||||
try {
|
||||
TraceNode update = new TraceNode();
|
||||
update.setStatus(status);
|
||||
update.setErrorMessage(errorMessage);
|
||||
update.setOutputPayload(outputPayload);
|
||||
update.setEndTime(endTime);
|
||||
update.setDurationMs(durationMs);
|
||||
traceNodeMapper.update(update, Wrappers.lambdaUpdate(TraceNode.class)
|
||||
.eq(TraceNode::getTraceId, traceId)
|
||||
.eq(TraceNode::getNodeId, nodeId));
|
||||
} catch (Exception e) {
|
||||
log.warn("更新 trace node 失败,traceId={}, nodeId={}", traceId, nodeId, e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public TableDataInfo<TraceRunVo> pageRuns(TraceRunBo bo, PageQuery pageQuery) {
|
||||
LambdaQueryWrapper<TraceRun> wrapper = buildRunWrapper(bo == null ? new TraceRunBo() : bo);
|
||||
if (StringUtils.isBlank(pageQuery.getOrderByColumn())) {
|
||||
wrapper.orderByDesc(TraceRun::getStartTime);
|
||||
}
|
||||
Page<TraceRunVo> page = traceRunMapper.selectVoPage(pageQuery.build(), wrapper);
|
||||
if (page.getRecords() != null) {
|
||||
page.getRecords().forEach(this::enrichRunVo);
|
||||
}
|
||||
return TableDataInfo.build(page);
|
||||
}
|
||||
|
||||
@Override
|
||||
public TraceRunVo getRun(String traceId) {
|
||||
TraceRunVo run = traceRunMapper.selectVoOne(Wrappers.lambdaQuery(TraceRun.class).eq(TraceRun::getTraceId, traceId));
|
||||
if (run != null) {
|
||||
enrichRunVo(run);
|
||||
}
|
||||
return run;
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<TraceNodeVo> listNodes(String traceId) {
|
||||
List<TraceNodeVo> nodes = traceNodeMapper.selectVoList(Wrappers.lambdaQuery(TraceNode.class)
|
||||
.eq(TraceNode::getTraceId, traceId)
|
||||
.orderByAsc(TraceNode::getStartTime)
|
||||
.orderByAsc(TraceNode::getId));
|
||||
if (nodes != null) {
|
||||
nodes.forEach(this::enrichNodeVo);
|
||||
}
|
||||
return nodes;
|
||||
}
|
||||
|
||||
@Override
|
||||
public TraceDetailVo getDetail(String traceId) {
|
||||
TraceDetailVo detail = new TraceDetailVo();
|
||||
TraceRunVo run = getRun(traceId);
|
||||
detail.setRun(run);
|
||||
|
||||
List<TraceNodeVo> nodes = listNodes(traceId);
|
||||
// 返回扁平列表,前端自行按 parentNodeId 建树
|
||||
detail.setNodes(nodes != null ? nodes : new ArrayList<>());
|
||||
|
||||
// 计算统计信息
|
||||
if (nodes != null) {
|
||||
detail.setStatistics(buildStatistics(nodes, run));
|
||||
}
|
||||
return detail;
|
||||
}
|
||||
|
||||
// ======================== VO 加工 ========================
|
||||
|
||||
/**
|
||||
* 为 TraceRunVo 补充展示标签和解析后的 metadata。
|
||||
*/
|
||||
private void enrichRunVo(TraceRunVo vo) {
|
||||
if (vo == null) {
|
||||
return;
|
||||
}
|
||||
vo.setStatusLabel(TraceDisplayConstants.statusLabel(vo.getStatus()));
|
||||
vo.setBusinessTypeLabel(TraceDisplayConstants.businessTypeLabel(vo.getBusinessType()));
|
||||
vo.setParsedMetadata(TracePayloadUtils.parseJsonToMap(vo.getMetadata()));
|
||||
}
|
||||
|
||||
/**
|
||||
* 为 TraceNodeVo 补充展示标签和解析后的 payload。
|
||||
*/
|
||||
private void enrichNodeVo(TraceNodeVo vo) {
|
||||
if (vo == null) {
|
||||
return;
|
||||
}
|
||||
vo.setNodeTypeLabel(TraceDisplayConstants.nodeTypeLabel(vo.getNodeType()));
|
||||
vo.setStatusLabel(TraceDisplayConstants.statusLabel(vo.getStatus()));
|
||||
vo.setNodeDisplayName(TraceDisplayConstants.prettifyNodeName(vo.getNodeName()));
|
||||
vo.setParsedInput(TracePayloadUtils.parseJsonToMap(vo.getInputPayload()));
|
||||
vo.setParsedOutput(TracePayloadUtils.parseJsonToMap(vo.getOutputPayload()));
|
||||
vo.setParsedMetadata(TracePayloadUtils.parseJsonToMap(vo.getMetadata()));
|
||||
}
|
||||
|
||||
// ======================== 统计计算 ========================
|
||||
private TraceDetailVo.TraceStatistics buildStatistics(List<TraceNodeVo> nodes, TraceRunVo run) {
|
||||
TraceDetailVo.TraceStatistics stats = new TraceDetailVo.TraceStatistics();
|
||||
|
||||
int total = nodes == null ? 0 : nodes.size();
|
||||
int success = 0;
|
||||
int failed = 0;
|
||||
int running = 0;
|
||||
int maxDepth = 0;
|
||||
long totalDuration = 0;
|
||||
|
||||
for (TraceNodeVo node : nodes) {
|
||||
if (node == null) {
|
||||
continue;
|
||||
}
|
||||
if (TraceDisplayConstants.isSuccess(node.getStatus())) {
|
||||
success++;
|
||||
} else if (TraceDisplayConstants.isFailed(node.getStatus())) {
|
||||
failed++;
|
||||
} else if (TraceDisplayConstants.isRunning(node.getStatus())) {
|
||||
running++;
|
||||
}
|
||||
|
||||
int depth = node.getDepth() == null ? 0 : node.getDepth();
|
||||
if (depth > maxDepth) {
|
||||
maxDepth = depth;
|
||||
}
|
||||
|
||||
long dur = node.getDurationMs() == null ? 0 : node.getDurationMs();
|
||||
totalDuration += dur;
|
||||
}
|
||||
|
||||
stats.setTotalNodes(total);
|
||||
stats.setSuccessCount(success);
|
||||
stats.setFailedCount(failed);
|
||||
stats.setRunningCount(running);
|
||||
stats.setMaxDepth(maxDepth);
|
||||
stats.setAvgDurationMs(total > 0 ? totalDuration / total : 0);
|
||||
stats.setTotalDurationMs(run != null && run.getDurationMs() != null ? run.getDurationMs() : totalDuration);
|
||||
|
||||
// 慢节点 Top N
|
||||
long totalMs = stats.getTotalDurationMs();
|
||||
List<TraceDetailVo.SlowNodeInfo> topSlow = nodes.stream()
|
||||
.filter(n -> n != null && n.getDurationMs() != null && n.getDurationMs() > 0)
|
||||
.sorted(Comparator.comparingLong(TraceNodeVo::getDurationMs).reversed())
|
||||
.limit(TOP_SLOW_NODES_LIMIT)
|
||||
.map(n -> {
|
||||
TraceDetailVo.SlowNodeInfo info = new TraceDetailVo.SlowNodeInfo();
|
||||
info.setNodeId(n.getNodeId());
|
||||
info.setNodeDisplayName(n.getNodeDisplayName() != null ? n.getNodeDisplayName() : n.getNodeName());
|
||||
info.setNodeTypeLabel(n.getNodeTypeLabel());
|
||||
info.setDurationMs(n.getDurationMs());
|
||||
info.setPercentOfTotal(totalMs > 0 ? Math.round(n.getDurationMs() * 1000.0 / totalMs) / 10.0 : 0);
|
||||
return info;
|
||||
})
|
||||
.collect(Collectors.toList());
|
||||
stats.setTopSlowNodes(topSlow);
|
||||
|
||||
return stats;
|
||||
}
|
||||
|
||||
// ======================== 查询辅助 ========================
|
||||
|
||||
private LambdaQueryWrapper<TraceRun> buildRunWrapper(TraceRunBo bo) {
|
||||
Map<String, Object> params = bo.getParams();
|
||||
return new LambdaQueryWrapper<TraceRun>()
|
||||
.eq(StringUtils.isNotBlank(bo.getTraceId()), TraceRun::getTraceId, bo.getTraceId())
|
||||
.like(StringUtils.isNotBlank(bo.getTraceName()), TraceRun::getTraceName, bo.getTraceName())
|
||||
.eq(StringUtils.isNotBlank(bo.getBusinessType()), TraceRun::getBusinessType, bo.getBusinessType())
|
||||
.eq(StringUtils.isNotBlank(bo.getBusinessId()), TraceRun::getBusinessId, bo.getBusinessId())
|
||||
.eq(StringUtils.isNotBlank(bo.getStatus()), TraceRun::getStatus, bo.getStatus())
|
||||
.eq(bo.getUserId() != null, TraceRun::getUserId, bo.getUserId())
|
||||
.eq(StringUtils.isNotBlank(bo.getTenantId()), TraceRun::getTenantId, bo.getTenantId())
|
||||
.between(params.get("beginTime") != null && params.get("endTime") != null,
|
||||
TraceRun::getStartTime, params.get("beginTime"), params.get("endTime"));
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,82 @@
|
||||
package org.ruoyi.common.trace.util;
|
||||
|
||||
import com.fasterxml.jackson.core.JsonProcessingException;
|
||||
import com.fasterxml.jackson.core.type.TypeReference;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.fasterxml.jackson.databind.module.SimpleModule;
|
||||
import com.fasterxml.jackson.databind.ser.std.ToStringSerializer;
|
||||
import org.ruoyi.common.json.handler.BigNumberSerializer;
|
||||
import org.ruoyi.common.trace.config.TraceProperties;
|
||||
|
||||
import java.math.BigDecimal;
|
||||
import java.math.BigInteger;
|
||||
import java.util.Collections;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* Trace payload 序列化、截断与解析工具。
|
||||
*/
|
||||
public final class TracePayloadUtils {
|
||||
|
||||
private static final ObjectMapper OBJECT_MAPPER;
|
||||
|
||||
static {
|
||||
OBJECT_MAPPER = new ObjectMapper();
|
||||
SimpleModule jsSafeModule = new SimpleModule("js-safe-numbers");
|
||||
jsSafeModule.addSerializer(Long.class, BigNumberSerializer.INSTANCE);
|
||||
jsSafeModule.addSerializer(Long.TYPE, BigNumberSerializer.INSTANCE);
|
||||
jsSafeModule.addSerializer(BigInteger.class, BigNumberSerializer.INSTANCE);
|
||||
jsSafeModule.addSerializer(BigDecimal.class, ToStringSerializer.instance);
|
||||
OBJECT_MAPPER.registerModule(jsSafeModule);
|
||||
}
|
||||
|
||||
private TracePayloadUtils() {
|
||||
}
|
||||
|
||||
public static String error(Throwable throwable, TraceProperties properties) {
|
||||
if (throwable == null) {
|
||||
return null;
|
||||
}
|
||||
int maxLength = properties == null ? 1000 : properties.getPayload().getMaxErrorLength();
|
||||
String message = throwable.getClass().getSimpleName() + ": " + (throwable.getMessage() == null ? "" : throwable.getMessage());
|
||||
return truncate(message, maxLength);
|
||||
}
|
||||
|
||||
public static String toJson(Object value) {
|
||||
if (value == null) {
|
||||
return null;
|
||||
}
|
||||
try {
|
||||
return OBJECT_MAPPER.writeValueAsString(value);
|
||||
} catch (JsonProcessingException e) {
|
||||
return "{\"summary\":\"payload serialization failed\"}";
|
||||
}
|
||||
}
|
||||
|
||||
public static String truncate(String value, int maxLength) {
|
||||
if (value == null) {
|
||||
return null;
|
||||
}
|
||||
if (maxLength <= 0 || value.length() <= maxLength) {
|
||||
return value;
|
||||
}
|
||||
return value.substring(0, maxLength);
|
||||
}
|
||||
|
||||
/**
|
||||
* 将 JSON 字符串解析为 Map,用于前端展示结构化数据。
|
||||
* 解析失败时返回空 Map,不影响主流程。
|
||||
*/
|
||||
@SuppressWarnings("unchecked")
|
||||
public static Map<String, Object> parseJsonToMap(String json) {
|
||||
if (json == null || json.isBlank()) {
|
||||
return Collections.emptyMap();
|
||||
}
|
||||
try {
|
||||
return OBJECT_MAPPER.readValue(json, new TypeReference<Map<String, Object>>() {});
|
||||
} catch (JsonProcessingException e) {
|
||||
// 非 JSON 字符串,返回包含原始文本的 Map
|
||||
return Collections.singletonMap("_raw", json);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
org.ruoyi.common.trace.config.TraceAutoConfiguration
|
||||
@@ -0,0 +1,83 @@
|
||||
package org.ruoyi.common.trace.core;
|
||||
|
||||
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.config.TraceProperties;
|
||||
import org.ruoyi.common.trace.constant.TraceConstants;
|
||||
import org.ruoyi.common.trace.service.TraceRecordService;
|
||||
|
||||
import java.util.Date;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertNull;
|
||||
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.times;
|
||||
import static org.mockito.Mockito.verify;
|
||||
|
||||
@ExtendWith(MockitoExtension.class)
|
||||
class DefaultTraceStreamSpanTest {
|
||||
|
||||
@Mock
|
||||
private TraceRecordService traceRecordService;
|
||||
|
||||
@AfterEach
|
||||
void tearDown() {
|
||||
TraceContext.clear();
|
||||
}
|
||||
|
||||
@Test
|
||||
void finishShouldBeIdempotent() {
|
||||
DefaultTraceStreamSpan span = new DefaultTraceStreamSpan(traceRecordService, new TraceProperties(), "trace-1", "node-1", System.currentTimeMillis());
|
||||
|
||||
span.finishSuccess();
|
||||
span.finishError(new IllegalStateException("ignored"));
|
||||
|
||||
verify(traceRecordService, times(1)).finishNode(eq("trace-1"), eq("node-1"), eq(TraceConstants.STATUS_SUCCESS),
|
||||
isNull(), isNull(), any(Date.class), anyLong());
|
||||
}
|
||||
|
||||
@Test
|
||||
void finishErrorShouldRecordTruncatedError() {
|
||||
TraceProperties properties = new TraceProperties();
|
||||
properties.getPayload().setMaxErrorLength(8);
|
||||
DefaultTraceStreamSpan span = new DefaultTraceStreamSpan(traceRecordService, properties, "trace-1", "node-1", System.currentTimeMillis());
|
||||
|
||||
span.finishError(new IllegalStateException("abcdef"));
|
||||
|
||||
ArgumentCaptor<String> errorCaptor = ArgumentCaptor.forClass(String.class);
|
||||
verify(traceRecordService).finishNode(eq("trace-1"), eq("node-1"), eq(TraceConstants.STATUS_ERROR),
|
||||
errorCaptor.capture(), isNull(), any(Date.class), anyLong());
|
||||
assertEquals("IllegalS", errorCaptor.getValue());
|
||||
}
|
||||
|
||||
@Test
|
||||
void detachShouldPopOnlyOnce() {
|
||||
TraceContext.pushNode("root");
|
||||
TraceContext.pushNode("stream");
|
||||
DefaultTraceStreamSpan span = new DefaultTraceStreamSpan(traceRecordService, new TraceProperties(), "trace-1", "stream", System.currentTimeMillis());
|
||||
|
||||
span.detach();
|
||||
span.detach();
|
||||
|
||||
assertEquals("root", TraceContext.currentNodeId());
|
||||
}
|
||||
|
||||
@Test
|
||||
void writeFailureShouldNotEscape() {
|
||||
doThrow(new IllegalStateException("db down")).when(traceRecordService)
|
||||
.finishNode(eq("trace-1"), eq("node-1"), eq(TraceConstants.STATUS_SUCCESS), isNull(), isNull(), any(Date.class), anyLong());
|
||||
DefaultTraceStreamSpan span = new DefaultTraceStreamSpan(traceRecordService, new TraceProperties(), "trace-1", "node-1", System.currentTimeMillis());
|
||||
|
||||
span.finishSuccess();
|
||||
|
||||
assertNull(TraceContext.currentNodeId());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,45 @@
|
||||
package org.ruoyi.common.trace.core;
|
||||
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertNull;
|
||||
|
||||
class TraceContextTest {
|
||||
|
||||
@AfterEach
|
||||
void tearDown() {
|
||||
TraceContext.clear();
|
||||
}
|
||||
|
||||
@Test
|
||||
void beginShouldExposeTraceAndBusinessIdentity() {
|
||||
try (TraceScope ignored = TraceContext.begin("trace-1", "RAG_CHAT", "session-1", 10L, "000000")) {
|
||||
assertEquals("trace-1", TraceContext.getTraceId());
|
||||
assertEquals("RAG_CHAT", TraceContext.getBusinessType());
|
||||
assertEquals("session-1", TraceContext.getBusinessId());
|
||||
assertEquals(10L, TraceContext.getUserId());
|
||||
assertEquals("000000", TraceContext.getTenantId());
|
||||
}
|
||||
assertNull(TraceContext.getTraceId());
|
||||
}
|
||||
|
||||
@Test
|
||||
void nodeStackShouldTrackParentAndDepth() {
|
||||
TraceContext.pushNode("root");
|
||||
assertEquals("root", TraceContext.currentNodeId());
|
||||
assertEquals(1, TraceContext.depth());
|
||||
|
||||
TraceContext.pushNode("child");
|
||||
assertEquals("child", TraceContext.currentNodeId());
|
||||
assertEquals(2, TraceContext.depth());
|
||||
|
||||
TraceContext.popNode();
|
||||
assertEquals("root", TraceContext.currentNodeId());
|
||||
|
||||
TraceContext.popNode();
|
||||
assertNull(TraceContext.currentNodeId());
|
||||
assertEquals(0, TraceContext.depth());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
package org.ruoyi.common.trace.util;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.util.Map;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
class TracePayloadUtilsTest {
|
||||
|
||||
@Test
|
||||
void truncateShouldRespectMaxLength() {
|
||||
assertEquals("abc", TracePayloadUtils.truncate("abcdef", 3));
|
||||
}
|
||||
|
||||
@Test
|
||||
void truncateShouldHandleNull() {
|
||||
assertNull(TracePayloadUtils.truncate(null, 3));
|
||||
}
|
||||
|
||||
@Test
|
||||
void toJsonShouldSerializeMap() {
|
||||
String json = TracePayloadUtils.toJson(Map.of("count", 2));
|
||||
assertTrue(json.contains("\"count\":2"));
|
||||
}
|
||||
}
|
||||
@@ -56,10 +56,10 @@ public class WorkflowStarter implements IWorkFlowStarterService {
|
||||
public SseEmitter streaming(User user, String workflowUuid, List<ObjectNode> 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;
|
||||
}
|
||||
|
||||
@@ -37,6 +37,11 @@
|
||||
<artifactId>ruoyi-common-sensitive</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.ruoyi</groupId>
|
||||
<artifactId>ruoyi-common-trace</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>dev.langchain4j</groupId>
|
||||
<artifactId>langchain4j-open-ai</artifactId>
|
||||
|
||||
@@ -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());
|
||||
|
||||
@@ -56,13 +56,22 @@ import org.ruoyi.common.core.utils.StringUtils;
|
||||
import org.ruoyi.common.satoken.utils.LoginHelper;
|
||||
import org.ruoyi.common.sse.core.SseEmitterManager;
|
||||
import org.ruoyi.common.sse.utils.SseMessageUtils;
|
||||
import org.ruoyi.common.trace.config.TraceProperties;
|
||||
import org.ruoyi.common.trace.constant.TraceConstants;
|
||||
import org.ruoyi.common.trace.core.DefaultTraceStreamSpan;
|
||||
import org.ruoyi.common.trace.core.TraceContext;
|
||||
import org.ruoyi.common.trace.core.TraceScope;
|
||||
import org.ruoyi.common.trace.core.TraceStreamSpan;
|
||||
import org.ruoyi.common.trace.domain.TraceNode;
|
||||
import org.ruoyi.common.trace.domain.TraceRun;
|
||||
import org.ruoyi.common.trace.service.TraceRecordService;
|
||||
import org.ruoyi.common.trace.util.TracePayloadUtils;
|
||||
import org.ruoyi.config.agent.SkillsPathResolver;
|
||||
import org.ruoyi.domain.bo.vector.QueryVectorBo;
|
||||
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;
|
||||
@@ -71,16 +80,20 @@ 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;
|
||||
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Date;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
/**
|
||||
* 聊天服务门面层
|
||||
@@ -105,8 +118,6 @@ public class ChatServiceFacade implements IChatService {
|
||||
|
||||
private final IKnowledgeInfoService knowledgeInfoService;
|
||||
|
||||
private final VectorStoreService vectorStoreService;
|
||||
|
||||
private final KnowledgeRetrievalService knowledgeRetrievalService;
|
||||
|
||||
private final SseEmitterManager sseEmitterManager;
|
||||
@@ -115,12 +126,15 @@ public class ChatServiceFacade implements IChatService {
|
||||
|
||||
private final IWorkFlowStarterService workFlowStarterService;
|
||||
|
||||
private final ToolProviderFactory toolProviderFactory;
|
||||
|
||||
private final IAgentService agentService;
|
||||
|
||||
private final LangChain4jMcpToolProviderService langChain4jMcpToolProviderService;
|
||||
|
||||
private final TraceRecordService traceRecordService;
|
||||
|
||||
private final TraceProperties traceProperties;
|
||||
|
||||
/**
|
||||
* 内存实例缓存,避免同一会话重复创建
|
||||
* Key: sessionId, Value: MessageWindowChatMemory实例
|
||||
@@ -137,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());
|
||||
@@ -157,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<ChatMessage> contextMessages = buildContextMessages(chatRequest, agentVo);
|
||||
|
||||
chatRequest.setEmitter(emitter);
|
||||
@@ -176,8 +190,10 @@ public class ChatServiceFacade implements IChatService {
|
||||
// 保存用户消息
|
||||
chatMessageService.saveChatMessage(userId, chatRequest.getSessionId(), chatRequest.getContent(), RoleType.USER.getName(), chatRequest.getModel());
|
||||
|
||||
TraceRunHandle traceRun = Boolean.TRUE.equals(chatRequest.getEnableWorkFlow())
|
||||
? null : startRagTraceRun(chatRequest, userId);
|
||||
// 3. 路由对话模式:工作流对话 / 智能体对话(两者均返回各自的 SseEmitter)
|
||||
return handleSpecialChatModes(chatRequest, agentVo);
|
||||
return handleSpecialChatModes(chatRequest, agentVo, traceRun);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -187,7 +203,7 @@ public class ChatServiceFacade implements IChatService {
|
||||
* @param agentVo 智能体配置(可为 null)
|
||||
* @return 对应模式的 SseEmitter
|
||||
*/
|
||||
private SseEmitter handleSpecialChatModes(ChatRequest chatRequest, AgentVo agentVo) {
|
||||
private SseEmitter handleSpecialChatModes(ChatRequest chatRequest, AgentVo agentVo, TraceRunHandle traceRun) {
|
||||
// 模式1:工作流对话(前端应用市场选工作流后携带 workFlowRunner)
|
||||
if (Boolean.TRUE.equals(chatRequest.getEnableWorkFlow())) {
|
||||
log.info("处理工作流对话,会话: {}", chatRequest.getSessionId());
|
||||
@@ -203,7 +219,7 @@ public class ChatServiceFacade implements IChatService {
|
||||
);
|
||||
}
|
||||
// 模式2:智能体对话(默认走 Supervisor 多 Agent 编排)
|
||||
return handleAgentChat(chatRequest, agentVo);
|
||||
return handleAgentChat(chatRequest, agentVo, traceRun);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -212,7 +228,7 @@ public class ChatServiceFacade implements IChatService {
|
||||
* @param chatRequest 聊天请求
|
||||
* @param agentVo 智能体配置(可为 null,无智能体时用请求 model 兜底)
|
||||
*/
|
||||
private SseEmitter handleAgentChat(ChatRequest chatRequest, AgentVo agentVo) {
|
||||
private SseEmitter handleAgentChat(ChatRequest chatRequest, AgentVo agentVo, TraceRunHandle traceRun) {
|
||||
ChatModelVo chatModelVo = chatRequest.getChatModelVo();
|
||||
|
||||
// 配置监督者模型:统一按 providerCode 走对应 AbstractChatService.buildChatModel,
|
||||
@@ -221,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,否则加载全部
|
||||
@@ -300,33 +317,148 @@ public class ChatServiceFacade implements IChatService {
|
||||
promptBuilder.append(augmentedInput);
|
||||
String prompt = promptBuilder.toString();
|
||||
|
||||
String tokenValue = chatRequest.getTokenValue();
|
||||
|
||||
// 异步执行 supervisor,避免阻塞 HTTP 请求线程导致 SSE 事件被缓冲
|
||||
CompletableFuture.runAsync(() -> {
|
||||
try {
|
||||
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(),
|
||||
result, RoleType.ASSISTANT.getName(), chatRequest.getModel());
|
||||
}
|
||||
if (llmSpan != null) {
|
||||
llmSpan.finishSuccess(RagTracePayloadBuilder.streamOutputSummary(
|
||||
result == null ? 0 : result.length()));
|
||||
}
|
||||
finishTraceRun(traceRun, TraceConstants.STATUS_SUCCESS, null);
|
||||
} catch (Exception e) {
|
||||
if (llmSpan != null) {
|
||||
llmSpan.finishError(e);
|
||||
}
|
||||
finishTraceRun(traceRun, TraceConstants.STATUS_ERROR, e);
|
||||
log.error("Supervisor 执行失败", e);
|
||||
SseMessageUtils.sendError(userId, e.getMessage());
|
||||
SseMessageUtils.sendError(sessionId, e.getMessage());
|
||||
} finally {
|
||||
SseMessageUtils.completeConnection(userId, tokenValue);
|
||||
if (llmSpan != null) {
|
||||
llmSpan.detach();
|
||||
}
|
||||
SseMessageUtils.completeConnection(sessionId);
|
||||
}
|
||||
});
|
||||
return chatRequest.getEmitter();
|
||||
}
|
||||
|
||||
private TraceRunHandle startRagTraceRun(ChatRequest chatRequest, Long userId) {
|
||||
if (!traceProperties.isEnabled()) {
|
||||
return null;
|
||||
}
|
||||
|
||||
String traceId = UUID.randomUUID().toString().replace("-", "");
|
||||
long startMillis = System.currentTimeMillis();
|
||||
TraceRun run = new TraceRun();
|
||||
run.setTraceId(traceId);
|
||||
run.setTraceName(RagTraceNodeTypes.TRACE_NAME_RAG_CHAT);
|
||||
run.setBusinessType(RagTraceNodeTypes.BUSINESS_TYPE_RAG_CHAT);
|
||||
run.setBusinessId(chatRequest.getSessionId() == null ? null : chatRequest.getSessionId().toString());
|
||||
run.setUserId(userId);
|
||||
run.setTenantId(safeGetTenantId());
|
||||
run.setStatus(TraceConstants.STATUS_RUNNING);
|
||||
run.setStartTime(new Date(startMillis));
|
||||
run.setMetadata(RagTracePayloadBuilder.chatRequestSummary(chatRequest));
|
||||
|
||||
try {
|
||||
traceRecordService.startRun(run);
|
||||
} catch (Exception e) {
|
||||
log.warn("写入 RAG chat trace run 失败,traceId={}", traceId, e);
|
||||
}
|
||||
return new TraceRunHandle(traceId, startMillis, run.getBusinessId(), run.getTenantId());
|
||||
}
|
||||
|
||||
private TraceScope openTraceScope(TraceRunHandle traceRun, Long userId) {
|
||||
if (traceRun == null) {
|
||||
return null;
|
||||
}
|
||||
return TraceContext.begin(traceRun.traceId, RagTraceNodeTypes.BUSINESS_TYPE_RAG_CHAT,
|
||||
traceRun.businessId, userId, traceRun.tenantId);
|
||||
}
|
||||
|
||||
private TraceStreamSpan startLlmCallSpan(TraceRunHandle traceRun, ChatRequest chatRequest) {
|
||||
if (traceRun == null || StringUtils.isBlank(TraceContext.getTraceId())) {
|
||||
return null;
|
||||
}
|
||||
|
||||
String nodeId = UUID.randomUUID().toString().replace("-", "");
|
||||
long startMillis = System.currentTimeMillis();
|
||||
TraceNode node = new TraceNode();
|
||||
node.setTraceId(traceRun.traceId);
|
||||
node.setNodeId(nodeId);
|
||||
node.setParentNodeId(TraceContext.currentNodeId());
|
||||
node.setDepth(TraceContext.depth());
|
||||
node.setNodeName("llm-call");
|
||||
node.setNodeType(RagTraceNodeTypes.NODE_LLM_CALL);
|
||||
node.setClassName(ChatServiceFacade.class.getName());
|
||||
node.setMethodName("handleAgentChat");
|
||||
node.setStatus(TraceConstants.STATUS_RUNNING);
|
||||
node.setStartTime(new Date(startMillis));
|
||||
node.setInputPayload(RagTracePayloadBuilder.streamInputSummary(chatRequest));
|
||||
|
||||
try {
|
||||
traceRecordService.startNode(node);
|
||||
TraceContext.pushNode(nodeId);
|
||||
return new DefaultTraceStreamSpan(traceRecordService, traceProperties, traceRun.traceId, nodeId, startMillis);
|
||||
} catch (Exception e) {
|
||||
log.warn("写入 LLM trace 节点失败,traceId={}, nodeId={}", traceRun.traceId, nodeId, e);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
private void finishTraceRun(TraceRunHandle traceRun, String status, Throwable error) {
|
||||
if (traceRun == null || !traceRun.finished.compareAndSet(false, true)) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
traceRecordService.finishRun(traceRun.traceId, status, TracePayloadUtils.error(error, traceProperties),
|
||||
new Date(), System.currentTimeMillis() - traceRun.startMillis);
|
||||
} catch (Exception e) {
|
||||
log.warn("结束 RAG chat trace run 失败,traceId={}", traceRun.traceId, e);
|
||||
}
|
||||
}
|
||||
|
||||
private String safeGetTenantId() {
|
||||
try {
|
||||
return LoginHelper.getTenantId();
|
||||
} catch (Exception e) {
|
||||
log.warn("获取 trace tenantId 失败: {}", e.getMessage());
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
private static final class TraceRunHandle {
|
||||
|
||||
private final String traceId;
|
||||
private final long startMillis;
|
||||
private final String businessId;
|
||||
private final String tenantId;
|
||||
private final AtomicBoolean finished = new AtomicBoolean(false);
|
||||
|
||||
private TraceRunHandle(String traceId, long startMillis, String businessId, String tenantId) {
|
||||
this.traceId = traceId;
|
||||
this.startMillis = startMillis;
|
||||
this.businessId = businessId;
|
||||
this.tenantId = tenantId;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 兜底 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"))
|
||||
@@ -334,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");
|
||||
@@ -345,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()
|
||||
@@ -430,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);
|
||||
@@ -663,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() {
|
||||
|
||||
@@ -681,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) {
|
||||
@@ -692,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) {
|
||||
@@ -704,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) {
|
||||
@@ -721,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
|
||||
@@ -732,4 +862,3 @@ public class ChatServiceFacade implements IChatService {
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -4,6 +4,13 @@ import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.ruoyi.common.core.utils.StringUtils;
|
||||
import org.ruoyi.common.core.exception.ServiceException;
|
||||
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.core.TraceNodeTemplate;
|
||||
import org.ruoyi.common.trace.domain.TraceNode;
|
||||
import org.ruoyi.common.trace.service.TraceRecordService;
|
||||
import org.ruoyi.common.trace.util.TracePayloadUtils;
|
||||
import org.ruoyi.domain.bo.rerank.RerankRequest;
|
||||
import org.ruoyi.domain.bo.rerank.RerankResult;
|
||||
import org.ruoyi.domain.bo.vector.QueryVectorBo;
|
||||
@@ -14,12 +21,15 @@ import org.ruoyi.mapper.knowledge.KnowledgeFragmentMapper;
|
||||
import org.ruoyi.service.rerank.RerankModelService;
|
||||
import org.ruoyi.service.retrieval.KnowledgeRetrievalService;
|
||||
import org.ruoyi.service.vector.VectorStoreService;
|
||||
import org.ruoyi.trace.RagTraceNodeTypes;
|
||||
import org.ruoyi.trace.RagTracePayloadBuilder;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import java.util.*;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
/**
|
||||
@@ -37,6 +47,8 @@ public class KnowledgeRetrievalServiceImpl implements KnowledgeRetrievalService
|
||||
private final VectorStoreService vectorStoreService;
|
||||
private final RerankModelFactory rerankModelFactory;
|
||||
private final KnowledgeFragmentMapper fragmentMapper;
|
||||
private final TraceRecordService traceRecordService;
|
||||
private final TraceProperties traceProperties;
|
||||
|
||||
/**
|
||||
* 粗召回默认扩大倍数
|
||||
@@ -64,6 +76,20 @@ public class KnowledgeRetrievalServiceImpl implements KnowledgeRetrievalService
|
||||
}
|
||||
log.info("开始知识库检索, kid={}, query={}", queryVectorBo.getKid(), queryVectorBo.getQuery());
|
||||
|
||||
String retrievalInputPayload = traceActive()
|
||||
? RagTracePayloadBuilder.retrievalInputSummary(queryVectorBo) : null;
|
||||
List<KnowledgeRetrievalVo> finalResults = TraceNodeTemplate.withNode(traceRecordService, traceProperties,
|
||||
"retrieval", RagTraceNodeTypes.NODE_RETRIEVAL,
|
||||
KnowledgeRetrievalServiceImpl.class.getName(), "retrieve",
|
||||
retrievalInputPayload,
|
||||
() -> retrieveUncached(queryVectorBo),
|
||||
RagTracePayloadBuilder::retrievalOutputSummary);
|
||||
cache(cacheKey, finalResults);
|
||||
return copyResults(finalResults);
|
||||
}
|
||||
|
||||
private List<KnowledgeRetrievalVo> retrieveUncached(QueryVectorBo queryVectorBo) {
|
||||
|
||||
// 1. 粗召回阶段 (向量检索 + 关键词搜索)
|
||||
List<KnowledgeRetrievalVo> coarseResults = performCoarseRetrieval(queryVectorBo);
|
||||
log.debug("粗召回返回 {} 条结果", coarseResults.size());
|
||||
@@ -92,8 +118,7 @@ public class KnowledgeRetrievalServiceImpl implements KnowledgeRetrievalService
|
||||
.filter(res -> res.getScore() != null && res.getScore() >= threshold)
|
||||
.collect(Collectors.toList());
|
||||
}
|
||||
cache(cacheKey, finalResults);
|
||||
return copyResults(finalResults);
|
||||
return finalResults;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -182,6 +207,11 @@ public class KnowledgeRetrievalServiceImpl implements KnowledgeRetrievalService
|
||||
* 重排序阶段
|
||||
*/
|
||||
private List<KnowledgeRetrievalVo> performRerank(QueryVectorBo queryVectorBo, List<KnowledgeRetrievalVo> coarseResults) {
|
||||
int topN = queryVectorBo.getRerankTopN() != null ? queryVectorBo.getRerankTopN() : queryVectorBo.getMaxResults();
|
||||
String rerankInputPayload = traceActive()
|
||||
? RagTracePayloadBuilder.rerankInputSummary(queryVectorBo, coarseResults.size(), topN) : null;
|
||||
TraceNodeHandle traceNode = startTraceNode("rerank", RagTraceNodeTypes.NODE_RERANK, "performRerank",
|
||||
rerankInputPayload);
|
||||
try {
|
||||
RerankModelService rerankModel = rerankModelFactory.createModel(queryVectorBo.getRerankModelName());
|
||||
|
||||
@@ -190,8 +220,6 @@ public class KnowledgeRetrievalServiceImpl implements KnowledgeRetrievalService
|
||||
.collect(Collectors.toList());
|
||||
|
||||
// topN 默认为 maxResults
|
||||
int topN = queryVectorBo.getRerankTopN() != null ? queryVectorBo.getRerankTopN() : queryVectorBo.getMaxResults();
|
||||
|
||||
RerankRequest rerankRequest = RerankRequest.builder()
|
||||
.query(queryVectorBo.getQuery())
|
||||
.documents(contents)
|
||||
@@ -215,12 +243,18 @@ public class KnowledgeRetrievalServiceImpl implements KnowledgeRetrievalService
|
||||
reranked.sort((a, b) -> b.getScore().compareTo(a.getScore()));
|
||||
|
||||
// 截断到 topN
|
||||
return reranked.subList(0, Math.min(topN, reranked.size()));
|
||||
List<KnowledgeRetrievalVo> results = reranked.subList(0, Math.min(topN, reranked.size()));
|
||||
finishTraceNode(traceNode, TraceConstants.STATUS_SUCCESS, null,
|
||||
RagTracePayloadBuilder.rerankOutputSummary(results));
|
||||
return results;
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("重排序流程失败: {}", e.getMessage());
|
||||
int limit = queryVectorBo.getMaxResults() != null ? queryVectorBo.getMaxResults() : 10;
|
||||
return coarseResults.subList(0, Math.min(limit, coarseResults.size()));
|
||||
List<KnowledgeRetrievalVo> fallback = coarseResults.subList(0, Math.min(limit, coarseResults.size()));
|
||||
finishTraceNode(traceNode, TraceConstants.STATUS_ERROR, e,
|
||||
RagTracePayloadBuilder.rerankOutputSummary(fallback));
|
||||
return fallback;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -263,6 +297,56 @@ public class KnowledgeRetrievalServiceImpl implements KnowledgeRetrievalService
|
||||
return fusedResults;
|
||||
}
|
||||
|
||||
private TraceNodeHandle startTraceNode(String nodeName, String nodeType, String methodName, String inputPayload) {
|
||||
if (!traceProperties.isEnabled() || StringUtils.isBlank(TraceContext.getTraceId())) {
|
||||
return null;
|
||||
}
|
||||
|
||||
String traceId = TraceContext.getTraceId();
|
||||
String nodeId = UUID.randomUUID().toString().replace("-", "");
|
||||
long startMillis = System.currentTimeMillis();
|
||||
TraceNode node = new TraceNode();
|
||||
node.setTraceId(traceId);
|
||||
node.setNodeId(nodeId);
|
||||
node.setParentNodeId(TraceContext.currentNodeId());
|
||||
node.setDepth(TraceContext.depth());
|
||||
node.setNodeName(nodeName);
|
||||
node.setNodeType(nodeType);
|
||||
node.setClassName(KnowledgeRetrievalServiceImpl.class.getName());
|
||||
node.setMethodName(methodName);
|
||||
node.setStatus(TraceConstants.STATUS_RUNNING);
|
||||
node.setStartTime(new Date(startMillis));
|
||||
node.setInputPayload(inputPayload);
|
||||
|
||||
try {
|
||||
traceRecordService.startNode(node);
|
||||
TraceContext.pushNode(nodeId);
|
||||
return new TraceNodeHandle(traceId, nodeId, startMillis);
|
||||
} catch (Exception e) {
|
||||
log.warn("写入 RAG 检索 trace 节点失败,traceId={}, nodeId={}", traceId, nodeId, e);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
private boolean traceActive() {
|
||||
return traceProperties.isEnabled() && StringUtils.isNotBlank(TraceContext.getTraceId());
|
||||
}
|
||||
|
||||
private void finishTraceNode(TraceNodeHandle traceNode, String status, Throwable error, String outputPayload) {
|
||||
if (traceNode == null || !traceNode.finished.compareAndSet(false, true)) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
traceRecordService.finishNode(traceNode.traceId, traceNode.nodeId, status,
|
||||
TracePayloadUtils.error(error, traceProperties), outputPayload,
|
||||
new Date(), System.currentTimeMillis() - traceNode.startMillis);
|
||||
} catch (Exception e) {
|
||||
log.warn("结束 RAG 检索 trace 节点失败,traceId={}, nodeId={}", traceNode.traceId, traceNode.nodeId, e);
|
||||
} finally {
|
||||
TraceContext.popNode();
|
||||
}
|
||||
}
|
||||
|
||||
private QueryVectorBo copyOf(QueryVectorBo original, int maxResults) {
|
||||
QueryVectorBo copy = new QueryVectorBo();
|
||||
copy.setQuery(original.getQuery());
|
||||
@@ -312,4 +396,18 @@ public class KnowledgeRetrievalServiceImpl implements KnowledgeRetrievalService
|
||||
}
|
||||
|
||||
private record CacheEntry(long createdAt, List<KnowledgeRetrievalVo> results) { }
|
||||
|
||||
private static final class TraceNodeHandle {
|
||||
|
||||
private final String traceId;
|
||||
private final String nodeId;
|
||||
private final long startMillis;
|
||||
private final AtomicBoolean finished = new AtomicBoolean(false);
|
||||
|
||||
private TraceNodeHandle(String traceId, String nodeId, long startMillis) {
|
||||
this.traceId = traceId;
|
||||
this.nodeId = nodeId;
|
||||
this.startMillis = startMillis;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,17 @@
|
||||
package org.ruoyi.trace;
|
||||
|
||||
/**
|
||||
* RAG trace 业务与节点类型常量。
|
||||
*/
|
||||
public final class RagTraceNodeTypes {
|
||||
|
||||
private RagTraceNodeTypes() {
|
||||
}
|
||||
|
||||
public static final String BUSINESS_TYPE_RAG_CHAT = "RAG_CHAT";
|
||||
public static final String TRACE_NAME_RAG_CHAT = "rag-chat";
|
||||
|
||||
public static final String NODE_RETRIEVAL = "RETRIEVAL";
|
||||
public static final String NODE_RERANK = "RERANK";
|
||||
public static final String NODE_LLM_CALL = "LLM_CALL";
|
||||
}
|
||||
@@ -0,0 +1,131 @@
|
||||
package org.ruoyi.trace;
|
||||
|
||||
import org.ruoyi.common.chat.domain.dto.request.ChatRequest;
|
||||
import org.ruoyi.common.chat.domain.vo.chat.ChatModelVo;
|
||||
import org.ruoyi.common.trace.util.TracePayloadUtils;
|
||||
import org.ruoyi.domain.bo.vector.QueryVectorBo;
|
||||
import org.ruoyi.domain.vo.knowledge.KnowledgeRetrievalVo;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* RAG trace payload 摘要构建器。
|
||||
*/
|
||||
public final class RagTracePayloadBuilder {
|
||||
|
||||
private static final int MAX_RESULT_SUMMARY_SIZE = 5;
|
||||
|
||||
private RagTracePayloadBuilder() {
|
||||
}
|
||||
|
||||
public static String chatRequestSummary(ChatRequest request) {
|
||||
Map<String, Object> payload = new LinkedHashMap<>();
|
||||
payload.put("requestPresent", request != null);
|
||||
if (request == null) {
|
||||
return TracePayloadUtils.toJson(payload);
|
||||
}
|
||||
|
||||
ChatModelVo model = request.getChatModelVo();
|
||||
payload.put("sessionId", request.getSessionId() == null ? null : request.getSessionId().toString());
|
||||
payload.put("model", request.getModel());
|
||||
payload.put("providerCode", model == null ? null : model.getProviderCode());
|
||||
payload.put("knowledgeId", request.getKnowledgeId());
|
||||
payload.put("hasKnowledge", request.getKnowledgeId() != null);
|
||||
payload.put("contentLength", length(request.getContent()));
|
||||
payload.put("contextMessageCount", request.getContextMessages() == null ? null : request.getContextMessages().size());
|
||||
payload.put("enableWorkFlow", request.getEnableWorkFlow());
|
||||
payload.put("enableThinking", request.getEnableThinking());
|
||||
return TracePayloadUtils.toJson(payload);
|
||||
}
|
||||
|
||||
public static String retrievalInputSummary(QueryVectorBo query) {
|
||||
Map<String, Object> payload = new LinkedHashMap<>();
|
||||
payload.put("queryPresent", query != null);
|
||||
if (query == null) {
|
||||
return TracePayloadUtils.toJson(payload);
|
||||
}
|
||||
|
||||
payload.put("kid", query.getKid());
|
||||
payload.put("queryLength", length(query.getQuery()));
|
||||
payload.put("maxResults", query.getMaxResults());
|
||||
payload.put("vectorModelName", query.getVectorModelName());
|
||||
payload.put("embeddingModelName", query.getEmbeddingModelName());
|
||||
payload.put("enableHybrid", query.getEnableHybrid());
|
||||
payload.put("hybridAlpha", query.getHybridAlpha());
|
||||
payload.put("similarityThreshold", query.getSimilarityThreshold());
|
||||
payload.put("enableRerank", query.getEnableRerank());
|
||||
payload.put("rerankModel", query.getRerankModelName());
|
||||
payload.put("rerankTopN", query.getRerankTopN());
|
||||
payload.put("rerankScoreThreshold", query.getRerankScoreThreshold());
|
||||
return TracePayloadUtils.toJson(payload);
|
||||
}
|
||||
|
||||
public static String retrievalOutputSummary(List<KnowledgeRetrievalVo> results) {
|
||||
Map<String, Object> payload = new LinkedHashMap<>();
|
||||
payload.put("resultCount", results == null ? 0 : results.size());
|
||||
payload.put("results", summarizeResults(results));
|
||||
return TracePayloadUtils.toJson(payload);
|
||||
}
|
||||
|
||||
public static String rerankInputSummary(QueryVectorBo query, int candidateCount, Integer topN) {
|
||||
Map<String, Object> payload = new LinkedHashMap<>();
|
||||
payload.put("candidateCount", candidateCount);
|
||||
payload.put("rerankModel", query == null ? null : query.getRerankModelName());
|
||||
payload.put("topN", topN);
|
||||
return TracePayloadUtils.toJson(payload);
|
||||
}
|
||||
|
||||
public static String rerankOutputSummary(List<KnowledgeRetrievalVo> results) {
|
||||
Map<String, Object> payload = new LinkedHashMap<>();
|
||||
payload.put("resultCount", results == null ? 0 : results.size());
|
||||
payload.put("results", summarizeResults(results));
|
||||
return TracePayloadUtils.toJson(payload);
|
||||
}
|
||||
|
||||
public static String streamInputSummary(ChatRequest request) {
|
||||
Map<String, Object> payload = new LinkedHashMap<>();
|
||||
payload.put("sessionId", request == null || request.getSessionId() == null ? null : request.getSessionId().toString());
|
||||
payload.put("model", request == null ? null : request.getModel());
|
||||
payload.put("contextMessageCount", request == null || request.getContextMessages() == null ? null : request.getContextMessages().size());
|
||||
payload.put("contentLength", request == null ? null : length(request.getContent()));
|
||||
return TracePayloadUtils.toJson(payload);
|
||||
}
|
||||
|
||||
public static String streamOutputSummary(int responseLength) {
|
||||
Map<String, Object> payload = new LinkedHashMap<>();
|
||||
payload.put("responseLength", responseLength);
|
||||
return TracePayloadUtils.toJson(payload);
|
||||
}
|
||||
|
||||
private static List<Map<String, Object>> summarizeResults(List<KnowledgeRetrievalVo> results) {
|
||||
List<Map<String, Object>> summaries = new ArrayList<>();
|
||||
if (results == null || results.isEmpty()) {
|
||||
return summaries;
|
||||
}
|
||||
|
||||
int limit = Math.min(MAX_RESULT_SUMMARY_SIZE, results.size());
|
||||
for (int i = 0; i < limit; i++) {
|
||||
KnowledgeRetrievalVo result = results.get(i);
|
||||
Map<String, Object> item = new LinkedHashMap<>();
|
||||
item.put("rank", i + 1);
|
||||
item.put("id", result == null ? null : result.getId());
|
||||
item.put("docId", result == null ? null : result.getDocId());
|
||||
item.put("knowledgeId", result == null || result.getKnowledgeId() == null ? null : result.getKnowledgeId().toString());
|
||||
item.put("idx", result == null ? null : result.getIdx());
|
||||
item.put("score", result == null ? null : result.getScore());
|
||||
item.put("rawScore", result == null ? null : result.getRawScore());
|
||||
item.put("originalIndex", result == null ? null : result.getOriginalIndex());
|
||||
item.put("sourceName", result == null ? null : result.getSourceName());
|
||||
item.put("contentLength", result == null ? null : length(result.getContent()));
|
||||
summaries.add(item);
|
||||
}
|
||||
return summaries;
|
||||
}
|
||||
|
||||
private static Integer length(String value) {
|
||||
return value == null ? null : value.length();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,78 @@
|
||||
package org.ruoyi.trace;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.ruoyi.common.chat.domain.dto.request.ChatRequest;
|
||||
import org.ruoyi.common.chat.domain.vo.chat.ChatModelVo;
|
||||
import org.ruoyi.domain.bo.vector.QueryVectorBo;
|
||||
import org.ruoyi.domain.vo.knowledge.KnowledgeRetrievalVo;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
class RagTracePayloadBuilderTest {
|
||||
|
||||
@Test
|
||||
void chatRequestSummaryShouldHandleNullsAndAvoidPromptBody() {
|
||||
ChatRequest request = new ChatRequest();
|
||||
request.setSessionId(100L);
|
||||
request.setModel("qwen-plus");
|
||||
request.setKnowledgeId("200");
|
||||
request.setContent("secret prompt body");
|
||||
|
||||
ChatModelVo model = new ChatModelVo();
|
||||
model.setProviderCode("dashscope");
|
||||
request.setChatModelVo(model);
|
||||
|
||||
String payload = RagTracePayloadBuilder.chatRequestSummary(request);
|
||||
|
||||
assertTrue(payload.contains("\"sessionId\":100"));
|
||||
assertTrue(payload.contains("\"contentLength\":18"));
|
||||
assertTrue(payload.contains("\"providerCode\":\"dashscope\""));
|
||||
assertFalse(payload.contains("secret prompt body"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void retrievalInputSummaryShouldUseSummaryOnly() {
|
||||
QueryVectorBo query = new QueryVectorBo();
|
||||
query.setKid("200");
|
||||
query.setQuery("private retrieval query");
|
||||
query.setMaxResults(5);
|
||||
query.setEnableRerank(true);
|
||||
query.setRerankModelName("gte-rerank");
|
||||
query.setRerankTopN(null);
|
||||
|
||||
String payload = RagTracePayloadBuilder.retrievalInputSummary(query);
|
||||
|
||||
assertTrue(payload.contains("\"kid\":\"200\""));
|
||||
assertTrue(payload.contains("\"queryLength\":23"));
|
||||
assertTrue(payload.contains("\"enableRerank\":true"));
|
||||
assertFalse(payload.contains("private retrieval query"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void retrievalOutputSummaryShouldAvoidFragmentContent() {
|
||||
KnowledgeRetrievalVo result = new KnowledgeRetrievalVo();
|
||||
result.setId("fragment-1");
|
||||
result.setDocId("doc-1");
|
||||
result.setIdx(1);
|
||||
result.setScore(0.85);
|
||||
result.setContent("sensitive knowledge fragment");
|
||||
|
||||
String payload = RagTracePayloadBuilder.retrievalOutputSummary(List.of(result));
|
||||
|
||||
assertTrue(payload.contains("\"resultCount\":1"));
|
||||
assertTrue(payload.contains("\"contentLength\":28"));
|
||||
assertTrue(payload.contains("\"fragment-1\""));
|
||||
assertFalse(payload.contains("sensitive knowledge fragment"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void summariesShouldAcceptNullValuesWithoutMapOfNpe() {
|
||||
assertTrue(RagTracePayloadBuilder.chatRequestSummary(null).contains("\"requestPresent\":false"));
|
||||
assertTrue(RagTracePayloadBuilder.retrievalInputSummary(null).contains("\"queryPresent\":false"));
|
||||
assertTrue(RagTracePayloadBuilder.retrievalOutputSummary(null).contains("\"resultCount\":0"));
|
||||
assertTrue(RagTracePayloadBuilder.rerankInputSummary(null, 0, null).contains("\"candidateCount\":0"));
|
||||
}
|
||||
}
|
||||
@@ -48,6 +48,11 @@
|
||||
<artifactId>ruoyi-common-log</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.ruoyi</groupId>
|
||||
<artifactId>ruoyi-common-trace</artifactId>
|
||||
</dependency>
|
||||
|
||||
<!-- excel-->
|
||||
<dependency>
|
||||
<groupId>org.ruoyi</groupId>
|
||||
|
||||
@@ -0,0 +1,68 @@
|
||||
package org.ruoyi.system.controller.monitor;
|
||||
|
||||
import cn.dev33.satoken.annotation.SaCheckPermission;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import org.ruoyi.common.core.domain.R;
|
||||
import org.ruoyi.common.mybatis.core.page.PageQuery;
|
||||
import org.ruoyi.common.mybatis.core.page.TableDataInfo;
|
||||
import org.ruoyi.common.trace.domain.bo.TraceRunBo;
|
||||
import org.ruoyi.common.trace.domain.vo.TraceDetailVo;
|
||||
import org.ruoyi.common.trace.domain.vo.TraceNodeVo;
|
||||
import org.ruoyi.common.trace.domain.vo.TraceRunVo;
|
||||
import org.ruoyi.common.trace.service.TraceRecordService;
|
||||
import org.ruoyi.common.web.core.BaseController;
|
||||
import org.springframework.validation.annotation.Validated;
|
||||
import org.springframework.web.bind.annotation.GetMapping;
|
||||
import org.springframework.web.bind.annotation.PathVariable;
|
||||
import org.springframework.web.bind.annotation.RequestMapping;
|
||||
import org.springframework.web.bind.annotation.RestController;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* 链路追踪监控
|
||||
*/
|
||||
@Validated
|
||||
@RequiredArgsConstructor
|
||||
@RestController
|
||||
@RequestMapping("/monitor/trace")
|
||||
public class TraceController extends BaseController {
|
||||
|
||||
private final TraceRecordService traceRecordService;
|
||||
|
||||
/**
|
||||
* 获取链路追踪运行列表
|
||||
*/
|
||||
@SaCheckPermission("monitor:trace:list")
|
||||
@GetMapping("/run/list")
|
||||
public TableDataInfo<TraceRunVo> list(TraceRunBo bo, PageQuery pageQuery) {
|
||||
return traceRecordService.pageRuns(bo, pageQuery);
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取链路追踪运行详情
|
||||
*/
|
||||
@SaCheckPermission("monitor:trace:query")
|
||||
@GetMapping("/run/{traceId}")
|
||||
public R<TraceRunVo> run(@PathVariable String traceId) {
|
||||
return R.ok(traceRecordService.getRun(traceId));
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取链路追踪节点列表
|
||||
*/
|
||||
@SaCheckPermission("monitor:trace:query")
|
||||
@GetMapping("/node/list/{traceId}")
|
||||
public R<List<TraceNodeVo>> nodes(@PathVariable String traceId) {
|
||||
return R.ok(traceRecordService.listNodes(traceId));
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取链路追踪完整详情
|
||||
*/
|
||||
@SaCheckPermission("monitor:trace:query")
|
||||
@GetMapping("/detail/{traceId}")
|
||||
public R<TraceDetailVo> detail(@PathVariable String traceId) {
|
||||
return R.ok(traceRecordService.getDetail(traceId));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user