Skip to content

Commit 93fdb34

Browse files
jixuajixua
authored
Merge pull request #152 from ql-link/claude/reverent-jennings-1f448e
docs(internals): 补全稀疏向量/ES 检索/预分词/缓存模块文档并刷新结构索引
2 parents df277bf + eb2b0c8 commit 93fdb34

6 files changed

Lines changed: 534 additions & 16 deletions

File tree

‎docs/README.md‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -33,12 +33,17 @@
3333
| [pipeline_architecture.md](internals/pipeline_architecture.md) | 解析 Pipeline 架构 |
3434
| [parse_task_pipeline.md](internals/parse_task_pipeline.md) | 解析任务流水线状态机 |
3535
| [recall_pipeline.md](internals/recall_pipeline.md) | 召回 Pipeline 架构 |
36+
| [recall_http_api.md](internals/recall_http_api.md) | 召回 HTTP 入口与会话/鉴权 |
3637
| [file_parser.md](internals/file_parser.md) | 文件解析器（含回退链） |
3738
| [markdown_parser.md](internals/markdown_parser.md) | Markdown 解析与 LLM 增强 |
3839
| [chunking.md](internals/chunking.md) | 分块策略与流水线 |
39-
| [vectorization.md](internals/vectorization.md) | 向量化模块 |
40+
| [vectorization.md](internals/vectorization.md) | 向量化模块（dense） |
41+
| [sparse_vector.md](internals/sparse_vector.md) | 稀疏向量（BGE-M3）编码与索引 |
42+
| [preprocessor.md](internals/preprocessor.md) | ES 预分词（RAGFlow） |
43+
| [es_index_storage.md](internals/es_index_storage.md) | ES 索引与 BM25 检索 |
4044
| [mq.md](internals/mq.md) | MQ 中间件实现 |
4145
| [llm.md](internals/llm.md) | LLM 调用模块 |
46+
| [cache.md](internals/cache.md) | 缓存基础设施（Redis） |
4247
| [object_storage.md](internals/object_storage.md) | 对象存储 |
4348
| [naming_conventions.md](internals/naming_conventions.md) | 命名约定 |
4449

‎docs/internals/cache.md‎

