feat: add knowledge import preflight
This commit is contained in:
张胜宇
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,75 @@
|
||||
import json
|
||||
|
||||
import pytest
|
||||
|
||||
from tools.knowledge_import_preflight import build_import_manifest
|
||||
|
||||
|
||||
def vector_candidate() -> dict[str, object]:
|
||||
return {
|
||||
"qa_id": "RAG-PUB-001",
|
||||
"title": "基金交易确认时间说明",
|
||||
"question": "基金什么时候确认？",
|
||||
"paraphrases": ["申购何时确认", "赎回多久确认"],
|
||||
"answer": "交易确认时间以产品规则和实际交易日为准。",
|
||||
"scope": "public",
|
||||
"intent": "policy_explain",
|
||||
"collection": "fin_policy_collection",
|
||||
"execution_mode": "vector_search",
|
||||
"retrieval_status": "approved_candidate",
|
||||
"audience": ["visitor", "authenticated_user"],
|
||||
"agent_data_access": "none",
|
||||
"tags": ["交易规则", "确认"],
|
||||
"phase": "phase_1",
|
||||
"source_type": "qa_pair",
|
||||
"source_file": "qa-v5.8.txt",
|
||||
"source_url": None,
|
||||
"source_version": "v5.8",
|
||||
"review_status": "approved_candidate",
|
||||
"status": "active",
|
||||
"effective_date": None,
|
||||
"expire_date": None,
|
||||
}
|
||||
|
||||
|
||||
def rule_only_record() -> dict[str, object]:
|
||||
return {
|
||||
"qa_id": "RAG-SEC-001",
|
||||
"scope": "security_notice",
|
||||
"collection": None,
|
||||
"execution_mode": "fixed_route",
|
||||
"retrieval_status": "rule_only",
|
||||
}
|
||||
|
||||
|
||||
def test_manifest_contains_only_eligible_public_records_pending_admin_review() -> None:
|
||||
manifest = build_import_manifest(
|
||||
[vector_candidate(), rule_only_record()], source_name="qa-v5.8.jsonl"
|
||||
)
|
||||
|
||||
assert manifest["summary"] == {
|
||||
"total_records": 2,
|
||||
"eligible_records": 1,
|
||||
"excluded_rule_records": 1,
|
||||
"publication_state": "pending_review",
|
||||
}
|
||||
entry = manifest["records"][0]
|
||||
assert entry["qa_id"] == "RAG-PUB-001"
|
||||
assert entry["knowledge_type"] == "policy_explain"
|
||||
assert entry["milvus_collection"] == "fin_policy_collection"
|
||||
assert entry["review_status"] == "pending_review"
|
||||
assert entry["status"] == "active"
|
||||
assert entry["retrieval_text"] == (
|
||||
"标准问题：基金什么时候确认？\n"
|
||||
"相似问法：申购何时确认；赎回多久确认\n"
|
||||
"标签：交易规则、确认"
|
||||
)
|
||||
assert json.loads(entry["content_text"])["answer"] == "交易确认时间以产品规则和实际交易日为准。"
|
||||
|
||||
|
||||
def test_invalid_public_record_is_rejected_instead_of_silently_entering_manifest() -> None:
|
||||
invalid = vector_candidate()
|
||||
invalid["agent_data_access"] = "account"
|
||||
|
||||
with pytest.raises(ValueError, match="RAG-PUB-001"):
|
||||
build_import_manifest([invalid], source_name="qa-v5.8.jsonl")
|
||||
@@ -0,0 +1,221 @@
|
||||
"""将一期公开 QA 候选转换为不含外部写入的待审核导入清单。"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
from collections.abc import Mapping, Sequence
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from app.core.knowledge_contracts import ALLOWED_KNOWLEDGE_COLLECTIONS
|
||||
from app.service.knowledge_config import KnowledgeRuntimeConfig
|
||||
|
||||
|
||||
# 仅允许一期三类公开知识按既定路由进入后续发布流程。
|
||||
EXPECTED_COLLECTIONS = {
|
||||
intent: collection
|
||||
for intent, (collection, _top_k) in KnowledgeRuntimeConfig.DEFAULT_ROUTES.items()
|
||||
}
|
||||
# 公开候选必须具备的字段，缺失时不能生成不完整的发布清单。
|
||||
REQUIRED_PUBLIC_FIELDS = frozenset({
|
||||
"qa_id",
|
||||
"title",
|
||||
"question",
|
||||
"paraphrases",
|
||||
"answer",
|
||||
"scope",
|
||||
"intent",
|
||||
"collection",
|
||||
"execution_mode",
|
||||
"retrieval_status",
|
||||
"audience",
|
||||
"agent_data_access",
|
||||
"tags",
|
||||
"source_type",
|
||||
"source_file",
|
||||
"source_version",
|
||||
"review_status",
|
||||
"status",
|
||||
})
|
||||
|
||||
|
||||
def _required_string(record: Mapping[str, object], field: str, qa_id: str) -> str:
|
||||
"""读取非空字符串字段，拒绝将不完整资料带入后续发布阶段。"""
|
||||
value = record.get(field)
|
||||
if not isinstance(value, str) or not value.strip():
|
||||
raise ValueError(f"{qa_id}: {field} must be a non-empty string")
|
||||
return value.strip()
|
||||
|
||||
|
||||
def _required_strings(record: Mapping[str, object], field: str, qa_id: str) -> list[str]:
|
||||
"""读取非空字符串数组，并保持来源中已经审核的条目顺序。"""
|
||||
value = record.get(field)
|
||||
if not isinstance(value, list) or not value:
|
||||
raise ValueError(f"{qa_id}: {field} must be a non-empty string list")
|
||||
strings = [item.strip() for item in value if isinstance(item, str) and item.strip()]
|
||||
if len(strings) != len(value):
|
||||
raise ValueError(f"{qa_id}: {field} must contain only non-empty strings")
|
||||
return strings
|
||||
|
||||
|
||||
def _is_rule_only(record: Mapping[str, object]) -> bool:
|
||||
"""控制类记录必须由应用层固定路由处理，绝不能进入向量导入清单。"""
|
||||
return (
|
||||
record.get("collection") is None
|
||||
and record.get("execution_mode") == "fixed_route"
|
||||
and record.get("retrieval_status") == "rule_only"
|
||||
)
|
||||
|
||||
|
||||
def _validate_public_record(record: Mapping[str, object]) -> None:
|
||||
"""验证候选是否符合一期公开知识与最小权限边界。"""
|
||||
qa_id = _required_string(record, "qa_id", "<unknown>")
|
||||
missing = sorted(field for field in REQUIRED_PUBLIC_FIELDS if field not in record)
|
||||
if missing:
|
||||
raise ValueError(f"{qa_id}: missing required fields: {', '.join(missing)}")
|
||||
intent = _required_string(record, "intent", qa_id)
|
||||
collection = _required_string(record, "collection", qa_id)
|
||||
if intent not in EXPECTED_COLLECTIONS:
|
||||
raise ValueError(f"{qa_id}: unsupported public intent: {intent}")
|
||||
if collection not in ALLOWED_KNOWLEDGE_COLLECTIONS:
|
||||
raise ValueError(f"{qa_id}: collection is not allowlisted: {collection}")
|
||||
if collection != EXPECTED_COLLECTIONS[intent]:
|
||||
raise ValueError(f"{qa_id}: collection does not match intent route")
|
||||
if record.get("scope") != "public":
|
||||
raise ValueError(f"{qa_id}: public vector record must have scope=public")
|
||||
if record.get("execution_mode") != "vector_search":
|
||||
raise ValueError(f"{qa_id}: public record must use vector_search")
|
||||
if record.get("retrieval_status") != "approved_candidate":
|
||||
raise ValueError(f"{qa_id}: record is not an approved candidate")
|
||||
if record.get("review_status") != "approved_candidate":
|
||||
raise ValueError(f"{qa_id}: review state cannot enter preflight")
|
||||
if record.get("status") != "active":
|
||||
raise ValueError(f"{qa_id}: inactive record cannot enter preflight")
|
||||
if record.get("agent_data_access") != "none":
|
||||
raise ValueError(f"{qa_id}: agent data access must remain none")
|
||||
audience = _required_strings(record, "audience", qa_id)
|
||||
if set(audience) != {"visitor", "authenticated_user"}:
|
||||
raise ValueError(f"{qa_id}: audience must be visitor and authenticated_user")
|
||||
_required_string(record, "title", qa_id)
|
||||
_required_string(record, "question", qa_id)
|
||||
_required_strings(record, "paraphrases", qa_id)
|
||||
_required_string(record, "answer", qa_id)
|
||||
_required_strings(record, "tags", qa_id)
|
||||
_required_string(record, "source_type", qa_id)
|
||||
_required_string(record, "source_file", qa_id)
|
||||
_required_string(record, "source_version", qa_id)
|
||||
|
||||
|
||||
def _public_entry(record: Mapping[str, object]) -> dict[str, object]:
|
||||
"""构造供管理员审核的确定性条目，不生成数据库主键或向量。"""
|
||||
qa_id = _required_string(record, "qa_id", "<unknown>")
|
||||
question = _required_string(record, "question", qa_id)
|
||||
paraphrases = _required_strings(record, "paraphrases", qa_id)
|
||||
tags = _required_strings(record, "tags", qa_id)
|
||||
answer = _required_string(record, "answer", qa_id)
|
||||
content = {
|
||||
"qa_id": qa_id,
|
||||
"question": question,
|
||||
"paraphrases": paraphrases,
|
||||
"answer": answer,
|
||||
"audience": _required_strings(record, "audience", qa_id),
|
||||
"agent_data_access": "none",
|
||||
"source_version": _required_string(record, "source_version", qa_id),
|
||||
}
|
||||
return {
|
||||
"qa_id": qa_id,
|
||||
"knowledge_type": _required_string(record, "intent", qa_id),
|
||||
"title": _required_string(record, "title", qa_id),
|
||||
"milvus_collection": _required_string(record, "collection", qa_id),
|
||||
"version": _required_string(record, "source_version", qa_id),
|
||||
"source_file": _required_string(record, "source_file", qa_id),
|
||||
"source_type": _required_string(record, "source_type", qa_id),
|
||||
"source_url": record.get("source_url"),
|
||||
"effective_date": record.get("effective_date"),
|
||||
"expire_date": record.get("expire_date"),
|
||||
"tags": tags,
|
||||
"content_text": json.dumps(content, ensure_ascii=False, separators=(",", ":")),
|
||||
"retrieval_text": (
|
||||
f"标准问题：{question}\n"
|
||||
f"相似问法：{'；'.join(paraphrases)}\n"
|
||||
f"标签：{'、'.join(tags)}"
|
||||
),
|
||||
"snippet": question[:300],
|
||||
# 管理员填入真实审核人并批准前，预检清单绝不伪装成已发布数据。
|
||||
"review_status": "pending_review",
|
||||
"status": "active",
|
||||
}
|
||||
|
||||
|
||||
def build_import_manifest(
|
||||
records: Sequence[Mapping[str, object]], *, source_name: str
|
||||
) -> dict[str, object]:
|
||||
"""构建可复查的导入清单，并拒绝任何不满足公开边界的非控制类记录。"""
|
||||
entries: list[dict[str, object]] = []
|
||||
seen_ids: set[str] = set()
|
||||
excluded_rule_records = 0
|
||||
for record in records:
|
||||
if _is_rule_only(record):
|
||||
excluded_rule_records += 1
|
||||
continue
|
||||
_validate_public_record(record)
|
||||
entry = _public_entry(record)
|
||||
qa_id = str(entry["qa_id"])
|
||||
if qa_id in seen_ids:
|
||||
raise ValueError(f"duplicate qa_id: {qa_id}")
|
||||
seen_ids.add(qa_id)
|
||||
entries.append(entry)
|
||||
return {
|
||||
"source_name": source_name,
|
||||
"summary": {
|
||||
"total_records": len(records),
|
||||
"eligible_records": len(entries),
|
||||
"excluded_rule_records": excluded_rule_records,
|
||||
"publication_state": "pending_review",
|
||||
},
|
||||
"records": entries,
|
||||
}
|
||||
|
||||
|
||||
def load_jsonl(source: Path) -> list[dict[str, object]]:
|
||||
"""读取 UTF-8 JSONL，并为每个无效 JSON 行返回带行号的明确错误。"""
|
||||
records: list[dict[str, object]] = []
|
||||
for line_number, line in enumerate(source.read_text(encoding="utf-8").splitlines(), start=1):
|
||||
if not line.strip():
|
||||
continue
|
||||
try:
|
||||
value: Any = json.loads(line)
|
||||
except json.JSONDecodeError as exc:
|
||||
raise ValueError(f"line {line_number}: invalid JSON") from exc
|
||||
if not isinstance(value, dict):
|
||||
raise ValueError(f"line {line_number}: record must be an object")
|
||||
records.append(value)
|
||||
return records
|
||||
|
||||
|
||||
def write_manifest(target: Path, manifest: Mapping[str, object]) -> None:
|
||||
"""仅在调用者显式传入输出路径时，写入本地待审核 JSON 清单。"""
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
target.write_text(
|
||||
json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
|
||||
)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
"""执行本地预检；该入口不读取配置也不连接任何外部服务。"""
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--input", type=Path, required=True)
|
||||
parser.add_argument("--output", type=Path)
|
||||
arguments = parser.parse_args()
|
||||
manifest = build_import_manifest(
|
||||
load_jsonl(arguments.input), source_name=arguments.input.name
|
||||
)
|
||||
if arguments.output is None:
|
||||
print(json.dumps(manifest, ensure_ascii=False, indent=2))
|
||||
else:
|
||||
write_manifest(arguments.output, manifest)
|
||||
print(f"WROTE: {arguments.output}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user