wangjianlong_0626
451aa4e915
## 现象（2026-09-12 端到端跑通后查数据时发现） 客户 9102 的画像快照状态**彼此矛盾**： | version | is_current | current_customer_id | |---|---|---| | 2 | `0` | `9102` ← 清旧时没清空 | | 3 | `1` | `NULL` ← 建新时没写入 | ## 根因 `ProfileAssemblyService._write_snapshot` 把 `current_customer_id` **当成了生成列**： - 方法 docstring 原文写着"唯一键 `uk_profile_snapshot_current` 建立在**生成列** `current_customer_id` 上" - 因此两处都不赋值（以为 DB 会自动填） **但该列不是生成列** —— `alembic/baseline_generated.sql` 与真实库都是**普通可空列 + 唯一键**， `app/model/profile.py` 的模块 docstring 第 2 条已明确："当前版本必须由写入方**显式写入**客户 ID （历史版本写 NULL），才能保证「每个客户最多一条当前快照」"。 后果：唯一键**形同虚设**（多个 NULL 不冲突）⇒ 不变式失效；且旧版本残留的值 一旦与新版本补上的值相同，就会**直接撞唯一键**。 > 这与本线先前修的 `CustomerProfileCandidateService._write_profile_snapshot` 是**同一个缺陷的另一处** > —— 当时只找到一处，这次是靠真实链路跑出数据后核对才暴露出来。 ## 改动 `app/service/profile_assembly_service.py`： - 旧版本：`is_current = False` 的同时 `current_customer_id = None` - 新版本：`is_current=True` 的同时 `current_customer_id=customer_id` - 订正方法 docstring 的错误认知（"生成列"→ 普通可空列 + 唯一键），并写明后果 ## 已有数据订正 新增 `tools/fix_profile_snapshot_current.py`（**默认 dry-run**、幂等、`--apply` 才提交）： 1. 先清空 `is_current=0` 却残留值的行 2. 再补写 `is_current=1` 却是 NULL 的行 3. **顺序要紧**：反过来的话第 2 步会与残留值撞唯一键 本机实测：清空 1 行、补写 1 行，复核两类异常均归零。 ## 验证 - `mypy app` → 0 错 / 245 文件 - `pytest tests`（全量）→ `2 failed, 1418 passed, 1 skipped` （2 个失败为既有环境项：httpx 把中文序列化成 `\uXXXX`，非本次引入） - 端到端：真实对话 → 记忆抽取 → `memory_unit` 落库已实测通过（客户 9102 `preference:risk_level = "低风险"`，候选态）
280 lines
13 KiB
Python
280 lines
13 KiB
Python
"""画像组装：中期记忆 → 长期事实 → 画像 + 版本快照。
|
||
|
||
这是"记忆系统为画像服务"的落地环节。三段职责：
|
||
|
||
1. **事实提升（中期 → 长期）**：把 `memory_unit` 里证据足够的记忆提炼进 `user_facts`。
|
||
门槛是 `evidence_count >= 2` **或** `confidence >= 0.90` —— 这条门槛就是
|
||
"客户随口一说不能变成画像结论"的落地方式。
|
||
2. **画像组装（长期 → 画像）**：把 `user_facts` 按**白名单**映射进 `fin_customer_profile`。
|
||
未列入白名单的事实只进 `user_facts`，不进画像，避免画像被噪声撑大。
|
||
3. **版本留痕**：每次重建写一条 `profile_snapshots`，并用 `generation_basis` 记录
|
||
**每个字段分别来自哪里**——风控与合规复盘时要能回答"当时凭什么这么判断"。
|
||
|
||
## 两条必须由代码保证的红线
|
||
|
||
- **`investor_type` 只来自问卷测评**（`fin_risk_assessment` 最新一条）。下面的实现里
|
||
它只从问卷查询取数，任何记忆路径都碰不到它。客户在对话里说"我是激进型"不会改变它
|
||
——这是合规底线，不能只靠约定。
|
||
- **按字段所有权写入**：交易侧的客观字段（`total_asset`/`trading_frequency`/`behavior_score`）
|
||
本服务**不写**，留给交易模块，避免两个模块抢写同一列。
|
||
"""
|
||
|
||
import json as _json
|
||
from datetime import UTC, datetime
|
||
from hashlib import sha256
|
||
from typing import Any
|
||
from uuid import uuid4
|
||
|
||
from sqlalchemy import or_, select, text
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.model.fund import FundCustomerProfile
|
||
from app.model.memory import MemoryUnit
|
||
from app.model.profile import ProfileSnapshot, UserFact
|
||
|
||
# 提升门槛
|
||
MIN_EVIDENCE = 2
|
||
HIGH_CONFIDENCE = 0.90
|
||
|
||
# 事实键 → 画像字段的**白名单**映射。没列在这里的事实（如 profile:family）只进 user_facts，
|
||
# 不进画像字段——画像要保持"能直接支撑决策"的信噪比。
|
||
FACT_TO_PROFILE_FIELD: dict[str, str] = {
|
||
"preference:asset_class": "preferred_asset_class",
|
||
"preference:horizon": "investment_horizon",
|
||
}
|
||
|
||
# 自述类事实（客户自己说的偏好）统一进 risk_tags，并标注来源为"自述"。
|
||
# 保留它们的价值在于：当出现「问卷 C4 / 自述稳健 / 行为买 R4」三方不一致时，
|
||
# 这种矛盾本身就是风控信号——但绝不能与问卷等级混进同一个字段。
|
||
SELF_REPORTED_PREFIXES = ("preference:risk_level", "preference:", "profile:")
|
||
|
||
# 关键事实：参与决策，标记出来便于下游优先读取
|
||
CRITICAL_FACTS = frozenset({
|
||
"preference:risk_level", "preference:horizon", "preference:asset_class",
|
||
})
|
||
|
||
# 画像中允许本服务写入的字段（其余字段归交易/注册侧所有）
|
||
PROFILE_OWNED_FIELDS = ("investor_type", "preferred_asset_class", "investment_horizon", "risk_tags")
|
||
|
||
|
||
def _now() -> datetime:
|
||
return datetime.now(UTC).replace(tzinfo=None)
|
||
|
||
|
||
def _fact_id() -> int:
|
||
"""`user_facts.id` 没有 auto_increment，主键由应用生成。
|
||
|
||
用微秒时间戳：单调递增、无需额外序列、同客户同微秒重复在单进程写入下不可能发生。
|
||
"""
|
||
return int(datetime.now(UTC).timestamp() * 1_000_000)
|
||
|
||
|
||
class ProfileAssemblyService:
|
||
def __init__(self, session: AsyncSession) -> None:
|
||
self.session = session
|
||
|
||
# ---------- 中期 → 长期 ----------
|
||
|
||
async def promote_facts(self, customer_id: int) -> list[str]:
|
||
"""把证据足够的记忆提炼为长期事实；返回本次提升的事实键。"""
|
||
now = _now()
|
||
rows = list(await self.session.scalars(
|
||
select(MemoryUnit).where(
|
||
MemoryUnit.customer_id == customer_id,
|
||
MemoryUnit.status == "active",
|
||
or_(
|
||
MemoryUnit.evidence_count >= MIN_EVIDENCE,
|
||
MemoryUnit.confidence >= HIGH_CONFIDENCE,
|
||
),
|
||
or_(MemoryUnit.valid_until.is_(None), MemoryUnit.valid_until > now),
|
||
)
|
||
))
|
||
promoted: list[str] = []
|
||
for memory in rows:
|
||
key = str(memory.memory_key)
|
||
value = self._fact_value(memory)
|
||
existing = await self.session.scalar(
|
||
select(UserFact).where(
|
||
UserFact.customer_id == customer_id, UserFact.fact_key == key
|
||
)
|
||
)
|
||
if existing is None:
|
||
self.session.add(UserFact(
|
||
# 主键显式赋值：该表无 auto_increment
|
||
id=_fact_id(),
|
||
customer_id=customer_id,
|
||
fact_key=key,
|
||
fact_value=value,
|
||
source_portal=str(memory.source_type or "conversation"),
|
||
source_episode_id=None,
|
||
confidence=float(memory.confidence or 0.0),
|
||
is_critical=key in CRITICAL_FACTS,
|
||
created_at=now,
|
||
))
|
||
else:
|
||
existing.fact_value = value
|
||
existing.confidence = float(memory.confidence or 0.0)
|
||
existing.is_critical = key in CRITICAL_FACTS
|
||
promoted.append(key)
|
||
await self.session.flush()
|
||
return promoted
|
||
|
||
@staticmethod
|
||
def _fact_value(memory: MemoryUnit) -> Any:
|
||
"""事实值优先取结构化值，回退到正文；始终以 JSON 可存的形式返回。"""
|
||
structured = memory.structured_value
|
||
if isinstance(structured, dict) and "value" in structured:
|
||
return structured["value"]
|
||
if structured is not None:
|
||
return structured
|
||
return memory.content or ""
|
||
|
||
# ---------- 长期 → 画像 ----------
|
||
|
||
async def rebuild_profile(self, customer_id: int) -> dict[str, Any]:
|
||
"""用长期事实 + 问卷重建画像，并写一条版本快照。"""
|
||
now = _now()
|
||
facts = list(await self.session.scalars(
|
||
select(UserFact).where(UserFact.customer_id == customer_id)
|
||
))
|
||
assessment = (await self.session.execute(text(
|
||
"""
|
||
SELECT investor_type, questionnaire_version, assessed_at, valid_until
|
||
FROM fin_risk_assessment
|
||
WHERE customer_id = :customer_id
|
||
ORDER BY assessed_at DESC, id DESC
|
||
LIMIT 1
|
||
"""
|
||
), {"customer_id": customer_id})).first()
|
||
|
||
values: dict[str, Any] = {}
|
||
basis: dict[str, Any] = {}
|
||
|
||
# 红线：风险等级只从问卷取；记忆里哪怕有 preference:risk_level 也不写这个字段
|
||
if assessment is not None and assessment[0]:
|
||
values["investor_type"] = str(assessment[0])
|
||
basis["investor_type"] = {
|
||
"source": "fin_risk_assessment",
|
||
"questionnaire_version": assessment[1],
|
||
"assessed_at": str(assessment[2]),
|
||
"valid_until": str(assessment[3]),
|
||
}
|
||
|
||
tags: list[str] = []
|
||
for fact in facts:
|
||
key = str(fact.fact_key)
|
||
field = FACT_TO_PROFILE_FIELD.get(key)
|
||
if field is not None:
|
||
values[field] = self._as_text(fact.fact_value)
|
||
basis[field] = {
|
||
"source": "user_facts", "fact_key": key,
|
||
"confidence": float(fact.confidence or 0.0),
|
||
}
|
||
elif key.startswith(SELF_REPORTED_PREFIXES):
|
||
# 自述信息进标签，并显式标注"自述"，与问卷等级区分开
|
||
tags.append(f"自述:{key}={self._as_text(fact.fact_value)}")
|
||
basis.setdefault("risk_tags", {"source": "user_facts", "items": []})
|
||
basis["risk_tags"]["items"].append(key)
|
||
if tags:
|
||
values["risk_tags"] = "；".join(tags)
|
||
|
||
profile = await self.session.get(FundCustomerProfile, customer_id)
|
||
if profile is None:
|
||
# 画像行由开户流程创建：`trade_account` 等身份字段在库里是 NOT NULL，属注册／账户侧
|
||
# 所有。本服务不代替开户去造这些数据——否则会写出一条**假的**开户记录，
|
||
# 而画像恰恰是风控要读的东西，假数据比没有数据更危险。未开户时如实报告。
|
||
return {
|
||
"profile": None,
|
||
"reason": "profile_row_not_opened",
|
||
"generation_basis": basis,
|
||
"promoted": len(facts),
|
||
}
|
||
for field in PROFILE_OWNED_FIELDS:
|
||
if field == "investor_type":
|
||
# 问卷是唯一权威：本轮没有问卷记录时**保持原值**（既不写入也不清空）。
|
||
# 否则重测前的空档会把开户时的等级抹掉，而该列是 NOT NULL。
|
||
if field in values:
|
||
setattr(profile, field, values[field])
|
||
continue
|
||
# 其余字段由本服务独占：本轮没有对应事实即清空。
|
||
# 这不只是洁癖——记忆失效后若不清理，画像会留着一个已经作废的投资期限，
|
||
# 投顾据此给建议，而客户从未授权这条信息继续生效（实测踩到过）。
|
||
setattr(profile, field, values.get(field))
|
||
profile.updated_at = now
|
||
await self.session.flush()
|
||
|
||
snapshot = {
|
||
**{field: getattr(profile, field, None) for field in PROFILE_OWNED_FIELDS},
|
||
"generated_at": now.isoformat(),
|
||
}
|
||
await self._write_snapshot(customer_id, snapshot, basis, now)
|
||
return {"profile": snapshot, "generation_basis": basis, "promoted": len(facts)}
|
||
|
||
@staticmethod
|
||
def _as_text(value: Any) -> str:
|
||
"""把 JSON 列里取出的值渲染成可读字符串。
|
||
|
||
字符串类型的值可能带着 JSON 序列化时的外层引号（取决于驱动如何回读 JSON 列），
|
||
这里去掉它们——`risk_tags` 是给风控与投顾看的，多一对引号会让人以为值本身包含引号。
|
||
"""
|
||
if isinstance(value, str):
|
||
return value.strip().strip('"')
|
||
return _json.dumps(value, ensure_ascii=False)
|
||
|
||
async def _write_snapshot(
|
||
self, customer_id: int, snapshot: dict[str, Any], basis: dict[str, Any], now: datetime
|
||
) -> None:
|
||
"""写入新版本快照并把旧版本置为非当前。
|
||
|
||
唯一键 `uk_profile_snapshot_current` 建立在 `current_customer_id` 上，
|
||
保证「每个客户最多一条 current」；因此必须先清旧再写新，顺序不能反。
|
||
|
||
⚠️ **该列不是生成列**（订正 2026-09-12）：它由 `alembic/baseline_generated.sql`
|
||
与真实库确认为**普通可空列 + 唯一键**，`app/model/profile.py` 的模块 docstring
|
||
第 2 条已写明"当前版本必须由写入方**显式写入**客户 ID（历史版本写 NULL）"。
|
||
此处原先按"生成列"理解而两处都不赋值，后果实测到：
|
||
旧版本 `is_current=0` 却仍留着 `current_customer_id`，新版本 `is_current=1`
|
||
却是 NULL —— **唯一键形同虚设**（多个 NULL 不冲突），
|
||
而且旧值一旦残留、新值再补写就会直接撞键。
|
||
"""
|
||
previous = list(await self.session.scalars(
|
||
select(ProfileSnapshot).where(
|
||
ProfileSnapshot.customer_id == customer_id, ProfileSnapshot.is_current.is_(True)
|
||
)
|
||
))
|
||
for row in previous:
|
||
row.is_current = False
|
||
# 必须同时清空，否则与即将插入的新当前版本撞唯一键（见 docstring）。
|
||
row.current_customer_id = None
|
||
row.updated_at = now
|
||
await self.session.flush()
|
||
|
||
latest = await self.session.scalar(text(
|
||
"SELECT COALESCE(MAX(version), 0) FROM profile_snapshots WHERE customer_id = :cid"
|
||
), {"cid": customer_id})
|
||
version = int(latest or 0) + 1
|
||
payload = _json.dumps(snapshot, ensure_ascii=False, sort_keys=True)
|
||
self.session.add(ProfileSnapshot(
|
||
profile_uuid=str(uuid4()),
|
||
customer_id=customer_id,
|
||
version=version,
|
||
snapshot=snapshot,
|
||
generation_basis=basis,
|
||
snapshot_hash=sha256(payload.encode("utf-8")).hexdigest(),
|
||
is_current=True,
|
||
# 当前版本必须显式写入客户 ID（历史版本为 NULL），见 docstring。
|
||
current_customer_id=customer_id,
|
||
generated_at=now,
|
||
created_at=now,
|
||
updated_at=now,
|
||
))
|
||
await self.session.flush()
|
||
|
||
# ---------- 完整链路 ----------
|
||
|
||
async def rebuild(self, customer_id: int) -> dict[str, Any]:
|
||
promoted = await self.promote_facts(customer_id)
|
||
outcome = await self.rebuild_profile(customer_id)
|
||
outcome["promoted_keys"] = promoted
|
||
return outcome
|