#ai 思维链

zhou zhou

15 小时以前 6477d7156272a6f1fe126c781958369bb10970c6

 rsf-server/src/main/java/com/vincent/rsf/server/ai/service/impl/AiChatServiceImpl.java

@@ -14,6 +14,7 @@
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;
@@ -51,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;

@@ -88,6 +86,10 @@
    @Qualifier("aiChatTaskExecutor")
    private final Executor aiChatTaskExecutor;

    /**
     * 获取当前对话抽屉初始化所需的运行时数据。
     * 该方法不会触发模型调用，而是把配置解析结果和会话记忆聚合成前端一次渲染所需的快照。
     */
    @Override
    public AiChatRuntimeDto getRuntime(String promptCode, Long sessionId, Long userId, Long tenantId) {
        AiResolvedConfig config = aiConfigResolverService.resolve(promptCode, tenantId);
@@ -109,6 +111,9 @@
                .build();
    }

    /**
     * 查询指定 Prompt 场景下的历史会话摘要列表。
     */
    @Override
    public List<AiChatSessionDto> listSessions(String promptCode, String keyword, Long userId, Long tenantId) {
        AiResolvedConfig config = aiConfigResolverService.resolve(promptCode, tenantId);
@@ -140,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);
@@ -148,6 +157,14 @@
    }

    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<>();
@@ -157,6 +174,7 @@
        Long sessionId = request.getSessionId();
        Long callLogId = null;
        String model = null;
        ThinkingTraceEmitter thinkingTraceEmitter = null;
        try {
            ensureIdentity(userId, tenantId);
            AiResolvedConfig config = resolveConfig(request, tenantId);
@@ -205,10 +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,
                        toolSuccessCount, toolFailureCount, callLogId, userId, tenantId
                        toolSuccessCount, toolFailureCount, callLogId, userId, tenantId, activeThinkingTraceEmitter
                );
                Prompt prompt = new Prompt(
                        buildPromptMessages(memory, mergedMessages, config.getPrompt(), request.getMetadata()),
@@ -221,9 +242,10 @@
                    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(
@@ -251,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));
                            }
@@ -262,6 +284,7 @@
                            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(
@@ -281,12 +304,12 @@
            }
        } catch (AiChatException e) {
            handleStreamFailure(emitter, requestId, sessionId, model, startedAt, firstTokenAtRef.get(), e,
                    callLogId, toolSuccessCount.get(), toolFailureCount.get());
                    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),
                    callLogId, toolSuccessCount.get(), toolFailureCount.get());
                    callLogId, toolSuccessCount.get(), toolFailureCount.get(), thinkingTraceEmitter);
        } finally {
            log.debug("AI chat stream finished, requestId={}", requestId);
        }
@@ -302,6 +325,7 @@
    }

    private AiResolvedConfig resolveConfig(AiChatRequest request, Long tenantId) {
        /** 把请求里的 Prompt 场景解析成一份可直接执行的 AI 配置。 */
        try {
            return aiConfigResolverService.resolve(request.getPromptCode(), tenantId);
        } catch (Exception e) {
@@ -311,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) {
@@ -320,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) {
@@ -329,6 +355,7 @@
    }

    private McpMountRuntimeFactory.McpMountRuntime createRuntime(AiResolvedConfig config, Long userId) {
        /** 按配置中的 MCP 挂载记录构造本轮对话专属的工具运行时。 */
        try {
            return mcpMountRuntimeFactory.create(config.getMcpMounts(), userId);
        } catch (Exception e) {
@@ -355,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)
@@ -388,10 +419,14 @@

    private void handleStreamFailure(SseEmitter emitter, String requestId, Long sessionId, String model, long startedAt,
                                     Long firstTokenAt, AiChatException exception, Long callLogId,
                                     long toolSuccessCount, long toolFailureCount) {
                                     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,
@@ -409,6 +444,9 @@
        }
        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)
@@ -456,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);
        }
@@ -507,7 +539,9 @@
    private ToolCallback[] wrapToolCallbacks(ToolCallback[] toolCallbacks, SseEmitter emitter, String requestId,
                                             Long sessionId, AtomicLong toolCallSequence,
                                             AtomicLong toolSuccessCount, AtomicLong toolFailureCount,
                                             Long callLogId, Long userId, Long tenantId) {
                                             Long callLogId, Long userId, Long tenantId,
                                             ThinkingTraceEmitter thinkingTraceEmitter) {
        /** 给所有工具回调套上一层可观测包装，用于实时 SSE 轨迹和审计日志落库。 */
        if (Cools.isEmpty(toolCallbacks)) {
            return toolCallbacks;
        }
@@ -517,12 +551,16 @@
                continue;
            }
            wrappedCallbacks.add(new ObservableToolCallback(callback, emitter, requestId, sessionId, toolCallSequence,
                    toolSuccessCount, toolFailureCount, callLogId, userId, tenantId));
                    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("对话消息不能为空");
        }
@@ -569,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);
@@ -634,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)
@@ -662,6 +702,7 @@
    }

    private void emitStrict(SseEmitter emitter, String eventName, Object payload) {
        /** 严格发送 SSE 事件；一旦发送失败，直接上抛为流式输出异常。 */
        try {
            String data = objectMapper.writeValueAsString(payload);
            emitter.send(SseEmitter.event()
@@ -673,6 +714,7 @@
    }

    private void emitSafely(SseEmitter emitter, String eventName, Object payload) {
        /** 尝试发送非关键事件，发送失败只记录日志，不打断主对话流程。 */
        try {
            emitStrict(emitter, eventName, payload);
        } catch (Exception e) {
@@ -702,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;
@@ -714,11 +875,13 @@
        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,
                                       AtomicLong toolSuccessCount, AtomicLong toolFailureCount,
                                       Long callLogId, Long userId, Long tenantId) {
                                       Long callLogId, Long userId, Long tenantId,
                                       ThinkingTraceEmitter thinkingTraceEmitter) {
            this.delegate = delegate;
            this.emitter = emitter;
            this.requestId = requestId;
@@ -729,6 +892,7 @@
            this.callLogId = callLogId;
            this.userId = userId;
            this.tenantId = tenantId;
            this.thinkingTraceEmitter = thinkingTraceEmitter;
        }

        @Override
@@ -748,10 +912,18 @@

        @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)
@@ -777,6 +949,9 @@
                        .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),
@@ -796,6 +971,9 @@
                        .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(),