AI260626/group_fqcd_jr
13
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat(customer-service): 把闲聊提示词补回生效版本，并修掉发布脚本的两个坑

Browse Source 
**背景**：提示词配过（挂在 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 全绿。
This commit is contained in:
lzf_0626
2026-09-11 13:26:43 +08:00
parent e017bcc9bb
commit 8283f6ab69
1 changed files with 87 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/publish_chitchat_prompt.py
+87 -65
View File
@@ -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