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

refactor: unify workflow/websocket logging and extract conversation state store

Browse Source
This commit is contained in:
jimi
2026-03-01 16:35:39 +08:00
parent 8dd5a11b4b
commit 4a07f9c726
4 changed files with 219 additions and 181 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
core/conversation_state_store.py Normal file
View File

@@ -0,0 +1,95 @@
from __future__ import annotations
from datetime import datetime
from typing import Any
from core.quote_state_machine import QuoteStateMachine
def refresh_quote_phase(state: Any, phase_hint: str = "") -> None:
"""统一维护收图报价状态机。"""
QuoteStateMachine().refresh(state, phase_hint=phase_hint)
def sync_pending_quote_state(agent: Any, customer_id: str, state: Any) -> None:
"""把待报价队列同步到客户库，避免重启丢失。"""
try:
refresh_quote_phase(state)
from db.customer_db import db
db.update_pending_quote_state(
customer_id,
state.pending_image_urls,
state.pending_requirements,
)
except Exception:
pass
def restore_pending_quote_state(customer_id: str, state: Any) -> None:
"""从客户库恢复待报价队列。"""
try:
from db.customer_db import db
profile = db.get_customer(customer_id)
state.pending_image_urls = list(getattr(profile, "pending_quote_images", []) or [])
state.pending_requirements = list(getattr(profile, "pending_quote_requirements", []) or [])
state.image_count = len(state.pending_image_urls)
refresh_quote_phase(state)
except Exception:
pass
def cleanup_inactive(conversations: dict, message_histories: dict, now: datetime) -> None:
"""清理超过 7 天没有消息的对话状态，释放内存。"""
if len(conversations) % 100 != 0:
return
expired = [
cid
for cid, state in conversations.items()
if state.last_update and (now - datetime.fromisoformat(state.last_update)).days > 7
]
for cid in expired:
conversations.pop(cid, None)
message_histories.pop(cid, None)
def get_conversation_state(agent: Any, customer_id: str) -> Any:
"""获取或创建对话状态，超时自动重置。"""
now = datetime.now()
if customer_id in agent.conversations:
state = agent.conversations[customer_id]
if state.last_update:
try:
last = datetime.fromisoformat(state.last_update)
hours = (now - last).total_seconds() / 3600
if hours > agent.CONVERSATION_TIMEOUT_HOURS:
state.stage = "售前"
state.discount_count = 0
agent.message_histories.pop(customer_id, None)
except Exception:
pass
if not state.pending_image_urls and not state.pending_requirements:
restore_pending_quote_state(customer_id, state)
else:
agent.conversations[customer_id] = agent.ConversationStateClass(
customer_id=customer_id,
last_update=now.isoformat(),
)
restore_pending_quote_state(customer_id, agent.conversations[customer_id])
cleanup_inactive(agent.conversations, agent.message_histories, now)
return agent.conversations[customer_id]
def should_defer_batch_quote(agent: Any, state: Any, mark_ready: bool = False) -> bool:
"""批量报价延后控制。"""
agent.quote_state_machine.delay_turns = max(0, int(agent.batch_quote_delay_turns))
return agent.quote_state_machine.should_defer_batch_quote(state, mark_ready=mark_ready)
def mark_quote_ready(agent: Any, state: Any) -> None:
"""仅标记 ready 状态，不消费等待轮次。"""
agent.quote_state_machine.delay_turns = max(0, int(agent.batch_quote_delay_turns))
agent.quote_state_machine.mark_ready(state)

93
core/pydantic_ai_agent.py
View File