Lines changed: 71 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,71 @@
1+
# 缓存基础设施
2+
3+
本文说明 `src/cache/`：基于 Redis 的缓存层，主要服务于 LLM 配置/系统厂商的读多写少场景，并配合 MQ 缓存同步消息做失效。
4+
5+
```text
6+
src/cache/
7+
├── __init__.py # 导出 redis_client / cache_manager 两个全局单例
8+
├── redis_client.py # 异步 Redis 连接单例
9+
└── cache_manager.py # CacheManager + 后端抽象（Redis / Null）
10+
```
11+
12+
---
13+
14+
## 1. Redis 客户端（redis_client.py）
15+
16+
`RedisClient` 是进程级单例（`__new__` 控制），封装 `redis.asyncio`：
17+
18+
- `initialize()`：懒建连接，`redis.from_url(REDIS_URL or REDIS_HOST, password=REDIS_PASSWORD, db=REDIS_DB, decode_responses=True)`。在应用启动（[src/main.py](../../src/main.py)）时调用。
19+
- `close()`：应用关闭时释放连接。
20+
- 模块级全局 `redis_client` 供 `cache_manager` 使用。
21+
22+
配置项：`REDIS_HOST` / `REDIS_PORT` / `REDIS_DB` / `REDIS_PASSWORD`，或直接给 `REDIS_URL`（`config.py` 有 `field_validator` 在未显式提供时由 host/port/db/password 拼装）。
23+
24+
---
25+
26+
## 2. 缓存管理器（cache_manager.py）
27+
28+
### 2.1 后端抽象
29+
30+
`CacheBackend`（ABC）定义 `get` / `set` / `delete` / `keys` 四个异步方法，两个实现：
31+
32+
| 后端 | 用途 |
33+
| --- | --- |
34+
| `RedisCacheBackend` | 生产环境，委托给全局 `redis_client` |
35+
| `NullCacheBackend` | 测试环境，所有操作 no-op（`get` 返 `None`，`keys` 返 `[]`） |
36+
37+
抽象后端的意义：测试不必起 Redis，注入 `NullCacheBackend` 即可让缓存逻辑"透明穿透"。
38+
39+
### 2.2 `CacheManager`
40+
41+
在后端之上提供序列化、键管理与失效：
42+
43+
- **JSON 序列化**：`set` 用 `json.dumps(value, default=str)` 写入；`get` 读出后 `json.loads`，解析失败回退原始字符串。
44+
- **TTL**：默认 `DEFAULT_TTL = 600`（10 分钟），`set(key, value, ttl=...)` 可覆盖。
45+
- **键前缀**（集中定义，避免散落硬编码）：
46+
- `llm:user:{user_id}:config` / `:configs` / `:default` —— 用户 LLM 配置
47+
- `llm:system:providers` / `llm:system:provider:<type>` —— 系统厂商
48+
- 配套静态方法 `user_config_key` / `user_configs_key` / `user_default_key` / `system_providers_key` / `system_provider_key` 生成键。
49+
- **批量失效**：`clear_user_cache(user_id)` 按 `llm:user:{id}:*` 删除，`clear_system_cache()` 按 `llm:system:*` 删除。
50+
- 模块级全局 `cache_manager = CacheManager()`，默认 `RedisCacheBackend`。
51+
52+
`__init__.py` 导出 `redis_client` 与 `cache_manager` 两个单例。
53+
54+
---
55+
56+
## 3. 谁在用
57+
58+
| 调用方 | 用途 |
59+
| --- | --- |
60+
| [src/main.py](../../src/main.py) | 启动/关闭时初始化、释放 Redis |
61+
| [src/services/config_reader_service.py](../../src/services/config_reader_service.py) | 读 LLM 用户配置/系统厂商，缓存命中优先 |
62+
| [src/services/cache_sync_service.py](../../src/services/cache_sync_service.py) | 消费 MQ 缓存同步消息后失效对应键 |
63+
| [src/api/recall_session_auth.py](../../src/api/recall_session_auth.py) | 召回会话鉴权相关缓存 |
64+
65+
LLM 配置读取链路见 [llm.md](llm.md)，缓存同步消息契约见 [mq_contracts.md](../api/mq_contracts.md)。
66+
67+
---
68+
69+
## 4. 测试约定
70+
71+
涉及缓存的服务测试注入 `NullCacheBackend`（或 `CacheManager(backend=NullCacheBackend())`），不连真实 Redis；Redis 连通性属于集成测试范围。

‎docs/internals/es_index_storage.md‎

