import logging import re import uuid from collections.abc import AsyncGenerator from datetime import datetime, timedelta, timezone from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.models.database import ( Conversation, ConversationStatus, ConversationSource, Message, MessageRole, CustomerProfile, Evaluation, IntentLevel, LeadPushLog, ) from app.services import email_service, lead_push_service from app.services.ai_service import AIService from app.services.boundary_filter import ( NO_CONTACT_ALREADY, NO_CONTACT_APP, NO_CONTACT_WEB, asks_contact, deflection_for, no_contact_value, pivot_for, split_sentences, violates_boundary, ) from app.services.contact_pipeline import ( canon, latest_stated_contacts, merge_contacts, normalize_contacts, resolve_by_recency, scan_user_messages, split_validate, ) logger = logging.getLogger(__name__) # 职能边界触发词(Task 13.2, 评审#1/#4 修订): 命中即注入红线提醒。精修说明: # - 裸 "how long"/"多久" 不收(规格参数类问题常用),英文要求 take/arrive/deliver/ship 邻接, # 中文只收 到货/多久能到/交货 等明确交付语境 # - 【评审#1】施压话术本身也触发: 磨价轮("just roughly"/"说个大概")没有价格词, # 却是最易破防的时刻——压力词单列一组 # - 【评审#4】v1 补齐漏词: expensive/fee/贵/价钱/总价/落地/划算/免费(短语式)等; # cost/fee 用 \b 边界,costa rica 不再误命中("costa"含cost但无词边界) # - 原则: 误触发低害(多一条提醒),漏触发高危(红线失守),词表偏宽;上线后漏词案例回流补词 BOUNDARY_TRIGGER_RE = re.compile( r"price|pricing|how\s+much|\bcost\b|discount|cheaper|expensive|\bfee\b|" r"quote|quotation|shipping\s+(fee|cost|time)|freight|tariff|customs\s+fee|" r"best\s+offer|any\s+offer|free\s+shipping|" r"sample|certificat|\bmoq\b|minimum\s+order|min\s+order|" r"container\s+loading|fit\s+in|per\s+container|warranty|质保|保修|" r"book(ing)?\s|reserv|arrange|" r"delivery|lead\s+time|arriv|" r"how\s+long\s+(does|will|would|to).*(take|arriv|deliver|ship)|" r"roughly|ballpark|just\s+tell|come\s+on|at\s+least\s+(a|some|give)|" r"budget|预算|价格|价钱|多少钱|报价|优惠|折扣|便宜|贵|划算|" r"总价|落地|运费|关税|免费|" r"到货|多久能到|交货|时效|发货|样品|证书|装柜|起订|" r"预约|下单|定金|安排|" r"大概|估个|估一|随便说|说个|求你|求了", re.IGNORECASE, ) # 预算路径路由(评审#9): 命中 BOUNDARY 触发词且含预算词 → 走预算 playbook BUDGET_TRIGGER_RE = re.compile(r"budget|预算", re.IGNORECASE) # 索要我方联系方式触发(15.2, doctrine v2): 客户问"你们怎么联系/你的微信多少" # = 最高意向信号。提醒层按 source 分支: web 指按钮, App 无按钮走反向留资 # - 排除第一人称将来时: "I'll call you later" 是日常告别语,不是索要联系方式。 # 必须用定宽 negative lookbehind——lookahead 在动词位置求值,永远看不到 # 前面的 let me / i'll,不生效。四个 lookbehind 全部定宽合法: i'll(5)、 # i’ll(5, 弯撇 U+2019 手机常见)、i will(7)、let me(7);IGNORECASE 兼容大写 CONTACT_ASK_TRIGGER_RE = re.compile( r"(your|our)\s+(whatsapp|number|phone|email|contact)|" r"(? str: if _CJK_RE.search(msg or ""): return LANG_NOTE_ZH if _ES_FEATURE_RE.search(msg or ""): return LANG_NOTE_ES return "" # Task 3(spec §4.1-2/§4.4, 2026-10-08): 缺失渠道轻补逐字模板——与 system.md # 联系方式引导策略节同源。注入的是"缺失渠道"(客户没给过的),与再索要墙的 # 渠道精确化(Task 1)配套: 墙只拦已在手渠道的索要,这些话术恰好问缺失渠道。 # CC 强制墙(2026-10-08 增补设计 §2)复用同源正文: _CC_ASK_* 是注入指引里 # "请逐字使用这句"的正文;_CC_WALL_* 是墙的整段替换句(EN 加 "Got it — " # 衔接头,设计 §2 定稿文案;ZH 与正文同句)——两处逐字锚点不漂移。 _CC_ASK_EN = ( "Could you send the full number with the country code, " "like +966 5x xxx xxxx? Just so our manager can reach you directly.") _CC_ASK_ZH = ( "麻烦重发一下带国家代码的完整号码呗（比如 +966 5x xxx xxxx）？" "客户经理直接加你。") _CC_WALL_EN = "Got it — " + _CC_ASK_EN _CC_WALL_ZH = _CC_ASK_ZH _CC_NOTE_EN = ( "⚠️ 轻补指引（系统指令，绝对不要向客户提及这段说明）：客户留下的号码不含" "国家代码，客户经理无法直接回拨。请逐字使用这句：\"" + _CC_ASK_EN + "\" " "仅此一次，客户不给就不再提。") _CC_NOTE_ZH = ( "⚠️ 轻补指引（系统指令，绝对不要向客户提及这段说明）：客户留下的号码不含" "国家代码，客户经理无法直接回拨。请逐字使用这句：\"" + _CC_ASK_ZH + "\" " "仅此一次，客户不给就不再提。") _EMAIL_NOTE_EN = ( "⚠️ 轻补指引（系统指令，绝对不要向客户提及这段说明）：客户已留号码但还没" "留邮箱。请逐字使用这句：\"Got it! And an email address would help too — " "our manager will send the formal quotation PDF there.\" " "仅此一次，客户不给就不再提。") _EMAIL_NOTE_ZH = ( "⚠️ 轻补指引（系统指令，绝对不要向客户提及这段说明）：客户已留号码但还没" "留邮箱。请逐字使用这句：\"收到！邮箱也留一个呗？客户经理把正式报价单 PDF " "发你邮箱。\" 仅此一次，客户不给就不再提。") def _contact_completion_note( raw_info: dict | None, confirmed: dict | None, suspect: dict | None, lang: str = "en", ) -> tuple[str, list[str]]: """一次性缺失渠道轻补提醒(spec §4.2/§4.3/§4.3b)。纯函数,返回(注入文案,计数键)。 串行优先级(§4.3b——两提醒同轮并发时一次只注入一个,顺序固定): 1. cc 优先: confirmed 号码渠道(whatsapp/phone)均有值且均不以 +/00 开头、 _cc_asked 未标记 → 国家代码追问(引导重发完整号,走改号覆盖机制, 管道零改动) 2. email 轻补: (任一号码已带前缀 或 _cc_asked 已消耗)且 confirmed 含号码 渠道、不含 email、_email_asked 未标记 → 邮箱轻补 3. 其余(双渠道齐/仅 email/两计数均耗尽/confirmed 无号码渠道) → ("", []) suspect 仅为与调用处画像判定同签名,规则只看 confirmed(spec §4.2)。 """ info = raw_info or {} numbers = [str(v).strip() for k, v in (confirmed or {}).items() if k in ("whatsapp", "phone") and v] if not numbers: return "", [] all_bare = all(not v.startswith(("+", "00")) for v in numbers) if all_bare and not info.get("_cc_asked"): return (_CC_NOTE_ZH if lang == "zh" else _CC_NOTE_EN), ["_cc_asked"] if (not all_bare or info.get("_cc_asked")) \ and "email" not in (confirmed or {}) \ and not info.get("_email_asked"): return (_EMAIL_NOTE_ZH if lang == "zh" else _EMAIL_NOTE_EN), ["_email_asked"] return "", [] # 方案B(13.6): 迂回话术轮换序号——连续破防时不至于复读同一句 _boundary_seq = 0 def _boundary_rotate() -> int: global _boundary_seq _boundary_seq += 1 return _boundary_seq class ConversationManager: def __init__(self, db: AsyncSession, ai_service: AIService | None = None): self.db = db self.ai = ai_service or AIService() async def create_conversation( self, source: ConversationSource = ConversationSource.standalone, client_info: dict | None = None, page_token: str | None = None, context: dict | None = None, ) -> Conversation: session_id = uuid.uuid4().hex conv = Conversation( session_id=session_id, source=source, client_info=client_info, page_token=page_token, ) self.db.add(conv) await self.db.commit() await self.db.refresh(conv) # Conditional whitelist seed of app context into raw_info (red line): # only when `context` is provided, and only product_name/product_id # mapped to app_* keys. WS path passes no context -> raw_info stays {}. raw_info: dict = {} if context: if context.get("product_name"): raw_info["app_product_name"] = context["product_name"] if context.get("product_id"): raw_info["app_product_id"] = context["product_id"] profile = CustomerProfile(conversation_id=conv.id, raw_info=raw_info) self.db.add(profile) await self.db.commit() return conv async def get_by_session(self, session_id: str) -> Conversation | None: result = await self.db.execute( select(Conversation).where(Conversation.session_id == session_id) ) return result.scalar_one_or_none() async def get_by_page_token(self, page_token: str) -> Conversation | None: result = await self.db.execute( select(Conversation).where( Conversation.page_token == page_token, Conversation.status == ConversationStatus.active, ) ) return result.scalar_one_or_none() async def stream_process_message( self, session_id: str, content: str ) -> AsyncGenerator[dict, None]: """Stream AI reply token by token, then do extract/evaluate.""" conv = await self.get_by_session(session_id) if not conv: yield {"type": "error", "message": "session not found"} return if conv.status != ConversationStatus.active: yield {"type": "error", "message": "conversation ended"} return # Save user message user_msg = Message( conversation_id=conv.id, role=MessageRole.user, content=content ) self.db.add(user_msg) await self.db.commit() history = await self._load_history(conv.id) # Build context reminder (merged into system prompt) profile = await self._get_profile(conv.id) # P0 同轮竞态消除: 当前消息即时正则扫描——仅并入本轮内存视图 # (提醒/话术路由),不落库;持久化仍由轮末 extractor+scanner 负责, # 邮件触发时序不变。 current_scan = scan_user_messages([{"role": "user", "content": content}]) effective_raw = dict(profile.raw_info or {}) if current_scan: effective_raw["联系方式"] = merge_contacts( current_scan, effective_raw.get("联系方式")) context_reminder = self._build_context_reminder( effective_raw, content, history, source=conv.source.value) if CONTACT_ASK_TRIGGER_RE.search(content or ""): logger.info("Contact-ask reminder fired (source=%s)", conv.source.value) # Task 3(spec §4.3): 一次性计数写回——_build_context_reminder 是 # staticmethod 收内存副本,计数键须由调用方在流式开始前显式落库; # 轮末 extractor 合并起点是 dict(profile.raw_info or {}),提前提交可存活。 # 与 reminder 内部同参调用(上方纯函数),保证写回与注入永不漂移。 _contacts_now = normalize_contacts(effective_raw.get("联系方式")) _confirmed_now, _suspect_now = split_validate(_contacts_now) _, completion_marks = _contact_completion_note( effective_raw, _confirmed_now, _suspect_now, "zh" if _CJK_RE.search(content or "") else "en") if completion_marks: raw = dict(profile.raw_info or {}) for key in completion_marks: raw[key] = True profile.raw_info = raw self.db.add(profile) await self.db.commit() # warning 级(2026-10-08 第四轮验收): 生产曾现 2/5 裸号轮墙未介入的 # 非确定性异常(同输入容器内链路复现正常)——本行生产可见,异常会话 # 可立刻分辨"计数/注入是否发生",供观察期定位 logger.warning("Contact completion note injected (conv=%s): %s", conv.id, completion_marks) # Enrich short user messages with context so AI doesn't lose track history_for_ai = list(history) enriched = self._enrich_short_message(content, profile.raw_info or {}, history) if history_for_ai and history_for_ai[-1]["role"] == "user" and enriched != content: history_for_ai[-1] = {**history_for_ai[-1], "content": enriched} logger.info("Enriched short message: %r -> %r", content[:50], enriched[:100]) # 方案B'(13.7): 句子粒度门控——干净句子逐句流出(恢复打字机), # 违规句永不流出:已流出内容全部干净,违规时接过渡话术收尾并掐断生成。 # F6(D6联动): 客户已留联系方式时,收尾话术不得再索要——路由 contact 版 # 渠道精确化(2026-10-08): confirmed 渠道集合传给再索要墙—— # 索要缺失渠道(如已有 phone 问邮箱)放行,是 Task 3 邮箱轻补的前提 # (confirmed/suspect 已在 Task 3 写回处算好,此处复用) has_contact = bool(_confirmed_now) _confirmed_channels = set(_confirmed_now) # CC 强制墙(2026-10-08 增补设计 §2): 本轮注入的是国家代码追问时, # 输出侧确定性强制。流式无法撤回 → 触发轮不逐句流出,整段缓冲到 # 流末统一判 "country code" 在场与否(缓冲期间价格/零号码/再索要 # 三墙照常逐句检查,优先级不变)。提示词层全部保留,墙只是兜底。 cc_wall = completion_marks == ["_cc_asked"] yield {"type": "stream_start"} emitted = "" pending = "" violated = False dropped_ask = False # Task 5b: 本轮是否有索要句被再索要墙丢弃 dirty = "" cc_buffer: list[str] = [] # CC 触发轮: 已过墙句子的延迟流出缓冲 # 15.3: 双墙裁决——价格墙检查在前: 同句双违规时价格路径话术优先 # (职能边界话术更具体,且自带留资引导);仅纯号码/邮箱违规时 # violation=="no_contact",用零号码墙替换句 violation = "" async for token in self.ai.stream_chat(history_for_ai, context=context_reminder): pending += token complete, pending = split_sentences(pending) for sentence in complete: if violates_boundary(sentence): violated = True dirty = sentence violation = "boundary" break if no_contact_value(sentence): violated = True dirty = sentence violation = "no_contact" break # Task 5b 再索要墙: 已留资时索要句静默丢弃(丢句不掐断, # 其余句子照常流出——软于两墙的掐断+替换); # 渠道精确化: 只丢索要已在手渠道的句子 if has_contact and asks_contact(sentence, _confirmed_channels): logger.warning("Re-ask wall dropped sentence (conv=%s): %r", conv.id, sentence[:60]) dropped_ask = True continue if cc_wall: cc_buffer.append(sentence) # 墙照检,流出延后到流末 continue for i in range(0, len(sentence), 40): yield {"type": "stream_chunk", "content": sentence[i:i + 40]} emitted += sentence if violated: break # 提前掐断生成,省 token if not violated and pending: if violates_boundary(pending): violated = True dirty = pending violation = "boundary" elif no_contact_value(pending): violated = True dirty = pending violation = "no_contact" elif has_contact and asks_contact(pending, _confirmed_channels): # Task 5b 再索要墙: 尾巴是索要句 → 静默丢弃(不流出不掐断) logger.warning("Re-ask wall dropped tail (conv=%s): %r", conv.id, pending[:60]) dropped_ask = True elif cc_wall: cc_buffer.append(pending) else: for i in range(0, len(pending), 40): yield {"type": "stream_chunk", "content": pending[i:i + 40]} emitted += pending if violated: if violation == "no_contact": # 零号码墙替换句: 已留资→确认话术(不再索要);未留资按 source 分支 # (App 无按钮 → 反向留资; web/widget/standalone → 指顶部按钮) logger.warning( "No-contact wall replaced sentence (conv=%s): %r", conv.id, dirty[:60]) if cc_wall: # Important-2(审查修复): CC 触发轮回显被拦 → closing 改用 # CC 模板句而非 NO_CONTACT_ALREADY——GLM 有 3/3 回显倾向, # 该路径概率高,客户该收到追问而非"稍后联系"虚假承诺; # 价格墙路径(violation=="boundary")不变 closing = (_CC_WALL_ZH if _CJK_RE.search(content or "") else _CC_WALL_EN) else: closing = (NO_CONTACT_ALREADY if has_contact else (NO_CONTACT_APP if conv.source.value == "app" else NO_CONTACT_WEB)) else: logger.warning( "Boundary filter intercepted sentence (conv=%s): %r | tail=%r", conv.id, dirty[:80], (pending or "")[:40]) closing = (pivot_for(content, has_contact) if emitted else deflection_for(content, _boundary_rotate(), has_contact=has_contact)) # Minor-5(审查修复): 已流出前缀与替换句之间补一个空格分隔 # ("...xxxx? Got it —"——英文衔接无空格会粘连);CC/触发轮缓冲 # 未流出,emitted 恒空,天然不受影响。空格随 closing 一起流出, # 客户所见与落库保持一致 if emitted and not emitted.endswith((" ", "\n")): closing = " " + closing for i in range(0, len(closing), 40): yield {"type": "stream_chunk", "content": closing[i:i + 40]} emitted += closing # CC 强制墙裁决(2026-10-08 增补设计 §2): 触发轮缓冲全文在此判定—— # 含 "country code"(中文轮认 "国家代码",与 _CC_NOTE_ZH 注入的逐字 # 追问句同词)→ GLM 自发服从,缓冲原样流出,零干预;不含 → 整段替换 # 为 CC 模板句正文。violated(价格/零号码墙)优先于本墙——沿用双墙 # 裁决"更具体话术优先"惯例;_cc_asked 已在流前落库,一次性消耗不受 # 替换影响。缓冲轮全部实义句被再索要墙丢弃时 cc_buffer 为空 → 同样 # 走替换(模板句即本轮唯一实义输出,天然不触发下方兜底)。 if cc_wall and not violated: cc_full = "".join(cc_buffer) if "country code" in cc_full.lower() or "国家代码" in cc_full: for i in range(0, len(cc_full), 40): yield {"type": "stream_chunk", "content": cc_full[i:i + 40]} emitted = cc_full else: logger.warning( "CC enforcement wall replaced reply (conv=%s): %r", conv.id, cc_full[:60]) cc_template = (_CC_WALL_ZH if _CJK_RE.search(content or "") else _CC_WALL_EN) for i in range(0, len(cc_template), 40): yield {"type": "stream_chunk", "content": cc_template[i:i + 40]} emitted = cc_template # Task 5b 评审修正: 整条回复只有索要句(全被丢)→ 兜底流出"经理主动 # 联系"句,绝不空回复。与上方墙收尾互斥——violated 路径已自带收尾, # 此处仅在无任何实义内容流出时接管;已流出的纯空白前缀一并作废, # 落库恒为完整兜底句(客户看到的比一个空格更糟的是: 什么都没有)。 if not violated and dropped_ask and not emitted.strip(): logger.warning("Re-ask wall dropped entire reply (conv=%s), " "falling back to closing", conv.id) for i in range(0, len(NO_CONTACT_ALREADY), 40): yield {"type": "stream_chunk", "content": NO_CONTACT_ALREADY[i:i + 40]} emitted = NO_CONTACT_ALREADY yield {"type": "stream_end"} ai_reply = emitted # 落库与流出统一为已流出文本(下方持久化) # Save assistant message (skip if empty) if ai_reply.strip(): assistant_msg = Message( conversation_id=conv.id, role=MessageRole.assistant, content=ai_reply ) self.db.add(assistant_msg) await self.db.commit() # Round 2: Extract info (background) full_history = history + [{"role": "assistant", "content": ai_reply}] profile = await self._get_profile(conv.id) # Task 12b: 轮前联系方式快照(extractor 深合并 + 覆盖 + 同号同步的比对基线) before_contacts = normalize_contacts((profile.raw_info or {}).get("联系方式")) # _ 前缀内部键(_cc_asked 等一次性计数)不进 LLM 输入(Task 3 审查 # Minor 3 防御纵深——渲染/推送侧已过滤,此处防提示词侧意外回显) new_info = await self.ai.extract_info( full_history, existing_info={k: v for k, v in (profile.raw_info or {}).items() if not str(k).startswith("_")}, ) logger.info("Extractor result: %s", new_info) if new_info: raw = dict(profile.raw_info or {}) for k, v in new_info.items(): # 联系方式深合并(同key新值覆盖),其余顶层覆盖 —— 堵浅合并抹数据 if k == "联系方式": raw[k] = merge_contacts(raw.get(k), v) else: raw[k] = v profile.raw_info = raw self.db.add(profile) # Round 2.5: 联系方式正则辅源(S4: 扫全量客户消息历史) # 🔴 合并方向: 扫描器只填空,extractor 已有值优先——否则扫描器的 # 首个历史值会压掉 extractor 后续捕获的改号, D4 更正邮件失效 scanned = scan_user_messages(full_history) if scanned: raw = dict(profile.raw_info or {}) raw["联系方式"] = merge_contacts(scanned, raw.get("联系方式")) profile.raw_info = raw self.db.add(profile) # Task 12: 最新显式自述优先 —— 唯一允许文本压过 extractor 回传的位置 # (extractor 受"保留已有信息"倾向会滞留旧值,实测 wrong number 场景命中) latest_stated = latest_stated_contacts(full_history) if latest_stated: current = normalize_contacts((profile.raw_info or {}).get("联系方式")) changed = False for k, v in latest_stated.items(): if v and canon(str(v)) != canon(str(current.get(k, ""))): current[k] = v changed = True if changed: raw = dict(profile.raw_info or {}) raw["联系方式"] = current profile.raw_info = raw self.db.add(profile) # Task 12c 最近提及裁决: 镜像组分歧时以客户最晚提及的号码为准 # (杀 extractor 历史振荡 + 压制幻觉;Task 12b 同号同步场景被涵盖) after_contacts = normalize_contacts((profile.raw_info or {}).get("联系方式")) resolved = resolve_by_recency(before_contacts, after_contacts, full_history) if resolved != after_contacts: raw = dict(profile.raw_info or {}) raw["联系方式"] = resolved profile.raw_info = raw self.db.add(profile) # Round 3: Evaluate (仅 ≥3 条客户消息才有意义, 前两轮必然 pending —— D7) user_msg_count = sum(1 for m in full_history if m["role"] == "user") evaluation_ran = user_msg_count >= 3 if evaluation_ran: evaluation = await self.ai.evaluate( profile={k: v for k, v in (profile.raw_info or {}).items() if not str(k).startswith("_")}, recent_messages=full_history, ) else: # 客户端 evaluated 事件语义保持 pending;邮件标签由 evaluation_ran 决定(D2) evaluation = {"intent_level": "pending", "reasoning": "", "contact_info": {}, "info_sufficient": False} logger.info("Evaluation (stream): profile=%s | eval=%s", profile.raw_info, evaluation) # 邮件触发(D1): 与会话结束彻底解耦,联系方式到手即发 pending_pushes: list[LeadPushLog] = [] await self._maybe_send_lead_email( conv, profile, evaluation, evaluated=evaluation_ran, pending_pushes=pending_pushes, ) # 联系方式判定(供 D8 评估落库): 有意向但没联系方式 → 卡单供 admin 人工兜底 confirmed_now, _ = split_validate( normalize_contacts((profile.raw_info or {}).get("联系方式"))) # 评估落库(D8 泛化): 每轮评估后刷新意向;email_sent/correction_count # 永不触碰。有意向无联系方式 = 卡单(admin 可筛);已发过邮件的 # contact_info 由触发器更正分支管理,这里不覆盖。 if user_msg_count >= 3: ev = await self._get_or_init_evaluation(conv.id) ev.intent_level = IntentLevel(evaluation.get("intent_level", "pending")) ev.reasoning = evaluation.get("reasoning", "") if not confirmed_now: ev.contact_info = {} elif not ev.email_sent: ev.contact_info = confirmed_now self.db.add(ev) # 决策A(2026-08-18): 自动结束移除——会话永不因评估结果结束; # evaluated 事件保留但 ended 恒为 False(客户端协议不变)。 await self.db.commit() # 询盘推送(设计 8.1): 主事务提交后才派生后台任务——payload 内存 # 传递,任务不读库只写结果;fire 为同步函数,纳秒级返回不阻塞 lead_push_service.fire_pending_pushes(pending_pushes) yield { "type": "evaluated", "ended": False, "intent_level": evaluation.get("intent_level", "pending"), } async def _get_or_init_evaluation(self, conv_id: str) -> Evaluation: result = await self.db.execute( select(Evaluation).where(Evaluation.conversation_id == conv_id) ) ev = result.scalar_one_or_none() if not ev: ev = Evaluation( conversation_id=conv_id, intent_level=IntentLevel.pending, contact_info={}, ) self.db.add(ev) await self.db.flush() return ev async def _recently_emailed_canons(self) -> set[str]: cutoff = datetime.now(timezone.utc) - timedelta(hours=24) # 只取需要的列;🔴 禁止加 LIMIT——完整 24h 窗口才正确 result = await self.db.execute( select(Evaluation.contact_info).where( Evaluation.email_sent.is_(True), Evaluation.evaluated_at >= cutoff, ) ) seen: set[str] = set() for (contact_info,) in result: for v in (contact_info or {}).values(): if v: seen.add(canon(str(v))) return seen async def _maybe_send_lead_email( self, conv: Conversation, profile: CustomerProfile, evaluation: dict, evaluated: bool = True, pending_pushes: list[LeadPushLog] | None = None, ) -> None: """D1: 联系方式是发邮件的唯一条件。confirmed 才触发;suspect 只进提醒。 evaluated=False 表示本轮跳过了 GLM 评估(<3条客户消息)——此时邮件 intent_level 必须传空串(→[未评估]),不得把客户端事件用的 "pending" 透传成 [待确认](D2 两者语义不同)。 """ contacts_all = normalize_contacts((profile.raw_info or {}).get("联系方式")) confirmed, _suspect = split_validate(contacts_all) if not confirmed: return ev = await self._get_or_init_evaluation(conv.id) if not ev.email_sent: # S3: 任一值已见于24h内其他会话 → 轻提醒;否则全量 seen = await self._recently_emailed_canons() if any(canon(str(v)) in seen for v in confirmed.values()): await self._collect_push( pending_pushes, ev, conv, profile, confirmed, "returning", evaluation, evaluated) sent = await email_service.send_returning_lead_email(confirmed) else: await self._collect_push( pending_pushes, ev, conv, profile, confirmed, "lead", evaluation, evaluated) messages = await self._load_history(conv.id) sent = await email_service.send_lead_email( customer_profile=profile.raw_info or {}, contact_info=confirmed, reasoning=evaluation.get("reasoning", ""), conversation_messages=messages, intent_level=evaluation.get("intent_level", "") if evaluated else "", ) if sent: ev.email_sent = True ev.contact_info = confirmed self.db.add(ev) else: old_canons = {canon(str(v)) for v in (ev.contact_info or {}).values()} if any(canon(str(v)) not in old_canons for v in confirmed.values()): if ev.correction_count < 3: count = ev.correction_count + 1 await self._collect_push( pending_pushes, ev, conv, profile, confirmed, "correction", evaluation, evaluated) if await email_service.send_correction_email(confirmed, count=count): ev.correction_count = count ev.contact_info = confirmed self.db.add(ev) else: # D4'(2026-08-18): 锁邮件不锁快照 —— 上限已到, # admin 仍要看到客户最新号码 ev.contact_info = confirmed self.db.add(ev) async def _collect_push( self, pending, ev, conv, profile, confirmed, push_type, evaluation, evaluated, ) -> None: """询盘推送派发(设计 §8): 触发条件=进入发信分支,与 SMTP 成败解耦。 dispatch 内部吞异常(失败只记日志),这里无需再兜。 """ log = await lead_push_service.dispatch_lead_push( self.db, ev, conv, profile, confirmed, push_type, evaluation, evaluated, ) if log is not None and pending is not None: pending.append(log) async def _load_history(self, conv_id: str) -> list[dict]: result = await self.db.execute( select(Message) .where(Message.conversation_id == conv_id) .order_by(Message.created_at) ) messages = result.scalars().all() return [{"role": m.role.value, "content": m.content} for m in messages] async def _get_profile(self, conv_id: str) -> CustomerProfile: result = await self.db.execute( select(CustomerProfile).where(CustomerProfile.conversation_id == conv_id) ) return result.scalar_one() @staticmethod def _enrich_short_message(content: str, raw_info: dict, history: list[dict] | None = None) -> str: """Enrich short user messages with conversation context. Two sources: extractor output (primary), last substantive user message (fallback when extractor hasn't run yet). """ if len(content.strip()) >= 30: return content facts = [] # Primary: extractor output if raw_info: facts = [f"{k}: {v}" for k, v in raw_info.items() if v and not str(k).startswith("_") and k != "联系方式"] # Fallback: use last substantive user message as context if not facts and history: for msg in reversed(history[:-1]): if msg.get("role") == "user" and len(msg.get("content", "").strip()) >= 5: facts.append("客户之前说: " + msg["content"][:80]) break if not facts: return content return ( content + "\n\n【系统上下文 - 系统标注，不要向客户提及】已知: " + " | ".join(facts) + "。不要重复询问这些已确认的信息！" ) @staticmethod def _build_context_reminder(raw_info: dict, latest_msg: str, history: list[dict] | None = None, source: str = "") -> str: """Build a context reminder from extracted profile info.""" # 职能边界提醒(评审#7: 前置到早退之前,首条消息画像为空也要生效) boundary_note = "" if BOUNDARY_TRIGGER_RE.search(latest_msg or ""): if BUDGET_TRIGGER_RE.search(latest_msg or ""): # 评审#9: 预算求推荐 → 预算路径(推荐+反问+引导),不一刀切要号 boundary_note = ( "⚠️ 职能边界提醒：客户报预算求推荐。推荐产品方案但绝不复述预算数字、" "不报价格 → 反问意向产品与需求规格 → 顺势引导留联系方式。") else: boundary_note = ( "⚠️ 职能边界提醒：客户正在询问价格/优惠/交付时效/实质安排。绝对禁止" "给出任何价格数字（包括复述客户自己的预算）、交付时间或任何承诺，" "按职能边界对应话术处理。") # 联系方式状态前移(供 boundary/contact 提醒共用判定) contacts_all = normalize_contacts(raw_info.get("联系方式")) confirmed, suspect = split_validate(contacts_all) # 索要我方联系方式提醒(15.2): 与 boundary_note 同样前移到早退之前, # 空画像首条消息也要生效;按 source 分支(widget/standalone 有按钮, app 无)。 # 冒烟实证: 抽象指令("用App话术")GLM 会无视并背 prompt 里更显眼的 web 脚本—— # 提醒必须携带该分支的逐字话术,不让模型自己挑。 contact_note = "" if CONTACT_ASK_TRIGGER_RE.search(latest_msg or ""): if source == "app": if confirmed or suspect: # 已留资: 索要我方联系方式 → 逐字"经理主动联系"版, # 与"禁止再索要"提醒不再打架 # Task 3 收尾: 第三处禁令同步改写(禁令集完整)——与前两处 # 同语义,只禁已提供过的渠道,缺失渠道放行轻补一次 contact_note = ( "⚠️ 客户正在索要我方联系方式，且客户已留过自己的联系方式。" "请逐字使用这句：\"Our manager will reach out to you at your " "WhatsApp shortly with the full quote!\"（中文同理）。" "绝对禁止输出任何号码或邮箱，也绝对禁止再索要客户已提供过" "的联系方式（缺失渠道可按轻补指引提一次；号码缺国家代码时" "可请客户重发完整号一次）。") else: contact_note = ("⚠️ 客户正在索要我方联系方式（最高意向信号）。当前为 App 会话，" "没有页面按钮——禁止使用任何 tap/button 按钮话术！请逐字使用这句：" "\"Drop your WhatsApp and our manager will reach out to you directly " "with the full quote!\"（中文同理引导客户留联系方式）。" "绝对禁止输出任何号码或邮箱。") else: contact_note = ("⚠️ 客户正在索要我方联系方式（最高意向信号）。请逐字使用这句指路：" f"\"{NO_CONTACT_WEB}\"（邮箱同理指 Email 按钮）。" "绝对禁止口头报出任何号码或邮箱。") # 语言跟随提醒(Task 11): 与 boundary/contact 提醒同挂载位—— # 首条消息画像为空也要生效(中文客户开场即漂英文是实测重灾区) lang_note = _lang_note_for(latest_msg) if not raw_info: # 空画像:仅返回边界/联系/语言提醒(可能为空串) return "\n".join(x for x in (boundary_note, contact_note, lang_note) if x) parts = ["[上下文提醒 - 这是一条系统指令，绝对不要向客户提及这段内容]"] parts.append("客户已明确告知以下信息，你绝对不能再问这些：") for key, val in raw_info.items(): if str(key).startswith("_") or key == "联系方式": continue # 内部状态与联系方式单独渲染 if val: parts.append(f"{key}: {val}") # Task 3(spec §4.2/§4.3b): 缺失渠道一次性轻补——cc 优先、邮箱次之,串行 # 注入。计数键(_marks)此处弃用: 本函数是 staticmethod 收内存副本,写回由 # stream 调用方以同参调用纯函数取回并落库(纯函数保证两处判定不漂移)。 # 计算前移到 confirmed 块之前: CC 追问轮 confirmed 块的确认句要让位。 completion_note, completion_marks = _contact_completion_note( raw_info, confirmed, suspect, lang="zh" if _CJK_RE.search(latest_msg or "") else "en") # 联系方式专项(D6) 与 职能边界(D6联动) if confirmed: shown = ", ".join( f"{k}: {' '.join(str(v).split())[:60]}" for k, v in confirmed.items()) # 逐字话术直接引用 NO_CONTACT_ALREADY 常量(Task 5 已 import)—— # 与零号码墙替换句同源,三处"经理主动联系"话术不漂移 # Task 3 防自锁改写(spec 4.1-6): 原"绝对禁止再索要任何联系方式！" # 会抑制缺失渠道轻补——只禁已提供过的渠道,缺失渠道放行一次 parts.append(f"⚠️ 客户已提供联系方式（{shown}）。绝对禁止再索要客户已提供过" f"的联系方式；缺失渠道仅可按轻补指引提一次；号码缺国家代码" f"时可请客户重发完整号一次。") if completion_marks == ["_cc_asked"]: # 生产验收场景①修复(2026-10,三轮): CC 追问是对已在手渠道的重发 # 请求,不在"缺失渠道"例外内——同轮两个"逐字使用这句"竞争时 GLM # 稳定服从更显眼的确认句,压制追问。本轮 confirmed 块让位: 唯一 # 逐字锚点=下方轻补指引的追问句;禁数字回显堵住"让位后 GLM 自创 # 确认句带出真实号码"的回归(墙的客户回显豁免会放行) parts.append("本轮不要使用确认句，请逐字使用下面这条国家代码追问句请客户" "重发带国家代码的完整号码（仅此一次，客户重发或明确不给后" "下一轮恢复正常）。回复中绝不写出号码数字——提到号码一律用 " "your WhatsApp 指代。") else: parts.append(f"需要回应时可直接逐字使用：\"{NO_CONTACT_ALREADY}\"（中文同理）。" f"确认收到后继续解答客户问题即可，不必急着结束对话。") if boundary_note: parts.append(boundary_note) # Task 3 防自锁改写: 尾部补缺失渠道豁免(与上方改写同因) parts.append("客户已留过联系方式——边界请求直接告知'客户经理会把准确报价/" "安排发到你 WhatsApp'即可，禁止再索要联系方式" "（缺失渠道可按轻补指引提一次；号码缺国家代码时可请客户" "重发完整号一次）。") elif suspect: shown = ", ".join( f"{k}: {' '.join(str(v).split())[:60]}" for k, v in suspect.items()) parts.append(f"⚠️ 客户疑似已提供联系方式（{shown}）。禁止重新索要；" f"可礼貌地请客户确认一次。") if boundary_note: parts.append(boundary_note) if boundary_note and not confirmed and not suspect: parts.append(boundary_note) # 未留资:边界提醒引导留资(职能边界话术自带) # Task 3 轻补注入(计算已前移到 confirmed 块之前——CC 轮让位判定需要) if completion_note: parts.append(completion_note) # 15.2: 索要我方联系方式提醒注入(镜像 boundary_note 的挂载位)。 # 互留提示仅限 web 来源且客户未留资(confirmed/suspect 都算"已留"); # App 的反向留资话术本身就是主话术,不再叠加 if contact_note: note = contact_note if source != "app" and not confirmed and not suspect: note += "客户尚未留过联系方式，可顺势轻问一句（每会话最多一次）。" parts.append(note) if len(latest_msg.strip()) < 25: parts.append("⚠️ 客户发的是短消息！必须结合上面已知信息来理解，绝对不能当作独立问题！") if lang_note: parts.append(lang_note) return "\n".join(parts)