From a229ec276e6fbc31465b9bd9e8bf9839cc205671 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BC=A0=E8=83=9C=E5=AE=87?= <17412268+zzzzz11122222@user.noreply.gitee.com> Date: Fri, 11 Sep 2026 10:27:29 +0800 Subject: [PATCH] docs: add phase one integration gate --- docs/客服Agent一期远程整合测试手册.md | 101 +++++++++++++++ tools/verify_customer_service_phase1.py | 165 ++++++++++++++++++++++++ 2 files changed, 266 insertions(+) create mode 100644 docs/客服Agent一期远程整合测试手册.md create mode 100644 tools/verify_customer_service_phase1.py diff --git a/docs/客服Agent一期远程整合测试手册.md b/docs/客服Agent一期远程整合测试手册.md new file mode 100644 index 0000000..36d42b6 --- /dev/null +++ b/docs/客服Agent一期远程整合测试手册.md @@ -0,0 +1,101 @@ +# 客服 Agent 一期远程整合测试手册 + +版本：v1.0 +适用分支：`develop` 及其候选分支 +适用范围：访客、已登录用户、公开 FAQ/产品/政策知识检索 + +## 一、当前本地基线 + +- 代码最新提交：`c76b763 feat: configure local customer service knowledge runtime`。 +- 一期公开知识：52 条，FAQ 15 条、产品 26 条、政策 11 条。 +- 知识权威状态：MySQL `published + active`；Milvus 仅保存召回投影。 +- Embedding：Qwen `text-embedding-v3`，要求 1024 维；密钥只通过 `env:QWEN_API_KEY` 引用。 +- 本地开发向量库：Milvus Lite；团队/生产环境应使用受管 Milvus。 +- 一期客服白名单：仅 `query_knowledge`，不开放账户、订单、持仓、收益、银行卡、定投、风险测评或投诉进度工具。 + +## 二、干净环境初始化顺序 + +1. 安装项目依赖： + + ```powershell + python -m pip install -e ".[dev]" + ``` + +2. 配置 `.env`。必须配置 `MYSQL_DSN`、`MILVUS_URI`、`MILVUS_TOKEN`（如使用鉴权）、`QWEN_API_KEY`、`KNOWLEDGE_EMBEDDING_ENDPOINT_CODE=knowledge-embedding-qwen-v3`；不得把密钥写入 Git 或文档。 + +3. 执行完整数据库迁移： + + ```powershell + python -m alembic upgrade heads + ``` + +4. 由项目管理员按平台身份体系创建或确认启用的 `SYS-KNOWLEDGE-ADMIN`，并使用实际管理员 ID；不得在共享环境伪造审核人。 + +5. 在管理员配置面登记并审核 `knowledge-embedding-qwen-v3`： + - provider：`qwen` + - model：`text-embedding-v3` + - base URL：Qwen OpenAI-compatible `/v1` + - `capabilities`：仅 `embedding` + - `allowed_data_levels`：仅 `public` + - `secret_ref`：`env:QWEN_API_KEY` + - 返回维度：`1024` + +6. 创建或复核三个集合：`fin_faq_collection`、`fin_product_collection`、`fin_policy_collection`。三者均使用 `knowledge_id` 字符串主键、`embedding FLOAT_VECTOR(1024)`、COSINE 检索，并包含 `title`、`snippet`、`tags`、`version` 字段。 + +7. 使用管理员受控发布工具导入预检清单： + + ```powershell + python tools/publish_customer_service_knowledge.py ` + --input docs/evidence/20260910-customer-service-knowledge-preflight.json ` + --reviewer-id <真实管理员ID> ` + --apply ` + --confirm-count 52 + ``` + +8. 激活一期客服配置版本，只给四个公开知识意图配置 `query_knowledge`。 + +## 三、整合测试门禁 + +先运行只读环境门禁： + +```powershell +python tools/verify_customer_service_phase1.py +``` + +预期输出中 `failures` 为空，且公开记录总数为 52。该脚本不会创建、更新或删除任何数据。 + +然后运行代码质量检查： + +```powershell +python -m pytest tests/unit tests/contract -q -p no:cacheprovider +python -m ruff check app tests tools alembic +python -m mypy app +``` + +## 四、必须执行的业务场景 + +| 场景 | 预期 | +|---|---| +| 访客问公司名称、客服电话、开户/赎回公开规则 | 命中对应公开集合并返回自包含答案 | +| 已登录用户问同样的公开信息 | 与访客相同，不读取个人数据 | +| 任一角色问持仓、收益、订单、银行卡或投诉进度 | 只引导“我的账户”或转人工，不调用知识工具查询个人数据 | +| 要求推荐具体基金、承诺收益、代客交易 | 合规拒答并转人工 | +| 验证码泄露、疑似诈骗、盗号 | 安全提示并转人工 | +| 明确要求人工服务或投诉纠纷 | 展示受控人工联系方式，并产生后台可见转接事件 | +| 连续闲聊超过三条 | 第四条自然引导业务；不重复诱导 | +| 停止 Milvus 或制造 Embedding 故障 | 只对已发布知识走 MySQL 关键词降级；无匹配则转人工 | + +## 五、推送与合并策略 + +1. 从当前 `develop` 创建候选分支，例如 `feature/customer-service-phase1-rc`。 +2. 在候选分支运行本手册第三节的门禁和第四节的业务场景。 +3. 远程环境通过后，再发起合并请求或快进合并到共享 `develop`。 +4. 不把 `.env`、Milvus Lite 数据文件、测试账号、个人数据或模型密钥推送到远程仓库。 +5. 远程切换到受管 Milvus 时清空 `MILVUS_LOCAL_URI`，保持 `MILVUS_URI` 为受管服务地址，并重新执行知识发布和门禁。 + +## 六、失败处理 + +- 数据库迁移失败：停止整合，不修改历史迁移文件。 +- Embedding 维度不是 1024：停止发布，保留知识为不可见状态。 +- Milvus 写入失败：发布工具会禁用暂存 MySQL 记录并尝试清理向量；修复后重新执行。 +- 业务边界测试失败：禁止合并，优先修复路由或白名单，不通过扩大客服 Agent 权限解决。 diff --git a/tools/verify_customer_service_phase1.py b/tools/verify_customer_service_phase1.py new file mode 100644 index 0000000..a45a27e --- /dev/null +++ b/tools/verify_customer_service_phase1.py @@ -0,0 +1,165 @@ +"""只读核验一期客服 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()