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()
|