You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
339 lines
15 KiB
339 lines
15 KiB
package com.wok.supportbot.app;
|
|
|
|
import com.wok.supportbot.advisor.ContentSafetyAdvisor;
|
|
import com.wok.supportbot.advisor.MyLoggerAdvisor;
|
|
import com.wok.supportbot.chatmemory.DatabaseChatMemory;
|
|
import com.wok.supportbot.config.ChatModelFactory;
|
|
import com.wok.supportbot.config.SimpleCircuitBreaker;
|
|
import com.wok.supportbot.mcp.McpToolCallback;
|
|
import com.wok.supportbot.mcp.McpToolCallbackAdapter;
|
|
import jakarta.annotation.Resource;
|
|
import lombok.extern.slf4j.Slf4j;
|
|
import org.springframework.ai.chat.client.ChatClient;
|
|
import org.springframework.ai.chat.client.advisor.MessageChatMemoryAdvisor;
|
|
import org.springframework.ai.chat.model.ChatModel;
|
|
import org.springframework.ai.document.Document;
|
|
import org.springframework.ai.tool.ToolCallback;
|
|
import org.springframework.ai.vectorstore.VectorStore;
|
|
import org.springframework.beans.factory.annotation.Value;
|
|
import org.springframework.stereotype.Component;
|
|
import org.springframework.util.StringUtils;
|
|
import reactor.core.Disposable;
|
|
import reactor.core.publisher.Flux;
|
|
import reactor.core.publisher.FluxSink;
|
|
|
|
import java.util.ArrayList;
|
|
import java.util.Collections;
|
|
import java.util.LinkedHashMap;
|
|
import java.util.List;
|
|
import java.util.Map;
|
|
|
|
import static org.springframework.ai.chat.memory.ChatMemory.CONVERSATION_ID;
|
|
|
|
/**
|
|
* AI 对话执行层 —— ChatClient 构建、缓存管理与 LLM 调用执行。
|
|
* <p>
|
|
* 本类负责 ChatClient 生命周期（按 appType + allowedMcpTools 缓存，LRU 淘汰），
|
|
* Advisor 链装配（ContentSafetyAdvisor → MessageChatMemoryAdvisor → MyLoggerAdvisor），
|
|
* 以及同步/流式对话执行（含熔断保护）。对话编排决策由 {@link ChatPipeline} 完成。
|
|
* <p>
|
|
* 核心方法：
|
|
* <ul>
|
|
* <li>{@link #chat(ChatContext)} — 同步对话（返回纯文本）</li>
|
|
* <li>{@link #chatWithEvents(ChatContext)} — 同步对话 + MCP 工具调用事件</li>
|
|
* <li>{@link #chatStream(ChatContext)} — 流式对话（返回 Flux<String>）</li>
|
|
* </ul>
|
|
* <p>
|
|
* {@code @pipeline} execution-layer order=2<br>
|
|
* {@code @pipeline-step} 熔断检查: SimpleCircuitBreaker(3次失败/5分钟恢复) → 熔断时返回降级提示<br>
|
|
* {@code @pipeline-step} Advisor链: ContentSafetyAdvisor(HIGHEST) → MessageChatMemoryAdvisor → MyLoggerAdvisor<br>
|
|
* {@code @pipeline-step} ChatClient.call/stream: LLM 大模型调用 → 成功/失败记录到熔断器<br>
|
|
* 同步至: frontend/src/views/PipelineFlow.vue Advisor 子图 & 熔断节点
|
|
*
|
|
* @author lyx
|
|
* @version 1.0.0
|
|
* @date 2025/06/27 14:11
|
|
*/
|
|
@Component
|
|
@Slf4j
|
|
public class AssistantApp {
|
|
|
|
@Resource
|
|
private ContentSafetyAdvisor contentSafetyAdvisor;
|
|
|
|
@Resource
|
|
private McpToolCallbackAdapter mcpToolCallbackAdapter;
|
|
|
|
@Resource
|
|
private ChatPipeline chatPipeline;
|
|
|
|
/** MCP 工具开关，默认启用，可通过 application.yml 的 chat.mcp.enabled 关闭 */
|
|
@Value("${chat.mcp.enabled:true}")
|
|
private boolean enableMcpTools;
|
|
|
|
private final ChatModelFactory chatModelFactory;
|
|
|
|
private final DatabaseChatMemory chatMemory;
|
|
|
|
/** ChatClient 缓存上限，超出后按 LRU 自动淘汰最早最少使用的实例 */
|
|
private static final int CHAT_CLIENT_CACHE_MAX = 32;
|
|
|
|
/** ChatClient 缓存（LinkedHashMap + removeEldestEntry 实现 LRU 淘汰） */
|
|
private final Map<String, ChatClient> chatClientCache =
|
|
Collections.synchronizedMap(new LinkedHashMap<>(16, 0.75f, true) {
|
|
@Override
|
|
protected boolean removeEldestEntry(Map.Entry<String, ChatClient> eldest) {
|
|
if (size() > CHAT_CLIENT_CACHE_MAX) {
|
|
log.info("ChatClient 缓存淘汰: key={}", eldest.getKey());
|
|
return true;
|
|
}
|
|
return false;
|
|
}
|
|
});
|
|
private final SimpleCircuitBreaker aiCircuitBreaker = new SimpleCircuitBreaker(3, 5 * 60 * 1000L, "AI");
|
|
/** 全局 AI 调用的熔断 key（非模型配置级，整体熔断） */
|
|
private static final Long AI_CIRCUIT_KEY = -1L;
|
|
|
|
/** AI 熔断时的降级提示语 */
|
|
private static final String CIRCUIT_OPEN_MESSAGE = "AI 服务暂时不可用，请稍后重试。";
|
|
|
|
private static final String SYSTEM_PROMPT = "";
|
|
|
|
/** 尾部空白缓冲上限：超过后强制发出，避免纯空白输出导致 buffer 无界增长 */
|
|
private static final int MAX_TRAILING_WHITESPACE_BUFFER = 256;
|
|
|
|
/**
|
|
* 初始化 ChatClient
|
|
*
|
|
* @param chatModelFactory
|
|
*/
|
|
public AssistantApp(ChatModelFactory chatModelFactory, DatabaseChatMemory chatMemory) {
|
|
this.chatModelFactory = chatModelFactory;
|
|
this.chatMemory = chatMemory;
|
|
}
|
|
|
|
private ChatClient getChatClient(String appType) {
|
|
return getChatClient(appType, enableMcpTools ? List.of("*") : null);
|
|
}
|
|
|
|
/**
|
|
* 获取 ChatClient（支持按角色过滤 MCP 工具），缓存上限 32 个，按 LRU 淘汰。
|
|
*/
|
|
private ChatClient getChatClient(String appType, List<String> allowedMcpTools) {
|
|
// 仅 null（无角色）降级为 ["*"]（向后兼容允许全部）；空列表表示有角色但未授权，保持空（不注册任何工具）
|
|
final List<String> effectiveTools;
|
|
if (enableMcpTools && allowedMcpTools == null) {
|
|
effectiveTools = List.of("*");
|
|
} else {
|
|
effectiveTools = allowedMcpTools;
|
|
}
|
|
String cacheKey = appType + ":" + (effectiveTools == null ? "none" : String.join(",", effectiveTools));
|
|
|
|
ChatClient cached = chatClientCache.get(cacheKey);
|
|
if (cached != null) return cached;
|
|
|
|
return chatClientCache.computeIfAbsent(cacheKey, key -> {
|
|
ChatModel chatModel = chatModelFactory.getChatModel(appType);
|
|
var builder = ChatClient.builder(chatModel);
|
|
if (StringUtils.hasText(SYSTEM_PROMPT)) {
|
|
builder.defaultSystem(SYSTEM_PROMPT);
|
|
}
|
|
// 注册 MCP 工具（按角色权限过滤）
|
|
if (enableMcpTools && effectiveTools != null && !effectiveTools.isEmpty()) {
|
|
ToolCallback[] mcpTools = mcpToolCallbackAdapter.getFilteredToolCallbacks(effectiveTools);
|
|
if (mcpTools.length > 0) {
|
|
builder.defaultToolCallbacks(mcpTools);
|
|
log.info("已注册 {} 个 MCP 工具到 ChatClient [{}]", mcpTools.length, cacheKey);
|
|
for (ToolCallback tc : mcpTools) {
|
|
log.info(" 工具: {} — {}", tc.getToolDefinition().name(), tc.getToolDefinition().description());
|
|
}
|
|
} else {
|
|
log.warn("MCP 工具已启用但无可注册的工具 [{}]，请检查：", cacheKey);
|
|
log.warn(" 1. 是否在「MCP 服务管理」页面添加了 MCP Server 配置？");
|
|
log.warn(" 2. 配置是否已启用（is_active=true）？");
|
|
log.warn(" 3. MCP Server 是否连接成功（点击「测试连接」验证）？");
|
|
log.warn(" 4. MCP Server 是否暴露了工具（listTools 返回非空）？");
|
|
}
|
|
} else {
|
|
log.info("MCP 工具未启用 [enableMcpTools={}, effectiveTools={}]", enableMcpTools, effectiveTools);
|
|
}
|
|
return builder
|
|
.defaultAdvisors(
|
|
contentSafetyAdvisor,
|
|
MessageChatMemoryAdvisor.builder(chatMemory).build(),
|
|
new MyLoggerAdvisor()
|
|
)
|
|
.build();
|
|
});
|
|
}
|
|
|
|
public void clearCache() {
|
|
chatClientCache.clear();
|
|
log.info("AssistantApp ChatClient cache cleared");
|
|
}
|
|
|
|
// ==================== 统一入口（新） ====================
|
|
|
|
/**
|
|
* 同步对话（新入口，委托 {@link ChatPipeline} 编排）。
|
|
*
|
|
* @param ctx 对话上下文
|
|
* @return AI 回答文本
|
|
*/
|
|
public String chat(ChatContext ctx) {
|
|
return chatWithEvents(ctx).text();
|
|
}
|
|
|
|
/**
|
|
* 同步对话 + MCP 工具调用事件（新入口）。
|
|
* 比 {@link #chat(ChatContext)} 多返回本次触发的工具调用事件，供需要展示调用过程的场景。
|
|
*
|
|
* @param ctx 对话上下文
|
|
* @return 回答文本 + MCP 事件
|
|
*/
|
|
public ChatResult chatWithEvents(ChatContext ctx) {
|
|
// 熔断：全局 AI 调用处于熔断状态，直接返回降级提示
|
|
if (aiCircuitBreaker.isOpen(AI_CIRCUIT_KEY)) {
|
|
log.warn("AI 调用熔断中，返回降级提示");
|
|
return new ChatResult(CIRCUIT_OPEN_MESSAGE, List.of());
|
|
}
|
|
ChatRequest req = chatPipeline.buildRequest(ctx);
|
|
if (req.faqHit()) {
|
|
return new ChatResult(req.faqAnswer().get(), List.of());
|
|
}
|
|
McpToolCallback.resetEvents();
|
|
McpToolCallback.resetCallRounds();
|
|
try {
|
|
ChatClient.ChatClientRequestSpec spec = getChatClient(ctx.appType(), ctx.allowedMcpTools())
|
|
.prompt()
|
|
.user(req.finalMessage())
|
|
.advisors(s -> s.param(CONVERSATION_ID, ctx.chatId()));
|
|
if (StringUtils.hasText(req.finalSystemPrompt())) {
|
|
spec = spec.system(req.finalSystemPrompt());
|
|
}
|
|
String text = spec.call().chatResponse().getResult().getOutput().getText();
|
|
aiCircuitBreaker.recordSuccess(AI_CIRCUIT_KEY);
|
|
|
|
// 推荐问题已不再由主回复同步生成，改由 SuggestionGenerator 异步按需生成
|
|
return new ChatResult(text, McpToolCallback.drainEvents(), List.of());
|
|
} catch (Exception e) {
|
|
aiCircuitBreaker.recordFailure(AI_CIRCUIT_KEY);
|
|
log.error("AI 同步调用失败: chatId={}, error={}", ctx.chatId(), e.getMessage());
|
|
return new ChatResult("抱歉，AI 服务调用失败：" + e.getMessage(), List.of());
|
|
}
|
|
}
|
|
|
|
/**
|
|
* 流式对话（新入口，委托 {@link ChatPipeline} 编排）。
|
|
* <p>
|
|
* 注意：返回类型为 Flux<String>，Spring WebFlux 会对每个元素做 data: 帧包装。
|
|
* 因此 Flux 元素必须只是纯文本内容，不应包含 event: / data: 等 SSE 协议行。
|
|
*
|
|
* @param ctx 对话上下文
|
|
* @return 纯文本流式回答（每个元素为一段自然语言文本）
|
|
*/
|
|
public Flux<String> chatStream(ChatContext ctx) {
|
|
// 熔断：全局 AI 调用处于熔断状态
|
|
if (aiCircuitBreaker.isOpen(AI_CIRCUIT_KEY)) {
|
|
log.warn("AI 调用熔断中（流式），返回降级提示");
|
|
return Flux.just(CIRCUIT_OPEN_MESSAGE);
|
|
}
|
|
ChatRequest req = chatPipeline.buildRequest(ctx);
|
|
if (req.faqHit()) {
|
|
// FAQ 命中：整段答案原样输出，由 SSE 编码器处理内部换行，
|
|
// 后端不做任何格式增删（不拆行、不加换行、不补空格）。
|
|
return Flux.just(req.faqAnswer().get());
|
|
}
|
|
McpToolCallback.resetEvents();
|
|
McpToolCallback.resetCallRounds();
|
|
ChatClient.ChatClientRequestSpec spec = getChatClient(ctx.appType(), ctx.allowedMcpTools())
|
|
.prompt()
|
|
.user(req.finalMessage())
|
|
.advisors(s -> s.param(CONVERSATION_ID, ctx.chatId()));
|
|
if (StringUtils.hasText(req.finalSystemPrompt())) {
|
|
spec = spec.system(req.finalSystemPrompt());
|
|
}
|
|
// 原始文本流；推荐问题已不再由主回复同步生成，改由 SuggestionGenerator 异步按需生成
|
|
Flux<String> rawStream = spec.stream().content();
|
|
return preserveTrailingWhitespace(rawStream)
|
|
.doOnComplete(() -> aiCircuitBreaker.recordSuccess(AI_CIRCUIT_KEY))
|
|
.doOnError(e -> {
|
|
aiCircuitBreaker.recordFailure(AI_CIRCUIT_KEY);
|
|
log.error("AI 流式调用失败: chatId={}, error={}", ctx.chatId(), e.getMessage());
|
|
})
|
|
.doFinally(signalType -> {
|
|
// 确保 ThreadLocal 清理，防止线程池复用时数据残留
|
|
McpToolCallback.resetEvents();
|
|
McpToolCallback.resetCallRounds();
|
|
})
|
|
.onErrorResume(e -> Flux.just("抱歉，AI 服务调用失败：" + e.getMessage()));
|
|
}
|
|
|
|
/**
|
|
* 缓冲以空白字符结尾的 chunk，将其与下一个 chunk 合并后再发出。
|
|
* <p>
|
|
* 原因：前端 SSE 解析会对每行执行 trim()，如果 chunk 以空白结尾（如 Markdown 标题 "## "），
|
|
* 行尾空白会被削掉，导致 "## " + "一、..." 拼成 "##一、..."，破坏 Markdown 渲染。
|
|
* 本方法不改变大模型输出的文本内容，只调整 chunk 边界以避开 SSE 的空白截断。
|
|
*
|
|
* @param source 原始大模型输出流
|
|
* @return 调整后的流，每个元素均不以空白字符结尾（流末尾除外）
|
|
*/
|
|
private static Flux<String> preserveTrailingWhitespace(Flux<String> source) {
|
|
return Flux.create(sink -> {
|
|
StringBuilder buffer = new StringBuilder();
|
|
// 捕获内部订阅，下游取消/释放时同步取消上游，避免资源泄漏
|
|
Disposable subscription = source.subscribe(
|
|
chunk -> {
|
|
if (chunk == null) {
|
|
return;
|
|
}
|
|
buffer.append(chunk);
|
|
String current = buffer.toString();
|
|
if (current.isEmpty()) {
|
|
return;
|
|
}
|
|
char last = current.charAt(current.length() - 1);
|
|
// 当前累积内容不以空白结尾，可以安全发出
|
|
if (!Character.isWhitespace(last)) {
|
|
sink.next(current);
|
|
buffer.setLength(0);
|
|
} else if (buffer.length() >= MAX_TRAILING_WHITESPACE_BUFFER) {
|
|
// 尾部空白过长（如纯空白输出），强制发出以限制内存占用
|
|
sink.next(current);
|
|
buffer.setLength(0);
|
|
}
|
|
// 若以空白结尾，继续缓存，等下一个 chunk
|
|
},
|
|
e -> {
|
|
// 出错前先把已缓冲内容发出，避免已生成文本丢失
|
|
if (buffer.length() > 0) {
|
|
sink.next(buffer.toString());
|
|
buffer.setLength(0);
|
|
}
|
|
sink.error(e);
|
|
},
|
|
() -> {
|
|
if (buffer.length() > 0) {
|
|
sink.next(buffer.toString());
|
|
}
|
|
sink.complete();
|
|
}
|
|
);
|
|
sink.onCancel(subscription);
|
|
sink.onDispose(subscription);
|
|
}, FluxSink.OverflowStrategy.BUFFER);
|
|
}
|
|
|
|
/**
|
|
* 统一检索引用来源（新入口，委托 {@link ChatPipeline#retrieveSources}）。
|
|
*
|
|
* @param ctx 对话上下文（使用 message / rewriteStrategy / categoryIds）
|
|
* @return 命中的知识库片段，含 metadata（documentId/title/sourceName/chunkIndex/distance）
|
|
*/
|
|
public List<Document> retrieveSources(ChatContext ctx) {
|
|
return chatPipeline.retrieveSources(ctx);
|
|
}
|
|
|
|
// ==================== 内部工具方法已移除，不再需要 ====================
|
|
}
|