lzf_0626
8283f6ab69
**背景**：提示词配过（挂在 release 174），但 174 已被取代，而 load_active_prompt 是先定位 active 版本、再按 release_id 查的 —— 于是读不到，Agent 回落到代码默认值。功能看着正常 （_chitchat_prompt 有逐字段兜底），所以一直没人发现，也没有任何告警。 发布脚本原先有两个坑，这次一并修掉： 1. **version 写死为 1**。prompt_template_version 的唯一键是 (prompt_code, version)， 客服那条已经占了 v1，照搬旧行会主键冲突。改为取现有最大值 +1（本次自动分到 v2）。 2. **继承只读 platform_config_item**。改用 ConfigReleaseService.effective_snapshot()， 它覆盖全部三张受管表；并且做**字段名映射**（库的 config_key → API 的 item_key、 value_json 归一化）—— 快照行是库的形状，直接 POST 会 422。 实测： - 发布走路径二（路径一如预期被状态机拒：409 RUN_NOT_CANCELLABLE「只能修改草稿发布版本」） - 新版本 201 继承 9 条配置项、一条没丢；提示词 v2 随之生效 - load_active_prompt 现在返回 release_id=201 / version=2（此前为 None） - 闲聊链路：status=succeeded、intent=chitchat，回答「您好，我是南方科技智能客服， 想了解基金、理财还是账户服务？」—— 简洁、自然引导到业务，符合提示词要求 提示词正文与代码默认值**刻意保持一致**：发布前后行为不变，变的只是"能不能改" （改话术从此要经审核并留痕）。Agent 侧仍保留代码默认值作为兜底。 ruff / mypy(136 文件) / 612 unit+contract / 29 integration 全绿。
243 lines
10 KiB
Python
243 lines
10 KiB
Python
"""把客服闲聊提示词发布为可审核、可回滚的配置版本。
|
||
|
||
业务方选定「提示词走发布配置」而不是写死在代码里：改话术要经过审核并留痕，
|
||
这也符合金融场景对口径变更的要求。
|
||
|
||
**为什么要重发**：提示词配过（挂在 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
|
||
"""
|
||
|
||
import asyncio
|
||
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.config_release_service import ConfigReleaseService
|
||
|
||
ADMIN = "9003"
|
||
PROMPT_CODE = "customer_service_chitchat"
|
||
TASK_TYPE = "chat"
|
||
AGENT_TYPE = "customer_service"
|
||
|
||
# 提示词正文：与 Agent 代码里的默认值保持一致 —— 发布后这份配置即成为话术的唯一来源
|
||
# （Agent 侧仍保留代码默认值作为兜底，两者内容相同，所以发布前后行为一致、
|
||
# 区别只在于"能不能改"）。
|
||
SYSTEM_PROMPT = (
|
||
"你是南方科技的智能客服助手。回应要简短、礼貌，并自然引导用户提出与基金、理财、"
|
||
"账户相关的问题。禁止承诺收益，禁止出现「保本」「稳赚」「无风险」「保证收益」"
|
||
"「预期收益率」「年化收益率」「安全」等表述。"
|
||
)
|
||
USER_PROMPT_TEMPLATE = "用户说：{message}\n请用不超过 40 字回应，并把话题引导到业务上。"
|
||
|
||
|
||
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 active_release_id() -> int | None:
|
||
async with SessionFactory() as session:
|
||
return await ConfigReleaseService(session).active_release_id()
|
||
|
||
|
||
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,
|
||
PromptTemplateVersion.prompt_code == PROMPT_CODE,
|
||
PromptTemplateVersion.agent_type == AGENT_TYPE,
|
||
)
|
||
)
|
||
return row is not None
|
||
|
||
|
||
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
|
||
)
|
||
)
|
||
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(
|
||
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")
|
||
|
||
|
||
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": version,
|
||
"system_prompt": SYSTEM_PROMPT,
|
||
"user_prompt_template": USER_PROMPT_TEMPLATE,
|
||
}
|
||
|
||
|
||
async def main() -> int:
|
||
if await prompt_already_in_active_release():
|
||
print("当前生效版本已包含该提示词，无需发布")
|
||
return 0
|
||
|
||
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(
|
||
transport=httpx.ASGITransport(app=app), base_url="http://test", timeout=60
|
||
) as client:
|
||
# 路径一：直接挂到当前生效版本
|
||
direct = await post(
|
||
client, "/api/v1/admin/prompt-templates", auth=auth,
|
||
payload=prompt_payload(release, version),
|
||
)
|
||
print(f"直接挂到生效版本 {release}：{direct.status_code} {direct.text[:160]}")
|
||
if direct.status_code in (200, 201):
|
||
print("完成：提示词已挂到当前生效版本")
|
||
return 0
|
||
|
||
# 路径二：新建发布版本，继承现有配置项后追加提示词
|
||
print("生效版本不可追加，改为新建发布版本并继承现有配置项")
|
||
created = await post(client, "/api/v1/admin/config-releases", auth=auth, payload={
|
||
"release_no": f"cs-prompt-{uuid.uuid4().hex[:12]}",
|
||
"title": "客服闲聊提示词",
|
||
"change_summary": "发布客服闲聊提示词，并继承现有全部配置项",
|
||
})
|
||
if created.status_code != 201:
|
||
print(f" 创建发布版本失败：{created.status_code} {created.text[:200]}")
|
||
return 1
|
||
new_release = int(created.json()["data"]["id"])
|
||
print(f" 新发布版本 id={new_release}")
|
||
|
||
for item in items:
|
||
response = await post(
|
||
client, f"/api/v1/admin/config-releases/{new_release}/platform-config-items",
|
||
auth=auth, payload=item,
|
||
)
|
||
if response.status_code != 201:
|
||
print(f" 继承 {item['item_key']} 失败：{response.text[:160]}")
|
||
return 1
|
||
print(f" 已继承 {len(items)} 条配置项")
|
||
|
||
added = await post(
|
||
client, "/api/v1/admin/prompt-templates", auth=auth,
|
||
payload=prompt_payload(new_release, version),
|
||
)
|
||
print(f" 添加提示词 v{version}：{added.status_code} {added.text[:160]}")
|
||
if added.status_code not in (200, 201):
|
||
return 1
|
||
|
||
base = f"/api/v1/admin/config-releases/{new_release}"
|
||
await post(client, f"{base}/validations", auth=auth, payload={},
|
||
if_match=await etag_of(client, base, auth))
|
||
await post(client, f"{base}/reviews", auth=auth,
|
||
payload={"decision": "approved", "comment": "客服闲聊提示词"},
|
||
if_match=await etag_of(client, base, auth))
|
||
activated = await post(client, f"{base}/activations", auth=auth, payload={},
|
||
if_match=await etag_of(client, base, auth))
|
||
print(f" 激活：{activated.status_code}")
|
||
if activated.status_code not in (200, 201):
|
||
print(f" 失败：{activated.text[:200]}")
|
||
return 1
|
||
print(f"完成：新发布版本 {new_release} 已生效，提示词 v{version} 随之生效")
|
||
return 0
|
||
|
||
|
||
sys.exit(asyncio.run(main()))
|