import os import re import hashlib import logging from typing import List, Optional, Any, Dict from pydantic_ai import Agent, RunContext from pydantic_ai.models.openai import OpenAIChatModel from pydantic_ai.providers.openai import OpenAIProvider from core.schema import StandardMessage, StandardResponse from core.agent_tools import register_agent_tools logger = logging.getLogger("cs_agent") from core.skill_manager import skill_manager def _clip(text: str, limit: int = 1200) -> str: if text is None: return "" text = str(text) if len(text) <= limit: return text return f"{text[:limit]}...(截断, 共{len(text)}字)" def _fmt_time(ts: Any) -> str: s = str(ts or "").strip() if not s: return "--:--:--" if " " in s: return s.split(" ", 1)[1] return s class CustomerServiceBrain: """ 重构后的单一 Agent 大脑： 【全能终极版】统一称呼为“设计师”，支持下线安抚。 """ def __init__(self, model_name: str = None): self.api_key = os.getenv("OPENAI_API_KEY") self.base_url = os.getenv("OPENAI_BASE_URL") self.model_name = model_name or os.getenv("OPENAI_MODEL", "gpt-4o-mini") model = OpenAIChatModel( model_name=self.model_name, provider=OpenAIProvider(api_key=self.api_key, base_url=self.base_url) ) exclude_names = os.getenv("SKILL_EXCLUDE_FROM_PROMPT", "pricing-skill") excluded_skills = [s.strip().lower() for s in exclude_names.split(",") if s.strip()] all_skills = skill_manager.get_all_skills_text(exclude=excluded_skills) logger.info(f"[SkillManager] 已从提示词排除技能: {excluded_skills}") # --- 统一口径后的 System Prompt --- system_prompt = ( "你是一位专注【高清修复】和【找原图】的专业店主。性格干脆，说话自然、专业。\n\n" "【统一称呼规范】\n" "1. 严禁使用'师傅'、'客服'、'专员'等词汇！必须统一称为【设计师】。\n" "2. 未转接前，用第一人称（我/我这边）。例如：'我叫设计师看下'。\n\n" "【核心逻辑】\n" "1. 业务：只聊高清修复和找原图。核心链路：引导发图 -> 问需求 -> 找设计师。\n" "2. **主动引导（关键）**：如果客户【没发图】就问能不能做、问收费，你必须回：'亲亲先发图我看下哈'。\n" "3. **非业务问题**：如果客户问招聘、合作、闲聊等与做图无关的话题，礼貌拒绝：'亲亲咱这边只做图哦，暂不招人哈'。\n" "4. **客户说没有参考图**：如果客户明确说'没有图'、'找不到'、'想让你们帮找'，直接转人工：'好的，我这就叫设计师帮您找哈'。\n" "5. **客户问尺寸/能否打印/退款**：这类问题需要设计师判断，直接转人工：'这个设计师帮您看下哈'。\n" "6. 转接时机：收到图片并明确需求后，立即调用转人工工具，并告知：'收到，正在呼叫设计师核价，稍等哈'。\n" "7. **下线安抚（重要）**：只有当【本次】工具返回 'ERROR_NO_DESIGNER_ONLINE' 时才能说下班。不能根据历史对话或自己猜测说下班！\n" "8. 正在转接中：如果系统提示已在转接，回：'设计师正在赶来，我再帮你催下哈！'。\n" "9. **每次转接必须调用工具**：不要根据之前的结果猜测，每次需要转接都必须重新调用工具检查设计师是否在线。\n\n" "【必杀令 - 严格遵守】\n" "1. 每句回复严禁超过15个字！语气淘宝亲切风，多用'哈'、'呢'。\n" "2. 严禁报价，严禁复读图片已收到的情况。\n" "3. 必须原样输出工具返回的'正在为您转接|'指令。\n" "4. **严禁**说'在呢铁子'！只能说'在呢'或'在呢亲'。\n" "5. **严禁**重复发送相同内容！如果刚说过的话，换一种说法。\n" "6. **严禁**输出任何代码、标记、括号等乱码！只输出自然语言。\n" "7. **严禁**自己臆造'下班'！只有工具返回ERROR才能说下班。\n\n" f"业务参考：\n{all_skills}" ) self.agent = Agent(model=model, system_prompt=system_prompt) register_agent_tools(self.agent) async def think_and_reply(self, msg: StandardMessage, history: Optional[List[dict]] = None) -> StandardResponse: if history is None: history = [] try: user_content = msg.content or "" # 客户已发图：在上下文中明确告知 AI，避免再回"先发图" if msg.image_urls: user_content = ( f"【系统通知：客户已发送 {len(msg.image_urls)} 张图片，不要再让客户发图！" f"请直接问客户需求（找原图还是修复），然后转接设计师】\n{user_content}" ) recent_context = "" if history: lines = [] for h in history[-6:]: role = "客户" if h.get("role") == "user" else "我" content = h.get("content", "") lines.append(f"[{_fmt_time(h.get('timestamp'))}] {role}：{content}") recent_context = "【近期对话回顾】\n" + "\n".join(lines) + "\n----------------\n" full_input = f"{recent_context}现在的对话：{user_content}" logger.info( f"[PROMPT->AI] user={msg.user_id} acc={msg.acc_id} images={len(msg.image_urls)}\n" f"{_clip(full_input)}" ) result = await self.agent.run(full_input, message_history=history) # --- 终极修复：强制截获工具返回的转接指令 --- reply_text = "" # pydantic-ai 1.x 使用 result.output（旧版 0.x 使用 result.data） raw_output = getattr(result, 'output', None) or getattr(result, 'data', None) if isinstance(raw_output, str): reply_text = raw_output # 暴力扫描所有消息片段，寻找转接暗号 found_magic = "" for m in result.all_messages(): if hasattr(m, 'parts'): for part in m.parts: # 检查是否是工具返回片段 if getattr(part, 'part_kind', '') == 'tool-return': content = str(getattr(part, 'content', '')) if "[转移会话]" in content: found_magic = content # 如果 AI 弄丢了暗号，我们强行给它补回来 if found_magic and "[转移会话]" not in reply_text: logger.info(f"[Brain] 检测到 AI 弄丢了转接暗号，正在强制恢复: {found_magic[:30]}...") reply_text = found_magic # ---------------------------------------- # 清理模型泄露的内部标记/乱码（覆盖所有已知格式） reply_text = re.sub(r'\[\]<\|[^|]+\|>', '', reply_text) reply_text = re.sub(r'<\|[^|]*\|>', '', reply_text) reply_text = re.sub(r'\[Function[^\]]*\]', '', reply_text) reply_text = re.sub(r'\[/?Tool[^\]]*\]', '', reply_text) reply_text = re.sub(r']*>', '', reply_text, flags=re.IGNORECASE) reply_text = re.sub(r']*>.*?]*>', '', reply_text, flags=re.DOTALL) reply_text = re.sub(r']*>.*', '', reply_text, flags=re.DOTALL) reply_text = re.sub(r']*>', '', reply_text) reply_text = re.sub(r'```[^`]*```', '', reply_text) reply_text = re.sub(r'\{["\'][^}]+\}', '', reply_text) reply_text = re.sub(r'AgentRunResult\([^)]*\)', '', reply_text) reply_text = re.sub(r'\[/?[A-Z][a-zA-Z]*(?:Call|End|Start|Result|Return)[^\]]*\]', '', reply_text) reply_text = re.sub(r'[\[\]]{2,}', '', reply_text) reply_text = reply_text.strip() # 过滤"在呢铁子" if "在呢铁子" in reply_text: reply_text = reply_text.replace("在呢铁子", "在呢亲") if not reply_text: reply_text = "稍等我看看。" logger.info(f"[THINK/RAW_OUTPUT] user={msg.user_id}\n{_clip(reply_text)}") need_transfer = "[转移会话]" in reply_text return StandardResponse( reply_content=reply_text, need_transfer=need_transfer, metadata={"acc_id": msg.acc_id, "acc_type": msg.acc_type} ) except Exception as e: logger.error(f"[Brain Error]: {e}") return StandardResponse(reply_content="好哒，设计师正在看图，稍等回你。", metadata={"acc_id": msg.acc_id})