r"""合规层种子数据：7 个零容忍负面词（+4 变体）与 6 类固定话术。 背景：`docs/02` §10.2/§10.3 要求这些数据，但迁移只建表、没有 seed。而 `app/service/agent/governance.py` L62-65 的取数条件是 `status='active' AND reviewer_id IS NOT NULL AND reviewed_at IS NOT NULL`， 两张表 0 行就意味着一**条规则都查不到**：客服 Agent 说"这只基金稳赚"不会被拦， 免责声明也不会出现在任何回复里。合规过滤是静默失效的——不报错，只是不生效。 取值来源（不臆造，均以真实 DDL 为准）： - `agent_negative_word.match_type` CHECK：`exact` / `contains` / `regex`， 这里只用 `contains`；`regex` 被 governance.py L75-77 显式拒绝 （Python `re` 无执行超时，底座不跑未经约束的管理员正则）。 - `agent_negative_word.severity` CHECK：`warn` / `regenerate` / `block`， 零容忍词一律 `block`。 - `agent_negative_word.category` **无 CHECK 约束**，按语义取值（《理财销售办法》 的口径：误导性 / 绝对化 / 承诺 / 收益）。 - `agent_reply_template.scene` CHECK：`disclaimer` / `low_confidence` / `compliance_block` / `transfer` / `model_failure` / `system_busy` / `clarification`。 - `agent_reply_template.active_key` 是 **生成列**（`status='active'` 时等于 `template_code:locale`，否则 NULL）：不能写、也不该写，由库自己算。 它上面有唯一键 `uk_reply_template_active_one`，所以同一 template_code 只能有一条 active。 幂等：按 `rule_code`（唯一键 `uk_negative_rule_code`）与 `template_code`+`version`（唯一键 `uk_reply_template_version`）upsert，可重复执行； 重复执行不会新增行，只会把已存在的行重新置为 active 并刷新审核信息。 执行：`.\.venv\Scripts\python.exe tools\seed_compliance_baseline.py`（可重复执行） """ from __future__ import annotations import asyncio import sys from datetime import UTC, datetime from pathlib import Path ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from sqlalchemy import text # noqa: E402 from app.infrastructure.db import SessionFactory # noqa: E402 #: 审核人与创建人：`sys_user.id=9003`（T-ADMIN，由 tools/seed_test_rbac.py 种入）。 #: `created_by`/`reviewer_id` 都有指向 `sys_user` 的外键，写不存在的用户会直接失败。 ADMIN_ID = 9003 #: 只对客服 Agent 生效，避免影响其他 Agent 的表达自由度。 APPLICABLE_AGENTS = '["customer_service"]' #: 命中后统一的合规拒答话术（就是本脚本要种的 6 条之一，所以话术先插）。 SAFE_REPLY_TEMPLATE_CODE = "TPL_COMPLIANCE_REFUSAL" #: `docs/02` §10.2 逐字要求的 7 个零容忍词。前 7 条规则必须恰好覆盖它们。 REQUIRED_ZERO_TOLERANCE_WORDS = ( "保本", "稳赚", "无风险", "保证收益", "预期收益率", "年化收益率", "安全", ) #: (rule_code, word_pattern, category, reason) #: 前 7 条是 `docs/02` §10.2 逐字要求的零容忍词，一个都不能少； #: 后 4 条是专项设计 §6.5 评审建议的变体（换个说法绕过关键词的常见写法）。 NEGATIVE_RULES: tuple[tuple[str, str, str, str], ...] = ( ("NEG-001", "保本", "guarantee", "资管新规后非保本"), ("NEG-002", "稳赚", "misleading", "误导性表述"), ("NEG-003", "无风险", "absolute", "绝对化表述"), ("NEG-004", "保证收益", "guarantee", "禁止刚兑承诺"), ("NEG-005", "预期收益率", "yield_claim", "《理财销售办法》明文禁止"), ("NEG-006", "年化收益率", "yield_claim", "监管处罚点名措辞"), ("NEG-007", "安全", "absolute", "绝对化表述"), ("NEG-008", "零风险", "absolute", "绝对化表述变体"), ("NEG-009", "稳赚不赔", "misleading", "误导性表述变体"), ("NEG-010", "躺着赚", "misleading", "误导性表述变体"), ("NEG-011", "坐享收益", "misleading", "误导性表述变体"), ) #: (template_code, scene, title, content_text) #: scene 取自 `chk_template_scene` 的 7 个合法值里的 6 个，覆盖 6 类固定话术。 #: TPL_AI_NOTICE 用 `clarification`（AI 生成标识属主动澄清披露）； #: TPL_COMPLIANCE_REFUSAL 用 `compliance_block`（合规拦截后替换的安全话术）。 #: #: ⚠️ **话术文本本身不得命中 NEGATIVE_RULES 里的任何词**（`verify()` 会强制校验）。 #: 治理层（governance.py L127-128）对 `contains` 是**朴素子串匹配**，没有否定式豁免： #: 旧文案里的「非保本」会命中「保本」规则，形成"合规话术被自己的规则拦截"的自绊， #: Task 2 接入话术后就会变成"替换文本又被过滤"的循环。因此合规语义必须用 #: **不含禁用字面**的说法表达（"不承诺本金不受损失" 而不是 "非保本"）。 REPLY_TEMPLATES: tuple[tuple[str, str, str, str], ...] = ( ("TPL_COMPLIANCE_REFUSAL", "compliance_block", "合规拒答", "根据监管要求，我不能对收益做出任何承诺，也不承诺本金不受损失。" "本产品收益可能为负，请以产品说明书为准。"), ("TPL_DISCLAIMER", "disclaimer", "固定免责声明", "本内容仅为投资分析参考，不构成任何直接投资建议，不构成对任何产品的收益承诺，" "据此操作风险自负，请谨慎对待。"), ("TPL_AI_NOTICE", "clarification", "AI 生成标识", "本回答由 AI 生成，仅供参考。"), ("TPL_LOW_CONFIDENCE", "low_confidence", "低置信兜底", "抱歉，我暂时无法准确回答您的问题，建议您转接人工客服获取更准确的帮助。"), ("TPL_TRANSFER_HUMAN", "transfer", "转人工提示", "已为您转接人工客服，工作时间为工作日 09:00-18:00，客服电话 15936583816。"), ("TPL_SYSTEM_BUSY", "system_busy", "系统繁忙/模型故障", "系统繁忙，暂时无法回答，请稍后重试或联系人工客服 15936583816。"), ) #: template_code -> content_text，供 `verify()` 做自绊检查（须定义在 REPLY_TEMPLATES 之后）。 CONTENT_BY_TEMPLATE_CODE = {code: content for code, _scene, _title, content in REPLY_TEMPLATES} INSERT_NEGATIVE_RULE = text( "INSERT INTO agent_negative_word" " (rule_code, word_pattern, match_type, category, severity, applicable_agents," " safe_reply_template_code, status, version, created_by, reviewer_id, reviewed_at," " created_at, updated_at)" " VALUES (:rule_code, :word, 'contains', :category, 'block', :agents," " :safe_code, 'active', 1, :admin, :admin, :now, :now, :now)" " ON DUPLICATE KEY UPDATE word_pattern=:word, match_type='contains', category=:category," " severity='block', applicable_agents=:agents, safe_reply_template_code=:safe_code," " status='active', reviewer_id=:admin, reviewed_at=:now, updated_at=:now" ) INSERT_REPLY_TEMPLATE = text( "INSERT INTO agent_reply_template" " (template_code, scene, title, content_text, variables, locale, version, status," " created_by, reviewer_id, reviewed_at, created_at, updated_at)" " VALUES (:code, :scene, :title, :content, '{}', 'zh-CN', 1, 'active'," " :admin, :admin, :now, :now, :now)" " ON DUPLICATE KEY UPDATE scene=:scene, title=:title, content_text=:content," " status='active', reviewer_id=:admin, reviewed_at=:now, updated_at=:now" ) async def seed() -> None: """幂等落库。话术先于负面词插入：负面词会引用话术 code。""" now = datetime.now(UTC).replace(tzinfo=None) async with SessionFactory() as session, session.begin(): for code, scene, title, content in REPLY_TEMPLATES: await session.execute(INSERT_REPLY_TEMPLATE, { "code": code, "scene": scene, "title": title, "content": content, "admin": ADMIN_ID, "now": now, }) for rule_code, word, category, _reason in NEGATIVE_RULES: await session.execute(INSERT_NEGATIVE_RULE, { "rule_code": rule_code, "word": word, "category": category, "agents": APPLICABLE_AGENTS, "safe_code": SAFE_REPLY_TEMPLATE_CODE, "admin": ADMIN_ID, "now": now, }) async def verify() -> None: """按 governance.py 的**同一口径**复查，确认真的能被查到。""" async with SessionFactory() as session: rules = (await session.execute(text( "SELECT rule_code, word_pattern, category, severity, match_type" " FROM agent_negative_word" " WHERE status='active' AND reviewer_id IS NOT NULL AND reviewed_at IS NOT NULL" " ORDER BY rule_code" ))).all() templates = (await session.execute(text( "SELECT template_code, scene, status FROM agent_reply_template" " WHERE status='active' AND reviewer_id IS NOT NULL AND reviewed_at IS NOT NULL" " ORDER BY template_code" ))).all() for code, word, category, severity, match_type in rules: print(f" rule {code} {word!r} category={category} severity={severity}" f" match_type={match_type}") for code, scene, status in templates: print(f" tpl {code} scene={scene} status={status}") # 种子层的自绊检查：话术不得命中自己这套规则。治理层是朴素子串匹配， # 没有否定式豁免——「非保本」这类写法会命中「保本」规则。 # 在种子阶段就拦住，比等 Task 2 接入话术后再发现在替换结果里循环排查便宜得多。 self_trips = [ (code, word) for code, content in CONTENT_BY_TEMPLATE_CODE.items() for word in (word for _, word, *_ in rules) if word in content ] if self_trips: raise SystemExit(f"话术命中了自己的禁用词（自绊）：{self_trips}") patterns = {word for _, word, *_ in rules} missing = { word for word in REQUIRED_ZERO_TOLERANCE_WORDS if not any(word in pattern for pattern in patterns) } if missing: raise SystemExit(f"零容忍词未生效：{sorted(missing)}") scene_count = len({scene for _, scene, _ in templates}) if scene_count < 6: raise SystemExit(f"生效话术场景不足 6 类：{scene_count}") print(f"verified: {len(rules)} active rules (含 7 个零容忍词), " f"{len(templates)} active templates covering {scene_count} scenes") async def main() -> None: await seed() print(f"seeded {len(NEGATIVE_RULES)} negative rules, {len(REPLY_TEMPLATES)} reply templates") await verify() if __name__ == "__main__": asyncio.run(main())