"""把已审核的 QA 源文件一次性导入 `fin_knowledge_meta` + `agent_faq_synonym`，并投递向量同步事件。 为什么是一个独立的种子脚本、而不是"跑一次 SQL"：这 105 条（+1 条验收补充）是客服 Agent RAG 检索的**全部**知识来源。它们必须能被重复导入——Task 2 改治理层、Task 4 建文档解析、 Task 5 写 Outbox Worker 都可能要重放这次导入；一旦脚本不幂等，重放就会产生重复知识、 重复同义问法，并在 Milvus 里留下同一段文本的多个向量（检索结果会重复命中同一条）。 幂等口径（控制者裁定 4）： 1. 知识行按 `tags.qa_id` 定位——存在则 UPDATE、不存在则 INSERT。`qa_id` 是源文件的稳定编号， 不依赖自增主键，因此重复导入不会新增行。 2. 同义问法先按 `(knowledge_id, source_type='import')` 删除再插入：`import` 来源的整批 由本脚本独占，删除-重插能得到源文件的精确镜像；手工/会话/工单来源的同义问法不受影响。 3. 去重按 `phrase_hash`（SHA-256(归一化问法)）而非原文字符串——`uk_faq_synonym (knowledge_id, phrase_hash)` 就是这么定义的，且「问题」本身也作为一个 phrase 写入 （控制者裁定 5），所以标准问法与相似问法之间的重复也在同一套哈希下去重。 本脚本**不写 Milvus**：向量由 `knowledge.vector_sync_requested` 事件的 Worker（Task 5）落库。 这里只负责把事件投进 `domain_event_outbox`。 写入的固定值（控制者裁定 1/2/3，全部有据可查）： - `milvus_collection='fin_faq_collection'`（与 Task 6 的三集合白名单一致）。 - `effective_date`/`expire_date` 留空 = 长期有效；读路径对空值按"长期有效"处理。 - `review_status='published'` + `status='active'`（源文件本身即 `approved_candidate` 且已人工审核）； 同义问法 `status='approved'` + `source_type='import'`（`docs/02` §7.3 DDL 的合法枚举，带 CHECK）。 - `created_by`/`reviewer_id` = `sys_user.id`（现库 `9003`）——`agent_faq_synonym.created_by` 非空且带外键 `fk_synonym_created_by`，写不存在的用户会直接失败。 执行（可重复执行）： .\\.venv\\Scripts\\python.exe tools\\import_knowledge_seed.py \\ --source "C:\\...\\客服Agent知识库_QA问答对_v5_RAG发布候选版.txt" --created-by 9003 """ from __future__ import annotations import argparse import asyncio import json import sys from datetime import UTC, datetime from pathlib import Path from uuid import uuid4 ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from sqlalchemy import text # noqa: E402 from sqlalchemy.ext.asyncio import AsyncSession # noqa: E402 from app.infrastructure.db import SessionFactory # noqa: E402 from tools.qa_source_parser import ( # noqa: E402 QaRecord, normalize_phrase, parse_qa_source, phrase_hash, ) #: 三个 Milvus 集合里的 FAQ 集合（Task 6 的白名单口径）。 FAQ_COLLECTION = "fin_faq_collection" #: `fin_knowledge_meta.knowledge_type`：FAQ 文本知识。 KNOWLEDGE_TYPE = "faq" #: 源文件版本，写入 `version` 与 `tags.source_version`，便于回溯"这条来自哪一版"。 SOURCE_VERSION = "v5.8" #: 向量同步事件类型（Task 5 的 Worker 消费它）。 VECTOR_SYNC_EVENT = "knowledge.vector_sync_requested" #: 本脚本独占的 `agent_faq_synonym.source_type`；重导时只删这一来源的行。 IMPORT_SOURCE_TYPE = "import" #: 老师 Phase 1 验收点名的测试问题「基金申购后多久确认」**不在 105 条内** #: （源文件里只有 `RAG-CONFIG-006`「申购和赎回什么时候确认、到账？」，措辞不同）， #: 这里补一条同义的验收条目，使验收用例能真正命中知识库而不是落到兜底话术。 ACCEPTANCE_SUPPLEMENT = QaRecord( qa_id="SUP-001", question="基金申购后多久确认？", synonyms=("申购多久确认份额？", "买基金几天能确认？", "基金申购确认时间"), answer=( "交易日 15:00 前提交的申购申请，T+1 日确认份额（QDII 基金为 T+2 日）；" "15:00 后提交则顺延至下一交易日。非交易日（周末及法定节假日）提交的申请顺延至" "下一交易日处理。确认后即可查看持仓。具体以产品说明书和交易页面显示为准。" ), ) SELECT_KNOWLEDGE_ID_BY_QA_ID = text( "SELECT id FROM fin_knowledge_meta" " WHERE JSON_UNQUOTE(JSON_EXTRACT(tags, '$.qa_id')) = :qa_id" " ORDER BY id LIMIT 1" ) INSERT_KNOWLEDGE_META = text( "INSERT INTO fin_knowledge_meta" " (knowledge_type, title, source_file, milvus_collection, version, content_text," " tags, reviewer_id, review_status, status, created_at, updated_at)" " VALUES (:knowledge_type, :title, :source_file, :collection, :version, :content_text," " :tags, :reviewer_id, 'published', 'active', :now, :now)" ) UPDATE_KNOWLEDGE_META = text( "UPDATE fin_knowledge_meta SET title=:title, source_file=:source_file," " milvus_collection=:collection, version=:version, content_text=:content_text," " tags=:tags, reviewer_id=:reviewer_id, review_status='published', status='active'," " updated_at=:now WHERE id=:id" ) SELECT_LAST_INSERT_ID = text("SELECT LAST_INSERT_ID()") DELETE_IMPORTED_SYNONYMS = text( "DELETE FROM agent_faq_synonym WHERE knowledge_id=:knowledge_id" " AND source_type=:source_type" ) INSERT_SYNONYM = text( "INSERT INTO agent_faq_synonym" " (knowledge_id, phrase, normalized_phrase, phrase_hash, language_code, source_type," " hit_count, status, created_by, reviewer_id, reviewed_at, created_at, updated_at)" " VALUES (:knowledge_id, :phrase, :normalized_phrase, :phrase_hash, 'zh-CN'," " :source_type, 0, 'approved', :created_by, :created_by, :now, :now, :now)" ) INSERT_OUTBOX_EVENT = text( "INSERT INTO domain_event_outbox" " (event_id, event_type, aggregate_type, aggregate_id, trace_id, payload, status," " retry_count, occurred_at, created_at, updated_at)" " VALUES (:event_id, :event_type, 'knowledge_meta', :aggregate_id, :trace_id, :payload," " 'pending', 0, :now, :now, :now)" ) def _knowledge_tags(record: QaRecord) -> str: return json.dumps( {"qa_id": record.qa_id, "phase": "phase_1", "source_version": SOURCE_VERSION}, ensure_ascii=False, ) async def upsert_knowledge_meta( session: AsyncSession, record: QaRecord, *, source_file: str, created_by: int, now: datetime, ) -> int: """按 `tags.qa_id` 幂等写入知识行，返回 `fin_knowledge_meta.id`。""" params = { "knowledge_type": KNOWLEDGE_TYPE, "title": record.question, "source_file": source_file, "collection": FAQ_COLLECTION, "version": SOURCE_VERSION, "content_text": record.answer, "tags": _knowledge_tags(record), "reviewer_id": created_by, "now": now, } existing = await session.scalar(SELECT_KNOWLEDGE_ID_BY_QA_ID, {"qa_id": record.qa_id}) if existing is None: await session.execute(INSERT_KNOWLEDGE_META, params) inserted = await session.scalar(SELECT_LAST_INSERT_ID) if inserted is None: # pragma: no cover - 只在非 MySQL 后端上发生 raise RuntimeError(f"{record.qa_id} 插入后取不到 LAST_INSERT_ID()") return int(inserted) knowledge_id = int(existing) await session.execute(UPDATE_KNOWLEDGE_META, {**params, "id": knowledge_id}) return knowledge_id async def replace_imported_synonyms( session: AsyncSession, knowledge_id: int, record: QaRecord, *, created_by: int, now: datetime, ) -> int: """重建该知识的 `import` 来源同义问法（含标准问法本身），返回写入行数。""" await session.execute( DELETE_IMPORTED_SYNONYMS, {"knowledge_id": knowledge_id, "source_type": IMPORT_SOURCE_TYPE}, ) digests: set[str] = set() written = 0 for phrase in (record.question, *record.synonyms): digest = phrase_hash(phrase) if digest in digests: continue digests.add(digest) await session.execute(INSERT_SYNONYM, { "knowledge_id": knowledge_id, "phrase": phrase, "normalized_phrase": normalize_phrase(phrase), "phrase_hash": digest, "source_type": IMPORT_SOURCE_TYPE, "created_by": created_by, "now": now, }) written += 1 return written async def enqueue_vector_sync( session: AsyncSession, knowledge_id: int, *, now: datetime, ) -> None: """投递向量同步事件；Milvus 的写入由 Task 5 的 Outbox Worker 负责。""" await session.execute(INSERT_OUTBOX_EVENT, { "event_id": str(uuid4()), "event_type": VECTOR_SYNC_EVENT, "aggregate_id": str(knowledge_id), "trace_id": str(uuid4()), "payload": json.dumps({"knowledge_id": str(knowledge_id)}), "now": now, }) async def import_records( records: list[QaRecord], *, source_file: str, created_by: int, now: datetime, ) -> tuple[int, int]: """单事务导入全部记录，返回 `(知识行数, 同义问法行数)`。""" synonyms_written = 0 async with SessionFactory() as session, session.begin(): for record in records: knowledge_id = await upsert_knowledge_meta( session, record, source_file=source_file, created_by=created_by, now=now, ) synonyms_written += await replace_imported_synonyms( session, knowledge_id, record, created_by=created_by, now=now, ) await enqueue_vector_sync(session, knowledge_id, now=now) return len(records), synonyms_written async def verify(expected_knowledge: int, expected_synonyms: int) -> None: """按读路径的**同一口径**复查落库结果：任一项不符就 `SystemExit`（非零退出码）。 四项断言，全部是"== 期望值"：知识行数、同义问法行数、被向量同步事件覆盖的 `aggregate_id` 数、验收补充条目可查。**不打印了事**——见函数体内的说明。 """ async with SessionFactory() as session: published = await session.scalar(text( "SELECT COUNT(*) FROM fin_knowledge_meta" " WHERE review_status='published' AND status='active'" " AND milvus_collection=:collection" " AND (effective_date IS NULL OR effective_date <= UTC_DATE())" " AND (expire_date IS NULL OR expire_date > UTC_DATE())" ), {"collection": FAQ_COLLECTION}) imported_synonyms = await session.scalar(text( "SELECT COUNT(*) FROM agent_faq_synonym" " WHERE source_type=:source_type AND status='approved'" ), {"source_type": IMPORT_SOURCE_TYPE}) events = await session.scalar(text( "SELECT COUNT(*) FROM domain_event_outbox WHERE event_type=:event_type" ), {"event_type": VECTOR_SYNC_EVENT}) covered = await session.scalar(text( "SELECT COUNT(DISTINCT aggregate_id) FROM domain_event_outbox" " WHERE event_type=:event_type" ), {"event_type": VECTOR_SYNC_EVENT}) supplement = await session.scalar(text( "SELECT id FROM fin_knowledge_meta WHERE title=:title" " AND knowledge_type=:knowledge_type AND milvus_collection=:collection" " AND review_status='published' AND status='active'" ), {"title": ACCEPTANCE_SUPPLEMENT.question, "knowledge_type": KNOWLEDGE_TYPE, "collection": FAQ_COLLECTION}) # 断言而不是打印：`verify()` 存在的意义就是"每轮导入都确认真的能被读路径查到"。 # 只打印不断言的话，一轮投递 0 个事件、知识行被别的流程删掉，脚本照样打印 # `verified:` 并退出 0——「静默失效」比报错难查得多（本 Task 审查时真的发生过： # 底座配置发布测试的 `finally` 清理删 Outbox 时漏了 `event_type` 条件， # `aggregate_id` 是字符串列而 release_id 也是小整数，数值碰撞把知识事件误删了）。 print(f" published+active+有效 知识行: {published}（本次导入 {expected_knowledge} 条）") print(f" approved+import 同义问法行: {imported_synonyms}（本次写入 {expected_synonyms} 条）") print(f" {VECTOR_SYNC_EVENT} 事件行: {events}，覆盖 {covered} 个 aggregate_id" f"（本次投递 {expected_knowledge} 条）") # 用 `==` 而不是 `>=`：多出来的行同样是坏消息（别的流程往同一集合里写了不在 # 本次导入范围内的知识），放过去等于把这个集合的"精确镜像"性质丢掉。 if published is None or int(published) != expected_knowledge: raise SystemExit(f"知识行不符：期望 == {expected_knowledge}，实际 {published}") if imported_synonyms is None or int(imported_synonyms) != expected_synonyms: raise SystemExit(f"同义问法行不符：期望 {expected_synonyms}，实际 {imported_synonyms}") # 本脚本对 `import_records` 的每条知识都投一条事件（见 `import_records` 循环）， # 所以"事件覆盖的 aggregate_id 数 == 知识行数"在本设计下必须成立： # 少一个就说明有知识行没进入向量同步队列（就是那个被误删的场景）。 if covered is None or int(covered) != expected_knowledge: raise SystemExit( f"向量同步事件未覆盖全部知识行：期望 {expected_knowledge} 个 aggregate_id，" f"实际 {covered}" ) if supplement is None: raise SystemExit( f"验收补充条目查不到：{ACCEPTANCE_SUPPLEMENT.question}" f"（knowledge_type={KNOWLEDGE_TYPE}, collection={FAQ_COLLECTION}, published+active）" ) print(f" verified: 验收补充条目 id={supplement} 可查；" f"{expected_knowledge} 个知识行全部有向量同步事件") async def main() -> None: parser = argparse.ArgumentParser(description="导入已审核 QA 知识库（幂等，可重复执行）") parser.add_argument("--source", required=True, type=Path, help="已审核 QA 源文件") parser.add_argument("--created-by", required=True, type=int, help="sys_user.id（外键）") parser.add_argument("--dry-run", action="store_true", help="只解析、不写库") args = parser.parse_args() records = parse_qa_source(args.source.read_text(encoding="utf-8")) records.append(ACCEPTANCE_SUPPLEMENT) print(f"parsed {len(records)} records" f" (含 1 条验收补充：{ACCEPTANCE_SUPPLEMENT.qa_id})") if args.dry_run: print("dry-run: 未写库") return now = datetime.now(UTC).replace(tzinfo=None) knowledge_count, synonym_count = await import_records( records, source_file=str(args.source), created_by=args.created_by, now=now, ) print(f"imported {knowledge_count} knowledge rows, {synonym_count} synonym rows;" f" {knowledge_count} vector sync events queued") await verify(knowledge_count, synonym_count) if __name__ == "__main__": asyncio.run(main())