From 6c09cdecbd45531f9026e6e8be05f810d14346bf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=8D=BF=E4=BA=91=E7=A7=8B=E6=9C=88?= <15273589815@163.com> Date: Fri, 11 Sep 2026 19:51:44 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E8=A1=A5=E5=8F=91=E7=94=BB?= =?UTF-8?q?=E5=83=8F=E5=B7=A5=E5=85=B7=E7=99=BD=E5=90=8D=E5=8D=95=E7=9A=84?= =?UTF-8?q?=E8=84=9A=E6=9C=AC=EF=BC=88dry-run=20+=20=E5=90=88=E5=B9=B6?= =?UTF-8?q?=E5=89=8D=E9=98=B2=E5=91=86=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit tools/publish_profile_tool_whitelist.py：把 query_customer_profile 加进 customer_service:faq，供 NL_develop 合并后在本环境补发配置。 关键设计都来自这轮评审核实到的事实： 1. 继承走 ConfigReleaseService.effective_snapshot()，一次拿全三张受管表 （platform_config_item / prompt_template_version / model_routing_rule）。 既有的两个 agent_tools 发布脚本都绕开它、自己写 SQL 只查第一张 —— 用它们发版 会把 201 里那条客服闲聊提示词一起清掉，而 Agent 侧有兜底、功能看着正常、没有告警 （这个事故在本项目真实发生过一次）。 2. 同 key 覆盖：继承项里的 customer_service:faq 必须被本次新值覆盖，否则旧值 ["search_knowledge"] 会被 admin 端校验拦下、整次发布失败。 3. 提示词搬运重分配 version（唯一键含 version），并按 PromptPayload 的 9 个字段挑字段、 归一化 input_schema/output_schema —— 漏带就是静默丢失。 4. 前置防呆：admin 端校验「配置 ⊆ 代码 allowed_tools」（admin_service.py:219-220）， 合并前跑必然 422，而报错只有"配置超出 Agent 工具上限"。脚本先查代码上限： dry-run 给警告后继续预览，正式跑直接中止。 5. JSON 列归一化：SELECT * 读出来的 JSON 列可能是字符串，不解析会静默走进 "生效版本里没有该 key"的分支（本脚本第一版就是这么错的，诊断打印才定位到）。 实测：--dry-run 打印 9 条配置项 + 1 条提示词（v2→v3 重分配）与唯一变化； 正式跑因代码尚未合并而中止，未产生任何写入。 --- tools/publish_profile_tool_whitelist.py | 337 ++++++++++++++++++++++++ 1 file changed, 337 insertions(+) create mode 100644 tools/publish_profile_tool_whitelist.py diff --git a/tools/publish_profile_tool_whitelist.py b/tools/publish_profile_tool_whitelist.py new file mode 100644 index 0000000..22c3c8d --- /dev/null +++ b/tools/publish_profile_tool_whitelist.py @@ -0,0 +1,337 @@ +"""补发客服工具白名单：把 `query_customer_profile` 加进 `customer_service:faq`。 + +## 为什么必须补发 + +`ToolExecutor` 取「代码 `AgentDefinition.allowed_tools` ∩ 发布配置 +`agent_tools/:`」的**交集**，缺配置即**失败关闭**。 + +`NL_develop` 引入画像问答出口后，客服的代码上限变成 +`(search_knowledge, check_suitability, query_customer_profile)`，而本环境生效版本 **201** 的 +`customer_service:faq` 只有 `["search_knowledge"]` —— 画像出口一调用就会被 +`AGENT_PERMISSION_DENIED` 拒掉，表现为"问等级/画像一律转人工"。 + +## 为什么不能只发这一条 + +`config_release` 是**整版本替换**语义：新版本没带上的配置项**等于被删除**。 +所以要先把当前生效版本在**全部三张受管表**（`platform_config_item` / +`prompt_template_version` / `model_routing_rule`）里的内容搬进新版本，再覆盖本次要改的那一条。 + +本环境 201 里有 **9 条 `agent_tools` + 1 条提示词**；只发白名单会把那条提示词一起清掉， +而 Agent 侧有逐字段兜底、会回落代码默认值 —— **功能看着正常、没有任何告警**， +这个事故在本项目真实发生过一次（提示词挂在 release 174、active 变成 181 后读不到）。 + +## 两个已踩过的坑 + +1. **同 key 的继承项必须被本次新定义覆盖。** 否则旧值（这里是 `["search_knowledge"]`） + 会被 admin 端的子集校验 422 拦下，**整次发布失败**，而报错只有 + "配置超出 Agent 工具上限"，看不出是继承造成的。 +2. **提示词搬运必须重分配 `version`。** `prompt_template_version` 的唯一键含 `version`， + 照搬旧行会冲突；同时**必须带上 `input_schema` / `output_schema`**， + `PromptPayload` 只收 9 个字段，漏了就是静默丢失。 + +跑法： + + python tools/publish_profile_tool_whitelist.py --dry-run # 只看要写什么，不动库 + python tools/publish_profile_tool_whitelist.py # 真正发布 + +## ⚠️ 必须在 `NL_develop` 合并进来**之后**才能跑 + +admin 端会校验「配置项 ⊆ 代码 `allowed_tools`」（`app/service/admin_service.py:219-220`）： + + if not set(raw) <= set(definition.allowed_tools): + raise ValidationAgentError("配置超出 Agent 工具上限") + +合并前代码上限还是 `(search_knowledge, check_suitability)`，**不含** `query_customer_profile` +⇒ 这一版会被 422 拒绝，而报错只有"配置超出 Agent 工具上限"，看起来像白名单写错了。 +所以脚本启动时会先查代码上限并给出明确原因，不让你去猜。 +""" + +import asyncio +import argparse +import datetime as dt +import json +import sys +import uuid +from pathlib import Path +from typing import Any + +import httpx +import jwt +from sqlalchemy import func, select + +from app.core.config import get_settings +from app.infrastructure.db import SessionFactory +from app.main import create_app +from app.model.configuration import PromptTemplateVersion +from app.service.agent.bootstrap import get_agent_factory +from app.service.config_release_service import ConfigReleaseService + +# Windows 控制台常是 GBK：输出里的部分符号会直接抛 UnicodeEncodeError，让脚本在 +# "看起来是逻辑错误"的地方失败（本脚本第一版就踩了）。统一兜住，别让编码问题掩盖真问题。 +if hasattr(sys.stdout, "reconfigure"): + sys.stdout.reconfigure(errors="replace") # type: ignore[union-attr] + +ADMIN = "9003" +AGENT_TYPE = "customer_service" +INTENT_KEY = "customer_service:faq" +PROFILE_TOOL = "query_customer_profile" + +# 本次唯一的变化。其余全部原样继承。 +NEW_FAQ_TOOLS: tuple[str, ...] = ("search_knowledge", PROFILE_TOOL) + +# `PromptPayload` 接受的字段（`app/api/schemas/admin.py:56-65`）。 +# 表里还有 checksum / created_by / created_at，那三个由服务端生成，不能搬。 +PROMPT_API_FIELDS = ( + "prompt_code", "task_type", "agent_type", + "system_prompt", "user_prompt_template", "input_schema", "output_schema", +) + + +def _as_json(value: Any) -> Any: + """`SELECT *` 读出来的 JSON 列，在不同驱动下可能是字符串、也可能已解析。 + + 不归一化的后果很隐蔽：`value_json` 保持字符串时，`isinstance(value, dict)` 为假， + 脚本会走到"生效版本里没有该 key"那条分支 —— **看起来像配置缺失，实际只是没解析**。 + 本脚本第一版就是这么错的。`input_schema` / `output_schema` 同理，而且它们是 + `PromptPayload` 的 `dict | None` 字段，传字符串会直接 422。 + """ + return json.loads(value) if isinstance(value, str) else value + + +def token(subject: str) -> str: + settings = get_settings() + private_key = Path(settings.jwt_private_key_path).read_text(encoding="utf-8") + now = dt.datetime.now(dt.UTC) + return jwt.encode( + { + "sub": subject, "iss": settings.jwt_issuer, "aud": settings.jwt_audience, + "exp": now + dt.timedelta(minutes=30), "nbf": now - dt.timedelta(seconds=5), + "jti": str(uuid.uuid4()), + }, + private_key, + algorithm="RS256", + ) + + +async def snapshot() -> dict[str, list[dict[str, Any]]]: + """当前生效版本在全部受管表里的内容；每行已剥掉 id / release_id。""" + async with SessionFactory() as session: + return await ConfigReleaseService(session).effective_snapshot() + + +async def next_prompt_versions(rows: list[dict[str, Any]]) -> list[int]: + """给每个待搬运的 prompt_code 分配一个新 version。 + + 唯一键含 version，照搬旧行会冲突，所以逐个 prompt_code 取现有最大值 +1。 + """ + assigned: list[int] = [] + async with SessionFactory() as session: + for row in rows: + latest = await session.scalar( + select(func.max(PromptTemplateVersion.version)).where( + PromptTemplateVersion.prompt_code == row["prompt_code"] + ) + ) + assigned.append(int(latest or 0) + 1) + return assigned + + +def code_allows_profile_tool() -> bool: + """代码上限里有没有 `query_customer_profile`。 + + admin 端要求「配置 ⊆ 代码 allowed_tools」，所以**合并之前发这一版必然 422**， + 而报错只是"配置超出 Agent 工具上限"。这里提前判定，把真实原因说出来。 + """ + definition = get_agent_factory().definition(AGENT_TYPE) + return PROFILE_TOOL in set(definition.allowed_tools) + + +def current_faq_tools(items: list[dict[str, Any]]) -> list[str] | None: + for item in items: + if str(item.get("item_key")) == INTENT_KEY: + value = item.get("value_json") + if isinstance(value, dict): + tools = value.get("allowed_tools") + if isinstance(tools, list): + return [str(tool) for tool in tools] + return None + + +def merged_items(items: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], list[str]]: + """继承全部配置项，并把 `customer_service:faq` 覆盖成本次的新值。 + + 返回 (新列表, 被替换掉的旧值描述)。**同 key 覆盖**是关键：留给继承的旧值会被 + admin 端的子集校验 422 拦下，整次发布失败。 + """ + replaced: list[str] = [] + merged: list[dict[str, Any]] = [] + for item in items: + if str(item.get("item_key")) == INTENT_KEY: + old = current_faq_tools([item]) or [] + replaced.append(f"{INTENT_KEY}: {old} → {list(NEW_FAQ_TOOLS)}") + merged.append({**item, "value_json": {"allowed_tools": list(NEW_FAQ_TOOLS)}}) + else: + merged.append(item) + return merged, replaced + + +def prompt_payloads( + rows: list[dict[str, Any]], versions: list[int], release_id: int +) -> list[dict[str, Any]]: + payloads: list[dict[str, Any]] = [] + for row, version in zip(rows, versions, strict=True): + payload: dict[str, Any] = {"release_id": release_id, "version": version} + for field in PROMPT_API_FIELDS: + value = row.get(field) + payload[field] = _as_json(value) if field.endswith("_schema") else value + payloads.append(payload) + return payloads + + +async def post( + client: httpx.AsyncClient, path: str, *, auth: dict[str, str], + payload: dict[str, object] | None = None, if_match: str | None = None, +) -> httpx.Response: + headers = {**auth, "Idempotency-Key": uuid.uuid4().hex} + if if_match: + headers["If-Match"] = if_match + return await client.post(path, json=payload, headers=headers) + + +async def etag_of(client: httpx.AsyncClient, path: str, auth: dict[str, str]) -> str | None: + return (await client.get(path, headers=auth)).headers.get("ETag") + + +async def main() -> int: + parser = argparse.ArgumentParser(description="补发客服工具白名单（画像工具）") + parser.add_argument("--dry-run", action="store_true", help="只打印将写入的内容，不调用任何写接口") + args = parser.parse_args() + + if not code_allows_profile_tool(): + print("[警告] 代码里的 customer_service.allowed_tools 还没有 " + f"{PROFILE_TOOL}（仍为合并前状态）。") + print(" admin 端会以「配置超出 Agent 工具上限」422 拒绝这一版 —— " + "请先合并 NL_develop。") + if not args.dry_run: + print("[失败] 正式发布已中止。") + return 1 + print(" dry-run 继续，仅预览将要写入的内容。\n") + + snap = await snapshot() + items = [ + { + "namespace": row["namespace"], + "item_key": row["config_key"], + "value_json": _as_json(row["value_json"]), + "schema_version": row["schema_version"], + } + for row in snap["platform_config_item"] + ] + prompts = list(snap["prompt_template_version"]) + rules = list(snap["model_routing_rule"]) + + faq = current_faq_tools(items) + if faq is None: + print(f"[失败] 生效版本里没有 {INTENT_KEY}，先跑 tools/publish_customer_service_config.py") + print(f" 快照键：{sorted(snap)}") + print(f" 配置项 {len(items)} 条：{[item.get('item_key') for item in items]}") + return 1 + if PROFILE_TOOL in faq: + print(f"✅ {INTENT_KEY} 已包含 {PROFILE_TOOL}（{faq}），无需发布") + return 0 + + merged, replaced = merged_items(items) + versions = await next_prompt_versions(prompts) + + print(f"当前生效版本：配置项 {len(items)} 条、提示词 {len(prompts)} 条、路由规则 {len(rules)} 条") + print("\n本次变更：") + for line in replaced: + print(f" · {line}") + print(f"\n将继承：{len(merged)} 条配置项、{len(prompts)} 条提示词") + for prompt, version in zip(prompts, versions, strict=True): + print(f" · 提示词 {prompt['prompt_code']} v{prompt['version']} → v{version}（重分配）") + + if rules: + # 本环境当前为空；不为空就必须先支持搬运，否则激活后会静默丢规则。 + print(f"\n❌ 生效版本里有 {len(rules)} 条 model_routing_rule，本脚本尚未支持搬运。") + print(" 直接发布会把它们清空 —— 先补上搬运逻辑再跑。") + return 1 + + if args.dry_run: + print("\n[dry-run] 未调用任何写接口。去掉 --dry-run 即真正发布。") + return 0 + + app = create_app() + auth = {"Authorization": f"Bearer {token(ADMIN)}"} + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://test", timeout=60 + ) as client: + created = await post(client, "/api/v1/admin/config-releases", auth=auth, payload={ + "release_no": f"cs-profile-tool-{uuid.uuid4().hex[:12]}", + "title": "客服画像工具白名单", + "change_summary": ( + f"{INTENT_KEY} 增加 {PROFILE_TOOL}（画像问答出口），" + f"并继承既有 {len(merged)} 条配置项与 {len(prompts)} 条提示词" + ), + }) + if created.status_code != 201: + print(f"创建发布版本失败：{created.status_code} {created.text[:200]}") + return 1 + release_id = int(created.json()["data"]["id"]) + print(f"\n新发布版本 id={release_id}") + + base = f"/api/v1/admin/config-releases/{release_id}/platform-config-items" + for item in merged: + response = await post(client, base, auth=auth, payload=item) + if response.status_code != 201: + print(f" 写配置项 {item['item_key']} 失败：{response.status_code} {response.text[:200]}") + return 1 + print(f" 已写入 {len(merged)} 条配置项") + + for payload in prompt_payloads(prompts, versions, release_id): + response = await post(client, "/api/v1/admin/prompt-templates", auth=auth, payload=payload) + if response.status_code not in (200, 201): + print(f" 写提示词 {payload['prompt_code']} 失败：{response.text[:200]}") + return 1 + print(f" 已写入 {len(prompts)} 条提示词") + + release_base = f"/api/v1/admin/config-releases/{release_id}" + submitted = await post(client, f"{release_base}/validations", auth=auth, payload={}, + if_match=await etag_of(client, release_base, auth)) + print(f"\n提交复核：{submitted.status_code}") + if submitted.status_code not in (200, 201): + print(f" 失败：{submitted.text[:300]}") + return 1 + reviewed = await post(client, f"{release_base}/reviews", auth=auth, + payload={"decision": "approved", "comment": "客服画像工具白名单"}, + if_match=await etag_of(client, release_base, auth)) + print(f"审核：{reviewed.status_code}") + activated = await post(client, f"{release_base}/activations", auth=auth, payload={}, + if_match=await etag_of(client, release_base, auth)) + print(f"激活：{activated.status_code}") + if activated.status_code not in (200, 201): + print(f" 失败：{activated.text[:200]}") + return 1 + + after = await snapshot() + restored = current_faq_tools([ + { + "item_key": row["config_key"], + "value_json": _as_json(row["value_json"]), + } + for row in after["platform_config_item"] + ]) + print(f"\n激活后：{INTENT_KEY} = {restored}") + print(f" 配置项 {len(after['platform_config_item'])} 条、" + f"提示词 {len(after['prompt_template_version'])} 条") + if PROFILE_TOOL not in (restored or []): + print("[失败] 激活后白名单里仍没有画像工具，请人工核查") + return 1 + if len(after["prompt_template_version"]) != len(prompts): + print("[失败] 提示词条数变了，可能发生了静默丢失，请人工核查") + return 1 + print("[OK] 完成") + return 0 + + +sys.exit(asyncio.run(main()))