张胜宇
9675df8453
- 分支内容对齐 qyqy_develop b6ec3aa，树完全一致（同步后 git diff 为空） - 覆盖本轮全部交付：客服 Agent 重构（安全路由 / 五出口 / 记忆与画像 / RAG 全链路） + 开发文档 62 份编号体系（D1.1 v1.17 索引） + 新增 D2.10-客服Agent端到端答辩文档-2026-09-21.html - 基线：e239eb7（2026-09-17 品牌口径统一快照），本提交为其直接后继
183 lines
7.5 KiB
Python
183 lines
7.5 KiB
Python
"""画像生成的数据访问层（Repository）。
|
||
|
||
按 MVC+S 边界：**Service 不得直接建 Session / 直接查 Model**，所有库交互收口在这里。
|
||
|
||
只做两件事：
|
||
1. **只读**组装画像所需的权威事实（`docs/00` §6.4.6：画像生成只允许读取**已提交的**业务事实
|
||
和状态为**有效**的中期记忆）；
|
||
2. **写入**画像版本与两条跨存储同步事件（同事务，由调用方管理事务）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import datetime
|
||
from typing import Any
|
||
|
||
from sqlalchemy import text
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.model.memory import MemorySyncOutbox
|
||
|
||
#: 画像快照：查询客户主表 + 客户分层。
|
||
#:
|
||
#: **为什么必须 JOIN `sys_user`**：`customer_tier`（客户分层：金卡 / 白金 / 钻石 / 尊享）的
|
||
#: 权威列在 `sys_user` 上，`fin_customer_profile` **没有**这一列。而 `D3.1` §3.5 与 `DEC-19`
|
||
#: 裁定 (a) 都把「读 `customer_level`」列为客服侧的**受准只读字段**（供确定性规则与画像问答），
|
||
#: `PROFILE_FIELD_POLICY` / `customer_profile_service.ALLOWED_FIELDS` 也早已把它写进白名单 ——
|
||
#: 唯独没有取数来源，于是它在"白名单里躺了很久却永远是空"。这里补上取数，字段才真正可达。
|
||
#:
|
||
#: 用 `LEFT JOIN`（不是 `INNER JOIN`）：客户可能没有 `sys_user` 行（历史数据），
|
||
#: 此时画像仍应生成，只是分层为空 —— 不能因为缺一行账号把整张画像判成"不存在"。
|
||
_SQL_PROFILE = text("""
|
||
SELECT p.customer_id, p.investor_type, p.investment_horizon, p.preferred_asset_class,
|
||
p.trading_frequency, p.total_asset, p.behavior_score, p.risk_tags, p.last_active_at,
|
||
u.customer_tier
|
||
FROM fin_customer_profile AS p
|
||
LEFT JOIN sys_user AS u ON u.id = p.customer_id
|
||
WHERE p.customer_id = :customer_id
|
||
""")
|
||
|
||
#: 风险测评：取**最新一条**，含有效期（用于判定是否过期）。
|
||
_SQL_ASSESSMENT = text("""
|
||
SELECT id, questionnaire_version, investor_type, total_score, assessed_at, valid_until
|
||
FROM fin_risk_assessment
|
||
WHERE customer_id = :customer_id
|
||
ORDER BY assessed_at DESC, id DESC
|
||
LIMIT 1
|
||
""")
|
||
|
||
#: 当前画像版本（取 `is_current=1` 那一条）。
|
||
_SQL_CURRENT_SNAPSHOT = text("""
|
||
SELECT id, profile_uuid, version, snapshot, snapshot_hash, generation_basis
|
||
FROM profile_snapshots
|
||
WHERE customer_id = :customer_id AND is_current = 1
|
||
LIMIT 1
|
||
""")
|
||
|
||
#: 下一版本号 = 当前最大版本 + 1
|
||
#: （`uk_profile_snapshot_version (customer_id, version)` 要求版本号按客户唯一）。
|
||
_SQL_NEXT_VERSION = text("""
|
||
SELECT COALESCE(MAX(version), 0) + 1
|
||
FROM profile_snapshots WHERE customer_id = :customer_id
|
||
""")
|
||
|
||
#: 该客户**有效**的长期记忆（投影到 Milvus 长期记忆向量集合的数据源）。
|
||
#:
|
||
#: 只取 `status='active'`：失效/被取代的记忆不应再进入向量召回，否则投顾会召回
|
||
#: 已过期偏好。字段与 `MilvusProfileProjection` 的 `memory_sources` 契约一一对应。
|
||
_SQL_ACTIVE_MEMORIES = text("""
|
||
SELECT memory_uuid, memory_key, content, memory_type, confidence, version, valid_until
|
||
FROM memory_unit
|
||
WHERE customer_id = :customer_id AND status = 'active'
|
||
ORDER BY id
|
||
""")
|
||
|
||
#: 旧画像置非当前（`current_customer_id` 有唯一键，必须先置 0 才能插新的当前版本）。
|
||
_SQL_CLEAR_CURRENT = text("""
|
||
UPDATE profile_snapshots
|
||
SET is_current = 0, current_customer_id = NULL, updated_at = :now
|
||
WHERE customer_id = :customer_id AND is_current = 1
|
||
""")
|
||
|
||
_SQL_INSERT_SNAPSHOT = text("""
|
||
INSERT INTO profile_snapshots
|
||
(profile_uuid, customer_id, version, snapshot, generation_basis, snapshot_hash,
|
||
is_current, current_customer_id, generated_at, created_at, updated_at)
|
||
VALUES
|
||
(:profile_uuid, :customer_id, :version, :snapshot, :generation_basis, :snapshot_hash,
|
||
1, :customer_id, :now, :now, :now)
|
||
""")
|
||
|
||
|
||
class ProfileRepository:
|
||
def __init__(self, session: AsyncSession) -> None:
|
||
self._session = session
|
||
|
||
# --- 只读事实 ---------------------------------------------------------
|
||
|
||
async def profile_row(self, customer_id: int) -> dict[str, Any] | None:
|
||
result = await self._session.execute(_SQL_PROFILE, {"customer_id": customer_id})
|
||
row = result.mappings().first()
|
||
return dict(row) if row is not None else None
|
||
|
||
async def latest_assessment(self, customer_id: int) -> dict[str, Any] | None:
|
||
result = await self._session.execute(_SQL_ASSESSMENT, {"customer_id": customer_id})
|
||
row = result.mappings().first()
|
||
return dict(row) if row is not None else None
|
||
|
||
async def current_snapshot(self, customer_id: int) -> dict[str, Any] | None:
|
||
result = await self._session.execute(_SQL_CURRENT_SNAPSHOT, {"customer_id": customer_id})
|
||
row = result.mappings().first()
|
||
return dict(row) if row is not None else None
|
||
|
||
async def active_memories(self, customer_id: int) -> list[dict[str, Any]]:
|
||
"""该客户所有 `status='active'` 的长期记忆，按 id 稳定排序。"""
|
||
result = await self._session.execute(
|
||
_SQL_ACTIVE_MEMORIES, {"customer_id": customer_id}
|
||
)
|
||
return [dict(row) for row in result.mappings().all()]
|
||
|
||
async def next_version(self, customer_id: int) -> int:
|
||
return int(await self._session.scalar(_SQL_NEXT_VERSION, {"customer_id": customer_id}))
|
||
|
||
# --- 写入（调用方管事务） ---------------------------------------------
|
||
|
||
async def clear_current(self, customer_id: int, *, now: datetime) -> None:
|
||
"""把旧画像的 `is_current` 置 0 —— 必须在插入新版本**之前**调用。
|
||
|
||
`uk_profile_snapshot_current` 建在 `current_customer_id` 上（唯一），
|
||
所以同一客户同时只能有一条 `is_current=1`；顺序颠倒会撞唯一键。
|
||
"""
|
||
await self._session.execute(_SQL_CLEAR_CURRENT, {"customer_id": customer_id, "now": now})
|
||
|
||
async def insert_snapshot(
|
||
self,
|
||
*,
|
||
customer_id: int,
|
||
version: int,
|
||
profile_uuid: str,
|
||
snapshot_json: str,
|
||
generation_basis_json: str,
|
||
snapshot_hash: str,
|
||
now: datetime,
|
||
) -> None:
|
||
await self._session.execute(_SQL_INSERT_SNAPSHOT, {
|
||
"profile_uuid": profile_uuid,
|
||
"customer_id": customer_id,
|
||
"version": version,
|
||
"snapshot": snapshot_json,
|
||
"generation_basis": generation_basis_json,
|
||
"snapshot_hash": snapshot_hash,
|
||
"now": now,
|
||
})
|
||
|
||
def add_sync_event(
|
||
self,
|
||
*,
|
||
event_uuid: str,
|
||
target_store: str,
|
||
aggregate_uuid: str,
|
||
aggregate_version: int,
|
||
payload: dict[str, Any],
|
||
now: datetime,
|
||
status: str,
|
||
operation: str,
|
||
) -> None:
|
||
"""追加一条跨存储同步事件。
|
||
|
||
唯一键 `uk_memory_sync_event (event_uuid, target_store)`：**同一 `event_uuid` 允许
|
||
对两个目标库各写一条**，因此 Milvus 与 Neo4j 共用同一个 `event_uuid`。
|
||
"""
|
||
self._session.add(MemorySyncOutbox(
|
||
event_uuid=event_uuid,
|
||
aggregate_type="profile",
|
||
aggregate_uuid=aggregate_uuid,
|
||
aggregate_version=aggregate_version,
|
||
target_store=target_store,
|
||
operation=operation,
|
||
payload=payload,
|
||
status=status,
|
||
retry_count=0,
|
||
created_at=now,
|
||
))
|