lzf_0626
9490e5043b
## 新增：docs/演示用/记忆系统演示文档-2026-09-14.md 按"操作 → 看到什么 → 这体现什么"写，五个场景（探针看存储 / 记住一件事 / 谁能读谁读不到 / 画像版本与投影 / 停 Worker），每个场景配可复现命令与**实测输出**，另附建议顺序与时长、 六条问答话术、演示前自检清单、受控信号词表摘录。 配套两个新工具（都已实跑）： - `tools/memory_demo_chain.py`：一键走完"客户说 → 候选 → 客户确认 → 管理员批准 → 记忆与画像自动收敛"，打印演示前后对比（实测 2 秒收敛）。 - `tools/memory_recall_demo.py`：同一客户换六个身份召回，当场看出"客户只读自己 / 员工要权限码+归属 / 运营读不到 / 管理员有权限码但没归属也读不到"。 ## 修掉两处会让演示当场断链的问题（都是真机复现的） 1. **候选批准后画像字段不收敛** `CustomerProfileCandidateService._promote` 只写画像快照、**不投** `profile.rebuild_requested`，而 `user_facts` 与画像字段是 `ProfileAssemblyService` 的重建链路写的。后果：批准后记忆变 active、快照版本 +1，但**画像字段停在旧值** （客户说"三年以内"，画像里还是"十年以上"），要等一次无关的重建才收敛 —— 而"客户说完 → 批准 → 画像变了"正是演示主线。 现补一条重建事件（走事件而不是就地重建：本方法所在事务还没提交， 另开 session 看不到刚写入的记忆）。 2. **画像快照的唯一键从来没起作用，而且埋雷** `uk_profile_snapshot_current` 建在列 `current_customer_id` 上（不是 `is_current`）， 但 `ProfileAssemblyService._write_snapshot` 旧行只置 `is_current=False`（不清该列）、 新行**不写**该列（实测 13 行该列全 NULL）。 后果：只要客户**先被重建过一次**，旧 current 行仍占着 `current_customer_id=9001`， 下一次"批准画像候选"就会撞 `Duplicate entry '9001' for key uk_profile_snapshot_current` → **整次批准 500**（本次实测踩到）。 现按唯一键的真实语义写：清旧行的该列、新行显式写客户号。 （`ProfileGenerationService` / 候选路径本来就是这么写的，只有这一处没对齐。） ## 守卫 新增 `tests/integration/test_profile_snapshot_current_invariant_mysql.py`： 按真实顺序"先重建再批准候选"，断言 ① 只有一条 current 且它占着唯一键、 ② 历史版本已归还该列、③ 批准不再 500、④ 批准投出了重建事件。 修复前这条用例会在第 ② 步失败。 ## 验证 - `pytest tests/unit tests/contract` → 1490 passed, 2 skipped, 0 failed - 新增集成用例通过；真机实测演示主线：候选 → verified → active → **2 秒内** `user_facts` 与 `fin_customer_profile.investment_horizon` 都变成新值 - `mypy tools/memory_demo_chain.py tools/memory_recall_demo.py` → 0 错；ruff 全绿 ## 文档 `docs/44-演示流程.md`：配套文档清单与"记忆链路"备选场景都指向新演示文档。
299 lines
14 KiB
Python
299 lines
14 KiB
Python
"""客户画像候选的确认、审核与晋升服务。
|
||
|
||
候选与正式记忆共用 ``memory_unit``，但状态转换必须经过本服务；客服 Agent 不具备
|
||
调用权限。用户确认只把候选标记为 ``verified``，管理员批准后才切换为 ``active``。
|
||
"""
|
||
|
||
import hashlib
|
||
import json
|
||
from datetime import UTC, datetime
|
||
from typing import Any, Literal
|
||
from uuid import uuid4
|
||
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.core.contracts import RequestContext
|
||
from app.core.errors import GenericResourceNotFoundError, InvalidStateError
|
||
from app.infrastructure.db import SessionFactory
|
||
from app.model.audit import InteractionAudit
|
||
from app.model.memory import MemoryConflict, MemorySyncOutbox, MemoryUnit
|
||
from app.model.platform import DomainEventOutbox
|
||
from app.model.profile import ProfileSnapshot
|
||
from app.service.agent.bootstrap import get_memory_cache_adapter
|
||
from app.service.authorization_service import AuthorizationService
|
||
from app.service.memory_service import MemoryService
|
||
|
||
CandidateDecision = Literal["confirmed", "rejected"]
|
||
ReviewDecision = Literal["approved", "rejected"]
|
||
USER_CONFIRM_PERMISSION = "memory:candidate:confirm"
|
||
ADMIN_REVIEW_PERMISSION = "memory:candidate:review"
|
||
|
||
|
||
class CustomerProfileCandidateService:
|
||
"""候选状态机的唯一应用服务入口。"""
|
||
|
||
async def list_for_customer(
|
||
self, context: RequestContext, *, limit: int = 20
|
||
) -> dict[str, Any]:
|
||
"""返回当前登录用户自己的候选，不暴露证据原文或其他客户数据。"""
|
||
await AuthorizationService.require(context, "memory:read:self")
|
||
customer_id = int(context.user_id)
|
||
async with SessionFactory() as session:
|
||
rows = await session.scalars(
|
||
select(MemoryUnit)
|
||
.where(
|
||
MemoryUnit.customer_id == customer_id,
|
||
MemoryUnit.status.in_(("candidate", "verified")),
|
||
)
|
||
.order_by(MemoryUnit.updated_at.desc())
|
||
.limit(max(1, min(limit, 100)))
|
||
)
|
||
data = [self._view(item) for item in rows]
|
||
return {"data": data, "meta": {"trace_id": context.trace_id}}
|
||
|
||
async def list_for_admin(
|
||
self, context: RequestContext, *, limit: int = 20
|
||
) -> dict[str, Any]:
|
||
"""管理员查看所有待处理候选；列表仍不返回证据原文。"""
|
||
await AuthorizationService.require(context, ADMIN_REVIEW_PERMISSION, admin=True)
|
||
async with SessionFactory() as session:
|
||
rows = await session.scalars(
|
||
select(MemoryUnit)
|
||
.where(MemoryUnit.status.in_(("candidate", "verified")))
|
||
.order_by(MemoryUnit.updated_at.asc())
|
||
.limit(max(1, min(limit, 100)))
|
||
)
|
||
data = [self._view(item) for item in rows]
|
||
return {"data": data, "meta": {"trace_id": context.trace_id}}
|
||
|
||
async def decide_by_customer(
|
||
self, candidate_id: int, decision: CandidateDecision, context: RequestContext
|
||
) -> dict[str, Any]:
|
||
"""用户确认或拒绝自己的候选；确认不会直接激活正式记忆。"""
|
||
await AuthorizationService.require(context, USER_CONFIRM_PERMISSION)
|
||
customer_id = int(context.user_id)
|
||
target_status = "verified" if decision == "confirmed" else "rejected"
|
||
async with SessionFactory() as session, session.begin():
|
||
candidate = await self._locked_candidate(session, candidate_id, customer_id)
|
||
if candidate.status != "candidate":
|
||
raise InvalidStateError("候选已处理，不能重复确认")
|
||
candidate.status = target_status
|
||
candidate.updated_at = self._now()
|
||
session.add(self._audit(
|
||
context, customer_id, "memory.candidate_user_decision",
|
||
{"candidate_id": candidate_id, "decision": decision},
|
||
))
|
||
await session.flush()
|
||
return {"data": self._view(candidate), "meta": {"trace_id": context.trace_id}}
|
||
|
||
async def review_by_admin(
|
||
self, candidate_id: int, decision: ReviewDecision, context: RequestContext,
|
||
*, comment: str = "",
|
||
) -> dict[str, Any]:
|
||
"""管理员审核候选；批准时处理同键正式记忆冲突并激活候选。"""
|
||
await AuthorizationService.require(context, ADMIN_REVIEW_PERMISSION, admin=True)
|
||
async with SessionFactory() as session, session.begin():
|
||
candidate = await self._locked_candidate(session, candidate_id, None)
|
||
if candidate.status not in {"candidate", "verified"}:
|
||
raise InvalidStateError("候选已处理，不能重复审核")
|
||
if decision == "rejected":
|
||
candidate.status = "rejected"
|
||
else:
|
||
await self._promote(session, candidate, int(context.user_id))
|
||
candidate.updated_at = self._now()
|
||
session.add(self._audit(
|
||
context, candidate.customer_id, "memory.candidate_admin_review",
|
||
{"candidate_id": candidate_id, "decision": decision, "comment": comment[:1000]},
|
||
))
|
||
await session.flush()
|
||
return {"data": self._view(candidate), "meta": {"trace_id": context.trace_id}}
|
||
|
||
async def _locked_candidate(
|
||
self, session: AsyncSession, candidate_id: int, customer_id: int | None
|
||
) -> MemoryUnit:
|
||
"""按身份范围加锁读取候选，找不到时统一隐藏资源存在性。"""
|
||
conditions = [MemoryUnit.id == candidate_id]
|
||
if customer_id is not None:
|
||
conditions.append(MemoryUnit.customer_id == customer_id)
|
||
candidate = await session.scalar(select(MemoryUnit).where(*conditions).with_for_update())
|
||
if candidate is None:
|
||
raise GenericResourceNotFoundError("候选不存在")
|
||
return candidate
|
||
|
||
async def _promote(
|
||
self, session: AsyncSession, candidate: MemoryUnit, reviewer_id: int
|
||
) -> None:
|
||
"""同一客户同一键只保留一条 active，旧值失效并留下冲突审计记录。"""
|
||
current = await session.scalar(
|
||
select(MemoryUnit)
|
||
.where(
|
||
MemoryUnit.customer_id == candidate.customer_id,
|
||
MemoryUnit.memory_key == candidate.memory_key,
|
||
MemoryUnit.status == "active",
|
||
)
|
||
.with_for_update()
|
||
)
|
||
now = self._now()
|
||
if current is not None and current.id != candidate.id:
|
||
current.status = "invalidated"
|
||
current.updated_at = now
|
||
session.add(MemoryConflict(
|
||
left_memory_id=current.id, right_memory_id=candidate.id,
|
||
conflict_type="candidate_promoted", severity="medium", status="resolved",
|
||
resolution="管理员审核候选后替换旧正式记忆", winner_memory_id=candidate.id,
|
||
resolved_by=reviewer_id, resolved_at=now, created_at=now,
|
||
))
|
||
candidate.status = "active"
|
||
candidate.promoted_at = now
|
||
candidate.version += 1
|
||
await self._write_profile_snapshot(session, candidate, reviewer_id, now)
|
||
# 追加一条**画像重建**事件：`_write_profile_snapshot` 只写"画像快照"，
|
||
# 而 `user_facts`（记忆 → 画像字段的中间层）与 `fin_customer_profile.risk_tags`
|
||
# 是由 `ProfileAssemblyService.rebuild()` 里的 `promote_facts` 写的。
|
||
#
|
||
# 不投这条事件的后果（2026-09-14 实测）：批准候选后 `memory_unit` 立刻变 active、
|
||
# 快照版本 +1、投影也投出去了，但**画像字段仍停在旧值**（客户说"稳健型"，
|
||
# 画像里还是"进取型"），要等一次无关的重建（或手动 `tools/rebuild_profile.py`）
|
||
# 才收敛 —— 演示"说完 → 批准 → 画像变了"会当场翻车。
|
||
#
|
||
# 走事件而不是在这里直接重建：本方法所在事务**还没提交**，
|
||
# 另开 session 去重建看不到刚写下的记忆（这是 `runtime.dispatch_profile_rebuild`
|
||
# 注释里记录的同一个坑）。事件只可能在提交之后被消费。
|
||
session.add(DomainEventOutbox(
|
||
# `id=0`：该 ORM 类没声明 autoincrement，不显式给主键会让 INSERT 报错
|
||
# （MySQL 把 0 当作"取下一个自增值"）。与 `episode_worker` 的写法一致。
|
||
id=0,
|
||
event_id=str(uuid4()),
|
||
event_type="profile.rebuild_requested",
|
||
aggregate_type="customer_profile",
|
||
aggregate_id=str(candidate.customer_id),
|
||
trace_id=str(candidate.memory_uuid),
|
||
payload={"customer_id": candidate.customer_id, "trigger": "candidate_promoted"},
|
||
status="pending",
|
||
retry_count=0,
|
||
occurred_at=now,
|
||
created_at=now,
|
||
updated_at=now,
|
||
))
|
||
await MemoryService(session, cache=get_memory_cache_adapter()).invalidate_recall_cache(
|
||
int(candidate.customer_id)
|
||
)
|
||
|
||
async def _write_profile_snapshot(
|
||
self, session: AsyncSession, candidate: MemoryUnit, reviewer_id: int, now: datetime
|
||
) -> None:
|
||
"""生成新的当前画像版本，并为两个派生存储写入可靠同步事件。"""
|
||
current = await session.scalar(
|
||
select(ProfileSnapshot)
|
||
.where(
|
||
ProfileSnapshot.customer_id == candidate.customer_id,
|
||
ProfileSnapshot.is_current.is_(True),
|
||
)
|
||
.with_for_update()
|
||
)
|
||
previous = dict(current.snapshot) if current is not None else {}
|
||
preferences = dict(previous.get("customer_service_preferences", {}))
|
||
preferences[candidate.memory_key] = {
|
||
"value": candidate.content,
|
||
"memory_type": candidate.memory_type,
|
||
"confidence": float(candidate.confidence),
|
||
}
|
||
snapshot = {**previous, "customer_service_preferences": preferences}
|
||
version = (int(current.version) + 1) if current is not None else 1
|
||
profile_uuid = str(uuid4())
|
||
snapshot_hash = hashlib.sha256(
|
||
json.dumps(snapshot, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
||
.encode("utf-8")
|
||
).hexdigest()
|
||
if current is not None:
|
||
current.is_current = False
|
||
# 必须**同时清空** `current_customer_id`：唯一键 `uk_profile_snapshot_current`
|
||
# 建在这一列上（不是 `is_current`），旧当前版本不清就会与新版本撞键。
|
||
# 与 `ProfileGenerationService._SQL_CLEAR_CURRENT` 的做法一致。
|
||
current.current_customer_id = None
|
||
current.updated_at = now
|
||
created = ProfileSnapshot(
|
||
profile_uuid=profile_uuid, customer_id=candidate.customer_id,
|
||
version=version, snapshot=snapshot,
|
||
generation_basis={
|
||
"source": "customer_profile_candidate",
|
||
"candidate_id": candidate.id,
|
||
"reviewer_id": reviewer_id,
|
||
},
|
||
snapshot_hash=snapshot_hash, is_current=True,
|
||
# 当前版本必须**显式写入**客户 ID（历史版本为 NULL），见 `app/model/profile.py`
|
||
# 的模块 docstring 第 2 条：该列不是生成列，不显式写就形同虚设，
|
||
# 「每个客户最多一条当前快照」这条不变式会失效。
|
||
current_customer_id=candidate.customer_id,
|
||
generated_at=now,
|
||
created_at=now, updated_at=now,
|
||
)
|
||
session.add(created)
|
||
await session.flush()
|
||
active_memories = list(await session.scalars(
|
||
select(MemoryUnit).where(
|
||
MemoryUnit.customer_id == candidate.customer_id,
|
||
MemoryUnit.status == "active",
|
||
)
|
||
))
|
||
memory_sources = [
|
||
{
|
||
"memory_uuid": item.memory_uuid,
|
||
"memory_key": item.memory_key,
|
||
"content": item.content,
|
||
"memory_type": item.memory_type,
|
||
"confidence": float(item.confidence),
|
||
"version": int(item.version),
|
||
"valid_until": item.valid_until.isoformat() if item.valid_until else None,
|
||
}
|
||
for item in active_memories
|
||
]
|
||
for target_store in ("milvus", "neo4j"):
|
||
session.add(MemorySyncOutbox(
|
||
event_uuid=str(uuid4()), aggregate_type="profile_snapshot",
|
||
aggregate_uuid=profile_uuid, aggregate_version=version,
|
||
target_store=target_store, operation="upsert",
|
||
payload={
|
||
"customer_id": candidate.customer_id,
|
||
"profile_uuid": profile_uuid,
|
||
"profile_version": version,
|
||
"snapshot": snapshot,
|
||
"memory_sources": memory_sources,
|
||
},
|
||
status="pending", retry_count=0, created_at=now,
|
||
))
|
||
|
||
@staticmethod
|
||
def _view(item: MemoryUnit) -> dict[str, Any]:
|
||
"""只返回结构化候选值，不返回对话证据摘录。"""
|
||
return {
|
||
"candidate_id": int(item.id),
|
||
"customer_id": str(item.customer_id),
|
||
"memory_key": item.memory_key,
|
||
"value": item.content,
|
||
"memory_type": item.memory_type,
|
||
"confidence": float(item.confidence),
|
||
"status": item.status,
|
||
"version": item.version,
|
||
"created_at": item.created_at.isoformat(),
|
||
"updated_at": item.updated_at.isoformat(),
|
||
}
|
||
|
||
@staticmethod
|
||
def _audit(
|
||
context: RequestContext, customer_id: int, action_type: str, detail: dict[str, Any]
|
||
) -> InteractionAudit:
|
||
"""统一生成候选状态变更审计，不携带证据原文。"""
|
||
return InteractionAudit(
|
||
actor_type="user", actor_id=int(context.user_id), target_customer_id=customer_id,
|
||
session_id=None, portal=context.portal, action_type=action_type,
|
||
detail={**detail, "trace_id": context.trace_id},
|
||
created_at=CustomerProfileCandidateService._now(),
|
||
)
|
||
|
||
@staticmethod
|
||
def _now() -> datetime:
|
||
"""使用 UTC 无时区值，与现有数据库 DATETIME 字段保持一致。"""
|
||
return datetime.now(UTC).replace(tzinfo=None)
|