feat: enforce activity logs and tighten sizing/map reply policies

core/websocket_client.py

@@ -130,7 +130,20 @@ class QingjianAPIClient:
         if workflow:
             workflow.register_send_callback(self._workflow_send)
             workflow.register_agent_notify_callback(self._workflow_agent_notify)
-        
+
+    def _activity_log(self, event: str, **kwargs):
+        """统一活动日志，便于按 event 检索完整链路。"""
+        safe = {}
+        for k, v in kwargs.items():
+            if isinstance(v, str):
+                safe[k] = v[:200]
+            else:
+                safe[k] = v
+        try:
+            logger.info(f"[ACTIVITY] event={event} data={json.dumps(safe, ensure_ascii=False)}")
+        except Exception:
+            logger.info(f"[ACTIVITY] event={event} data={safe}")
+
 
     async def connect(self):
         """连接WebSocket服务器"""
@@ -417,12 +430,19 @@ class QingjianAPIClient:
     async def _debounce_agent_reply(self, data: dict):
         """
         消息防抖：同一客户在 _DEBOUNCE_SECONDS 内的连续消息合并后再处理。
-        订单通知、图片URL、付款相关消息不走防抖，立即处理。
+        订单通知、付款相关消息不走防抖，立即处理。
         """
         msg_body = data.get('msg', '')
         # 以下情况跳过防抖，立即处理（后台执行，不阻塞接收循环）
         immediate_keywords = ["买家已付款", "已付款", "[系统订单信息]"]
-        if any(kw in msg_body for kw in immediate_keywords) or self._msg_has_image_url(msg_body):
+        if any(kw in msg_body for kw in immediate_keywords):
+            self._activity_log(
+                "debounce_bypass_immediate",
+                acc_id=data.get("acc_id", ""),
+                customer_id=data.get("from_id", ""),
+                reason="payment_or_order",
+                msg=msg_body,
+            )
             self._fire_and_forget(self._agent_reply_serialized(data))
             return
 
@@ -432,6 +452,12 @@ class QingjianAPIClient:
         if key not in self._pending_msgs:
             self._pending_msgs[key] = []
         self._pending_msgs[key].append(msg_body)
+        self._activity_log(
+            "debounce_enqueue",
+            key=key,
+            queue_size=len(self._pending_msgs[key]),
+            msg=msg_body,
+        )
 
         # 取消上一个等待任务（如果有）
         old_task = self._debounce_tasks.get(key)
@@ -451,6 +477,12 @@ class QingjianAPIClient:
             else:
                 merged_msg = "、".join(m for m in msgs if m.strip())
                 print(f"[{self.get_time()}] 防抖合并 {len(msgs)} 条消息: {merged_msg[:60]}")
+            self._activity_log(
+                "debounce_flush",
+                key=capture_key,
+                merged_count=len(msgs),
+                merged_msg=merged_msg,
+            )
             merged_data = dict(capture_data)
             merged_data['msg'] = merged_msg
             await self._agent_reply_serialized(merged_data)
@@ -513,7 +545,11 @@ class QingjianAPIClient:
         if intent == "打招呼":
             low, high = 1.0, min(3.0, base)
         elif intent in ("询价", "砍价"):
-            low, high = 2.0, min(5.0, base)
+            # 询价先略等一会，给客户补发图片/需求的窗口，减少机械两连回
+            low, high = 4.0, min(7.0, max(base, 7.0))
+        elif intent == "image":
+            # 文本里直接贴图链接：短等合并上下文，避免和上一条询价并发
+            low, high = 2.2, 4.2
         elif intent in ("修改", "批量"):
             low, high = max(3.0, base * 0.65), min(18.0, base + 2.0)
         elif intent == "转接":
@@ -825,6 +861,12 @@ class QingjianAPIClient:
                     return
 
             logger.info("Agent 正在处理消息...")
+            self._activity_log(
+                "agent_process_start",
+                acc_id=data.get("acc_id", ""),
+                customer_id=data.get("from_id", ""),
+                msg=msg_text,
+            )
             
             # 调用 Agent
             response = await self.agent.process_message(customer_msg)
@@ -832,6 +874,12 @@ class QingjianAPIClient:
             # 检查是否需要转接人工
             if response.need_transfer:
                 logger.info("Agent 决定转接人工")
