From 3a21afa8478ee03332aeb21d5e27c59463c0f8da 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:16:43 +0800 Subject: [PATCH] feat: add customer profile candidate flow --- app/service/agent_persistence_service.py | 15 +++- app/service/memory_service.py | 8 +- .../customer_profile_candidate_worker.py | 50 +++++++++++ app/worker/memory_extraction_worker.py | 18 +++- app/worker/runtime.py | 26 ++++++ docs/客服Agent二期_客户画像候选流程_v1.md | 84 +++++++++++++++++++ .../test_agent_persistence_handover.py | 19 +++++ tests/unit/service/test_memory_service.py | 18 ++++ .../worker/test_memory_extraction_worker.py | 38 +++++++++ .../worker/test_runtime_worker_dispatch.py | 25 ++++++ 10 files changed, 295 insertions(+), 6 deletions(-) create mode 100644 app/worker/customer_profile_candidate_worker.py create mode 100644 docs/客服Agent二期_客户画像候选流程_v1.md diff --git a/app/service/agent_persistence_service.py b/app/service/agent_persistence_service.py index 8a3f704..29ecd09 100644 --- a/app/service/agent_persistence_service.py +++ b/app/service/agent_persistence_service.py @@ -25,7 +25,7 @@ class AgentPersistenceService: async def complete_run( self, run_id: str, result: AgentResult, memory_extraction_requested: bool = True, - *, worker_id: str | None = None, + *, worker_id: str | None = None, profile_candidate_requested: bool = False, ) -> int: now = datetime.now(UTC).replace(tzinfo=None) async with self.session.begin(): @@ -167,6 +167,19 @@ class AgentPersistenceService: payload={"run_id": run_id, "message_id": message.id, "customer_id": run.user_id}, occurred_at=now, )) + if profile_candidate_requested: + # 候选事件只允许由已登录客服会话触发，消费者仍会再次校验身份标记。 + events.append(DomainEvent( + event_id=str(uuid4()), + event_type="customer_profile.candidate_requested", + aggregate_type="agent_run", aggregate_id=run_id, trace_id=run.trace_id, + payload={ + "run_id": run_id, "message_id": message.id, + "customer_id": run.user_id, + "actor_type": "authenticated_customer", + }, + occurred_at=now, + )) if handover_ticket is not None: # Outbox 事件由后续管理员通知/工单消费方可靠投递，服务层不直接通知外部系统。 assert handover_context is not None diff --git a/app/service/memory_service.py b/app/service/memory_service.py index f71f389..366cc7a 100644 --- a/app/service/memory_service.py +++ b/app/service/memory_service.py @@ -118,6 +118,7 @@ class MemoryService: confidence: float = 0.5, source_type: str = "conversation", structured_value: dict[str, Any] | None = None, + status: str = "active", ) -> MemoryUnit: """按 (customer_id, active_memory_key) 语义更新唯一有效记忆。 @@ -125,8 +126,11 @@ class MemoryService: 内容变化时记录一条冲突：左侧为被覆盖的旧值所在记忆行，右侧为该记忆的新版本 标识（见 `_conflict_right_id`）。两侧绝不指向同一条记录，避免自引用冲突。 """ + if status not in {"active", "candidate"}: + raise ValueError("status must be active or candidate") now = datetime.now(UTC).replace(tzinfo=None) - memory = await self._active(customer_id, memory_key) + # 候选记录不能覆盖现有有效记忆，必须等待确认或审核后再晋升。 + memory = await self._active(customer_id, memory_key) if status == "active" else None if memory is not None: updated = await self._update(memory, content, confidence, now, structured_value) await self.invalidate_recall_cache(customer_id) @@ -137,7 +141,7 @@ class MemoryService: source_type=source_type, source_confidence=confidence, confidence=confidence, structured_value=structured_value, evidence_count=0, conflict_count=0, recall_count=0, - status="active", valid_from=now, version=1, created_at=now, updated_at=now, + status=status, valid_from=now, version=1, created_at=now, updated_at=now, ) self.session.add(memory) try: diff --git a/app/worker/customer_profile_candidate_worker.py b/app/worker/customer_profile_candidate_worker.py new file mode 100644 index 0000000..20947fe --- /dev/null +++ b/app/worker/customer_profile_candidate_worker.py @@ -0,0 +1,50 @@ +"""客服对话画像候选消费者。 + +该消费者只接收已登录用户的候选事件，并将脱敏后的模型抽取结果写为 +``memory_unit.status='candidate'``。候选不会进入客服召回，也不会修改正式画像。 +""" + +from datetime import datetime +from typing import Any + +from sqlalchemy.ext.asyncio import AsyncSession + +from app.service.memory_extraction_service import ( + MemoryExtractionService, + get_memory_extraction_service, +) +from app.service.memory_service import CacheDeleteAdapter +from app.worker.memory_extraction_worker import MemoryExtractionWorker + + +class CustomerProfileCandidateWorker: + """消费已登录客服会话的画像候选事件；访客事件失败关闭。""" + + def __init__( + self, + session: AsyncSession, + *, + extractor: MemoryExtractionService | None = None, + cache: CacheDeleteAdapter | None = None, + ) -> None: + self.worker = MemoryExtractionWorker( + session, + extractor=extractor or get_memory_extraction_service(), + cache=cache, + memory_status="candidate", + source_type="AI对话提取", + event_type="customer_profile.candidate_requested", + sanitize_source=True, + ) + + async def handle( + self, + payload: dict[str, Any], + *, + event_id: str | None = None, + occurred_at: datetime | None = None, + ) -> bool: + """只允许受理服务标记的已登录客户事件，防止访客写入画像候选。""" + if payload.get("actor_type") != "authenticated_customer": + return False + return await self.worker.handle(payload, event_id=event_id, occurred_at=occurred_at) diff --git a/app/worker/memory_extraction_worker.py b/app/worker/memory_extraction_worker.py index 921029c..5fd32bd 100644 --- a/app/worker/memory_extraction_worker.py +++ b/app/worker/memory_extraction_worker.py @@ -5,6 +5,7 @@ from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession +from app.core.conversation_privacy import sanitize_customer_service_message from app.model.conversation import ConversationMessage from app.model.memory import MemoryEvidence from app.model.platform import AgentRun, DomainEventOutbox @@ -41,6 +42,10 @@ class MemoryExtractionWorker: *, extractor: MemoryExtractionService | None = None, cache: CacheDeleteAdapter | None = None, + memory_status: str = "active", + source_type: str = SOURCE_TYPE, + event_type: str = "memory.extraction_requested", + sanitize_source: bool = False, ) -> None: self.session = session # 默认走生产装配（与业务 Agent 同一个 ModelGenerationService）；验收探针 @@ -48,6 +53,10 @@ class MemoryExtractionWorker: self.extractor = extractor if extractor is not None else get_memory_extraction_service() # 召回热缓存适配器：写入生效后必须失效，否则新记忆在 TTL 内召回不到。 self.cache = cache + self.memory_status = memory_status + self.source_type = source_type + self.event_type = event_type + self.sanitize_source = sanitize_source async def handle( self, payload: dict[str, Any], *, event_id: str | None = None, @@ -73,7 +82,7 @@ class MemoryExtractionWorker: # 没有可追溯的事件 id 就不能建立幂等边界，重复消费将无法去重。 logger.warning("memory extraction skipped: event id not found run_id=%s", run_id) return False - idempotency_key = f"memory.extraction_requested:{event_id}" + idempotency_key = f"{self.event_type}:{event_id}" seen = await self.session.scalar( select(MemoryEvidence.id).where(MemoryEvidence.idempotency_key == idempotency_key)) if seen is not None: @@ -91,7 +100,8 @@ class MemoryExtractionWorker: extracted.value, memory_type=extracted.memory_type, confidence=extracted.confidence, - source_type=SOURCE_TYPE, + source_type=self.source_type, + status=self.memory_status, structured_value={ "memory_key": extracted.memory_key, "value": extracted.value, @@ -125,7 +135,7 @@ class MemoryExtractionWorker: async def _event_id(self, run_id: str, result_message_id: int) -> str | None: """按 payload 回查本事件的事件 id，作为重复消费的幂等边界。""" - conditions = [DomainEventOutbox.event_type == "memory.extraction_requested"] + conditions = [DomainEventOutbox.event_type == self.event_type] if run_id: conditions.append(DomainEventOutbox.aggregate_id == run_id) else: @@ -158,6 +168,8 @@ class MemoryExtractionWorker: content = (message.content or "").strip() if not content: return None, message.id + if self.sanitize_source: + content = sanitize_customer_service_message(content) return content, message.id async def _request_message( diff --git a/app/worker/runtime.py b/app/worker/runtime.py index 5fd7e80..97eb99b 100644 --- a/app/worker/runtime.py +++ b/app/worker/runtime.py @@ -40,6 +40,7 @@ from app.service.memory_recall_service import MemoryRecallService from app.service.memory_service import CacheDeleteAdapter, MemoryService from app.service.memory_taxonomy import BUSINESS_EVENT_TYPES from app.service.model_gateway import DatabaseModelEndpointResolver, ModelGenerationService +from app.worker.customer_profile_candidate_worker import CustomerProfileCandidateWorker from app.worker.episode_worker import ( EpisodeConsumptionResult, EpisodeExtractionConsumer, @@ -145,6 +146,20 @@ class WorkerRuntime: signals=MemoryService.detect_memory_signals(message), ) + @staticmethod + def should_request_profile_candidate( + *, agent_type: str, context: RequestContext, message: str, + ) -> bool: + """客服只为已登录客户生成候选，不为访客生成正式画像数据。""" + if agent_type != "customer_service" or "visitor" in context.roles: + return False + if not {"customer", "authenticated_user"}.intersection(context.roles): + return False + if context.data_scope != "self": + return False + # 只有明确的长期偏好、约束或目标才进入候选队列，普通问答不触发模型抽取。 + return bool(MemoryService.detect_memory_signals(message)) + async def dispatch_one(self, *, run_id: str | None = None) -> bool: # Outbox acknowledges a durable SQL queue entry, not an in-memory task. async with SessionFactory() as session: @@ -162,6 +177,13 @@ class WorkerRuntime: session, extractor=self.memory_extraction, cache=self.memory_cache ).handle(payload) + async def dispatch_profile_candidate(payload: dict[str, Any]) -> None: + if "message_id" not in payload or "customer_id" not in payload: + raise ValueError("profile candidate payload is incomplete") + await CustomerProfileCandidateWorker( + session, extractor=self.memory_extraction, cache=self.memory_cache + ).handle(payload) + async def dispatch_run_completed(payload: dict[str, Any]) -> None: # 结果消息与审计已由 complete_run 同事务落库，此事件只承担 # "运行已完成"的对外通知职责。当前没有独立外部消费者， @@ -224,6 +246,7 @@ class WorkerRuntime: handlers: dict[str, Callable[[dict[str, Any]], Awaitable[None]]] = { "agent.run_requested": dispatch, "memory.extraction_requested": dispatch_memory_extraction, + "customer_profile.candidate_requested": dispatch_profile_candidate, "agent.run_completed": dispatch_run_completed, "config.cache_invalidate_requested": dispatch_cache_invalidate, "memory.deletion_requested": dispatch_memory_deletion, @@ -536,6 +559,9 @@ class WorkerRuntime: result=result, business_events=business_events, ), + profile_candidate_requested=self.should_request_profile_candidate( + agent_type=run.agent_type, context=context, message=request.message, + ), ) # 只有数据库成功保存完整用户/助手轮次后才写短期 Redis；Redis 故障不应让已完成 # 的客服回答回滚或重试。短期组件会再次脱敏，形成持久化链路的第二道保护。 diff --git a/docs/客服Agent二期_客户画像候选流程_v1.md b/docs/客服Agent二期_客户画像候选流程_v1.md new file mode 100644 index 0000000..888b24e --- /dev/null +++ b/docs/客服Agent二期_客户画像候选流程_v1.md @@ -0,0 +1,84 @@ +# 客服 Agent 二期：客户画像候选流程 + +版本：v1.0 +适用分支：`ZSY_develop2` +状态：候选提取链路已实现，确认/审核入口待后续迭代 + +## 1. 业务边界 + +本阶段只为**已登录用户**的客服对话生成画像候选，不为访客生成任何客户画像数据。 +客服 Agent 仍然不读取长期画像、不读取持仓/收益/订单/银行卡/投诉进度，也不直接修改 +`fin_customer_profile` 或 `profile_snapshots`。 + +候选数据不是正式画像，不能用于客服回答、产品推荐、风险等级判断或交易决策。 + +## 2. 处理流程 + +```text +已登录用户客服消息 + -> 识别明确的长期偏好/约束/目标信号 + -> 完成客服回答并在同一事务写入候选 Outbox + -> Worker 回查权威用户消息 + -> 二次脱敏 + -> 受控模型抽取 memory_key/value/type/confidence + -> 写入 memory_unit(status='candidate') + memory_evidence + -> 等待用户确认或管理员审核 + -> 后续流程再决定是否晋升为 active/profile_snapshots +``` + +普通公开问答、闲聊、一次性操作问题不触发候选抽取；访客、非客服 Agent、非 self 数据范围 +和缺少已登录身份标记的事件均失败关闭。 + +## 3. 事件契约 + +事件类型：`customer_profile.candidate_requested` + +事件只携带定位和身份信息，不携带用户原文： + +```json +{ + "run_id": "运行 ID", + "message_id": 123, + "customer_id": 456, + "actor_type": "authenticated_customer" +} +``` + +事件由 `AgentPersistenceService.complete_run()` 与客服回答、审计和运行终态在同一事务写入， +由 `WorkerRuntime` 异步消费。重复消费使用事件 ID 作为证据幂等键。 + +## 4. 数据与安全约束 + +- 候选写入既有 `memory_unit`，状态固定为 `candidate`，不会覆盖同键的 `active` 记忆。 +- 证据写入既有 `memory_evidence`，幂等键格式为 + `customer_profile.candidate_requested:{event_id}`。 +- 候选抽取前对消息执行客服隐私脱敏；证据摘录不得保存密码、验证码、完整手机号、 + 完整证件号或完整银行卡号。 +- 模型输出继续使用既有受控词表和严格 JSON 校验；抽取失败时不写任何记忆。 +- 访客候选事件必须被消费者拒绝，不能仅依赖上游路由判断。 +- `MemoryRecallService` 只召回 `active` 状态，因此候选不会进入任何 Agent 的长期记忆上下文。 + +## 5. 当前已实现文件 + +- `app/service/memory_service.py`：支持候选状态写入，并保证候选不覆盖正式记忆。 +- `app/worker/memory_extraction_worker.py`：支持事件类型、状态、来源和脱敏策略配置。 +- `app/worker/customer_profile_candidate_worker.py`：已登录客服候选专用消费者。 +- `app/service/agent_persistence_service.py`：完成客服运行时写入候选 Outbox 事件。 +- `app/worker/runtime.py`：候选触发判定与事件处理器。 + +## 6. 尚未实现的后续能力 + +1. 用户确认候选的接口和页面。 +2. 管理员候选列表、审核、驳回和审计接口。 +3. 候选晋升为 `active` 的冲突检测、版本切换和 `profile_snapshots` 生成。 +4. 候选撤回、过期、删除和隐私授权管理。 +5. 候选流程的 MySQL 集成测试和管理员端到端验收。 + +在上述能力完成前，禁止把 `candidate` 状态直接作为正式画像对外展示或用于业务决策。 + +## 7. 验证结果 + +- 客服画像候选专项测试：通过。 +- 一期单元与契约回归：`683 passed`。 +- Ruff：通过。 +- Mypy：通过。 diff --git a/tests/unit/service/test_agent_persistence_handover.py b/tests/unit/service/test_agent_persistence_handover.py index a58ecb3..53b1a22 100644 --- a/tests/unit/service/test_agent_persistence_handover.py +++ b/tests/unit/service/test_agent_persistence_handover.py @@ -239,3 +239,22 @@ async def test_normal_result_does_not_create_handover_ticket_or_event() -> None: event.event_type != "conversation.transfer_requested" for event in added_of(session.added, DomainEventOutbox) ) + + +@pytest.mark.asyncio +async def test_profile_candidate_event_is_written_without_modifying_profile() -> None: + """画像候选只写 Outbox，正式画像更新留给后续确认/审核服务。""" + session = FakeSession(queued_run()) + + await AgentPersistenceService(session).complete_run( + "run-transfer-1", result(transfer_required=False), + memory_extraction_requested=False, profile_candidate_requested=True, + ) + + events = added_of(session.added, DomainEventOutbox) + candidate = next( + event for event in events if event.event_type == "customer_profile.candidate_requested" + ) + assert candidate.payload["customer_id"] == 7 + assert candidate.payload["actor_type"] == "authenticated_customer" + assert added_of(session.added, HandoverTicket) == [] diff --git a/tests/unit/service/test_memory_service.py b/tests/unit/service/test_memory_service.py index 57f7a7e..871d46a 100644 --- a/tests/unit/service/test_memory_service.py +++ b/tests/unit/service/test_memory_service.py @@ -116,6 +116,24 @@ async def test_upsert_persists_structured_value_when_created() -> None: assert created.memory_key == "preference:risk_level" +@pytest.mark.asyncio +async def test_upsert_candidate_never_updates_active_memory() -> None: + """候选写入不读取或覆盖同键正式记忆，等待确认后再晋升。""" + session = fake_session() + session.scalar.side_effect = [None] + added: list[Any] = [] + session.add = Mock(side_effect=added.append) + + created = await MemoryService(session).upsert( + 7, "preference:risk_level", "稳健型", memory_type="preference", + confidence=0.9, status="candidate", + ) + + assert created.status == "candidate" + assert created.content == "稳健型" + assert added == [created] + + @pytest.mark.asyncio async def test_upsert_refreshes_structured_value_on_existing_memory() -> None: existing = memory(5, key="preference:risk_level", content="保守型") diff --git a/tests/unit/worker/test_memory_extraction_worker.py b/tests/unit/worker/test_memory_extraction_worker.py index 2d0a0e2..9111d0a 100644 --- a/tests/unit/worker/test_memory_extraction_worker.py +++ b/tests/unit/worker/test_memory_extraction_worker.py @@ -8,6 +8,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.core.errors import RecoverableAgentError from app.model.conversation import ConversationMessage from app.service.memory_extraction_service import ExtractedMemory +from app.worker.customer_profile_candidate_worker import CustomerProfileCandidateWorker from app.worker.memory_extraction_worker import MemoryExtractionWorker NOW = datetime(2026, 9, 9, 0, 0, 0, tzinfo=UTC).replace(tzinfo=None) @@ -194,3 +195,40 @@ async def test_missing_message_is_skipped_without_write() -> None: assert not await MemoryExtractionWorker(session, extractor=StubExtractor(EXTRACTED)).handle( {"run_id": "run-1", "message_id": 11, "customer_id": 7}, event_id="event-1") session.add.assert_not_called() + + +@pytest.mark.asyncio +async def test_profile_candidate_is_sanitized_and_not_active() -> None: + """客服画像候选只保存脱敏证据，并且状态必须是 candidate。""" + session = link() + added: list[Any] = [] + session.add = Mock(side_effect=added.append) + extractor = StubExtractor(EXTRACTED) + + assert await CustomerProfileCandidateWorker(session, extractor=extractor).handle( + { + "run_id": "run-1", "message_id": 11, "customer_id": 7, + "actor_type": "authenticated_customer", + }, + event_id="event-candidate", + ) + + memory = next(item for item in added if type(item).__name__ == "MemoryUnit") + evidence = next(item for item in added if type(item).__name__ == "MemoryEvidence") + assert memory.status == "candidate" + assert memory.source_type == "AI对话提取" + assert evidence.idempotency_key == "customer_profile.candidate_requested:event-candidate" + assert extractor.calls == [USER_FACT] + + +@pytest.mark.asyncio +async def test_profile_candidate_rejects_visitor_event() -> None: + """消费者对缺少已登录身份标记的事件失败关闭。""" + session = link() + extractor = StubExtractor(EXTRACTED) + + assert not await CustomerProfileCandidateWorker(session, extractor=extractor).handle( + {"run_id": "run-1", "message_id": 11, "customer_id": 7, "actor_type": "visitor"}, + event_id="event-visitor", + ) + assert extractor.calls == [] diff --git a/tests/unit/worker/test_runtime_worker_dispatch.py b/tests/unit/worker/test_runtime_worker_dispatch.py index cb30ecc..9a0d379 100644 --- a/tests/unit/worker/test_runtime_worker_dispatch.py +++ b/tests/unit/worker/test_runtime_worker_dispatch.py @@ -18,6 +18,7 @@ PAYLOAD: dict[str, Any] = { OUTBOX = { "agent.run_requested", "memory.extraction_requested", + "customer_profile.candidate_requested", "agent.run_completed", "config.cache_invalidate_requested", # 投影清理事件必须有消费者，否则 memory.invalidated/memory.deleted 永久 pending。 @@ -77,6 +78,30 @@ def test_authenticated_customer_service_does_not_request_memory_extraction() -> assert requested is False +def test_authenticated_customer_service_requests_profile_candidate() -> None: + """已登录客户明确陈述偏好时只生成候选，不复用正式记忆事件。""" + context = RequestContext( + user_id="7", trace_id="customer-trace", roles=("customer",), + permissions=("agent:run",), data_scope="self", + ) + + assert WorkerRuntime.should_request_profile_candidate( + agent_type="customer_service", context=context, message="我的风险偏好是稳健型" + ) is True + + +def test_visitor_does_not_request_profile_candidate() -> None: + """访客即使陈述偏好也不能创建画像候选。""" + context = RequestContext( + user_id="visitor:test", trace_id="visitor-trace", roles=("visitor",), + permissions=("agent:run",), data_scope="public", + ) + + assert WorkerRuntime.should_request_profile_candidate( + agent_type="customer_service", context=context, message="我的风险偏好是稳健型" + ) is False + + @pytest.mark.asyncio async def test_customer_service_short_memory_append_is_available_on_runtime() -> None: """Worker 的短期会话依赖与长期画像抽取依赖必须彼此独立。"""