151 lines
6.1 KiB
Python
151 lines
6.1 KiB
Python
"""解析已审核的 QA 源文件（`客服Agent知识库_QA问答对_v5_RAG发布候选版.txt`）。
|
||
|
||
为什么需要它：这份 105 条的文本是**业务侧唯一的人工审核产物**，后续入库的知识、
|
||
同义问法、以及 Milvus 向量全部由它派生。手工抄写 105 条 × (1 问题 + 若干相似问法 + 1 回答)
|
||
既不可审计也不可重放，所以把"解析"单独做成一个无副作用的纯函数模块：解析能被单测固定住，
|
||
导入脚本只负责落库。
|
||
|
||
源文件的真实形状（实测，不是推测）：
|
||
|
||
- 每条固定三行：`[编号]` / `问题：` / `相似问法：`（`|` 分隔）/ `回答：`。
|
||
- 体量：**105 条**，其中 `RAG-*` **62** 条、`NF-*` **43** 条；相似问法合计 **421** 条。
|
||
- 编号前缀实测有 **12 种**：`RAG-PER`/`RAG-CHAT`/`RAG-PUB`/`RAG-RVW`/`RAG-HUM`/`RAG-CONFIG`/
|
||
`RAG-P1` 与 `NF-SVC`/`NF-TRD`/`NF-CMP`/`NF-STS`/`NF-ACC`。
|
||
计划里写的"11 种"漏了 `NF-ACC`（2 条：`NF-ACC-001`/`NF-ACC-003`，在"三、真实直销业务场景问答"节内）。
|
||
**按前缀白名单解析是错的方向**——只按 `RAG-` 匹配会静默丢掉 43 条 `NF-*`，
|
||
所以这里的编号正则刻意不枚举前缀，`[任意字母开头的连字符编号]` 一律接受，
|
||
由 `expected_count` 兜底：数量对不上就报错，而不是少导了还说自己成功。
|
||
|
||
`EXPECTED_RECORD_COUNT=105` 是**默认开启的断言**：调用方必须显式传
|
||
`expected_count=None` 才能跳过计数校验——默认安全，测试样例才需要绕过。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from dataclasses import dataclass
|
||
from hashlib import sha256
|
||
|
||
#: 已审核源文件的条目数（业务侧 v5.8 交付物）。验收补充条目另算，不在此数内。
|
||
EXPECTED_RECORD_COUNT = 105
|
||
|
||
#: 编号行：`[RAG-PER-001]` / `[NF-ACC-003]`……**不枚举前缀**，否则会漏记录。
|
||
_ID_PATTERN = re.compile(r"^\[([A-Za-z][A-Za-z0-9]*(?:-[A-Za-z0-9]+)+)\]$")
|
||
|
||
#: 三个字段的固定前缀（全角冒号）。
|
||
_QUESTION = "问题："
|
||
_SYNONYMS = "相似问法："
|
||
_ANSWER = "回答："
|
||
_FIELD_PREFIXES = (_QUESTION, _SYNONYMS, _ANSWER)
|
||
|
||
#: 相似问法分隔符。
|
||
_SYNONYM_SEPARATOR = "|"
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class QaRecord:
|
||
"""一条已审核 QA。`synonyms` 为不可变元组，且在**归一化后**无重复。"""
|
||
|
||
qa_id: str
|
||
question: str
|
||
synonyms: tuple[str, ...]
|
||
answer: str
|
||
|
||
|
||
def normalize_phrase(phrase: str) -> str:
|
||
"""归一化问法：折叠所有空白并转小写。
|
||
|
||
与 `docs/02` §7.3 L355 的口径一致——`phrase_hash` 算的是**完整**归一化文本的
|
||
SHA-256，去重必须在归一化之后做，否则「你 是 谁？」与「你是谁？」会被当成两条。
|
||
"""
|
||
return " ".join(phrase.split()).strip().lower()
|
||
|
||
|
||
def phrase_hash(phrase: str) -> str:
|
||
"""`agent_faq_synonym.phrase_hash`：归一化文本的 SHA-256（hex，64 字符）。"""
|
||
return sha256(normalize_phrase(phrase).encode("utf-8")).hexdigest()
|
||
|
||
|
||
def _blocks(text: str) -> list[list[str]]:
|
||
"""按编号行切块，每块只保留编号行与三字段行；章节标题、说明文字、分隔线一律丢弃。"""
|
||
blocks: list[list[str]] = []
|
||
current: list[str] | None = None
|
||
for raw in text.splitlines():
|
||
line = raw.strip()
|
||
if _ID_PATTERN.match(line):
|
||
if current is not None:
|
||
blocks.append(current)
|
||
current = [line]
|
||
continue
|
||
if current is None or not line:
|
||
continue
|
||
if line.startswith(_FIELD_PREFIXES):
|
||
current.append(line)
|
||
if current is not None:
|
||
blocks.append(current)
|
||
return blocks
|
||
|
||
|
||
def _parse_block(block: list[str], seen: set[str]) -> QaRecord:
|
||
match = _ID_PATTERN.match(block[0])
|
||
if match is None: # pragma: no cover - `_blocks` 只产出匹配行
|
||
raise ValueError(f"非法的编号行：{block[0]}")
|
||
qa_id = match.group(1)
|
||
if qa_id in seen:
|
||
# 重复编号意味着后一条会覆盖前一条，或同义词被挂到错的知识上，必须当场失败。
|
||
raise ValueError(f"重复的 qa_id：{qa_id}")
|
||
seen.add(qa_id)
|
||
|
||
fields: dict[str, str] = {}
|
||
for line in block[1:]:
|
||
for prefix in _FIELD_PREFIXES:
|
||
if line.startswith(prefix):
|
||
if prefix in fields:
|
||
raise ValueError(f"{qa_id} 的 {prefix.rstrip('：')} 出现了多次")
|
||
fields[prefix] = line[len(prefix):].strip()
|
||
break
|
||
|
||
question = fields.get(_QUESTION, "")
|
||
synonyms_raw = fields.get(_SYNONYMS, "")
|
||
answer = fields.get(_ANSWER, "")
|
||
if not question:
|
||
raise ValueError(f"{qa_id} 缺少问题")
|
||
if not synonyms_raw:
|
||
raise ValueError(f"{qa_id} 缺少相似问法")
|
||
if not answer:
|
||
raise ValueError(f"{qa_id} 缺少回答")
|
||
|
||
synonyms: list[str] = []
|
||
digests: set[str] = set()
|
||
for item in synonyms_raw.split(_SYNONYM_SEPARATOR):
|
||
phrase = item.strip()
|
||
if not phrase:
|
||
continue
|
||
digest = phrase_hash(phrase)
|
||
if digest in digests:
|
||
# `uk_faq_synonym (knowledge_id, phrase_hash)` 是硬约束：在解析层就去重，
|
||
# 免得到库上撞唯一键时报 MySQL 错误（那种报错看不到是哪条 QA 有问题）。
|
||
continue
|
||
digests.add(digest)
|
||
synonyms.append(phrase)
|
||
if not synonyms:
|
||
raise ValueError(f"{qa_id} 相似问法为空")
|
||
return QaRecord(qa_id, question, tuple(synonyms), answer)
|
||
|
||
|
||
def parse_qa_source(
|
||
text: str, *, expected_count: int | None = EXPECTED_RECORD_COUNT
|
||
) -> list[QaRecord]:
|
||
"""解析全文并返回按出现顺序排列的 `QaRecord` 列表。
|
||
|
||
`expected_count` 默认 `105`：数量不符直接抛 `ValueError`，让"少解析了 43 条
|
||
NF-*"这类错误在解析阶段就炸出来，而不是安静地导进去 62 条。
|
||
"""
|
||
records: list[QaRecord] = []
|
||
seen: set[str] = set()
|
||
for block in _blocks(text):
|
||
records.append(_parse_block(block, seen))
|
||
if expected_count is not None and len(records) != expected_count:
|
||
raise ValueError(f"记录数不符：期望 {expected_count}，实际 {len(records)}")
|
||
return records
|