@@ -71,6 +71,15 @@ from core.batch_quote_helpers import (
from core.prompt_builder import build_prompt as build_agent_prompt, split_customer_text
from core.image_workflow_router import handle_image_workflow as route_image_workflow
from core.message_orchestrator import process_incoming_message
from core.conversation_state_store import (
get_conversation_state as load_conversation_state,
mark_quote_ready as state_mark_quote_ready,
refresh_quote_phase as state_refresh_quote_phase,
should_defer_batch_quote as state_should_defer_batch_quote,
sync_pending_quote_state as state_sync_pending_quote_state,
restore_pending_quote_state as state_restore_pending_quote_state,
cleanup_inactive as state_cleanup_inactive,
)
load_dotenv()
@@ -285,6 +294,7 @@ class CustomerServiceAgent:
# 对话状态管理
self.conversations: Dict[str, ConversationState] = {}
self.ConversationStateClass = ConversationState
# 多轮对话历史（PydanticAI ModelMessage 列表，按客户ID存储）
self.message_histories: Dict[str, list] = {}
self.evolution_candidate = self._load_evolution_candidate()
@@ -584,95 +594,26 @@ class CustomerServiceAgent:
CONVERSATION_TIMEOUT_HOURS = 12
def _get_conversation_state(self, customer_id: str) -> ConversationState:
"""获取或创建对话状态，超时自动重置"""
now = datetime.now()
if customer_id in self.conversations:
state = self.conversations[customer_id]
# 超过 12 小时没有消息，重置阶段和压价次数
if state.last_update:
try:
last = datetime.fromisoformat(state.last_update)
hours = (now - last).total_seconds() / 3600
if hours > self.CONVERSATION_TIMEOUT_HOURS:
state.stage = "售前"
state.discount_count = 0
# 同时清理对话历史，避免发送过期上下文
self.message_histories.pop(customer_id, None)
except Exception:
pass
# 进程内状态为空时，尝试从持久化恢复
if not state.pending_image_urls and not state.pending_requirements:
self._restore_pending_quote_state(customer_id, state)
else:
self.conversations[customer_id] = ConversationState(
customer_id=customer_id,
last_update=now.isoformat()
)
self._restore_pending_quote_state(customer_id, self.conversations[customer_id])
# 定期清理长期不活跃客户（超过 7 天）
self._cleanup_inactive(now)
return self.conversations[customer_id]
return load_conversation_state(self, customer_id)
def _cleanup_inactive(self, now: datetime):
"""清理超过 7 天没有消息的对话状态，释放内存"""
# 每 100 次调用清理一次，避免每次都遍历
if len(self.conversations) % 100 != 0:
return
expired = [
cid for cid, state in self.conversations.items()
if state.last_update and
(now - datetime.fromisoformat(state.last_update)).days > 7
]
for cid in expired:
self.conversations.pop(cid, None)
self.message_histories.pop(cid, None)
state_cleanup_inactive(self.conversations, self.message_histories, now)
def _sync_pending_quote_state(self, customer_id: str, state: ConversationState):
"""把待报价队列同步到客户库，避免重启丢失。"""
try:
self._refresh_quote_phase(state)
from db.customer_db import db
db.update_pending_quote_state(
customer_id,
state.pending_image_urls,
state.pending_requirements,
)
except Exception:
pass
state_sync_pending_quote_state(self, customer_id, state)
def _restore_pending_quote_state(self, customer_id: str, state: ConversationState):
"""从客户库恢复待报价队列。"""
try:
from db.customer_db import db
profile = db.get_customer(customer_id)
state.pending_image_urls = list(getattr(profile, "pending_quote_images", []) or [])
state.pending_requirements = list(getattr(profile, "pending_quote_requirements", []) or [])
state.image_count = len(state.pending_image_urls)
self._refresh_quote_phase(state)
except Exception:
pass
state_restore_pending_quote_state(customer_id, state)
@staticmethod
def _refresh_quote_phase(state: ConversationState, phase_hint: str = ""):
"""统一维护收图报价状态机。"""
QuoteStateMachine().refresh(state, phase_hint=phase_hint)
state_refresh_quote_phase(state, phase_hint=phase_hint)
def _should_defer_batch_quote(self, state: ConversationState, mark_ready: bool = False) -> bool:
"""
批量报价延后控制：
- 首次进入 ready_to_quote 时按配置等待 N 轮
- 等待轮次归零后，本轮即可报价
"""
self.quote_state_machine.delay_turns = max(0, int(self.batch_quote_delay_turns))
return self.quote_state_machine.should_defer_batch_quote(state, mark_ready=mark_ready)
return state_should_defer_batch_quote(self, state, mark_ready=mark_ready)
def _mark_quote_ready(self, state: ConversationState):
"""仅标记 ready 状态，不消费等待轮次。"""
self.quote_state_machine.delay_turns = max(0, int(self.batch_quote_delay_turns))
self.quote_state_machine.mark_ready(state)
state_mark_quote_ready(self, state)
def _build_reject_message(self, reason: str = "") -> str:
templates = [

141
core/websocket_client.py
View File

@@ -67,9 +67,9 @@ except Exception as e:
workflow = None
_get_shop_type = lambda acc_id, goods_name: "find_image"
import traceback
print(f"警告: Agent 模块导入失败: {e}")
logger.info(f"警告: Agent 模块导入失败: {e}")
traceback.print_exc()
print("将使用基础回复功能")
logger.info("将使用基础回复功能")
class QingjianAPIClient:
@@ -123,9 +123,9 @@ class QingjianAPIClient:
if self.enable_agent:
try:
self.agent = CustomerServiceAgent()
print(f"[{self.get_time()}] Agent 初始化成功")
logger.info(f"[{self.get_time()}] Agent 初始化成功")
except Exception as e:
print(f"[{self.get_time()}] Agent 初始化失败: {e}")
logger.info(f"[{self.get_time()}] Agent 初始化失败: {e}")
self.enable_agent = False
# 注册 workflow 消息发送回调（供图片AI完成后推送消息用）
@@ -149,15 +149,15 @@ class QingjianAPIClient:
"""连接WebSocket服务器"""
while self.running:
try:
print(f"[{self.get_time()}] 正在连接轻简API {self.uri}...")
logger.info(f"[{self.get_time()}] 正在连接轻简API {self.uri}...")
async with websockets.connect(self.uri) as websocket:
self.websocket = websocket
from utils.health_check import set_qingjian_connected
set_qingjian_connected(True)
print(f"[{self.get_time()}] 连接成功!")
logger.info(f"[{self.get_time()}] 连接成功!")
if self.enable_agent:
print(f"[{self.get_time()}] AI Agent 已启用，将自动处理消息")
print(f"[{self.get_time()}] 等待接收消息...")
logger.info(f"[{self.get_time()}] AI Agent 已启用，将自动处理消息")
logger.info(f"[{self.get_time()}] 等待接收消息...")
# 持续接收消息
await self.receive_messages()
@@ -165,19 +165,19 @@ class QingjianAPIClient:
except ConnectionRefusedError:
from utils.health_check import set_qingjian_connected
set_qingjian_connected(False)
print(f"[{self.get_time()}] 连接被拒绝，请检查轻简软件是否已启动")
logger.info(f"[{self.get_time()}] 连接被拒绝，请检查轻简软件是否已启动")
except websockets.exceptions.InvalidURI:
from utils.health_check import set_qingjian_connected
set_qingjian_connected(False)
print(f"[{self.get_time()}] URI格式错误")
logger.info(f"[{self.get_time()}] URI格式错误")
except Exception as e:
from utils.health_check import set_qingjian_connected
set_qingjian_connected(False)
print(f"[{self.get_time()}] 连接错误: {e}")
logger.info(f"[{self.get_time()}] 连接错误: {e}")
# 等待5秒后重连
if self.running:
print(f"[{self.get_time()}] 5秒后尝试重连...")
logger.info(f"[{self.get_time()}] 5秒后尝试重连...")
await asyncio.sleep(5)
def _customer_key(self, data: dict) -> str:
@@ -295,11 +295,11 @@ class QingjianAPIClient:
except websockets.exceptions.ConnectionClosed:
from utils.health_check import set_qingjian_connected
set_qingjian_connected(False)
print(f"[{self.get_time()}] 连接已关闭")
logger.info(f"[{self.get_time()}] 连接已关闭")
except Exception as e:
from utils.health_check import set_qingjian_connected
set_qingjian_connected(False)
print(f"[{self.get_time()}] 接收消息错误: {e}")
logger.info(f"[{self.get_time()}] 接收消息错误: {e}")
async def handle_message(self, message):
"""处理接收到的消息"""
@@ -317,25 +317,25 @@ class QingjianAPIClient:
self.last_msg = data
# 打印格式化的消息
print(f"\n{'='*50}")
print(f"[{timestamp}] 收到新消息:")
print(f"{'='*50}")
print(f" 消息ID: {data.get('msg_id', 'N/A')}")
print(f" 账号ID: {self.to_chinese(data.get('acc_id', 'N/A'))}")
print(f" 发送者ID: {self.to_chinese(data.get('from_id', 'N/A'))}")
print(f" 发送者名称: {self.to_chinese(data.get('from_name', 'N/A'))}")
print(f" 会话ID: {self.to_chinese(data.get('cy_id', 'N/A'))}")
print(f" 平台类型: {data.get('acc_type', 'N/A')}")
print(f" 消息类型: {self.get_msg_type_name(data.get('msg_type', 0))}")
print(f" 消息内容: {self.to_chinese(data.get('msg', 'N/A'))}")
logger.info(f"\n{'='*50}")
logger.info(f"[{timestamp}] 收到新消息:")
logger.info(f"{'='*50}")
logger.info(f" 消息ID: {data.get('msg_id', 'N/A')}")
logger.info(f" 账号ID: {self.to_chinese(data.get('acc_id', 'N/A'))}")
logger.info(f" 发送者ID: {self.to_chinese(data.get('from_id', 'N/A'))}")
logger.info(f" 发送者名称: {self.to_chinese(data.get('from_name', 'N/A'))}")
logger.info(f" 会话ID: {self.to_chinese(data.get('cy_id', 'N/A'))}")
logger.info(f" 平台类型: {data.get('acc_type', 'N/A')}")
logger.info(f" 消息类型: {self.get_msg_type_name(data.get('msg_type', 0))}")
logger.info(f" 消息内容: {self.to_chinese(data.get('msg', 'N/A'))}")
# 显示商品信息（如果有）
if data.get('goods_name'):
print(f" 商品名称: {self.to_chinese(data.get('goods_name', ''))}")
logger.info(f" 商品名称: {self.to_chinese(data.get('goods_name', ''))}")
if data.get('goods_order'):
print(f" 订单信息: {self.to_chinese(data.get('goods_order', ''))}")
logger.info(f" 订单信息: {self.to_chinese(data.get('goods_order', ''))}")
print(f"{'='*50}\n")
logger.info(f"{'='*50}\n")
# 消息去重：同一条消息不重复处理
msg_id = data.get('msg_id', '')
@@ -350,14 +350,14 @@ class QingjianAPIClient:
acc_id = data.get('acc_id', '')
msg_body = data.get('msg', '')
if not from_id or from_id == 'N/A' or not acc_id or acc_id == 'N/A':
print(f"[{self.get_time()}] 空消息跳过（from_id={from_id!r} acc_id={acc_id!r}）")
logger.info(f"[{self.get_time()}] 空消息跳过（from_id={from_id!r} acc_id={acc_id!r}）")
return
self._log_inbound_once(data)
# Gemini 店铺：不回复，直接跳过
goods_name = self.to_chinese(data.get('goods_name', '') or '')
if _get_shop_type(acc_id, goods_name) == "gemini_api":
print(f"[{self.get_time()}] Gemini 店铺消息，跳过")
logger.info(f"[{self.get_time()}] Gemini 店铺消息，跳过")
try:
from utils.wechat_chat_log import push_chat_to_wechat
asyncio.create_task(push_chat_to_wechat(
@@ -378,7 +378,7 @@ class QingjianAPIClient:
if msg_type == 0:
if self._is_transfer_msg(data):
# 会话转交 → 主动打招呼
print(f"[{self.get_time()}] 收到转交消息，发送问候")
logger.info(f"[{self.get_time()}] 收到转交消息，发送问候")
greeting = self._pick_transfer_greeting()
await self.send_reply(data, greeting)
try:
@@ -397,9 +397,9 @@ class QingjianAPIClient:
# 进店卡片：有历史对话就不回复，没有才打招呼（Gemini 已在上面统一跳过）
cid = data.get('from_id', '')
if self._has_chat_history(cid):
print(f"[{self.get_time()}] 进店卡片（已有记录），跳过")
logger.info(f"[{self.get_time()}] 进店卡片（已有记录），跳过")
else:
print(f"[{self.get_time()}] 进店卡片（新客户），发送问候")
logger.info(f"[{self.get_time()}] 进店卡片（新客户），发送问候")
greeting = "在呢，发图来我看看"
await self.send_reply(data, greeting)
try:
@@ -415,9 +415,9 @@ class QingjianAPIClient:
except Exception:
pass
elif await self._handle_system_inquiry(data):
print(f"[{self.get_time()}] 系统客服询单消息，已按规则处理")
logger.info(f"[{self.get_time()}] 系统客服询单消息，已按规则处理")
elif self._should_ignore(data):
print(f"[{self.get_time()}] 系统通知，跳过回复")
logger.info(f"[{self.get_time()}] 系统通知，跳过回复")
else:
await self._debounce_agent_reply(data)
elif msg_type == 1:
@@ -425,7 +425,7 @@ class QingjianAPIClient:
await self.handle_image_message(data)
except json.JSONDecodeError:
print(f"[{timestamp}] 收到非JSON消息: {message}")
logger.info(f"[{timestamp}] 收到非JSON消息: {message}")
async def _debounce_agent_reply(self, data: dict):
"""
@@ -476,7 +476,7 @@ class QingjianAPIClient:
merged_msg = msgs[0]
else:
merged_msg = "、".join(m for m in msgs if m.strip())
print(f"[{self.get_time()}] 防抖合并 {len(msgs)} 条消息: {merged_msg[:60]}")
logger.info(f"[{self.get_time()}] 防抖合并 {len(msgs)} 条消息: {merged_msg[:60]}")
self._activity_log(
"debounce_flush",
key=capture_key,
@@ -1514,7 +1514,7 @@ class QingjianAPIClient:
设计师在线状态：仅在转人工时按需查询，不轮询。
"""
if not self.websocket:
print(f"[{self.get_time()}] 错误: 未连接到服务器")
logger.info(f"[{self.get_time()}] 错误: 未连接到服务器")
return
acc_id = data.get("acc_id", "")
@@ -1569,14 +1569,14 @@ class QingjianAPIClient:
if assigned_to:
cmd = f"正在为你转接人工|[转移会话],{assigned_to},无原因"
await self.send_reply(data, cmd)
print(f"[{self.get_time()}] 已发送转接请求 (店铺:{acc_id or '未知'} -> 设计师:{assigned_to})")
logger.info(f"[{self.get_time()}] 已发送转接请求 (店铺:{acc_id or '未知'} -> 设计师:{assigned_to})")
return
if not group_id:
group_id = _get_transfer_group(acc_id)
cmd = f"话术|[转移会话],分组{group_id},无原因"
await self.send_reply(data, cmd)
print(f"[{self.get_time()}] 已发送转接请求 (店铺:{acc_id or '未知'} -> 分组:{group_id})")
logger.info(f"[{self.get_time()}] 已发送转接请求 (店铺:{acc_id or '未知'} -> 分组:{group_id})")
async def _save_conversation_summary(self, customer_id: str, buyer_msg: str, agent_reply: str):
"""用 AI 生成一句话对话摘要并持久化"""
@@ -1678,7 +1678,7 @@ class QingjianAPIClient:
"""
trace_id = original_msg.get("_trace_id", "")
if not self.websocket:
print(f"[{self.get_time()}] 错误: 未连接到服务器")
logger.info(f"[{self.get_time()}] 错误: 未连接到服务器")
self._activity_log(
"send_reply_skipped",
trace_id=trace_id,
@@ -1839,7 +1839,7 @@ class QingjianAPIClient:
msg_json = json.dumps(message, ensure_ascii=False)
await self.websocket.send(msg_json)
pretty = json.dumps(message, ensure_ascii=False, indent=2)
print(f"[{self.get_time()}] 发送成功:\n{pretty}")
logger.info(f"[{self.get_time()}] 发送成功:\n{pretty}")
self._activity_log(
"send_message_success",
trace_id=message.get("_trace_id", ""),
@@ -1849,7 +1849,7 @@ class QingjianAPIClient:
msg=message.get("msg", ""),
)
except Exception as e:
print(f"[{self.get_time()}] 发送失败: {e}")
logger.info(f"[{self.get_time()}] 发送失败: {e}")
self._activity_log(
"send_message_error",
trace_id=message.get("_trace_id", ""),
@@ -1858,7 +1858,7 @@ class QingjianAPIClient:
error=str(e),
)
else:
print(f"[{self.get_time()}] 错误: 连接未打开")
logger.info(f"[{self.get_time()}] 错误: 连接未打开")
self._activity_log(
"send_message_skipped",
trace_id=message.get("_trace_id", ""),
@@ -1873,13 +1873,13 @@ class QingjianAPIClient:
async def command_handler(self):
"""命令行交互"""
print("\n命令帮助:")
print(" reply <内容> - 回复最后一条消息")
print(" text <id> <平台> <内容> - 发送文本消息")
print(" img <id> <平台> <路径> - 发送图片")
print(" setid <id> - 设置回复ID")
print(" agent on/off - 开启/关闭 Agent")
print(" exit/quit - 退出\n")
logger.info("\n命令帮助:")
logger.info(" reply <内容> - 回复最后一条消息")
logger.info(" text <id> <平台> <内容> - 发送文本消息")
logger.info(" img <id> <平台> <路径> - 发送图片")
logger.info(" setid <id> - 设置回复ID")
logger.info(" agent on/off - 开启/关闭 Agent")
logger.info(" exit/quit - 退出\n")
while self.running:
try:
@@ -1893,7 +1893,7 @@ class QingjianAPIClient:
cmd = parts[0].lower()
if cmd in ["exit", "quit", "q"]:
print(f"[{self.get_time()}] 正在关闭...")
logger.info(f"[{self.get_time()}] 正在关闭...")
self.running = False
if self.websocket:
await self.websocket.close()
@@ -1901,21 +1901,21 @@ class QingjianAPIClient:
elif cmd == "setid" and len(parts) > 1:
self.reply_id = parts[1]
print(f"[{self.get_time()}] 回复ID已设置为: {self.reply_id}")
logger.info(f"[{self.get_time()}] 回复ID已设置为: {self.reply_id}")
elif cmd == "agent" and len(parts) > 1:
if parts[1].lower() == "on":
self.enable_agent = True
print(f"[{self.get_time()}] Agent 已开启")
logger.info(f"[{self.get_time()}] Agent 已开启")
elif parts[1].lower() == "off":
self.enable_agent = False
print(f"[{self.get_time()}] Agent 已关闭")
logger.info(f"[{self.get_time()}] Agent 已关闭")
elif cmd == "reply" and len(parts) > 1:
if self.last_msg:
await self.send_reply(self.last_msg, parts[1])
else:
print(f"[{self.get_time()}] 错误: 还没有收到任何消息")
logger.info(f"[{self.get_time()}] 错误: 还没有收到任何消息")
elif cmd == "text" and len(parts) > 1:
# text cy_id acc_type content
@@ -1923,7 +1923,7 @@ class QingjianAPIClient:
if len(args) >= 3:
await self.send_text(args[0], args[1], args[2])
else:
print(f"[{self.get_time()}] 格式: text <cy_id> <acc_type> <内容>")
logger.info(f"[{self.get_time()}] 格式: text <cy_id> <acc_type> <内容>")
elif cmd == "img" and len(parts) > 1:
# img cy_id acc_type image_path
@@ -1931,13 +1931,13 @@ class QingjianAPIClient:
if len(args) >= 3:
await self.send_image(args[0], args[1], args[2])
else:
print(f"[{self.get_time()}] 格式: img <cy_id> <acc_type> <图片路径>")
logger.info(f"[{self.get_time()}] 格式: img <cy_id> <acc_type> <图片路径>")
else:
print(f"[{self.get_time()}] 未知命令: {cmd}")
logger.info(f"[{self.get_time()}] 未知命令: {cmd}")
except Exception as e:
print(f"[{self.get_time()}] 命令错误: {e}")
logger.info(f"[{self.get_time()}] 命令错误: {e}")
def get_time(self):
"""获取当前时间字符串"""
@@ -1951,20 +1951,20 @@ class QingjianAPIClient:
try:
from mail.email_receiver import email_receiver
if email_receiver.username:
print(f"[{self.get_time()}] 邮件接收已启动，监控: {email_receiver.username}")
logger.info(f"[{self.get_time()}] 邮件接收已启动，监控: {email_receiver.username}")
tasks.append(email_receiver.start())
else:
print(f"[{self.get_time()}] 未配置邮件账号，跳过邮件接收")
logger.info(f"[{self.get_time()}] 未配置邮件账号，跳过邮件接收")
except Exception as e:
print(f"[{self.get_time()}] 邮件接收模块加载失败: {e}")
logger.info(f"[{self.get_time()}] 邮件接收模块加载失败: {e}")
# 启动每日汇总定时任务
try:
from utils.daily_summary import scheduler as daily_scheduler
tasks.append(daily_scheduler())
print(f"[{self.get_time()}] 每日日报定时任务已启动")
logger.info(f"[{self.get_time()}] 每日日报定时任务已启动")
except Exception as e:
print(f"[{self.get_time()}] 日报模块加载失败: {e}")
logger.info(f"[{self.get_time()}] 日报模块加载失败: {e}")
# 设计师在线状态：转人工时按需查询，不再轮询
@@ -1974,17 +1974,17 @@ class QingjianAPIClient:
def _qingjian_ok():
return self.websocket is not None and not getattr(self.websocket, "closed", True)
tasks.append(health_check_loop(_qingjian_ok))
print(f"[{self.get_time()}] 健康检查已启动")
logger.info(f"[{self.get_time()}] 健康检查已启动")
except Exception as e:
print(f"[{self.get_time()}] 健康检查模块加载失败: {e}")
logger.info(f"[{self.get_time()}] 健康检查模块加载失败: {e}")
# 每天早上8点发送启动消息到企微群
try:
from utils.wechat_chat_log import morning_startup_scheduler
tasks.append(morning_startup_scheduler())
print(f"[{self.get_time()}] 早8点企微启动消息已启动")
logger.info(f"[{self.get_time()}] 早8点企微启动消息已启动")
except Exception as e:
print(f"[{self.get_time()}] 企微启动消息模块加载失败: {e}")
logger.info(f"[{self.get_time()}] 企微启动消息模块加载失败: {e}")
await asyncio.gather(*tasks)
@@ -1999,7 +1999,7 @@ if __name__ == "__main__":
try:
asyncio.run(client.run())
except KeyboardInterrupt:
print("\n已停止")
logger.info("\n已停止")
async def _load_task_modules(self):
@@ -2091,3 +2091,4 @@ async def check_and_trigger_tasks_v2(self, data: dict):
except Exception as e:
logger.error(f"检查任务触发失败：{e}")

71
core/workflow.py
View File

@@ -6,7 +6,7 @@
- 图片AI接入点：调用 workflow.image_ai_submit_result(task_id, result_url)
- 消息回调接口：通过 register_send_callback 注入发送函数
"""
import asyncio
import asyncio`r`nimport logging
import os
import uuid
from enum import Enum
@@ -14,7 +14,7 @@ from typing import Optional, Dict, Callable, Awaitable, Any, List
from datetime import datetime
from dataclasses import dataclass, field
_WECHAT_WEBHOOK = os.getenv("WECHAT_WEBHOOK", "")
_WECHAT_WEBHOOK = os.getenv("WECHAT_WEBHOOK", "")`r`nlogger = logging.getLogger("cs_agent")
async def _wechat_notify(content: str):
@@ -30,11 +30,11 @@ async def _wechat_notify(content: str):
})
data = resp.json()
if data.get("errcode") == 0:
print(f"[Workflow通知] 企业微信推送成功 ✓")
logger.info(f"[Workflow通知] 企业微信推送成功 ✓")
else:
print(f"[Workflow通知] 企业微信推送失败: {data}")
logger.info(f"[Workflow通知] 企业微信推送失败: {data}")
except Exception as e:
print(f"[Workflow通知] 推送异常: {e}")
logger.info(f"[Workflow通知] 推送异常: {e}")
from db.customer_db import db
@@ -142,7 +142,7 @@ class CustomerServiceWorkflow:
if requirements:
db.add_requirement(customer_id, requirements)
print(f"[Workflow] 创建任务 {task_id} | 客户: {customer_name} | 操作: {operation}")
logger.info(f"[Workflow] 创建任务 {task_id} | 客户: {customer_name} | 操作: {operation}")
return task_id
def get_task(self, task_id: str) -> Optional[ImageTask]:
@@ -216,7 +216,7 @@ class CustomerServiceWorkflow:
requirements=requirements,
)
print(f"[Workflow] 图片识别完成 | 客户:{customer_id} | 复杂度:{complexity} | 建议报价:{price_hint}")
logger.info(f"[Workflow] 图片识别完成 | 客户:{customer_id} | 复杂度:{complexity} | 建议报价:{price_hint}")
# 通知客服AI报价（把识别结果注入消息，让AI根据结果报价）
if self._send_message:
@@ -266,10 +266,10 @@ class CustomerServiceWorkflow:
if not task:
# 内存任务丢失（重启场景）→ 从客户档案重建
print(f"[Workflow] 付款触发：内存无任务，尝试从客户档案重建 | 客户: {customer_id}")
logger.info(f"[Workflow] 付款触发：内存无任务，尝试从客户档案重建 | 客户: {customer_id}")
task = await self._rebuild_task_from_profile(customer_id, acc_id, acc_type)
if not task:
print(f"[Workflow] 付款触发：客户 {customer_id} 无图片记录，无法重建任务，跳过")
logger.info(f"[Workflow] 付款触发：客户 {customer_id} 无图片记录，无法重建任务，跳过")
await _wechat_notify(
f"⚠️ **付款但无图片**\n"
f"客户：{customer_id}\n"
@@ -279,11 +279,11 @@ class CustomerServiceWorkflow:
return False
if task.status not in (TaskStatus.PENDING,):
print(f"[Workflow] 付款触发：任务 {task.task_id[:8]}... 状态={task.status.value}，跳过")
logger.info(f"[Workflow] 付款触发：任务 {task.task_id[:8]}... 状态={task.status.value}，跳过")
return False
task.operation = task.operation or "enhance"
print(f"[Workflow] 付款确认，启动 Gemini 处理 | 客户: {customer_id} | 任务: {task.task_id[:8]}...")
logger.info(f"[Workflow] 付款确认，启动 Gemini 处理 | 客户: {customer_id} | 任务: {task.task_id[:8]}...")
asyncio.create_task(self._auto_process(task.task_id, acc_id=acc_id, acc_type=acc_type))
return True
@@ -325,10 +325,10 @@ class CustomerServiceWorkflow:
)
self.tasks[task_id] = task
self.customer_active_task[customer_id] = task_id
print(f"[Workflow] 任务已重建 | 客户: {customer_id} | 图片: {image_url[:60]}...")
logger.info(f"[Workflow] 任务已重建 | 客户: {customer_id} | 图片: {image_url[:60]}...")
return task
except Exception as e:
print(f"[Workflow] 任务重建失败: {e}")
logger.info(f"[Workflow] 任务重建失败: {e}")
return None
@staticmethod
@@ -367,7 +367,7 @@ class CustomerServiceWorkflow:
if revision_note:
gemini_prompt = (gemini_prompt or "") + f"\n【客户修改要求】{revision_note}"
print(f"[Workflow] Gemini 开始处理 | 任务: {task_id[:8]}... | 比例: {aspect_ratio} | 透视: {perspective} | 图片: {task.original_image}")
logger.info(f"[Workflow] Gemini 开始处理 | 任务: {task_id[:8]}... | 比例: {aspect_ratio} | 透视: {perspective} | 图片: {task.original_image}")
try:
from image.image_processor import image_processor
from utils.image_queue import run_with_queue
@@ -387,7 +387,7 @@ class CustomerServiceWorkflow:
qa_score = result.get("qa_score", 0)
qa_pass = result.get("qa_pass", True)
qa_issue = result.get("qa_issue", "")
print(f"[Workflow] Gemini 处理完成 | 任务: {task_id[:8]}... | 质检: {qa_score}分 | 尝试: {attempts}次")
logger.info(f"[Workflow] Gemini 处理完成 | 任务: {task_id[:8]}... | 质检: {qa_score}分 | 尝试: {attempts}次")
# 质检未通过（已达重试上限，保留结果但人工跟进）
if not qa_pass:
@@ -408,7 +408,7 @@ class CustomerServiceWorkflow:
)
else:
err_msg = result['message']
print(f"[Workflow] Gemini 处理失败: {err_msg}")
logger.info(f"[Workflow] Gemini 处理失败: {err_msg}")
task.update_status(TaskStatus.FAILED)
# 企业微信预警
await _wechat_notify(
@@ -428,7 +428,7 @@ class CustomerServiceWorkflow:
msg_type=0,
)
except Exception as e:
print(f"[Workflow] 自动处理异常: {e}")
logger.info(f"[Workflow] 自动处理异常: {e}")
task.update_status(TaskStatus.FAILED)
await _wechat_notify(
f"⚠️ **Workflow处理异常**\n"
@@ -459,13 +459,13 @@ class CustomerServiceWorkflow:
"""
task = self.tasks.get(task_id)
if not task:
print(f"[Workflow] 任务不存在: {task_id}")
logger.info(f"[Workflow] 任务不存在: {task_id}")
return False
task.result_url = result_url
task.update_status(TaskStatus.AWAITING_CONFIRM)
print(f"[Workflow] 任务 {task_id} 处理完成，发送给客户确认")
logger.info(f"[Workflow] 任务 {task_id} 处理完成，发送给客户确认")
# 先发结果图片
if self._send_message:
@@ -573,7 +573,7 @@ class CustomerServiceWorkflow:
)
return result.get("success", False)
except Exception as e:
print(f"[Workflow] 邮件发送失败: {e}")
logger.info(f"[Workflow] 邮件发送失败: {e}")
await _wechat_notify(
f"⚠️ **邮件发送失败**\n"
f"客户：{task.customer_id}\n"
@@ -621,17 +621,17 @@ class CustomerServiceWorkflow:
# 尝试从数据库加载
db_task = self.db.get_task(task_id)
if db_task:
print(f"[Workflow] 从数据库加载任务：{task_id[:8]}...")
logger.info(f"[Workflow] 从数据库加载任务：{task_id[:8]}...")
# 可以在这里重建内存任务
else:
print(f"[Workflow] 任务不存在：{task_id}")
logger.info(f"[Workflow] 任务不存在：{task_id}")
return False
# 添加到数据库
success = self.db.add_customer_note(task_id, requirement, changed_by)
if success:
print(f"[Workflow] 客户添加需求：{task_id[:8]}... | {requirement}")
logger.info(f"[Workflow] 客户添加需求：{task_id[:8]}... | {requirement}")
# 如果任务还在待处理状态，通知 AI 客服
if task and task.status.value == 'pending':
@@ -664,12 +664,12 @@ class CustomerServiceWorkflow:
if not task:
db_task = self.db.get_task(task_id)
if not db_task:
print(f"[Workflow] 任务不存在：{task_id}")
logger.info(f"[Workflow] 任务不存在：{task_id}")
return False
# 检查状态，已处理完成的不允许修改
if task and task.status.value in ['completed', 'processing']:
print(f"[Workflow] 任务已开始处理，不允许修改操作：{task_id}")
logger.info(f"[Workflow] 任务已开始处理，不允许修改操作：{task_id}")
if self._send_message:
await self._send_message(
customer_id=customer_id,
@@ -685,7 +685,7 @@ class CustomerServiceWorkflow:
if success and task:
task.operation = new_operation
print(f"[Workflow] 修改操作类型：{task_id[:8]}... -> {new_operation}")
logger.info(f"[Workflow] 修改操作类型：{task_id[:8]}... -> {new_operation}")
if self._send_message:
await self._send_message(
@@ -720,7 +720,7 @@ class CustomerServiceWorkflow:
bool: 是否成功
"""
try:
print(f"[Workflow] 启动查找图片工作流 | 客户：{customer_id}")
logger.info(f"[Workflow] 启动查找图片工作流 | 客户：{customer_id}")
# 1. 创建任务
task_id = self.create_image_task(
@@ -754,7 +754,7 @@ class CustomerServiceWorkflow:
msg_type=0,
)
print(f"[Workflow] 查找图片完成 | 客户：{customer_id} | URL: {tuhui_url}")
logger.info(f"[Workflow] 查找图片完成 | 客户：{customer_id} | URL: {tuhui_url}")
return True
except Exception as e:
@@ -777,7 +777,7 @@ class CustomerServiceWorkflow:
bool: 是否成功
"""
try:
print(f"[Workflow] 启动处理图片工作流 | 客户：{customer_id}")
logger.info(f"[Workflow] 启动处理图片工作流 | 客户：{customer_id}")
# 1. 创建任务
task_id = self.create_image_task(
@@ -803,7 +803,7 @@ class CustomerServiceWorkflow:
# 3. 启动处理
await self.trigger_processing_on_payment(customer_id, acc_id, acc_type)
print(f"[Workflow] 处理图片已启动 | 客户：{customer_id}")
logger.info(f"[Workflow] 处理图片已启动 | 客户：{customer_id}")
return True
except Exception as e:
@@ -828,7 +828,7 @@ class CustomerServiceWorkflow:
bool: 是否成功
"""
try:
print(f"[Workflow] 启动转人工派单工作流 | 客户：{customer_id} | 原因：{reason}")
logger.info(f"[Workflow] 启动转人工派单工作流 | 客户：{customer_id} | 原因：{reason}")
# 1. 创建任务
task_id = self.create_image_task(
@@ -864,7 +864,7 @@ class CustomerServiceWorkflow:
f"请安排设计师上线"
)
print(f"[Workflow] 无人在线 | 客户：{customer_id}")
logger.info(f"[Workflow] 无人在线 | 客户：{customer_id}")
return False
# 3. 派单给在线设计师
@@ -885,7 +885,7 @@ class CustomerServiceWorkflow:
msg_type=0,
)
print(f"[Workflow] 已派单给设计师：{designer} | 客户：{customer_id}")
logger.info(f"[Workflow] 已派单给设计师：{designer} | 客户：{customer_id}")
return True
except Exception as e:
@@ -901,7 +901,7 @@ class CustomerServiceWorkflow:
"""
try:
designers = await self.dispatch_client.get_online_designers()
print(f"[Workflow] 查询在线设计师：{len(designers)}人在线 | {designers}")
logger.info(f"[Workflow] 查询在线设计师：{len(designers)}人在线 | {designers}")
return designers
except Exception as e:
@@ -945,7 +945,7 @@ class CustomerServiceWorkflow:
)
if success:
print(f"[Workflow] 派单成功：{dispatch_task_id} → {designer_name} | 客户：{customer_id}")
logger.info(f"[Workflow] 派单成功：{dispatch_task_id} → {designer_name} | 客户：{customer_id}")
# 企业微信通知
await _wechat_notify(
@@ -969,3 +969,4 @@ class CustomerServiceWorkflow:
# ========== 全局实例 ==========
workflow = CustomerServiceWorkflow()