Lines changed: 157 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,157 @@
1+
# Elasticsearch 索引与 BM25 检索
2+
3+
本文说明 `src/core/es_index_storage/`。该模块承担**两个职责**：
4+
5+
1. **写入侧（入库）**：把预分词后的 chunk token 文档批量写进 Elasticsearch。
6+
2. **召回侧（检索）**：对预分词字段做 BM25 检索，返回 topK chunk。
7+
8+
它消费上游 [preprocessor](preprocessor.md) 产出的 `FilePostIndexPlan`（预分词计划），召回侧通过适配器接入 [recall_pipeline.md](recall_pipeline.md)。ES 索引结构权威说明见 [schemas/elasticsearch.md](../api/schemas/elasticsearch.md)。
9+
10+
---
11+
12+
## 1. 包结构
13+
14+
```text
15+
src/core/es_index_storage/
16+
├── __init__.py # 公共入口：入库 Pipeline + 召回 Retriever + 模型 + 异常
17+
├── client.py # 进程级 AsyncElasticsearch 客户端单例
18+
├── mapping.py # ES index settings + mappings（analyzer / 字段）
19+
├── document_factory.py # chunk token plan → ES bulk action（瘦文档）
20+
├── batcher.py # 按字节/条数把 bulk action 分批
21+
├── pipeline.py # EsIndexingPipeline：文件级入库阶段
22+
├── models.py # 入库结果模型（EsIndexingResult / BulkBatchResult）
23+
├── retrieval.py # EsBm25Retriever：BM25 topK 检索
24+
├── retrieval_models.py # Bm25RecallRequest / Bm25ChunkHit
25+
├── bm25_retriever.py # 召回 Pipeline 适配器（Bm25Retriever）
26+
├── exceptions.py # ES 入库/召回异常族
27+
└── smoke.py # 集成测试用最小 keyword query 冒烟工具
28+
```
29+
30+
---
31+
32+
## 2. ES 索引结构
33+
34+
`mapping.py::build_es_index_body(shards, replicas)` 定义索引。要点：
35+
36+
- **瘦文档**：`_source` 排除 `coarse_tokens` / `fine_tokens`，token 只进倒排索引不回存正文，控制存储与 `_source` 体积。
37+
- **routing 必填**：`routing.required = True`，写入与检索都按 `dataset_id` 路由，保证同数据集 chunk 落同分片。
38+
- **字段**：定位字段 `chunk_id`（keyword）、`user_id` / `dataset_id` / `doc_id`（long）、`task_id`（keyword）、`chunk_index`（integer）；检索字段 `coarse_tokens` / `fine_tokens`（text）。
39+
- **分词器**：token 已由上游 RAGFlow 预分词为空格分隔词串，ES 侧 `chunk_index_analyzer` / `chunk_search_analyzer` 都用 `whitespace` tokenizer + `lowercase` filter，不在 ES 内二次分词——索引侧和召回侧用同一份分词产物，避免 token 分布漂移。
40+
41+
---
42+
43+
## 3. 写入路径
44+
45+
```text
46+
FilePostIndexPlan（来自 preprocessor）
47+
→ EsIndexingPipeline.run(...)
48+
→ EsDocumentFactory.build_action() # 每个 chunk 转 bulk action，校验大小
49+
→ TokenBatcher # 按字节/条数分批
50+
→ client.bulk(...) # 逐批写 ES
51+
→ 汇总为 EsIndexingResult
52+
```
53+
54+
- **`EsDocumentFactory`（document_factory.py）**：把 `ChunkWithTokens` + `FileIndexMeta` 转成 `EsBulkAction`（operation + document + estimated_bytes），超过 `ES_MAX_DOCUMENT_BYTES` 的文档抛 `EsDocumentValidationError`。
55+
- **`TokenBatcher`（batcher.py）**：按 `ES_MAX_TOKEN_BATCH_BYTES` / `ES_MAX_TOKEN_BATCH_CHUNKS` 把 action 切成多个 `TokenBatch`，校验失败的 chunk 收集进 `failed_errors`。
56+
- **`EsIndexingPipeline`（pipeline.py）**：编排上述步骤，依赖 `ChunkRepository` 推进 chunk 的 ES 索引状态，ES 服务级失败抛 `EsBulkError`。
57+
58+
结果模型在 `models.py`：`EsIndexingResult`（`total_items` / `indexed_items` / `failed_item_ids` / `succeeded_item_ids` / `skipped_item_ids`，`is_success` 判定全成功）、`BulkBatchResult`（单次 bulk 的成功/失败明细）。
59+
60+
---
61+
62+
## 4. 召回路径
63+
64+
### 4.1 `EsBm25Retriever`（retrieval.py）
65+
66+
底层 BM25 检索器，输入 `Bm25RecallRequest`，输出 `list[Bm25ChunkHit]`：
67+
68+
```python
69+
@dataclass(frozen=True)
70+
class Bm25RecallRequest:
71+
user_id: int
72+
dataset_id: int
73+
tokens: Sequence[str] # 已分词的 query token
74+
top_k: int
75+
doc_id: int | None = None
76+
77+
@dataclass(frozen=True)
78+
class Bm25ChunkHit:
79+
chunk_id: str
80+
doc_id: int # 同步返回，省去召回后回查 MySQL
81+
score: float # ES 原始 BM25 分
82+
```
83+
84+
查询构造（`_build_query`）：
85+
86+
- **filter**（不打分，做范围裁剪）：`user_id` term + `dataset_id` term，可选 `doc_id` term。
87+
- **must**（打分）：`multi_match` 在 `coarse_tokens^2`（权重 2）和 `fine_tokens` 上做 `best_fields`，query 为 token 空格拼接。
88+
- 检索按 `dataset_id` 路由（`routing=str(dataset_id)`），`_source` 只取 `chunk_id` / `doc_id`，`size=top_k`。
89+
- 空 token 直接返空；ES 调用异常包成 `EsRetrievalError`，非法请求抛 `EsRecallValidationError`。
90+
91+
### 4.2 `Bm25Retriever`（bm25_retriever.py）—— 召回 Pipeline 适配器
92+
93+
实现 `Retriever` 协议（见 [recall_pipeline.md §4](recall_pipeline.md#4-retriever-协议)），`source = "bm25"`。只做形状翻译：
94+
95+
```text
96+
Retriever.recall(query, dataset_ids, doc_ids, *, user_id, top_k)
97+
↓ tokenizer.tokenize(query) 取 coarse_tokens
98+
↓ 对每个 dataset_id（×doc_id）构造 Bm25RecallRequest
99+
EsBm25Retriever.recall_topk_chunks(request)
100+
↓ 合并、按 ES 原始分降序、截断 top_k
101+
list[RetrieverHit]
102+
```
103+
104+
- **分词器复用写入侧**：`tokenizer` 装配期注入，生产上用 `preprocessor.RagFlowTokenizer`，召回只取 `coarse_tokens` 切回 list——和入库用同一份分词器，避免召回/索引 token 分布漂移。
105+
- `user_id` / `top_k` 由 pipeline 执行期透传并校验为正。
106+
- `dataset_ids` 为空 → 返空（BM25 依赖 dataset routing，放弃"全库"语义）；多 dataset/doc 做笛卡儿积逐次下发，合并截断。
107+
108+
---
109+
110+
## 5. ES 客户端
111+
112+
`client.py` 维护**进程级单例** `AsyncElasticsearch`：`get_async_es_client(settings)` 懒初始化（`asyncio.Lock` 保护），`close_async_es_client()` 在应用关闭时释放。连接参数取自 settings：`ES_HOST`、`ES_BULK_REQUEST_TIMEOUT_SECONDS`（request_timeout），`ES_USER` + `ES_PASSWORD` 都存在时启用 basic auth。
113+
114+
---
115+
116+
## 6. 配置项
117+
118+
| 配置 | 默认 | 说明 |
119+
| --- | --- | --- |
120+
| `ES_HOST` | `http://localhost:9200` | ES 地址 |
121+
| `ES_USER` / `ES_PASSWORD` | — | basic auth（两者都设才启用） |
122+
| `ES_INDEX_NAME` | `tolink_rag_index` | 索引名 |
123+
| `ES_INDEX_SHARDS` / `ES_INDEX_REPLICAS` | `3` / `1` | 分片与副本 |
124+
| `ES_MAX_DOCUMENT_BYTES` | `131072` | 单文档字节上限 |
125+
| `ES_MAX_TOKEN_BATCH_BYTES` / `ES_MAX_TOKEN_BATCH_CHUNKS` | `5242880` / `500` | 单次 bulk 字节/条数上限 |
126+
| `ES_BULK_REQUEST_TIMEOUT_SECONDS` | `30` | bulk/search 请求超时 |
127+
| `ES_SMOKE_ENABLED` | `False` | 是否启用 smoke 冒烟 |
128+
129+
配置详解见 [ops/configure.md](../ops/configure.md)。
130+
131+
---
132+
133+
## 7. 异常族
134+
135+
| 异常 | 触发 |
136+
| --- | --- |
137+
| `EsIndexingError` | 入库异常基类 |
138+
| `EsDocumentValidationError` | chunk 无法转成合法 ES 文档（如超大） |
139+
| `EsBulkError` | ES 服务级操作失败 |
140+
| `EsRecallValidationError`（继承 `ValueError`） | 召回请求非法 |
141+
| `EsRetrievalError` | 召回检索失败 |
142+
143+
---
144+
145+
## 8. 公共入口
146+
147+
`__init__.py` 导出：`EsIndexingPipeline`、`EsIndexingResult`、`EsBm25Retriever`、`Bm25Retriever`、`Bm25RecallRequest`、`Bm25ChunkHit`，以及全部异常。`smoke.py::run_es_index_smoke()` 仅供集成测试做最小 keyword query 验证。
148+
149+
---
150+
151+
## 9. 测试约定
152+
153+
| 测试目标 | 入口 |
154+
| --- | --- |
155+
| 入库阶段、分批、文档工厂 | `tests/unit/core/es_index_storage/` |
156+
| BM25 召回适配器 | `tests/unit/core/es_index_storage/test_bm25_retriever.py` |
157+
| 真实 ES（需开关） | `ES_SMOKE_ENABLED=True` + 集成测试 |

‎docs/internals/preprocessor.md‎

Lines changed: 81 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,81 @@
1+
# 预分词模块（Preprocessor）
2+
3+
本文说明 `src/core/preprocessor/`。它是 ES/BM25 链路的**上游**：把已落库的 chunk 正文用 RAGFlow 分词器预先切成 token，产出文件级"ES 后置索引计划"（`FilePostIndexPlan`），交给 [es_index_storage](es_index_storage.md) 写入 Elasticsearch。
4+
5+
"预分词"指**写入侧提前完成分词**：ES 索引文档里直接存空格分隔的 token 串，ES 端只做 `whitespace` 切分不再二次分词，从而索引侧与 BM25 召回侧共用同一份分词产物，避免 token 分布漂移（召回侧分词见 [es_index_storage.md §4.2](es_index_storage.md#42-bm25retrieverbm25_retrieverpy--召回-pipeline-适配器)）。
6+
7+
---
8+
9+
## 1. 包结构
10+
11+
```text
12+
src/core/preprocessor/
13+
├── __init__.py # 包说明（不导出符号）
14+
├── models.py # 预分词产物契约：FileIndexMeta / ChunkWithTokens / FilePostIndexPlan
15+
├── ragflow_tokenizer.py # RagFlowTokenizer：RAGFlow 分词器适配
16+
└── service.py # Preprocessor：从库里读 chunk 构建 FilePostIndexPlan
17+
```
18+
19+
---
20+
21+
## 2. 产物模型（models.py）
22+
23+
这些 dataclass 是预分词与 ES 入库之间的**共享契约**（`frozen` + `slots`）：
24+
25+
| 模型 | 字段 | 说明 |
26+
| --- | --- | --- |
27+
| `FileIndexMeta` | `user_id` / `dataset_id` / `doc_id` / `task_id` | 文件级归属元信息 |
28+
| `ChunkWithTokens` | `chunk_id` / `chunk_index` / `coarse_tokens` / `fine_tokens` | 单 chunk 的两级 token 串 |
29+
| `FilePostIndexPlan` | `file_meta` + `chunks_with_tokens: list[ChunkWithTokens]` | 一个文件的完整 ES 后置索引计划 |
30+
31+
`coarse_tokens`（粗粒度）与 `fine_tokens`（细粒度）对应 ES mapping 里的两个检索字段，召回时 `coarse_tokens` 权重更高（见 ES BM25 查询构造）。
32+
33+
---
34+
35+
## 3. 分词器（ragflow_tokenizer.py）
36+
37+
`RagFlowTokenizer` 是对 RAGFlow 分词实现的薄封装：
38+
39+
- 依赖 `infinity.rag_tokenizer.RagTokenizer`（来自 `infinity-sdk`）。依赖缺失时构造抛 `RuntimeError`，提示安装或注入替身。
40+
- `tokenize(text) -> TokenizedText`：先用 `TABLE_TAG_RE` 把 `<table>/<td>/<tr>/<th>/<caption>` 等表格标签替换为空格，再 `tokenize` 得 `coarse_tokens`、`fine_grained_tokenize` 得 `fine_tokens`，二者均为空格分隔词串。
41+
- `TokenizedText` 是 `(coarse_tokens, fine_tokens)` 的轻量载体。
42+
43+
> RAGFlow/infinity 分词器需要本地 NLTK 数据，路径引导见 [src/nltk_bootstrap.py](../../src/nltk_bootstrap.py)。
44+
45+
---
46+
47+
## 4. 服务（service.py）
48+
49+
`Preprocessor` 从 MySQL 读 chunk 并构建计划，核心方法 `build_file_post_index_plan(doc_id, task_id)`：
50+
51+
1. 通过 `_list_chunks_for_pretokenization` 查该文档**全部有效 chunk**（Issue #57：ES 文档级全量重建，不再按 `es_status` 过滤）。筛选条件：`doc_id` 匹配、`dense_vector_status = INDEXED`（dense 已就绪是前置依赖）、`lifecycle_status = ACTIVE`，按 `chunk_index` 升序。
52+
2. 无记录 → 返回空计划（`file_meta` 用占位 0 值，`chunks_with_tokens=[]`）。
53+
3. 对每条记录调 `_tokenize_record`：校验 `chunk_index` 合法、分词、`strip` 后校验 `coarse_tokens` / `fine_tokens` 非空，产出 `ChunkWithTokens`。
54+
4. 文件级 all-or-nothing：任一 chunk 预分词失败 → 整体抛 `PreprocessorError`，不写任何 chunk 的 es_status；失败终态由上游解析阶段落地。
55+
5. `file_meta` 的 `user_id` / `dataset_id`（取自 `set_id`）/ `doc_id` 从首条记录取。
56+
57+
依赖通过构造注入，便于测试：`session_factory`（默认 `get_async_session_factory()`）、`tokenizer` 或 `tokenizer_factory`（默认 `RagFlowTokenizer`，懒加载）。`ChunkTokenizer` Protocol 定义了 `tokenize` 最小契约。
58+
59+
---
60+
61+
## 5. 与相邻模块的关系
62+
63+
```text
64+
chunk_fact_storage (ChunkRecordDB) ← 数据来源（dense 已 INDEXED 的 active chunk）
65+
│
66+
▼
67+
preprocessor.Preprocessor ← 本模块：读 chunk → 预分词
68+
│ FilePostIndexPlan
69+
▼
70+
es_index_storage.EsIndexingPipeline ← 下游：批量写 ES
71+
⋮
72+
es_index_storage.Bm25Retriever ← 召回侧复用 RagFlowTokenizer 分词 query
73+
```
74+
75+
在解析主流水线里，预分词对应 parse_task 的 `pretokenize` 阶段，紧接其后是 `es_indexing` 阶段（见 [parse_task_pipeline.md](parse_task_pipeline.md)）。
76+
77+
---
78+
79+
## 6. 测试约定
80+
81+
`Preprocessor` 用注入的 fake tokenizer / session 测试，不依赖真实 infinity-sdk；`RagFlowTokenizer` 的真实分词行为属于集成测试范围。

0 commit comments

Comments
 (0)