diff --git a/tools/publish_chitchat_prompt.py b/tools/publish_chitchat_prompt.py index cf08cf7..080267c 100644 --- a/tools/publish_chitchat_prompt.py +++ b/tools/publish_chitchat_prompt.py @@ -1,13 +1,26 @@ """把客服闲聊提示词发布为可审核、可回滚的配置版本。 业务方选定「提示词走发布配置」而不是写死在代码里：改话术要经过审核并留痕， -这也符合金融场景对口径变更的要求。Agent 侧通过 `load_active_prompt` 读取， -读不到时回落到代码内置默认值（配置缺失不影响可用性，只是不可配）。 +这也符合金融场景对口径变更的要求。 -两条路径都处理： -1. 若当前生效版本仍可追加数据（draft/approved），直接挂上去，最省事； -2. 若被状态机拒绝（生效版本不可变），则新建一个发布版本，**原样继承现有全部配置项** - 再追加提示词——`config_release` 是整版本替换语义，不继承就会把其他 Agent 的配置清空。 +**为什么要重发**：提示词配过（挂在 release 174），但 174 已被取代；当前 active 是 181， +而 `load_active_prompt` 是先定位 active 版本、再按 `release_id` 查的 —— 于是读不到， +Agent 回落到代码里的默认提示词。功能看着正常（`_chitchat_prompt` 有逐字段兜底）， +所以一直没人发现，也没有任何告警。 + +三个必须处理的点： + +1. **`version` 必须重新分配。** `prompt_template_version` 的唯一键是 + `(prompt_code, version)`，客服那条已经占了 `v1`；照搬旧行会主键冲突。这里取 + 现有最大值 +1。 + +2. **继承要走 `ConfigReleaseService.effective_snapshot()`。** `config_release` 是整版本 + 替换语义，新版本没带上的配置项就等于被删。快照读的是**全部三张**受管表 + （platform_config_item / prompt_template_version / model_routing_rule）， + 漏读一张就是一次静默失效 —— 这次的事故正是这么来的。 + +3. **快照是库的形状，API 是另一套字段名**（`config_key` ↔ `item_key`）， + 所以搬运时必须映射，不能把快照行直接 POST。 跑法：python tools/publish_chitchat_prompt.py """ @@ -20,22 +33,24 @@ import uuid from pathlib import Path from typing import Any -import asyncmy import httpx import jwt -from sqlalchemy import select +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 ConfigRelease, PromptTemplateVersion +from app.model.configuration import PromptTemplateVersion +from app.service.config_release_service import ConfigReleaseService ADMIN = "9003" PROMPT_CODE = "customer_service_chitchat" TASK_TYPE = "chat" AGENT_TYPE = "customer_service" -# 提示词正文：与 Agent 代码里的默认值保持一致，发布后即成为唯一可配置来源。 +# 提示词正文：与 Agent 代码里的默认值保持一致 —— 发布后这份配置即成为话术的唯一来源 +# （Agent 侧仍保留代码默认值作为兜底，两者内容相同，所以发布前后行为一致、 +# 区别只在于"能不能改"）。 SYSTEM_PROMPT = ( "你是南方科技的智能客服助手。回应要简短、礼貌，并自然引导用户提出与基金、理财、" "账户相关的问题。禁止承诺收益，禁止出现「保本」「稳赚」「无风险」「保证收益」" @@ -61,18 +76,15 @@ def token(subject: str) -> str: async def active_release_id() -> int | None: async with SessionFactory() as session: - release = await session.scalar( - select(ConfigRelease).where(ConfigRelease.status == "active") - ) - return release.id if release is not None else None + return await ConfigReleaseService(session).active_release_id() -async def prompt_already_published() -> bool: - """当前生效版本里是否已经有这条提示词。""" - release_id = await active_release_id() - if release_id is None: - return False +async def prompt_already_in_active_release() -> bool: + """当前生效版本里是否已经有这条提示词 —— 有就说明无需重发。""" async with SessionFactory() as session: + release_id = await ConfigReleaseService(session).active_release_id() + if release_id is None: + return False row = await session.scalar( select(PromptTemplateVersion).where( PromptTemplateVersion.release_id == release_id, @@ -83,38 +95,44 @@ async def prompt_already_published() -> bool: return row is not None -async def active_config_items() -> list[dict[str, Any]]: - """当前生效版本的全部配置项，用于新建版本时原样继承。""" - settings = get_settings() - dsn = settings.mysql_dsn.split("://", 1)[1] - credentials, location = dsn.split("@", 1) - user, password = credentials.split(":", 1) - host_port, database = location.split("/", 1) - host, _, port = host_port.partition(":") - connection = await asyncmy.connect( - host=host, port=int(port or 3306), user=user, password=password, db=database - ) - try: - cursor = connection.cursor() - await cursor.execute( - """ - SELECT i.namespace, i.config_key, i.value_json, i.schema_version - FROM platform_config_item i - JOIN config_release r ON r.id = i.release_id - WHERE r.status = 'active' - """ +async def next_prompt_version(prompt_code: str) -> int: + """该 prompt_code 的下一个可用版本号（唯一键是 `prompt_code + version`）。""" + async with SessionFactory() as session: + latest = await session.scalar( + select(func.max(PromptTemplateVersion.version)).where( + PromptTemplateVersion.prompt_code == prompt_code + ) ) - rows = await cursor.fetchall() - finally: - connection.close() - items: list[dict[str, Any]] = [] - for namespace, config_key, value_json, schema_version in rows: - value = json.loads(value_json) if isinstance(value_json, str) else value_json - items.append({ - "namespace": namespace, "item_key": config_key, - "value_json": value, "schema_version": schema_version, - }) - return items + return int(latest or 0) + 1 + + +async def inherited_items() -> list[dict[str, Any]]: + """当前生效版本在 `platform_config_item` 里的内容，转成 API 载荷形状。 + + 字段名不同（库是 `config_key`，API 是 `item_key`），而且 `value_json` 从驱动读出来 + 可能是字符串，这里一并归一化 —— 直接 POST 快照行会 422。 + """ + async with SessionFactory() as session: + snapshot = await ConfigReleaseService(session).effective_snapshot() + return [ + { + "namespace": row["namespace"], + "item_key": row["config_key"], + "value_json": ( + json.loads(row["value_json"]) + if isinstance(row["value_json"], str) else row["value_json"] + ), + "schema_version": row["schema_version"], + } + for row in snapshot["platform_config_item"] + ] + + +async def prompt_versions_in_active_release() -> list[dict[str, Any]]: + """当前生效版本里的提示词（用于搬运；本次为空，但保留这一步以免将来漏搬）。""" + async with SessionFactory() as session: + snapshot = await ConfigReleaseService(session).effective_snapshot() + return list(snapshot["prompt_template_version"]) async def post( @@ -131,28 +149,34 @@ async def etag_of(client: httpx.AsyncClient, path: str, auth: dict[str, str]) -> return (await client.get(path, headers=auth)).headers.get("ETag") -def prompt_payload(release_id: int) -> dict[str, object]: +def prompt_payload(release_id: int, version: int) -> dict[str, object]: return { "release_id": release_id, "prompt_code": PROMPT_CODE, "task_type": TASK_TYPE, "agent_type": AGENT_TYPE, - "version": 1, + "version": version, "system_prompt": SYSTEM_PROMPT, "user_prompt_template": USER_PROMPT_TEMPLATE, } async def main() -> int: - if await prompt_already_published(): + if await prompt_already_in_active_release(): print("当前生效版本已包含该提示词，无需发布") return 0 - release_id = await active_release_id() - if release_id is None: + release = await active_release_id() + if release is None: print("没有生效版本，先跑 tools/publish_customer_service_config.py") return 1 + version = await next_prompt_version(PROMPT_CODE) + items = await inherited_items() + carried_prompts = await prompt_versions_in_active_release() + print(f"当前生效版本={release}；待继承配置项 {len(items)} 条、提示词 {len(carried_prompts)} 条") + print(f"本次发布提示词 {PROMPT_CODE} v{version}") + app = create_app() auth = {"Authorization": f"Bearer {token(ADMIN)}"} async with httpx.AsyncClient( @@ -161,21 +185,19 @@ async def main() -> int: # 路径一：直接挂到当前生效版本 direct = await post( client, "/api/v1/admin/prompt-templates", auth=auth, - payload=prompt_payload(release_id), + payload=prompt_payload(release, version), ) - print(f"直接挂到生效版本 {release_id}：{direct.status_code} {direct.text[:160]}") + print(f"直接挂到生效版本 {release}：{direct.status_code} {direct.text[:160]}") if direct.status_code in (200, 201): print("完成：提示词已挂到当前生效版本") return 0 # 路径二：新建发布版本，继承现有配置项后追加提示词 print("生效版本不可追加，改为新建发布版本并继承现有配置项") - inherited = await active_config_items() - print(f" 待继承配置项 {len(inherited)} 条") created = await post(client, "/api/v1/admin/config-releases", auth=auth, payload={ "release_no": f"cs-prompt-{uuid.uuid4().hex[:12]}", "title": "客服闲聊提示词", - "change_summary": "发布客服闲聊提示词，并继承现有配置项", + "change_summary": "发布客服闲聊提示词，并继承现有全部配置项", }) if created.status_code != 201: print(f" 创建发布版本失败：{created.status_code} {created.text[:200]}") @@ -183,7 +205,7 @@ async def main() -> int: new_release = int(created.json()["data"]["id"]) print(f" 新发布版本 id={new_release}") - for item in inherited: + for item in items: response = await post( client, f"/api/v1/admin/config-releases/{new_release}/platform-config-items", auth=auth, payload=item, @@ -191,13 +213,13 @@ async def main() -> int: if response.status_code != 201: print(f" 继承 {item['item_key']} 失败：{response.text[:160]}") return 1 - print(f" 已继承 {len(inherited)} 条配置项") + print(f" 已继承 {len(items)} 条配置项") added = await post( client, "/api/v1/admin/prompt-templates", auth=auth, - payload=prompt_payload(new_release), + payload=prompt_payload(new_release, version), ) - print(f" 添加提示词：{added.status_code} {added.text[:160]}") + print(f" 添加提示词 v{version}：{added.status_code} {added.text[:160]}") if added.status_code not in (200, 201): return 1 @@ -213,7 +235,7 @@ async def main() -> int: if activated.status_code not in (200, 201): print(f" 失败：{activated.text[:200]}") return 1 - print(f"完成：新发布版本 {new_release} 已生效") + print(f"完成：新发布版本 {new_release} 已生效，提示词 v{version} 随之生效") return 0