# -*- 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)