# 潜在治疗召回 · 数据流转全景图

> **用途**:业务方评审,讲清楚一条召回计划是如何从"宿主原始数据"一路走到"客服可执行任务"的。
>
> **场景**:**潜在治疗新链召回**(treatment_initiation_recall)— 患者有诊断 / 医生建议,但未启动相应治疗。
>
> **真实例子**:卜晓平,2025-12-21 诊断 K08 缺失牙 17 号,151 天后(2026-05-21)仍未启动种植/桥/义齿 → PAC 算法产 87 分召回 plan。
>
> **架构口诀**(W3 末确立):**形态归 A.5 / 翻译归 A / 校验归 B / 抽取归 C** — 4 层各管一类语义,互不复制。
>   - A.5 transforms(yaml 声明式 6 operator):**形态改造**(拆行/派生/分流/过滤)
>   - A AssemblerEngine(yaml field_mapping/enum_mapping):**命名 + 枚举翻译**
>   - B Parser + zod(per fact_type TS):**语义校验**(强 schema 闭集)
>   - C Signal Extractor(LLM + 规则,W5+):**自由文本抽取**

---

## 总览图

```mermaid
%%{init: {'theme':'neutral','flowchart':{'curve':'basis'}}}%%
flowchart TB

%% ───── L0 · 数据源(4 入口统一) ─────
subgraph L0["🌐 L0 · 数据源(4 入口统一,W3 末确立)"]
  direction LR
  COLD["①Cold-import<br/>━━━━━━━━━━<br/>SQL 直连 / CSV dump<br/>历史一次性灌"]
  DW["②DW 直连增量<br/>━━━━━━━━━━<br/>SQL + cursor 水位<br/>(W5+,框架预留)"]
  PULL["③Pull HTTP<br/>━━━━━━━━━━<br/>5i5ya/Friday API<br/>(W5+)"]
  PUSH["④Push Webhook<br/>━━━━━━━━━━<br/>host 主动推 event<br/>action/subjectType 自带"]
end

%% ───── Layer A.5 · transforms(yaml 声明式形态改造) ─────
subgraph LA5["🔧 Layer A.5 · transforms(yaml 声明式,6 operator 白名单)"]
  direction LR
  T1["project<br/>选列/改名"]
  T2["split_json_array<br/>JSON 数组拆行"]
  T3["derive<br/>substring/concat"]
  T4["route_by_pattern<br/>关键词分流"]
  T5["pick_first_nonzero<br/>多列推断单列"]
  T6["filter<br/>多字段行过滤"]
end

%% ───── Layer A · AssemblerEngine(yaml 翻译) ─────
LA["🔄 Layer A · AssemblerEngine<br/>━━━━━━━━━━━━━━━━━━<br/>field_mapping(命名翻译)<br/>enum_mapping(枚举闭集 + _default 兜底)<br/>emits(action/subjectType/occurredAtField)<br/>dedup_by(炸行兜底)"]

%% ───── L1 · 患者事实层 ─────
subgraph L1["📘 L1 · 患者事实层"]
  direction TB
  TX["patient_transactions<br/>━━━━━━━━━━━━<br/>append-only 账本<br/>raw_payload + canonical_payload 双留底<br/>source_event_id 幂等键"]
  PARSER{{"Parser + zod 强校验<br/>━━━━━━━━━━<br/>17 个 parser<br/>FactContentSchemas 闭集<br/>违反 schema 抛错不入库"}}
  FACTS["patient_facts<br/>━━━━━━━━━━━━<br/>事实单元版本流<br/>━━━━━━━━━━━━<br/>diagnosis_record  K08/牙位<br/>treatment_record  actual/planned<br/>· 11 category(含 review)<br/>recommendation_record  建议XX<br/>emr_record    自由文本<br/>encounter_record  接诊元数据<br/>appointment_record  预约状态<br/>payment_record  amount_cents"]
  TX -->|1 transaction<br/>→ 0/1/N fact 版本| PARSER
  PARSER --> FACTS
end

%% ───── L2 · Persona ─────
subgraph L2["🟠 L2 · Persona 画像层"]
  PERSONA["personas<br/>━━━━━━━━━━<br/>版本流 + eventWatermark 幂等水位"]
  FEAT["persona_features<br/>━━━━━━━━━━<br/>v1 起步 4 特征:<br/>value(钻/金/银/普通/新客)<br/>treatment_chain_status<br/>recall_risk<br/>do_not_contact_status"]
  PERSONA --> FEAT
end

%% ───── L3 · 召回计划 ─────
subgraph L3["🟣 L3 · 召回计划层"]
  SCEN["Scenario · treatment_initiation_recall<br/>━━━━━━━━━━━━━━━━━━━━<br/>3 子规则(SQL EXISTS / NOT EXISTS):<br/>#A K08 缺牙 → implant 未做<br/>#B K02 龋齿 → restorative 未做<br/>#C K05 牙周 → periodontic 未做<br/>(review 类不算已做)"]
  SCORER["6 因子 Priority Scorer<br/>━━━━━━━━━━━━━━━━━━<br/>①ClinicalBase ②TimeWindowFactor<br/>③ValueBonus ④LikelihoodBonus<br/>⑤UrgencyBonus ⑥SignalQualityDiscount<br/>score = clamp((①×② + ③+④+⑤) × ⑥, 0-100)"]
  PLAN["followup_plans + plan_reasons<br/>━━━━━━━━━━━━━━━━━━<br/>plan = patient 级触达单元<br/>(同 patient 多 sub-rule 合并 1 plan)<br/>breakdown(6 因子 JSONB)"]
  SCEN -->|hit| SCORER
  SCORER -->|score+breakdown| PLAN
end

%% ───── UI ─────
subgraph UI["👤 客服 UI · /plans"]
  LIST["召回池列表<br/>按 priority_score 排序"]
  DETAIL["详情页<br/>━━━━━━━━━━<br/>WhyCard:6 因子拆解<br/>FactsTimeline:fact 时间轴<br/>ScriptStream:话术(DeepSeek)"]
end

%% ───── 横切 ─────
subgraph CROSS["🔧 横切支撑"]
  direction LR
  SL["sync_logs<br/>同步运行"]
  PRL["persona_recompute_logs<br/>画像重算"]
  PGL["plan_generation_logs<br/>召回生成"]
end

%% ───── 主流(实线) ─────
COLD ==> LA5
DW ==> LA5
PULL ==> LA5
PUSH -.->|绕过 transforms<br/>action 自带| LA
LA5 ==> LA
LA ==> TX
FACTS -->|触发| PERSONA
FEAT -->|输入| SCEN
FACTS -->|主信号| SCEN
PLAN ==> LIST
LIST -->|点击| DETAIL

%% ───── 编排关联(虚线) ─────
L0 -.-> SL
PERSONA -.-> PRL
PLAN -.-> PGL

%% ───── 样式 ─────
style L0  fill:#fff3e0,stroke:#f57c00,stroke-width:2px
style LA5 fill:#fffde7,stroke:#fbc02d,stroke-width:2px
style LA  fill:#e0f7fa,stroke:#00838f,stroke-width:2px
style L1  fill:#e3f2fd,stroke:#1976d2,stroke-width:2px
style L2  fill:#fff3e0,stroke:#f57c00,stroke-width:2px
style L3  fill:#f3e5f5,stroke:#7b1fa2,stroke-width:2px
style UI  fill:#e8f5e9,stroke:#388e3c,stroke-width:2px
style CROSS fill:#f5f5f5,stroke:#616161
```

