From b05f094ac51dce91eb8c00235226d54a04658c6d Mon Sep 17 00:00:00 2001
From: zhou zhou <3272660260@qq.com>
Date: 星期一, 23 三月 2026 15:51:17 +0800
Subject: [PATCH] #ai 页面优化
---
/dev/null | 129 --------
rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/chat/AiChatOrchestrator.java | 25
rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/chat/AiSseEventPublisher.java | 6
rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/chat/AiChatFailureHandler.java | 10
rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/chat/AiToolObservationService.java | 104 ++----
rsf-server/src/test/java/com/vincent/rsf/server/AI/service/impl/chat/AiToolObservationServiceTest.java | 70 ++++
rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/chat/AiChatTraceEmitter.java | 179 +++++++++++
rsf-admin/src/i18n/zh.js | 4
rsf-server/src/test/java/com/vincent/rsf/server/AI/service/impl/chat/AiChatTraceEmitterTest.java | 59 +++
rsf-server/src/main/java/com/vincent/rsf/server/ai/dto/AiChatTraceEventDto.java | 18 +
rsf-admin/src/i18n/en.js | 4
rsf-admin/src/layout/AiChatDrawer.jsx | 264 ++++++----------
12 files changed, 484 insertions(+), 388 deletions(-)
diff --git a/rsf-admin/src/i18n/en.js b/rsf-admin/src/i18n/en.js
index 5069c5b..8d4d70e 100644
--- a/rsf-admin/src/i18n/en.js
+++ b/rsf-admin/src/i18n/en.js
@@ -519,6 +519,7 @@
renameAction: "Rename session",
deleteAction: "Delete session",
toolTrace: "Tool Trace",
+ activityTrace: "Thinking & Tool Trace",
thinkingProcess: "Thinking Process",
thinkingExpand: "Show Thinking Process",
thinkingCollapse: "Hide Thinking Process",
@@ -529,7 +530,10 @@
thinkingStatusFailed: "Failed",
thinkingStatusAborted: "Aborted",
noToolTrace: "No tool call was triggered in this round",
+ noActivityTrace: "No thinking or tool trace in this round",
unknownTool: "Unknown tool",
+ traceTypeThinking: "Thinking",
+ traceTypeTool: "Tool",
toolStatusFailed: "Failed",
toolStatusCompleted: "Completed",
toolStatusRunning: "Running",
diff --git a/rsf-admin/src/i18n/zh.js b/rsf-admin/src/i18n/zh.js
index fd62ee9..6759812 100644
--- a/rsf-admin/src/i18n/zh.js
+++ b/rsf-admin/src/i18n/zh.js
@@ -535,6 +535,7 @@
renameAction: "閲嶅懡鍚嶄細璇�",
deleteAction: "鍒犻櫎浼氳瘽",
toolTrace: "宸ュ叿璋冪敤杞ㄨ抗",
+ activityTrace: "鎬濈淮閾句笌宸ュ叿杞ㄨ抗",
thinkingProcess: "鎬濊�冭繃绋�",
thinkingExpand: "灞曞紑鎬濊�冭繃绋�",
thinkingCollapse: "鏀惰捣鎬濊�冭繃绋�",
@@ -545,7 +546,10 @@
thinkingStatusFailed: "澶辫触",
thinkingStatusAborted: "宸蹭腑姝�",
noToolTrace: "褰撳墠杞湭瑙﹀彂宸ュ叿璋冪敤",
+ noActivityTrace: "褰撳墠杞皻鏃犳�濊�冩垨宸ュ叿杞ㄨ抗",
unknownTool: "鏈煡宸ュ叿",
+ traceTypeThinking: "鎬濈淮閾�",
+ traceTypeTool: "宸ュ叿",
toolStatusFailed: "澶辫触",
toolStatusCompleted: "瀹屾垚",
toolStatusRunning: "鎵ц涓�",
diff --git a/rsf-admin/src/layout/AiChatDrawer.jsx b/rsf-admin/src/layout/AiChatDrawer.jsx
index fc1eb06..cbbd584 100644
--- a/rsf-admin/src/layout/AiChatDrawer.jsx
+++ b/rsf-admin/src/layout/AiChatDrawer.jsx
@@ -49,11 +49,6 @@
const DEFAULT_PROMPT_CODE = "home.default";
const AI_CHAT_DRAWER_Z_INDEX = 1400;
const AI_CHAT_DIALOG_Z_INDEX = AI_CHAT_DRAWER_Z_INDEX + 20;
-const THINKING_PHASE_ORDER = {
- ANALYZE: 0,
- TOOL_CALL: 1,
- ANSWER: 2,
-};
const normalizeMarkdownContent = (content) => {
if (!content) {
@@ -239,10 +234,8 @@
const [sessions, setSessions] = useState([]);
const [persistedMessages, setPersistedMessages] = useState([]);
const [messages, setMessages] = useState([]);
- const [toolEvents, setToolEvents] = useState([]);
- const [expandedToolIds, setExpandedToolIds] = useState([]);
- const [thinkingEvents, setThinkingEvents] = useState([]);
- const [thinkingExpanded, setThinkingExpanded] = useState(true);
+ const [traceEvents, setTraceEvents] = useState([]);
+ const [expandedTraceIds, setExpandedTraceIds] = useState([]);
const [input, setInput] = useState("");
const [loadingRuntime, setLoadingRuntime] = useState(false);
const [streaming, setStreaming] = useState(false);
@@ -286,18 +279,6 @@
};
}, [runtime]);
- const currentThinkingMessageIndex = useMemo(() => {
- if (!thinkingEvents.length || !messages.length) {
- return -1;
- }
- for (let i = messages.length - 1; i >= 0; i -= 1) {
- if (messages[i]?.role === "assistant") {
- return i;
- }
- }
- return -1;
- }, [messages, thinkingEvents]);
-
useEffect(() => {
if (open) {
setRuntimePanelExpanded(false);
@@ -322,10 +303,8 @@
}, [open, messages, streaming]);
const initializeDrawer = async (targetSessionId = null) => {
- setToolEvents([]);
- setExpandedToolIds([]);
- setThinkingEvents([]);
- setThinkingExpanded(true);
+ setTraceEvents([]);
+ setExpandedTraceIds([]);
await Promise.all([
loadRuntime(targetSessionId),
loadSessions(sessionKeyword),
@@ -370,10 +349,8 @@
setSessionId(null);
setPersistedMessages([]);
setMessages([]);
- setToolEvents([]);
- setExpandedToolIds([]);
- setThinkingEvents([]);
- setThinkingExpanded(true);
+ setTraceEvents([]);
+ setExpandedTraceIds([]);
setUsage(null);
setDrawerError("");
};
@@ -393,10 +370,8 @@
return;
}
setUsage(null);
- setToolEvents([]);
- setExpandedToolIds([]);
- setThinkingEvents([]);
- setThinkingExpanded(true);
+ setTraceEvents([]);
+ setExpandedTraceIds([]);
await loadRuntime(targetSessionId);
};
@@ -562,42 +537,15 @@
return next;
};
- const upsertToolEvent = (payload) => {
- if (!payload?.toolCallId) {
+ const appendTraceEvent = (payload) => {
+ if (!payload?.traceId) {
return;
}
- setToolEvents((prev) => {
- const index = prev.findIndex((item) => item.toolCallId === payload.toolCallId);
- if (index < 0) {
- return [...prev, payload];
- }
- const next = [...prev];
- next[index] = { ...next[index], ...payload };
- return next;
- });
- };
-
- const toggleToolEventExpanded = (toolCallId) => {
- if (!toolCallId) {
- return;
- }
- setExpandedToolIds((prev) => (
- prev.includes(toolCallId)
- ? prev.filter((item) => item !== toolCallId)
- : [...prev, toolCallId]
- ));
- };
-
- const upsertThinkingEvent = (payload) => {
- if (!payload?.phase) {
- return;
- }
- setThinkingEvents((prev) => {
- const index = prev.findIndex((item) => item.phase === payload.phase);
+ setTraceEvents((prev) => {
+ const index = prev.findIndex((item) => item.traceId === payload.traceId);
if (index < 0) {
return [...prev, payload].sort((left, right) => (
- (THINKING_PHASE_ORDER[left.phase] ?? Number.MAX_SAFE_INTEGER)
- - (THINKING_PHASE_ORDER[right.phase] ?? Number.MAX_SAFE_INTEGER)
+ (left?.sequence ?? 0) - (right?.sequence ?? 0)
));
}
const next = [...prev];
@@ -606,8 +554,15 @@
});
};
- const toggleThinkingExpanded = () => {
- setThinkingExpanded((prev) => !prev);
+ const toggleTraceEventExpanded = (traceId) => {
+ if (!traceId) {
+ return;
+ }
+ setExpandedTraceIds((prev) => (
+ prev.includes(traceId)
+ ? prev.filter((item) => item !== traceId)
+ : [...prev, traceId]
+ ));
};
const getThinkingStatusLabel = (status) => {
@@ -626,6 +581,16 @@
return translate("ai.drawer.thinkingStatusStarted");
};
+ const getToolStatusLabel = (status) => {
+ if (status === "FAILED") {
+ return translate("ai.drawer.toolStatusFailed");
+ }
+ if (status === "COMPLETED") {
+ return translate("ai.drawer.toolStatusCompleted");
+ }
+ return translate("ai.drawer.toolStatusRunning");
+ };
+
const handleSend = async () => {
const content = input.trim();
if (!content || streaming) {
@@ -636,10 +601,8 @@
setInput("");
setUsage(null);
setDrawerError("");
- setToolEvents([]);
- setExpandedToolIds([]);
- setThinkingEvents([]);
- setThinkingExpanded(true);
+ setTraceEvents([]);
+ setExpandedTraceIds([]);
setMessages(ensureAssistantPlaceholder(nextMessages));
setStreaming(true);
@@ -676,11 +639,8 @@
if (eventName === "delta") {
appendAssistantDelta(payload?.content || "");
}
- if (eventName === "tool_start" || eventName === "tool_result" || eventName === "tool_error") {
- upsertToolEvent(payload);
- }
- if (eventName === "thinking") {
- upsertThinkingEvent(payload);
+ if (eventName === "trace") {
+ appendTraceEvent(payload);
}
if (eventName === "done") {
setUsage(payload);
@@ -864,71 +824,106 @@
>
<Box px={2} py={1.5} display="flex" flexDirection="column" minHeight={0}>
<Typography variant="subtitle2" mb={1}>
- {translate("ai.drawer.toolTrace")}
+ {translate("ai.drawer.activityTrace")}
</Typography>
<Paper variant="outlined" sx={{ flex: 1, minHeight: { xs: 140, md: 0 }, overflow: "hidden", bgcolor: "grey.50" }}>
- {!toolEvents.length ? (
+ {!traceEvents.length ? (
<Box px={1.5} py={1.25}>
<Typography variant="body2" color="text.secondary">
- {translate("ai.drawer.noToolTrace")}
+ {translate("ai.drawer.noActivityTrace")}
</Typography>
</Box>
) : (
<Stack spacing={1} sx={{ p: 1.25, maxHeight: { xs: 220, md: "calc(100vh - 180px)" }, overflow: "auto" }}>
- {toolEvents.map((item) => (
+ {traceEvents.map((item) => (
<Paper
- key={item.toolCallId}
+ key={item.traceId}
variant="outlined"
sx={{
p: 1.25,
- bgcolor: item.status === "FAILED" ? "error.lighter" : "common.white",
- borderColor: item.status === "FAILED" ? "error.light" : "divider",
+ bgcolor: item.status === "FAILED"
+ ? "error.lighter"
+ : item.traceType === "thinking"
+ ? "info.lighter"
+ : "common.white",
+ borderColor: item.status === "FAILED"
+ ? "error.light"
+ : item.traceType === "thinking"
+ ? "info.light"
+ : "divider",
}}
>
<Stack direction="row" spacing={1} alignItems="center" flexWrap="wrap" useFlexGap>
+ <Chip
+ size="small"
+ variant="outlined"
+ color={item.traceType === "thinking" ? "info" : "primary"}
+ label={translate(item.traceType === "thinking" ? "ai.drawer.traceTypeThinking" : "ai.drawer.traceTypeTool")}
+ />
<Typography variant="body2" fontWeight={700}>
- {item.toolName || translate("ai.drawer.unknownTool")}
+ {item.traceType === "thinking"
+ ? (item.title || translate("ai.drawer.thinkingProcess"))
+ : (item.toolName || item.title || translate("ai.drawer.unknownTool"))}
</Typography>
<Chip
size="small"
- color={item.status === "FAILED" ? "error" : item.status === "COMPLETED" ? "success" : "info"}
- label={translate(item.status === "FAILED" ? "ai.drawer.toolStatusFailed" : item.status === "COMPLETED" ? "ai.drawer.toolStatusCompleted" : "ai.drawer.toolStatusRunning")}
+ color={item.status === "FAILED"
+ ? "error"
+ : item.status === "COMPLETED"
+ ? "success"
+ : item.status === "ABORTED"
+ ? "warning"
+ : "info"}
+ label={item.traceType === "thinking"
+ ? getThinkingStatusLabel(item.status)
+ : getToolStatusLabel(item.status)}
/>
{item.durationMs != null && (
<Typography variant="caption" color="text.secondary">
{item.durationMs} ms
</Typography>
)}
- {(item.inputSummary || item.outputSummary || item.errorMessage) && (
+ {item.traceType === "tool" && (item.inputSummary || item.outputSummary || item.errorMessage) && (
<Button
size="small"
- onClick={() => toggleToolEventExpanded(item.toolCallId)}
- endIcon={expandedToolIds.includes(item.toolCallId)
+ onClick={() => toggleTraceEventExpanded(item.traceId)}
+ endIcon={expandedTraceIds.includes(item.traceId)
? <ExpandLessOutlinedIcon fontSize="small" />
: <ExpandMoreOutlinedIcon fontSize="small" />}
sx={{ ml: "auto", minWidth: "auto", px: 0.5 }}
>
- {expandedToolIds.includes(item.toolCallId) ? translate("ai.drawer.collapseDetail") : translate("ai.drawer.viewDetail")}
+ {expandedTraceIds.includes(item.traceId) ? translate("ai.drawer.collapseDetail") : translate("ai.drawer.viewDetail")}
</Button>
)}
</Stack>
- <Collapse in={expandedToolIds.includes(item.toolCallId)} timeout="auto" unmountOnExit>
- {!!item.inputSummary && (
- <Typography variant="caption" display="block" sx={{ mt: 0.75, whiteSpace: "pre-wrap" }}>
- {translate("ai.drawer.toolInput", { value: item.inputSummary })}
- </Typography>
- )}
- {!!item.outputSummary && (
- <Typography variant="caption" display="block" sx={{ mt: 0.75, whiteSpace: "pre-wrap" }}>
- {translate("ai.drawer.toolOutput", { value: item.outputSummary })}
- </Typography>
- )}
- {!!item.errorMessage && (
- <Typography variant="caption" color="error.main" display="block" sx={{ mt: 0.75, whiteSpace: "pre-wrap" }}>
- {translate("ai.drawer.toolError", { value: item.errorMessage })}
- </Typography>
- )}
- </Collapse>
+ {item.traceType === "thinking" ? (
+ <Typography variant="caption" display="block" color="text.secondary" sx={{ mt: 0.75, whiteSpace: "pre-wrap" }}>
+ {item.content || translate("ai.drawer.thinkingEmpty")}
+ </Typography>
+ ) : (
+ <Collapse in={expandedTraceIds.includes(item.traceId)} timeout="auto" unmountOnExit>
+ {!!item.title && (
+ <Typography variant="caption" display="block" color="text.secondary" sx={{ mt: 0.75, whiteSpace: "pre-wrap" }}>
+ {item.title}
+ </Typography>
+ )}
+ {!!item.inputSummary && (
+ <Typography variant="caption" display="block" sx={{ mt: 0.75, whiteSpace: "pre-wrap" }}>
+ {translate("ai.drawer.toolInput", { value: item.inputSummary })}
+ </Typography>
+ )}
+ {!!item.outputSummary && (
+ <Typography variant="caption" display="block" sx={{ mt: 0.75, whiteSpace: "pre-wrap" }}>
+ {translate("ai.drawer.toolOutput", { value: item.outputSummary })}
+ </Typography>
+ )}
+ {!!item.errorMessage && (
+ <Typography variant="caption" color="error.main" display="block" sx={{ mt: 0.75, whiteSpace: "pre-wrap" }}>
+ {translate("ai.drawer.toolError", { value: item.errorMessage })}
+ </Typography>
+ )}
+ </Collapse>
+ )}
</Paper>
))}
</Stack>
@@ -1051,61 +1046,6 @@
justifyContent={message.role === "user" ? "flex-end" : "flex-start"}
>
<Stack spacing={1} sx={{ maxWidth: "85%", width: "100%" }} alignItems={message.role === "user" ? "flex-end" : "flex-start"}>
- {message.role === "assistant" && index === currentThinkingMessageIndex && !!thinkingEvents.length && (
- <Paper
- variant="outlined"
- sx={{
- width: "100%",
- borderRadius: 2,
- overflow: "hidden",
- bgcolor: "grey.50",
- }}
- >
- <Button
- fullWidth
- size="small"
- onClick={toggleThinkingExpanded}
- endIcon={thinkingExpanded
- ? <ExpandLessOutlinedIcon fontSize="small" />
- : <ExpandMoreOutlinedIcon fontSize="small" />}
- sx={{
- justifyContent: "space-between",
- px: 1.25,
- py: 0.75,
- color: "text.primary",
- }}
- >
- {thinkingExpanded ? translate("ai.drawer.thinkingCollapse") : translate("ai.drawer.thinkingExpand")}
- </Button>
- <Collapse in={thinkingExpanded} timeout="auto" unmountOnExit>
- <Stack spacing={1} sx={{ px: 1.25, pb: 1.25 }}>
- {thinkingEvents.map((item) => (
- <Paper key={item.phase} variant="outlined" sx={{ px: 1, py: 0.9, bgcolor: "common.white" }}>
- <Stack direction="row" spacing={1} alignItems="center" flexWrap="wrap" useFlexGap>
- <Typography variant="body2" fontWeight={700}>
- {item.title || translate("ai.drawer.thinkingProcess")}
- </Typography>
- <Chip
- size="small"
- color={item.status === "FAILED"
- ? "error"
- : item.status === "COMPLETED"
- ? "success"
- : item.status === "ABORTED"
- ? "warning"
- : "info"}
- label={getThinkingStatusLabel(item.status)}
- />
- </Stack>
- <Typography variant="caption" display="block" color="text.secondary" sx={{ mt: 0.75, whiteSpace: "pre-wrap" }}>
- {item.content || translate("ai.drawer.thinkingEmpty")}
- </Typography>
- </Paper>
- ))}
- </Stack>
- </Collapse>
- </Paper>
- )}
<Paper
elevation={0}
sx={{
diff --git a/rsf-server/src/main/java/com/vincent/rsf/server/ai/dto/AiChatThinkingEventDto.java b/rsf-server/src/main/java/com/vincent/rsf/server/ai/dto/AiChatThinkingEventDto.java
deleted file mode 100644
index 12183c7..0000000
--- a/rsf-server/src/main/java/com/vincent/rsf/server/ai/dto/AiChatThinkingEventDto.java
+++ /dev/null
@@ -1,25 +0,0 @@
-package com.vincent.rsf.server.ai.dto;
-
-import lombok.Builder;
-import lombok.Data;
-
-@Data
-@Builder
-public class AiChatThinkingEventDto {
-
- private String requestId;
-
- private Long sessionId;
-
- private String phase;
-
- private String status;
-
- private String title;
-
- private String content;
-
- private String toolCallId;
-
- private Long timestamp;
-}
diff --git a/rsf-server/src/main/java/com/vincent/rsf/server/ai/dto/AiChatToolEventDto.java b/rsf-server/src/main/java/com/vincent/rsf/server/ai/dto/AiChatTraceEventDto.java
similarity index 68%
rename from rsf-server/src/main/java/com/vincent/rsf/server/ai/dto/AiChatToolEventDto.java
rename to rsf-server/src/main/java/com/vincent/rsf/server/ai/dto/AiChatTraceEventDto.java
index dbeb6de..2273171 100644
--- a/rsf-server/src/main/java/com/vincent/rsf/server/ai/dto/AiChatToolEventDto.java
+++ b/rsf-server/src/main/java/com/vincent/rsf/server/ai/dto/AiChatTraceEventDto.java
@@ -5,19 +5,31 @@
@Data
@Builder
-public class AiChatToolEventDto {
+public class AiChatTraceEventDto {
private String requestId;
private Long sessionId;
+
+ private String traceId;
+
+ private Long sequence;
+
+ private String traceType;
+
+ private String phase;
+
+ private String status;
+
+ private String title;
+
+ private String content;
private String toolCallId;
private String toolName;
private String mountName;
-
- private String status;
private String inputSummary;
diff --git a/rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/chat/AiChatFailureHandler.java b/rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/chat/AiChatFailureHandler.java
index 9fa35c5..d9f9d91 100644
--- a/rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/chat/AiChatFailureHandler.java
+++ b/rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/chat/AiChatFailureHandler.java
@@ -28,13 +28,13 @@
public void handleStreamFailure(SseEmitter emitter, String requestId, Long sessionId, String model, long startedAt,
Long firstTokenAt, AiChatException exception, Long callLogId,
long toolSuccessCount, long toolFailureCount,
- AiThinkingTraceEmitter thinkingTraceEmitter,
+ AiChatTraceEmitter traceEmitter,
Long tenantId, Long userId, String promptCode) {
if (isClientAbortException(exception)) {
log.warn("AI chat aborted by client, requestId={}, sessionId={}, stage={}, message={}",
requestId, sessionId, exception.getStage(), exception.getMessage());
- if (thinkingTraceEmitter != null) {
- thinkingTraceEmitter.markTerminated("ABORTED");
+ if (traceEmitter != null) {
+ traceEmitter.markTerminated("ABORTED");
}
aiSseEventPublisher.emitSafely(emitter, "status",
aiSseEventPublisher.buildTerminalStatus(requestId, sessionId, "ABORTED", model, startedAt, firstTokenAt));
@@ -55,8 +55,8 @@
}
log.error("AI chat failed, requestId={}, sessionId={}, category={}, stage={}, message={}",
requestId, sessionId, exception.getCategory(), exception.getStage(), exception.getMessage(), exception);
- if (thinkingTraceEmitter != null) {
- thinkingTraceEmitter.markTerminated("FAILED");
+ if (traceEmitter != null) {
+ traceEmitter.markTerminated("FAILED");
}
aiSseEventPublisher.emitSafely(emitter, "status",
aiSseEventPublisher.buildTerminalStatus(requestId, sessionId, "FAILED", model, startedAt, firstTokenAt));
diff --git a/rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/chat/AiChatOrchestrator.java b/rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/chat/AiChatOrchestrator.java
index 055192b..8b6b510 100644
--- a/rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/chat/AiChatOrchestrator.java
+++ b/rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/chat/AiChatOrchestrator.java
@@ -57,6 +57,7 @@
String requestId = request.getRequestId();
long startedAt = System.currentTimeMillis();
AtomicReference<Long> firstTokenAtRef = new AtomicReference<>();
+ AtomicLong traceSequence = new AtomicLong(0);
AtomicLong toolCallSequence = new AtomicLong(0);
AtomicLong toolSuccessCount = new AtomicLong(0);
AtomicLong toolFailureCount = new AtomicLong(0);
@@ -64,7 +65,7 @@
Long callLogId = null;
String model = null;
String resolvedPromptCode = request.getPromptCode();
- AiThinkingTraceEmitter thinkingTraceEmitter = null;
+ AiChatTraceEmitter traceEmitter = null;
try {
ensureIdentity(userId, tenantId);
AiResolvedConfig config = resolveConfig(request, tenantId);
@@ -115,13 +116,13 @@
.build());
log.info("AI chat started, requestId={}, userId={}, tenantId={}, sessionId={}, model={}",
requestId, userId, tenantId, session.getId(), resolvedModel);
- thinkingTraceEmitter = new AiThinkingTraceEmitter(aiSseEventPublisher, emitter, requestId, session.getId());
- thinkingTraceEmitter.startAnalyze();
- AiThinkingTraceEmitter activeThinkingTraceEmitter = thinkingTraceEmitter;
+ traceEmitter = new AiChatTraceEmitter(aiSseEventPublisher, emitter, requestId, session.getId(), traceSequence);
+ traceEmitter.startAnalyze();
+ AiChatTraceEmitter activeTraceEmitter = traceEmitter;
ToolCallback[] observableToolCallbacks = aiToolObservationService.wrapToolCallbacks(
- runtime.getToolCallbacks(), emitter, requestId, session.getId(), toolCallSequence,
- toolSuccessCount, toolFailureCount, callLogId, userId, tenantId, activeThinkingTraceEmitter
+ runtime.getToolCallbacks(), requestId, session.getId(), toolCallSequence,
+ toolSuccessCount, toolFailureCount, callLogId, userId, tenantId, activeTraceEmitter
);
Prompt prompt = new Prompt(
aiPromptMessageBuilder.buildPromptMessages(memory, mergedMessages, config.getPrompt(), request.getMetadata()),
@@ -134,10 +135,10 @@
String content = extractContent(response);
aiChatMemoryService.saveRound(session, userId, tenantId, request.getMessages(), content);
if (StringUtils.hasText(content)) {
- aiSseEventPublisher.markFirstToken(firstTokenAtRef, emitter, requestId, session.getId(), resolvedModel, startedAt, activeThinkingTraceEmitter);
+ aiSseEventPublisher.markFirstToken(firstTokenAtRef, emitter, requestId, session.getId(), resolvedModel, startedAt, activeTraceEmitter);
aiSseEventPublisher.emitStrict(emitter, "delta", aiSseEventPublisher.buildMessagePayload("requestId", requestId, "content", content));
}
- activeThinkingTraceEmitter.completeCurrentPhase();
+ activeTraceEmitter.completeCurrentPhase();
aiSseEventPublisher.emitDone(emitter, requestId, response.getMetadata(), config.getAiParam().getModel(),
session.getId(), startedAt, firstTokenAtRef.get());
aiSseEventPublisher.emitSafely(emitter, "status",
@@ -169,7 +170,7 @@
lastMetadata.set(response.getMetadata());
String content = extractContent(response);
if (StringUtils.hasText(content)) {
- aiSseEventPublisher.markFirstToken(firstTokenAtRef, emitter, requestId, session.getId(), resolvedModel, startedAt, activeThinkingTraceEmitter);
+ aiSseEventPublisher.markFirstToken(firstTokenAtRef, emitter, requestId, session.getId(), resolvedModel, startedAt, activeTraceEmitter);
assistantContent.append(content);
aiSseEventPublisher.emitStrict(emitter, "delta",
aiSseEventPublisher.buildMessagePayload("requestId", requestId, "content", content));
@@ -181,7 +182,7 @@
e == null ? "AI 妯″瀷娴佸紡璋冪敤澶辫触" : e.getMessage(), e);
}
aiChatMemoryService.saveRound(session, userId, tenantId, request.getMessages(), assistantContent.toString());
- activeThinkingTraceEmitter.completeCurrentPhase();
+ activeTraceEmitter.completeCurrentPhase();
aiSseEventPublisher.emitDone(emitter, requestId, lastMetadata.get(), config.getAiParam().getModel(),
session.getId(), startedAt, firstTokenAtRef.get());
aiSseEventPublisher.emitSafely(emitter, "status",
@@ -205,13 +206,13 @@
}
} catch (AiChatException e) {
aiChatFailureHandler.handleStreamFailure(emitter, requestId, sessionId, model, startedAt, firstTokenAtRef.get(), e,
- callLogId, toolSuccessCount.get(), toolFailureCount.get(), thinkingTraceEmitter,
+ callLogId, toolSuccessCount.get(), toolFailureCount.get(), traceEmitter,
tenantId, userId, resolvedPromptCode);
} catch (Exception e) {
aiChatFailureHandler.handleStreamFailure(emitter, requestId, sessionId, model, startedAt, firstTokenAtRef.get(),
aiChatFailureHandler.buildAiException("AI_INTERNAL_ERROR", AiErrorCategory.INTERNAL, "INTERNAL",
e == null ? "AI 瀵硅瘽澶辫触" : e.getMessage(), e),
- callLogId, toolSuccessCount.get(), toolFailureCount.get(), thinkingTraceEmitter,
+ callLogId, toolSuccessCount.get(), toolFailureCount.get(), traceEmitter,
tenantId, userId, resolvedPromptCode);
} finally {
log.debug("AI chat stream finished, requestId={}", requestId);
diff --git a/rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/chat/AiChatTraceEmitter.java b/rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/chat/AiChatTraceEmitter.java
new file mode 100644
index 0000000..ccedff3
--- /dev/null
+++ b/rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/chat/AiChatTraceEmitter.java
@@ -0,0 +1,179 @@
+package com.vincent.rsf.server.ai.service.impl.chat;
+
+import com.vincent.rsf.server.ai.dto.AiChatTraceEventDto;
+import org.springframework.util.StringUtils;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+import java.time.Instant;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+public class AiChatTraceEmitter {
+
+ private static final String TRACE_EVENT_NAME = "trace";
+
+ private final AiSseEventPublisher aiSseEventPublisher;
+ private final SseEmitter emitter;
+ private final String requestId;
+ private final Long sessionId;
+ private final AtomicLong traceSequence;
+ private final Map<String, Long> traceOrderMap = new ConcurrentHashMap<>();
+ private String currentPhase;
+ private String currentStatus;
+
+ public AiChatTraceEmitter(AiSseEventPublisher aiSseEventPublisher, SseEmitter emitter, String requestId,
+ Long sessionId, AtomicLong traceSequence) {
+ this.aiSseEventPublisher = aiSseEventPublisher;
+ this.emitter = emitter;
+ this.requestId = requestId;
+ this.sessionId = sessionId;
+ this.traceSequence = traceSequence;
+ }
+
+ public void startAnalyze() {
+ if (currentPhase != null) {
+ return;
+ }
+ currentPhase = "ANALYZE";
+ currentStatus = "STARTED";
+ emitThinkingEvent("ANALYZE", "STARTED", "姝ｅ湪鍒嗘瀽闂",
+ "宸叉帴鏀朵綘鐨勯棶棰橈紝姝ｅ湪鐞嗚В鎰忓浘骞跺垽鏂槸鍚﹂渶瑕佽皟鐢ㄥ伐鍏枫��", null);
+ }
+
+ public void onToolStart(String toolName, String mountName, String toolCallId, String inputSummary, long timestamp) {
+ completeCurrentPhase();
+ emitToolEvent("STARTED", "寮�濮嬭皟鐢ㄥ伐鍏�", null, toolCallId, toolName, mountName, inputSummary,
+ null, null, null, timestamp);
+ }
+
+ public void onToolResult(String toolName, String mountName, String toolCallId, String inputSummary,
+ String outputSummary, String errorMessage, Long durationMs, long timestamp,
+ boolean failed) {
+ emitToolEvent(failed ? "FAILED" : "COMPLETED",
+ failed ? "宸ュ叿璋冪敤澶辫触" : "宸ュ叿璋冪敤瀹屾垚",
+ null,
+ toolCallId,
+ toolName,
+ mountName,
+ inputSummary,
+ outputSummary,
+ errorMessage,
+ durationMs,
+ timestamp);
+ }
+
+ public void startAnswer() {
+ switchPhase("ANSWER", "STARTED", "姝ｅ湪鏁寸悊绛旀", "宸插畬鎴愬垎鏋愶紝姝ｅ湪缁勭粐鏈�缁堝洖澶嶅唴瀹广��", null);
+ }
+
+ public void completeCurrentPhase() {
+ if (!StringUtils.hasText(currentPhase) || isTerminalStatus(currentStatus)) {
+ return;
+ }
+ currentStatus = "COMPLETED";
+ emitThinkingEvent(currentPhase, "COMPLETED", resolveCompleteTitle(currentPhase),
+ resolveCompleteContent(currentPhase), null);
+ }
+
+ public void markTerminated(String terminalStatus) {
+ if (!StringUtils.hasText(currentPhase) || isTerminalStatus(currentStatus)) {
+ return;
+ }
+ currentStatus = terminalStatus;
+ emitThinkingEvent(currentPhase, terminalStatus,
+ "ABORTED".equals(terminalStatus) ? "鎬濊�冨凡涓" : "鎬濊�冨け璐�",
+ "ABORTED".equals(terminalStatus)
+ ? "鏈疆瀵硅瘽宸茶涓锛屾�濊�冭繃绋嬫彁鍓嶇粨鏉熴��"
+ : "鏈疆瀵硅瘽鍦ㄧ敓鎴愮瓟妗堝墠澶辫触锛屽綋鍓嶆�濊�冭繃绋嬪凡鍋滄銆�",
+ null);
+ }
+
+ private void switchPhase(String nextPhase, String nextStatus, String title, String content, String toolCallId) {
+ if (!Objects.equals(currentPhase, nextPhase)) {
+ completeCurrentPhase();
+ }
+ currentPhase = nextPhase;
+ currentStatus = nextStatus;
+ emitThinkingEvent(nextPhase, nextStatus, title, content, toolCallId);
+ }
+
+ private void emitThinkingEvent(String phase, String status, String title, String content, String toolCallId) {
+ emitTraceEvent(AiChatTraceEventDto.builder()
+ .requestId(requestId)
+ .sessionId(sessionId)
+ .traceType("thinking")
+ .phase(phase)
+ .status(status)
+ .title(title)
+ .content(content)
+ .toolCallId(toolCallId)
+ .timestamp(Instant.now().toEpochMilli())
+ .build(), buildThinkingTraceId(phase));
+ }
+
+ private void emitToolEvent(String status, String title, String content, String toolCallId, String toolName,
+ String mountName, String inputSummary, String outputSummary, String errorMessage,
+ Long durationMs, long timestamp) {
+ emitTraceEvent(AiChatTraceEventDto.builder()
+ .requestId(requestId)
+ .sessionId(sessionId)
+ .traceType("tool")
+ .status(status)
+ .title(title)
+ .content(content)
+ .toolCallId(toolCallId)
+ .toolName(toolName)
+ .mountName(mountName)
+ .inputSummary(inputSummary)
+ .outputSummary(outputSummary)
+ .errorMessage(errorMessage)
+ .durationMs(durationMs)
+ .timestamp(timestamp)
+ .build(), buildToolTraceId(toolCallId));
+ }
+
+ private void emitTraceEvent(AiChatTraceEventDto payload, String traceId) {
+ long sequence = traceOrderMap.computeIfAbsent(traceId, ignored -> traceSequence.incrementAndGet());
+ payload.setSequence(sequence);
+ payload.setTraceId(traceId);
+ aiSseEventPublisher.emitSafely(emitter, TRACE_EVENT_NAME, payload);
+ }
+
+ private String buildThinkingTraceId(String phase) {
+ return requestId + "-thinking-" + phase;
+ }
+
+ private String buildToolTraceId(String toolCallId) {
+ return toolCallId;
+ }
+
+ private boolean isTerminalStatus(String status) {
+ return "COMPLETED".equals(status) || "FAILED".equals(status) || "ABORTED".equals(status);
+ }
+
+ private String resolveCompleteTitle(String phase) {
+ if ("ANSWER".equals(phase)) {
+ return "绛旀鏁寸悊瀹屾垚";
+ }
+ if ("TOOL_CALL".equals(phase)) {
+ return "宸ュ叿鍒嗘瀽瀹屾垚";
+ }
+ return "闂鍒嗘瀽瀹屾垚";
+ }
+
+ private String resolveCompleteContent(String phase) {
+ if ("ANSWER".equals(phase)) {
+ return "鏈�缁堢瓟澶嶅凡鐢熸垚瀹屾垚銆�";
+ }
+ if ("TOOL_CALL".equals(phase)) {
+ return "宸ュ叿璋冪敤闃舵宸茬粨鏉燂紝鐩稿叧淇℃伅宸叉暣鐞嗗畬姣曘��";
+ }
+ return "闂鎰忓浘鍜屽鐞嗘柟鍚戝凡鍒嗘瀽瀹屾垚銆�";
+ }
+
+ private String safeLabel(String value, String fallback) {
+ return StringUtils.hasText(value) ? value : fallback;
+ }
+}
diff --git a/rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/chat/AiSseEventPublisher.java b/rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/chat/AiSseEventPublisher.java
index aace1a3..0590643 100644
--- a/rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/chat/AiSseEventPublisher.java
+++ b/rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/chat/AiSseEventPublisher.java
@@ -28,12 +28,12 @@
private final ObjectMapper objectMapper;
public void markFirstToken(AtomicReference<Long> firstTokenAtRef, SseEmitter emitter, String requestId,
- Long sessionId, String model, long startedAt, AiThinkingTraceEmitter thinkingTraceEmitter) {
+ Long sessionId, String model, long startedAt, AiChatTraceEmitter traceEmitter) {
if (!firstTokenAtRef.compareAndSet(null, System.currentTimeMillis())) {
return;
}
- if (thinkingTraceEmitter != null) {
- thinkingTraceEmitter.startAnswer();
+ if (traceEmitter != null) {
+ traceEmitter.startAnswer();
}
emitSafely(emitter, "status", AiChatStatusDto.builder()
.requestId(requestId)
diff --git a/rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/chat/AiThinkingTraceEmitter.java b/rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/chat/AiThinkingTraceEmitter.java
deleted file mode 100644
index 206ad51..0000000
--- a/rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/chat/AiThinkingTraceEmitter.java
+++ /dev/null
@@ -1,129 +0,0 @@
-package com.vincent.rsf.server.ai.service.impl.chat;
-
-import com.vincent.rsf.server.ai.dto.AiChatThinkingEventDto;
-import org.springframework.util.StringUtils;
-import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
-
-import java.time.Instant;
-import java.util.Objects;
-
-public class AiThinkingTraceEmitter {
-
- private final AiSseEventPublisher aiSseEventPublisher;
- private final SseEmitter emitter;
- private final String requestId;
- private final Long sessionId;
- private String currentPhase;
- private String currentStatus;
-
- public AiThinkingTraceEmitter(AiSseEventPublisher aiSseEventPublisher, SseEmitter emitter, String requestId, Long sessionId) {
- this.aiSseEventPublisher = aiSseEventPublisher;
- this.emitter = emitter;
- this.requestId = requestId;
- this.sessionId = sessionId;
- }
-
- public void startAnalyze() {
- if (currentPhase != null) {
- return;
- }
- currentPhase = "ANALYZE";
- currentStatus = "STARTED";
- emitThinkingEvent("ANALYZE", "STARTED", "姝ｅ湪鍒嗘瀽闂",
- "宸叉帴鏀朵綘鐨勯棶棰橈紝姝ｅ湪鐞嗚В鎰忓浘骞跺垽鏂槸鍚﹂渶瑕佽皟鐢ㄥ伐鍏枫��", null);
- }
-
- public void onToolStart(String toolName, String toolCallId) {
- switchPhase("TOOL_CALL", "STARTED", "姝ｅ湪璋冪敤宸ュ叿", "宸插垽鏂渶瑕佽皟鐢ㄥ伐鍏凤紝姝ｅ湪鏌ヨ鐩稿叧淇℃伅銆�", null);
- currentStatus = "UPDATED";
- emitThinkingEvent("TOOL_CALL", "UPDATED", "姝ｅ湪璋冪敤宸ュ叿",
- "姝ｅ湪璋冪敤宸ュ叿 " + safeLabel(toolName, "鏈煡宸ュ叿") + " 鑾峰彇鎵�闇�淇℃伅銆�", toolCallId);
- }
-
- public void onToolResult(String toolName, String toolCallId, boolean failed) {
- currentPhase = "TOOL_CALL";
- currentStatus = failed ? "FAILED" : "UPDATED";
- emitThinkingEvent("TOOL_CALL", failed ? "FAILED" : "UPDATED",
- failed ? "宸ュ叿璋冪敤澶辫触" : "宸ュ叿璋冪敤瀹屾垚",
- failed
- ? "宸ュ叿 " + safeLabel(toolName, "鏈煡宸ュ叿") + " 璋冪敤澶辫触锛屾鍦ㄨ瘎浼板け璐ュ奖鍝嶅苟鏁寸悊鍙敤淇℃伅銆�"
- : "宸ュ叿 " + safeLabel(toolName, "鏈煡宸ュ叿") + " 宸茶繑鍥炵粨鏋滐紝姝ｅ湪缁х画鍒嗘瀽骞舵彁鐐煎叧閿俊鎭��",
- toolCallId);
- }
-
- public void startAnswer() {
- switchPhase("ANSWER", "STARTED", "姝ｅ湪鏁寸悊绛旀", "宸插畬鎴愬垎鏋愶紝姝ｅ湪缁勭粐鏈�缁堝洖澶嶅唴瀹广��", null);
- }
-
- public void completeCurrentPhase() {
- if (!StringUtils.hasText(currentPhase) || isTerminalStatus(currentStatus)) {
- return;
- }
- currentStatus = "COMPLETED";
- emitThinkingEvent(currentPhase, "COMPLETED", resolveCompleteTitle(currentPhase),
- resolveCompleteContent(currentPhase), null);
- }
-
- public void markTerminated(String terminalStatus) {
- if (!StringUtils.hasText(currentPhase) || isTerminalStatus(currentStatus)) {
- return;
- }
- currentStatus = terminalStatus;
- emitThinkingEvent(currentPhase, terminalStatus,
- "ABORTED".equals(terminalStatus) ? "鎬濊�冨凡涓" : "鎬濊�冨け璐�",
- "ABORTED".equals(terminalStatus)
- ? "鏈疆瀵硅瘽宸茶涓锛屾�濊�冭繃绋嬫彁鍓嶇粨鏉熴��"
- : "鏈疆瀵硅瘽鍦ㄧ敓鎴愮瓟妗堝墠澶辫触锛屽綋鍓嶆�濊�冭繃绋嬪凡鍋滄銆�",
- null);
- }
-
- private void switchPhase(String nextPhase, String nextStatus, String title, String content, String toolCallId) {
- if (!Objects.equals(currentPhase, nextPhase)) {
- completeCurrentPhase();
- }
- currentPhase = nextPhase;
- currentStatus = nextStatus;
- emitThinkingEvent(nextPhase, nextStatus, title, content, toolCallId);
- }
-
- private void emitThinkingEvent(String phase, String status, String title, String content, String toolCallId) {
- aiSseEventPublisher.emitSafely(emitter, "thinking", AiChatThinkingEventDto.builder()
- .requestId(requestId)
- .sessionId(sessionId)
- .phase(phase)
- .status(status)
- .title(title)
- .content(content)
- .toolCallId(toolCallId)
- .timestamp(Instant.now().toEpochMilli())
- .build());
- }
-
- private boolean isTerminalStatus(String status) {
- return "COMPLETED".equals(status) || "FAILED".equals(status) || "ABORTED".equals(status);
- }
-
- private String resolveCompleteTitle(String phase) {
- if ("ANSWER".equals(phase)) {
- return "绛旀鏁寸悊瀹屾垚";
- }
- if ("TOOL_CALL".equals(phase)) {
- return "宸ュ叿鍒嗘瀽瀹屾垚";
- }
- return "闂鍒嗘瀽瀹屾垚";
- }
-
- private String resolveCompleteContent(String phase) {
- if ("ANSWER".equals(phase)) {
- return "鏈�缁堢瓟澶嶅凡鐢熸垚瀹屾垚銆�";
- }
- if ("TOOL_CALL".equals(phase)) {
- return "宸ュ叿璋冪敤闃舵宸茬粨鏉燂紝鐩稿叧淇℃伅宸叉暣鐞嗗畬姣曘��";
- }
- return "闂鎰忓浘鍜屽鐞嗘柟鍚戝凡鍒嗘瀽瀹屾垚銆�";
- }
-
- private String safeLabel(String value, String fallback) {
- return StringUtils.hasText(value) ? value : fallback;
- }
-}
diff --git a/rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/chat/AiToolObservationService.java b/rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/chat/AiToolObservationService.java
index 82d2019..72b6385 100644
--- a/rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/chat/AiToolObservationService.java
+++ b/rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/chat/AiToolObservationService.java
@@ -1,7 +1,6 @@
package com.vincent.rsf.server.ai.service.impl.chat;
import com.vincent.rsf.framework.exception.CoolException;
-import com.vincent.rsf.server.ai.dto.AiChatToolEventDto;
import com.vincent.rsf.server.ai.service.AiCallLogService;
import com.vincent.rsf.server.ai.service.MountedToolCallback;
import com.vincent.rsf.server.ai.store.AiCachedToolResult;
@@ -11,7 +10,6 @@
import org.springframework.ai.tool.ToolCallback;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
-import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.util.ArrayList;
import java.util.List;
@@ -21,15 +19,14 @@
@RequiredArgsConstructor
public class AiToolObservationService {
- private final AiSseEventPublisher aiSseEventPublisher;
private final AiToolResultStore aiToolResultStore;
private final AiCallLogService aiCallLogService;
- public ToolCallback[] wrapToolCallbacks(ToolCallback[] toolCallbacks, SseEmitter emitter, String requestId,
+ public ToolCallback[] wrapToolCallbacks(ToolCallback[] toolCallbacks, String requestId,
Long sessionId, AtomicLong toolCallSequence,
AtomicLong toolSuccessCount, AtomicLong toolFailureCount,
Long callLogId, Long userId, Long tenantId,
- AiThinkingTraceEmitter thinkingTraceEmitter) {
+ AiChatTraceEmitter traceEmitter) {
if (toolCallbacks == null || toolCallbacks.length == 0) {
return toolCallbacks;
}
@@ -38,8 +35,8 @@
if (callback == null) {
continue;
}
- wrappedCallbacks.add(new ObservableToolCallback(callback, emitter, requestId, sessionId, toolCallSequence,
- toolSuccessCount, toolFailureCount, callLogId, userId, tenantId, thinkingTraceEmitter));
+ wrappedCallbacks.add(new ObservableToolCallback(callback, requestId, sessionId, toolCallSequence,
+ toolSuccessCount, toolFailureCount, callLogId, userId, tenantId, traceEmitter));
}
return wrappedCallbacks.toArray(new ToolCallback[0]);
}
@@ -58,7 +55,6 @@
private class ObservableToolCallback implements ToolCallback {
private final ToolCallback delegate;
- private final SseEmitter emitter;
private final String requestId;
private final Long sessionId;
private final AtomicLong toolCallSequence;
@@ -67,15 +63,14 @@
private final Long callLogId;
private final Long userId;
private final Long tenantId;
- private final AiThinkingTraceEmitter thinkingTraceEmitter;
+ private final AiChatTraceEmitter traceEmitter;
- private ObservableToolCallback(ToolCallback delegate, SseEmitter emitter, String requestId,
+ private ObservableToolCallback(ToolCallback delegate, String requestId,
Long sessionId, AtomicLong toolCallSequence,
AtomicLong toolSuccessCount, AtomicLong toolFailureCount,
Long callLogId, Long userId, Long tenantId,
- AiThinkingTraceEmitter thinkingTraceEmitter) {
+ AiChatTraceEmitter traceEmitter) {
this.delegate = delegate;
- this.emitter = emitter;
this.requestId = requestId;
this.sessionId = sessionId;
this.toolCallSequence = toolCallSequence;
@@ -84,7 +79,7 @@
this.callLogId = callLogId;
this.userId = userId;
this.tenantId = tenantId;
- this.thinkingTraceEmitter = thinkingTraceEmitter;
+ this.traceEmitter = traceEmitter;
}
@Override
@@ -108,95 +103,56 @@
String mountName = delegate instanceof MountedToolCallback ? ((MountedToolCallback) delegate).getMountName() : null;
String toolCallId = requestId + "-tool-" + toolCallSequence.incrementAndGet();
long startedAt = System.currentTimeMillis();
+ String inputSummary = summarizeToolPayload(toolInput, 400);
AiCachedToolResult cachedToolResult = aiToolResultStore.getToolResult(tenantId, requestId, toolName, toolInput);
if (cachedToolResult != null) {
- aiSseEventPublisher.emitSafely(emitter, "tool_result", AiChatToolEventDto.builder()
- .requestId(requestId)
- .sessionId(sessionId)
- .toolCallId(toolCallId)
- .toolName(toolName)
- .mountName(mountName)
- .status(cachedToolResult.isSuccess() ? "COMPLETED" : "FAILED")
- .inputSummary(summarizeToolPayload(toolInput, 400))
- .outputSummary(summarizeToolPayload(cachedToolResult.getOutput(), 600))
- .errorMessage(cachedToolResult.getErrorMessage())
- .durationMs(0L)
- .timestamp(System.currentTimeMillis())
- .build());
- if (thinkingTraceEmitter != null) {
- thinkingTraceEmitter.onToolResult(toolName, toolCallId, !cachedToolResult.isSuccess());
+ String outputSummary = summarizeToolPayload(cachedToolResult.getOutput(), 600);
+ String errorMessage = cachedToolResult.getErrorMessage();
+ if (traceEmitter != null) {
+ traceEmitter.onToolResult(toolName, mountName, toolCallId, inputSummary, outputSummary,
+ errorMessage, 0L, System.currentTimeMillis(), !cachedToolResult.isSuccess());
}
if (cachedToolResult.isSuccess()) {
toolSuccessCount.incrementAndGet();
aiCallLogService.saveMcpCallLog(callLogId, requestId, sessionId, toolCallId, mountName, toolName,
- "COMPLETED", summarizeToolPayload(toolInput, 400), summarizeToolPayload(cachedToolResult.getOutput(), 600),
+ "COMPLETED", inputSummary, outputSummary,
null, 0L, userId, tenantId);
return cachedToolResult.getOutput();
}
toolFailureCount.incrementAndGet();
aiCallLogService.saveMcpCallLog(callLogId, requestId, sessionId, toolCallId, mountName, toolName,
- "FAILED", summarizeToolPayload(toolInput, 400), null, cachedToolResult.getErrorMessage(),
+ "FAILED", inputSummary, null, errorMessage,
0L, userId, tenantId);
- throw new CoolException(cachedToolResult.getErrorMessage());
+ throw new CoolException(errorMessage);
}
- if (thinkingTraceEmitter != null) {
- thinkingTraceEmitter.onToolStart(toolName, toolCallId);
+ if (traceEmitter != null) {
+ traceEmitter.onToolStart(toolName, mountName, toolCallId, inputSummary, startedAt);
}
- aiSseEventPublisher.emitSafely(emitter, "tool_start", AiChatToolEventDto.builder()
- .requestId(requestId)
- .sessionId(sessionId)
- .toolCallId(toolCallId)
- .toolName(toolName)
- .mountName(mountName)
- .status("STARTED")
- .inputSummary(summarizeToolPayload(toolInput, 400))
- .timestamp(startedAt)
- .build());
try {
String output = toolContext == null ? delegate.call(toolInput) : delegate.call(toolInput, toolContext);
long durationMs = System.currentTimeMillis() - startedAt;
- aiSseEventPublisher.emitSafely(emitter, "tool_result", AiChatToolEventDto.builder()
- .requestId(requestId)
- .sessionId(sessionId)
- .toolCallId(toolCallId)
- .toolName(toolName)
- .mountName(mountName)
- .status("COMPLETED")
- .inputSummary(summarizeToolPayload(toolInput, 400))
- .outputSummary(summarizeToolPayload(output, 600))
- .durationMs(durationMs)
- .timestamp(System.currentTimeMillis())
- .build());
- if (thinkingTraceEmitter != null) {
- thinkingTraceEmitter.onToolResult(toolName, toolCallId, false);
+ String outputSummary = summarizeToolPayload(output, 600);
+ if (traceEmitter != null) {
+ traceEmitter.onToolResult(toolName, mountName, toolCallId, inputSummary, outputSummary,
+ null, durationMs, System.currentTimeMillis(), false);
}
aiToolResultStore.cacheToolResult(tenantId, requestId, toolName, toolInput, true, output, null);
toolSuccessCount.incrementAndGet();
aiCallLogService.saveMcpCallLog(callLogId, requestId, sessionId, toolCallId, mountName, toolName,
- "COMPLETED", summarizeToolPayload(toolInput, 400), summarizeToolPayload(output, 600),
+ "COMPLETED", inputSummary, outputSummary,
null, durationMs, userId, tenantId);
return output;
} catch (RuntimeException e) {
long durationMs = System.currentTimeMillis() - startedAt;
- aiSseEventPublisher.emitSafely(emitter, "tool_error", AiChatToolEventDto.builder()
- .requestId(requestId)
- .sessionId(sessionId)
- .toolCallId(toolCallId)
- .toolName(toolName)
- .mountName(mountName)
- .status("FAILED")
- .inputSummary(summarizeToolPayload(toolInput, 400))
- .errorMessage(e.getMessage())
- .durationMs(durationMs)
- .timestamp(System.currentTimeMillis())
- .build());
- if (thinkingTraceEmitter != null) {
- thinkingTraceEmitter.onToolResult(toolName, toolCallId, true);
+ String errorMessage = e.getMessage();
+ if (traceEmitter != null) {
+ traceEmitter.onToolResult(toolName, mountName, toolCallId, inputSummary, null,
+ errorMessage, durationMs, System.currentTimeMillis(), true);
}
- aiToolResultStore.cacheToolResult(tenantId, requestId, toolName, toolInput, false, null, e.getMessage());
+ aiToolResultStore.cacheToolResult(tenantId, requestId, toolName, toolInput, false, null, errorMessage);
toolFailureCount.incrementAndGet();
aiCallLogService.saveMcpCallLog(callLogId, requestId, sessionId, toolCallId, mountName, toolName,
- "FAILED", summarizeToolPayload(toolInput, 400), null, e.getMessage(),
+ "FAILED", inputSummary, null, errorMessage,
durationMs, userId, tenantId);
throw e;
}
diff --git a/rsf-server/src/test/java/com/vincent/rsf/server/AI/service/impl/chat/AiChatTraceEmitterTest.java b/rsf-server/src/test/java/com/vincent/rsf/server/AI/service/impl/chat/AiChatTraceEmitterTest.java
new file mode 100644
index 0000000..53b3dfe
--- /dev/null
+++ b/rsf-server/src/test/java/com/vincent/rsf/server/AI/service/impl/chat/AiChatTraceEmitterTest.java
@@ -0,0 +1,59 @@
+package com.vincent.rsf.server.ai.service.impl.chat;
+
+import com.vincent.rsf.server.ai.dto.AiChatTraceEventDto;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+@ExtendWith(MockitoExtension.class)
+class AiChatTraceEmitterTest {
+
+ @Mock
+ private AiSseEventPublisher aiSseEventPublisher;
+
+ @Captor
+ private ArgumentCaptor<AiChatTraceEventDto> payloadCaptor;
+
+ @Test
+ void shouldReuseTraceIdentityForLogicalTraceCards() {
+ AiChatTraceEmitter traceEmitter = new AiChatTraceEmitter(
+ aiSseEventPublisher,
+ new SseEmitter(1000L),
+ "req-1",
+ 11L,
+ new AtomicLong(0)
+ );
+
+ traceEmitter.startAnalyze();
+ traceEmitter.onToolStart("inventory.lookup", "builtin-stock", "tool-1", "{\"code\":\"A01\"}", 100L);
+ traceEmitter.onToolResult("inventory.lookup", "builtin-stock", "tool-1", "{\"code\":\"A01\"}",
+ "{\"stock\":12}", null, 64L, 164L, false);
+
+ verify(aiSseEventPublisher, times(4)).emitSafely(any(), eq("trace"), payloadCaptor.capture());
+
+ List<AiChatTraceEventDto> payloads = payloadCaptor.getAllValues();
+ assertThat(payloads).extracting(AiChatTraceEventDto::getSequence)
+ .containsExactly(1L, 1L, 2L, 2L);
+ assertThat(payloads).extracting(AiChatTraceEventDto::getTraceId)
+ .containsExactly("req-1-thinking-ANALYZE", "req-1-thinking-ANALYZE", "tool-1", "tool-1");
+ assertThat(payloads).extracting(AiChatTraceEventDto::getTraceType)
+ .containsExactly("thinking", "thinking", "tool", "tool");
+ assertThat(payloads.get(1).getStatus()).isEqualTo("COMPLETED");
+ assertThat(payloads.get(3).getStatus()).isEqualTo("COMPLETED");
+ assertThat(payloads.get(3).getToolCallId()).isEqualTo("tool-1");
+ assertThat(payloads.get(3).getInputSummary()).isEqualTo("{\"code\":\"A01\"}");
+ }
+}
diff --git a/rsf-server/src/test/java/com/vincent/rsf/server/AI/service/impl/chat/AiToolObservationServiceTest.java b/rsf-server/src/test/java/com/vincent/rsf/server/AI/service/impl/chat/AiToolObservationServiceTest.java
new file mode 100644
index 0000000..5c0f320
--- /dev/null
+++ b/rsf-server/src/test/java/com/vincent/rsf/server/AI/service/impl/chat/AiToolObservationServiceTest.java
@@ -0,0 +1,70 @@
+package com.vincent.rsf.server.ai.service.impl.chat;
+
+import com.vincent.rsf.server.ai.service.AiCallLogService;
+import com.vincent.rsf.server.ai.service.MountedToolCallback;
+import com.vincent.rsf.server.ai.store.AiCachedToolResult;
+import com.vincent.rsf.server.ai.store.AiToolResultStore;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.ai.tool.ToolCallback;
+import org.springframework.ai.tool.definition.ToolDefinition;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class AiToolObservationServiceTest {
+
+ @Mock
+ private AiToolResultStore aiToolResultStore;
+ @Mock
+ private AiCallLogService aiCallLogService;
+ @Mock
+ private AiChatTraceEmitter traceEmitter;
+ @Mock
+ private MountedToolCallback mountedToolCallback;
+ @Mock
+ private ToolDefinition toolDefinition;
+
+ @Test
+ void shouldSkipStartTraceWhenToolResultComesFromCache() {
+ AiToolObservationService aiToolObservationService = new AiToolObservationService(aiToolResultStore, aiCallLogService);
+ when(mountedToolCallback.getToolDefinition()).thenReturn(toolDefinition);
+ when(toolDefinition.name()).thenReturn("inventory.lookup");
+ when(mountedToolCallback.getMountName()).thenReturn("builtin-stock");
+ when(aiToolResultStore.getToolResult(1L, "req-1", "inventory.lookup", "{\"code\":\"A01\"}"))
+ .thenReturn(AiCachedToolResult.builder()
+ .success(true)
+ .output("{\"stock\":12}")
+ .build());
+
+ ToolCallback[] callbacks = aiToolObservationService.wrapToolCallbacks(
+ new ToolCallback[]{mountedToolCallback},
+ "req-1",
+ 11L,
+ new AtomicLong(0),
+ new AtomicLong(0),
+ new AtomicLong(0),
+ 21L,
+ 31L,
+ 1L,
+ traceEmitter
+ );
+
+ String output = callbacks[0].call("{\"code\":\"A01\"}");
+
+ assertThat(output).isEqualTo("{\"stock\":12}");
+ verify(traceEmitter, never()).onToolStart(eq("inventory.lookup"), eq("builtin-stock"), eq("req-1-tool-1"), eq("{\"code\":\"A01\"}"), anyLong());
+ verify(traceEmitter).onToolResult(eq("inventory.lookup"), eq("builtin-stock"), eq("req-1-tool-1"),
+ eq("{\"code\":\"A01\"}"), eq("{\"stock\":12}"), eq(null), eq(0L), anyLong(), eq(false));
+ verify(mountedToolCallback, never()).call("{\"code\":\"A01\"}");
+ }
+}
--
Gitblit v1.9.1