zuowei/tw
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: queue pending transfers until designers are available

Browse Source
This commit is contained in:
jimi
2026-03-08 12:43:40 +08:00
parent 5a5bde1ba5
commit 2e3409d8c5
3 changed files with 288 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

106
core/orchestrator.py
View File

@@ -10,6 +10,13 @@ from core.adapters.qianniu_adapter import QianniuAdapter
from core.pydantic_ai_agent_v2 import CustomerServiceBrain
from core.events.event_bus import bus
from core.repository import repo
from db.pending_transfer_db import (
enqueue_pending_transfer,
claim_due_pending_transfers,
complete_pending_transfer,
retry_pending_transfer,
)
from services.dispatch_service import dispatch_service
logger = logging.getLogger("cs_agent")
@@ -17,6 +24,8 @@ logger = logging.getLogger("cs_agent")
MSG_DEDUP_CAPACITY = 200 # 消息 ID 去重缓存容量
TRANSFER_COOLDOWN_SEC = 120 # 转接冷却时间（秒）—— 转接后2分钟内不再调用AI
DEBOUNCE_SECONDS = 2.0 # 消息防抖延迟（秒）
PENDING_TRANSFER_POLL_SECONDS = 30
PENDING_TRANSFER_RETRY_SECONDS = 60
# 转接后安抚话术池（轮换使用，避免复读）
_TRANSFER_CALM_REPLIES = [
@@ -70,6 +79,7 @@ class SystemOrchestrator:
self._debounce_tasks: Dict[str, asyncio.Task] = {}
self._pending_messages: Dict[str, List[StandardMessage]] = {}
self._user_locks: Dict[str, asyncio.Lock] = {}
self._pending_transfer_task: Optional[asyncio.Task] = None
bus.subscribe("MESSAGE_OUTBOUND", self.handle_outbound_event)
@@ -86,6 +96,17 @@ class SystemOrchestrator:
self._user_locks[user_id] = asyncio.Lock()
return self._user_locks[user_id]
def _should_run_pending_transfer_worker(self) -> bool:
worker_id = getattr(self.ws_client, "worker_id", -1) if self.ws_client else -1
return worker_id in (-1, 0)
def _ensure_background_tasks(self):
if not self._should_run_pending_transfer_worker():
return
if self._pending_transfer_task is None or self._pending_transfer_task.done():
self._pending_transfer_task = asyncio.create_task(self._process_pending_transfers_loop())
logger.info("[Orchestrator] 待转接轮询任务已启动")
@staticmethod
def _sanitize_outbound_text(text: str) -> str:
if not text:
@@ -107,6 +128,7 @@ class SystemOrchestrator:
"""链路入口"""
try:
if platform != "qianniu": return
self._ensure_background_tasks()
std_msg, direction = await self.qianniu_adapter.translate_inbound(raw_data)
@@ -343,7 +365,9 @@ class SystemOrchestrator:
# E. 发送并记录时间
if std_res.should_reply:
std_res.reply_content = self._sanitize_outbound_text(std_res.reply_content)
std_res.metadata = {"acc_id": acc_id, "acc_type": acc_type}
meta = dict(std_res.metadata or {})
meta.update({"acc_id": acc_id, "acc_type": acc_type})
std_res.metadata = meta
# 转接场景：先发一句安抚话，再发转接指令
if "[转移会话]" in std_res.reply_content:
@@ -358,6 +382,21 @@ class SystemOrchestrator:
await self.qianniu_adapter.translate_outbound(std_res, user_id)
await repo.save_chat(platform, user_id, std_res.reply_content, "out", acc_id=acc_id)
if std_res.metadata.get("pending_transfer"):
reason = str(std_res.metadata.get("pending_transfer_reason") or "").strip()
if reason:
pending_id = await asyncio.to_thread(
enqueue_pending_transfer,
customer_id=user_id,
acc_id=acc_id,
acc_type=acc_type,
platform=platform,
reason=reason,
)
logger.info(
f"[Orchestrator] 已加入待转接池: pending_id={pending_id} user={user_id} acc={acc_id}"
)
if "[转移会话]" in std_res.reply_content:
self._last_transfer_time[session_key] = time.time()
@@ -370,6 +409,71 @@ class SystemOrchestrator:
response.reply_content = self._sanitize_outbound_text(response.reply_content)
await self.qianniu_adapter.translate_outbound(response, user_id)
async def _process_pending_transfers_loop(self):
while True:
try:
if not self.ws_client or not getattr(self.ws_client, "websocket", None):
await asyncio.sleep(PENDING_TRANSFER_POLL_SECONDS)
continue
rows = await asyncio.to_thread(claim_due_pending_transfers, 10)
if not rows:
await asyncio.sleep(PENDING_TRANSFER_POLL_SECONDS)
continue
for row in rows:
row_id = int(row["id"])
customer_id = str(row.get("customer_id") or "")
acc_id = str(row.get("acc_id") or "")
acc_type = str(row.get("acc_type") or "AliWorkbench")
reason = str(row.get("reason") or "").strip()
try:
designer_name = await dispatch_service.assign_designer(user_id=customer_id)
if not designer_name:
await asyncio.to_thread(
retry_pending_transfer,
row_id,
PENDING_TRANSFER_RETRY_SECONDS,
"designer_unavailable",
)
continue
notify = StandardResponse(
reply_content="设计师上线了，我给您转过去哈",
metadata={"acc_id": acc_id, "acc_type": acc_type},
)
transfer = StandardResponse(
reply_content=f"正在为您转接|[转移会话],{designer_name},无原因",
need_transfer=True,
metadata={"acc_id": acc_id, "acc_type": acc_type},
)
await self.qianniu_adapter.translate_outbound(notify, customer_id)
await repo.save_chat("qianniu", customer_id, notify.reply_content, "out", acc_id=acc_id)
await asyncio.sleep(0.5)
await self.qianniu_adapter.translate_outbound(transfer, customer_id)
await repo.save_chat("qianniu", customer_id, transfer.reply_content, "out", acc_id=acc_id)
self._last_transfer_time[f"{customer_id}@{acc_id}"] = time.time()
await asyncio.to_thread(complete_pending_transfer, row_id)
logger.info(
f"[Orchestrator] 待转接自动完成: pending_id={row_id} user={customer_id} designer={designer_name} reason={reason}"
)
except Exception as e:
logger.warning(f"[Orchestrator] 待转接处理失败 pending_id={row_id}: {e}")
await asyncio.to_thread(
retry_pending_transfer,
row_id,
PENDING_TRANSFER_RETRY_SECONDS,
str(e),
)
except asyncio.CancelledError:
break
except Exception as e:
logger.warning(f"[Orchestrator] 待转接轮询异常: {e}")
await asyncio.sleep(PENDING_TRANSFER_RETRY_SECONDS)
# 全局单例
orchestrator: Optional[SystemOrchestrator] = None
def init_orchestrator(ws_client):