---

## 4 入口统一管道(W3 末确立)

PAC 的 ingestion 是一条管道,4 个入口接不同 stage,后续完全共用:

| 入口 | Stage 0 取数 | Stage 1 transforms | Stage 2 Assembler | Stage 3 Synthesizer emits | Stage 4 Parser |
|---|---|---|---|---|---|
| **Cold-import**(SQL/CSV) | 文件/SQL 朴素 SELECT | ✅ 跑 | ✅ 跑(yaml emits) | yaml | ✅ 共用 |
| **DW 直连增量**(W5+) | SQL + cursor | ✅ 跑 | ✅ 跑 | yaml | ✅ 共用 |
| **Pull HTTP**(W5+ 5i5ya) | HTTP GET | ✅ 跑 | ✅ 跑 | yaml | ✅ 共用 |
| **Push Webhook**(W5+) | host 直推 event | ❌ 跳过 | 视形态可选 | **event 自带** | ✅ 共用 |

**纪律**:**Stage 3-4 后所有路径汇合,`fact.content` 是 PAC 唯一真理源**。后续 Persona/Plan/AI 永远只读 fact,不读 raw_payload。

---

## 各层做了什么(评审用速读)

| 层 | 输入 | 处理 | 输出 | 关键设计 |
|---|---|---|---|---|
| **L0 数据源** | — | 4 入口取 raw rows(SQL/CSV/HTTP/event)| `Record<tableName, Row[]>` | 4 入口形态一致,下游无感知 |
| **Layer A.5 transforms** | raw tables | yaml 声明式形态改造(6 operator 白名单)| 整理后的 tables(JSON 已拆行 / 多列推断完成 / 关键词分流到 3 路) | **无副作用纯函数**,撞墙必须加新 operator + PR |
| **Layer A AssemblerEngine** | 整理后 tables | field_mapping + enum_mapping(`_default` 兜底)+ dedup_by + emits 推断 | canonical rows + emits | **集团统一引擎**,字典闭集 |
| **L1 患者事实** | canonical rows + emits | TransactionSynthesizer + Parser + zod | `patient_transactions`(双留底)+ `patient_facts`(版本流)| **fact 是 PAC 唯一真理源**,下游只读 |
| **L2 Persona 画像** | active facts | 4 个 rule-based 特征(纯 TS 算法)| `personas` + `persona_features`(eventWatermark 幂等) | "特征 + 描述 + 证据 + 版本",不字段化拆 |
| **L3 召回计划** | facts + personas | scenario SQL 入池 + 6 因子算分 + 同 patient 多 sub-rule 合并 | `followup_plans` + `plan_reasons`(含 breakdown JSONB)| **plan = patient 级触达单元**,UNIQUE 保证 1 patient 1 active plan |
| **UI · 客服工作台** | plan-aggregate API | 列表排序 + 详情可解释性 | 客服可点开"为什么 87 分" | 6 因子 breakdown 完全可解释,无黑盒 |