+                self._activity_log(
+                    "agent_transfer",
+                    acc_id=data.get("acc_id", ""),
+                    customer_id=data.get("from_id", ""),
+                    transfer_msg=response.transfer_msg,
+                )
                 await self.transfer_to_human(data, response.transfer_msg)
                 # 推送到企微：客户消息+转接回复成对
                 try:
@@ -881,6 +929,12 @@ class QingjianAPIClient:
                     # 模拟真人打字延迟，避免瞬间回复太机械
                     await asyncio.sleep(0.8)
                     logger.info(f"Agent 回复: {response.reply}")
+                    self._activity_log(
+                        "agent_reply",
+                        acc_id=data.get("acc_id", ""),
+                        customer_id=data.get("from_id", ""),
+                        reply=response.reply,
+                    )
                     await self.send_reply(data, response.reply)
                     # 推送到企微：客户消息+AI回复成对
                     try:
@@ -897,9 +951,20 @@ class QingjianAPIClient:
                         pass
             elif not response.need_transfer:
                 logger.info("Agent 决定不回复此消息")
-                
+                self._activity_log(
+                    "agent_no_reply",
+                    acc_id=data.get("acc_id", ""),
+                    customer_id=data.get("from_id", ""),
+                )
+
         except Exception as e:
             logger.error(f"Agent 处理失败: {e}")
+            self._activity_log(
+                "agent_process_error",
+                acc_id=data.get("acc_id", ""),
+                customer_id=data.get("from_id", ""),
+                error=str(e),
+            )
 
     async def _analyze_multi_and_reply(self, data: dict, urls: list):
         try:
@@ -1513,6 +1578,12 @@ class QingjianAPIClient:
         """
         if not self.websocket:
             print(f"[{self.get_time()}] 错误: 未连接到服务器")
+            self._activity_log(
+                "send_reply_skipped",
+                reason="websocket_not_connected",
+                acc_id=original_msg.get("acc_id", ""),
+                customer_id=original_msg.get("from_id", ""),
+            )
             return
 
         reply_content = self._colloquialize_outbound_reply(reply_content)
@@ -1531,6 +1602,12 @@ class QingjianAPIClient:
                 logger.info(
                     f"外发限流命中，跳过发送 | 客户:{ckey} | cooldown:{cooldown}s | msg:{str(reply_content)[:40]}"
                 )
+                self._activity_log(
+                    "send_reply_throttled",
+                    key=ckey,
+                    cooldown_s=cooldown,
+                    msg=str(reply_content),
+                )
                 return
             self._last_reply_sent_at[ckey] = now_mono
         
@@ -1554,6 +1631,12 @@ class QingjianAPIClient:
             "cy_name": customer_name
         }
         self._log_outbound_once(original_msg, str(reply_content))
+        self._activity_log(
+            "send_reply_attempt",
+            acc_id=shop_id,
+            customer_id=customer_id,
+            msg=str(reply_content),
+        )
         await self.send_message(reply)
 
     def _colloquialize_outbound_reply(self, text: Any) -> Any:
@@ -1652,10 +1735,29 @@ class QingjianAPIClient:
                 await self.websocket.send(msg_json)
                 pretty = json.dumps(message, ensure_ascii=False, indent=2)
                 print(f"[{self.get_time()}] 发送成功:\n{pretty}")
+                self._activity_log(
+                    "send_message_success",
+                    acc_id=message.get("acc_id", ""),
+                    customer_id=message.get("from_id", ""),
+                    msg_type=message.get("msg_type", 0),
+                    msg=message.get("msg", ""),
+                )
             except Exception as e:
                 print(f"[{self.get_time()}] 发送失败: {e}")
+                self._activity_log(
+                    "send_message_error",
+                    acc_id=message.get("acc_id", ""),
+                    customer_id=message.get("from_id", ""),
+                    error=str(e),
+                )
         else:
             print(f"[{self.get_time()}] 错误: 连接未打开")
+            self._activity_log(
+                "send_message_skipped",
+                reason="socket_not_open",
+                acc_id=message.get("acc_id", ""),
+                customer_id=message.get("from_id", ""),
+            )
 
     async def auto_reply(self, data):
         """自动回复示例（已弃用，使用 agent_reply 替代）"""