From 6477d7156272a6f1fe126c781958369bb10970c6 Mon Sep 17 00:00:00 2001
From: zhou zhou <3272660260@qq.com>
Date: 星期六, 21 三月 2026 11:15:50 +0800
Subject: [PATCH] #ai 思维链
---
rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/AiChatServiceImpl.java | 328 ++++++++++++++++++++++++++++++++++++++++++++++++++----
1 files changed, 302 insertions(+), 26 deletions(-)
diff --git a/rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/AiChatServiceImpl.java b/rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/AiChatServiceImpl.java
index 5e3c7ab..e7d842e 100644
--- a/rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/AiChatServiceImpl.java
+++ b/rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/AiChatServiceImpl.java
@@ -14,16 +14,20 @@
import com.vincent.rsf.server.ai.dto.AiChatSessionDto;
import com.vincent.rsf.server.ai.dto.AiChatSessionPinRequest;
import com.vincent.rsf.server.ai.dto.AiChatSessionRenameRequest;
+import com.vincent.rsf.server.ai.dto.AiChatThinkingEventDto;
import com.vincent.rsf.server.ai.dto.AiChatToolEventDto;
import com.vincent.rsf.server.ai.dto.AiResolvedConfig;
+import com.vincent.rsf.server.ai.entity.AiCallLog;
import com.vincent.rsf.server.ai.entity.AiParam;
import com.vincent.rsf.server.ai.entity.AiPrompt;
import com.vincent.rsf.server.ai.entity.AiChatSession;
import com.vincent.rsf.server.ai.enums.AiErrorCategory;
import com.vincent.rsf.server.ai.exception.AiChatException;
+import com.vincent.rsf.server.ai.service.AiCallLogService;
import com.vincent.rsf.server.ai.service.AiChatService;
import com.vincent.rsf.server.ai.service.AiChatMemoryService;
import com.vincent.rsf.server.ai.service.AiConfigResolverService;
+import com.vincent.rsf.server.ai.service.MountedToolCallback;
import com.vincent.rsf.server.ai.service.McpMountRuntimeFactory;
import io.micrometer.observation.ObservationRegistry;
import lombok.RequiredArgsConstructor;
@@ -48,12 +52,9 @@
import org.springframework.ai.util.json.schema.SchemaType;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.http.MediaType;
-import org.springframework.http.client.SimpleClientHttpRequestFactory;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
-import org.springframework.web.client.RestClient;
-import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import reactor.core.publisher.Flux;
@@ -78,12 +79,17 @@
private final AiConfigResolverService aiConfigResolverService;
private final AiChatMemoryService aiChatMemoryService;
private final McpMountRuntimeFactory mcpMountRuntimeFactory;
+ private final AiCallLogService aiCallLogService;
private final GenericApplicationContext applicationContext;
private final ObservationRegistry observationRegistry;
private final ObjectMapper objectMapper;
@Qualifier("aiChatTaskExecutor")
private final Executor aiChatTaskExecutor;
+ /**
+ * 鑾峰彇褰撳墠瀵硅瘽鎶藉眽鍒濆鍖栨墍闇�鐨勮繍琛屾椂鏁版嵁銆�
+ * 璇ユ柟娉曚笉浼氳Е鍙戞ā鍨嬭皟鐢紝鑰屾槸鎶婇厤缃В鏋愮粨鏋滃拰浼氳瘽璁板繂鑱氬悎鎴愬墠绔竴娆℃覆鏌撴墍闇�鐨勫揩鐓с��
+ */
@Override
public AiChatRuntimeDto getRuntime(String promptCode, Long sessionId, Long userId, Long tenantId) {
AiResolvedConfig config = aiConfigResolverService.resolve(promptCode, tenantId);
@@ -105,6 +111,9 @@
.build();
}
+ /**
+ * 鏌ヨ鎸囧畾 Prompt 鍦烘櫙涓嬬殑鍘嗗彶浼氳瘽鎽樿鍒楄〃銆�
+ */
@Override
public List<AiChatSessionDto> listSessions(String promptCode, String keyword, Long userId, Long tenantId) {
AiResolvedConfig config = aiConfigResolverService.resolve(promptCode, tenantId);
@@ -136,6 +145,10 @@
aiChatMemoryService.retainLatestRound(userId, tenantId, sessionId);
}
+ /**
+ * 鍚姩涓�娆℃柊鐨� SSE 瀵硅瘽娴併��
+ * 鎺у埗绾跨▼绔嬪嵆杩斿洖 emitter锛岀湡姝ｇ殑妯″瀷璋冪敤涓庡伐鍏锋墽琛屼氦缁� AI 涓撶敤绾跨▼姹犲紓姝ュ鐞嗐��
+ */
@Override
public SseEmitter stream(AiChatRequest request, Long userId, Long tenantId) {
SseEmitter emitter = new SseEmitter(AiDefaults.SSE_TIMEOUT_MS);
@@ -144,12 +157,24 @@
}
private void doStream(AiChatRequest request, Long userId, Long tenantId, SseEmitter emitter) {
+ /**
+ * AI 瀵硅瘽鐨勬牳蹇冩墽琛岄摼璺細
+ * 1. 鏍￠獙韬唤鍜岃В鏋愮鎴烽厤缃�
+ * 2. 瑙ｆ瀽鎴栧垱寤轰細璇濓紝鍔犺浇璁板繂
+ * 3. 鍔ㄦ�佹寕杞� MCP 宸ュ叿
+ * 4. 鍙戣捣妯″瀷娴佸紡/闈炴祦寮忚皟鐢�
+ * 5. 鎸佷箙鍖栨湰杞秷鎭紝杈撳嚭 SSE 浜嬩欢骞惰褰曞璁℃棩蹇�
+ */
String requestId = request.getRequestId();
long startedAt = System.currentTimeMillis();
AtomicReference<Long> firstTokenAtRef = new AtomicReference<>();
AtomicLong toolCallSequence = new AtomicLong(0);
+ AtomicLong toolSuccessCount = new AtomicLong(0);
+ AtomicLong toolFailureCount = new AtomicLong(0);
Long sessionId = request.getSessionId();
+ Long callLogId = null;
String model = null;
+ ThinkingTraceEmitter thinkingTraceEmitter = null;
try {
ensureIdentity(userId, tenantId);
AiResolvedConfig config = resolveConfig(request, tenantId);
@@ -159,6 +184,19 @@
sessionId = session.getId();
AiChatMemoryDto memory = loadMemory(userId, tenantId, config.getPromptCode(), session.getId());
List<AiChatMessageDto> mergedMessages = mergeMessages(memory.getShortMemoryMessages(), request.getMessages());
+ AiCallLog callLog = aiCallLogService.startCallLog(
+ requestId,
+ session.getId(),
+ userId,
+ tenantId,
+ config.getPromptCode(),
+ config.getPrompt().getName(),
+ config.getAiParam().getModel(),
+ config.getMcpMounts().size(),
+ config.getMcpMounts().size(),
+ config.getMcpMounts().stream().map(item -> item.getName()).toList()
+ );
+ callLogId = callLog.getId();
try (McpMountRuntimeFactory.McpMountRuntime runtime = createRuntime(config, userId)) {
emitStrict(emitter, "start", AiChatRuntimeDto.builder()
.requestId(requestId)
@@ -185,9 +223,13 @@
.build());
log.info("AI chat started, requestId={}, userId={}, tenantId={}, sessionId={}, model={}",
requestId, userId, tenantId, session.getId(), resolvedModel);
+ thinkingTraceEmitter = new ThinkingTraceEmitter(emitter, requestId, session.getId());
+ thinkingTraceEmitter.startAnalyze();
+ ThinkingTraceEmitter activeThinkingTraceEmitter = thinkingTraceEmitter;
ToolCallback[] observableToolCallbacks = wrapToolCallbacks(
- runtime.getToolCallbacks(), emitter, requestId, session.getId(), toolCallSequence
+ runtime.getToolCallbacks(), emitter, requestId, session.getId(), toolCallSequence,
+ toolSuccessCount, toolFailureCount, callLogId, userId, tenantId, activeThinkingTraceEmitter
);
Prompt prompt = new Prompt(
buildPromptMessages(memory, mergedMessages, config.getPrompt(), request.getMetadata()),
@@ -200,11 +242,23 @@
String content = extractContent(response);
aiChatMemoryService.saveRound(session, userId, tenantId, request.getMessages(), content);
if (StringUtils.hasText(content)) {
- markFirstToken(firstTokenAtRef, emitter, requestId, session.getId(), resolvedModel, startedAt);
+ markFirstToken(firstTokenAtRef, emitter, requestId, session.getId(), resolvedModel, startedAt, activeThinkingTraceEmitter);
emitStrict(emitter, "delta", buildMessagePayload("requestId", requestId, "content", content));
}
+ activeThinkingTraceEmitter.completeCurrentPhase();
emitDone(emitter, requestId, response.getMetadata(), config.getAiParam().getModel(), session.getId(), startedAt, firstTokenAtRef.get());
emitSafely(emitter, "status", buildTerminalStatus(requestId, session.getId(), "COMPLETED", resolvedModel, startedAt, firstTokenAtRef.get()));
+ aiCallLogService.completeCallLog(
+ callLogId,
+ "COMPLETED",
+ System.currentTimeMillis() - startedAt,
+ resolveFirstTokenLatency(startedAt, firstTokenAtRef.get()),
+ response.getMetadata() == null || response.getMetadata().getUsage() == null ? null : response.getMetadata().getUsage().getPromptTokens(),
+ response.getMetadata() == null || response.getMetadata().getUsage() == null ? null : response.getMetadata().getUsage().getCompletionTokens(),
+ response.getMetadata() == null || response.getMetadata().getUsage() == null ? null : response.getMetadata().getUsage().getTotalTokens(),
+ toolSuccessCount.get(),
+ toolFailureCount.get()
+ );
log.info("AI chat completed, requestId={}, sessionId={}, elapsedMs={}, firstTokenLatencyMs={}",
requestId, session.getId(), System.currentTimeMillis() - startedAt, resolveFirstTokenLatency(startedAt, firstTokenAtRef.get()));
emitter.complete();
@@ -219,7 +273,7 @@
lastMetadata.set(response.getMetadata());
String content = extractContent(response);
if (StringUtils.hasText(content)) {
- markFirstToken(firstTokenAtRef, emitter, requestId, session.getId(), resolvedModel, startedAt);
+ markFirstToken(firstTokenAtRef, emitter, requestId, session.getId(), resolvedModel, startedAt, activeThinkingTraceEmitter);
assistantContent.append(content);
emitStrict(emitter, "delta", buildMessagePayload("requestId", requestId, "content", content));
}
@@ -230,18 +284,32 @@
e == null ? "AI 妯″瀷娴佸紡璋冪敤澶辫触" : e.getMessage(), e);
}
aiChatMemoryService.saveRound(session, userId, tenantId, request.getMessages(), assistantContent.toString());
+ activeThinkingTraceEmitter.completeCurrentPhase();
emitDone(emitter, requestId, lastMetadata.get(), config.getAiParam().getModel(), session.getId(), startedAt, firstTokenAtRef.get());
emitSafely(emitter, "status", buildTerminalStatus(requestId, session.getId(), "COMPLETED", resolvedModel, startedAt, firstTokenAtRef.get()));
+ aiCallLogService.completeCallLog(
+ callLogId,
+ "COMPLETED",
+ System.currentTimeMillis() - startedAt,
+ resolveFirstTokenLatency(startedAt, firstTokenAtRef.get()),
+ lastMetadata.get() == null || lastMetadata.get().getUsage() == null ? null : lastMetadata.get().getUsage().getPromptTokens(),
+ lastMetadata.get() == null || lastMetadata.get().getUsage() == null ? null : lastMetadata.get().getUsage().getCompletionTokens(),
+ lastMetadata.get() == null || lastMetadata.get().getUsage() == null ? null : lastMetadata.get().getUsage().getTotalTokens(),
+ toolSuccessCount.get(),
+ toolFailureCount.get()
+ );
log.info("AI chat completed, requestId={}, sessionId={}, elapsedMs={}, firstTokenLatencyMs={}",
requestId, session.getId(), System.currentTimeMillis() - startedAt, resolveFirstTokenLatency(startedAt, firstTokenAtRef.get()));
emitter.complete();
}
} catch (AiChatException e) {
- handleStreamFailure(emitter, requestId, sessionId, model, startedAt, firstTokenAtRef.get(), e);
+ handleStreamFailure(emitter, requestId, sessionId, model, startedAt, firstTokenAtRef.get(), e,
+ callLogId, toolSuccessCount.get(), toolFailureCount.get(), thinkingTraceEmitter);
} catch (Exception e) {
handleStreamFailure(emitter, requestId, sessionId, model, startedAt, firstTokenAtRef.get(),
buildAiException("AI_INTERNAL_ERROR", AiErrorCategory.INTERNAL, "INTERNAL",
- e == null ? "AI 瀵硅瘽澶辫触" : e.getMessage(), e));
+ e == null ? "AI 瀵硅瘽澶辫触" : e.getMessage(), e),
+ callLogId, toolSuccessCount.get(), toolFailureCount.get(), thinkingTraceEmitter);
} finally {
log.debug("AI chat stream finished, requestId={}", requestId);
}
@@ -257,6 +325,7 @@
}
private AiResolvedConfig resolveConfig(AiChatRequest request, Long tenantId) {
+ /** 鎶婅姹傞噷鐨� Prompt 鍦烘櫙瑙ｆ瀽鎴愪竴浠藉彲鐩存帴鎵ц鐨� AI 閰嶇疆銆� */
try {
return aiConfigResolverService.resolve(request.getPromptCode(), tenantId);
} catch (Exception e) {
@@ -266,6 +335,7 @@
}
private AiChatSession resolveSession(AiChatRequest request, Long userId, Long tenantId, String promptCode) {
+ /** 鏍规嵁 sessionId 澶嶇敤鍘嗗彶浼氳瘽锛屾垨鍦ㄩ娆℃彁闂椂鍒涘缓鏂颁細璇濄�� */
try {
return aiChatMemoryService.resolveSession(userId, tenantId, promptCode, request.getSessionId(), resolveTitleSeed(request.getMessages()));
} catch (Exception e) {
@@ -275,6 +345,7 @@
}
private AiChatMemoryDto loadMemory(Long userId, Long tenantId, String promptCode, Long sessionId) {
+ /** 璇诲彇浼氳瘽鐨勭煭鏈熻蹇嗐�佹憳瑕佽蹇嗗拰浜嬪疄璁板繂锛屼緵妯″瀷缁勮涓婁笅鏂囥�� */
try {
return aiChatMemoryService.getMemory(userId, tenantId, promptCode, sessionId);
} catch (Exception e) {
@@ -284,6 +355,7 @@
}
private McpMountRuntimeFactory.McpMountRuntime createRuntime(AiResolvedConfig config, Long userId) {
+ /** 鎸夐厤缃腑鐨� MCP 鎸傝浇璁板綍鏋勯�犳湰杞璇濅笓灞炵殑宸ュ叿杩愯鏃躲�� */
try {
return mcpMountRuntimeFactory.create(config.getMcpMounts(), userId);
} catch (Exception e) {
@@ -310,9 +382,13 @@
}
}
- private void markFirstToken(AtomicReference<Long> firstTokenAtRef, SseEmitter emitter, String requestId, Long sessionId, String model, long startedAt) {
+ private void markFirstToken(AtomicReference<Long> firstTokenAtRef, SseEmitter emitter, String requestId,
+ Long sessionId, String model, long startedAt, ThinkingTraceEmitter thinkingTraceEmitter) {
if (!firstTokenAtRef.compareAndSet(null, System.currentTimeMillis())) {
return;
+ }
+ if (thinkingTraceEmitter != null) {
+ thinkingTraceEmitter.startAnswer();
}
emitSafely(emitter, "status", AiChatStatusDto.builder()
.requestId(requestId)
@@ -341,16 +417,36 @@
return firstTokenAt == null ? null : Math.max(0L, firstTokenAt - startedAt);
}
- private void handleStreamFailure(SseEmitter emitter, String requestId, Long sessionId, String model, long startedAt, Long firstTokenAt, AiChatException exception) {
+ private void handleStreamFailure(SseEmitter emitter, String requestId, Long sessionId, String model, long startedAt,
+ Long firstTokenAt, AiChatException exception, Long callLogId,
+ long toolSuccessCount, long toolFailureCount,
+ ThinkingTraceEmitter thinkingTraceEmitter) {
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");
+ }
emitSafely(emitter, "status", buildTerminalStatus(requestId, sessionId, "ABORTED", model, startedAt, firstTokenAt));
+ aiCallLogService.failCallLog(
+ callLogId,
+ "ABORTED",
+ exception.getCategory().name(),
+ exception.getStage(),
+ exception.getMessage(),
+ System.currentTimeMillis() - startedAt,
+ resolveFirstTokenLatency(startedAt, firstTokenAt),
+ toolSuccessCount,
+ toolFailureCount
+ );
emitter.completeWithError(exception);
return;
}
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");
+ }
emitSafely(emitter, "status", buildTerminalStatus(requestId, sessionId, "FAILED", model, startedAt, firstTokenAt));
emitSafely(emitter, "error", AiChatErrorDto.builder()
.requestId(requestId)
@@ -361,6 +457,17 @@
.message(exception.getMessage())
.timestamp(Instant.now().toEpochMilli())
.build());
+ aiCallLogService.failCallLog(
+ callLogId,
+ "FAILED",
+ exception.getCategory().name(),
+ exception.getStage(),
+ exception.getMessage(),
+ System.currentTimeMillis() - startedAt,
+ resolveFirstTokenLatency(startedAt, firstTokenAt),
+ toolSuccessCount,
+ toolFailureCount
+ );
emitter.completeWithError(exception);
}
@@ -387,21 +494,15 @@
}
private OpenAiApi buildOpenAiApi(AiParam aiParam) {
- int timeoutMs = aiParam.getTimeoutMs() == null ? AiDefaults.DEFAULT_TIMEOUT_MS : aiParam.getTimeoutMs();
- SimpleClientHttpRequestFactory requestFactory = new SimpleClientHttpRequestFactory();
- requestFactory.setConnectTimeout(timeoutMs);
- requestFactory.setReadTimeout(timeoutMs);
-
- return OpenAiApi.builder()
- .baseUrl(aiParam.getBaseUrl())
- .apiKey(aiParam.getApiKey())
- .restClientBuilder(RestClient.builder().requestFactory(requestFactory))
- .webClientBuilder(WebClient.builder())
- .build();
+ return AiOpenAiApiSupport.buildOpenAiApi(aiParam);
}
private OpenAiChatOptions buildChatOptions(AiParam aiParam, ToolCallback[] toolCallbacks, Long userId, Long tenantId,
String requestId, Long sessionId, Map<String, Object> metadata) {
+ /**
+ * 缁勮涓�娆¤亰澶╄皟鐢ㄧ殑鍏ㄩ儴妯″瀷閫夐」鍜� Tool Context銆�
+ * Tool Context 浼氶�忎紶缁欏唴缃伐鍏峰拰澶栭儴 MCP锛屼繚璇佸伐鍏峰湪绉熸埛鍜屼細璇濊寖鍥村唴鎵ц銆�
+ */
if (userId == null) {
throw buildAiException("AI_AUTH_USER_MISSING", AiErrorCategory.AUTH, "OPTIONS_BUILD", "褰撳墠鐧诲綍鐢ㄦ埛涓嶅瓨鍦�", null);
}
@@ -436,7 +537,11 @@
}
private ToolCallback[] wrapToolCallbacks(ToolCallback[] toolCallbacks, SseEmitter emitter, String requestId,
- Long sessionId, AtomicLong toolCallSequence) {
+ Long sessionId, AtomicLong toolCallSequence,
+ AtomicLong toolSuccessCount, AtomicLong toolFailureCount,
+ Long callLogId, Long userId, Long tenantId,
+ ThinkingTraceEmitter thinkingTraceEmitter) {
+ /** 缁欐墍鏈夊伐鍏峰洖璋冨涓婁竴灞傚彲瑙傛祴鍖呰锛岀敤浜庡疄鏃� SSE 杞ㄨ抗鍜屽璁℃棩蹇楄惤搴撱�� */
if (Cools.isEmpty(toolCallbacks)) {
return toolCallbacks;
}
@@ -445,12 +550,17 @@
if (callback == null) {
continue;
}
- wrappedCallbacks.add(new ObservableToolCallback(callback, emitter, requestId, sessionId, toolCallSequence));
+ wrappedCallbacks.add(new ObservableToolCallback(callback, emitter, requestId, sessionId, toolCallSequence,
+ toolSuccessCount, toolFailureCount, callLogId, userId, tenantId, thinkingTraceEmitter));
}
return wrappedCallbacks.toArray(new ToolCallback[0]);
}
private List<Message> buildPromptMessages(AiChatMemoryDto memory, List<AiChatMessageDto> sourceMessages, AiPrompt aiPrompt, Map<String, Object> metadata) {
+ /**
+ * 缁勮鏈�缁堟彁浜ょ粰妯″瀷鐨勬秷鎭垪琛ㄣ��
+ * 椤哄簭涓婂缁堟槸锛氱郴缁� Prompt -> 鍘嗗彶鎽樿 -> 鍏抽敭浜嬪疄 -> 鏈�杩戝璇� -> 褰撳墠鐢ㄦ埛杈撳叆銆�
+ */
if (Cools.isEmpty(sourceMessages)) {
throw new CoolException("瀵硅瘽娑堟伅涓嶈兘涓虹┖");
}
@@ -497,6 +607,7 @@
}
private List<AiChatMessageDto> mergeMessages(List<AiChatMessageDto> persistedMessages, List<AiChatMessageDto> memoryMessages) {
+ /** 鎶婅惤搴撳巻鍙蹭笌鏈疆鍓嶇鍐呭瓨澧為噺鍚堝苟鎴愭ā鍨嬪彲娑堣垂鐨勫畬鏁翠笂涓嬫枃銆� */
List<AiChatMessageDto> merged = new ArrayList<>();
if (!Cools.isEmpty(persistedMessages)) {
merged.addAll(persistedMessages);
@@ -562,6 +673,7 @@
}
private void emitDone(SseEmitter emitter, String requestId, ChatResponseMetadata metadata, String fallbackModel, Long sessionId, long startedAt, Long firstTokenAt) {
+ /** 杈撳嚭瀵硅瘽瀹屾垚浜嬩欢锛岀粺涓�灏佽鑰楁椂銆侀鍖呭欢杩熷拰 token 鐢ㄩ噺銆� */
Usage usage = metadata == null ? null : metadata.getUsage();
emitStrict(emitter, "done", AiChatDoneDto.builder()
.requestId(requestId)
@@ -590,6 +702,7 @@
}
private void emitStrict(SseEmitter emitter, String eventName, Object payload) {
+ /** 涓ユ牸鍙戦�� SSE 浜嬩欢锛涗竴鏃﹀彂閫佸け璐ワ紝鐩存帴涓婃姏涓烘祦寮忚緭鍑哄紓甯搞�� */
try {
String data = objectMapper.writeValueAsString(payload);
emitter.send(SseEmitter.event()
@@ -601,6 +714,7 @@
}
private void emitSafely(SseEmitter emitter, String eventName, Object payload) {
+ /** 灏濊瘯鍙戦�侀潪鍏抽敭浜嬩欢锛屽彂閫佸け璐ュ彧璁板綍鏃ュ織锛屼笉鎵撴柇涓诲璇濇祦绋嬨�� */
try {
emitStrict(emitter, eventName, payload);
} catch (Exception e) {
@@ -630,6 +744,125 @@
return false;
}
+ private class ThinkingTraceEmitter {
+
+ private final SseEmitter emitter;
+ private final String requestId;
+ private final Long sessionId;
+ private String currentPhase;
+ private String currentStatus;
+
+ private ThinkingTraceEmitter(SseEmitter emitter, String requestId, Long sessionId) {
+ this.emitter = emitter;
+ this.requestId = requestId;
+ this.sessionId = sessionId;
+ }
+
+ private void startAnalyze() {
+ if (currentPhase != null) {
+ return;
+ }
+ currentPhase = "ANALYZE";
+ currentStatus = "STARTED";
+ emitThinkingEvent("ANALYZE", "STARTED", "姝ｅ湪鍒嗘瀽闂",
+ "宸叉帴鏀朵綘鐨勯棶棰橈紝姝ｅ湪鐞嗚В鎰忓浘骞跺垽鏂槸鍚﹂渶瑕佽皟鐢ㄥ伐鍏枫��", null);
+ }
+
+ private void onToolStart(String toolName, String toolCallId) {
+ switchPhase("TOOL_CALL", "STARTED", "姝ｅ湪璋冪敤宸ュ叿", "宸插垽鏂渶瑕佽皟鐢ㄥ伐鍏凤紝姝ｅ湪鏌ヨ鐩稿叧淇℃伅銆�", null);
+ currentStatus = "UPDATED";
+ emitThinkingEvent("TOOL_CALL", "UPDATED", "姝ｅ湪璋冪敤宸ュ叿",
+ "姝ｅ湪璋冪敤宸ュ叿 " + safeLabel(toolName, "鏈煡宸ュ叿") + " 鑾峰彇鎵�闇�淇℃伅銆�", toolCallId);
+ }
+
+ private 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);
+ }
+
+ private void startAnswer() {
+ switchPhase("ANSWER", "STARTED", "姝ｅ湪鏁寸悊绛旀", "宸插畬鎴愬垎鏋愶紝姝ｅ湪缁勭粐鏈�缁堝洖澶嶅唴瀹广��", null);
+ }
+
+ private void completeCurrentPhase() {
+ if (!StringUtils.hasText(currentPhase) || isTerminalStatus(currentStatus)) {
+ return;
+ }
+ currentStatus = "COMPLETED";
+ emitThinkingEvent(currentPhase, "COMPLETED", resolveCompleteTitle(currentPhase),
+ resolveCompleteContent(currentPhase), null);
+ }
+
+ private 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) {
+ 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;
+ }
+ }
+
private class ObservableToolCallback implements ToolCallback {
private final ToolCallback delegate;
@@ -637,14 +870,29 @@
private final String requestId;
private final Long sessionId;
private final AtomicLong toolCallSequence;
+ private final AtomicLong toolSuccessCount;
+ private final AtomicLong toolFailureCount;
+ private final Long callLogId;
+ private final Long userId;
+ private final Long tenantId;
+ private final ThinkingTraceEmitter thinkingTraceEmitter;
private ObservableToolCallback(ToolCallback delegate, SseEmitter emitter, String requestId,
- Long sessionId, AtomicLong toolCallSequence) {
+ Long sessionId, AtomicLong toolCallSequence,
+ AtomicLong toolSuccessCount, AtomicLong toolFailureCount,
+ Long callLogId, Long userId, Long tenantId,
+ ThinkingTraceEmitter thinkingTraceEmitter) {
this.delegate = delegate;
this.emitter = emitter;
this.requestId = requestId;
this.sessionId = sessionId;
this.toolCallSequence = toolCallSequence;
+ this.toolSuccessCount = toolSuccessCount;
+ this.toolFailureCount = toolFailureCount;
+ this.callLogId = callLogId;
+ this.userId = userId;
+ this.tenantId = tenantId;
+ this.thinkingTraceEmitter = thinkingTraceEmitter;
}
@Override
@@ -664,44 +912,72 @@
@Override
public String call(String toolInput, ToolContext toolContext) {
+ /**
+ * 宸ュ叿鎵ц瑙傛祴鍖呰鍣ㄣ��
+ * 鍦ㄧ湡瀹炶皟鐢ㄥ墠鍚庡垎鍒彂閫� tool_start / tool_result / tool_error锛�
+ * 鍚屾椂鎶婅皟鐢ㄦ憳瑕佸啓鍏� MCP 璋冪敤鏃ュ織琛ㄣ��
+ */
String toolName = delegate.getToolDefinition() == null ? "unknown" : delegate.getToolDefinition().name();
+ String mountName = delegate instanceof MountedToolCallback ? ((MountedToolCallback) delegate).getMountName() : null;
String toolCallId = requestId + "-tool-" + toolCallSequence.incrementAndGet();
long startedAt = System.currentTimeMillis();
+ if (thinkingTraceEmitter != null) {
+ thinkingTraceEmitter.onToolStart(toolName, toolCallId);
+ }
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;
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(System.currentTimeMillis() - startedAt)
+ .durationMs(durationMs)
.timestamp(System.currentTimeMillis())
.build());
+ if (thinkingTraceEmitter != null) {
+ thinkingTraceEmitter.onToolResult(toolName, toolCallId, false);
+ }
+ toolSuccessCount.incrementAndGet();
+ aiCallLogService.saveMcpCallLog(callLogId, requestId, sessionId, toolCallId, mountName, toolName,
+ "COMPLETED", summarizeToolPayload(toolInput, 400), summarizeToolPayload(output, 600),
+ null, durationMs, userId, tenantId);
return output;
} catch (RuntimeException e) {
+ long durationMs = System.currentTimeMillis() - startedAt;
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(System.currentTimeMillis() - startedAt)
+ .durationMs(durationMs)
.timestamp(System.currentTimeMillis())
.build());
+ if (thinkingTraceEmitter != null) {
+ thinkingTraceEmitter.onToolResult(toolName, toolCallId, true);
+ }
+ toolFailureCount.incrementAndGet();
+ aiCallLogService.saveMcpCallLog(callLogId, requestId, sessionId, toolCallId, mountName, toolName,
+ "FAILED", summarizeToolPayload(toolInput, 400), null, e.getMessage(),
+ durationMs, userId, tenantId);
throw e;
}
}
--
Gitblit v1.9.1