lzf_0626
d4836ed4df
一、装配 projection_cleaner 新增 app/service/projection_cleanup_service.py，并在**组装层**（app/worker/__main__.py） 注入。此前该客户端一直未装配，清理链路只记录降级（skipped_no_client），实际后果是 **销户后图里仍留着偏好关系**——投顾仍能通过关系网络看到这个人。 清理策略是"以画像为准"而不是按标识直接删边：先删掉该记忆对应的长期事实，再重建画像， 最后用对账修复让图跟着画像收敛。这样即使一条记忆影响多条派生边也能删干净，不依赖 "记得它当初投影成了什么"。Milvus 侧按 memory_uuid 删除；集合不存在（语义召回未启用） 时视为无需清理，客户端不可用则如实报告未清理，绝不伪造成功。 放在组装层而不是 runtime 内部兜底：组件内部给默认实现会把"尚未装配"这一事实悄悄盖住， 而"未注入即显式降级并留痕"是 runtime 刻意保留的语义。最初的改法写成内部兜底，被 5 个 既有单测拦下——那些测试是对的，因此改为在入口处注入。 二、顺带修复：事实消失后画像字段残留 实测：让 preference:horizon 记忆失效并清理后，user_facts 与图边都正确清除，但 fin_customer_profile.investment_horizon 仍是"长期（5年以上）"。根因是 rebuild_profile 只写"本轮有新值"的字段；事实被删后该字段没有新值，旧值就留在画像里——于是记忆已经作废， 投顾还能看到一个客户从未授权继续生效的投资期限。 改为：本轮没有对应事实时**清空**该字段；investor_type 例外——问卷是唯一权威，本轮没有 问卷记录时保持原值（该列 NOT NULL，且清空会在重测空档抹掉开户时的等级）。 三、实测 · 让 preference:horizon 失效后清理：cleaned=True， detail=profile_rebuilt; graph_cleaned; vector_collection_absent； · user_facts 2→1；图边由 ['HAS_GOAL','PREFERS'] 变为 ['PREFERS']； · 画像 investment_horizon 由"长期（5年以上）"变为 None；investor_type 保持 C2、 risk_tags 不受影响（与 horizon 无关）； · 恢复记忆状态后重建，画像与图边均正确复原； · ruff 通过、mypy 113 文件无错、unit+contract 447 passed、integration 29 passed。
268 lines
12 KiB
Python
268 lines
12 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」；因此必须先清旧再写新，顺序不能反。
|
||
"""
|
||
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
|
||
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,
|
||
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
|