98 lines
4.2 KiB
Python
98 lines
4.2 KiB
Python
import logging
|
||
import asyncio
|
||
from datetime import datetime
|
||
from typing import List, Optional, Dict, Any
|
||
from pydantic import BaseModel, Field
|
||
from pydantic_ai import RunContext
|
||
from core.schema import StandardResponse
|
||
from services.dispatch_service import dispatch_service
|
||
from db.chat_log_db import get_conversation
|
||
|
||
logger = logging.getLogger("cs_agent")
|
||
|
||
|
||
async def transfer_to_human_tool(ctx: RunContext[Any], reason: str = Field(description="转人工的原因")) -> str:
|
||
"""
|
||
【核心工具】执行转人工逻辑。
|
||
获取设计师姓名并生成精准转接指令。
|
||
"""
|
||
logger.info(f"[Tool] 尝试呼叫设计师接手: {reason}")
|
||
|
||
designer_name = await dispatch_service.assign_designer()
|
||
|
||
if designer_name:
|
||
magic_cmd = f"正在为您转接|[转移会话],{designer_name},无原因"
|
||
logger.info(f"[Tool] 成功呼叫设计师: {designer_name}")
|
||
return magic_cmd
|
||
else:
|
||
hour = datetime.now().hour
|
||
logger.warning(f"[Tool] 派单失败：设计师们不在位 (当前{hour}点)")
|
||
if 0 <= hour < 9:
|
||
return "ERROR_DESIGNER_NOT_STARTED：现在设计师还没上班，你告诉客户需求记下了，上班后第一时间处理。不要说下班。"
|
||
elif 22 <= hour or hour < 1:
|
||
return "ERROR_DESIGNER_OFFLINE：设计师已下班，你告诉客户需求记下了，明天第一时间回复。"
|
||
else:
|
||
return "ERROR_DESIGNER_BUSY：设计师暂时不在位，你告诉客户稍等，马上帮忙联系设计师。不要说下班。"
|
||
|
||
|
||
async def check_order_status_tool(ctx: RunContext[Any], customer_id: str = Field(description="客户ID")) -> str:
|
||
"""查询订单状态。"""
|
||
return "我在帮你加急处理中，稍等哈。"
|
||
|
||
|
||
async def lookup_chat_history_tool(
|
||
ctx: RunContext[Any],
|
||
customer_id: str = Field(description="客户ID，从当前对话上下文中获取"),
|
||
num_messages: int = Field(default=30, description="要查询的历史消息条数，默认30条"),
|
||
) -> str:
|
||
"""
|
||
【历史记录查询工具】查询该客户的历史聊天记录。
|
||
使用场景：
|
||
- 客户说"之前聊过"、"上次"、"你看聊天记录"、"我发过图了"等暗示有历史对话时
|
||
- 客户第二次来访、追问进度、催单时
|
||
- 你不确定客户之前是否发过图或说过需求时
|
||
必须先调用此工具回顾历史，再回复客户，避免重复要求客户发图。
|
||
"""
|
||
logger.info(f"[Tool] 查询历史记录: customer_id={customer_id}, limit={num_messages}")
|
||
try:
|
||
rows = await asyncio.to_thread(get_conversation, customer_id, limit=num_messages)
|
||
if not rows:
|
||
return f"该客户({customer_id})暂无历史聊天记录。"
|
||
|
||
lines = []
|
||
has_images = False
|
||
customer_needs = []
|
||
for r in rows:
|
||
role = "客户" if r["direction"] == "in" else "客服"
|
||
ts = str(r.get("timestamp", ""))
|
||
msg = r.get("message", "")
|
||
line = f"[{ts}] {role}：{msg}"
|
||
lines.append(line)
|
||
if r["direction"] == "in":
|
||
if "已收到" in msg and "图" in msg:
|
||
has_images = True
|
||
if any(k in msg for k in ["找原图", "修复", "高清", "去背景", "抠图", "做衣服", "打印"]):
|
||
customer_needs.append(msg[:60])
|
||
|
||
summary_parts = [f"共{len(rows)}条历史消息。"]
|
||
if has_images:
|
||
summary_parts.append("⚠️ 客户之前已经发过图片！不要再让客户发图！")
|
||
if customer_needs:
|
||
summary_parts.append(f"客户曾表达的需求：{'；'.join(customer_needs[:3])}")
|
||
|
||
summary = " ".join(summary_parts)
|
||
history_text = "\n".join(lines[-30:])
|
||
return f"【历史记录摘要】{summary}\n\n【详细记录】\n{history_text}"
|
||
|
||
except Exception as e:
|
||
logger.error(f"[Tool] 查询历史记录失败: {e}")
|
||
return f"查询历史记录失败: {e}"
|
||
|
||
|
||
def register_agent_tools(agent: Any):
|
||
"""注册工具"""
|
||
agent.tool(transfer_to_human_tool)
|
||
agent.tool(check_order_status_tool)
|
||
agent.tool(lookup_chat_history_tool)
|
||
logger.info("[Agent] 工具箱已更新：含转人工、订单查询、历史记录查询。")
|