- * 在请求线程捕获租户上下文快照，后台启动 Plan → Execute 流程，同时返回事件流供前端订阅。 - * 编排流程在 boundedElastic 调度器上执行，不阻塞当前线程。 + * 重构后流程（接入 Memory 模块）： + *
- * 关键点：必须在请求线程捕获 {@link TenantContext}（含 tenantId/userId/token/RequestAttributes/Exchange）， - * 因为 AuthInfoUtils 和 RequestContextHolder 不支持异步线程访问，编排切到 boundedElastic 后 - * 需要用快照构建 ToolContext 和 ChatClient。 - *
- *- * 文件解析：若请求携带 fileIds，在请求线程同步加载文件内容（Feign 调用 mf-storage）， - * 拼接到用户提示词前。必须在请求线程执行，因为 Feign 的 BearerTokenInterceptor 依赖 - * RequestContextHolder 中继令牌，异步线程拿不到。 + * 关键点：必须在请求线程完成 Memory 写入（TenantContext 绑定 + 文件解析）， + * 因为 AuthInfoUtils 和 Feign BearerTokenInterceptor 都依赖 RequestContextHolder， + * 切到 boundedElastic 异步线程后拿不到请求上下文。 *
*
* 所有返回的 {@link ChatResponseVo} 的 id 字段填充为 {@link AiRequest#getId()}，
@@ -88,14 +94,25 @@ public class AgentRuntime {
String requestId = aiRequest.getId();
String sessionId = aiRequest.getSessionId();
String prompt = aiRequest.getMessage() != null ? aiRequest.getMessage().getContent() : null;
- // 用请求 id 作为事件 id，与普通聊天返回结构一致
EventBus eventBus = new EventBus(requestId);
- // 在请求线程捕获租户上下文快照，供异步编排使用
- TenantContext tenantContext = captureTenantContext();
+ // 获取（或创建）会话 Memory
+ ConversationMemory memory = memoryStore.getOrCreate(sessionId);
- // 文件解析必须在请求线程执行：Feign BearerTokenInterceptor 依赖 RequestContextHolder
- prompt = resolveFileContents(prompt, aiRequest.getFileIds());
+ // 在请求线程捕获租户上下文并绑定到 Memory 的 Vars Context
+ TenantContext tenantContext = captureTenantContext();
+ memory.bindTenantContext(tenantContext);
+
+ // 文件解析 + 注入 Memory 的 Document Context（必须在请求线程执行）
+ List
- * 若 fileIds 为空或全部加载失败，返回原始 prompt。
- * 文件内容加载失败时记录日志但不阻断流程，降级为纯文本对话。
- *
diff --git a/mf-business/mf-ai/src/main/java/cn/com/mfish/ai/agent/BaseAssistant.java b/mf-business/mf-ai/src/main/java/cn/com/mfish/ai/agent/BaseAssistant.java
index 198a1001..d9f75be2 100644
--- a/mf-business/mf-ai/src/main/java/cn/com/mfish/ai/agent/BaseAssistant.java
+++ b/mf-business/mf-ai/src/main/java/cn/com/mfish/ai/agent/BaseAssistant.java
@@ -7,6 +7,8 @@ import cn.com.mfish.common.ai.engine.ApiToolEngine;
import cn.com.mfish.common.ai.entity.AiRequest;
import cn.com.mfish.common.ai.entity.ChatResponseVo;
import cn.com.mfish.common.ai.agent.ToolCapable;
+import cn.com.mfish.common.ai.memory.ConversationMemory;
+import cn.com.mfish.common.ai.memory.ConversationMemoryStore;
import cn.com.mfish.common.core.constants.RPCConstants;
import cn.com.mfish.common.core.utils.AuthInfoUtils;
import cn.com.mfish.common.core.utils.ServletUtils;
@@ -61,6 +63,19 @@ public abstract class BaseAssistant implements IClientAssistant, ToolCapable {
@Autowired
protected FileParseService fileParseService;
+ /**
+ * 会话记忆存储：三驾马车 Memory 模块的入口
+ *
+ * 用于按 sessionId 获取 {@link ConversationMemory}，将文件解析结果、租户上下文、
+ * 业务变量等统一收纳到 Memory，供后续 Planner 拼接上下文。
+ *
+ * 字段注入避免修改所有子类构造函数。
+ *
- * 若请求携带fileIds，会先通过FileParseService从文件服务获取文件内容，
- * 将其拼接到用户提示词前，再交由子类chat(sessionId, prompt)处理。
+ * 重构后流程（接入 Memory 模块）：
+ *
+ * 关键约束：文件解析必须在请求线程执行（Feign BearerTokenInterceptor 依赖
+ * RequestContextHolder），Memory 写入也在请求线程完成（保证一致性）。
*
* @return 聊天信息
*/
@Override
public Flux
- * 调用 LLM 将用户原始需求拆解为结构化的 {@link AgentPlan}：
+ * 重构后职责（第二阶段存量平移）：
*
+ * 与旧版差异：
+ *
+ *
+ *
- *
*
+ *
+ *
* 采用结构化输出（responseEntity），LLM 直接返回 JSON 反序列化为 AgentPlan。 * 必须在请求线程解析租户并构建 ChatClient（AuthInfoUtils 不支持异步）。 *
@@ -45,22 +62,36 @@ public class Planner { private final ChatMemory chatMemory; private final LlmModelRouter llmModelRouter; + private final CapabilityEngine capabilityEngine; + private final ConversationMemoryStore memoryStore; - public Planner(ChatMemory chatMemory, LlmModelRouter llmModelRouter) { + public Planner(ChatMemory chatMemory, LlmModelRouter llmModelRouter, + CapabilityEngine capabilityEngine, ConversationMemoryStore memoryStore) { this.chatMemory = chatMemory; this.llmModelRouter = llmModelRouter; + this.capabilityEngine = capabilityEngine; + this.memoryStore = memoryStore; } /** * 规划：将用户需求拆解为步骤列表 *- * 注意：本方法会被异步调度器（boundedElastic）调用，因此 ChatClient 必须在请求线程 - * 预构建并通过闭包传入。{@link TenantContext} 中包含请求线程捕获的 tenantId， + * 重构后流程： + *
+ * 注意：本方法会被异步调度器（boundedElastic）调用，因此 ChatClient 和系统提示词 + * 必须在请求线程预构建并通过闭包传入。{@link TenantContext} 中包含请求线程捕获的 tenantId， * 用于路由到该租户的 ChatModel。 *
* * @param sessionId 会话ID - * @param prompt 用户原始需求 + * @param prompt 用户原始需求（已由 AgentRuntime 拼接 Document Context） * @param tenantContext 请求线程捕获的租户上下文 * @return 执行计划 */ @@ -70,16 +101,26 @@ public class Planner { ? tenantContext.getTenantId() : llmModelRouter.currentTenantId(); ChatClient chatClient = getChatClient(tenantId); + + // 从 Memory 读取系统上下文（Vars + Document），拼接到 prompt 前 + String systemContext = resolveSystemContext(sessionId); + + // 从 CapabilityEngine 获取动作列表，构建规划提示词 String systemPrompt = buildPlannerPrompt(); + // 拼接最终 prompt：系统上下文 + 用户需求 + String finalPrompt = StringUtils.isNotEmpty(systemContext) + ? systemContext + "\n用户需求：" + prompt + : prompt; + return Mono.fromCallable(() -> { var responseEntity = chatClient.prompt() .system(systemPrompt) - .user(prompt) + .user(finalPrompt) .advisors(a -> a.param(CONVERSATION_ID, sessionId)) .call() .responseEntity(AgentPlan.class); - AgentPlan plan = Objects.requireNonNullElseGet(responseEntity.entity(), () -> fallbackPlan(prompt)); + AgentPlan plan = Objects.requireNonNullElseGet(responseEntity.entity(), () -> fallbackPlan(finalPrompt)); plan.setOriginalPrompt(prompt); log.info("[Planner] 规划完成, 步骤数={}, summary={}", plan.getSteps() != null ? plan.getSteps().size() : 0, plan.getSummary()); @@ -88,17 +129,63 @@ public class Planner { .subscribeOn(Schedulers.boundedElastic()) .onErrorResume(ex -> { log.error("[Planner] 规划失败，降级为单步执行", ex); - return Mono.just(fallbackPlan(prompt)); + return Mono.just(fallbackPlan(finalPrompt)); }); } /** - * 构建规划师系统提示词，包含可用服务列表 + * 从 Memory 读取系统上下文（Vars Context + Document Context） + *+ * AgentRuntime 已在请求线程将 DocumentContext 拼接到 prompt 中， + * 此处进一步补充 Vars Context（租户信息、业务变量）， + * 使 Planner 能感知用户身份和业务上下文。 + *
+ */ + private String resolveSystemContext(String sessionId) { + try { + ConversationMemory memory = memoryStore.getOrCreate(sessionId); + return memory.getSystemContext(); + } catch (Exception e) { + log.warn("[Planner] 读取 Memory 系统上下文失败 sessionId={}", sessionId, e); + return ""; + } + } + + /** + * 构建规划师系统提示词，包含可用服务列表和动作摘要 + *+ * 重构后从 {@link CapabilityEngine#getAvailableActions()} 动态获取动作列表， + * 按服务分组展示。相比旧版从 {@code ServiceConstants.MfService.values()} 硬编码枚举获取， + * 新版自动反映已注册的子引擎状态（如某服务未启动则不出现在列表中）。 + *
*/ private String buildPlannerPrompt() { - String serviceList = Arrays.stream(ServiceConstants.MfService.values()) - .map(s -> String.format(" - %s: %s", s.getValue(), s.getGatewayPrefix())) - .collect(Collectors.joining("\n")); + // 从 CapabilityEngine 获取所有动作，按 serviceId 分组 + List+ * 重构后从 {@link CapabilityEngine#getAvailableActions()} 提取所有 serviceId， + * 相比旧版从 {@code ServiceConstants.MfService.values()} 硬编码枚举获取， + * 新版仅聚合实际有动作的服务，避免向未启动的服务发送请求。 + *
*/ private AgentPlan fallbackPlan(String prompt) { - List* 遍历fileKey列表，逐个获取文件元数据与内容，将文本/Office文档内容拼接到StringBuilder。 * 单个文件读取失败不影响其他文件，异常会被记录日志并跳过。 + *
+ * 旧版接口，保留向后兼容。新调用方建议使用 {@link #loadAsChunks} 获取结构化的 + * {@link DocumentChunk} 列表，再注入到 Memory 模块。 * * @param fileIds 文件fileKey列表 * @return 拼接好的文件内容提示词片段；列表为空或全部失败时返回空字符串 @@ -141,6 +147,45 @@ public class FileParseService { return sb.toString(); } + /** + * 加载文件并解析为 {@link DocumentChunk} 列表 + *
+ * 与 {@link #loadFileContents} 的区别：返回结构化的文档块列表，便于 Memory 模块 + * 按 chunk 维度管理文档上下文，支持后续按相关性检索（RAG）、按 token 上限分片注入等扩展。 + *
+ *+ * 当前实现：每个文件产出 1 个 Chunk（小文件场景），未来可扩展为按段落/页码/字符数分片。 + * Chunk 内容已截断到 {@link #MAX_CONTENT_CHARS}，可直接注入 LLM 上下文。 + *
+ *+ * 必须在请求线程调用：内部通过 Feign 调用 mf-storage，BearerTokenInterceptor 依赖 + * RequestContextHolder 中继令牌。 + *
+ * + * @param fileIds 文件fileKey列表 + * @return 文档块列表；列表为空或全部失败时返回空列表（不会返回 null） + */ + public List@@ -176,6 +221,47 @@ public class FileParseService { return buildBinaryFileHint(fileName, fileType, storageInfo.getFileSize()); } + /** + * 加载单个文件并构建为 {@link DocumentChunk} + *
+ * 与 {@link #loadOneFile} 共享文件元数据查询和解析路径，但返回结构化 Chunk 而非拼接文本。 + * 二进制文件（无法解析内容的）返回 null 而非提示片段，避免无效 Chunk 进入 Memory。 + *
+ * + * @param fileKey 文件key + * @return 文档块；文件不存在或无法解析返回 null + */ + private DocumentChunk loadOneFileAsChunk(String fileKey) { + Result+ * 能力引擎向 Planner 暴露的单个可执行动作的元数据，遵循 OpenAI Tool 格式。 + * 一个 {@link CapabilitySubEngine} 可暴露多个 Action，{@link CapabilityEngine} + * 将所有子引擎的 Action 合并为统一列表供 Planner 选择。 + *
+ *+ * 与 Spring AI ToolCallback 的关系： + *
+ * 示例： + *
{@code
+ * {
+ * "type": "object",
+ * "properties": {
+ * "userId": { "type": "string", "description": "用户ID" }
+ * },
+ * "required": ["userId"]
+ * }
+ * }
+ *
+ */
+ private String inputSchema;
+
+ /**
+ * 来源引擎类型
+ */
+ private EngineType engineType;
+
+ /**
+ * 来源服务ID（仅 TOOL 引擎有值，标识来自哪个微服务；其他引擎为 null）
+ */
+ private String serviceId;
+}
diff --git a/mf-common/mf-common-ai/src/main/java/cn/com/mfish/common/ai/capability/ActionResult.java b/mf-common/mf-common-ai/src/main/java/cn/com/mfish/common/ai/capability/ActionResult.java
new file mode 100644
index 00000000..999d851f
--- /dev/null
+++ b/mf-common/mf-common-ai/src/main/java/cn/com/mfish/common/ai/capability/ActionResult.java
@@ -0,0 +1,69 @@
+package cn.com.mfish.common.ai.capability;
+
+import lombok.Data;
+import lombok.experimental.Accessors;
+
+/**
+ * 动作执行结果
+ * + * 由 {@link CapabilitySubEngine#execute} 返回，封装执行状态、输出内容和错误信息。 + * 设计为不可变值对象：执行完成后构造，调用方只读。 + *
+ * + * @author: mfish + * @date: 2026/07/21 + */ +@Data +@Accessors(chain = true) +public class ActionResult { + + /** + * 是否执行成功 + */ + private boolean success; + + /** + * 执行输出（LLM 可读的文本，通常为 JSON 字符串或纯文本） + *+ * 成功时填充结果内容，失败时可为 null。 + *
+ */ + private String output; + + /** + * 错误信息（失败时填充，成功时为 null） + */ + private String error; + + /** + * 来源引擎类型（便于 Planner 日志追踪和结果分流处理） + */ + private EngineType engineType; + + /** + * 执行耗时（毫秒） + */ + private long durationMs; + + /** + * 构建成功结果 + */ + public static ActionResult success(EngineType engineType, String output, long durationMs) { + return new ActionResult() + .setSuccess(true) + .setOutput(output) + .setEngineType(engineType) + .setDurationMs(durationMs); + } + + /** + * 构建失败结果 + */ + public static ActionResult failure(EngineType engineType, String error, long durationMs) { + return new ActionResult() + .setSuccess(false) + .setError(error) + .setEngineType(engineType) + .setDurationMs(durationMs); + } +} diff --git a/mf-common/mf-common-ai/src/main/java/cn/com/mfish/common/ai/capability/CapabilityAutoConfiguration.java b/mf-common/mf-common-ai/src/main/java/cn/com/mfish/common/ai/capability/CapabilityAutoConfiguration.java new file mode 100644 index 00000000..11a5dae0 --- /dev/null +++ b/mf-common/mf-common-ai/src/main/java/cn/com/mfish/common/ai/capability/CapabilityAutoConfiguration.java @@ -0,0 +1,123 @@ +package cn.com.mfish.common.ai.capability; + +import cn.com.mfish.common.ai.engine.ApiToolEngine; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.SmartInitializingSingleton; +import org.springframework.boot.autoconfigure.AutoConfiguration; +import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.context.annotation.Bean; + +import java.util.List; + +/** + * 能力引擎自动配置 + *+ * 注册 {@link CapabilityEngine} 作为能力引擎门面，自动发现所有 {@link CapabilitySubEngine} 实现， + * 在所有单例 Bean 初始化完成后将子引擎注册到 CapabilityEngine。 + *
+ *+ * 默认注册两个子引擎： + *
+ * 与 ApiToolAutoConfiguration 的关系： + *
+ * 通过 META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports 注册。 + *
+ * + * @author: mfish + * @date: 2026/07/21 + */ +@Slf4j +@AutoConfiguration +public class CapabilityAutoConfiguration { + + /** + * 能力引擎门面 + */ + @Bean + @ConditionalOnMissingBean + public CapabilityEngine capabilityEngine() { + return new CapabilityEngine(); + } + + /** + * 工具能力引擎适配器（包装 ApiToolEngine 为 CapabilitySubEngine） + *+ * 依赖 ApiToolEngine Bean，由 ApiToolAutoConfiguration 注册。 + *
+ */ + @Bean + public ToolCapabilityEngine toolCapabilityEngine(ApiToolEngine apiToolEngine) { + return new ToolCapabilityEngine(apiToolEngine); + } + + /** + * MCP 能力引擎（条件注册） + *+ * 仅当 classpath 中存在 MCP SDK 类 且 容器中有 {@link McpServerConfigProvider} 实现时才注册。 + *
+ *+ * 条件说明： + *
+ * 放在此处而非 CapabilityEngine 内，避免 CapabilityEngine 依赖 Spring 容器回调接口，保持可测试性。 + * 与 ApiToolAutoConfiguration 的 apiToolEngineInitializer 模式一致。 + *
+ *+ * 对于 McpCapabilityEngine，在注册前先调用 {@code refresh()} 连接所有 MCP 服务器并发现工具， + * 确保 {@code getActions()} 在注册到 CapabilityEngine 时返回完整的动作列表。 + *
+ */ + @Bean + public SmartInitializingSingleton capabilityEngineInitializer(CapabilityEngine capabilityEngine, + List+ * 组合（Composite）所有 {@link CapabilitySubEngine} 实例，向上为 Planner 提供统一的能力发现与执行入口。 + * 屏蔽异构执行细节：Planner 无需关心动作来自 TOOL / SKILL / MCP / WORKFLOW 哪个引擎。 + *
+ *+ * 架构定位： + *
+ * CapabilityEngine（门面） + * │ + * ┌───────────────┼───────────────┬───────────────┬───────────────┐ + * │ │ │ │ + * ToolCapability SkillCapability McpCapability WorkflowCapability + * (Engine) (Engine) (Engine) (Engine) + * │ │ │ │ + * ApiToolEngine Java Skill MCP Client Flowable/BPMN + * (Feign+HTTP) + *+ * + *
+ * 与现有 ApiToolEngine 的关系： + *
+ * 线程安全：子引擎注册在初始化期完成，运行期只读；动作查找通过 ConcurrentHashMap 支持并发。 + *
+ * + * @author: mfish + * @date: 2026/07/21 + */ +@Slf4j +public class CapabilityEngine { + + /** + * 已注册的子引擎列表（按注册顺序） + */ + private final List+ * 初始化时构建，子引擎动态新增动作时需调用 {@link #refreshActionIndex()} 重建。 + * 同名动作按子引擎注册顺序，先注册的覆盖（与 ApiToolEngine 去重策略一致）。 + *
+ */ + private volatile Map+ * 在 {@link CapabilityAutoConfiguration} 初始化期调用，运行期不应动态增减。 + * 注册后自动重建动作索引。 + *
+ * + * @param subEngine 子引擎实例 + */ + public synchronized void registerSubEngine(CapabilitySubEngine subEngine) { + if (subEngine == null) { + return; + } + subEngines.add(subEngine); + log.info("[CapabilityEngine] 注册子引擎 type={} actions={}", + subEngine.getEngineType(), + subEngine.getActions() != null ? subEngine.getActions().size() : 0); + rebuildActionIndex(); + } + + /** + * 批量注册子引擎 + * + * @param engines 子引擎列表 + */ + public synchronized void registerSubEngines(List+ * Planner 调用此方法获取全局动作列表，可选择： + *
+ * 按 actionName 路由到对应子引擎执行。Planner 显式调用模式走此方法。 + *
+ * + * @param actionName 动作名称（需与 ActionDefinition.name 一致） + * @param params 动作参数 + * @param ctx 执行上下文 + * @return 执行结果；动作不存在时返回 failure + */ + public ActionResult executeAction(String actionName, Map+ * 同名动作按子引擎注册顺序，先注册的保留（与 ApiToolEngine 去重策略一致）。 + *
+ */ + private void rebuildActionIndex() { + Map+ * 屏蔽异构执行细节的统一抽象。每种能力来源（Tool / Skill / MCP / Workflow） + * 实现此接口，由 {@link CapabilityEngine} 门面组合所有子引擎实例。 + *
+ *+ * 三种执行模式的关系： + *
+ * 线程安全：实现类需保证 getActions() 和 execute() 线程安全， + * 因为可能被多个请求线程并发调用。 + *
+ * + * @author: mfish + * @date: 2026/07/21 + */ +public interface CapabilitySubEngine { + + /** + * 引擎类型识别 + * + * @return 引擎类型枚举（TOOL / SKILL / MCP / WORKFLOW） + */ + EngineType getEngineType(); + + /** + * 向 Planner 暴露的可用动作元数据 + *+ * 遵循 OpenAI Tool 格式，{@link CapabilityEngine} 将所有子引擎的返回值合并为统一列表。 + * 动作名需全局唯一（建议加引擎前缀，如 {@code tool.xxx}、{@code skill.xxx}）。 + *
+ * + * @return 动作定义列表，无可用动作时返回空列表 + */ + List+ * 显式调用入口：按 actionName 查找并执行对应动作。 + * 实现类需处理 actionName 不存在的情况（返回 failure 结果，不抛异常）。 + *
+ * + * @param actionName 动作名称（需与 {@link ActionDefinition#getName()} 一致） + * @param params 动作参数（key 为参数名，value 为参数值） + * @param ctx 执行上下文（租户信息、会话ID、自定义属性） + * @return 执行结果 + */ + ActionResult execute(String actionName, Map+ * 标识 {@link CapabilitySubEngine} 的具体来源，供 Planner 决策时区分能力来源， + * 也用于日志和调试。新增引擎类型时在此枚举扩展。 + *
+ * + * @author: mfish + * @date: 2026/07/21 + */ +public enum EngineType { + + /** + * 工具引擎：基于 Feign/OpenAPI 的微服务接口工具 + *+ * 适配现有 {@code ApiToolEngine} + {@code ToolProvider} 体系， + * 将 Spring AI {@code ToolCallback} 转换为 {@link ActionDefinition}。 + *
+ */ + TOOL, + + /** + * 技能引擎：Java 编写的预置技能（如文件解析、数据格式化、SQL 生成等） + *+ * 与 TOOL 的区别：TOOL 是远程接口的包装，SKILL 是本地 Java 逻辑的包装。 + * SKILL 不依赖网络调用，执行延迟低、结果确定性强。 + *
+ */ + SKILL, + + /** + * MCP 引擎：Model Context Protocol 客户端工具 + *+ * 对接外部 MCP Server，将 MCP 工具转换为统一的 ActionDefinition。 + * 当前为预留类型，待 Spring AI MCP 集成后实现。 + *
+ */ + MCP, + + /** + * 工作流引擎：基于 Flowable/BPMN 的流程编排能力 + *+ * 将工作流的"发起流程/审批/查询任务"等操作封装为 Action， + * 供 Planner 在需要人工审批或复杂流程编排时调用。 + *
+ */ + WORKFLOW; +} diff --git a/mf-common/mf-common-ai/src/main/java/cn/com/mfish/common/ai/capability/ExecutionContext.java b/mf-common/mf-common-ai/src/main/java/cn/com/mfish/common/ai/capability/ExecutionContext.java new file mode 100644 index 00000000..bd3b881d --- /dev/null +++ b/mf-common/mf-common-ai/src/main/java/cn/com/mfish/common/ai/capability/ExecutionContext.java @@ -0,0 +1,88 @@ +package cn.com.mfish.common.ai.capability; + +import cn.com.mfish.common.ai.agent.TenantContext; +import lombok.Data; +import lombok.experimental.Accessors; + +import java.util.HashMap; +import java.util.Map; + +/** + * 能力执行上下文 + *+ * 封装动作执行时所需的环境信息，由调用方（Planner / Executor / BaseAssistant）构建， + * 传递给 {@link CapabilitySubEngine#execute}。 + *
+ *+ * 与 Spring AI ToolContext 的关系： + *
+ * 由请求线程捕获，供异步执行线程恢复认证上下文。 + * 可为 null（如离线测试场景），各子引擎需做空值防御。 + *
+ */ + private TenantContext tenantContext; + + /** + * 会话ID（用于日志追踪和会话级状态关联） + */ + private String sessionId; + + /** + * 自定义属性（扩展通道，子引擎可按约定存取特定 key） + *+ * 常用 key（可选，非强制约定）： + *
+ * 实现 {@link CapabilitySubEngine} 接口，对接外部 MCP (Model Context Protocol) Server， + * 将远程工具转换为统一的 {@link ActionDefinition}。 + *
+ *+ * 架构定位： + *
+ * CapabilityEngine（门面） + * │ + * ├── ToolCapabilityEngine (Feign/OpenAPI) + * ├── McpCapabilityEngine (本类，MCP 协议) + * ├── [SkillCapabilityEngine] (未来扩展) + * └── [WorkflowCapabilityEngine] (未来扩展) + *+ * + *
+ * 初始化流程： + *
+ * 动作名约定：{@code mcp.{serverName}.{toolName}}
+ *
示例：{@code mcp.filesystem.readFile}、{@code mcp.github.searchRepos}
+ *
+ * 线程安全：clients 和 actions 映射使用 ConcurrentHashMap，支持并发读取。 + * 初始化在 {@link cn.com.mfish.common.ai.capability.CapabilityAutoConfiguration} 的 + * SmartInitializingSingleton 中触发。 + *
+ * + * @author: mfish + * @date: 2026/07/21 + */ +@Slf4j +@SuppressWarnings("deprecation") +public class McpCapabilityEngine implements CapabilitySubEngine { + + /** + * stdio 传输类型标识 + */ + private static final String TRANSPORT_STDIO = "stdio"; + + /** + * SSE 传输类型标识 + */ + private static final String TRANSPORT_SSE = "sse"; + + /** + * Streamable HTTP 传输类型标识（MCP 2025-03-26 规范，推荐用于新服务器） + */ + private static final String TRANSPORT_STREAMABLE = "streamable"; + + /** + * MCP 动作名前缀 + */ + private static final String MCP_ACTION_PREFIX = "mcp."; + + /** + * MCP 服务器配置提供者（由业务层注入，查询数据库） + */ + private final McpServerConfigProvider configProvider; + + /** + * MCP 客户端映射：serverName → McpSyncClient + */ + private final Map+ * 返回的是初始化时缓存的快照，不会实时调用 MCP 服务器。 + * 如需刷新，调用 {@link #refresh()}。 + *
+ */ + @Override + public List+ * 按 actionName 从 actionRegistry 查找对应的 McpSyncClient 和原始 toolName， + * 调用 {@code client.callTool(CallToolRequest)} 执行工具。 + *
+ */ + @Override + public ActionResult execute(String actionName, Map+ * 由 {@link CapabilityAutoConfiguration} 的 SmartInitializingSingleton 触发。 + * 某个服务器连接失败不影响其他服务器。 + *
+ */ + public synchronized void refresh() { + if (configProvider == null) { + log.warn("[McpCapabilityEngine] 无 McpServerConfigProvider，跳过初始化"); + return; + } + + List+ * 与 SSE 的区别： + *
+ * 字段复用：sseUrl 作为基础 URL，sseEndpoint 作为端点路径（通常为 /mcp）。 + *
+ */ + private McpSyncClient connectStreamable(McpServerInfo config) { + if (StringUtils.isEmpty(config.getSseUrl())) { + log.warn("[McpCapabilityEngine] streamable 模式缺少 sseUrl 参数 server={}", config.getServerName()); + return null; + } + + WebClient.Builder webClientBuilder = WebClient.builder() + .baseUrl(config.getSseUrl()); + + // 注入认证 Token + if (StringUtils.isNotEmpty(config.getAuthToken())) { + webClientBuilder.defaultHeader("Authorization", "Bearer " + config.getAuthToken()); + } + + WebClientStreamableHttpTransport.Builder transportBuilder = WebClientStreamableHttpTransport.builder(webClientBuilder); + if (StringUtils.isNotEmpty(config.getSseEndpoint())) { + transportBuilder.endpoint(config.getSseEndpoint()); + } + + WebClientStreamableHttpTransport transport = transportBuilder.build(); + return createAndInitialize(transport, config.getServerName()); + } + + /** + * 创建 McpSyncClient 并初始化连接 + */ + private McpSyncClient createAndInitialize(McpClientTransport transport, String serverName) { + McpSyncClient client = McpClient.sync(transport) + .requestTimeout(Duration.ofSeconds(60)) + .build(); + client.initialize(); + log.info("[McpCapabilityEngine] MCP 客户端已初始化 server={}", serverName); + return client; + } + + /** + * 构建 MCP 动作名：mcp.{serverName}.{toolName} + */ + private String buildActionName(String serverName, String toolName) { + return MCP_ACTION_PREFIX + serverName + "." + toolName; + } + + /** + * 从 CallToolResult 提取文本内容 + */ + private String extractTextContent(McpSchema.CallToolResult result) { + if (result == null || result.content() == null || result.content().isEmpty()) { + return ""; + } + StringBuilder sb = new StringBuilder(); + for (McpSchema.Content content : result.content()) { + if (content instanceof McpSchema.TextContent tc) { + sb.append(tc.text()); + } else { + // 非 TextContent 类型，序列化为 JSON + sb.append(JSON.toJSONString(content)); + } + sb.append("\n"); + } + return sb.toString().trim(); + } + + /** + * 将 MCP Tool 的 inputSchema 序列化为 JSON 字符串 + */ + private String serializeSchema(Object inputSchema) { + if (inputSchema == null) { + return "{}"; + } + try { + return JSON.toJSONString(inputSchema); + } catch (Exception e) { + log.warn("[McpCapabilityEngine] 序列化 inputSchema 失败", e); + return "{}"; + } + } + + /** + * 清理所有连接和动作注册表 + */ + private void clearAll() { + for (Map.Entry+ * 跨模块解耦：mf-common-ai 模块的 {@link McpCapabilityEngine} 通过此接口获取数据库中的 MCP 配置， + * 由 mf-ai 业务模块实现（查询 ai_mcp_server_config 表）。 + *
+ *+ * 若 Spring 容器中无此接口的实现 Bean（如 mf-common-ai 独立测试场景）， + * {@link McpCapabilityEngine} 不报错，返回空动作列表。 + *
+ * + * @author: mfish + * @date: 2026/07/21 + */ +public interface McpServerConfigProvider { + + /** + * 获取所有状态为"正常"的 MCP 服务器配置 + * + * @return MCP 服务器配置列表，无配置时返回空列表 + */ + List+ * 与 {@code McpServerConfig} 实体类解耦：mf-common-ai 不依赖 mf-ai-api 模块， + * 通过 {@link McpServerConfigProvider} 接口由业务层注入配置数据。 + *
+ * + * @author: mfish + * @date: 2026/07/21 + */ +@Data +@Accessors(chain = true) +public class McpServerInfo { + + /** + * MCP 服务器名称（唯一标识） + */ + private String serverName; + + /** + * 传输类型：stdio / sse / streamable + *+ * 将现有 {@link ApiToolEngine} 包装为 {@link CapabilitySubEngine}， + * 使 Feign/OpenAPI 工具能纳入 CapabilityEngine 的统一能力发现与执行体系。 + *
+ *+ * 适配逻辑： + *
+ * 与 LLM 驱动模式的关系： + *
+ * 动作名直接使用 ToolCallback 的原始 name（与 LLM 驱动模式保持一致）， + * 跨服务同名动作由 ApiToolEngine 去重策略保证唯一。 + * serviceId 字段标识动作来源服务，便于 Planner 按服务过滤。 + *
+ */ + @Override + public List+ * 通过 ApiToolEngine 查找 actionName 对应的 ToolCallback， + * 将 params 序列化为 JSON 字符串作为 toolInput， + * 从 ExecutionContext 提取租户信息构建 ToolContext， + * 调用 {@code ToolCallback.call(toolInput, toolContext)}。 + *
+ */ + @Override + public ActionResult execute(String actionName, Map+ * 复用 {@code BaseAssistant.buildToolContextFromSnapshot} 的 key 约定 + * （RPCConstants.REQ_* 常量），保证 FeignToolCallback/HttpToolCallback + * 的内部参数自动填充逻辑一致。 + *
+ */ + private ToolContext buildToolContext(ExecutionContext ctx) { + Map+ * 供 {@code ToolCapabilityEngine} 枚举所有服务的工具，将其暴露为 ActionDefinition。 + *
+ * + * @return 已注册的 serviceId 集合（不可变快照） + */ + public Set+ * 供 {@code ToolCapabilityEngine} 构建全局动作列表。 + * 去重策略与 {@link #getToolCallbacks(Collection)} 一致：先注册的保留。 + *
+ * + * @return 全部工具列表（去重后的快照） + */ + public List+ * 供 {@code ToolCapabilityEngine.execute()} 通过 actionName 路由到具体工具。 + * 遍历所有服务，返回首个名称匹配的 ToolCallback。 + *
+ * + * @param actionName 动作名（即 ToolCallback.getToolDefinition().name()） + * @return 匹配的 ToolCallback，未找到时返回 null + */ + public ToolCallback findToolCallback(String actionName) { + if (actionName == null || actionName.isEmpty()) { + return null; + } + for (List+ * 三驾马车驱动的 AI 架构 —— Memory 模块的核心抽象。 + * 每个 sessionId 对应一个 ConversationMemory 实例，聚合会话内 Spring AI 管不到的两类上下文： + *
+ *+ * 为什么不管理 Short-term Memory（Chat 历史）？ + * Spring AI 的 {@code ChatMemory} + {@code MessageChatMemoryAdvisor} 已经完整覆盖 + * "对话历史存储 + 自动注入 LLM 上下文" 两个职责，本模块不再重复实现，避免双写不一致。 + * Planner / Assistant 若需要读取对话历史，直接注入 Spring AI 的 {@code ChatMemory} 即可。 + *
+ *+ * 接口设计原则： + *
+ * 若变量已存在则覆盖。常见用法： + *
+ *+ * Planner 拼接上下文时调用，避免逐个 getVariable 的开销。 + * 返回的 Map 不可变，修改需通过 {@link #updateVariable}。 + *
+ * + * @return 变量快照（不可变 Map） + */ + Map+ * 将 TenantContext 的 5 个字段（tenantId / userId / accessToken / requestAttributes / + * serverWebExchange）按约定的 key 写入 Vars Context，供后续工具调用恢复认证态。 + *
+ *+ * 这是对原 AgentRuntime.captureTenantContext + BaseAssistant.buildToolContext 流程的统一收纳： + * 原本散落在两处的"捕获-传递-解包"逻辑，现在统一为 Memory 的一次 bind 调用。 + *
+ * + * @param tenantContext 租户上下文快照（null 表示清除绑定的租户信息） + */ + void bindTenantContext(TenantContext tenantContext); + + /** + * 清空 Vars Context + */ + void clearVars(); + + // ==================== Document Context ==================== + + /** + * 批量添加文档块到 Document Context + *+ * FileParseService 解析文件后调用此方法注入。重复添加相同 chunkId 的块会被去重。 + *
+ * + * @param chunks 文档块列表（不能为 null，可为空列表） + */ + void addDocumentChunks(List+ * 返回的是副本或不可变视图，外部修改不影响内部状态。 + *
+ * + * @return 文档块列表（可能为空，不会为 null） + */ + List+ * 按 chunkIndex 顺序拼接所有文档块内容，每个块带文件名分隔符。 + * Planner / Assistant 注入 LLM 上下文时调用此方法，无需自己拼接。 + *
+ *+ * 全量注入语义：本方法返回当前会话的全部文档块内容，适合文档量较小的场景。 + * 文档量较大时（几十份 PDF / 大型代码库），建议改用 {@link #searchDocumentContext} 按相关性检索， + * 避免 LLM 上下文窗口爆炸。 + *
+ *+ * 示例输出： + *
+ *+ * === 文件: example.docx === + * 文件内容... + * === 文件结束: example.docx === + *+ * + * @return 拼接后的文本；无文档时返回空字符串 + */ + String getDocumentContext(); + + /** + * 按相关性检索文档块（向量库扩展点） + *
+ * 默认实现（内存版）：忽略 query，返回全部文档块（截断到 topK）。 + * 这等价于"全量注入"，保持与 {@link #getDocumentChunks()} 一致的行为。 + *
+ *+ * 向量库版实现：将 query 文本向量化，按余弦相似度返回最相关的 topK 个 chunk。 + * 这是 RAG（检索增强生成）的核心入口，让 LLM 只看到最相关的文档片段而非全部文档。 + *
+ *+ * 调用方建议： + *
+ * 默认实现：调用 {@link #searchDocuments(query, topK)} 检索，然后按 + * {@link #getDocumentContext()} 的格式拼接。 + *
+ *+ * 向量库版可重写此方法以使用更高效的批量检索 API，或在拼接时附加相关性分数。 + *
+ * + * @param query 检索查询文本；内存版可忽略 + * @param topK 返回的最大 chunk 数；<= 0 表示不限制 + * @return 拼接后的文本；无文档或检索无结果时返回空字符串 + */ + default String searchDocumentContext(String query, int topK) { + List+ * Planner 在构建 prompt 时调用此方法，获取 Vars Context 和 Document Context 的拼接文本， + * 注入到 system prompt 中。Short-term Memory（Chat 历史）由 Spring AI + * MessageChatMemoryAdvisor 自动注入，不在此方法产出范围内。 + *
+ *+ * 输出格式示例： + *
+ *+ * 【用户上下文】 + * 租户: tenant_001 + * 用户: user_123 + * + * 【业务变量】 + * formId: 1001 + * businessKey: ORDER-2026-001 + * + * 【上传文件】 + * === 文件: example.docx === + * 文件内容... + * === 文件结束: example.docx === + *+ * + * @return 系统上下文文本；无任何上下文时返回空字符串 + */ + String getSystemContext(); + + /** + * 清空所有记忆（Vars + Documents） + *
+ * 用于会话彻底销毁场景。会话本身的存储条目由 {@link ConversationMemoryStore} 管理。 + *
+ *+ * 注意：此方法不清空 Spring AI 的 ChatMemory（对话历史），那部分由 Spring AI + * 自己管理，会话结束时业务方需另行调用 {@code ChatMemory.clear(sessionId)}。 + *
+ */ + void clearAll(); +} diff --git a/mf-common/mf-common-ai/src/main/java/cn/com/mfish/common/ai/memory/ConversationMemoryStore.java b/mf-common/mf-common-ai/src/main/java/cn/com/mfish/common/ai/memory/ConversationMemoryStore.java new file mode 100644 index 00000000..f84a2443 --- /dev/null +++ b/mf-common/mf-common-ai/src/main/java/cn/com/mfish/common/ai/memory/ConversationMemoryStore.java @@ -0,0 +1,85 @@ +package cn.com.mfish.common.ai.memory; + +import java.util.List; + +/** + * 会话记忆存储（Conversation Memory Store） + *+ * Memory 模块的入口，负责按 sessionId 创建、获取、销毁 {@link ConversationMemory} 实例。 + * 对外暴露类似 Map 的语义，但底层可以是内存 / Redis / 数据库。 + *
+ *+ * 使用方式： + *
+ *
+ * // 在请求入口获取（或创建）会话 Memory
+ * ConversationMemory memory = memoryStore.getOrCreate(sessionId);
+ *
+ * // 注入上下文
+ * memory.bindTenantContext(tenantContext);
+ * memory.addDocumentChunks(chunks);
+ * memory.updateVariable("biz.formId", 1001);
+ *
+ * // Planner / Assistant 读取上下文
+ * String systemContext = memory.getSystemContext();
+ *
+ * // 会话结束后清理
+ * memoryStore.remove(sessionId);
+ *
+ * + * 设计为 Spring Bean，由 {@code MemoryAutoConfiguration} 注册。 + * Controller / Assistant / Runtime 通过依赖注入获取。 + *
+ * + * @author: mfish + * @date: 2026/07/20 + */ +public interface ConversationMemoryStore { + + /** + * 获取或创建会话 Memory + *+ * 首次访问某 sessionId 时创建新实例并缓存；后续访问返回同一实例。 + * 实例的生命周期由实现类管理（如基于 TTL 自动过期）。 + *
+ * + * @param sessionId 会话ID（不能为 null 或空） + * @return 会话 Memory 实例 + */ + ConversationMemory getOrCreate(String sessionId); + + /** + * 获取会话 Memory（不创建） + * + * @param sessionId 会话ID + * @return 会话 Memory 实例；不存在返回 null + */ + ConversationMemory get(String sessionId); + + /** + * 移除会话 Memory 并清理其内部状态 + *+ * 触发 {@link ConversationMemory#clearAll()}，并从存储中移除。 + * 用于会话主动关闭、用户登出等场景。 + *
+ * + * @param sessionId 会话ID + * @return 被移除的实例；不存在返回 null + */ + ConversationMemory remove(String sessionId); + + /** + * 获取当前活跃的会话 ID 列表 + *+ * 主要用于监控、调试、定期清理任务。 + *
+ * + * @return 会话 ID 列表（不可变） + */ + List+ * 由 FileParseService 解析文件后产出的最小内容单元。一个文件可能产生多个 Chunk， + * 也可能只有一个 Chunk（小文件）。Memory 模块按 Chunk 维度管理文档上下文， + * 便于后续扩展：按相关性检索（RAG）、按 token 上限分片注入等。 + *
+ *+ * 设计为不可变值对象：构造后内容不再修改，需要更新时替换整个 Chunk。 + *
+ * + * @author: mfish + * @date: 2026/07/20 + */ +@Data +@Accessors(chain = true) +public class DocumentChunk { + /** + * 块ID（建议使用 UUID 或 fileKey + 序号） + */ + private String chunkId; + /** + * 来源文件的 fileKey（Storage 服务的唯一标识） + */ + private String fileKey; + /** + * 来源文件名（用于 LLM 提示词展示） + */ + private String fileName; + /** + * 文件 MIME 类型（用于 LLM 提示词展示和后续分流处理） + */ + private String fileType; + /** + * 块文本内容（已提取为纯文本，可直接注入 LLM 上下文） + */ + private String content; + /** + * 在原文件中的序号（从 0 开始；单文件单块时为 0） + */ + private int chunkIndex; + /** + * 块字符数（便于 token 估算和上下文裁剪） + */ + private int length; + /** + * 文本向量（embedding），用于向量库相似度检索 + *+ * 维度由 embedding model 决定（如 OpenAI text-embedding-3-small 为 1536 维）。 + *
+ *+ * 使用约定： + *
+ * 类型为 {@code float[]} 而非 {@code List
+ * 两段记忆的存储策略： + *
+ * 不管理 Short-term Memory：对话历史由 Spring AI 的 {@code ChatMemory} + + * {@code MessageChatMemoryAdvisor} 自动管理，本类不涉及，避免双写不一致。 + *
+ *+ * 线程安全：本类所有方法线程安全，可被多个线程（请求线程 + 异步编排线程）并发访问。 + * 实例由 {@link InMemoryConversationMemoryStore} 按 sessionId 单例化，无需关心实例级并发。 + *
+ * + * @author: mfish + * @date: 2026/07/20 + */ +@Slf4j +public class InMemoryConversationMemory implements ConversationMemory { + + /** + * Vars Context 中租户上下文的标准 key 前缀 + */ + private static final String TENANT_KEY_PREFIX = "sys.tenant."; + /** tenantId 在 Vars Context 中的 key */ + public static final String KEY_TENANT_ID = "sys.tenant.tenantId"; + /** userId 在 Vars Context 中的 key */ + public static final String KEY_USER_ID = "sys.tenant.userId"; + /** accessToken 在 Vars Context 中的 key */ + public static final String KEY_ACCESS_TOKEN = "sys.tenant.accessToken"; + /** requestAttributes 在 Vars Context 中的 key（Servlet） */ + public static final String KEY_REQUEST_ATTRIBUTES = "sys.tenant.requestAttributes"; + /** serverWebExchange 在 Vars Context 中的 key（WebFlux） */ + public static final String KEY_SERVER_WEB_EXCHANGE = "sys.tenant.serverWebExchange"; + + private final String sessionId; + + /** Vars Context：业务变量 + 租户上下文，统一存储 */ + private final Map+ * 使用 {@link ConcurrentHashMap} 按 sessionId 缓存 {@link ConversationMemory} 实例。 + * 实例首次访问时创建并缓存，后续访问返回同一实例（sessionId 级单例）。 + *
+ *+ * 限制：进程内存储，重启丢失；不支持分布式部署下的会话漂移。 + * 后续可替换为 Redis 实现以支持多实例共享。 + *
+ *+ * 注意：本 Store 只管理 Vars Context 和 Document Context，不管理 Spring AI 的 + * ChatMemory（对话历史）。会话结束时如需清空对话历史，业务方需另行调用 + * {@code ChatMemory.clear(sessionId)}。 + *
+ * + * @author: mfish + * @date: 2026/07/20 + */ +public class InMemoryConversationMemoryStore implements ConversationMemoryStore { + + private final ConcurrentMap+ * 注册 {@link ConversationMemoryStore} 的默认实现 {@link InMemoryConversationMemoryStore}。 + *
+ *+ * 本模块只管理 Spring AI 管不到的两类上下文：Vars Context（变量/凭证）和 + * Document Context（文件解析块）。对话历史（Short-term Memory）由 Spring AI 的 + * {@code ChatMemory} + {@code MessageChatMemoryAdvisor} 自动管理，本模块不涉及。 + *
+ *+ * 业务方可通过自定义 {@link ConversationMemoryStore} Bean 覆盖默认实现， + * 例如替换为 Redis 版本以支持多实例部署。 + *
+ * + * @author: mfish + * @date: 2026/07/20 + */ +@Configuration +public class MemoryAutoConfiguration { + + @Bean + @ConditionalOnMissingBean + public ConversationMemoryStore conversationMemoryStore() { + return new InMemoryConversationMemoryStore(); + } +} diff --git a/mf-common/mf-common-ai/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/mf-common/mf-common-ai/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports index f2170c30..83ecfe48 100644 --- a/mf-common/mf-common-ai/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports +++ b/mf-common/mf-common-ai/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports @@ -1 +1,3 @@ cn.com.mfish.common.ai.engine.ApiToolAutoConfiguration +cn.com.mfish.common.ai.memory.MemoryAutoConfiguration +cn.com.mfish.common.ai.capability.CapabilityAutoConfiguration