面向MCP服务器1C文档搜索的多层级RAG架构
MCP服务器documents1c和metadata1c实现了一套多层级RAG系统，用于精准搜索1C文档和配置。每个排序层级都针对特定挑战：从块上下文到向量搜索与全文搜索的结合。系统采用PostgreSQL搭配pgvector、Redis用于缓存，以及针对1C术语的自定义分词。
为何放弃带层级结构的向量数据库
像Qdrant或Weaviate这样的标准向量存储难以应对1C文档的动态特性。文档脱离上下文就无法自足：一篇标题为“设置”的文章需要指定所属章节，而像Reference.Employees这样的对象则需要其属性和模块的描述。
文档更新频繁——新的BSP版本、平台方法。在向量数据库中重建树结构成本高昂。解决方案是：PostgreSQL搭配pgvector，并将heading_path嵌入每个块中。嵌入向量基于heading_path + text生成，仅存储块文本。
节省之处： 在article_chunks表中使用embedding halfvec(1024)列，而非单独的向量数据库。
层级1：带标题上下文的块
将Markdown分割成块时考虑了结构：移除YAML，按标题划分，不拆分代码块，重叠220个字符。
def split_markdown_into_chunks(
text: str,
max_len: int = 800,
overlap: int = 220,
include_heading_path: str = "full_path",
) -> list[str]:
body = _strip_frontmatter(text)
sections = _split_by_headers(body)
chunks: list[str] = []
for section, heading_path, current_heading in sections:
context_heading = " > ".join(heading_path)
section_with_context = f"{context_heading}\n\n{section}"
if len(section_with_context) <= max_len:
chunks.append(section_with_context)
else:
sub_chunks = _split_section_by_paragraphs(section_with_context, max_len)
chunks.extend(sub_chunks)
return _add_overlap(chunks, overlap)示例块：
薪资与人力资源管理 > 薪资计算 > 设置个人所得税
要正确计算个人所得税，必须指定税率...在数据库中，meta单独存储heading_path用于过滤和链接。嵌入向量基于：heading_path + chunk_text。
层级2：Redis中的嵌入向量缓存
生成嵌入向量（text-embedding-qwen3-embedding-4b）需要50–200毫秒。对于重复查询——使用带TTL的Redis，键为model:dimensions:text的SHA256哈希值。
async def embed_text(text: str) -> list[float]:
model = _get_embed_model_name()
dims = _get_embed_dimensions()
if _cache_enabled():
cached = await cache_get(text, model, dims)
if cached is not None:
return cached
response = await to_thread(client.embeddings.create, model=model, input=text)
vector = response.data[0].embedding
if _cache_enabled():
await cache_set(text, model, dims, vector, _cache_ttl())
return vector模型变更时缓存自动失效。
层级3：使用halfvec的向量搜索
通过pgvector中的<=>运算符计算余弦距离，halfvec(1024)类型节省内存。
SELECT
ac.id, ac.article_id, ac.chunk_text, ac.meta,
ac.embedding::halfvec(1024) <=> CAST(:query_embedding AS halfvec(1024)) AS distance
FROM article_chunks ac
INNER JOIN articles a ON ac.article_id = a.id
WHERE a.job_id IN (SELECT id FROM jobs WHERE load_mode IN ('hbk', 'docs'))
AND a.destination = :destination
ORDER BY distance
LIMIT :limit;得分 = 1 - 距离。
层级4：使用RRF合并排序结果
向量搜索对于精确术语如&OnClientOnServer效果不佳。RRF融合向量和全文搜索列表：
@staticmethod
def _rrf_merge(vector_results, fts_results, top_k: int, k: int = 60) -> list[dict]:
scores: dict[int, float] = {}
items: dict[int, dict] = {}
for rank, item in enumerate(vector_results, start=1):
chunk_id = item["id"]
scores[chunk_id] = scores.get(chunk_id, 0.0) + 1.0 / (k + rank)
items[chunk_id] = item
for rank, item in enumerate(fts_results, start=1):
chunk_id = item["id"]
scores[chunk_id] = scores.get(chunk_id, 0.0) + 1.0 / (k + rank)
if chunk_id not in items:
items[chunk_id] = item
sorted_ids = sorted(scores, key=lambda cid: scores[cid], reverse=True)
# 归一化并返回top_kHBK提升： 对于没有过滤器的查询，1C内置语言块（load_mode='hbk'）会被添加到列表开头。
- 按query_emb排序的顶部HBK块
- 排除重复项
- 置于主要结果之前
层级5：使用1C分词的BM25重排序
对于来自RRF的50–100个候选结果，应用带权重（向量0.6，bm25 0.4）的BM25。对以下情况有效：
- 特定术语：
CommonModule、InformationRegister、&OnServer - 对象名称：
Reference.Nomenclature - 简短技术查询
自定义分词：
GUID_RE = re.compile(r"\b[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\b", re.I)
VERSION_RE = re.compile(r"\bv?8\.\d+(?:\.\d+){1,3}\b", re.I)
DIRECTIVE_RE = re.compile(r"[&#][And-IA-Z][And-YaYoA-Za-z0-9_]*", re.I)
def _tokenize(self, text: str, expand_query: bool = False) -> list[str]:
text = _norm_text(text)
out = []
for g in GUID_RE.findall(text): out.append(g.lower())
for v in VERSION_RE.findall(text): out.append(v.lower())
for d in DIRECTIVE_RE.findall(text): out.append(d.lower())
for t in TOKEN_RE.findall(text):
if t in RU_STOP: continue
for p in _split_identifier(t):
out.append(p.lower())
return out拆分CamelCase、snake_case，提取指令和GUID。
核心要点
- 块中的heading_path 无需文档树即可提供上下文
- Redis缓存 将嵌入向量延迟从100+毫秒降至0毫秒
- RRF 无需归一化即可结合语义和精确搜索
- 带1C分词的BM25 提升相关技术术语的排名
- halfvec(1024) 在pgvector中节省内存，同时保持余弦精度
— Editorial Team
暂无评论。