wangjianlong_0626
8a0cbab636
背景：memory_sync_outbox 这条链此前**完全没有消费者**，且生产端照 docs/00 §6.4.6
写成大写 MILVUS/NEO4J + 中文「待处理」，而消费端按 target_store 的**值**分派 handler、
且只领 status in {pending, failed} —— 两个条件都不满足，事件任何消费者都领不到、
永久滞留且不报错（唯一键 (event_uuid, target_store) 对大小写无约束，MySQL 也不报错）。
根因是代码与测试都硬编码字面量，所以测试跟着一起错、谁也没拦住。
订正
- profile_generation_service：取值改为全仓一致的小写（milvus/neo4j/upsert/pending）
- 测试改为引用常量并断言消费端契约，不再硬编码（硬编码是本次跑偏的直接原因）
- 新增契约回归测试：断言大写值分派不到 handler、会进死信，谁改回大写立刻红
- 新增 tools/normalize_memory_sync_outbox.py：订正历史脏行（默认 dry-run、幂等）
接通投影链路（此前零消费者）
- 新增 Milvus 集合 user_long_term_memory_v1 及建集合工具（幂等、不覆盖已有集合）
- 新增 MilvusProfileProjection / MilvusProfileVectorClient，并修掉移植带来的两处必炸点：
customer_id 由「必须 int」放宽为接受数字字符串（本仓所有生产者都写 str，
不放宽则每个事件必然失败）；不可投影的 memory_key 由「整批 raise」改为跳过留痕
（否则一条 constraint: 记忆毒死该客户整批，而受控词表 13 个键里有 7 个不满足前缀）
- 新增 MemorySyncOutboxWorker（领取/指数退避/死信骨架保留原样）并接入 WorkerRuntime
- milvus → 向量投影；neo4j → 复用主干 ProfileGraphProjectionService（方案 A，
不引入第二套投影，避免同一事实在图中两种说法、违反主干既有的只投影已确认事实的不变式）
- 生产端从 memory_unit(status=active) 组装 memory_sources，随事件带上确定快照
- 前置移植 conversation_privacy：写外部存储前脱敏手机号/证件号/银行卡等
验证
- 新增 17 个单测；全量 2 failed, 1307 passed, 2 skipped
（2 个失败为既有环境项：断言请求体中文原文而 httpx 序列化成 \uXXXX，非本次引入）
- mypy app → 0 错（227 文件）；audit_schema → 89 张业务表无缺失/意外，未改动表结构
- 真机：真实 embedding(1024 维) + 真实 Milvus 写入并回读通过
- 整合链路（测试记忆 → 生产端组装 → outbox → 消费端投递 → Milvus 回读）通过，
且 MySQL 已回滚、Milvus 无残留
文档
- 新增 docs/32-记忆投影链路实现说明.md：真实口径、根因、契约与验证证据（供接手）
- AGENTS.md：新增该易错点；新增 Windows 中文输出乱码的正确命令（-X utf8）；
校正测试基线与 mypy 文件数
未做：未改 docs/00 基线、未动数据库迁移、未改投顾线代码、未启动常驻 Worker。
遗留：投顾线两处生产者的 payload 缺 memory_sources，会被消费至死信，待架构师确认是否投影。
288 lines
12 KiB
Python
288 lines
12 KiB
Python
"""画像版本生成服务（`docs/00` §6.4.6 的生成器实现）。
|
||
|
||
## 为什么需要它
|
||
|
||
底座**规划了** `profile_snapshots` 版本投影表，也**写明了规则**，但**没有生成器**：
|
||
|
||
> 「画像生成只允许读取已提交的业务事实和状态为"有效"的中期记忆。新画像和两条同步事件
|
||
> 必须在同一个 MySQL 事务中写入，旧画像的 `is_current` 同时置为 0。」
|
||
> —— `docs/00` §6.4.6
|
||
|
||
本服务把这条规则落成代码。**在它之前**，`profile_snapshots` 里的数据只能由种子脚本直接写，
|
||
"画像怎么生成的"没有可运行答案。
|
||
|
||
## 落地的四条规则
|
||
|
||
1. **同事务**：新画像版本 + **两条**跨存储同步事件（MILVUS / NEO4J）一起提交，任一失败整体回滚。
|
||
2. **旧版本同时置 0**：`is_current` 是 `profile_snapshots` 的唯一当前标记
|
||
（`current_customer_id` 上有唯一键），必须先清旧再插新。
|
||
3. **幂等**：`snapshot_hash` 未变化时**不产生新版本**（返回 `changed=False`），
|
||
避免每次调用都把版本号 +1、并往 outbox 灌重复事件。
|
||
4. **只读已提交事实**：数据来源限定为 `fin_customer_profile` 与 `fin_risk_assessment`
|
||
（以及有效中期记忆的口径由 `generation_basis` 记录），不读取未提交状态。
|
||
|
||
## 与读取侧的关系
|
||
|
||
生成的 `snapshot` JSON 结构**必须与 `app/core/profile_projection.py` 的白名单字段一致** ——
|
||
投影是只读白名单，写侧多写字段不会被对外暴露，但白名单要求的字段**必须存在**，
|
||
否则端点会返回空画像。这一点由 `test_profile_generation_service.py` 固定。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
from collections.abc import Mapping
|
||
from dataclasses import dataclass
|
||
from datetime import UTC, datetime
|
||
from typing import Any
|
||
from uuid import uuid4
|
||
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.core.errors import ValidationAgentError
|
||
from app.core.profile_projection import PROFILE_FIELD_POLICY
|
||
from app.repository.profile_repository import ProfileRepository
|
||
|
||
#: `memory_sync_outbox` 的两个目标存储。
|
||
#:
|
||
#: ⚠️ **一律小写**：消费端按 `target_store` 的值分派 handler
|
||
#: （`MemorySyncOutboxWorker.handlers.get(event.target_store)`），且
|
||
#: `graph_projection_worker` / `projection_reconciliation_service` 的领取与重放
|
||
#: 都以 `status in {"pending","failed"}` + `"processed"` 为准。全仓（包含投顾线两处生产者）
|
||
#: 统一使用小写 `milvus`/`neo4j`、`upsert`、`pending`。
|
||
#:
|
||
#: 历史说明：`docs/00` §6.4.6 该栏曾写作大写 `MILVUS`/`NEO4J`、`UPSERT` 与中文 `待处理`，
|
||
#: 与上述实现从未对齐；本模块原先照文档写，是**全仓唯一的异类**，导致自己写的事件
|
||
#: 任何消费者都领不到。现统一为小写，并已把历史 2 行就地改齐（主键/唯一键不变）。
|
||
TARGET_MILVUS = "milvus"
|
||
TARGET_NEO4J = "neo4j"
|
||
SYNC_TARGETS: tuple[str, ...] = (TARGET_MILVUS, TARGET_NEO4J)
|
||
|
||
#: 同步操作与状态取值。
|
||
SYNC_OPERATION_UPSERT = "upsert"
|
||
SYNC_STATUS_PENDING = "pending"
|
||
AGGREGATE_TYPE_PROFILE = "profile"
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ProfileGenerationResult:
|
||
"""生成结果；`changed=False` 表示快照未变化、**未写任何行**。"""
|
||
|
||
customer_id: int
|
||
version: int
|
||
profile_uuid: str
|
||
snapshot_hash: str
|
||
changed: bool
|
||
sync_events: int
|
||
|
||
|
||
def build_snapshot(
|
||
profile_row: Mapping[str, Any],
|
||
assessment_row: Mapping[str, Any] | None,
|
||
*,
|
||
now: datetime | None = None,
|
||
) -> dict[str, object]:
|
||
"""把权威事实投影成画像快照（字段与读取侧白名单对齐）。
|
||
|
||
- `assessment_valid_until` / `assessment_expired` **由测评行实时推出**，不沿用旧快照的布尔值；
|
||
没有测评行时两者都**不写入**（读取侧据此表现为"无测评信息"，而不是伪造一个有效期）。
|
||
- 时间字段统一用 ISO 字符串，保证 JSON 稳定、`snapshot_hash` 可复现。
|
||
"""
|
||
current = now or datetime.now(UTC)
|
||
snapshot: dict[str, object] = {
|
||
"investor_type": profile_row["investor_type"],
|
||
"investment_horizon": profile_row["investment_horizon"],
|
||
"trading_frequency": profile_row["trading_frequency"],
|
||
"preferred_asset_class": _as_list(profile_row["preferred_asset_class"]),
|
||
"risk_tags": _as_list(profile_row["risk_tags"]),
|
||
"behavior_score": profile_row["behavior_score"],
|
||
"total_asset": _as_money(profile_row["total_asset"]),
|
||
}
|
||
if profile_row["last_active_at"] is not None:
|
||
snapshot["last_active_at"] = _iso(profile_row["last_active_at"])
|
||
|
||
if assessment_row is not None and assessment_row["valid_until"] is not None:
|
||
valid_until = _instant(assessment_row["valid_until"])
|
||
snapshot["assessment_valid_until"] = valid_until.isoformat()
|
||
snapshot["assessment_expired"] = valid_until <= current
|
||
return snapshot
|
||
|
||
|
||
def compute_hash(snapshot: dict[str, object]) -> str:
|
||
"""`snapshot_hash`：对**键排序后**的规范化 JSON 求 SHA-256。
|
||
|
||
排序是为了让"同一内容"总是得到同一哈希 —— 它是幂等判据，不能被字典插入顺序影响。
|
||
"""
|
||
canonical = json.dumps(snapshot, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
||
return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
|
||
|
||
|
||
class ProfileGenerationService:
|
||
"""生成客户画像版本；**不 commit**，事务归调用方。"""
|
||
|
||
def __init__(self, session: AsyncSession) -> None:
|
||
self._session = session
|
||
self._repo = ProfileRepository(session)
|
||
|
||
async def generate(
|
||
self, customer_id: int, *, now: datetime | None = None
|
||
) -> ProfileGenerationResult:
|
||
current = now or datetime.now(UTC).replace(tzinfo=None)
|
||
|
||
profile_row = await self._repo.profile_row(customer_id)
|
||
if profile_row is None:
|
||
# 失败关闭：没有客户主表行就不该凭空造画像。
|
||
raise ValidationAgentError(f"客户不存在或缺少画像主表行：{customer_id}")
|
||
assessment_row = await self._repo.latest_assessment(customer_id)
|
||
|
||
snapshot = build_snapshot(profile_row, assessment_row, now=current.replace(tzinfo=UTC))
|
||
snapshot_hash = compute_hash(snapshot)
|
||
|
||
existing = await self._repo.current_snapshot(customer_id)
|
||
if existing is not None and str(existing["snapshot_hash"] or "") == snapshot_hash:
|
||
# 幂等：内容没变就不升版本、不投事件。
|
||
return ProfileGenerationResult(
|
||
customer_id=customer_id,
|
||
version=int(existing["version"]),
|
||
profile_uuid=str(existing["profile_uuid"] or ""),
|
||
snapshot_hash=snapshot_hash,
|
||
changed=False,
|
||
sync_events=0,
|
||
)
|
||
|
||
# `profile_uuid` 上有**唯一键**（跨版本也唯一），所以每个版本必须用新 uuid ——
|
||
# 不能沿用以表示"同一实体"（实测会撞 `Duplicate entry ... for key profile_uuid`）。
|
||
# 待接入投影消费者时，若 Milvus/Neo4j 需要**跨版本稳定的实体标识**，
|
||
# 应另加一列（例如 `entity_uuid`）区分"版本标识"与"实体标识"，不要复用本列。
|
||
profile_uuid = str(uuid4())
|
||
version = await self._repo.next_version(customer_id)
|
||
|
||
# 顺序要紧：先清旧当前标记（唯一键约束），再插新版本。
|
||
await self._repo.clear_current(customer_id, now=current)
|
||
await self._repo.insert_snapshot(
|
||
customer_id=customer_id,
|
||
version=version,
|
||
profile_uuid=profile_uuid,
|
||
snapshot_json=json.dumps(snapshot, ensure_ascii=False, sort_keys=True),
|
||
generation_basis_json=json.dumps(
|
||
self._generation_basis(assessment_row), ensure_ascii=False, sort_keys=True
|
||
),
|
||
snapshot_hash=snapshot_hash,
|
||
now=current,
|
||
)
|
||
|
||
# 两条同步事件共用同一个 event_uuid（唯一键是 (event_uuid, target_store)）。
|
||
event_uuid = str(uuid4())
|
||
payload = {
|
||
"aggregate_uuid": profile_uuid,
|
||
"customer_id": str(customer_id),
|
||
"version": version,
|
||
# 别名：投影适配器契约用 `profile_version`。两个键都写，避免消费端
|
||
# 因生产者用词不同而取不到值（本仓三种 payload 形状的历史遗留）。
|
||
"profile_version": version,
|
||
"snapshot_hash": snapshot_hash,
|
||
"snapshot": snapshot,
|
||
"memory_sources": await self._memory_sources(customer_id),
|
||
}
|
||
for target in SYNC_TARGETS:
|
||
self._repo.add_sync_event(
|
||
event_uuid=event_uuid,
|
||
target_store=target,
|
||
aggregate_uuid=profile_uuid,
|
||
aggregate_version=version,
|
||
payload=payload,
|
||
now=current,
|
||
status=SYNC_STATUS_PENDING,
|
||
operation=SYNC_OPERATION_UPSERT,
|
||
)
|
||
|
||
return ProfileGenerationResult(
|
||
customer_id=customer_id,
|
||
version=version,
|
||
profile_uuid=profile_uuid,
|
||
snapshot_hash=snapshot_hash,
|
||
changed=True,
|
||
sync_events=len(SYNC_TARGETS),
|
||
)
|
||
|
||
async def _memory_sources(self, customer_id: int) -> list[dict[str, Any]]:
|
||
"""组装投影适配器要的 `memory_sources`（数据源：`memory_unit` 中 `status='active'`）。
|
||
|
||
为什么由生产端组装而不是消费端回查：消费端到时那条记忆可能已改版本，
|
||
让它自己去查会把"投递的是哪一版"变成不确定。事件里带上当时的确定快照，
|
||
投递语义才与 `aggregate_version` 一致（幂等判据也才有意义）。
|
||
|
||
为什么是 `memory_unit` 而不是 `user_facts`：两者用途不同——`user_facts` 是
|
||
**已确认的结构化事实**，喂画像与图投影；`memory_unit` 是**长期记忆条目**，
|
||
正是长期记忆向量集合要存的东西。这不与"图投影只读 user_facts"的不变式冲突，
|
||
因为 Milvus 存的就是记忆本身的向量，不是画像事实的另一种说法。
|
||
"""
|
||
rows = await self._repo.active_memories(customer_id)
|
||
return [
|
||
{
|
||
"memory_uuid": str(row["memory_uuid"]),
|
||
"memory_key": str(row["memory_key"]),
|
||
"content": str(row["content"]),
|
||
"memory_type": str(row["memory_type"]),
|
||
"confidence": float(row["confidence"]),
|
||
"version": int(row["version"]),
|
||
"valid_until": (
|
||
row["valid_until"].isoformat() if row["valid_until"] else None
|
||
),
|
||
}
|
||
for row in rows
|
||
]
|
||
|
||
@staticmethod
|
||
def _generation_basis(assessment_row: Mapping[str, Any] | None) -> dict[str, object]:
|
||
"""记录这次画像**依据了什么**（`docs/00`：使用的测评版本、交易窗口和记忆版本列表）。
|
||
|
||
当前只接入了测评版本；记忆与交易窗口待接入时补键，不臆造。
|
||
"""
|
||
basis: dict[str, object] = {"sources": ["fin_customer_profile", "fin_risk_assessment"]}
|
||
if assessment_row is not None:
|
||
basis["questionnaire_version"] = assessment_row["questionnaire_version"]
|
||
basis["assessment_id"] = int(assessment_row["id"])
|
||
return basis
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 取值归一化
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _as_list(value: object) -> list[object]:
|
||
if value is None:
|
||
return []
|
||
if isinstance(value, (list, tuple)):
|
||
return list(value)
|
||
if isinstance(value, str):
|
||
try:
|
||
parsed = json.loads(value)
|
||
except ValueError:
|
||
return [value]
|
||
return list(parsed) if isinstance(parsed, list) else [value]
|
||
return [value]
|
||
|
||
|
||
def _as_money(value: object) -> str:
|
||
"""金额统一成字符串（与读取侧 `public()` 的口径一致，避免浮点误差）。"""
|
||
return "" if value is None else str(value)
|
||
|
||
|
||
def _instant(value: object) -> datetime:
|
||
if isinstance(value, datetime):
|
||
return value if value.tzinfo is not None else value.replace(tzinfo=UTC)
|
||
return datetime.fromisoformat(str(value)).replace(tzinfo=UTC)
|
||
|
||
|
||
def _iso(value: object) -> str:
|
||
return _instant(value).isoformat()
|
||
|
||
|
||
#: 导出以便测试断言"写侧字段覆盖了读取侧白名单所需项"。
|
||
REQUIRED_SNAPSHOT_FIELDS: tuple[str, ...] = tuple(
|
||
field for field in PROFILE_FIELD_POLICY if field != "customer_tier"
|
||
)
|