张胜宇
e239eb778b
1) 客服 Agent 四份交付文档 + 构建脚手架：品牌由包装占位 XX科技 / 旧名 南方财富 统一为南方基金（热线 400-889-8899 / 官网 nffund.com），系统名改为「智能服务系统」； 同步追加 §0.4 修订记录行，工程记录行保留原占位字面以支撑硬编码扫描验收。 2) 开发文档：清理 28 份已作废/残留文档（14 份移出归档 + 14 份仓库副本）， 新增《文档规整方案与开发前待决事项-2026-09-17》。 3) 客服agent 四份交付文档首次纳入本分支。
80 lines
2.7 KiB
Python
80 lines
2.7 KiB
Python
"""只读探查：Milvus 知识集合的真实 schema（用于评审 NL_develop 的字段映射改动）。
|
||
|
||
只做 describe / query，不写入。结果写 `docs/evidence/knowledge-collections.json`：
|
||
|
||
python tools/probe_knowledge_collections.py
|
||
|
||
要回答的问题（对应交付说明 §4.2）：
|
||
1. 现库集合的**实际字段名**是什么 —— 是 `doc_id`/`content`/`visibility` 那套，
|
||
还是 `knowledge_id`/`snippet` 那套？他选的字段映射（A 方案）建立在这个前提上；
|
||
2. 每个集合有多少行；
|
||
3. 有没有 `visibility` 字段 —— 没有的话，检索层的内部资料硬隔离就是关着的。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from app.core.config import get_settings
|
||
|
||
OUTPUT = Path("docs/evidence/knowledge-collections.json")
|
||
|
||
|
||
def _client() -> Any:
|
||
from pymilvus import MilvusClient # type: ignore[import-untyped]
|
||
|
||
settings = get_settings()
|
||
return MilvusClient(uri=settings.milvus_uri, token=settings.milvus_token or None)
|
||
|
||
|
||
def collect() -> dict[str, Any]:
|
||
client = _client()
|
||
report: dict[str, Any] = {"collections": []}
|
||
for name in sorted(client.list_collections()):
|
||
try:
|
||
described = client.describe_collection(name)
|
||
fields = [
|
||
{
|
||
"name": str(field.get("name")),
|
||
"type": str(field.get("type")),
|
||
"is_primary": bool(field.get("is_primary", False)),
|
||
}
|
||
for field in described.get("fields", [])
|
||
]
|
||
stats = client.get_collection_stats(name)
|
||
row_count = stats.get("row_count") if isinstance(stats, dict) else None
|
||
except Exception as exc: # 单个集合探测失败不影响其余
|
||
report["collections"].append(
|
||
{"collection": name, "error": f"{type(exc).__name__}: {exc}"}
|
||
)
|
||
continue
|
||
report["collections"].append(
|
||
{
|
||
"collection": name,
|
||
"row_count": row_count,
|
||
"field_names": [field["name"] for field in fields],
|
||
"fields": fields,
|
||
"has_visibility": any(field["name"] == "visibility" for field in fields),
|
||
"looks_like_legacy_schema": any(
|
||
field["name"] in {"doc_id", "chapter", "doc_no"} for field in fields
|
||
),
|
||
}
|
||
)
|
||
return report
|
||
|
||
|
||
def main() -> None:
|
||
report = collect()
|
||
OUTPUT.parent.mkdir(parents=True, exist_ok=True)
|
||
OUTPUT.write_text(
|
||
json.dumps(report, ensure_ascii=False, indent=2, default=str),
|
||
encoding="utf-8",
|
||
)
|
||
print(f"wrote {OUTPUT}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|