RAG Knowledge Base System

基于 MySQL + Elasticsearch + Milvus 的三数据库架构 RAG 知识库检索系统

📖 目录

系统概述
核心特性
架构设计
快速开始
配置说明
项目结构
数据库设计
API接口
使用指南
测试
性能优化
常见问题
文档索引
更新日志

🎯 系统概述

本系统采用三数据库架构构建企业级 RAG（检索增强生成）知识库，实现文本和向量的混合检索能力。

核心理念

统一初始化 (/api/init)
    ↓ 一次性创建
MySQL (极简底座) + ES (文本索引 + doc_name冗余) + Milvus (向量索引 + doc_name冗余)
    ↓ 业务聚合
批量插入 (/api/documents/bulk) → 三库同步写入
    ↓ 检索
混合检索 (/api/search/hybrid) → ES/Milvus去重 → 补全content → 返回完整结果

✨ 核心特性

1. 多知识库逻辑隔离

基于 kb_id 字段实现多租户隔离：

支持多个独立知识库（财务、HR、技术研发等）
物理上共享一套数据库，逻辑上完全隔离
所有操作必须指定 kb_id，确保数据不串用

2. 极简Schema设计

MySQL - 仅4字段：

knowledge_document: doc_id, kb_id, doc_name, status
knowledge_chunk: chunk_id, kb_id, doc_id, content

Elasticsearch - 冗余doc_name：

chunk_id, doc_id, kb_id, doc_name, content

Milvus - 冗余doc_name：

chunk_id, doc_id, kb_id, doc_name, vector

3. 业务聚合接口

统一初始化: 单一 /api/init 入口，一次性创建所有数据库对象
批量插入: /api/documents/bulk 三库同步写入，支持自动覆盖更新，必须指定kb_id
混合检索: /api/search/hybrid ES+Milvus双路检索，自动去重补全，支持多知识库查询
删除文档: POST /api/documents/delete 统一的删除接口，支持单个和批量删除
删除知识库: POST /api/knowledge-bases/delete 删除整个知识库及其所有数据

4. 智能覆盖更新

批量插入接口支持Upsert（Update or Insert）：

文档ID重复时自动删除旧文档并插入新文档
无需先手动删除，简化更新流程
原子性保证，避免数据不一致
适用于文档内容更新、切片策略变更、向量重新生成等场景

5. 检索优化

ES/Milvus冗余 doc_name，减少MySQL查询
混合检索自动去重，优先ES结果（含content）
仅Milvus独有结果需从MySQL补全content
支持关键词精确匹配（ES）+ 语义相似匹配（Milvus）
强制kb_id过滤，确保搜索范围锁定在指定知识库

6. 架构优势

MySQL: 单一数据源（Single Source of Truth），极简Schema
Elasticsearch: 专业文本检索，支持中文ik分词
Milvus: 高性能向量检索，支持IVF_FLAT索引
数据一致性: 批量操作自动保证三库同步

🏗️ 架构设计

数据流向

写入流程

用户请求 → POST /api/documents/bulk
    ↓
检查文档ID是否存在？
    ↓ 是                ↓ 否
删除旧文档和切片      直接创建新文档
    ↓                    ↓
MySQL (status=SUCCESS) → MySQL chunks → ES chunks → Milvus vectors
    ↓ 任一步骤失败
MySQL文档 status=FAILED

检索流程

混合检索请求 → POST /api/search/hybrid
    ↓
ES文本检索 (含content, doc_name) + Milvus向量检索 (含doc_name, 无content)
    ↓ 按chunk_id去重
优先ES结果 → 对纯Milvus结果从MySQL补全content
    ↓
合并返回，标记 score_type

字段分离策略

数据库	存储字段	索引类型
MySQL	chunk_id, doc_id, content (极简)	B-Tree
Elasticsearch	chunk_id, doc_id, doc_name, content	倒排索引
Milvus	chunk_id, doc_id, doc_name, vector	IVF_FLAT

🚀 快速开始

前置要求

Python 3.8+
MySQL 8.0+
Elasticsearch 8.x (需安装ik分词器)
Milvus 2.x

安装步骤

克隆项目

git clone <repository-url>
cd code_rag_within_group

安装依赖

pip install -r requirements.txt

配置环境变量

cp .env.example .env
# 编辑 .env 文件，填写数据库连接信息

启动服务

python run.py

服务将在 http://localhost:8020 启动

访问文档

API 文档: http://localhost:8020/docs
OpenAPI: http://localhost:8020/openapi.json

📁 项目结构