---

## 关键技术决策(W3 末)

### 决策 1 · SQL 退化为"等价 csv 单表 dump"

SQL 只允许:`SELECT col1, col2 FROM single_table WHERE filter`。
**禁止**:ARRAY JOIN / JSONExtract / multiIf / substring / 跨表 JOIN(那是形态改造,归 transforms)。

**好处**:直连 DW / csv dump / HTTP pull 给的数据**完全等价**,host 任意切换数据交付方式,PAC 这边一行不动。

### 决策 2 · 形态改造统一 yaml(6 operator 闭集)

| operator | 干什么 | 例子 |
|---|---|---|
| `project` | 选列/改名 | `keep: [id, patient_id]`、`rename: { id: emr_id }` |
| `split_json_array` | JSON 数组字段拆行 | EMR.diag JSON → N 行诊断子记录 |
| `derive` | 字段派生(substring/concat/lower/upper/default)| `K05.300 → K05` |
| `route_by_pattern` | 按字段值分流到多输出表 | treatName → recommendation / review / treatment 三路 |
| `pick_first_nonzero` | 多个数值列挑第一个 > 0 对应 label | 17 个 pay_* 列推 payment_channel |
| `filter` | 多字段 AND 行过滤(等价 WHERE)| `in_time: { not_empty: true }` |

**纪律**:不允许任意 JS 表达式 / eval。撞墙必须加 operator(走 PR + review)。

### 决策 3 · canonical-codes 加第 11 category `review`

**问题**:DW EMR treat_plan 大量含"常规复查/拆线/暂观/无治疗"等流程性动作 — 是真实事实,但**不是治疗动作**。
**方案**:加 `review` category 承载这类事实,scenario 算法不消费(不算"已做治疗")。
- ✅ fact 层保真(2714 行 review fact 入库)
- ❌ 不污染 preventive(避免"建议种植"被误判为"已做预防")

### 决策 4 · enum_mapping `_default` 兜底键

`AssemblerEngine.applyEnum` 查表顺序:`fieldMap[key]` → `fieldMap._default` → 原值。
**用法**:settlement_type 全枚举 ~45 个 → PAC category,长尾未识别 → `_default: ""` 让 parser 跳过(silently warn drop)。

---

## 真实数据案例(评审"故事化"用)

