diff --git a/core/pydantic_ai_agent.py b/core/pydantic_ai_agent.py index 0d78a7e..aa7be2e 100755 --- a/core/pydantic_ai_agent.py +++ b/core/pydantic_ai_agent.py @@ -10,6 +10,7 @@ import glob import asyncio import random import hashlib +import re from typing import Optional, Dict, List, Any, Tuple from datetime import datetime from pydantic import BaseModel, Field @@ -1056,6 +1057,25 @@ class CustomerServiceAgent: - 黄色/擦边/涉政/政治人物/政治事件/政治图片等不接单,礼貌拒绝 - 输出不超过1句话""" + @staticmethod + def _is_political_inquiry(text: str) -> bool: + """文本前置风控:政治人物/政治事件/政治图片相关询问一律拒绝。""" + s = (text or "").strip().lower() + if not s: + return False + kw = ( + "政治", "涉政", "党政", "政治人物", "政治事件", "政治图片", "政治海报", "政治宣传", + "领导人", "伟人", "元帅", "将军", "红色人物", "党史", + "天安门", "人民大会堂", "中南海", + "习近平", "毛泽东", "邓小平", "江泽民", "胡锦涛", "李克强", "周恩来", + "特朗普", "拜登", "普京", "泽连斯基", + "trump", "biden", "putin", "zelensky", "xi jinping", + ) + if any(k in s for k in kw): + return True + # 兜底:类似“有没有十大元帅的照片/图片” + return bool(re.search(r"(元帅|将军|领导人|政治人物|政治事件).*(照片|图片|头像|原图)?", s)) + def _get_customer_profile_context(self, customer_id: str) -> str: """从数据库读取客户画像,注入给 AI。含个性化语气、报价策略、主动预测、近期对话。""" try: @@ -1314,6 +1334,21 @@ class CustomerServiceAgent: print(f"[Agent] 冷却期静默(距上次回复 {elapsed}s):{message.msg!r}") return AgentResponse(reply="", should_reply=False, need_transfer=False) + # 前置风控:客户文本一旦命中政治/敏感询问,直接拒绝,避免“发图我看看”类答非所问 + try: + from utils.content_filter import should_block_customer + if should_block_customer(message.msg) or self._is_political_inquiry(message.msg): + reply = "这类不做哈,政治相关图片和人物都不接。" + state.last_reply_at = datetime.now() + print(f"{self.C_REPLY}[REPLY->CUSTOMER]{self.C_RESET} {reply}") + return AgentResponse(reply=reply, should_reply=True, need_transfer=False) + except Exception: + if self._is_political_inquiry(message.msg): + reply = "这类不做哈,政治相关图片和人物都不接。" + state.last_reply_at = datetime.now() + print(f"{self.C_REPLY}[REPLY->CUSTOMER]{self.C_RESET} {reply}") + return AgentResponse(reply=reply, should_reply=True, need_transfer=False) + # 检测售前/售后 new_stage = self._detect_stage(message.msg) if new_stage != state.stage: @@ -1461,11 +1496,12 @@ class CustomerServiceAgent: "特朗普", "拜登", "普京", "泽连斯基", ] target_agent = self.agent_after_sale if state.stage == "售后" else self.agent - if any(k in msg_lower for k in risk_kw): + risk_hit = any(k in msg_lower for k in risk_kw) or self._is_political_inquiry(message.msg) + if risk_hit: target_agent = self.agent_risk elif any(k in message.msg for k in order_markers): target_agent = self.agent_order - if any(k in msg_lower for k in processing_kw): + elif any(k in msg_lower for k in processing_kw): target_agent = self.agent_processing elif any(k in msg_lower for k in pricing_kw): target_agent = self.agent_pricing