code_rag_within_group/
├── app/
│   ├── api/                      # API 路由层
│   │   └── aggregated_routes.py      # 聚合业务接口
│   ├── core/                     # 核心配置和连接
│   │   ├── config.py                 # 配置管理
│   │   ├── mysql_client.py           # MySQL 客户端
│   │   ├── es_client.py              # ES 客户端
│   │   └── milvus_client.py          # Milvus 客户端
│   ├── models/                   # 数据模型
│   │   ├── knowledge.py              # 知识库数据模型
│   │   └ aggregated.py              # 聚合接口模型
│   ├── services/                 # 业务逻辑层
│   │   ├── mysql_service.py          # MySQL 服务
│   │   ├── es_service.py             # ES 服务
│   │   ├── milvus_service.py         # Milvus 服务
│   │   └ aggregated_service.py      # 聚合业务服务
│   └── main.py                   # FastAPI 主应用
├── tests/
│   └ test_aggregated_api.py      # 聚合接口测试
│   └ test.md                       # 测试数据
├── .env                          # 环境配置
├── .env.example                  # 环境配置示例
├── requirements.txt              # Python 依赖
├── run.py                        # 应用启动入口
├── README.md                     # 本文档
├── API_DOCUMENTATION.md          # API 接口文档
└── FINAL_TEST_REPORT.md          # 测试报告

💾 数据库设计

MySQL 表结构（极简）

knowledge_document (文档主表)

字段名	类型	约束	描述
`doc_id`	`VARCHAR(64)`	PRIMARY KEY	文档唯一 ID (文件 MD5)
`kb_id`	`VARCHAR(64)`	INDEX	知识库 ID（多租户隔离）
`doc_name`	`VARCHAR(255)`	NOT NULL	文件名称 (用于前端溯源展示)
`status`	`VARCHAR(20)`	NOT NULL	状态: PENDING, SUCCESS, FAILED

knowledge_chunk (切片明细表)

字段名	类型	约束	描述
`chunk_id`	`VARCHAR(64)`	PRIMARY KEY	切片唯一 ID
`kb_id`	`VARCHAR(64)`	INDEX	知识库 ID（多租户隔离）
`doc_id`	`VARCHAR(64)`	INDEX	关联的文档 ID
`content`	`TEXT`	NOT NULL	切片的原始文本内容

Elasticsearch Index

{
  "mappings": {
    "properties": {
      "chunk_id": { "type": "keyword" },
      "doc_id": { "type": "keyword" },
      "kb_id": { "type": "keyword" },
      "doc_name": { "type": "keyword" },
      "content": {
        "type": "text",
        "analyzer": "ik_max_word",
        "search_analyzer": "ik_smart"
      }
    }
  }
}

Milvus Collection

fields = [
    FieldSchema(name="chunk_id", dtype=VARCHAR, max_length=64, is_primary=True),
    FieldSchema(name="doc_id", dtype=VARCHAR, max_length=64),
    FieldSchema(name="kb_id", dtype=VARCHAR, max_length=64),
    FieldSchema(name="doc_name", dtype=VARCHAR, max_length=255),
    FieldSchema(name="vector", dtype=FLOAT_VECTOR, dim=4096)
]

📚 使用指南

1. 初始化数据库

curl -X POST http://localhost:8020/api/init

响应:

{
  "success": true,
  "message": "All databases initialized successfully."
}

创建内容:

MySQL: 2张表（knowledge_document, knowledge_chunk）
Elasticsearch: 1个索引（rag_knowledge）
Milvus: 1个Collection（rag_knowledge）

2. 批量插入文档（支持覆盖更新）

功能特性:

✅ 新文档插入
✅ 文档ID重复时自动覆盖更新（Upsert）
✅ 必须指定 kb_id，数据归属明确
✅ 原子性保证，自动处理三库同步

请求示例 - 新文档插入:

curl -X POST http://localhost:8020/api/documents/bulk \
  -H "Content-Type: application/json" \
  -d '{
    "kb_id": "kb_finance",
    "doc_id": "doc_001",
    "doc_name": "example.txt",
    "chunks": [
      {
        "chunk_id": "doc_001_0",
        "content": "这是一段测试文本",
        "vector": [0.1, 0.2, ...]
      }
    ]
  }'

请求示例 - 文档更新（自动覆盖）:

# 相同doc_id和kb_id，自动删除旧版本并插入新版本
curl -X POST http://localhost:8020/api/documents/bulk \
  -H "Content-Type: application/json" \
  -d '{
    "kb_id": "kb_finance",
    "doc_id": "doc_001",
    "doc_name": "example_v2.txt",
    "chunks": [
      {
        "chunk_id": "doc_001_0",
        "content": "这是更新后的内容",
        "vector": [0.2, 0.3, ...]
      }
    ]
  }'

注意事项:

⚠️ 向量维度必须与.env中VECTOR_DIMENSION一致
⚠️ kb_id 必填：所有数据必须归属于某个知识库
✅ 支持覆盖更新: 相同doc_id+kb_id会自动删除旧文档并插入新文档
💡 建议单次批量不超过100个切片
⚠️ 数据不可恢复: 覆盖更新会完全删除旧数据，请谨慎操作

3. 混合检索

功能特性:

✅ 支持单知识库或多知识库查询，通过逗号分隔 kb_id
✅ ES文本检索 + Milvus向量检索双路并发
✅ 自动去重补全

请求示例 - 单知识库查询:

curl -X POST http://localhost:8020/api/search/hybrid \
  -H "Content-Type: application/json" \
  -d '{
    "kb_id": "kb_finance",
    "query_text": "报销标准",
    "query_vector": [0.1, 0.2, ...],
    "top_k": 10
  }'

请求示例 - 多知识库查询:

# 同时搜索财务和HR知识库
curl -X POST http://localhost:8020/api/search/hybrid \
  -H "Content-Type: application/json" \
  -d '{
    "kb_id": "kb_finance,kb_hr",
    "query_text": "报销 工作时间",
    "query_vector": [0.1, 0.2, ...],
    "top_k": 10
  }'

响应示例:

{
  "results": [
    {
      "chunk_id": "doc_001_0",
      "doc_id": "doc_001",
      "kb_id": "kb_finance",
      "doc_name": "财务制度.txt",
      "content": "差旅报销标准：员工出差可报销交通费、住宿费和餐饮费。",
      "score": 5.3578,
      "score_type": "es_score"
    },
    {
      "chunk_id": "doc_002_0",
      "doc_id": "doc_002",
      "kb_id": "kb_hr",
      "doc_name": "考勤制度.txt",
      "content": "工作时间：周一至周五，早9点到晚6点。",
      "score": 0.8885,
      "score_type": "milvus_score"
    }
  ]
}

kb_id 参数说明:

单个知识库: "kb_id": "kb_finance"
多个知识库: "kb_id": "kb_finance,kb_hr,kb_tech" (逗号分隔)
返回结果中包含 kb_id 字段，标识数据来源

分数说明:

es_score: BM25文本相关性分数，越高越相关
milvus_score: L2向量距离，越低越相似

4. 删除文档（统一接口）

功能特性:

✅ 统一接口: 单一POST接口，支持单个和批量删除
✅ 灵活使用: doc_ids 参数可以是单个或多个文档ID
✅ 详细统计: 返回成功/失败数量和失败文档列表
✅ kb_id 必填: 确保只删除指定知识库的数据

请求示例 - 删除单个文档:

curl -X POST http://localhost:8020/api/documents/delete \
  -H "Content-Type: application/json" \
  -d '{
    "kb_id": "kb_finance",
    "doc_ids": ["doc_001"]
  }'

请求示例 - 批量删除多个文档:

curl -X POST http://localhost:8020/api/documents/delete \
  -H "Content-Type: application/json" \
  -d '{
    "kb_id": "kb_finance",
    "doc_ids": ["doc_001", "doc_002", "doc_003"]
  }'

响应示例:

{
  "success": true,
  "success_count": 3,
  "failed_count": 0,
  "failed_docs": [],
  "message": "Batch delete completed: 3 successful, 0 failed"
}

执行流程:

验证 kb_id 和 doc_ids 参数
遍历 doc_ids 列表中的每个文档
对每个文档执行：
- 删除MySQL切片（先删除切片，因为有外键约束）
- 删除MySQL文档记录
- 删除ES切片
- 删除Milvus切片
统计成功和失败数量，记录失败文档ID
返回详细的删除统计信息

参数说明:

参数	类型	必填	描述
`kb_id`	string	✅	知识库ID
`doc_ids`	array[string]	✅	文档ID列表（可以是单个或多个）

注意事项:

🗑️ 不可逆: 删除操作不可恢复，请谨慎操作
⚠️ kb_id 必填: 必须指定知识库ID，确保只删除该知识库的数据
📊 返回详情: 成功和失败的文档都会在响应中统计
🔒 隔离保证: 只删除指定 kb_id 的数据，不影响其他知识库
✅ 幂等性: 删除不存在的文档不会报错，会被视为"已删除"

