+ * 用于调用方（如 Wiki 消化管线）已经有自己的外层重试策略， + * 希望绕过 Spring AI 内层重试、独占重试控制权的场景：传入 + * {@code RetryTemplate.builder().maxAttempts(1).build()} 即可把内层降级为"只跑一次"。 + *
+ * DashScope 和 OpenAI-ChatGPT 分支不走 Spring AI 的 RetryTemplate 接口， + * 本参数对它们无效（它们各自有内部重试或直通）。 + */ + public ChatModel buildRuntimeChatModel(ModelConfigEntity runtimeModel, RetryTemplate retryOverride) { ModelProviderEntity provider = modelProviderService.getProviderConfig(runtimeModel.getProvider()); ModelProtocol protocol = ModelProtocol.fromChatModel(provider.getChatModel()); @@ -490,7 +506,7 @@ public class AgentGraphBuilder { return OpenAiChatModel.builder() .openAiApi(api) .defaultOptions(options) - .retryTemplate(retryTemplate) + .retryTemplate(retryOverride) .observationRegistry(observationRegistryProvider.getIfAvailable(() -> ObservationRegistry.NOOP)) .build(); } @@ -501,7 +517,7 @@ public class AgentGraphBuilder { return AnthropicChatModel.builder() .anthropicApi(api) .defaultOptions(options) - .retryTemplate(retryTemplate) + .retryTemplate(retryOverride) .observationRegistry(observationRegistryProvider.getIfAvailable(() -> ObservationRegistry.NOOP)) .build(); } @@ -981,6 +997,11 @@ public class AgentGraphBuilder { builder.maxTokens(4096); } } + // RFC-014: 接入 Anthropic prompt caching（spring-ai 1.1.4 一等支持） + // 通过 cacheOptions 配置 system / tools / conversation history 自动打 cache_control， + // 多轮对话场景可节省 50–75% 输入 token 成本。 + builder.cacheOptions(anthropicCacheOptionsFactory.build()); + return builder.internalToolExecutionEnabled(false).build(); } diff --git a/mateclaw-server/src/main/java/vip/mate/agent/context/RuntimeContextInjector.java b/mateclaw-server/src/main/java/vip/mate/agent/context/RuntimeContextInjector.java index f3661a55..b664186c 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/context/RuntimeContextInjector.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/context/RuntimeContextInjector.java @@ -9,6 +9,12 @@ import java.time.format.DateTimeFormatter; *
* 参考 Claude Code 的 prependUserContext 模式：将时间信息作为首条 meta UserMessage 注入， * 而非修改 System Prompt，以保持 prompt cache 命中率。 + *
+ * RFC-014 协同要点：本类返回的内容必须始终包装为 {@code UserMessage} 注入，
+ * 严禁拼接进 SystemMessage —— 否则每次时间变化都会击穿 spring-ai 的
+ * {@code AnthropicCacheStrategy.SYSTEM_AND_TOOLS / CONVERSATION_HISTORY} system cache。
+ * 内容长度（约 70 字符）远小于 spring-ai 的 USER min-content-length 阈值（≥1024 字符），
+ * 因此不会被错误纳入对话 cache 块。新增调用点时请保持此约束。
*/
public final class RuntimeContextInjector {
diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java
index bccc2b9b..95a5e6c8 100644
--- a/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java
+++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java
@@ -272,6 +272,9 @@ public class NodeStreamingChatHelper {
AtomicReference 规则：
+ *
+ *
这是有状态对象，建议作为 Spring 单例 bean。所有计数用原子类型，无锁、零分配热路径。
+ */ +@Slf4j +public final class AdaptiveCacheStrategy implements PromptCacheStrategy { + + private final PromptCacheStrategy delegate; + private final int missThreshold; + private final long coolDownMillis; + + private final AtomicInteger consecutiveMisses = new AtomicInteger(); + private final AtomicLong degradedUntilEpochMs = new AtomicLong(); + + public AdaptiveCacheStrategy(PromptCacheStrategy delegate, int missThreshold, long coolDownMillis) { + this.delegate = (delegate instanceof AdaptiveCacheStrategy) + ? ((AdaptiveCacheStrategy) delegate).delegate + : delegate; + this.missThreshold = Math.max(1, missThreshold); + this.coolDownMillis = Math.max(0, coolDownMillis); + } + + @Override + public CachedPlan plan(CachePlanContext ctx) { + return isDegraded() ? CachedPlan.none() : delegate.plan(ctx); + } + + @Override + public boolean shouldCache(CachePlanContext ctx) { + return !isDegraded() && delegate.shouldCache(ctx); + } + + /** 命中：清零并立刻恢复（如已降级）。 */ + public void recordHit() { + consecutiveMisses.set(0); + degradedUntilEpochMs.set(0); + } + + /** 未命中：累加；越过阈值则进入冷却期。 */ + public void recordMiss() { + int n = consecutiveMisses.incrementAndGet(); + if (n >= missThreshold) { + long until = System.currentTimeMillis() + coolDownMillis; + degradedUntilEpochMs.set(until); + log.warn("Prompt cache strategy degraded after {} consecutive misses; coolDown={}ms", + n, coolDownMillis); + } + } + + private boolean isDegraded() { + long until = degradedUntilEpochMs.get(); + if (until == 0) return false; + if (System.currentTimeMillis() >= until) { + // 冷却期已过：原子清零（CAS 防止并发覆写） + degradedUntilEpochMs.compareAndSet(until, 0); + consecutiveMisses.set(0); + return false; + } + return true; + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/llm/cache/AnthropicCacheOptionsFactory.java b/mateclaw-server/src/main/java/vip/mate/llm/cache/AnthropicCacheOptionsFactory.java new file mode 100644 index 00000000..b76e4c9b --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/llm/cache/AnthropicCacheOptionsFactory.java @@ -0,0 +1,74 @@ +package vip.mate.llm.cache; + +import org.springframework.ai.anthropic.api.AnthropicCacheOptions; +import org.springframework.ai.anthropic.api.AnthropicCacheStrategy; +import org.springframework.ai.anthropic.api.AnthropicCacheTtl; +import org.springframework.ai.chat.messages.MessageType; +import org.springframework.stereotype.Component; + +import java.util.EnumMap; +import java.util.Map; + +/** + * 把 MateClaw 的 {@link CacheProperties} 翻译成 spring-ai 1.1.4 一等支持的 {@link AnthropicCacheOptions}。 + * + *设计说明：spring-ai 已内置完整的 Anthropic prompt cache 框架（{@code CacheEligibilityResolver} + + * {@code CacheBreakpointTracker}），它会按 {@link AnthropicCacheStrategy} 自动决策在 system / tools / + * conversation history 上挂 {@code cache_control}。我们只需配置好策略与 TTL/min-length 即可， + * 不必自己写 HTTP body 拦截器。
+ * + *映射规则（对应 RFC-014 Change 1 的 SystemAndTailCacheStrategy 默认行为）： + *
无状态、线程安全。
+ */ +@Component +public class AnthropicCacheOptionsFactory { + + /** token → 字符 的粗略乘子（英文 4 字符 ≈ 1 token；中文略低，但作为门槛足矣）。 */ + private static final int CHARS_PER_TOKEN_HEURISTIC = 4; + + private final CacheProperties props; + + public AnthropicCacheOptionsFactory(CacheProperties props) { + this.props = props; + } + + /** + * 构建启用了缓存的 {@link AnthropicCacheOptions}；当 {@link CacheProperties#isEnabled()} 为 false + * 时返回 {@link AnthropicCacheOptions#DISABLED}（spring-ai 内置的 NONE 策略快速路径）。 + */ + public AnthropicCacheOptions build() { + if (!props.isEnabled()) { + return AnthropicCacheOptions.DISABLED; + } + + AnthropicCacheStrategy strategy = props.isIncludeToolsBlock() + ? AnthropicCacheStrategy.CONVERSATION_HISTORY + : AnthropicCacheStrategy.SYSTEM_ONLY; + + AnthropicCacheTtl ttl = (props.resolveCacheTtl() == CacheTtl.EXTENDED_1H) + ? AnthropicCacheTtl.ONE_HOUR + : AnthropicCacheTtl.FIVE_MINUTES; + + Map断点位置用于序列化器决定把 {@code cache_control} 放在请求体的哪个内容块上。 + * 顺序与 Anthropic Messages API 的天然布局对齐：system → tools → memory/wiki → messages tail。
+ */ +public enum BreakpointKind { + SYSTEM_TAIL, + TOOLS_TAIL, + MEMORY_BLOCK, + MESSAGES_TAIL +} diff --git a/mateclaw-server/src/main/java/vip/mate/llm/cache/CacheDirectives.java b/mateclaw-server/src/main/java/vip/mate/llm/cache/CacheDirectives.java new file mode 100644 index 00000000..e7377551 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/llm/cache/CacheDirectives.java @@ -0,0 +1,38 @@ +package vip.mate.llm.cache; + +import java.util.List; +import java.util.Map; + +/** + * 协议无关的"缓存指令包"。 + * + *序列化器从 {@link CachedPlan} 推导出本对象后，由具体的请求拦截层 + * （如 {@code CachingChatModelDecorator} 注册到 RestClient 的 {@code ClientHttpRequestInterceptor}） + * 把指令落到具体协议的请求体上。
+ * + * @param breakpoints 最终要应用的断点（顺序敏感） + * @param httpHeaders 需要附加到 HTTP 请求的额外 header（如 Anthropic extended-cache-ttl beta header） + * @param ttl TTL 提示 + * @param protocol 生产此指令的协议；运行时拦截器据此选分支 + */ +public record CacheDirectives( + List不持有 {@link Prompt} 引用以外的可变状态；所有派生指标（system 长度、tool 数等）按需懒计算 + * 并由 record 字段缓存，避免重复扫描。
+ */ +public record CachePlanContext( + Prompt prompt, + ModelProtocol protocol, + int totalPromptTokens, + int turnCount, + int minPromptTokens, + int maxBreakpoints, + boolean includeToolsBlock, + CacheTtl ttl) { + + public CachePlanContext { + if (prompt == null) throw new IllegalArgumentException("prompt must not be null"); + if (protocol == null) throw new IllegalArgumentException("protocol must not be null"); + if (ttl == null) ttl = CacheTtl.DEFAULT_5M; + if (maxBreakpoints <= 0) maxBreakpoints = 4; + } + + /** 消息列表（不可变视图；上层不应修改）。 */ + public List+ * mateclaw: + * llm: + * cache: + * enabled: true + * min-prompt-tokens: 1024 + * max-breakpoints: 4 + * ttl: default # default | extended-1h + * include-tools-block: true + * adaptive: + * enabled: true + * miss-threshold: 5 + * cool-down: 60s + *+ */ +@ConfigurationProperties(prefix = "mateclaw.llm.cache") +public class CacheProperties { + + /** 总开关；false 时全部走 NoOp。 */ + private boolean enabled = true; + + /** 累计 prompt token 低于此值则跳过缓存（避免 cache write 倒亏）。 */ + private int minPromptTokens = 1024; + + /** 单请求最多打几个 cache_control 断点（Anthropic 上限 4）。 */ + private int maxBreakpoints = 4; + + /** 默认 TTL；{@code extended-1h} 需要 Anthropic beta header。 */ + private Ttl ttl = Ttl.DEFAULT; + + /** 是否把工具 schema 段也作为一个断点（独立断点，避免与 system 共享）。 */ + private boolean includeToolsBlock = true; + + private final Adaptive adaptive = new Adaptive(); + + public boolean isEnabled() { return enabled; } + public void setEnabled(boolean enabled) { this.enabled = enabled; } + + public int getMinPromptTokens() { return minPromptTokens; } + public void setMinPromptTokens(int minPromptTokens) { this.minPromptTokens = minPromptTokens; } + + public int getMaxBreakpoints() { return maxBreakpoints; } + public void setMaxBreakpoints(int maxBreakpoints) { this.maxBreakpoints = maxBreakpoints; } + + public Ttl getTtl() { return ttl; } + public void setTtl(Ttl ttl) { this.ttl = ttl; } + + public boolean isIncludeToolsBlock() { return includeToolsBlock; } + public void setIncludeToolsBlock(boolean includeToolsBlock) { this.includeToolsBlock = includeToolsBlock; } + + public Adaptive getAdaptive() { return adaptive; } + + public CacheTtl resolveCacheTtl() { + return ttl == Ttl.EXTENDED_1H ? CacheTtl.EXTENDED_1H : CacheTtl.DEFAULT_5M; + } + + public enum Ttl { DEFAULT, EXTENDED_1H } + + public static class Adaptive { + private boolean enabled = true; + private int missThreshold = 5; + /** 字符串解析在 application.yml 由 Spring Boot 自动转换；默认 60_000 ms。 */ + private long coolDownMs = 60_000L; + + public boolean isEnabled() { return enabled; } + public void setEnabled(boolean enabled) { this.enabled = enabled; } + public int getMissThreshold() { return missThreshold; } + public void setMissThreshold(int missThreshold) { this.missThreshold = missThreshold; } + public long getCoolDownMs() { return coolDownMs; } + public void setCoolDownMs(long coolDownMs) { this.coolDownMs = coolDownMs; } + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/llm/cache/CacheSerializer.java b/mateclaw-server/src/main/java/vip/mate/llm/cache/CacheSerializer.java new file mode 100644 index 00000000..41f66ad2 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/llm/cache/CacheSerializer.java @@ -0,0 +1,27 @@ +package vip.mate.llm.cache; + +/** + * 把 {@link CachedPlan} 转换为协议特化的 {@link CacheDirectives}。 + * + *
sealed 锁定可选实现，避免反射式插件加载并支持穷尽 switch。
+ * + *注意：本接口只决定"要打哪些断点 + 需要哪些 HTTP header"，不直接修改底层 HTTP body。 + * 真正写 JSON 的工作由调用层完成；这样保证序列化器是纯函数，便于单元测试。
+ * + *Anthropic 例外：spring-ai 1.1.4+ 已通过 {@code AnthropicCacheOptions} 提供一等支持， + * 由 {@link AnthropicCacheOptionsFactory} 直接装配到 {@code AnthropicChatOptions}，不走本接口。 + * 本接口仅用于我们自有客户端（如 OpenAI Responses）需要手动装配缓存指令的协议。
+ */ +public sealed interface CacheSerializer + permits OpenAIResponsesCacheSerializer { + + /** 此序列化器服务的协议；调度方据此挑选实现。 */ + vip.mate.llm.model.ModelProtocol protocol(); + + /** + * @param plan 策略产出的方案（可能为空 → 返回 {@link CacheDirectives#empty} ） + * @param ctx 与策略相同的上下文，便于序列化器看 messages 大小决定是否缩减断点 + * @return 供 HTTP 拦截器消费的指令 + */ + CacheDirectives serialize(CachedPlan plan, CachePlanContext ctx); +} diff --git a/mateclaw-server/src/main/java/vip/mate/llm/cache/CacheTtl.java b/mateclaw-server/src/main/java/vip/mate/llm/cache/CacheTtl.java new file mode 100644 index 00000000..98e8e130 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/llm/cache/CacheTtl.java @@ -0,0 +1,11 @@ +package vip.mate.llm.cache; + +/** + * Anthropic prompt cache 的存活时间。 + *{@link #EXTENDED_1H} 需要 beta header {@code anthropic-beta: extended-cache-ttl-2025-04-11}， + * 可通过 application.yml 的 {@code mateclaw.llm.cache.ttl=extended-1h} 启用。
+ */ +public enum CacheTtl { + DEFAULT_5M, + EXTENDED_1H +} diff --git a/mateclaw-server/src/main/java/vip/mate/llm/cache/CacheUsageExtractor.java b/mateclaw-server/src/main/java/vip/mate/llm/cache/CacheUsageExtractor.java new file mode 100644 index 00000000..dc94eea2 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/llm/cache/CacheUsageExtractor.java @@ -0,0 +1,91 @@ +package vip.mate.llm.cache; + +import org.springframework.ai.chat.metadata.Usage; + +import java.lang.reflect.Method; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; + +/** + * 从 Spring AI {@link Usage} 中提取 Anthropic 的 prompt cache token 计数。 + * + *spring-ai 的高层 {@code Usage} 接口只暴露 {@code promptTokens} / {@code completionTokens}， + * 没有 cache 维度；但 {@link Usage#getNativeUsage()} 会返回 provider 的原生 usage 对象。 + * 对 Anthropic 而言是 {@code AnthropicApi.Usage} record，含 {@code cacheCreationInputTokens} + * 与 {@code cacheReadInputTokens}。
+ * + *采用反射调用以避免： + *
不可变、线程安全。
+ */ +public final class CacheUsageExtractor { + + /** {@code (cacheReadTokens, cacheWriteTokens)}；任一字段不可得时为 0。 */ + public record CacheTokens(int cacheReadTokens, int cacheWriteTokens) { + public static final CacheTokens EMPTY = new CacheTokens(0, 0); + + public boolean isEmpty() { return cacheReadTokens == 0 && cacheWriteTokens == 0; } + } + + /** 缓存 (Class, methodName) → reflected Method（命中失败时为标记 NULL_METHOD）。 */ + private static final ConcurrentMap使用 JDK 21 {@link SequencedCollection} 保证断点的固定顺序（首尾稳定， + * 序列化器能直接按顺序写入），同时是不可变 record，便于在多线程间传递。
+ * + * @param breakpoints 断点序列；调用方应当按顺序遍历 + * @param ttl 缓存 TTL；{@code null} 等价 {@link CacheTtl#DEFAULT_5M} + */ +public record CachedPlan(SequencedCollection暴露 {@link PromptCacheStrategy} 作为 Spring bean：当 {@code mateclaw.llm.cache.adaptive.enabled=true} + * 时用 {@link AdaptiveCacheStrategy} 包装基础策略 {@link SystemAndTailCacheStrategy}， + * 后者负责实际断点计算；自适应层根据 cache miss 率自动降级。
+ * + *{@link AnthropicCacheOptionsFactory} 由 {@code @Component} 自动注册。
+ */ +@Configuration +@EnableConfigurationProperties(CacheProperties.class) +public class LlmCacheAutoConfiguration { + + @Bean + public PromptCacheStrategy promptCacheStrategy(CacheProperties props) { + if (!props.isEnabled()) { + return NoOpCacheStrategy.INSTANCE; + } + PromptCacheStrategy base = new SystemAndTailCacheStrategy(); + if (props.getAdaptive().isEnabled()) { + return new AdaptiveCacheStrategy( + base, + props.getAdaptive().getMissThreshold(), + props.getAdaptive().getCoolDownMs()); + } + return base; + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/llm/cache/NoOpCacheStrategy.java b/mateclaw-server/src/main/java/vip/mate/llm/cache/NoOpCacheStrategy.java new file mode 100644 index 00000000..73ec3ca3 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/llm/cache/NoOpCacheStrategy.java @@ -0,0 +1,24 @@ +package vip.mate.llm.cache; + +/** + * 占位策略：永不打缓存断点。 + * + *用于 {@link AdaptiveCacheStrategy} 在连续 cache miss 后降级， + * 或在 {@code mateclaw.llm.cache.enabled=false} 时全局短路。
+ */ +public final class NoOpCacheStrategy implements PromptCacheStrategy { + + public static final NoOpCacheStrategy INSTANCE = new NoOpCacheStrategy(); + + private NoOpCacheStrategy() {} + + @Override + public CachedPlan plan(CachePlanContext ctx) { + return CachedPlan.none(); + } + + @Override + public boolean shouldCache(CachePlanContext ctx) { + return false; + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/llm/cache/OpenAIResponsesCacheSerializer.java b/mateclaw-server/src/main/java/vip/mate/llm/cache/OpenAIResponsesCacheSerializer.java new file mode 100644 index 00000000..d965b1fc --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/llm/cache/OpenAIResponsesCacheSerializer.java @@ -0,0 +1,33 @@ +package vip.mate.llm.cache; + +import vip.mate.llm.model.ModelProtocol; + +import java.util.Map; + +/** + * OpenAI Responses API 的 cache_control 序列化器。 + * + *OpenAI Responses 协议使用与 Anthropic 同形的 {@code cache_control} 字段挂在 input 数组的 + * content block 上，无需额外 HTTP header。M2 阶段保留断点信息，由 + * {@code ChatGPTResponsesClient} 在 buildRequestBody 时按指令落字段（M3 完成）。
+ */ +public final class OpenAIResponsesCacheSerializer implements CacheSerializer { + + @Override + public ModelProtocol protocol() { + return ModelProtocol.OPENAI_CHATGPT; + } + + @Override + public CacheDirectives serialize(CachedPlan plan, CachePlanContext ctx) { + if (plan == null || plan.isEmpty()) { + return CacheDirectives.empty(ModelProtocol.OPENAI_CHATGPT); + } + // Responses API 没有 TTL 概念，全部按默认；headers 留空 + return new CacheDirectives( + plan.breakpoints().stream().toList(), + Map.of(), + CacheTtl.DEFAULT_5M, + ModelProtocol.OPENAI_CHATGPT); + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/llm/cache/PromptCacheStrategy.java b/mateclaw-server/src/main/java/vip/mate/llm/cache/PromptCacheStrategy.java new file mode 100644 index 00000000..5f9049fe --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/llm/cache/PromptCacheStrategy.java @@ -0,0 +1,28 @@ +package vip.mate.llm.cache; + +/** + * Prompt 缓存策略 SPI。 + * + *所有策略实现都被 sealed 锁定，便于 JDK 21 模式匹配 switch 穷尽分支、 + * 也避免运行期反射式插件加载带来的不可预期成本。
+ * + *调用契约： + *
断点选择规则（按优先级递减）： + *
这是无状态的纯函数实现，单例线程安全；任何调度方都可直接复用。
+ */ +public final class SystemAndTailCacheStrategy implements PromptCacheStrategy { + + /** system 段至少 256 字符才值得打断点（cache write 摊销下限）。 */ + private static final int MIN_SYSTEM_CHARS = 256; + + @Override + public CachedPlan plan(CachePlanContext ctx) { + if (!shouldCache(ctx)) { + return CachedPlan.none(); + } + + List
+ * 与 {@link #search} 共享同一套重试策略：408/429/5xx 状态码或 IO 异常时按
+ * 指数退避（800/1600/3200ms）重试，最多 {@code httpRetries} 次。
+ * 此外当 bundle 内容（{@code content}）为空时直接返回 null —— 空 bundle
+ * 重装会清空用户的 SKILL.md，是不可接受的"成功"。
*/
public SkillBundle fetchBundle(String slug, String version) {
String path = version != null && !version.isBlank()
@@ -100,27 +105,54 @@ public class SkillHubClient {
: "/api/v1/skills/" + slug;
String url = properties.getBaseUrl() + path;
- try {
- HttpRequest request = HttpRequest.newBuilder()
- .uri(URI.create(url))
- .timeout(Duration.ofSeconds(properties.getHttpTimeout()))
- .GET()
- .header("Accept", "application/json")
- .header("User-Agent", "MateClaw/1.0")
- .build();
+ for (int attempt = 0; attempt <= properties.getHttpRetries(); attempt++) {
+ try {
+ HttpRequest request = HttpRequest.newBuilder()
+ .uri(URI.create(url))
+ .timeout(Duration.ofSeconds(properties.getHttpTimeout()))
+ .GET()
+ .header("Accept", "application/json")
+ .header("User-Agent", "MateClaw/1.0")
+ .build();
- HttpResponse
+ * RFC-012 follow-up #3：原实现 phase B 的 create / merge 循环逐页串行调用 LLM，
+ * 单 chunk N 个 page 就要串行 N 次 LLM 调用——一个卡超时整条流水线停摆。
+ * 改为受此 Semaphore 控制的并行，默认 3。结合 maxParallelRawMaterials × maxParallelChunks
+ * × maxParallelPhaseBPages = 3 × 5 × 3 = 45 的理论最大并发，实际按 LLM 限流为准。
+ */
+ private int maxParallelPhaseBPages = 3;
+
/** 注入 agent prompt 的最大字符数 */
private int maxContextChars = 10000;
diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/controller/WikiController.java b/mateclaw-server/src/main/java/vip/mate/wiki/controller/WikiController.java
index 547d2fa9..f9925c26 100644
--- a/mateclaw-server/src/main/java/vip/mate/wiki/controller/WikiController.java
+++ b/mateclaw-server/src/main/java/vip/mate/wiki/controller/WikiController.java
@@ -3,9 +3,12 @@ package vip.mate.wiki.controller;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.tags.Tag;
import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartFile;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import vip.mate.common.result.R;
import vip.mate.exception.MateClawException;
import vip.mate.workspace.core.annotation.RequireWorkspaceRole;
@@ -19,6 +22,7 @@ import vip.mate.wiki.service.WikiKnowledgeBaseService;
import vip.mate.wiki.service.WikiPageService;
import vip.mate.wiki.service.WikiProcessingService;
import vip.mate.wiki.service.WikiRawMaterialService;
+import vip.mate.wiki.sse.WikiProgressBus;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
@@ -34,6 +38,7 @@ import java.util.Map;
*
* @author MateClaw Team
*/
+@Slf4j
@Tag(name = "Wiki 知识库")
@RestController
@RequestMapping("/api/v1/wiki")
@@ -47,6 +52,7 @@ public class WikiController {
private final WikiDirectoryScanService scanService;
private final WikiProperties properties;
private final ApplicationEventPublisher eventPublisher;
+ private final WikiProgressBus progressBus;
// ==================== Knowledge Base ====================
@@ -374,6 +380,59 @@ public class WikiController {
));
}
+ /**
+ * RFC-012 M3：订阅指定 KB 的处理进度 SSE 流。
+ *
+ * 客户端通过 {@code new EventSource('/api/v1/wiki/knowledge-bases/{kbId}/progress')} 订阅，
+ * 然后按事件名监听：
+ *
+ * SSE 是 best-effort：服务端断线、客户端断线、代理切流都可能丢事件，
+ * 因此前端仍需保留 60s 兜底轮询 {@code GET .../processing-status} 作为真源。
+ *
+ * Emitter 默认 30 分钟超时，足以覆盖最长的 raw 处理时间；超时后客户端
+ * 自动重连（EventSource 默认行为）。
+ */
+ @RequireWorkspaceRole("viewer")
+ @Operation(summary = "订阅处理进度 SSE")
+ @GetMapping(value = "/knowledge-bases/{kbId}/progress", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
+ public SseEmitter subscribeProgress(@PathVariable Long kbId,
+ @RequestHeader(value = "X-Workspace-Id", required = false) Long workspaceId) {
+ verifyKBWorkspace(kbId, workspaceId);
+ SseEmitter emitter = new SseEmitter(30L * 60 * 1000); // 30min
+ progressBus.subscribe(kbId, emitter);
+
+ emitter.onCompletion(() -> {
+ progressBus.unsubscribe(kbId, emitter);
+ log.debug("[Wiki SSE] emitter completed: kbId={}", kbId);
+ });
+ emitter.onTimeout(() -> {
+ progressBus.unsubscribe(kbId, emitter);
+ try { emitter.complete(); } catch (Exception ignore) { /* best-effort */ }
+ log.debug("[Wiki SSE] emitter timeout: kbId={}", kbId);
+ });
+ emitter.onError(e -> {
+ progressBus.unsubscribe(kbId, emitter);
+ log.debug("[Wiki SSE] emitter error: kbId={}, cause={}", kbId, e.getMessage());
+ });
+
+ // 立即发一个 hello 事件，确认连接已建立
+ try {
+ emitter.send(SseEmitter.event().name(WikiProgressBus.EVENT_HEARTBEAT)
+ .data("{\"ts\":" + System.currentTimeMillis() + ",\"hello\":true}"));
+ } catch (Exception e) {
+ log.debug("[Wiki SSE] initial heartbeat send failed: {}", e.getMessage());
+ }
+
+ return emitter;
+ }
+
// ==================== Workspace Verification ====================
private void verifyKBWorkspace(Long kbId, Long headerWorkspaceId) {
diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiPageService.java b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiPageService.java
index 154bdb5e..30f68252 100644
--- a/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiPageService.java
+++ b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiPageService.java
@@ -132,6 +132,38 @@ public class WikiPageService {
.eq(WikiPageEntity::getSlug, slug));
}
+ /**
+ * 把 slug 规范化为 canonical 形式：去掉所有连字符 / 下划线 + 转小写。
+ *
+ * 用于跨拼写匹配：{@code "shennong-bencao-jing"} 和 {@code "shen-nong-ben-cao-jing"}
+ * 都规范化为 {@code "shennongbencaojing"}，被视为同一概念。LLM 在并行处理大文档时
+ * 经常对同一概念给出不同 slug 拼写（按词分组 vs 按字分隔），这是兜底归一逻辑的基础。
+ */
+ public static String canonicalSlug(String slug) {
+ if (slug == null) return "";
+ return slug.toLowerCase().replace("-", "").replace("_", "");
+ }
+
+ /**
+ * 按 canonical slug 在指定 KB 中查找已存在的 page。
+ *
+ * 命中条件：现有 page 的 slug 经 {@link #canonicalSlug(String)} 后与给定 slug 的
+ * canonical 形式相等。复用 {@link #listSummaries(Long)} 的 5 分钟缓存，命中后再
+ * {@link #getBySlug(Long, String)} 拿完整 entity，避免额外全表扫描。
+ *
+ * @return 第一个 canonical 匹配的 page；找不到返回 {@code null}
+ */
+ public WikiPageEntity findByCanonicalSlug(Long kbId, String slug) {
+ String canonical = canonicalSlug(slug);
+ if (canonical.isEmpty()) return null;
+ for (WikiPageEntity p : listSummaries(kbId)) {
+ if (canonicalSlug(p.getSlug()).equals(canonical)) {
+ return getBySlug(kbId, p.getSlug());
+ }
+ }
+ return null;
+ }
+
public WikiPageEntity getById(Long id) {
return pageMapper.selectById(id);
}
diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiProcessingService.java b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiProcessingService.java
index b4550487..43b5ba98 100644
--- a/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiProcessingService.java
+++ b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiProcessingService.java
@@ -9,6 +9,7 @@ import org.springframework.ai.chat.messages.UserMessage;
import org.springframework.ai.chat.model.ChatModel;
import org.springframework.ai.chat.model.ChatResponse;
import org.springframework.ai.chat.prompt.Prompt;
+import org.springframework.retry.support.RetryTemplate;
import org.springframework.stereotype.Service;
import vip.mate.agent.AgentGraphBuilder;
import vip.mate.agent.prompt.PromptLoader;
@@ -18,6 +19,7 @@ import vip.mate.wiki.WikiProperties;
import vip.mate.wiki.model.WikiKnowledgeBaseEntity;
import vip.mate.wiki.model.WikiPageEntity;
import vip.mate.wiki.model.WikiRawMaterialEntity;
+import vip.mate.wiki.sse.WikiProgressBus;
import java.util.ArrayList;
import java.util.List;
@@ -48,6 +50,7 @@ public class WikiProcessingService {
private final ModelConfigService modelConfigService;
private final AgentGraphBuilder agentGraphBuilder;
private final ObjectMapper objectMapper;
+ private final WikiProgressBus progressBus;
/** 并行 chunk / 材料处理执行器（JDK 21 虚拟线程）；Listener 跨包需要引用，故 public */
public static final ExecutorService WIKI_EXECUTOR = Executors.newVirtualThreadPerTaskExecutor();
@@ -61,7 +64,19 @@ public class WikiProcessingService {
private static final class ProgressCounter {
final AtomicInteger total = new AtomicInteger(0);
final AtomicInteger done = new AtomicInteger(0);
+ /** Page-level 失败计数（chunk 内单页 create / merge 抛异常）。
+ * 注：DuplicateKeyException 触发的 fallback-to-update 不算 failure，
+ * 内容仍合入了同 slug page。仅 LLM 调用爆炸、JSON 解析失败、内容为空等真失败才递增。 */
+ final AtomicInteger failed = new AtomicInteger(0);
final AtomicBoolean phaseBStarted = new AtomicBoolean(false);
+ /**
+ * 跨 chunk slug 抢占表：canonical slug → 第一个声明该概念的实际 slug。
+ *
+ * 解决 LLM 在并行 chunk 中给同一概念起不同 slug 拼写（按词分组 vs 按字分隔）的问题。
+ * 使用 {@link ConcurrentHashMap#computeIfAbsent} 实现原子抢占：先到的 chunk 把自己的
+ * slug 注册为 winner，后到的 chunk 看到 winner 后会把内容写入 winner 对应的 page。
+ */
+ final ConcurrentHashMap
+ * 内存态：server 重启会丢，但原 raw 的 status 已被 reprocess 改为 pending，
+ * 重启后按正常 pending 流程跑（退化为「不删旧页的全量重跑」，功能不丢失只是
+ * 没有走 route 的 "update" 识别路径）。
+ */
+ private final Set
+ * 如果之前状态是 {@code partial}，把 rawId 加入 {@link #partialResumeIds}，
+ * 下游的 WikiProcessingService 会据此决定是否保留已生成的 exclusive page（续传语义）。
*/
@Transactional
public void reprocess(Long id) {
@@ -235,12 +251,28 @@ public class WikiRawMaterialService {
if (entity == null) {
throw new IllegalArgumentException("Raw material not found: " + id);
}
+ boolean wasPartial = "partial".equals(entity.getProcessingStatus());
entity.setProcessingStatus("pending");
entity.setErrorMessage(null);
rawMapper.updateById(entity);
+ if (wasPartial) {
+ partialResumeIds.add(id);
+ log.info("[Wiki] Raw material queued for PARTIAL RESUME: id={} (existing pages will be kept)", id);
+ } else {
+ log.info("[Wiki] Raw material queued for reprocessing: id={}", id);
+ }
eventPublisher.publishEvent(new WikiProcessingEvent(this, entity.getId(), entity.getKbId()));
- log.info("[Wiki] Raw material queued for reprocessing: id={}", id);
+ }
+
+ /**
+ * 消费 partial resume 标记：若存在则返回 true 并从 set 中移除（一次性）。
+ *
+ * 必须在 {@link #claimForProcessing(Long)} 之前调用：claim 会把 status 改成 processing，
+ * 此时已无法区分 raw 原本是从 partial 还是从 failed/pending 过来的。
+ */
+ public boolean consumePartialResumeFlag(Long id) {
+ return partialResumeIds.remove(id);
}
@Transactional
diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/sse/WikiProgressBus.java b/mateclaw-server/src/main/java/vip/mate/wiki/sse/WikiProgressBus.java
new file mode 100644
index 00000000..55ed27ee
--- /dev/null
+++ b/mateclaw-server/src/main/java/vip/mate/wiki/sse/WikiProgressBus.java
@@ -0,0 +1,116 @@
+package vip.mate.wiki.sse;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+/**
+ * RFC-012 M3：Wiki 处理进度事件总线。
+ *
+ * 维护 {kbId → SseEmitter[]} 订阅表，把后端处理过程中的关键事件
+ * （raw.started / chunk.done / raw.completed / raw.failed）实时推送给所有
+ * 订阅者（一个 KB 可被多 Tab / 多用户同时打开）。
+ *
+ * 设计要点：
+ *
+ *
+ *
+ *
+ */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class WikiProgressBus {
+
+ /** 事件名常量（前端按 EventSource.addEventListener 名称匹配） */
+ public static final String EVENT_RAW_STARTED = "raw.started";
+ public static final String EVENT_ROUTE_DONE = "route.done";
+ public static final String EVENT_CHUNK_DONE = "chunk.done";
+ public static final String EVENT_RAW_COMPLETED = "raw.completed";
+ public static final String EVENT_RAW_FAILED = "raw.failed";
+ public static final String EVENT_HEARTBEAT = "heartbeat";
+
+ private final ObjectMapper objectMapper;
+
+ /** kbId → 订阅者列表 */
+ private final ConcurrentHashMap