Files
ai-kefu-jc/docs/superpowers/plans/2026-08-17-contact-driven-lead-email.md
T
2026-09-29 14:15:12 +08:00

1536 lines
70 KiB
Markdown
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# 联系方式驱动的询盘邮件(Contact-Driven Lead Email)实施计划 v1.1
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** 联系方式成为发询盘邮件的唯一条件——统一管道检测到联系方式立刻发邮件(不看消息数、不看意向评级);无联系方式永不发邮件、永不结束会话;客户已给的联系方式绝不丢失、绝不重复索要。
**Architecture:** 新建确定性"统一联系方式管道"(`app/services/contact_pipeline.py`):extractor 提取(多语言 LLM 源)+ 正则高置信扫描(确定性辅源)→ key 归一化 → 深合并(同 key 新值覆盖)→ 格式校验(confirmed/suspect 分级)。管道结果作为唯一事实源,同时供邮件触发器(去重 + 更正 + 24h 跨会话去重)、定向提醒(禁止重复索要)、会话结束守卫(全来源)三处消费。邮件触发从路由层(ended && interested)整体迁移到 `ConversationManager` 内部,与结束逻辑彻底解耦。
**Tech Stack:** Python 3.13 / FastAPI / SQLAlchemy 2.0 async + SQLite(aiosqlite) / pytest(asyncio_mode=auto)
---
## 1. 背景与问题(为什么做)
线上痛点:大量**无联系方式**的询盘邮件推送至销售邮箱(无效询盘 + 有意向但无法联系的客户)。根因链(已代码验证):
1. widget/embedded/standalone 来源的 `should_end` 没有联系方式守卫(`_has_real_contact` 仅 app 生效,`conversation.py:230`);
2. evaluator.md 要求"无联系方式 info_sufficient 必须 false",但 GLM 不遵守(`conversation.py:253` 注释自证);
3. 判定 interested 即结束会话锁死客户(`conversation.py:350`),连补联系方式的机会都没有;
4. extractor 某轮返回空 `联系方式` 时浅合并整层覆盖,已提取的联系方式被静默抹掉(`conversation.py:119`)。
## 2. 已定决策(本计划的 spec,评审以此为准)
| # | 决策 | 内容 |
|---|---|---|
| D1 | **邮件唯一条件** | 管道 confirmed 联系方式 → 立即发邮件。≥3轮、interested、info_sufficient 一律不参与邮件门槛。第 1 条消息含联系方式也发 |
| D2 | **意向降级为标签** | 邮件主题打标签:interested→[意向强]、pending→[待确认]、not_interested→[无意向]、未评估(<3轮未跑评估)→[未评估]。**评估被跳过时邮件 intent_level 参数传空串**(→[未评估]),但发给客户端的 evaluated 事件仍报 `pending`(客户端语义不变)——两者不得混用 |
| D3 | **会话永不自动结束(2026-08-18 决策 A,取代原"全来源守卫")** | 邮件与结束解耦后,自动结束只剩收尾UX/停止计费两个弱价值,却会把还没问完的客户锁死(刷新丢上下文+销售响应空窗)。**任何情况都不自动结束**:会话保持 active 直到客户自然沉默,关闭交给超时清理(健康轨)。evaluated 事件 `ended` 恒 False(App 端需知晓);`_finalize` 删除,D10 的 upsert 语义由"每轮评估落库"承担 |
| D4 | **防重发** | 同会话 email_sent 标记,全量只发一次;联系方式变化(改号/新增key/无提示换值)→轻量[更正]邮件,**每会话≤3封(2026-08-18 用户测试反馈放宽)**;达到上限后"锁邮件不锁快照"——Evaluation.contact_info 仍刷新为最新(admin 永远看到客户最新号码) |
| D5 | **24h 跨会话去重** | 同一联系方式值(规范化比对)24h 内已发过全量 → 新会话只发一行"该客户又来聊了"轻提醒 |
| D6 | **不重复索要** | 管道 confirmed → 系统提醒注入"禁止再索要,确认收到后继续解答";suspect(校验不过) → "礼貌确认,禁止重新索要"。suspect 不触发邮件、不参与结束 |
| D7 | **评估前置** | 用户消息 <3 条跳过 GLM evaluate 调用(结果必然 pending),纯省成本;邮件标签用[未评估] |
| D8 | **卡单标记** | interested + 无联系方式 → upsert Evaluation(admin 后续可筛,人工兜底的数据基础) |
| D9 | **旧触发路径拆除** | `chat.py:94-97` 与 `chat_rest.py:140-142` 的 ended&&interested→`_try_send_email` 整体删除,与 D1 触发器**同一个 PR**(两套并存=双发邮件)。**批次 1 全部提交在同一分支、合并后一次部署——严禁部署分支中间提交**:Task 6(新触发器生效)与 Task 7(旧路径拆除)之间存在新旧双触发窗口,该窗口只允许存在于开发分支上 |
| D10 | **Evaluation upsert 语义** | `_get_or_init_evaluation` 是唯一建行入口(绝不 INSERT 已存在记录,唯一约束会 IntegrityError);每轮评估落库只更新 intent/reasoning/快照,**永不触碰 email_sent/correction_sent**(原 `_finalize` 已随决策 A 删除) |
| D11 | **话术**(system.md) | ①价值交换固定模板:只承诺"完整报价单",禁止自由发挥价格/时限;②先聊后引:聊天内要两次未果才引导点 WhatsApp 按钮;③拿到联系方式后确认一句、继续解答,不急着收尾 |
| D12 | **邮件转义(S2 提前)** | email_service 全量 `html.escape`(邮件触发频率上升,注入面同步放大,不等健康轨) |
| D13 | **行为变化已批准** | 本计划推翻"不改 widget 行为"红线(2026-08-17 产品决策)。上线时同步清理 `conversation.py:131-134、226-229` 的 app-only 注释与 `chat_rest.py` 模块 docstring 第 12-13 行 |
**多值/晋升细则(S1/S3/S4 钉死):**
- S1 suspect 只用于"防重复索要"提醒,永不自动晋升;客户后续重发或改发新值时由 extractor/正则重新捕获、重新校验(新值覆盖)。
- S3 24h 去重多值规则:confirmed 中**任一**值在 24h 内已发过 → 轻提醒(新值会显示在提醒主题里,信息不丢;全部值都未见 → 全量)。
- S4 正则每轮扫描**全部客户消息历史**(非仅当轮),兜住 GLM 失败轮次漏掉的联系方式。
**明确不做(防扩散):** 范式全量重构、超时清理、UI 留资卡片、WS 限流、会话持久化、admin 筛选视图 → 批次 2/3 或健康轨(见 §6)。`process_message`(`conversation.py:70`,grep 确认零调用方的死代码)在 Task 7 随守卫改造**一并删除**(若保留,其对 `_has_real_contact` 的调用会成为悬空引用);Task 7 之前不碰它。
## 3. 文件结构
| 文件 | 动作 | 职责 |
|---|---|---|
| `app/services/contact_pipeline.py` | 新建 | 纯函数管道:归一化/深合并/校验分级/正则扫描/规范化比对 |
| `app/services/conversation.py` | 修改 | 接线管道、邮件触发器、全来源结束守卫、卡单标记、评估前置、提醒注入、finalize UPDATE、深合并修复 |
| `app/services/email_service.py` | 修改 | 转义、标签、中文标签列、轻提醒/更正邮件、"对话进行中"提示 |
| `app/models/database.py` | 修改 | Evaluation 加 `correction_sent` 列 |
| `app/main.py` | 修改 | `_migrate` 加 correction_sent 的 PRAGMA 迁移 |
| `app/api/routes/chat.py` | 修改 | 删除 `_try_send_email` 与触发块 |
| `app/api/routes/chat_rest.py` | 修改 | 删除触发块与 lazy import,更新 docstring |
| `app/prompts/system.md` | 修改 | D11 话术 |
| `scripts/baseline_stats.py` | 新建 | 批次 0.5 基线统计(只读) |
| `tests/test_contact_pipeline.py` | 新建 | 管道单元测试 |
| `tests/test_lead_email_trigger.py` | 新建 | 触发器/结束守卫/卡单集成测试(FakeAI + async_db) |
| `tests/test_chat_rest_message.py` | 修改 | 删除钉死旧触发路径的 3 个测试(被新测试取代) |
> ⚠️ **行号锚点使用规则**:文中行号是计划撰写时(当前 HEAD)的定位辅助。Task 5/6 会向 `conversation.py` 插入代码,**后续任务(Task 7/8)引用的行号会漂移约 20 行**——实施时一律以任务中引用的**代码内容**定位(如 `should_end`、`_has_real_contact` 方法名),禁止盲改行号区域。
---
## 批次 0.5:基线统计
### Task 0: 基线脚本
**Files:**
- Create: `scripts/baseline_stats.py`
- [ ] **Step 0.1: 写脚本(只读,不写库)**
```python
"""批次0.5 基线统计:上线前存档,上线后同口径对比。只读。"""
import re
import sqlite3
import sys
DB = sys.argv[1] if len(sys.argv) > 1 else "./data/cc_ai_kefu.db"
EMAIL_RE = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}")
INTL_RE = re.compile(r"\+\d{6,15}")
def main() -> None:
conn = sqlite3.connect(DB)
cur = conn.cursor()
# 1) 历史已发邮件中无联系方式占比(按 source)
rows = cur.execute(
"""SELECT c.source,
COUNT(*) AS sent,
SUM(CASE WHEN COALESCE(e.contact_info, '{}') IN ('{}', 'null', '')
OR e.contact_info IS NULL THEN 1 ELSE 0 END) AS no_contact
FROM evaluations e JOIN conversations c ON c.id = e.conversation_id
WHERE e.email_sent = 1 GROUP BY c.source"""
).fetchall()
print("== 已发邮件联系方式缺失(按来源)==")
for source, sent, no_contact in rows:
pct = (no_contact / sent * 100) if sent else 0
print(f" {source}: 发送 {sent}, 缺失 {no_contact} ({pct:.0f}%)")
# 2) 用户消息中含联系方式样式的会话占比(新策略邮件量上界预估)
convs = cur.execute("SELECT id FROM conversations").fetchall()
with_contact = early_contact = total = 0
for (cid,) in convs:
msgs = cur.execute(
"SELECT role, content FROM messages WHERE conversation_id=? ORDER BY created_at",
(cid,),
).fetchall()
user_msgs = [m for m in msgs if m[0] == "user"]
if not user_msgs:
continue
total += 1
hit_idx = next(
(i for i, m in enumerate(user_msgs)
if EMAIL_RE.search(m[1]) or INTL_RE.search(m[1])),
None,
)
if hit_idx is not None:
with_contact += 1
if hit_idx < 2:
early_contact += 1
print("== 新策略邮件量预估 ==")
print(f" 有用户消息的会话: {total}")
print(f" 消息中含联系方式样式: {with_contact}"
f" ({with_contact / total * 100 if total else 0:.0f}%) ← 全量邮件上界")
print(f" 联系方式出现在前2条消息内: {early_contact}"
f" ({early_contact / total * 100 if total else 0:.0f}%) ← 即时触发增量")
conn.close()
if __name__ == "__main__":
main()
```
- [ ] **Step 0.2: 运行并存档**
Run: `python scripts/baseline_stats.py | tee docs/superpowers/plans/2026-08-17-baseline.txt`
Expected: 打印三组数字(存档文件用于上线后对比)
- [ ] **Step 0.3: Commit**
```bash
git add scripts/baseline_stats.py docs/superpowers/plans/2026-08-17-baseline.txt
git commit -m "chore(baseline): 询盘邮件改造前基线统计"
```
---
## 批次 1:核心实现
### Task 1: contact_pipeline — 归一化 / 深合并 / 校验分级
**Files:**
- Create: `app/services/contact_pipeline.py`
- Test: `tests/test_contact_pipeline.py`
- [ ] **Step 1.1: 写失败测试**
```python
from app.services.contact_pipeline import (
merge_contacts, normalize_contacts, split_validate,
)
def test_normalize_chinese_and_english_keys():
raw = {"手机": "+8613800138000", "邮箱": "a@b.com", "微信": "wxid_1",
"WhatsApp号码": "+254712345678", "空": ""}
out = normalize_contacts(raw)
assert out == {
"phone": "+8613800138000", "email": "a@b.com",
"wechat": "wxid_1", "whatsapp": "+254712345678",
}
def test_normalize_later_alias_wins_on_same_key():
# 同 key 别名冲突时后写者覆盖(上游 extractor 单轮内一般不会同时给 手机 与 tel,
# 但语义必须被钉死并被测试)
assert normalize_contacts({"手机": "1111111", "tel": "2222222"}) == {"phone": "2222222"}
```
```python
def test_merge_new_value_wins_same_key():
old = {"phone": "+254700000000"}
new = {"手机": "+254711111111"} # 客户改号
assert merge_contacts(old, new) == {"phone": "+254711111111"}
def test_merge_empty_new_keeps_old():
# T1 场景: extractor 某轮返回空联系方式,不得抹掉已有值
old = {"phone": "+254700000000"}
assert merge_contacts(old, {}) == {"phone": "+254700000000"}
assert merge_contacts(old, None) == {"phone": "+254700000000"}
def test_split_validate_confirmed_and_suspect():
confirmed, suspect = split_validate({
"email": "a@b.com", # 合法邮箱
"phone": "+254 712 345 678", # 数字 12 位 → confirmed
"whatsapp": "12345", # 数字 5 位 → suspect
})
assert confirmed == {"email": "a@b.com", "phone": "+254 712 345 678"}
assert suspect == {"whatsapp": "12345"}
def test_split_validate_bad_email_is_suspect():
confirmed, suspect = split_validate({"email": "not-an-email"})
assert confirmed == {} and suspect == {"email": "not-an-email"}
def test_split_validate_wechat_id_is_confirmed():
# 微信号无数字格式(wxid_x 等),非空≥4字符即 confirmed——否则只留微信的
# 客户永远无法触发邮件
confirmed, suspect = split_validate({"wechat": "wxid_1"})
assert confirmed == {"wechat": "wxid_1"} and suspect == {}
def test_split_validate_wechat_short_is_suspect():
confirmed, suspect = split_validate({"wechat": "abc"})
assert confirmed == {} and suspect == {"wechat": "abc"}
def test_normalize_coerces_int_values():
# extractor 偶尔会把号码输出成 int,不得崩溃且要能归一化
assert normalize_contacts({"手机": 254712345678}) == {"phone": "254712345678"}
def test_merge_none_old_keeps_new():
assert merge_contacts(None, {"email": "a@b.com"}) == {"email": "a@b.com"}
```
- [ ] **Step 1.2: 运行确认失败**
Run: `pytest tests/test_contact_pipeline.py -v`
Expected: FAIL(ModuleNotFoundError)
- [ ] **Step 1.3: 实现 contact_pipeline.py(归一化/合并/校验部分)**
```python
"""统一联系方式管道:联系方式检测的唯一事实源。
三个动作:
- 归一化:中英文 key 别名统一为 phone/whatsapp/email/wechat
- 深合并:同 key 新值覆盖旧值(客户改号场景),空值不覆盖
- 校验分级:confirmed(格式通过,可触发邮件/结束)/ suspect(疑似,
仅用于提醒 AI 礼貌确认,防止重复索要——D6/S1)
上游约定:raw_info 中 "_" 前缀 key 为内部状态,禁止进入提示词渲染。
"""
import re
KEY_ALIASES = {
"phone": "phone", "手机": "phone", "电话": "phone", "电话号码": "phone",
"tel": "phone", "telephone": "phone", "phone_number": "phone",
"phone number": "phone", "mobile": "phone", "mobile_number": "phone",
"mobile number": "phone",
"whatsapp": "whatsapp", "wa": "whatsapp", "whats app": "whatsapp",
"whatsapp号码": "whatsapp", "whatsapp number": "whatsapp",
"email": "email", "邮箱": "email", "e-mail": "email", "mail": "email",
"电子邮件": "email",
"wechat": "wechat", "微信": "wechat", "wx": "wechat", "weixin": "wechat",
}
def normalize_contacts(raw: dict | None) -> dict:
"""key 归一化 + 去空值。同 key 别名冲突时后写者覆盖。"""
out: dict[str, str] = {}
for k, v in (raw or {}).items():
key = KEY_ALIASES.get(str(k).strip().lower())
val = str(v or "").strip()
if key and val:
out[key] = val
return out
def merge_contacts(old: dict | None, new: dict | None) -> dict:
"""同 key 新值覆盖旧值;空/缺省不覆盖(堵浅合并抹数据 bug)。"""
return {**normalize_contacts(old), **normalize_contacts(new)}
def digits_only(s: str) -> str:
"""提取纯数字(split_validate 位数判断与后续 canon 复用)。"""
return re.sub(r"\D", "", s)
def split_validate(contacts: dict) -> tuple[dict, dict]:
"""返回 (confirmed, suspect)。电话/WhatsApp 取数字 7-15 位为合格;
微信号无数字格式,取非空 ≥4 字符为合格(否则 wxid_x 类真实微信号
会永远滞留 suspect,只留微信的客户无法触发邮件——违背 D1)。"""
confirmed: dict[str, str] = {}
suspect: dict[str, str] = {}
for k, v in normalize_contacts(contacts).items():
if k == "email":
ok = re.fullmatch(r"[^@\s]+@[^@\s]+\.[^@\s]+", v) is not None
elif k == "wechat":
ok = len(v.strip()) >= 4
else:
ok = 7 <= len(digits_only(v)) <= 15
(confirmed if ok else suspect)[k] = v
return confirmed, suspect
```
(`canon()` 的定义已移至 Task 6——它只被去重逻辑消费,与本任务的纯函数无关。)
- [ ] **Step 1.4: 运行确认通过**
Run: `pytest tests/test_contact_pipeline.py -v`
Expected: 7 PASS(6 项基础用例 + wechat 修正用例)
- [ ] **Step 1.5: Commit**
```bash
git add app/services/contact_pipeline.py tests/test_contact_pipeline.py
git commit -m "feat(pipeline): 联系方式归一化/深合并/校验分级"
```
### Task 2: contact_pipeline — 正则扫描(高置信,S4)
**Files:**
- Modify: `app/services/contact_pipeline.py`
- Test: `tests/test_contact_pipeline.py`
- [ ] **Step 2.1: 写失败测试(含误报反例——二手车语境全是数字)**
```python
from app.services.contact_pipeline import scan_user_messages
def m(*texts):
return [{"role": "user", "content": t} for t in texts] + [
{"role": "assistant", "content": "reply"}]
def test_scan_email():
found = scan_user_messages(m("my email is john.doe@x.co send quotes"))
assert found == {"email": "john.doe@x.co"}
def test_scan_intl_phone():
found = scan_user_messages(m("you can reach me +254712345678"))
assert found == {"phone": "+254712345678"}
def test_scan_whatsapp_context_word():
found = scan_user_messages(m("whatsapp: 0712 345 678"))
assert found.get("whatsapp") == "0712 345 678"
def test_scan_wa_me_link():
found = scan_user_messages(m("chat https://wa.me/254712345678"))
assert found == {"whatsapp": "254712345678"}
def test_scan_ignores_prices_years_and_ai_messages():
texts = m(
"budget $25,000 for 2024 BYD Tang", # 价格+年份
"50,000 km mileage", # 里程
"I want 5 units", # 数量
)
# 上一条含 assistant 消息,确保只扫 user
found = scan_user_messages(texts)
assert found == {}
def test_scan_full_history_not_only_last():
# S4: 扫全量历史,早轮的联系方式也要能捞到
msgs = [
{"role": "user", "content": "hi"},
{"role": "assistant", "content": "hey"},
{"role": "user", "content": "mail me a@b.com"},
{"role": "assistant", "content": "got it"},
{"role": "user", "content": "how much is tang"},
]
assert scan_user_messages(msgs) == {"email": "a@b.com"}
def test_scan_context_word_does_not_grab_unrelated_numbers():
# S1: 上下文词必须与数字邻接(同一匹配内),消息任意位置的7位数字不得被抓
assert scan_user_messages(m("call me on delivery, order 1234567")) == {}
assert scan_user_messages(m("车架号码是 LFV3A23C8K3123456 吗")) == {}
assert scan_user_messages(m("engine number 2ZRFE1234567?")) == {}
assert scan_user_messages(m("我手机坏了 budget 1500000 ksh")) == {}
def test_scan_adjacent_context_phone():
assert scan_user_messages(m("my phone number is 0712 345 678")) == {
"phone": "0712 345 678"}
assert scan_user_messages(m("call me at +86 138 0013 8000")) == {
"phone": "+86 138 0013 8000"}
```
- [ ] **Step 2.2: 运行确认失败**
Run: `pytest tests/test_contact_pipeline.py -v`
Expected: 新增用例 FAIL(scan_user_messages 未定义)
- [ ] **Step 2.3: 实现扫描器(追加到 contact_pipeline.py)**
```python
EMAIL_RE = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}")
INTL_PHONE_RE = re.compile(r"\+\d{6,15}")
# 邻接绑定:上下文词与数字必须在同一匹配内(仅空白衔接)——禁止"消息任意位置
# 有上下文词 + 任意位置有 7 位数字"的消息级误绑(S1:车架号/订单号/先令价格)
WHATSAPP_CTX_RE = re.compile(
r"(?:whats\s*app|wa\.?me)\D{0,12}(\+?\d[\d\s\-().]{5,17}\d)",
re.IGNORECASE,
)
PHONE_CTX_RE = re.compile(
r"(?:my\s+phone\s+(?:number\s+)?is|my\s+number\s+is|"
r"call\s+me\s+(?:at|on)|reach\s+me\s+(?:at|on)|"
r"phone\s*[::]|number\s*[::]|号码\s*[::]|手机号?\s*[::]|电话\s*[::])\s*"
r"(\+?\d[\d\s\-().]{5,17}\d)",
re.IGNORECASE,
)
def scan_user_messages(messages: list[dict]) -> dict:
"""S4: 扫描全部客户消息,返回高置信捕获(email / +国际号 / whatsapp·wa.me /
显式电话上下文邻接数字)。裸数字(价格/年份/里程/车架号)一律不收。"""
found: dict[str, str] = {}
for msg in messages:
if msg.get("role") != "user":
continue
text = msg.get("content", "")
if "email" not in found:
hit = EMAIL_RE.search(text)
if hit:
found["email"] = hit.group(0)
if "phone" not in found:
hit = INTL_PHONE_RE.search(text)
if hit:
found["phone"] = hit.group(0)
if "whatsapp" not in found:
hit = WHATSAPP_CTX_RE.search(text)
if hit:
found["whatsapp"] = hit.group(1)
if "phone" not in found:
hit = PHONE_CTX_RE.search(text)
if hit:
found["phone"] = hit.group(1)
return found
```
- [ ] **Step 2.4: 运行确认通过**
Run: `pytest tests/test_contact_pipeline.py -v`
Expected: 19 PASS(11 存量 + 8 本任务,含 S1 邻接反例与邻接正向用例)
- [ ] **Step 2.5: Commit**
```bash
git add app/services/contact_pipeline.py tests/test_contact_pipeline.py
git commit -m "feat(pipeline): 高置信正则扫描(仅客户消息/全量历史/价格年份不误捕)"
```
### Task 3: Evaluation 加 correction_sent 列 + 迁移(D4)
**Files:**
- Modify: `app/models/database.py:109-126`
- Modify: `app/main.py:16-22`
- [ ] **Step 3.1: database.py 的 Evaluation 类加列**(`email_sent` 行后)
```python
correction_sent: Mapped[bool] = mapped_column(Boolean, default=False)
```
- [ ] **Step 3.2: main.py 的 `_migrate` 追加迁移**(page_token 块之后)
```python
result = await conn.execute(text("PRAGMA table_info(evaluations)"))
columns = {row[1] for row in result}
if "correction_sent" not in columns:
await conn.execute(text(
"ALTER TABLE evaluations ADD COLUMN correction_sent BOOLEAN DEFAULT 0"
))
```
- [ ] **Step 3.3: 验证未破坏**
Run: `pytest tests/test_smoke.py -v`
Expected: PASS(create_all 在全新库上含新列;PRAGMA ALTER 分支与既有 page_token 迁移一样无测试覆盖,属既有模式,接受——不要把本步当作 ALTER 分支的验证)
- [ ] **Step 3.4: Commit**
```bash
git add app/models/database.py app/main.py
git commit -m "feat(db): evaluations.correction_sent 列 + PRAGMA 迁移"
```
### Task 4: email_service — 转义 / 标签 / 轻提醒 / 更正邮件(D2/D4/D12)
**Files:**
- Modify: `app/services/email_service.py`
- Test: `tests/test_lead_email_trigger.py`(本 Task 只改实现,集成验证在 Task 7)
- [ ] **Step 4.0: 写失败测试(先红)**
Create `tests/test_email_rendering.py`(monkeypatch 模块内 `_send`,不触网;断言主题标签、HTML 转义、内部 key 过滤):
```python
"""邮件渲染单测: 标签主题/HTML转义/内部key过滤(D2/D12)。不触网。"""
import asyncio
from app.services import email_service
def _capture_send(monkeypatch):
captured = {}
async def fake_send(msg):
captured["msg"] = msg
return True
monkeypatch.setattr(email_service, "_send", fake_send)
return captured
def _html(msg) -> str:
return msg.get_payload()[0].get_payload(decode=True).decode("utf-8")
def test_lead_email_subject_tag_and_escaped_body(monkeypatch):
captured = _capture_send(monkeypatch)
ok = asyncio.run(email_service.send_lead_email(
customer_profile={"目标国家": "肯尼亚", "_internal": "x"},
contact_info={"phone": "+254712345678"},
reasoning="<b>强意向</b>",
conversation_messages=[
{"role": "user", "content": "<script>alert(1)</script>"},
{"role": "assistant", "content": "ok"},
],
intent_level="interested",
))
assert ok is True
subject = str(captured["msg"]["Subject"])
assert "[新询盘·意向强]" in subject and "肯尼亚" in subject
body = _html(captured["msg"])
assert "<script>" not in body # D12: 转义生效
assert "&lt;script&gt;" in body
assert "_internal" not in body # 内部 key 不渲染
def test_lead_email_unevaluated_tag(monkeypatch):
captured = _capture_send(monkeypatch)
asyncio.run(email_service.send_lead_email(
customer_profile={}, contact_info={"email": "a@b.com"},
reasoning="", conversation_messages=[], intent_level="",
))
assert "[新询盘·未评估]" in str(captured["msg"]["Subject"])
def test_returning_and_correction_emails(monkeypatch):
captured = _capture_send(monkeypatch)
assert asyncio.run(
email_service.send_returning_lead_email({"whatsapp": "+254700000000"})
) is True
assert "[线索回访]" in str(captured["msg"]["Subject"])
assert asyncio.run(
email_service.send_correction_email({"email": "a@b.com"})
) is True
assert "[询盘·更正]" in str(captured["msg"]["Subject"])
```
- [ ] **Step 4.1: 重写 email_service.py**
```python
import html
import logging
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
import aiosmtplib
from app.config import settings
logger = logging.getLogger(__name__)
TAG_LABELS = {
"interested": "意向强", "pending": "待确认",
"not_interested": "无意向", "": "未评估",
}
CONTACT_LABELS = {"phone": "手机", "whatsapp": "WhatsApp", "email": "邮箱", "wechat": "微信"}
def _esc(v) -> str:
return html.escape(str(v)) if v else ""
def _contact_rows(contacts: dict) -> str:
return "".join(
f"<tr><td style='padding:6px 12px;font-weight:bold;'>{html.escape(CONTACT_LABELS.get(k, k))}</td>"
f"<td style='padding:6px 12px;'>{_esc(v)}</td></tr>"
for k, v in (contacts or {}).items() if v
)
async def _send(msg: MIMEMultipart) -> bool:
if not all([settings.SMTP_HOST, settings.SMTP_USER, settings.NOTIFY_EMAIL]):
logger.warning("SMTP not configured, skipping email")
return False
try:
await aiosmtplib.send(
msg, hostname=settings.SMTP_HOST, port=settings.SMTP_PORT,
username=settings.SMTP_USER, password=settings.SMTP_PASS,
use_tls=True,
)
logger.info("Lead email sent to %s", settings.NOTIFY_EMAIL)
return True
except Exception:
logger.exception("Failed to send lead email")
return False
async def send_lead_email(
customer_profile: dict,
contact_info: dict,
reasoning: str,
conversation_messages: list[dict],
intent_level: str = "",
) -> bool:
tag = TAG_LABELS.get(intent_level, "未评估")
contacts = dict(contact_info or {})
country = (customer_profile or {}).get("目标国家", "")
if not country:
country = (customer_profile or {}).get("target_country", "未知")
msg = MIMEMultipart("alternative")
contact_parts = [
f"{CONTACT_LABELS.get(k, k)}: {v}" for k, v in contacts.items() if v
]
subject = f"[新询盘·{tag}] {country}"
if contact_parts:
subject += " | " + ", ".join(contact_parts)
msg["Subject"] = subject
msg["From"] = f"【XYH】 <{settings.SMTP_USER}>"
msg["To"] = settings.NOTIFY_EMAIL
info_rows = "".join(
f"<tr><td style='padding:6px 12px;font-weight:bold;'>{_esc(k)}</td>"
f"<td style='padding:6px 12px;'>{_esc(v)}</td></tr>"
for k, v in (customer_profile or {}).items() if not str(k).startswith("_")
)
chat_lines = "".join(
f"<div style='margin:4px 0;'><b>{'客户' if m.get('role') == 'user' else 'AI'}:</b> {_esc(m.get('content', ''))}</div>"
for m in conversation_messages or []
)
html_body = f"""
<html><body style="font-family:sans-serif;color:#333;">
<h2>🎉 新询盘({html.escape(tag)})</h2>
<h3>联系方式</h3>
<table style="border-collapse:collapse;border:1px solid #ddd;">{_contact_rows(contacts)}</table>
<h3>客户画像</h3>
<table style="border-collapse:collapse;border:1px solid #ddd;">{info_rows}</table>
<h3>评估理由</h3><p>{_esc(reasoning)}</p>
<h3>完整对话记录</h3>
<div style="background:#f5f5f5;padding:12px;border-radius:6px;">{chat_lines}</div>
<p style="color:#999;font-size:12px;">注:邮件为触发时快照,对话可能仍在进行,最新记录请以后台为准。</p>
</body></html>
"""
msg.attach(MIMEText(html_body, "html", "utf-8"))
return await _send(msg)
async def send_returning_lead_email(contact_info: dict) -> bool:
"""D5: 24h 内同联系方式再访,只发一行轻提醒。"""
contact_parts = [
f"{CONTACT_LABELS.get(k, k)}: {v}" for k, v in (contact_info or {}).items() if v
]
msg = MIMEMultipart("alternative")
msg["Subject"] = f"[线索回访] {' | '.join(contact_parts) or '已知客户'}再次来访"
msg["From"] = f"【XYH】 <{settings.SMTP_USER}>"
msg["To"] = settings.NOTIFY_EMAIL
msg.attach(MIMEText(
f"<html><body><p>该联系方式 24h 内已发过完整询盘,客户又开启了新对话,"
f"详情请查后台。</p></body></html>", "html", "utf-8"))
return await _send(msg)
async def send_correction_email(contact_info: dict) -> bool:
"""D4: 联系方式变化(改号/新增)后的轻量更正邮件。"""
msg = MIMEMultipart("alternative")
msg["Subject"] = "[询盘·更正] 客户联系方式已更新"
msg["From"] = f"【XYH】 <{settings.SMTP_USER}>"
msg["To"] = settings.NOTIFY_EMAIL
msg.attach(MIMEText(
f"<html><body><p>客户更新了联系方式,请以本封为准:</p>"
f"<table>{_contact_rows(contact_info)}</table>"
f"<p style='color:#999;font-size:12px;'>每个会话最多发送一次更正。</p>"
f"</body></html>", "html", "utf-8"))
return await _send(msg)
```
- [ ] **Step 4.2: 确认测试转绿 + 导入无损**
Run: `python -m pytest tests/test_email_rendering.py -v` → 3 PASS
Run: `python -m pytest -q` → 39 passed
Run: `python -c "from app.services import email_service"` → 无输出(成功)
- [ ] **Step 4.3: Commit**
```bash
git add app/services/email_service.py
git commit -m "feat(email): 标签/转义/中文标签列/回访轻提醒/更正邮件"
```
### Task 5: conversation.py — 管道接线 + 深合并修复 + 评估前置(T1/T3/D7)
**Files:**
- Modify: `app/services/conversation.py`(仅 `stream_process_message`,154-246 行区域)
- Test: `tests/test_lead_email_trigger.py`
- [ ] **Step 5.1: 写失败测试(FakeAI + async_db 模式)**
```python
"""批次1 集成测试:管道接线/邮件触发/结束守卫/卡单标记。"""
import asyncio
from app.models.database import ConversationSource, ConversationStatus
from app.services.conversation import ConversationManager
from app.services import email_service
class FakeAI:
"""替代 GLM 的确定性桩:构造参数即每轮返回值。"""
def __init__(self, extract=None, evaluate=None):
self._extract_seq = extract if isinstance(extract, list) else [extract or {}]
self._evaluate = evaluate or {}
self.eval_calls = 0
async def chat(self, messages, context=""):
return "ok reply"
async def stream_chat(self, messages, context=""):
for token in ("ok ", "reply"):
yield token
async def extract_info(self, messages, existing_info):
# 第 N 轮客户消息 → seq[N-1](messages 为 [u,a,u,a...] 全量历史)
idx = min(max(len(messages) // 2 - 1, 0), len(self._extract_seq) - 1)
return self._extract_seq[idx]
async def evaluate(self, profile, recent_messages):
self.eval_calls += 1
return dict(self._evaluate)
INTERESTED = {
"intent_level": "interested", "reasoning": "预算国家车型齐全",
"contact_info": {}, "info_sufficient": True,
}
async def _send(manager, sid, text):
async for _ in manager.stream_process_message(sid, text):
pass
def _setup(async_db, monkeypatch, source=ConversationSource.widget, ai=None):
sent = []
async def fake_lead(customer_profile, contact_info, reasoning,
conversation_messages, intent_level=""):
sent.append(("lead", dict(contact_info), intent_level))
return True
async def fake_nudge(contact_info):
sent.append(("nudge", dict(contact_info), ""))
return True
async def fake_fix(contact_info):
sent.append(("fix", dict(contact_info), ""))
return True
monkeypatch.setattr(email_service, "send_lead_email", fake_lead)
monkeypatch.setattr(email_service, "send_returning_lead_email", fake_nudge)
monkeypatch.setattr(email_service, "send_correction_email", fake_fix)
async def run():
async with async_db() as db:
manager = ConversationManager(db, ai_service=ai or FakeAI())
conv = await manager.create_conversation(source=source)
return manager, conv
return asyncio.run(run()), sent
def test_extractor_empty_round_preserves_contact(async_db, monkeypatch):
# T1: extractor 第2轮返回空联系方式,不得抹掉第1轮已提取的
(manager, conv), _ = _setup(async_db, monkeypatch, ai=FakeAI(
extract=[{"联系方式": {"手机": "+8613800138000"}}, {"联系方式": {}}],
))
asyncio.run(_send(manager, conv.session_id, "my phone is +8613800138000"))
asyncio.run(_send(manager, conv.session_id, "how much"))
profile = asyncio.run(manager._get_profile(conv.id))
assert (profile.raw_info or {}).get("联系方式", {}).get("phone") == "+8613800138000"
def test_evaluation_skipped_below_3_user_messages(async_db, monkeypatch):
# D7: 前2条客户消息不调 evaluate
ai = FakeAI()
(manager, conv), _ = _setup(async_db, monkeypatch, ai=ai)
asyncio.run(_send(manager, conv.session_id, "hi"))
asyncio.run(_send(manager, conv.session_id, "price?"))
assert ai.eval_calls == 0
asyncio.run(_send(manager, conv.session_id, "kenya, 20k"))
assert ai.eval_calls == 1
```
(注:T3"正则兜住 extractor 漏提取 → 仍触发邮件"的用例依赖 Task 6 的触发器,**移至 Task 6 Step 6.1** 一并转绿,本 Task 不含。)
- [ ] **Step 5.2: 运行确认失败**
Run: `pytest tests/test_lead_email_trigger.py -v`
Expected: 2 用例 FAIL(T1 抹数据 / D7 评估未跳过)
- [ ] **Step 5.3: 修改 stream_process_message**
在 `stream_process_message` 中,将现有 Round 2(extract)与 Round 3(evaluate)区域(`conversation.py:206-224`)替换为:
```python
# Round 2: Extract info (background)
full_history = history + [{"role": "assistant", "content": ai_reply}]
profile = await self._get_profile(conv.id)
new_info = await self.ai.extract_info(
full_history, existing_info=profile.raw_info or {}
)
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)
# 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=profile.raw_info or {},
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)
```
文件顶部 import 区追加(Task 6 会再补 `normalize_contacts, split_validate, canon`——本任务只导入用到的,避免未用导入):
```python
from app.services.contact_pipeline import merge_contacts, scan_user_messages
```
注意:旧代码中 `user_msg_count` 原在 evaluate 之后计算(`conversation.py:225`),现在提前——旧的 `should_end` 区域在 Task 7 重写,本 Task 不动结束逻辑。`evaluation_ran` 在 Task 6 的触发器中使用(D2:跳过评估时邮件标签必须是[未评估]而不是[待确认])。
- [ ] **Step 5.4: 运行确认通过**
Run: `pytest tests/test_lead_email_trigger.py -v`
Expected: 2 PASS(T3 用例在 Task 6 引入)
- [ ] **Step 5.5: Commit**
```bash
git add app/services/conversation.py tests/test_lead_email_trigger.py
git commit -m "feat(pipeline): 接入统一管道/深合并修复/正则辅源/评估前置"
```
### Task 6: 邮件触发器 + 卡单标记 + finalize UPDATE(D1/D4/D5/D8/D10)
**Files:**
- Modify: `app/services/conversation.py`
- Test: `tests/test_lead_email_trigger.py`
- [ ] **Step 6.1: 写失败测试(追加)**
测试基建两条卫生规则(Task 5 质量评审采纳项,随本任务落地):
- 本文件每个用例结尾若持有 manager,追加 `asyncio.run(manager.db.close())`(含既有 T1/D7 两例)——消除悬挂读事务的 GC 警告;
- `FakeAI.evaluate` 按 `_normalize_evaluation` 的生产行为做默认补全/非法值矫正(4 个 key 全兜底),防止后续用例传部分 dict 时与生产行为分叉。
```python
from app.models.database import Evaluation, IntentLevel
from sqlalchemy import select
def _mk_ai(extract, evaluate=None):
return FakeAI(extract=extract, evaluate=evaluate)
def test_regex_catches_extractor_miss(async_db, monkeypatch):
# T3(自 Task 5 移入): extractor 全程漏提取,正则从消息里捞到 email → 仍然触发邮件
(manager, conv), sent = _setup(async_db, monkeypatch, ai=FakeAI(extract={}))
asyncio.run(_send(manager, conv.session_id, "my email is a@b.com"))
assert len(sent) == 1 and sent[0][0] == "lead"
assert sent[0][1]["email"] == "a@b.com"
def test_first_message_contact_sends_email_with_unevaluated_tag(async_db, monkeypatch):
# D1: 第1条消息含联系方式即发,标签[未评估](评估未跑 → intent_level 传空,D2)
(manager, conv), sent = _setup(async_db, monkeypatch,
ai=_mk_ai({"联系方式": {"手机": "+8613800138000"}}))
asyncio.run(_send(manager, conv.session_id, "wa me +8613800138000"))
assert len(sent) == 1 and sent[0][0] == "lead"
assert sent[0][2] == "" # 评估跳过 → 邮件 intent_level 参数为空串 → 标签[未评估]
def test_no_contact_never_emails_and_marks_stuck(async_db, monkeypatch):
# D8: interested+无联系方式 → 不发邮件,upsert Evaluation 卡单记录
(manager, conv), sent = _setup(async_db, monkeypatch,
ai=_mk_ai({}, evaluate=INTERESTED))
for text in ("hi", "kenya", "20k budget", "byd tang"):
asyncio.run(_send(manager, conv.session_id, text))
assert sent == []
async def check():
ev = (await manager.db.execute(
select(Evaluation).where(Evaluation.conversation_id == conv.id)
)).scalar_one()
return ev
ev = asyncio.run(check())
assert ev.intent_level == IntentLevel.interested
assert (ev.contact_info or {}) == {}
assert ev.email_sent is False
def test_email_sent_once_per_conversation(async_db, monkeypatch):
# D4: 联系方式持续存在,后续消息不得重发
(manager, conv), sent = _setup(async_db, monkeypatch,
ai=_mk_ai({"联系方式": {"email": "a@b.com"}}))
asyncio.run(_send(manager, conv.session_id, "mail me a@b.com"))
asyncio.run(_send(manager, conv.session_id, "another question"))
asyncio.run(_send(manager, conv.session_id, "more"))
assert len(sent) == 1
def test_correction_email_on_change_once(async_db, monkeypatch):
# D4: 改号 → 1封更正;再改 → 不再发(cap 必须被真正跑到第二次改号)
ai = FakeAI(extract=[{"联系方式": {"手机": "+8613800138000"}},
{"联系方式": {"手机": "+8613800138000"}},
{"联系方式": {"手机": "+8613800139999"}},
{"联系方式": {"手机": "+8613800139999"}},
{"联系方式": {"手机": "+8613800122222"}}])
(manager, conv), sent = _setup(async_db, monkeypatch, ai=ai)
asyncio.run(_send(manager, conv.session_id, "my number +8613800138000"))
asyncio.run(_send(manager, conv.session_id, "wrong, use +8613800139999"))
asyncio.run(_send(manager, conv.session_id, "again +8613800122222"))
asyncio.run(_send(manager, conv.session_id, "fourth message"))
asyncio.run(_send(manager, conv.session_id, "fifth message"))
kinds = [s[0] for s in sent]
assert kinds == ["lead", "fix"] # 第5轮 contact 再变(C),correction_sent 已锁 → 不发
def test_suspect_contact_never_triggers_email(async_db, monkeypatch):
# D1 边界: 校验不过的疑似联系方式(5位 whatsapp)绝不触发邮件
(manager, conv), sent = _setup(async_db, monkeypatch,
ai=_mk_ai({"联系方式": {"whatsapp": "12345"}}))
for text in ("hi", "second", "third"):
asyncio.run(_send(manager, conv.session_id, text))
assert sent == []
def test_24h_dedup_sends_nudge(async_db, monkeypatch):
# D5: 另一会话24h内已对同值发过全量 → 本会话只发轻提醒
from datetime import datetime, timezone
ai1 = _mk_ai({"联系方式": {"email": "same@x.com"}})
(manager, conv1), sent1 = _setup(async_db, monkeypatch, ai=ai1)
asyncio.run(_send(manager, conv1.session_id, "mail same@x.com"))
assert [s[0] for s in sent1] == ["lead"]
ai2 = _mk_ai({"联系方式": {"email": "same@x.com"}})
(manager, conv2), sent2 = _setup(async_db, monkeypatch, ai=ai2)
asyncio.run(_send(manager, conv2.session_id, "im back same@x.com"))
assert [s[0] for s in sent2] == ["nudge"]
```
- [ ] **Step 6.2: 运行确认失败**
Run: `pytest tests/test_lead_email_trigger.py -v`
Expected: 新增 6 用例 FAIL(含自 Task 5 移入的 T3 用例)
- [ ] **Step 6.3: 实现(conversation.py)**
(a0) 先在 `app/services/contact_pipeline.py` 追加 `canon`(含单测,先红后绿):
```python
def canon(value: str) -> str:
"""去重比对用的规范形:邮箱小写,其余取纯数字(S3/R3 best-effort)。"""
value = str(value or "").strip()
if "@" in value:
return value.lower()
return digits_only(value)
```
```python
def test_canon_normalizes_for_dedup():
from app.services.contact_pipeline import canon
assert canon("A@B.COM") == "a@b.com"
assert canon("+254 712-345 678") == "254712345678"
```
(a) 类内新增三个方法:
```python
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)
# 只取需要的列:该查询在每次首触/重试时执行,避免整行 ORM 实例化。
# 🔴 禁止加 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,
) -> 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()):
sent = await email_service.send_returning_lead_email(confirmed)
else:
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)
elif not ev.correction_sent:
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 await email_service.send_correction_email(confirmed):
ev.correction_sent = True
ev.contact_info = confirmed
self.db.add(ev)
```
顶部 import 追加:`from datetime import datetime, timedelta, timezone`、`from app.services import email_service`、`contact_pipeline` 导入补 `canon`。
(b) `stream_process_message` 在 evaluate 之后、`should_end` 之前插入:
```python
# 邮件触发(D1): 与会话结束彻底解耦,联系方式到手即发
await self._maybe_send_lead_email(
conv, profile, evaluation, evaluated=evaluation_ran
)
# 卡单标记(D8): 有意向但没联系方式 → 落库供 admin 人工兜底
confirmed_now, _ = split_validate(
normalize_contacts((profile.raw_info or {}).get("联系方式")))
if (user_msg_count >= 3
and evaluation.get("intent_level") == "interested"
and not confirmed_now):
ev = await self._get_or_init_evaluation(conv.id)
ev.intent_level = IntentLevel.interested
ev.reasoning = evaluation.get("reasoning", "")
ev.contact_info = {}
self.db.add(ev)
```
(c) `_finalize` 改 UPDATE 语义(D10),替换 `conversation.py:339-344` 的 INSERT:
```python
evaluation_record = await self._get_or_init_evaluation(conv.id)
evaluation_record.intent_level = intent_enum
evaluation_record.reasoning = evaluation.get("reasoning", "")
# 结束时的管道快照;保留 email_sent / correction_sent
confirmed, _ = split_validate(
normalize_contacts((profile.raw_info or {}).get("联系方式")))
evaluation_record.contact_info = confirmed or evaluation_record.contact_info or {}
self.db.add(evaluation_record)
```
- [ ] **Step 6.4: 运行确认通过**
Run: `pytest tests/test_lead_email_trigger.py -v`
Expected: 全部 PASS
- [ ] **Step 6.5: Commit**
```bash
git add app/services/conversation.py tests/test_lead_email_trigger.py
git commit -m "feat(trigger): 联系方式驱动邮件触发器/卡单标记/finalize改UPDATE"
```
### Task 7: 全来源结束守卫 + 拆除旧触发路径(D3/D9/D13)
**Files:**
- Modify: `app/services/conversation.py:226-236`(should_end 区域)与 `_has_real_contact`(248-262,删除)
- Modify: `app/api/routes/chat.py:10,94-97,108-144`
- Modify: `app/api/routes/chat_rest.py:138-142`(及 docstring 第 12-13 行)
- Modify: `tests/test_chat_rest_message.py`(删 3 个钉死旧路径的测试)
- Test: `tests/test_lead_email_trigger.py`
- [ ] **Step 7.1: 写失败测试(追加)**
```python
def test_widget_never_ends_without_contact_all_sources(async_db, monkeypatch):
# D3/P1: widget 来源 interested+sufficient 但无联系方式 → 永不结束
(manager, conv), _ = _setup(async_db, monkeypatch,
ai=_mk_ai({}, evaluate=INTERESTED))
for text in ("hi", "kenya", "20k", "tang", "ship how"):
asyncio.run(_send(manager, conv.session_id, text))
conv_now = asyncio.run(manager.get_by_session(conv.session_id))
assert conv_now.status == ConversationStatus.active
def test_ends_with_contact_and_interested(async_db, monkeypatch):
# D3: confirmed联系方式 + interested + ≥3轮 → 正常结束
ai = _mk_ai({"联系方式": {"email": "a@b.com"}}, evaluate=INTERESTED)
(manager, conv), _ = _setup(async_db, monkeypatch, ai=ai)
for text in ("hi", "kenya", "20k", "tang"):
asyncio.run(_send(manager, conv.session_id, text))
conv_now = asyncio.run(manager.get_by_session(conv.session_id))
assert conv_now.status == ConversationStatus.completed
```
- [ ] **Step 7.2: 运行确认失败**
Run: `pytest tests/test_lead_email_trigger.py -v`
Expected: widget 用例 FAIL(现状 widget 无守卫会结束)
- [ ] **Step 7.3: 修改 should_end(conversation.py:226-236 区域)**
```python
# 结束守卫(D3): 全来源必须有 confirmed 联系方式 —— 无联系方式永不结束,
# AI 保持争取机会。(推翻"不改widget"红线,2026-08-17 产品决策)
should_end = (
bool(confirmed_now)
and evaluation.get("info_sufficient", False)
and user_msg_count >= 3
and evaluation.get("intent_level") == "interested"
)
```
(`confirmed_now` 已在 Task 6 卡单标记处计算,紧邻复用。)随后:
- 删除 `_has_real_contact` 静态方法(248-262)及其上方的 app-only 注释(226-229);
- **删除整个 `process_message` 方法(70-152)**:grep 确认零调用方的死代码,且其对 `_has_real_contact` 的调用(137 行)在方法删除后会变成悬空引用——与其留一个必然 AttributeError 的死方法,不如随本提交一并清除(其 131-134 的红线注释同时消失);
- 顺手删除旧区域遗留的 `user_msg_count = sum(...)` 重复赋值行(`conversation.py:225`,Step 5.3 已提前计算)。
- [ ] **Step 7.4: 拆除路由层旧触发(D9,同一提交)**
`chat.py`:
- 删除 `if event["type"] == "evaluated" and event.get("ended")` 触发块(94-97 行区域,含 `_try_send_email` 调用);
- **保留** `ended = False` 标志 + `ended = True` + 外层 `if ended: break`(HEAD 的连接生命周期语义:ended 后关闭连接。只拆邮件触发,不改连接行为——若去掉外层 break,连接会悬挂到代理超时 1h);
- 删除 `_try_send_email` 整个函数(108-144)与 `from app.services import email_service`(第 10 行);随之 `from sqlalchemy.ext.asyncio import AsyncSession`(第 5 行)不再被使用,一并删除。
`chat_rest.py`:
- 删除 138-142 的 lazy import 与调用块;
- docstring 第 12-13 行的 "fires `_try_send_email` on `ended && intent_level == "interested"`" 改为 "lead email is triggered inside `ConversationManager` when contact info is detected (see contact-driven-lead-email plan)"。
`tests/test_chat_rest_message.py`:
- 删除 `test_email_fires_on_interested_end`、`test_email_does_not_fire_when_not_interested` 及模块 docstring 中的 🔴 段(其保护的行为已被 `tests/test_lead_email_trigger.py` 的 D1/D4 测试取代);
- 保留 reply/404/429 用例。
- [ ] **Step 7.4b: 确认死代码删除无残留引用(须在 7.4 删除 docstring 中的旧引用之后执行)**
Run: `grep -rn "process_message\b" app/ tests/ | grep -v stream_process_message`
Expected: 无输出。(注意顺序:若在 7.4 之前跑,`tests/test_chat_rest_message.py` docstring 里的裸 `process_message` 字样会产生一条命中——那是即将被 7.4 删除的旧文案。)
- [ ] **Step 7.5: 全量测试**
Run: `pytest -v`
Expected: 全部 PASS(含未改动的 context_seeding/enum/rate_limit/smoke)
- [ ] **Step 7.6: Commit**
```bash
git add app/services/conversation.py app/api/routes/chat.py app/api/routes/chat_rest.py tests/
git commit -m "feat(guard): 全来源结束守卫+拆除旧ended触发路径(邮件触发内聚到Manager)"
```
### Task 8: 定向提醒(D6)— 不重复索要
**Files:**
- Modify: `app/services/conversation.py`(`_build_context_reminder` 312-328、`_enrich_short_message` 279-310)
- Test: `tests/test_lead_email_trigger.py`
- [ ] **Step 8.1: 写失败测试(追加)**
```python
def test_reminder_says_no_reasking_when_contact_present():
raw = {"目标国家": "肯尼亚", "联系方式": {"whatsapp": "+254712345678"}}
text = ConversationManager._build_context_reminder(raw, "yes", [])
assert "禁止" in text and "+254712345678" in text
def test_reminder_suspect_variant_asks_to_confirm():
raw = {"联系方式": {"whatsapp": "12345"}} # 校验不过 → suspect
text = ConversationManager._build_context_reminder(raw, "yes", [])
assert "确认" in text and "禁止" in text
def test_reminder_hides_internal_keys():
raw = {"目标国家": "肯尼亚", "_internal": "x"}
text = ConversationManager._build_context_reminder(raw, "hi", [])
assert "_internal" not in text
```
- [ ] **Step 8.2: 运行确认失败**
Run: `pytest tests/test_lead_email_trigger.py -v`
Expected: 3 用例 FAIL
- [ ] **Step 8.3: 修改 `_build_context_reminder`**
```python
@staticmethod
def _build_context_reminder(raw_info: dict, latest_msg: str, history: list[dict] | None = None) -> str:
"""Build a context reminder from extracted profile info."""
if not raw_info:
return ""
parts = ["[上下文提醒 - 这是一条系统指令,绝对不要向客户提及这段内容]"]
parts.append("客户已明确告知以下信息,你绝对不能再问这些:")
for key, val in raw_info.items():
if str(key).startswith("_") or key == "联系方式":
continue # 内部状态与联系方式单独渲染
if val:
parts.append(f"{key}: {val}")
# 联系方式专项(D6): confirmed=禁止索要; suspect=礼貌确认。绝不重复索要已给的。
contacts_all = normalize_contacts(raw_info.get("联系方式"))
confirmed, suspect = split_validate(contacts_all)
if confirmed:
shown = ", ".join(
f"{k}: {' '.join(str(v).split())[:60]}" for k, v in confirmed.items())
parts.append(f"⚠️ 客户已提供联系方式({shown})。绝对禁止再索要任何联系方式!"
f"确认收到后继续解答客户问题即可,不必急着结束对话。")
elif suspect:
shown = ", ".join(
f"{k}: {' '.join(str(v).split())[:60]}" for k, v in suspect.items())
parts.append(f"⚠️ 客户疑似已提供联系方式({shown})。禁止重新索要;"
f"可礼貌地请客户确认一次。")
if len(latest_msg.strip()) < 25:
parts.append("⚠️ 客户发的是短消息!必须结合上面已知信息来理解,绝对不能当作独立问题!")
return "\n".join(parts)
```
`_enrich_short_message` 的 `facts` 构造行加同样过滤,并跳过 `联系方式`(避免 dict-repr 进提示词,与 reminder 的专项渲染保持一致):
```python
facts = [f"{k}: {v}" for k, v in raw_info.items()
if v and not str(k).startswith("_") and k != "联系方式"]
```
- [ ] **Step 8.4: 运行确认通过 + Commit**
Run: `pytest tests/test_lead_email_trigger.py -v`
Expected: 全部 PASS
```bash
git add app/services/conversation.py tests/test_lead_email_trigger.py
git commit -m "feat(reminder): 联系方式定向提醒(confirmed禁索要/suspect礼貌确认)"
```
### Task 9: system.md 话术(D11)
**Files:**
- Modify: `app/prompts/system.md:60-82`(对话流程 + 联系方式引导策略两节)
- [ ] **Step 9.1: 修改话术**
"## 对话流程"第 3、4 条替换为:
```markdown
3. 第3-4轮用"转接销售"或"发报价单"的方式引导联系方式(见下方策略)
4. 拿到联系方式后:确认一句("Got it!"),然后继续正常解答客户问题——不必急着结束对话,也别再提联系方式
```
"## 联系方式引导策略"节追加(放在"### 注意"之前):
```markdown
### 价值交换优先(比单纯索要多给一个理由)
- "I'll send you our full price list — what's your WhatsApp?"
- "Want me to send a proper quote? Drop your WhatsApp or email."
- 话术固定为"完整报价单/报价"级别,禁止承诺具体价格、折扣或时限(仅指索要联系方式时的话术,正常聊车报价照旧)
### 引流兜底(先聊后引,顺序不能反)
- 聊天内按上述方式要联系方式,最多两次
- 两次都没要到,才指引用户点面板上的 WhatsApp 按钮直接找我们:
"You can also just message us on WhatsApp (button at the top) — our guy replies fast."
- 禁止一上来就推 WhatsApp 按钮跳过聊天索要
### 系统提醒配合
- 系统会在上下文提醒里告诉你客户是否已留联系方式
- 提醒说"已提供"时:绝对禁止再要,确认收到即可
- 提醒说"疑似提供"时:礼貌请客户确认一次,不要当成没给过重新索要
```
"### 注意"节的"拿到后说一句…别再重复要"保持不变。
- [ ] **Step 9.2: 验证 prompt 加载**
Run: `python -c "from app.services.ai_service import SYSTEM_PROMPT; assert '价值交换优先' in SYSTEM_PROMPT; print('ok')"`
Expected: ok
- [ ] **Step 9.3: Commit**
```bash
git add app/prompts/system.md
git commit -m "feat(prompt): 价值交换/先聊后引/系统提醒配合话术"
```
### Task 10: 全量回归 + 人工验收
- [ ] **Step 10.1: 全量测试**
Run: `pytest -v`
Expected: 全部 PASS
- [ ] **Step 10.2: 本地起服务人工冒烟**
Run: `uvicorn app.main:app --port 8010`(.env 指向测试 SMTP/_NOTIFY_EMAIL)
操作:打开 `http://127.0.0.1:8010/`,第 1 条消息直接输入 "my whatsapp is +254712345678" → 收到 [新询盘·未评估] 邮件(含该号码、完整转义对话);继续聊 3+ 轮不再收邮件;再说 "wrong number, use +254711111111" → 收到 [询盘·更正];无联系方式的对话聊 5 轮 → 零邮件且输入框不锁死。
Expected: 全部符合
- [ ] **Step 10.3: 上线后一周观测(对照 2026-08-17-baseline.txt)**
指标:①新邮件 100% 带联系方式 ②全量邮件量 vs 基线上界 ③[更正]/[回访]占比 ④admin 中卡单(interested+无联系方式)数量
- [ ] **Step 10.4: 最终 Commit(如有遗漏小修)**
```bash
git add -A && git commit -m "chore: 批次1收尾"
```
### Task 11: 移除自动结束(决策 A,2026-08-18 追加)
**背景**:邮件已与结束解耦,自动结束只剩收尾 UX/停止计费两个弱价值,却会踢出还没问完的客户(锁输入+刷新丢上下文+销售响应空窗)。用户拍板方案 A。
**Files:**
- Modify: `app/services/conversation.py`(stream 结束区域 + `_finalize` 删除)
- Test: `tests/test_lead_email_trigger.py`
- [ ] **Step 11.1: 写失败测试(先红)**
```python
def test_conversation_never_auto_ends_even_with_contact_and_interested(async_db, monkeypatch):
# 决策A(2026-08-18): 自动结束移除——即使 contact+interested+≥3轮也保持 active
ai = _mk_ai({"联系方式": {"email": "a@b.com"}}, evaluate=INTERESTED)
(manager, conv), _ = _setup(async_db, monkeypatch, ai=ai)
for text in ("hi", "kenya", "20k", "tang", "more questions"):
asyncio.run(_send(manager, conv.session_id, text))
conv_now = asyncio.run(manager.get_by_session(conv.session_id))
assert conv_now.status == ConversationStatus.active
asyncio.run(manager.db.close())
def test_evaluation_upsert_keeps_flags_and_intent(async_db, monkeypatch):
# 评估落库泛化: intent 刷新为 interested,email_sent 不被触碰
ai = _mk_ai({"联系方式": {"email": "a@b.com"}}, evaluate=INTERESTED)
(manager, conv), sent = _setup(async_db, monkeypatch, ai=ai)
for text in ("hi", "kenya", "20k"):
asyncio.run(_send(manager, conv.session_id, text))
async def check():
ev = (await manager.db.execute(
select(Evaluation).where(Evaluation.conversation_id == conv.id)
)).scalar_one()
return ev
ev = asyncio.run(check())
assert ev.intent_level == IntentLevel.interested
assert ev.email_sent is True and ev.correction_sent is False
asyncio.run(manager.db.close())
```
同时把 `test_ends_with_contact_and_interested` **删除**(其断言的行为已被决策 A 废除,由上面的 never-auto-ends 用例取代)。
- [ ] **Step 11.2: 运行确认失败**
Run: `python -m pytest tests/test_lead_email_trigger.py -v -k "never_auto_ends or upsert_keeps"`
Expected: never_auto_ends FAIL(现状会结束);upsert_keeps 视现状可能通过(先记录)
- [ ] **Step 11.3: 实现**
(a) 删除结束区域(`should_end` 判定 + `if should_end: await self._finalize(...)`),evaluated 事件的 `ended` 恒为 `False`(事件本身保留,客户端协议不变)。
(b) 将 D8 卡单块泛化为每轮评估落库:
```python
# 评估落库(D8 泛化): 每轮评估后刷新意向;email_sent/correction_sent
# 永不触碰。有意向无联系方式 = 卡单(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)
```
(c) 删除 `_finalize` 方法整体(超时关闭功能未来另行实现,届时基于 `_get_or_init_evaluation` 重建)。删除前 grep 确认零调用方。
- [ ] **Step 11.4: 全量测试**
Run: `python -m pytest -q`
Expected: 57 passed(56 − 删1 + 增2)
- [ ] **Step 11.5: Commit** — 🔴 SKIP(沿用不提交约定)
### Task 12: 联系方式更正覆盖 + 更正上限 3 次(2026-08-18 用户实测反馈)
**背景**:用户实测 "wrong number, use X" 未触发更正——extractor 受"保留已有信息"倾向回传旧值,扫描器"首中+填空"结构性无法翻新(风险台账中该残余首次真实命中)。用户决策:①确定性"最新显式自述优先"覆盖;②extractor.md 教会覆盖语义;③更正上限放宽至 3 次;④锁邮件不锁快照。
**Files:** contact_pipeline.py(新增 `latest_stated_contacts` + `CORRECTION_CTX_RE` + PHONE_CTX 扩展 new/other/another)、conversation.py(Round 2.5 覆盖 + 触发器计数分支)、database.py + main.py(`correction_sent`→`correction_count`,迁移替换)、email_service.py(更正邮件带次数)、app/prompts/extractor.md(覆盖语义一行)、测试三文件。
**核心规则**:`latest_stated_contacts(messages)` 反向取每 key 全历史**最后一次**显式自述——上下文门控捕获(含 my new/other/another number is)与"更正语境 ∧ 裸国际号"有覆盖权;纯裸数字无覆盖权。Round 2.5 合并后:canon 与现值不同的 key 强制覆盖(唯一允许文本压过 extractor 的位置)。触发器更正分支:`correction_count < 3` → 发邮件(主题 `[询盘·更正N]`)并计数;已达上限 → 仅刷新 `ev.contact_info`。
**关键测试**:`test_correction_overrides_extractor_echo`(实测场景复刻:extractor 恒回旧值 + wrong number 消息 → 管道翻新 + lead/fix 各一封)、`test_bare_second_number_overrides`(无更正语境的 my new number is)、`test_correction_cap_three_then_snapshot_only`(A→B→C→D→E:lead+3 fix,第5值仅进快照)、原 `test_correction_email_on_change_once` 期望更新为 `["lead","fix","fix"]`、渲染测试断言 `[询盘·更正1]`、`latest_stated_*` 单测四例。
### Task 12b: 同号同步(2026-08-18 用户实测反馈 II)
**背景**:用户实测更正邮件里联系方式新旧混排(phone=新号, whatsapp=旧号)。根因:extractor 习惯把同一号码镜像进 phone/whatsapp 两个 key,更正只翻新命中的 key,兄弟 key 滞留旧值进邮件/快照。
**修复**:`contact_pipeline.sync_same_number_keys(before, after)` 纯函数——after 中某 key 翻新时,与其**旧值**同号的兄弟 key 一并翻新(email 身份独立不同步;两 key 本就不同号时不动)。接线在 Round 2 深合并前捕获 before、Task 12 覆盖后执行同步(覆盖 extractor 与 override 两条更新路径)。
**测试**:3 个纯函数单测(镜像同步/真不同号不动/email 独立)+ 1 个集成复刻(extractor 镜像旧号 → wrong number → 断言邮件与快照两 key 均为最新号)。
### Task 12c: 最近提及裁决(2026-08-18 用户实测反馈 III)
**背景**:用户实测更正配额被虚假更正烧光(仅 1 次真实变化却 correction_count=3,最后一封更正邮件显示旧号)。根因:extractor 在历史消息里的多个号码间振荡(GLM 每轮从全历史提取,回传值不稳定),深合并如实应用翻转,12b 同号同步进一步放大到镜像 key,每次翻转触发一封虚假更正。
**修复**:`resolve_by_recency(before, after, messages)` 取代 12b 的同号同步——镜像身份组(before 同 canon 的非 email key)内部分歧时,取"出现在最晚客户消息里"的值:旧号在更早消息里天然落败(杀振荡)、幻觉值 last_idx=-1 天然落败(压制幻觉)、客户最晚消息里的新号天然胜出(extractor 抓到的裸号正确翻新)。
**验收测试**:`test_oscillation_does_not_burn_correction_quota`——完整复刻用户序列(A→B→三轮振荡→裸号 C),断言仅 2 次更正且第 2 封内容就是 C;12b 的镜像同步场景由 recency 涵盖(原测试继续通过)。
---
## 4. 验收清单(对照已定决策)
| 决策 | 验收 |
|---|---|
| D1 邮件唯一条件 | `test_first_message_contact_sends_email_with_unevaluated_tag` |
| D3 永不自动结束(决策A) | `test_widget_never_ends_without_contact_all_sources` / `test_conversation_never_auto_ends_even_with_contact_and_interested` / `test_evaluation_upsert_keeps_flags_and_intent` |
| D4 防重发+更正 | `test_email_sent_once_per_conversation` / `test_correction_email_on_change_once` |
| D5 24h 去重 | `test_24h_dedup_sends_nudge` |
| D6 不重复索要 | `test_reminder_*` 三例 |
| D7 评估前置 | `test_evaluation_skipped_below_3_user_messages` |
| D8 卡单标记 | `test_no_contact_never_emails_and_marks_stuck` |
| D9/D10 旧路径拆除+UPDATE | Task 7 Step 7.4/7.5 全量回归绿 |
| 不丢失(T1/T2/T3) | `test_extractor_empty_round_preserves_contact` / 归一化用例 / `test_regex_catches_extractor_miss` |
| D12 转义 | email_service 全字段 `_esc` + 人工冒烟 |
## 5. 残余风险台账(明示接受)
| 风险 | 缓解 | 残余 |
|---|---|---|
| 正则误报 → 错发一封 + 提醒压制再索要 | 高置信模式/校验分级/上线监控无效率 | 低概率,方向安全 |
| 号码变体击穿 24h 去重 | canon 纯数字比对 | best-effort |
| extractor 连续漏提取"原值+更正值"时,扫描器首个历史值滞留 | 邻接绑定+extractor优先合并压低概率 | 双重漏提取,极低 |
| 给号即沉默 + SMTP 首发失败 | 下条消息自然重试(email_sent 未置位) | 双重失败丢线索(极低) |
| 发送成功后落库失败(毫秒级窗口)→ 重试时重复发送 | at-least-once 为有意选择:漏一条线索比重复一封邮件贵 | 接受 |
| App 端 `ended` 恒为 false(决策A) | App 若有"对话结束"闭环 UI 需知晓并调整;widget/独立页零影响 | 已知会用户 |
| 第 1 轮 extractor 幻觉出从未提及的号码 → 首封全量邮件 phone/whatsapp 混排 | 需"幻觉+首轮无历史组"双条件叠加,概率极低;whatsapp 侧仍真实可联 | 观察期,若实测出现再做 12e(无消息依据的 canon 降级) |
| 客户撤回已给联系方式 | — | 接受 |
| 邮件量上升(含无意向留号者) | 标签分流 + 基线上界预估 | 预期内 |
## 6. 后续批次(另行成 plan,本计划不含)
- **批次 2**:widget 会话持久化(localStorage 独立前缀 + 48h 过期)+ WS 限流/消息长度(消息级拒绝防重连风暴;session ≥60/h、IP 放宽;REST 长度复用 429)。
- **批次 3**:admin"缺联系方式线索"筛选(基于 D8 卡单记录)+ UI 留资卡片(提交即写管道;端点限流+校验+只弹一次)。
- **健康轨**:删 Redis、SQLite 备份+WAL、GLM 超时重试、admin 分页 total 修复、国家映射、ADMIN_PASSWORD 修改(强烈建议)、删除已无调用方的 `AIService.chat`(Task 7 后仅剩 provider 内部引用)。(`process_message` 死代码已并入本计划 Task 7 删除。)