From 80a6d9236ade191a5de0975abe4de5a6e7e63915 Mon Sep 17 00:00:00 2001
From: zhou zhou <3272660260@qq.com>
Date: 星期四, 19 三月 2026 14:03:10 +0800
Subject: [PATCH] #AI.注释
---
rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/AiChatServiceImpl.java | 157 +++++++++++++++++++++++++++++++++++++++++++++++++---
1 files changed, 148 insertions(+), 9 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..0430123 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
@@ -16,14 +16,17 @@
import com.vincent.rsf.server.ai.dto.AiChatSessionRenameRequest;
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;
@@ -78,12 +81,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 +113,9 @@
.build();
}
+ /**
+ * 鏌ヨ鎸囧畾 Prompt 鍦烘櫙涓嬬殑鍘嗗彶浼氳瘽鎽樿鍒楄〃銆�
+ */
@Override
public List<AiChatSessionDto> listSessions(String promptCode, String keyword, Long userId, Long tenantId) {
AiResolvedConfig config = aiConfigResolverService.resolve(promptCode, tenantId);
@@ -136,6 +147,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,11 +159,22 @@
}
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;
try {
ensureIdentity(userId, tenantId);
@@ -159,6 +185,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)
@@ -187,7 +226,8 @@
requestId, userId, tenantId, session.getId(), resolvedModel);
ToolCallback[] observableToolCallbacks = wrapToolCallbacks(
- runtime.getToolCallbacks(), emitter, requestId, session.getId(), toolCallSequence
+ runtime.getToolCallbacks(), emitter, requestId, session.getId(), toolCallSequence,
+ toolSuccessCount, toolFailureCount, callLogId, userId, tenantId
);
Prompt prompt = new Prompt(
buildPromptMessages(memory, mergedMessages, config.getPrompt(), request.getMetadata()),
@@ -205,6 +245,17 @@
}
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();
@@ -232,16 +283,29 @@
aiChatMemoryService.saveRound(session, userId, tenantId, request.getMessages(), assistantContent.toString());
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());
} 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());
} finally {
log.debug("AI chat stream finished, requestId={}", requestId);
}
@@ -257,6 +321,7 @@
}
private AiResolvedConfig resolveConfig(AiChatRequest request, Long tenantId) {
+ /** 鎶婅姹傞噷鐨� Prompt 鍦烘櫙瑙ｆ瀽鎴愪竴浠藉彲鐩存帴鎵ц鐨� AI 閰嶇疆銆� */
try {
return aiConfigResolverService.resolve(request.getPromptCode(), tenantId);
} catch (Exception e) {
@@ -266,6 +331,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 +341,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 +351,7 @@
}
private McpMountRuntimeFactory.McpMountRuntime createRuntime(AiResolvedConfig config, Long userId) {
+ /** 鎸夐厤缃腑鐨� MCP 鎸傝浇璁板綍鏋勯�犳湰杞璇濅笓灞炵殑宸ュ叿杩愯鏃躲�� */
try {
return mcpMountRuntimeFactory.create(config.getMcpMounts(), userId);
} catch (Exception e) {
@@ -341,11 +409,24 @@
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) {
if (isClientAbortException(exception)) {
log.warn("AI chat aborted by client, requestId={}, sessionId={}, stage={}, message={}",
requestId, sessionId, exception.getStage(), exception.getMessage());
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;
}
@@ -361,6 +442,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);
}
@@ -402,6 +494,10 @@
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 +532,10 @@
}
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) {
+ /** 缁欐墍鏈夊伐鍏峰洖璋冨涓婁竴灞傚彲瑙傛祴鍖呰锛岀敤浜庡疄鏃� SSE 杞ㄨ抗鍜屽璁℃棩蹇楄惤搴撱�� */
if (Cools.isEmpty(toolCallbacks)) {
return toolCallbacks;
}
@@ -445,12 +544,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));
}
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 +601,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 +667,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 +696,7 @@
}
private void emitStrict(SseEmitter emitter, String eventName, Object payload) {
+ /** 涓ユ牸鍙戦�� SSE 浜嬩欢锛涗竴鏃﹀彂閫佸け璐ワ紝鐩存帴涓婃姏涓烘祦寮忚緭鍑哄紓甯搞�� */
try {
String data = objectMapper.writeValueAsString(payload);
emitter.send(SseEmitter.event()
@@ -601,6 +708,7 @@
}
private void emitSafely(SseEmitter emitter, String eventName, Object payload) {
+ /** 灏濊瘯鍙戦�侀潪鍏抽敭浜嬩欢锛屽彂閫佸け璐ュ彧璁板綍鏃ュ織锛屼笉鎵撴柇涓诲璇濇祦绋嬨�� */
try {
emitStrict(emitter, eventName, payload);
} catch (Exception e) {
@@ -637,14 +745,26 @@
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 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) {
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;
}
@Override
@@ -664,7 +784,13 @@
@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();
emitSafely(emitter, "tool_start", AiChatToolEventDto.builder()
@@ -672,36 +798,49 @@
.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());
+ 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());
+ 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