97 lines
3.8 KiB
Python
97 lines
3.8 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""订单 / 快递意图识别。
|
||
|
||
只有客户在问订单、快递、物流、审核这类问题时，才把当前会话上下文交给订单 MCP。
|
||
否则每条消息都拉起一个只读订单服务子进程，既浪费，也让模型在闲聊时多了一个
|
||
可以误调用的工具。
|
||
|
||
关键词可以在 ``ai_config.AI_ORDER_INTENT_KEYWORDS`` 覆盖；留空则使用内置列表。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from typing import Any, Iterable
|
||
|
||
DEFAULT_ORDER_KEYWORDS: tuple[str, ...] = (
|
||
"快递", "物流", "订单", "单号", "运单", "快件", "包裹",
|
||
"发货", "发出", "寄出", "寄了", "寄过", "邮寄", "发了吗", "发没发",
|
||
"到哪", "到了吗", "到货", "到没到", "什么时候到", "啥时候到", "几天到",
|
||
"什么时候发", "啥时候发", "什么时候寄", "啥时候寄",
|
||
"签收", "派送", "派件", "配送", "送到", "收到药", "药到", "药寄", "药发",
|
||
"审核", "审核通过", "审没审", "过审", "支付审核", "处方审核",
|
||
"顺丰", "京东", "圆通", "中通", "申通", "韵达", "邮政", "ems", "极兔", "德邦",
|
||
"退款", "退货", "拒收", "改地址", "换地址", "地址写错",
|
||
)
|
||
|
||
# 客户在被要求“确认是哪一笔”之后的短回复，本身不含订单关键词，但仍属于订单话题。
|
||
_FOLLOW_UP_RE = re.compile(
|
||
r"^(第?[一二三两1-9]\s*(个|笔|单|次)?|最近(的|那)?(一)?(笔|个|单)?|上(一|次)(的|那)?(笔|个|单)?|"
|
||
r"(那|这)(笔|个|单)|都|全部|两个都|\d{1,2}月\d{1,2}(号|日)?(的|那)?(笔|个|单)?|"
|
||
r"[A-Za-z]{0,4}\d{6,}|(是|对|嗯|好)的?|不是|另一(个|笔))[。！!？?～~，,\s]*$"
|
||
)
|
||
|
||
|
||
def _keywords() -> tuple[str, ...]:
|
||
try:
|
||
import ai_config
|
||
raw = getattr(ai_config, "AI_ORDER_INTENT_KEYWORDS", None)
|
||
except Exception:
|
||
raw = None
|
||
if isinstance(raw, (list, tuple)):
|
||
cleaned = tuple(str(item).strip().lower() for item in raw if str(item).strip())
|
||
if cleaned:
|
||
return cleaned
|
||
return DEFAULT_ORDER_KEYWORDS
|
||
|
||
|
||
def _normalize(text: Any) -> str:
|
||
return re.sub(r"\s+", "", str(text or "")).lower()
|
||
|
||
|
||
def matches_order_keywords(text: Any, keywords: Iterable[str] | None = None) -> bool:
|
||
body = _normalize(text)
|
||
if not body:
|
||
return False
|
||
return any(word and word in body for word in (keywords or _keywords()))
|
||
|
||
|
||
def _history_texts(history: Any, limit: int = 4) -> list[str]:
|
||
if not isinstance(history, list):
|
||
return []
|
||
texts = []
|
||
for item in history[-limit:]:
|
||
if isinstance(item, dict):
|
||
texts.append(str(item.get("content") or ""))
|
||
else:
|
||
texts.append(str(item or ""))
|
||
return texts
|
||
|
||
|
||
def is_order_intent(customer_text: Any, history: Any = None) -> bool:
|
||
"""当前这轮是否属于订单 / 快递话题。
|
||
|
||
直接命中关键词即为订单意图；否则只有在最近几轮里刚聊过订单、且客户这轮是
|
||
“第一个 / 9 月 18 号那个 / 单号”这类短回复时才算延续订单话题。
|
||
"""
|
||
text = str(customer_text or "").strip()
|
||
keywords = _keywords()
|
||
if matches_order_keywords(text, keywords):
|
||
return True
|
||
if not text or len(_normalize(text)) > 24:
|
||
return False
|
||
if not _FOLLOW_UP_RE.match(text):
|
||
return False
|
||
return any(matches_order_keywords(item, keywords) for item in _history_texts(history))
|
||
|
||
|
||
def filter_context(
|
||
mcp_context: dict[str, Any] | None, customer_text: Any, history: Any = None
|
||
) -> dict[str, Any]:
|
||
"""只在订单意图成立时保留会话上下文，否则返回空 dict（不挂载订单 MCP）。"""
|
||
if not mcp_context or not isinstance(mcp_context, dict):
|
||
return {}
|
||
if not is_order_intent(customer_text, history):
|
||
return {}
|
||
return dict(mcp_context)
|