RAG Agent Platform：多租户智能体 SaaS 平台
GitHub：NEDONION/rag-agent-platform · 在线演示：39.97.58.27/explore
Java 17 Spring Boot 3.2.3 LangChain4j Next.js 15 React 19 PostgreSQL + PGVector RabbitMQ S3 MCP
本文是架构与设计沉淀；本地 环境搭建步骤见 RAG Agent Platform - LLM/SaaS 环境搭建。
1 这个项目在解决什么
公司里想用大模型查内部资料，市面上现成的 SaaS 都能做。但真到落地会卡在两件事上：
第一，文档得传到别人服务器上。合同、客户名单、内部方案，法务那一关就过不去。
第二，几个部门共用一套，谁能看哪个知识库、谁的额度用超了、这个月哪个部门花了多少钱 —— 这些问题现成产品要么不给拆，要么按人头另收一笔。
这个平台就是把整条链路自己做一遍：文档进来怎么处理、问题进来怎么检索、Agent 怎么调工具、多个租户怎么互不干扰。数据不出自己的机器，模型供应商想换就换。
覆盖的是一条完整链路，不是某个单点能力：
选择自托管而不是接第三方 SaaS，核心诉求是：数据、模型和部署方式全部可控，模型提供商可以一键切换。
2 整体架构
2.1 技术分层
| 区域 | 实现 |
|---|---|
| 后端 | Java 17、Spring Boot 3.2.3、LangChain4j 1.0.4.3-beta7 |
| 前端 | Next.js 15、React 19、Radix UI + Tailwind |
| 数据 | PostgreSQL 14+ 与 PGVector |
| ORM | MyBatis-Plus 3.5.11（多租户插件、乐观锁、逻辑删除） |
| 异步任务 | RabbitMQ（Direct/Topic Exchange + 手动 ACK + DLX） |
| 文件存储 | S3 兼容对象存储（七牛 KODO / AWS S3 / 腾讯 COS） |
| 流式响应 | Server-Sent Events (SSE) |
2.2 DDD 四层
严格分层，上层依赖下层，领域层通过 Repository 接口反向依赖基础设施：
Interfaces ──→ Application ──→ Domain ──→ Infrastructure
↑ │
└──────────────┘
| 层 | 职责 | 典型内容 |
|---|---|---|
| Interfaces | 收请求、转 DTO、参数校验、异常兜底 | @RestController、@WebSocketHandler |
| Application | 编排领域服务、事务边界、流程组合 | AgentAppService、RagAppService |
| Domain | 核心业务规则、领域实体、领域服务 | AgentDomainService、EmbeddingDomainService |
| Infrastructure | 数据访问实现、外部服务集成 | AgentRepositoryImpl、RerankForestApi |
应用层做编排，不写业务规则；领域层写业务规则，不碰技术细节。例如创建 Agent：
@Service
public class AgentAppService {
private final AgentDomainService agentDomainService;
private final ToolDomainService toolDomainService;
@Transactional
public AgentDTO createAgent(CreateAgentRequest request) {
// 应用层只负责编排多个领域服务 + 划事务边界
AgentEntity agent = agentDomainService.createAgent(
request.getName(), request.getSystemPrompt());
if (request.getToolIds() != null) {
toolDomainService.bindToolsToAgent(agent.getId(), request.getToolIds());
}
return AgentAssembler.toDTO(agent);
}
}
3 RAG 模块：文档处理流水线
这是整个项目工程量最大、也最能体现异步设计价值的部分。
3.1 三阶段异步流水线
为什么要拆成两段 MQ？ OCR 和向量化的失败原因、重试成本、资源瓶颈完全不同：OCR 卡在 Vision 模型的吞吐，向量化卡在 Embedding API 的限流。拆开之后可以独立设置并发消费者数量和重试策略，一段失败不会拖累另一段已完成的工作。
3.2 策略模式处理多格式
Spring 会把同类型 Bean 按名字注入 Map，这个特性让策略路由几乎零成本：
public interface RagDocSyncOcrStrategy {
void handle(RagDocSyncOcrMessage message, String strategy);
byte[] getFileData(RagDocSyncOcrMessage message, String strategy);
Map<Integer, String> processFile(byte[] fileBytes, int totalPages);
}
@Service("ragDocSyncOcr-PDF")
public class PDFRagDocSyncOcrStrategyImpl implements RagDocSyncOcrStrategy { /* ... */ }
@Service("ragDocSyncOcr-WORD")
public class WORDRagDocSyncOcrStrategyImpl implements RagDocSyncOcrStrategy { /* ... */ }
@Component
public class RagDocSyncOcrContext {
@Resource
private Map<String, RagDocSyncOcrStrategy> strategyMap; // Spring 自动注入
public RagDocSyncOcrStrategy getTaskExportStrategy(String fileType) {
return strategyMap.get("ragDocSyncOcr-" + fileType);
}
}
新增格式只要加一个 @Service("ragDocSyncOcr-XXX")，不用改任何调度代码。
3.3 状态机管理文件生命周期
文件处理有六个状态、两条异常分支，用 if-else 写必然失控，改成状态处理器：
public interface FileProcessingStateProcessor {
boolean canHandle(FileProcessingEventEnum event);
FileProcessingStatusEnum handle(String fileId, String userId, FileProcessingEventEnum event);
FileProcessingStatusEnum currentState();
}
@Service
public class FileProcessingStateMachineService {
private final Map<FileProcessingStatusEnum, FileProcessingStateProcessor> stateProcessors;
public boolean handleEvent(String fileId, String userId, FileProcessingEventEnum event) {
FileDetailEntity file = fileDetailRepository.selectById(fileId);
FileProcessingStatusEnum current =
FileProcessingStatusEnum.fromCode(file.getProcessingStatus());
FileProcessingStateProcessor processor = stateProcessors.get(current);
if (!processor.canHandle(event)) {
return false; // 非法状态转换直接拒绝，而不是写坏数据
}
FileProcessingStatusEnum next = processor.handle(fileId, userId, event);
fileDetailRepository.update(fileId, next);
return true;
}
}
收益：消息重复投递时，非法转换会被自然拒绝，等于免费获得了幂等性。
4 检索链路：召回 → 精排 → 扩展
这是决定 RAG 效果上限的地方。三段式设计：
三个关键参数的作用：
| 参数 | 默认值 | 作用 | 调优经验 |
|---|---|---|---|
candidateMultiplier | 2 | 召回候选倍数，先粗后精 | 精排模型质量越好，倍数可以越大 |
minScore | 0.7 | 相似度阈值 | 太高会空召回，所以配了降级到 0.3 的兜底 |
enableQueryExpansion | true | 取相邻页补全上下文 | 对被切断的表格、多页流程说明效果显著 |
Rerank 的实现是把召回结果的文本抽出来送给重排 API，再按返回顺序重排原始 match：
public List<EmbeddingMatch<TextSegment>> rerankDocument(
EmbeddingSearchResult<TextSegment> searchResult, String question) {
List<String> documents = searchResult.matches().stream()
.map(match -> match.embedded().text())
.toList();
RerankRequest request = new RerankRequest();
request.setModel(rerankProperties.getModel()); // bge-reranker-v2-m3
request.setQuery(question);
request.setDocuments(documents);
RerankResponse response = rerankForestApi.rerank(
rerankProperties.getApiUrl(), rerankProperties.getApiKey(), request);
List<EmbeddingMatch<TextSegment>> reranked = new ArrayList<>();
for (RerankResponse.SearchResult result : response.getResults()) {
reranked.add(searchResult.matches().get(result.getIndex()));
}
return reranked;
}
为什么必须要精排：向量召回是双塔结构，query 和 doc 各自独立编码，只能算粗粒度相似；Rerank 是 cross-encoder，query 和 doc 一起进模型，能捕捉细粒度交互。用「召回追求不漏，精排追求准」这个分工，比单纯调高召回阈值效果好得多。
5 知识库版本化：引用型 vs 快照型
知识库要能发布、被别人安装，就必须回答一个问题：安装后原作者更新了，安装方要不要跟着变？
这个项目给了两种答案，让用户自己选：
| 特性 | 引用型 REFERENCE | 快照型 SNAPSHOT |
|---|---|---|
| 版本号 | 固定 0.0.1 | >= 1.0.0 |
| 数据存储 | 引用原始数据集 | 完整复制到用户空间 |
| 向量存储 | 共享原始向量 | 独立向量副本 |
metadata.dataset_id | 原始 ragId | userRagId |
| 数据更新 | 实时同步 | 版本固化 |
| 存储成本 | 低（共享） | 高（独立副本） |
| 适用场 景 | 协作知识库、动态更新 | 稳定版本发布、数据隔离 |
快照安装要完整复制文件、文档单元，并用 userRagId 作为 dataset_id 重新向量化——这一步不能省，否则检索过滤条件会串到原始知识库上，造成跨租户数据泄漏。
@Transactional
public UserRagEntity createSnapshotInstall(String userId, String ragVersionId) {
RagVersionEntity ragVersion = ragVersionRepository.selectById(ragVersionId);
UserRagEntity userRag = new UserRagEntity();
userRag.setUserId(userId);
userRag.setRagVersionId(ragVersionId);
userRag.setInstallType(InstallType.SNAPSHOT);
userRagRepository.insert(userRag);
for (RagVersionFileEntity versionFile : listVersionFiles(ragVersionId)) {
UserRagFileEntity userFile = copyFile(userRag, versionFile);
for (RagVersionDocumentEntity versionDoc : listVersionDocs(versionFile.getId())) {
UserRagDocumentEntity userDoc = copyDoc(userRag, userFile, versionDoc);
// 关键：用 userRagId 作为 dataset_id 重新写向量
reEmbedDocumentToUserSpace(userDoc, userRag, userId);
}
}
return userRag;
}
6 Agent 模块与 MCP 集成
Agent 对话的完整链路：
数据库侧用 agent_execution_summary（汇总）+ agent_execution_details（逐步明细）两张表记录执行链路，这样既能在列表页快速展示成本和耗时，又能在详情页回放每一步工具调用。
7 高可用设计
三个层面的兜底，都是被线上问题倒逼出来的：
| 场景 | 策略 |
|---|---|
| 模型不可用 | 主模型失败 → 检查平替模型 → 切换 → 记录降级日志 |
| 消息处理失败 | 手动 NACK → 重新入队 → 最多重试 3 次 → 进入死信队列人工处理 |
| 数据库压力 | 主从复制、读写分离、连接池、慢查询监控 |
多租户隔离靠 MyBatis-Plus 的 TenantLineHandler 在 SQL 层自动拼租户条件，而不是靠业务代码每次手写 where user_id = ?——后者只要漏一处就是数据泄漏。
8 沉淀下来的经验
1. 异步流水线要按「失败域」切分，而不是按「步骤」切分。 OCR 与向量化拆开，是因为它们的失败原因和重试成本不同。如果只是按步骤机械拆分，会得到一堆职责相似、却要各自维护重试逻辑的队列。
2. 状态机不只是为了代码整洁，更是为了幂等。 消息队列的 at-least-once 语义意味着重复消费一定会发生。状态机拒绝非法转换，天然屏蔽了重复投递。
3. 检索质量的最大收益点是精排，不是调阈值。
双塔召回和 cross-encoder 精排是不同量级的能力。与其反复调 minScore，不如加一层 Rerank。
4. 多租户的向量隔离必须在 metadata 层面做实。
共享一张 embeddings 表时，dataset_id 就是唯一的隔离边界，快照安装必须重新写向量。
5. 部署脚本 ≠ 一键安装。 仓库里的 Docker Compose 是作者的部署参考，依赖外部配置的 PostgreSQL 和对象存储。把这一点在 README 里说清楚，比假装开箱即用更负责任。
9 待办与演进方向
- 微服务拆分（各模块已按领域边界隔离，具备拆分条件）
- 引入 Nacos/Apollo 做配置中心，SkyWalking 做分布式追踪
- Redis 缓存热点数据（知识库元信息、Agent 配置）
- MCP 工具的完整导入流程（当前需要手动配置 gateway 与容器）
参考
- 仓库文档：ARCHITECTURE · RAG_MODULE · AGENT_MODULE · DATABASE
- 相关笔记：RAG 面试题汇总 · AI Infra 学习资源导航