From 27174b9ce26abdd89c4c20d2bba6e57c27b798e1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BC=A0=E8=83=9C=E5=AE=87?= <17412268+zzzzz11122222@user.noreply.gitee.com> Date: Fri, 11 Sep 2026 17:58:46 +0800 Subject: [PATCH] feat: generate approved profile snapshots --- app/model/memory.py | 19 ++++++ .../customer_profile_candidate_service.py | 64 ++++++++++++++++++- docs/客服Agent二期_客户画像候选流程_v1.md | 11 ++-- ...test_customer_profile_candidate_service.py | 40 +++++++++++- 4 files changed, 128 insertions(+), 6 deletions(-) diff --git a/app/model/memory.py b/app/model/memory.py index 5a7eb95..bd8b353 100644 --- a/app/model/memory.py +++ b/app/model/memory.py @@ -75,6 +75,25 @@ class MemorySyncOutbox(Base): processed_at: Mapped[datetime | None] = mapped_column(DateTime) +class ProfileSnapshot(Base): + """客户画像版本快照；只有审核通过的候选才能生成当前版本。""" + + __tablename__ = "profile_snapshots" + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True) + profile_uuid: Mapped[str | None] = mapped_column(String(36), unique=True) + customer_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + version: Mapped[int] = mapped_column(BigInteger, nullable=False) + snapshot: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False) + generation_basis: Mapped[dict[str, Any] | None] = mapped_column(JSON) + snapshot_hash: Mapped[str | None] = mapped_column(String(64)) + is_current: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + current_customer_id: Mapped[int | None] = mapped_column(BigInteger, unique=True) + generated_at: Mapped[datetime | None] = mapped_column(DateTime) + created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + updated_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + + class MemoryConflict(Base): """记忆冲突；库列名为 `status`，`resolution`/`severity`/`resolved_by` 必须映射。""" diff --git a/app/service/customer_profile_candidate_service.py b/app/service/customer_profile_candidate_service.py index 7555245..bb2758f 100644 --- a/app/service/customer_profile_candidate_service.py +++ b/app/service/customer_profile_candidate_service.py @@ -4,8 +4,11 @@ 调用权限。用户确认只把候选标记为 ``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 @@ -14,7 +17,7 @@ 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, MemoryUnit +from app.model.memory import MemoryConflict, MemorySyncOutbox, MemoryUnit, 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 @@ -142,10 +145,69 @@ class CustomerProfileCandidateService: candidate.status = "active" candidate.promoted_at = now candidate.version += 1 + await self._write_profile_snapshot(session, candidate, reviewer_id, 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.current_customer_id == candidate.customer_id, + ) + .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 = 0 + 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=1, + current_customer_id=candidate.customer_id, generated_at=now, + created_at=now, updated_at=now, + ) + session.add(created) + await session.flush() + 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, + }, + status="pending", retry_count=0, created_at=now, + )) + @staticmethod def _view(item: MemoryUnit) -> dict[str, Any]: """只返回结构化候选值，不返回对话证据摘录。""" diff --git a/docs/客服Agent二期_客户画像候选流程_v1.md b/docs/客服Agent二期_客户画像候选流程_v1.md index 657eff2..9d50056 100644 --- a/docs/客服Agent二期_客户画像候选流程_v1.md +++ b/docs/客服Agent二期_客户画像候选流程_v1.md @@ -2,7 +2,7 @@ 版本：v1.0 适用分支：`ZSY_develop2` -状态：候选提取、用户确认和管理员审核入口已实现，画像快照同步待后续迭代 +状态：候选提取、用户确认、管理员审核、画像快照生成和投影事件写入已实现 ## 1. 业务边界 @@ -25,7 +25,7 @@ -> 用户确认或拒绝 -> 管理员审核 -> 批准后处理同键冲突并晋升为 active - -> 后续流程再生成 profile_snapshots + -> 生成 profile_snapshots 并写入 memory_sync_outbox ``` 普通公开问答、闲聊、一次性操作问题不触发候选抽取；访客、非客服 Agent、非 self 数据范围 @@ -89,10 +89,13 @@ - `app/api/controllers/public_platform.py`：用户候选查询和确认接口。 - `app/api/controllers/admin.py`：管理员候选查询和审核接口。 +管理员批准后，服务会创建新的 `profile_snapshots` 当前版本，并为 `milvus`、`neo4j` 各写入 +一条 `memory_sync_outbox` 待投影事件；旧当前版本在同一事务内标记为非当前。 + ## 7. 尚未实现的后续能力 1. 前端用户确认页面和管理员审核页面。 -2. 候选晋升后的 `profile_snapshots` 版本生成和 `memory_sync_outbox` 同步。 +2. `memory_sync_outbox` 到真实 Milvus/Neo4j 的消费者联调和失败重试验收。 3. 候选撤回、过期、删除和隐私授权管理。 4. 候选流程的 MySQL 集成测试和管理员端到端验收。 @@ -101,6 +104,6 @@ ## 8. 验证结果 - 客服画像候选专项测试：通过。 -- 一期单元与契约回归：`683 passed`。 +- 一期单元与契约回归：`687 passed`。 - Ruff：通过。 - Mypy：通过。 diff --git a/tests/unit/service/test_customer_profile_candidate_service.py b/tests/unit/service/test_customer_profile_candidate_service.py index 7da1b3d..c1c441f 100644 --- a/tests/unit/service/test_customer_profile_candidate_service.py +++ b/tests/unit/service/test_customer_profile_candidate_service.py @@ -7,7 +7,7 @@ import pytest from app.core.contracts import RequestContext from app.core.errors import ForbiddenAgentError, InvalidStateError -from app.model.memory import MemoryUnit +from app.model.memory import MemorySyncOutbox, MemoryUnit, ProfileSnapshot from app.service.customer_profile_candidate_service import ( ADMIN_REVIEW_PERMISSION, USER_CONFIRM_PERMISSION, @@ -142,3 +142,41 @@ async def test_candidate_transition_rejects_repeated_processing( await CustomerProfileCandidateService().decide_by_customer( 11, "confirmed", RequestContext(user_id="7", trace_id="candidate-test") ) + + +@pytest.mark.asyncio +async def test_profile_snapshot_and_projection_events_are_created_on_promotion( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """批准候选必须生成新画像版本和两个待投影事件。""" + class FakeSession: + def __init__(self) -> None: + self.added: list[object] = [] + + async def scalar(self, statement: object) -> None: + del statement + return None + + def add(self, item: object) -> None: + self.added.append(item) + + async def flush(self) -> None: + return None + + session = FakeSession() + target = candidate() + monkeypatch.setattr( + "app.service.customer_profile_candidate_service.get_memory_cache_adapter", + lambda: None, + ) + + await CustomerProfileCandidateService()._promote(session, target, reviewer_id=9003) + + assert target.status == "active" + snapshot = next(item for item in session.added if isinstance(item, ProfileSnapshot)) + assert snapshot.is_current == 1 + assert snapshot.version == 1 + value = snapshot.snapshot["customer_service_preferences"]["preference:risk_level"]["value"] + assert value == "稳健型" + events = [item for item in session.added if isinstance(item, MemorySyncOutbox)] + assert {item.target_store for item in events} == {"milvus", "neo4j"}