107 lines
4.3 KiB
Python
107 lines
4.3 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""订单回复事实校验。
|
|
|
|
模型带着订单工具作答之后，回复里出现的单号、订单号和物流 / 审核状态必须能在
|
|
工具返回的真实结果里找到。找不到就说明是编的：整条回复换成“正在核实”的安全话术，
|
|
而不是把一个假单号或假签收状态发给客户。
|
|
|
|
这一步是提示词之外的硬闸。提示词里写了“严禁编造单号”，模型照样会编。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from typing import Any, Iterable
|
|
|
|
SAFE_FALLBACK_REPLY = "这边先帮您去后台核实一下订单和物流情况，核实清楚马上回复您。"
|
|
|
|
# 快递单号 / 订单号：字母前缀 + 长数字，纯长数字，或已脱敏的 SF****7899。
|
|
_NUMBER_RE = re.compile(r"(?<![A-Za-z0-9])(?:[A-Za-z]{1,6}\*{2,}\d{3,8}|[A-Za-z]{0,6}\d{8,24})(?![A-Za-z0-9])")
|
|
|
|
# 只有真实查询结果才能支撑的状态性说法。
|
|
STATUS_CLAIMS: tuple[str, ...] = (
|
|
"已签收", "签收了", "运输中", "派送中", "派件中", "正在派送", "已揽收", "揽收了",
|
|
"已发货", "已发出", "已寄出", "发货了", "寄出了", "发出了", "已到达", "到达了",
|
|
"中转", "已完成", "已取消", "已退款", "审核通过", "已通过", "已驳回", "未通过",
|
|
"待审核", "审核中", "已支付", "待发货", "已制药", "制药中",
|
|
)
|
|
|
|
|
|
def _successful_results(tool_results: Iterable[str]) -> list[str]:
|
|
"""只有 ok 不为 False 的工具结果才算“查到了”。失败结果里的 message 不是事实。"""
|
|
texts = []
|
|
for raw in tool_results or []:
|
|
text = str(raw or "").strip()
|
|
if not text:
|
|
continue
|
|
try:
|
|
payload = json.loads(text)
|
|
except (TypeError, ValueError):
|
|
payload = None
|
|
if isinstance(payload, dict) and payload.get("ok") is False:
|
|
continue
|
|
texts.append(text)
|
|
return texts
|
|
|
|
|
|
def _tool_numbers(texts: Iterable[str]) -> set[str]:
|
|
numbers: set[str] = set()
|
|
for text in texts:
|
|
for match in _NUMBER_RE.findall(text):
|
|
numbers.add(match.upper())
|
|
numbers.add(re.sub(r"\D", "", match))
|
|
numbers.discard("")
|
|
return numbers
|
|
|
|
|
|
def find_unsupported_numbers(reply: str, tool_results: Iterable[str], customer_text: str = "") -> list[str]:
|
|
"""回复里出现、但工具结果和客户原话里都没有的单号 / 订单号。"""
|
|
known = _tool_numbers(_successful_results(tool_results))
|
|
known |= _tool_numbers([str(customer_text or "")])
|
|
unsupported = []
|
|
for match in _NUMBER_RE.findall(str(reply or "")):
|
|
digits = re.sub(r"\D", "", match)
|
|
# 纯 8~9 位数字大多是日期或金额（20260918），不当单号处理。
|
|
if match.isdigit() and len(match) < 10:
|
|
continue
|
|
if match.upper() in known or (digits and digits in known):
|
|
continue
|
|
unsupported.append(match)
|
|
return unsupported
|
|
|
|
|
|
def find_unsupported_claims(reply: str, tool_results: Iterable[str], customer_text: str = "") -> list[str]:
|
|
"""回复里的状态性说法，若真实结果和客户原话都没提到，视为编造。"""
|
|
body = str(reply or "")
|
|
if not body:
|
|
return []
|
|
evidence = "\n".join(_successful_results(tool_results)) + "\n" + str(customer_text or "")
|
|
unsupported = []
|
|
for claim in STATUS_CLAIMS:
|
|
if claim in body and claim not in evidence:
|
|
# “已发货”出现在证据的“待发货”里不算支撑，但那种情况下“待发货”会先被匹配。
|
|
unsupported.append(claim)
|
|
return unsupported
|
|
|
|
|
|
def enforce_order_facts(
|
|
reply: str,
|
|
tool_results: Iterable[str],
|
|
*,
|
|
customer_text: str = "",
|
|
fallback: str = SAFE_FALLBACK_REPLY,
|
|
) -> tuple[str, list[str]]:
|
|
"""返回 (最终回复, 违规原因列表)。无违规时原样返回。"""
|
|
results = list(tool_results or [])
|
|
reasons: list[str] = []
|
|
numbers = find_unsupported_numbers(reply, results, customer_text)
|
|
if numbers:
|
|
reasons.append("单号未经查询: " + ", ".join(numbers[:3]))
|
|
claims = find_unsupported_claims(reply, results, customer_text)
|
|
if claims:
|
|
reasons.append("状态未经查询: " + ", ".join(claims[:3]))
|
|
if reasons:
|
|
return fallback, reasons
|
|
return reply, []
|