jimi
c2d8853ae4
Some checks failed
Pre-commit / run (ubuntu-latest) (push) Has been cancelled
Deploy Sphinx documentation to Pages / build_en (ubuntu-latest, 3.10) (push) Has been cancelled
Deploy Sphinx documentation to Pages / build_zh (ubuntu-latest, 3.10) (push) Has been cancelled
Python Unittest Coverage / test (macos-15, 3.10) (push) Has been cancelled
Python Unittest Coverage / test (macos-15, 3.11) (push) Has been cancelled
Python Unittest Coverage / test (macos-15, 3.12) (push) Has been cancelled
Python Unittest Coverage / test (ubuntu-latest, 3.10) (push) Has been cancelled
Python Unittest Coverage / test (ubuntu-latest, 3.11) (push) Has been cancelled
Python Unittest Coverage / test (ubuntu-latest, 3.12) (push) Has been cancelled
Python Unittest Coverage / test (windows-latest, 3.10) (push) Has been cancelled
Python Unittest Coverage / test (windows-latest, 3.11) (push) Has been cancelled
Python Unittest Coverage / test (windows-latest, 3.12) (push) Has been cancelled
305 lines
12 KiB
Python
305 lines
12 KiB
Python
import asyncio
|
||
import json
|
||
import re
|
||
import time
|
||
from collections import defaultdict
|
||
|
||
import websockets
|
||
|
||
from .callbacks import post_tianwang_callback
|
||
from .config import (
|
||
AUTO_QUOTE_WAIT_SECONDS,
|
||
MESSAGE_DEBOUNCE_SECONDS,
|
||
QINGJIAN_WS_URI,
|
||
SHORT_REPLY_MAX_CHARS,
|
||
)
|
||
from .logger import setup_logger
|
||
from .observability import activity_event, build_trace_id
|
||
from .orchestrator import Orchestrator
|
||
from .rules import detect_intent, extract_image_urls, prefilter_message
|
||
|
||
|
||
class QingjianClient:
|
||
def __init__(self) -> None:
|
||
self.logger = setup_logger()
|
||
self.uri = QINGJIAN_WS_URI
|
||
self.reply_id = "tb001"
|
||
self.websocket = None
|
||
self.running = True
|
||
self.orchestrator = Orchestrator()
|
||
|
||
self.pending_msgs: dict[str, list[dict]] = defaultdict(list)
|
||
self.debounce_tasks: dict[str, asyncio.Task] = {}
|
||
self.pending_images: dict[str, list[str]] = defaultdict(list)
|
||
self.auto_quote_tasks: dict[str, asyncio.Task] = {}
|
||
self.last_reply_key: dict[str, str] = {}
|
||
self.recent_outbound: list[tuple[str, str, str, float]] = []
|
||
|
||
@staticmethod
|
||
def _customer_key(data: dict) -> str:
|
||
return f"{data.get('acc_id','')}:{data.get('from_id','')}"
|
||
|
||
@staticmethod
|
||
def _msg_text(data: dict) -> str:
|
||
return str(data.get("msg", "") or "").strip()
|
||
|
||
def _debounce_seconds(self, msg: str) -> float:
|
||
intent = detect_intent(msg)
|
||
if intent == "image":
|
||
return 2.5
|
||
if intent in {"pricing", "finish_or_quote_trigger"}:
|
||
return 2.0
|
||
if intent == "greeting":
|
||
return 1.5
|
||
return float(MESSAGE_DEBOUNCE_SECONDS)
|
||
|
||
async def send_message(self, message: dict) -> None:
|
||
if not self.websocket:
|
||
return
|
||
await self.websocket.send(json.dumps(message, ensure_ascii=False))
|
||
self.logger.info("[发送] %s", message.get("msg", ""))
|
||
|
||
async def send_reply(self, data: dict, text: str, trace_id: str = "-") -> None:
|
||
text = str(text or "").strip()
|
||
if not text:
|
||
return
|
||
text = self._shorten_reply(text)
|
||
msg = {
|
||
"msg_id": "",
|
||
"acc_id": data.get("acc_id", ""),
|
||
"msg": text,
|
||
"from_id": data.get("from_id", ""),
|
||
"from_name": data.get("from_name", data.get("from_id", "")),
|
||
"cy_id": data.get("from_id", ""),
|
||
"acc_type": data.get("acc_type", "AliWorkbench"),
|
||
"msg_type": 0,
|
||
"cy_name": data.get("from_name", data.get("from_id", "")),
|
||
}
|
||
activity_event(self.logger, "send_reply_attempt", trace_id=trace_id, customer_id=data.get("from_id", "-"), msg=text)
|
||
await self.send_message(msg)
|
||
self.recent_outbound.append((str(data.get("acc_id", "")), str(data.get("from_id", "")), text, time.monotonic()))
|
||
if len(self.recent_outbound) > 200:
|
||
self.recent_outbound = self.recent_outbound[-200:]
|
||
activity_event(self.logger, "send_reply_success", trace_id=trace_id, customer_id=data.get("from_id", "-"), msg=text)
|
||
|
||
@staticmethod
|
||
def _clean_text(text: str) -> str:
|
||
t = str(text or "").strip()
|
||
t = re.sub(r"\s+", "", t)
|
||
return t
|
||
|
||
def _shorten_reply(self, text: str) -> str:
|
||
max_len = max(8, int(SHORT_REPLY_MAX_CHARS))
|
||
t = str(text or "").strip()
|
||
if len(t) <= max_len:
|
||
return t
|
||
parts = re.split(r"[。！？!?]", t)
|
||
head = next((p.strip() for p in parts if p and p.strip()), t)
|
||
if len(head) > max_len:
|
||
head = head[:max_len].rstrip("，,;；:： ")
|
||
return head or t[:max_len]
|
||
|
||
@staticmethod
|
||
def _is_invalid_ai_reply(text: str) -> bool:
|
||
t = str(text or "").strip().lower()
|
||
if not t:
|
||
return True
|
||
if "i noticed that you have interrupted me" in t:
|
||
return True
|
||
if t.startswith("action:") or t.startswith("{"):
|
||
return True
|
||
return False
|
||
|
||
def _fallback_reply(self, action: str) -> str:
|
||
if action == "quote":
|
||
return "我先看下，马上给你报价。"
|
||
if action == "transfer":
|
||
return "我给你转人工处理哈。"
|
||
return "收到，我先看一下哈。"
|
||
|
||
def _is_outbound_echo(self, data: dict, msg: str) -> bool:
|
||
"""
|
||
轻简可能会把我方刚发送文本回推为“收到消息”。
|
||
对“短时间完全相同文本”做回环拦截，兼容 acc/from 对调回推，避免无限对话。
|
||
"""
|
||
in_acc = str(data.get("acc_id", ""))
|
||
in_from = str(data.get("from_id", ""))
|
||
in_msg = self._clean_text(msg)
|
||
now = time.monotonic()
|
||
if not in_msg:
|
||
return False
|
||
for out_acc, out_to, out_msg, ts in reversed(self.recent_outbound):
|
||
if (now - ts) > 120:
|
||
break
|
||
if self._clean_text(out_msg) != in_msg:
|
||
continue
|
||
if (out_acc == in_acc and out_to == in_from) or (out_acc == in_from and out_to == in_acc):
|
||
return True
|
||
return False
|
||
|
||
async def _handle_decision(self, data: dict, merged_msg: str, *, auto_quote: bool = False) -> None:
|
||
key = self._customer_key(data)
|
||
trace_id = build_trace_id(data.get("acc_id", ""), data.get("from_id", ""), merged_msg)
|
||
t0 = time.perf_counter()
|
||
|
||
urls = extract_image_urls(merged_msg)
|
||
if urls:
|
||
for u in urls:
|
||
if u not in self.pending_images[key]:
|
||
self.pending_images[key].append(u)
|
||
|
||
context = {
|
||
"customer_key": key,
|
||
"acc_id": data.get("acc_id", ""),
|
||
"customer_id": data.get("from_id", ""),
|
||
"goods_name": data.get("goods_name", ""),
|
||
"goods_order": data.get("goods_order", ""),
|
||
"msg": merged_msg,
|
||
"intent": detect_intent(merged_msg),
|
||
"pending_images": len(self.pending_images[key]),
|
||
"auto_quote_trigger": auto_quote,
|
||
"last_reply": self.last_reply_key.get(key, ""),
|
||
}
|
||
|
||
activity_event(self.logger, "agent_process_start", trace_id=trace_id, customer_id=context["customer_id"], acc_id=context["acc_id"], intent=context["intent"])
|
||
route, decision, state = await self.orchestrator.decide(context)
|
||
latency_ms = int((time.perf_counter() - t0) * 1000)
|
||
activity_event(
|
||
self.logger,
|
||
"agent_process_done",
|
||
trace_id=trace_id,
|
||
customer_id=context["customer_id"],
|
||
route=route,
|
||
action=decision.action,
|
||
reason=decision.reason,
|
||
latency_ms=latency_ms,
|
||
after_sales_stage=state.get("after_sales_stage", "new"),
|
||
)
|
||
|
||
if decision.action == "transfer":
|
||
text = (decision.transfer_msg or "").strip()
|
||
if self._is_invalid_ai_reply(text):
|
||
text = self._fallback_reply("transfer")
|
||
await self.send_reply(data, text, trace_id=trace_id)
|
||
self.last_reply_key[key] = text
|
||
await post_tianwang_callback("message_processed", data, extra={"trace_id": trace_id, "route": route, "action": "transfer", "reply": text})
|
||
return
|
||
|
||
if decision.action == "quote":
|
||
text = (decision.reply or "").strip()
|
||
if self._is_invalid_ai_reply(text):
|
||
text = self._fallback_reply("quote")
|
||
if self.last_reply_key.get(key) != text:
|
||
await self.send_reply(data, text, trace_id=trace_id)
|
||
self.last_reply_key[key] = text
|
||
await post_tianwang_callback("message_processed", data, extra={"trace_id": trace_id, "route": route, "action": "quote", "reply": text})
|
||
return
|
||
|
||
if decision.action == "noop":
|
||
await post_tianwang_callback("message_processed", data, extra={"trace_id": trace_id, "route": route, "action": "noop", "reply": ""})
|
||
return
|
||
|
||
text = (decision.reply or "").strip()
|
||
if self._is_invalid_ai_reply(text):
|
||
text = self._fallback_reply("reply")
|
||
if self.last_reply_key.get(key) != text:
|
||
await self.send_reply(data, text, trace_id=trace_id)
|
||
self.last_reply_key[key] = text
|
||
|
||
await post_tianwang_callback("message_processed", data, extra={"trace_id": trace_id, "route": route, "action": "reply", "reply": text})
|
||
|
||
if self.pending_images[key] and key not in self.auto_quote_tasks:
|
||
self.auto_quote_tasks[key] = asyncio.create_task(self._auto_quote_later(data))
|
||
|
||
async def _auto_quote_later(self, data: dict) -> None:
|
||
key = self._customer_key(data)
|
||
try:
|
||
await asyncio.sleep(AUTO_QUOTE_WAIT_SECONDS)
|
||
if self.pending_images.get(key):
|
||
await self._handle_decision(data, "发完了，报价吧", auto_quote=True)
|
||
finally:
|
||
self.auto_quote_tasks.pop(key, None)
|
||
|
||
async def _flush_customer(self, key: str) -> None:
|
||
queue = self.pending_msgs.get(key, [])
|
||
if not queue:
|
||
return
|
||
merged = "、".join([self._msg_text(x) for x in queue if self._msg_text(x)])
|
||
data = queue[-1]
|
||
self.pending_msgs[key].clear()
|
||
await self._handle_decision(data, merged)
|
||
|
||
async def _debounce_enqueue(self, data: dict) -> None:
|
||
key = self._customer_key(data)
|
||
msg = self._msg_text(data)
|
||
self.pending_msgs[key].append(data)
|
||
|
||
if key in self.debounce_tasks:
|
||
self.debounce_tasks[key].cancel()
|
||
|
||
wait_s = self._debounce_seconds(msg)
|
||
activity_event(self.logger, "debounce_enqueue", customer_id=data.get("from_id", "-"), key=key, queue_size=len(self.pending_msgs[key]), wait_s=wait_s)
|
||
|
||
async def later() -> None:
|
||
try:
|
||
await asyncio.sleep(wait_s)
|
||
await self._flush_customer(key)
|
||
except asyncio.CancelledError:
|
||
return
|
||
finally:
|
||
self.debounce_tasks.pop(key, None)
|
||
|
||
self.debounce_tasks[key] = asyncio.create_task(later())
|
||
|
||
async def _on_message(self, raw: str) -> None:
|
||
try:
|
||
data = json.loads(raw)
|
||
except Exception:
|
||
self.logger.info("[非JSON] %s", raw)
|
||
return
|
||
|
||
msg_type = int(data.get("msg_type", 0) or 0)
|
||
msg = self._msg_text(data)
|
||
rule = prefilter_message(msg, msg_type)
|
||
|
||
self.logger.info("[收消息] acc=%s from=%s type=%s msg=%s", data.get("acc_id", ""), data.get("from_id", ""), msg_type, msg)
|
||
await post_tianwang_callback("message_received", data, extra={"msg_type": msg_type})
|
||
|
||
if self._is_outbound_echo(data, msg):
|
||
activity_event(
|
||
self.logger,
|
||
"inbound_ignored",
|
||
customer_id=data.get("from_id", "-"),
|
||
reason="outbound_echo_loop_guard",
|
||
)
|
||
return
|
||
|
||
if rule.ignore:
|
||
activity_event(self.logger, "inbound_ignored", customer_id=data.get("from_id", "-"), reason=rule.reason)
|
||
return
|
||
|
||
patched = dict(data)
|
||
patched["msg"] = rule.normalized_msg or msg
|
||
|
||
if msg_type == 1:
|
||
await self._handle_decision(patched, patched["msg"])
|
||
return
|
||
|
||
await self._debounce_enqueue(patched)
|
||
|
||
async def _serve(self) -> None:
|
||
while self.running:
|
||
try:
|
||
self.logger.info("[连接] %s", self.uri)
|
||
async with websockets.connect(self.uri) as ws:
|
||
self.websocket = ws
|
||
self.logger.info("[连接成功]")
|
||
async for raw in ws:
|
||
await self._on_message(raw)
|
||
except Exception as e:
|
||
self.logger.info("[连接异常] %s", e)
|
||
await asyncio.sleep(3)
|
||
|
||
def run(self) -> None:
|
||
asyncio.run(self._serve())
|