5. 删除知识库

请求示例:

curl -X POST http://localhost:8020/api/knowledge-bases/delete \
  -H "Content-Type: application/json" \
  -d '{
    "kb_id": "kb_finance"
  }'

响应示例:

{
  "success": true,
  "deleted_doc_count": 10,
  "deleted_chunk_count": 50,
  "message": "Knowledge base deleted: 10 documents, 50 chunks removed"
}

功能说明:

⚠️ 危险操作: 删除整个知识库及其所有文档和切片
返回删除的文档数和切片数
不影响其他知识库的数据
此操作不可逆，请谨慎使用！

6. Python完整示例

import requests

BASE_URL = "http://localhost:8020"

# 初始化
requests.post(f"{BASE_URL}/api/init")

# 插入文档（单知识库）
doc_data = {
    "kb_id": "kb_finance",
    "doc_id": "doc_001",
    "doc_name": "example.txt",
    "chunks": [
        {
            "chunk_id": "doc_001_0",
            "content": "这是一段测试文本",
            "vector": [0.1] * 4096
        }
    ]
}
requests.post(f"{BASE_URL}/api/documents/bulk", json=doc_data)

# 检索（单知识库）
search_data = {
    "kb_id": "kb_finance",
    "query_text": "测试文本",
    "query_vector": [0.1] * 4096,
    "top_k": 10
}
result = requests.post(f"{BASE_URL}/api/search/hybrid", json=search_data)
print(result.json())

# 检索（多知识库）
search_data = {
    "kb_id": "kb_finance,kb_hr",
    "query_text": "测试",
    "query_vector": [0.1] * 4096,
    "top_k": 10
}
result = requests.post(f"{BASE_URL}/api/search/hybrid", json=search_data)
print(result.json())

# 删除单个文档（使用统一接口）
delete_data = {
    "kb_id": "kb_finance",
    "doc_ids": ["doc_001"]
}
result = requests.post(f"{BASE_URL}/api/documents/delete", json=delete_data)
print(result.json())

# 批量删除多个文档
batch_delete_data = {
    "kb_id": "kb_finance",
    "doc_ids": ["doc_001", "doc_002", "doc_003"]
}
result = requests.post(f"{BASE_URL}/api/documents/delete", json=batch_delete_data)
print(result.json())

# 删除知识库
delete_kb_data = {"kb_id": "kb_finance"}
result = requests.post(f"{BASE_URL}/api/knowledge-bases/delete", json=delete_kb_data)
print(result.json())

🎯 测试

运行测试脚本

python tests/test_aggregated_api.py

测试内容:

✅ 初始化三个数据库
✅ 预清理旧数据（支持重复测试）
✅ 读取测试文档并生成向量
✅ 批量插入数据到三库
✅ 测试混合检索功能（4个查询关键词）
✅ 测试文档删除功能

测试结果: 所有测试项完全通过 ✅

测试报告

完整测试报告: test_report.md

性能指标

指标	数值	备注
向量生成速度	~30ms/段落	Qwen3-Embedding-8B
批量插入耗时	~2.3秒	20个切片三库插入
混合检索延迟	~100ms	ES+Milvus双路检索
文档删除耗时	~8秒	三库同步删除

📊 性能优化

1. 批量插入优化

✅ 单次批量不超过100个切片
✅ 使用批量接口而非单个插入
✅ 向量维度建议不超过4096维

2. 检索优化

✅ 合理设置top_k参数（建议10-50）
✅ ES文本检索适合关键词精确匹配
✅ Milvus向量检索适合语义相似匹配
✅ 混合检索自动并发执行双路检索

3. 架构优化

✅ ES/Milvus冗余doc_name减少MySQL查询
✅ MySQL作为Single Source of Truth保证数据一致性
✅ 检索结果自动去重补全

❓ 常见问题

Q1: 向量维度不匹配怎么办？

A: 确保chunks[].vector和query_vector的维度与.env中VECTOR_DIMENSION配置一致。

# 检查配置
grep VECTOR_DIMENSION .env

Q2: 如何生成文档ID和切片ID？

A: 建议使用以下方式：

import hashlib

# 文档ID：文件MD5值（推荐，便于覆盖更新）
def generate_doc_id(file_path):
    with open(file_path, 'rb') as f:
        return hashlib.md5(f.read()).hexdigest()[:16]

# 切片ID：{doc_id}_para_{index}
def generate_chunk_id(doc_id, index):
    return f"{doc_id}_para_{index}"

