企业级记忆知识库：短期上下文与四层记忆的 RAG/向量检索实现

// 嵌入：文本 → 向量
public interface EmbeddingClient {
    float[] embed(String text);           // 例如返回 float[1024]
}

// 向量库：向量 + 元数据的存取
public interface VectorDatabase {
    void upsert(String collection, float[] vector, Map<String, Object> metadata);
    List<Hit> search(String collection, float[] queryVector, int topK, double threshold);
}

// 记忆门面：短期 + 长期的读写入口
public interface MemoryService {
    List<Message> loadShortTerm(String conversationId, int contextRounds);
    Map<String, String> loadLongTerm(Long tenantId, Long userId, Long agentId, String query);
}

// 1. 优先读 Redis List
List<String> raw = redisUtil.range(key, 0, Long.MAX_VALUE);
if (raw != null && !raw.isEmpty()) {
    // 命中：JSON 反序列化，翻转为 oldest-first
    return raw.stream().map(json -> JSON.parseObject(json, ChatMessageDto.class))
              .collect(Collectors.toList());
}
// 2. 未命中/异常：回源 MySQL，再按倒序 leftPush 回写
List<ChatMessageDto> fromDb = messageDao.selectByConversationId(cid);
for (int i = fromDb.size() - 1; i >= 0; i--) {
    redisUtil.leftPush(key, JSON.toJSONString(fromDb.get(i)));
}
redisUtil.expire(key, CACHE_TTL_SECONDS);
// 3. 超出上限从尾部淘汰最旧
while (redisUtil.size(key) > MAX_CACHED_MESSAGES) {
    redisUtil.rightPop(key);
}
return fromDb;

trimAlternationFromEnd(chatMsgs);                                  // 保证 user/assistant 交替
int windowSize = MAX_TOKEN_WINDOW_SIZE - SUMMARY_TOKEN_BUDGET;      // 预留摘要空间（约 2000 tokens）
int totalTokens = 0, recentTokens = 0;
boolean overLimit = false;
List<Message> recentWindow = new ArrayList<>();
for (int i = chatMsgs.size() - 1; i >= 0; i--) {
    Message message = chatMsgs.get(i);
    int tokens = TikTokensUtil.tikTokensCount(message.getText());
    totalTokens += tokens;
    if (!overLimit && recentTokens + tokens <= windowSize) {
        recentWindow.add(0, message);                               // 倒序插入 = 正序输出
        recentTokens += tokens;
    } else {
        overLimit = true;                                           // 一旦超限，只统计总量不再收纳
    }
}
if (totalTokens <= MAX_TOKEN_WINDOW_SIZE) {
    removeLeadingAssistant(chatMsgs);
    return mergeSystemAndChat(systemMsgs, chatMsgs);                // 没超：全量返回
}
removeLeadingAssistant(recentWindow);
String summaryMd = iMemoryRpcService.getConversationSummaryMd(Long.parseLong(cid));
if (StringUtils.isNotBlank(summaryMd)) {
    recentWindow.add(0, new SystemMessage(summaryMd));              // 超了：用摘要顶替被裁掉的历史
}
return recentWindow;

messages.forEach(message -> {
    String conversationId0 = conversationId;
    if (conversationId0.startsWith("agent:")) {
        // 子 Agent 的非 ChatMessage 不写入主会话，避免虚拟会话污染主链路
        if (message instanceof ChatMessageDto) {
            conversationId0 = conversationId.replace("agent:", "");
        } else {
            return;
        }
    }
    ChatMessageDto chatMessage = ...;   // 类型转换、清洗文本、补租户字段
    // 先落 MySQL，Redis 只作为可失效的热点缓存
    Long messageId = TenantFunctions.callWithTenantId(chatMessage.getTenantId(),
        () -> conversationDomainService.addConversationMessage(conversationMessage));
    chatMessage.setIndex(messageId);
    try {
        String key = generateConversationKey(conversationId0);
        redisUtil.leftPush(key, JSON.toJSONString(chatMessage));
        redisUtil.expire(key, CACHE_TTL_SECONDS);
        if (redisUtil.size(key) > MAX_CACHED_MESSAGES) {   // 最新在表头，超 200 条从尾部淘汰
            redisUtil.rightPop(key);
        }
    } catch (Exception e) {
        // 缓存失败不回滚已完成的 MySQL 写入，读取时会回源数据库
        log.warn("Failed to cache message to Redis, conversationId={}", conversationId0, e);
    }
});

@DubboService
public class OrderFacadeImpl implements OrderFacade {
    // 下单：先风控 → 冻结库存 → 核销优惠券 → 落库 → 扣减库存
    @Override
    public CreateOrderResponse createOrder(CreateOrderRequest req) { ... }
}

