166 lines
6.7 KiB
Python
166 lines
6.7 KiB
Python
"""只读核验一期客服 Agent 的数据库、知识库和向量运行环境。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import json
|
|
import sys
|
|
from collections.abc import Iterable
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from sqlalchemy import text
|
|
|
|
# 直接执行 tools 脚本时优先解析当前工作树，避免误导入相邻 worktree 的 app 包。
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from app.core.config import get_settings # noqa: E402
|
|
from app.infrastructure.db import SessionFactory # noqa: E402
|
|
|
|
COLLECTIONS = (
|
|
"fin_faq_collection",
|
|
"fin_product_collection",
|
|
"fin_policy_collection",
|
|
)
|
|
EXPECTED_KNOWLEDGE_COUNTS = {
|
|
"fin_faq_collection": 15,
|
|
"fin_product_collection": 26,
|
|
"fin_policy_collection": 11,
|
|
}
|
|
|
|
|
|
def _failures(values: Iterable[str]) -> list[str]:
|
|
"""统一收集失败项，保证脚本最后一次性输出可操作结果。"""
|
|
return [value for value in values if value]
|
|
|
|
|
|
async def _database_checks() -> list[str]:
|
|
"""只读检查管理员、Embedding 端点、配置版本与知识发布状态。"""
|
|
failures: list[str] = []
|
|
async with SessionFactory() as session:
|
|
admin = await session.execute(text("""
|
|
SELECT id FROM sys_user
|
|
WHERE id = 9003 AND user_no = 'SYS-KNOWLEDGE-ADMIN'
|
|
AND user_type IN ('employee', 'admin')
|
|
AND status IN ('正常', 'active')
|
|
"""))
|
|
if admin.scalar_one_or_none() is None:
|
|
failures.append("缺少启用的 SYS-KNOWLEDGE-ADMIN(9003)")
|
|
|
|
endpoint = await session.execute(text("""
|
|
SELECT endpoint_code, model_name, secret_ref, capabilities, status
|
|
FROM model_endpoint_config
|
|
WHERE endpoint_code = 'knowledge-embedding-qwen-v3'
|
|
AND status = 'active'
|
|
"""))
|
|
endpoint_row = endpoint.mappings().first()
|
|
if endpoint_row is None:
|
|
failures.append("Qwen Embedding 端点未激活")
|
|
else:
|
|
if endpoint_row["model_name"] != "text-embedding-v3":
|
|
failures.append("Embedding 模型不是 text-embedding-v3")
|
|
if not str(endpoint_row["secret_ref"]).startswith("env:"):
|
|
failures.append("Embedding 密钥不是 env: 引用")
|
|
|
|
release = await session.execute(text("""
|
|
SELECT id FROM config_release
|
|
WHERE release_no = 'customer-service-phase1-public-kb-v1'
|
|
AND status = 'active'
|
|
"""))
|
|
release_id = release.scalar_one_or_none()
|
|
if release_id is None:
|
|
failures.append("一期客服公开检索配置未激活")
|
|
else:
|
|
tools = await session.execute(text("""
|
|
SELECT config_key, value_json
|
|
FROM platform_config_item
|
|
WHERE release_id = :release_id AND namespace = 'agent_tools'
|
|
"""), {"release_id": release_id})
|
|
configured: dict[str, Any] = {}
|
|
for row in tools.mappings():
|
|
raw_value = row["value_json"]
|
|
configured[str(row["config_key"])] = (
|
|
json.loads(raw_value) if isinstance(raw_value, str) else raw_value
|
|
)
|
|
expected_keys = {
|
|
"customer_service:public_knowledge",
|
|
"customer_service:faq",
|
|
"customer_service:product_inquiry",
|
|
"customer_service:policy_explain",
|
|
}
|
|
if set(configured) != expected_keys:
|
|
failures.append("一期客服工具白名单缺失或包含额外意图")
|
|
if any(value != {"allowed_tools": ["query_knowledge"]} for value in configured.values()):
|
|
failures.append("一期客服工具白名单不是仅 query_knowledge")
|
|
|
|
knowledge = await session.execute(text("""
|
|
SELECT milvus_collection, review_status, status, COUNT(*) AS count
|
|
FROM fin_knowledge_meta
|
|
GROUP BY milvus_collection, review_status, status
|
|
"""))
|
|
actual: dict[str, int] = {}
|
|
for row in knowledge.mappings():
|
|
if row["review_status"] == "published" and row["status"] == "active":
|
|
actual[str(row["milvus_collection"])] = int(row["count"])
|
|
if actual != EXPECTED_KNOWLEDGE_COUNTS:
|
|
failures.append(f"公开知识数量不符合预期: {actual}")
|
|
return failures
|
|
|
|
|
|
async def _milvus_checks() -> list[str]:
|
|
"""只读检查三类集合的存在、维度、主键和行数。"""
|
|
settings = get_settings()
|
|
failures: list[str] = []
|
|
try:
|
|
from pymilvus import AsyncMilvusClient # type: ignore[import-untyped]
|
|
|
|
client: Any = AsyncMilvusClient(
|
|
uri=settings.resolved_milvus_uri, token=settings.milvus_token or None
|
|
)
|
|
for collection in COLLECTIONS:
|
|
if not await client.has_collection(collection_name=collection):
|
|
failures.append(f"集合不存在: {collection}")
|
|
continue
|
|
description = await client.describe_collection(collection_name=collection)
|
|
fields = {field["name"]: field for field in description.get("fields", [])}
|
|
embedding = fields.get("embedding", {})
|
|
if embedding.get("params", {}).get("dim") != 1024:
|
|
failures.append(f"集合 {collection} 不是 1024 维")
|
|
if not fields.get("knowledge_id", {}).get("is_primary"):
|
|
failures.append(f"集合 {collection} 缺少 knowledge_id 主键")
|
|
await client.load_collection(collection_name=collection)
|
|
stats = await client.get_collection_stats(collection_name=collection)
|
|
expected = EXPECTED_KNOWLEDGE_COUNTS[collection]
|
|
if int(stats.get("row_count", -1)) != expected:
|
|
failures.append(f"集合 {collection} 行数不符合预期: {stats}")
|
|
await client.close()
|
|
except Exception as exc:
|
|
failures.append(f"Milvus 检查失败: {type(exc).__name__}")
|
|
return failures
|
|
|
|
|
|
async def verify() -> int:
|
|
"""执行所有只读门禁并返回适合 CI 的退出码。"""
|
|
failures = _failures([*(await _database_checks()), *(await _milvus_checks())])
|
|
settings = get_settings()
|
|
print({
|
|
"milvus_uri_mode": "local" if settings.milvus_local_uri else "remote",
|
|
"knowledge_embedding_endpoint": settings.knowledge_embedding_endpoint_code,
|
|
"expected_public_records": sum(EXPECTED_KNOWLEDGE_COUNTS.values()),
|
|
"failures": failures,
|
|
})
|
|
return 1 if failures else 0
|
|
|
|
|
|
def main() -> None:
|
|
"""命令行入口；保留无参数形式，便于整合测试直接调用。"""
|
|
argparse.ArgumentParser(description=__doc__).parse_args()
|
|
raise SystemExit(asyncio.run(verify()))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|