> **卜晓平 · 男 · 1985** · 杭州大厦诊所
> **故事**:2025-12-21 接诊,医生 EMR 诊断 **K08 缺失牙 17 号**,医嘱建议种植修复。**151 天过去**,患者无任何 implant / prosthodontic 治疗记录,无新预约。
>
> ## PAC 算法决策
>
> | 阶段 | 输出 |
> |---|---|
> | **L0/A.5/A**(数据源 → transforms → assembler) | DW EMR 单行 → transforms 拆出 1 个 encounter + 1 个 diagnosis(K08/17)+ 1 个 emr_record + 1 个 treatment_record(review:常规复查)→ assembler 翻译入 canonical |
> | **L1 fact**(parser + zod 强校验) | `encounter_record` × 1 + `diagnosis_record:K08:17` × 1 + `emr_record` × 1 + `treatment_record:review` × 1 |
> | **L2 Persona** | value=钻卡(LTV ¥57k+)/ treatment_chain_status=gap(K08 无 implant)/ recall_risk=low |
> | **L3 Scenario `treatment_initiation_recall`** | 命中子规则 `missing_tooth`(K08 ∈ 30-180d 窗 + 无 implant actual treatment)|
> | **6 因子算分** | base 60 × tw 1.0 + value+20 + likelihood+2 + urgency+5(>120d 邻牙倾斜)= **87 分** |
> | **Plan 落库** | priorityScore=87 / goal="邀约启动缺失牙修复(种植/桥/义齿)" / recommendedRole=staff / recommendedChannel=phone |
> | **客服 UI** | 召回池排第一;详情页 WhyCard 自动渲染 6 因子拆解 + FactsTimeline 时间轴 |

**全程纯规则,无 LLM**(W5+ 接 LLM 后,只是把 emr_record 自由文本里的"建议种植"抽成独立 `recommendation_record`,算法不变)。

---

## 完备性 / 可靠性的几道闸(底色)

| # | 闸 | 防什么 |
|---|---|---|
| 1 | `source_event_id` partial UNIQUE | 同 transaction 重复入库 |
| 2 | `patient_facts.subjectId active UNIQUE` | 同 fact 多 active 版本冲突 |
| 3 | `FactContentSchemas` zod 强校验 | parser 字段漂移让脏数据入 fact |
| 4 | `canonical-codes.ts` 闭集 + `_default` 兜底 | yaml 翻译目标在闭集外;长尾自动 drop |
| 5 | `dedup_by` yaml 兜底 | DW ETL 炸行(EMR 5x / settlement 5x)|
| 6 | `event_seq` BigInt 水位 | Persona 重算丢漏事件 |
| 7 | 6 operator 白名单 | yaml 越权改 fact 主体(只允许声明式纯函数) |
| 8 | `breakdown` JSONB 落库 | 算法决策无解释性(可视性=可信任) |
| 9 | CI 防漂移测试 53 个 | 加 fact_type 忘加 zod / yaml enum 漂移 / operator 边界回归 |

---

## 数据规模(实测 W3 末)

| 数据 | 规模 |
|---|---|
| jvs-dw 5 试点诊所 | 14 万患者(瑞尔 8.4 万 + 瑞泰 4.6 万) |
| 一期 cohort 试运行 | **1,000 患者**(最近就诊优先,可调) |
| DW raw 行数(过滤后) | EMR 9k / 预约 18.8k / 结算 42.8k / 支付 5.2k |
| Layer A.5 transforms 产出 | diagnosis 560 / treatment(planned)6,788 / treatment(review)2,714 / treatment(actual)11,924 / recommendation 55 |
| 入库 transactions | **63,781**(0 failed) |
| 入库 facts | **~60,000**(0 failed,review category 2,714 行真实入库)|
| 召回率 | ~3.2%(W3 末跑通,完整数字待新一轮 plan recompute)|
| Top patient | 卜晓平 87 分(K08 缺牙 151 天) |

---

## W3 末实现状态

| 模块 | 状态 |
|---|---|
| Cold-import(SQL 直连 / CSV)| ✅ 跑通,0 fact failed |
| Layer A.5 transforms(6 operator)| ✅ 实现 + 49 测试通过 |
| Layer A AssemblerEngine(`_default` 兜底)| ✅ |
| Parser pipeline + FactContentSchemas zod | ✅ |
| Persona 4 起步特征 | ✅ |
| 6 因子 Priority Scorer + breakdown | ✅ |
| Scenario `treatment_initiation_recall`(3 子规则)| ✅ |
| 详情页 UI(WhyCard + FactsTimeline + ScriptStream)| ✅ |
| 4 入口统一管道(dispatcher emitsResolver)| ✅ P0 补完 |
| Mock Pull 端到端 | ✅ 自带 emitsPerResource |
| Push 形态 A(canonical payload)| ✅ event 自带 emits |
| Push 形态 B(host raw payload)| 🟡 框架预留,需 host 时实装 |
| Real HTTP Pull(api-key / login)| 🟡 框架预留,W5+ 接真 host |
| DW 直连增量 cursor | 🟡 框架预留(cold-import 已支持 SQL,加 cursor 即可)|
| Layer C 自由文本抽取(LLM)| 🔵 W5+ 计划 |
