fix: reply with standalone ping for meaningless short customer texts

core/pydantic_ai_agent.py

@@ -70,18 +70,29 @@ CASE_LIBRARY_LINK = "https://www.yuque.com/zuowei-dfvpq/kge0in/mynala0g35b8cec5"
 TAOBAO_REPLY_TAILS = ("嗯", "哦", "好的", "嗯咯", "嗯啦")
 
 
-def _ensure_taobao_reply_tail(text: str) -> str:
+def _is_ack_like_customer_text(text: str) -> bool:
-    """淘宝口吻收尾：最终回复结尾带简短口语尾词。"""
+    """客户是否为确认型短句（好的/嗯/收到/ok 等）。"""
-    t = (text or "").strip()
+    s = (text or "").strip().lower()
-    if not t:
+    if not s:
-        return ""
+        return False
-    transfer_keywords = ("TRANSFER_REQUESTED", "[转移会话]", "转移会话")
+    s = s.rstrip("。.!！？~～")
-    if any(k in t for k in transfer_keywords):
+    ack_set = {
-        return t
+        "好", "好的", "嗯", "嗯嗯", "收到", "知道了", "明白了",
-    trimmed = t.rstrip(" \t\r\n。.!！？~～")
+        "ok", "okay", "行", "可以", "好嘞", "好的呢",
-    if any(trimmed.endswith(tail) for tail in TAOBAO_REPLY_TAILS):
+    }
-        return trimmed
+    return s in ack_set
-    return f"{trimmed} 好的"
+
+
+def _is_meaningless_short_text(text: str) -> bool:
+    """识别无意义短句：仅需简短承接，不进入复杂流程。"""
+    s = (text or "").strip().lower().rstrip("。.!！？~～")
+    if not s:
+        return False
+    meaningless = {
+        "好", "好的", "嗯", "嗯嗯", "哦", "哦哦", "收到", "知道了", "明白了",
+        "ok", "okay", "行", "可以", "好嘞", "好的呢", "在吗", "有人吗", "在不在",
+    }
+    return s in meaningless
 
 
 # ========== 数据模型 ==========
@@ -137,8 +148,7 @@ class AgentResponse(BaseModel):
 
     @model_validator(mode="after")
     def _ensure_reply_tail(self):
-        if self.should_reply and self.reply:
+        # 统一在 process_message 中按客户输入决定是否补口语尾词
-            self.reply = _ensure_taobao_reply_tail(self.reply)
         return self
 
 
@@ -1782,6 +1792,13 @@ class CustomerServiceAgent:
         # 获取或创建对话状态
         state = self._get_conversation_state(message.from_id)
 
+        # 无意义短句承接：单独回一句口语，不进入复杂决策
+        if _is_meaningless_short_text(message.msg):
+            ping = random.choice(("嗯咯", "嗯啦", "嗯", "哦"))
+            state.last_reply_at = datetime.now()
+            self._activity_log("agent_ping_reply", customer_id=message.from_id, msg=message.msg, reply=ping)
+            return AgentResponse(reply=ping, should_reply=True, need_transfer=False)
+
         # 冷却期检测：近期已回复 + 纯打招呼 → 静默
         if self._in_cooldown(state, message.msg):
             elapsed = int((datetime.now() - state.last_reply_at).total_seconds())

tests/test_regression_pipeline.py

@@ -412,7 +412,25 @@ class RegressionPipelineTest(unittest.IsolatedAsyncioTestCase):
         self.assertTrue(resp.should_reply)
         self.assertIn("地图", resp.reply)
         self.assertIn("不做", resp.reply)
-        self.assertTrue(resp.reply.endswith(("嗯", "哦", "好的", "嗯咯", "嗯啦")))
+
+    async def test_meaningless_short_text_gets_ping_reply(self):
+        agent = CustomerServiceAgent()
+        msg = CustomerMessage(
+            msg_id="m-meaningless",
+            acc_id="test_shop",
+            msg="好的",
+            from_id=self.customer_id,
+            from_name="t",
+            cy_id=self.customer_id,
+            acc_type="AliWorkbench",
+            msg_type=0,
+            cy_name="t",
+            goods_name="专业找图",
+            goods_order="",
+        )
+        resp = await agent.process_message(msg)
+        self.assertTrue(resp.should_reply)
+        self.assertIn(resp.reply, ("嗯咯", "嗯啦", "嗯", "哦"))
 
     def tearDown(self):
         db.clear_pending_quote_state(self.customer_id)