274 lines
11 KiB
Python
274 lines
11 KiB
Python
|
# -*- coding: utf-8 -*-
|
|||
|
|
"""RAG 知识库模板化 —— 换行业只改模板配置，内核零改动（PRD 5.4 / Issue #46）。
|
|||
|
|
|
|||
|
|
领域 RAG 知识库按**模板配置**（对齐 data-bus 的模板化思想，见 PRD 5.4
|
|||
|
|
「知识库范围（由模板RAG库提供）」与 7.3「领域RAG知识库（一期）」）：
|
|||
|
|
|
|||
|
|
- 知识源分类固定为三类，模板配置声明每类收录的文档：
|
|||
|
|
- `process` 工艺规范（沸腾氯化工艺规范等）；
|
|||
|
|
- `sop` 标准作业程序 / 操作手册 / 异常处置 SOP；
|
|||
|
|
- `standard` 国标 / 行业标准 / 文献（氯、钛相关国标等）。
|
|||
|
|
- 全部外部资源命名由模板名推导（向量库 collection / 索引 / namespace），
|
|||
|
|
命名清洗规则对齐 data-bus（小写 + 非安全字符替换为 `_`），但**保留中文**
|
|||
|
|
（文档对象键 / doc_id 面向中文标题，MinIO 对象键支持 UTF-8）。
|
|||
|
|
|
|||
|
|
配置即「模板 RAG 库」资产（`config/kb.template.yaml`），运行时绑定模板
|
|||
|
|
版本，保证检索结果可复现（对齐 PRD 5.4 Prompt/知识库版本化）。
|
|||
|
|
"""
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import re
|
|||
|
|
from enum import Enum
|
|||
|
|
from typing import Dict, List, Tuple
|
|||
|
|
|
|||
|
|
# 保留字符集：小写字母、数字、`-` `_` `.` 与中文（对齐 data-bus 清洗规则，
|
|||
|
|
# 但保留中文以便文档对象键 / doc_id 直接使用中文标题）
|
|||
|
|
_KEEP = re.compile(r"[^a-z0-9_.\u4e00-\u9fff-]+")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def sanitize(name: str) -> str:
|
|||
|
|
"""通用命名清洗：小写 + 非安全字符（空格/标点等）替换为 `_`。
|
|||
|
|
|
|||
|
|
RAG 命名保留中文（文档标题即对象键），仅清洗 ASCII 不安全字符；
|
|||
|
|
与 data-bus 的 sanitize 不同点即在此。
|
|||
|
|
"""
|
|||
|
|
s = (name or "").strip().lower()
|
|||
|
|
s = _KEEP.sub("_", s)
|
|||
|
|
return s.strip("._") or "kb"
|
|||
|
|
|
|||
|
|
|
|||
|
|
class KnowledgeSourceKind(str, Enum):
|
|||
|
|
"""知识源分类（模板 RAG 库的三类知识范围，PRD 7.3）。"""
|
|||
|
|
|
|||
|
|
PROCESS = "process" # 工艺规范
|
|||
|
|
SOP = "sop" # 标准作业程序 / 操作手册 / 异常处置 SOP
|
|||
|
|
STANDARD = "standard" # 国标 / 行业标准 / 文献
|
|||
|
|
|
|||
|
|
@property
|
|||
|
|
def label(self) -> str:
|
|||
|
|
return {
|
|||
|
|
KnowledgeSourceKind.PROCESS: "工艺规范",
|
|||
|
|
KnowledgeSourceKind.SOP: "SOP/操作手册",
|
|||
|
|
KnowledgeSourceKind.STANDARD: "国标/标准",
|
|||
|
|
}[self]
|
|||
|
|
|
|||
|
|
|
|||
|
|
# 知识源分类注册表：模板配置中类别名 → 枚举（未知类别名直接拒绝，避免拼写漂移）
|
|||
|
|
SOURCE_KINDS: Dict[str, KnowledgeSourceKind] = {k.value: k for k in KnowledgeSourceKind}
|
|||
|
|
|
|||
|
|
|
|||
|
|
class KbTemplateNaming:
|
|||
|
|
"""按模板推导 RAG 知识库全部外部资源命名（向量库 / 索引 / namespace）。
|
|||
|
|
|
|||
|
|
命名对齐 data-bus（`tpl_{tpl}` schema 风格）：collection 名用下划线形态，
|
|||
|
|
namespace 用连字符形态（对象存储桶风格）。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
def __init__(self, template: str, index_suffix: str = "kb"):
|
|||
|
|
self.template = sanitize(template)
|
|||
|
|
# SQL/向量库标识符用下划线形态（避免 `-` 需引号转义）
|
|||
|
|
self.tpl_sql = self.template.replace("-", "_")
|
|||
|
|
self.index_suffix = sanitize(index_suffix)
|
|||
|
|
|
|||
|
|
def collection(self) -> str:
|
|||
|
|
"""向量库 collection：`{tpl}_kb`（对齐 `tpl_{tpl}` schema 命名）。"""
|
|||
|
|
return f"{self.tpl_sql}_{self.index_suffix}"
|
|||
|
|
|
|||
|
|
def index_name(self) -> str:
|
|||
|
|
"""检索索引名：`{tpl}_kb_idx`。"""
|
|||
|
|
return f"{self.collection()}_idx"
|
|||
|
|
|
|||
|
|
def namespace(self) -> str:
|
|||
|
|
"""对象存储 namespace：`tpl-{template}-kb`（S3 桶风格，允许 `-`）。"""
|
|||
|
|
return f"tpl-{self.template}-{self.index_suffix}"
|
|||
|
|
|
|||
|
|
def doc_object_key(self, doc_id: str, ext: str = "md") -> str:
|
|||
|
|
"""源文档对象键：`kb/{doc_id}.{ext}`（保留中文标题）。"""
|
|||
|
|
return f"kb/{sanitize(doc_id)}.{sanitize(ext) or 'md'}"
|
|||
|
|
|
|||
|
|
|
|||
|
|
class KbSourceConfig:
|
|||
|
|
"""单个知识源的配置（模板 RAG 库声明某类知识源收录哪些文档）。"""
|
|||
|
|
|
|||
|
|
def __init__(self, kind: KnowledgeSourceKind, documents: List[str]):
|
|||
|
|
self.kind = kind
|
|||
|
|
self.documents = list(documents)
|
|||
|
|
|
|||
|
|
def __repr__(self) -> str: # pragma: no cover - 调试辅助
|
|||
|
|
return f"KbSourceConfig({self.kind.value}, docs={len(self.documents)})"
|
|||
|
|
|
|||
|
|
|
|||
|
|
class KbTemplateConfig:
|
|||
|
|
"""模板 RAG 库配置：模板名 + 三类知识源文档清单。"""
|
|||
|
|
|
|||
|
|
def __init__(self, template: str, sources: List[KbSourceConfig], version: str = "1.0.0"):
|
|||
|
|
self.template = template
|
|||
|
|
self.sources = sources
|
|||
|
|
self.version = version
|
|||
|
|
|
|||
|
|
def documents_for(self, kind: KnowledgeSourceKind) -> List[str]:
|
|||
|
|
"""返回某类知识源收录的文档标题列表。"""
|
|||
|
|
for src in self.sources:
|
|||
|
|
if src.kind is kind:
|
|||
|
|
return list(src.documents)
|
|||
|
|
return []
|
|||
|
|
|
|||
|
|
def all_documents(self) -> List[str]:
|
|||
|
|
"""全部知识源文档标题（去重保序）。"""
|
|||
|
|
seen, out = set(), []
|
|||
|
|
for src in self.sources:
|
|||
|
|
for d in src.documents:
|
|||
|
|
if d not in seen:
|
|||
|
|
seen.add(d)
|
|||
|
|
out.append(d)
|
|||
|
|
return out
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
# 轻量 YAML 子集解析（零第三方依赖，递归下降）：map / list / scalar / 注释。
|
|||
|
|
# 足以解析 `config/kb.template.yaml` 这类模板资产；运行时零依赖可复现。
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
|
|||
|
|
def _parse_scalar(text: str) -> str:
|
|||
|
|
"""去掉标量两侧引号与行内注释（`key: value # comment`）。"""
|
|||
|
|
t = text.split(" #", 1)[0].strip()
|
|||
|
|
if len(t) >= 2 and t[0] == t[-1] and t[0] in ("'", '"'):
|
|||
|
|
return t[1:-1]
|
|||
|
|
return t
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _strip_comments(lines: List[str]) -> List[Tuple[str, int]]:
|
|||
|
|
"""剔除空行与整行注释，保留行号（1 起）用于报错定位。"""
|
|||
|
|
out = []
|
|||
|
|
for i, ln in enumerate(lines):
|
|||
|
|
s = ln.strip()
|
|||
|
|
if not s or s.startswith("#"):
|
|||
|
|
continue
|
|||
|
|
out.append((ln, i + 1))
|
|||
|
|
return out
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _parse_node(lines: List[Tuple[str, int]], i: int, indent: int):
|
|||
|
|
"""递归解析从 lines[i] 开始、缩进为 `indent` 的一个节点。
|
|||
|
|
|
|||
|
|
返回 `(value, next_i)`：value 为 dict / list / str，next_i 为下一个
|
|||
|
|
未消费行的下标。
|
|||
|
|
"""
|
|||
|
|
text, no = lines[i]
|
|||
|
|
lead = len(text) - len(text.lstrip(" "))
|
|||
|
|
|
|||
|
|
# ---- list 节点：`- item` 或 `- key: val`（map 项） ----
|
|||
|
|
if text.lstrip(" ").startswith("- "):
|
|||
|
|
items: List[object] = []
|
|||
|
|
while i < len(lines):
|
|||
|
|
t, no2 = lines[i]
|
|||
|
|
stripped = t.lstrip(" ")
|
|||
|
|
if not stripped.startswith("- "):
|
|||
|
|
break
|
|||
|
|
lead_j = len(t) - len(t.lstrip(" "))
|
|||
|
|
if lead_j != indent:
|
|||
|
|
break # 缩进不同的 `-` 项属于外层块，交还上层处理
|
|||
|
|
item_text = stripped[2:].strip()
|
|||
|
|
if not item_text:
|
|||
|
|
raise ValueError(f"kb.yaml 第 {no2} 行：list 项为空")
|
|||
|
|
if ":" in item_text:
|
|||
|
|
# map 项：把 `- key: val` 规范化为缩进 `map_indent` 的 map 首行
|
|||
|
|
map_indent = len(t) - len(t.lstrip(" ")) + 2
|
|||
|
|
lines[i] = (" " * map_indent + item_text, no2)
|
|||
|
|
v, i = _parse_node(lines, i, map_indent)
|
|||
|
|
items.append(v)
|
|||
|
|
else:
|
|||
|
|
items.append(_parse_scalar(item_text))
|
|||
|
|
i += 1
|
|||
|
|
return items, i
|
|||
|
|
|
|||
|
|
# ---- map 节点：`key: value` / `key:`（嵌套值） ----
|
|||
|
|
result: Dict[str, object] = {}
|
|||
|
|
while i < len(lines):
|
|||
|
|
t, no = lines[i]
|
|||
|
|
lead_j = len(t) - len(t.lstrip(" "))
|
|||
|
|
if lead_j < indent or t.lstrip(" ").startswith("- "):
|
|||
|
|
break
|
|||
|
|
if lead_j > indent:
|
|||
|
|
raise ValueError(f"kb.yaml 第 {no} 行缩进异常（期望 {indent}，实际 {lead_j}）")
|
|||
|
|
if ":" not in t:
|
|||
|
|
raise ValueError(f"kb.yaml 第 {no} 行不是合法键值对：{t!r}")
|
|||
|
|
key, _, rest = t.partition(":")
|
|||
|
|
key = key.strip()
|
|||
|
|
rest = rest.strip()
|
|||
|
|
if rest:
|
|||
|
|
result[key] = _parse_scalar(rest)
|
|||
|
|
i += 1
|
|||
|
|
continue
|
|||
|
|
# `key:` —— 值在后续行嵌套（list 或 map），缩进必须更深
|
|||
|
|
if i + 1 >= len(lines):
|
|||
|
|
raise ValueError(f"kb.yaml 第 {no} 行 {key!r} 缺少值")
|
|||
|
|
sub_indent = len(lines[i + 1][0]) - len(lines[i + 1][0].lstrip(" "))
|
|||
|
|
if sub_indent <= indent:
|
|||
|
|
raise ValueError(f"kb.yaml 第 {no} 行 {key!r} 缺少值（无嵌套内容）")
|
|||
|
|
v, i = _parse_node(lines, i + 1, sub_indent)
|
|||
|
|
result[key] = v
|
|||
|
|
return result, i
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _load_yaml_text(text: str) -> Dict[str, object]:
|
|||
|
|
"""解析 YAML 子集 → 嵌套 dict/list。顶层必须为 map。"""
|
|||
|
|
lines = _strip_comments(text.splitlines())
|
|||
|
|
if not lines:
|
|||
|
|
return {}
|
|||
|
|
top_indent = len(lines[0][0]) - len(lines[0][0].lstrip(" "))
|
|||
|
|
value, next_i = _parse_node(lines, 0, top_indent)
|
|||
|
|
if not isinstance(value, dict):
|
|||
|
|
raise ValueError("kb.yaml 顶层必须是 map")
|
|||
|
|
if next_i < len(lines):
|
|||
|
|
raise ValueError(
|
|||
|
|
f"kb.yaml 第 {lines[next_i][1]} 行：顶层存在多个节点（缩进不一致）"
|
|||
|
|
)
|
|||
|
|
return value
|
|||
|
|
|
|||
|
|
|
|||
|
|
def load_kb_config(path: str) -> KbTemplateConfig:
|
|||
|
|
"""从模板 RAG 库 YAML 资产加载配置（零第三方依赖）。
|
|||
|
|
|
|||
|
|
期望结构（详见 `config/kb.template.yaml`）：
|
|||
|
|
```yaml
|
|||
|
|
template: ti-cl4
|
|||
|
|
version: 1.0.0
|
|||
|
|
sources:
|
|||
|
|
- kind: process
|
|||
|
|
documents: [...]
|
|||
|
|
- kind: sop
|
|||
|
|
documents: [...]
|
|||
|
|
- kind: standard
|
|||
|
|
documents: [...]
|
|||
|
|
```
|
|||
|
|
"""
|
|||
|
|
with open(path, "r", encoding="utf-8") as fh:
|
|||
|
|
data = _load_yaml_text(fh.read())
|
|||
|
|
|
|||
|
|
template = str(data.get("template", "")).strip()
|
|||
|
|
if not template:
|
|||
|
|
raise ValueError("kb.yaml 缺少 template 字段")
|
|||
|
|
version = str(data.get("version", "1.0.0")).strip() or "1.0.0"
|
|||
|
|
|
|||
|
|
sources: List[KbSourceConfig] = []
|
|||
|
|
raw_sources = data.get("sources") or []
|
|||
|
|
if not isinstance(raw_sources, list):
|
|||
|
|
raise ValueError("kb.yaml sources 必须是 list")
|
|||
|
|
for item in raw_sources:
|
|||
|
|
if not isinstance(item, dict):
|
|||
|
|
raise ValueError(f"kb.yaml sources 项必须是 map，实际 {item!r}")
|
|||
|
|
kind_name = str(item.get("kind", "")).strip()
|
|||
|
|
if kind_name not in SOURCE_KINDS:
|
|||
|
|
raise ValueError(
|
|||
|
|
f"kb.yaml 未知知识源类别 {kind_name!r}（应为 {sorted(SOURCE_KINDS)}）"
|
|||
|
|
)
|
|||
|
|
docs = item.get("documents") or []
|
|||
|
|
if not isinstance(docs, list):
|
|||
|
|
raise ValueError(f"kb.yaml {kind_name} 的 documents 必须是 list")
|
|||
|
|
sources.append(
|
|||
|
|
KbSourceConfig(SOURCE_KINDS[kind_name], [str(d) for d in docs])
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
return KbTemplateConfig(template=template, sources=sources, version=version)
|