// 用 javaparser 还原调用链骨架（示意，简化了类型解析）
public String extractInvocationChain(String source) {
    CompilationUnit cu = StaticJavaParser.parse(source);
    StringBuilder sb = new StringBuilder();
    cu.findAll(MethodCallExpr.class).forEach(call -> {
        // 只保留本地 service/DAO 调用，过滤 JDK/工具类噪音
        if (isBusinessCall(call)) {
            sb.append(call.getScope().map(s -> s + ".").orElse(""))
              .append(call.getNameAsString()).append(" -> ");
        }
    });
    return sb.toString();
}

private MemoryJudgeResult judgeByLLM(Long tenantId, Long modelId,
        List<MemoryMessage> fullContext, List<MemoryMessage> newMessages) {
    Set<String> newMessageKeys = newMessages.stream()
        .map(m -> (m.getRole() != null ? m.getRole() : "user")
            + ":" + (m.getContent() != null ? m.getContent() : ""))
        .collect(Collectors.toSet());

    StringBuilder content = new StringBuilder("## 完整对话上下文\n\n");
    for (MemoryMessage message : fullContext) {
        String role = message.getRole() != null ? message.getRole() : "user";
        String text = message.getContent() != null ? message.getContent() : "";
        content.append(newMessageKeys.contains(role + ":" + text)
                ? "[[NEW] " : "[")
            .append(role).append("]: ").append(text).append("\n");
    }

    // 调用 LLM，约束输出 JSON：{ category, scope, importance, content }
    MemoryJudgeResult result = iModelRpcService.call(
        tenantId, modelId, JUDGE_SYSTEM_PROMPT, content.toString(),
        new ParameterizedTypeReference<MemoryJudgeResult>() {});
    return result == null ? null : validateAndFixResult(result);
}

Map<String, Object> metadata = Map.of(
    "tenantId",   tenantId,            // 多租户隔离
    "scope",      "agent_order_agent", // 对应 cube：user_profile 或 agent_{id}
    "type",       "tool_mem",          // text_mem / pref_mem / skill_mem / tool_mem
    "name",       "OrderFacade#createOrder",
    "importance", 8,
    "version",    "v3",                // 版本，旧版本要能降权/失效
    "source",     "codebase",          // 来源：codebase / conversation / doc / manual
    "updatedAt",  System.currentTimeMillis()
);

// 以接口为单位的 chunk（简化）
public List<Chunk> chunkCodeMemory(List<InterfaceMemory> interfaces) {
    return interfaces.stream().map(itf ->
        new Chunk(
            String.format("【接口】%s#%s\n调用顺序：%s\n异常约定：%s",
                itf.clazz(), itf.method(), itf.invocationChain(), itf.exceptionContract()),
            Map.of("type", "tool_mem", "name", itf.clazz() + "#" + itf.method())
        )
    ).collect(Collectors.toList());
}

public class OpenAiEmbeddingClient implements EmbeddingClient {
    private final String apiKey;
    private final RestClient rest = RestClient.create();

    @Override
    public float[] embed(String text) {
        // POST /v1/embeddings  { model: "text-embedding-3-small", input: text }
        EmbeddingResponse resp = rest.post()
            .uri("https://api.openai.com/v1/embeddings")
            .header("Authorization", "Bearer " + apiKey)
            .body(Map.of("model", "text-embedding-3-small", "input", text))
            .retrieve()
            .body(EmbeddingResponse.class);
        return resp.data().get(0).embedding();   // 例如 float[1536]
    }
}

public void persistMemory(Chunk chunk, String scope) {
    float[] vector = embeddingClient.embed(chunk.text());
    vectorDatabase.upsert(
        scope,                     // 集合名 = cube：user_profile 或 agent_{agentId}
        vector,
        chunk.metadata()
    );
}

// exact → contains → 字符级 Jaccard(0.7) → 短文本 Levenshtein(0.8)
public String calculateSimilarity(String a, String b) {
    if (a.equals(b)) return "exact";                      // 完全相同
    if (a.contains(b) || b.contains(a)) return "contains"; // 包含关系
    double jaccard = jaccard(a, b);                        // 字符级 Jaccard 相似度
    if (jaccard >= 0.7) return "similar";
    if (a.length() < 50 && b.length() < 50) {
        double lev = levenshteinRatio(a, b);               // 短文本用编辑距离
        if (lev >= 0.8) return "similar";
    }
    return "none";
}

// 先写新记忆，只有返回结果确认新增成功，才删除冲突旧记忆
AddMemoryResult added = memosClient.addMemory(request);  // async, fine 模式
if (added.hasNewMemories()) {
    if (!memoryIdsToDelete.isEmpty()) {
        try {
            memosClient.deleteMemories(memoryIdsToDelete);
        } catch (Exception e) {
            log.warn("delete conflict memories failed, keep new ones", e); // best-effort
        }
    }
}