提示: 使用稳定的文档ID可以方便地进行覆盖更新。

Q3: 混合检索的分数如何解读？

A:

es_score: BM25文本相关性分数，范围0-10+，越高越相关
milvus_score: L2向量距离，范围0-2+，越低越相似

Q4: 为什么删除文档要先改状态？

A: 保证事务性。删除流程：

MySQL文档状态改为FAILED（对检索隐形）
删除MySQL切片
删除ES切片
删除Milvus向量

这样即使删除过程中有检索请求，也不会返回部分删除的数据。

Q5: 可以只使用一个数据库吗？

A: 可以，但不推荐。三库架构各有优势：

MySQL: 数据持久化，单一数据源
Elasticsearch: 专业文本检索，支持中文分词
Milvus: 高性能向量检索

建议使用聚合接口（/api/）保证三库数据一致性。

Q6: 文档ID重复时会发生什么？

A: 系统会自动执行覆盖更新（Upsert）：

检测到文档ID已存在
自动删除旧文档及所有切片（MySQL + ES + Milvus）
插入新文档和切片
保证原子性，不会出现数据不一致

适用场景:

✅ 文档内容更新
✅ 切片策略变更
✅ 向量重新生成

注意事项:

⚠️ 覆盖更新会完全删除旧数据，无法恢复
⚠️ 重要数据建议使用新文档ID而不是覆盖

📝 更新日志

v2.1.0 (2026-06-23)

统一删除接口 - 合并删除文档和批量删除文档为一个POST接口

核心改进

统一API设计
- 新接口: POST /api/documents/delete
- 废弃: DELETE /api/documents/{doc_id} 和 POST /api/documents/batch-delete
灵活的参数
- doc_ids 支持单个或多个文档ID
- 单个删除: {"kb_id": "kb_finance", "doc_ids": ["doc_001"]}
- 批量删除: {"kb_id": "kb_finance", "doc_ids": ["doc_001", "doc_002"]}
详细的响应
- success_count: 成功删除数量
- failed_count: 失败数量
- failed_docs: 失败文档列表
增强的容错性
- 幂等性: 删除不存在的文档不报错
- 混合场景: 支持部分存在、部分不存在

迁移指南

# 旧方式（已废弃）
requests.delete(f"{BASE_URL}/api/documents/doc_001?kb_id=kb_finance")

# 新方式（推荐）
delete_data = {"kb_id": "kb_finance", "doc_ids": ["doc_001"]}
requests.post(f"{BASE_URL}/api/documents/delete", json=delete_data)

完整更新日志: CHANGELOG.md

📚 文档索引

API文档: API_DOCUMENTATION.md - 完整的API接口文档
更新日志: CHANGELOG.md - 版本更新记录
测试报告: test_report.md - 测试结果和性能指标

Name		Name	Last commit message	Last commit date
app		app
docs		docs
tests		tests
.env.example		.env.example
.gitignore		.gitignore
README.md		README.md
requirements.txt		requirements.txt
run.py		run.py

Latest commit

History

Folders and files

Repository files navigation

RAG Knowledge Base System

📖 目录

🎯 系统概述

核心理念

✨ 核心特性

1. 多知识库逻辑隔离

2. 极简Schema设计

3. 业务聚合接口

4. 智能覆盖更新

5. 检索优化

6. 架构优势

🏗️ 架构设计

数据流向

写入流程

检索流程

字段分离策略

🚀 快速开始

前置要求

安装步骤

📁 项目结构

💾 数据库设计

MySQL 表结构（极简）

knowledge_document (文档主表)

knowledge_chunk (切片明细表)

Elasticsearch Index

Milvus Collection

📚 使用指南

1. 初始化数据库

2. 批量插入文档（支持覆盖更新）

3. 混合检索

4. 删除文档（统一接口）

5. 删除知识库

6. Python完整示例

🎯 测试

运行测试脚本

测试报告

性能指标

📊 性能优化

1. 批量插入优化

2. 检索优化

3. 架构优化

❓ 常见问题

Q1: 向量维度不匹配怎么办？

Q2: 如何生成文档ID和切片ID？

Q3: 混合检索的分数如何解读？

Q4: 为什么删除文档要先改状态？

Q5: 可以只使用一个数据库吗？

Q6: 文档ID重复时会发生什么？

📝 更新日志

v2.1.0 (2026-06-23)

核心改进

迁移指南

📚 文档索引

About

Resources

Stars

Watchers

Forks

Releases

Packages

Contributors

Languages