final int finalContextRounds = contextRounds;
CompletableFuture<List<Message>> contextMessagesFuture = CompletableFuture.supplyAsync(() -> {
    if (finalContextRounds <= 0) {                       // contextRounds=0 合法，跳过短期
        return new ArrayList<Message>();
    }
    return new ArrayList<>(
        chatMemory.get(agentContext.getConversationId(), finalContextRounds * 3));
}, memoryLoadExecutor);

CompletableFuture<Map<String, String>> longMemoryFuture = CompletableFuture.supplyAsync(() -> {
    if (agentContext.getAgentConfig().getOpenLongMemory() != AgentConfig.OpenStatus.Open) {
        return Collections.emptyMap();                    // 未开启长期记忆：直接返回空
    }
    try {
        AgentComponentConfigDto modelConfig =
            agentContext.getAgentConfig().getModelComponentConfig();
        if (modelConfig == null || modelConfig.getTargetId() == null) {
            return Collections.emptyMap();                // 未绑定模型：跳过长期检索
        }
        boolean justKeywordMatch = resolveJustKeywordMatch(agentContext);
        return conversationApplicationService.queryMemory(
            agentContext.getUser().getTenantId(), agentContext.getUser().getId(),
            agentContext.getAgentConfig().getId(), modelConfig.getTargetId(),
            agentContext.getOriginalMessage(), "", justKeywordMatch,
            agentContext.isFilterSensitive());
    } catch (Exception e) {
        log.warn("查询长期记忆失败", e);                 // 长期记忆是增强能力，失败不阻塞主对话
        return Collections.emptyMap();
    }
}, memoryLoadExecutor);

// 汇合点：主流程等待两条链路都完成
agentContext.setContextMessages(contextMessagesFuture.join());
Map<String, String> longMemoryMap = longMemoryFuture.join();

public List<MemoryUnit> queryMemory(String userMessage, String context) {
    String query = buildSearchQuery(userMessage, context);   // context 非空时只取前 200 字符拼接
    float[] qv = embeddingClient.embed(query);               // ① query embedding

    List<Hit> hits = vectorDatabase.search(
        List.of("user_profile", "agent_" + agentId),         // ② 一次检索同时覆盖两个 cube
        qv,
        DEFAULT_TOP_K,                                        // ③ topK
        0.45);                                                // ④ 阈值：低于此分不返回

    // ⑤ 转换阶段：过滤低分、截断单条、按 score 降序
    List<MemoryUnit> result = hits.stream()
        .filter(h -> h.score() >= 0.3)                        // 低分过滤
        .map(h -> new MemoryUnit(h.content().substring(0, Math.min(1000, h.content().length()))))
        .collect(Collectors.toList());
    result.sort(Comparator.comparingDouble(MemoryUnit::score).reversed());
    return result;
}

List<Hit> sparse = esClient.bm25(query, topK);       // 词法路
List<Hit> dense  = vectorDB.search(queryVec, topK);  // 向量路
List<Hit> merged = rrfFusion(sparse, dense);          // 倒数排名融合

List<Hit> candidates = vectorDB.search(queryVec, 100);   // 粗排：快，召回高
List<Hit> reranked = reranker.rerank(query, candidates);  // 精排：慢，准
return reranked.subList(0, 5);

int budget = DEFAULT_LONG_MEMORY_TOKEN_BUDGET;              // 4000 tokens
int userProfileRawTokens =
    TikTokensUtil.tikTokensCount(userProfileRaw != null ? userProfileRaw : "");

if (userProfileRawTokens <= budget * 0.6) {                 // user_profile 最多占 60%
    userProfile = userProfileRaw;                           // 实际没占满，剩余让给 Agent 记忆
    agentMemory = truncateLongMemory(agentMemoryRaw, budget - userProfileRawTokens);
} else {
    userProfile = truncateLongMemory(userProfileRaw, (int) (budget * 0.6));
    int userProfileTokens =                                 // 重新计算截断后 token，别把估算带进来
        TikTokensUtil.tikTokensCount(userProfile != null ? userProfile : "");
    agentMemory = truncateLongMemory(agentMemoryRaw, budget - userProfileTokens);
}

String[] lines = longMemory.split("\n", -1);
for (String line : lines) {
    int lineTokens = TikTokensUtil.tikTokensCount(line + "\n");
    if (currentTokens + lineTokens > budget) {
        break;                                              // 达到预算停止，不拆散单条
    }
    truncated.append(line);
    currentTokens += lineTokens;
}