feat: add customer profile candidate flow

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

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:

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)

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(

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 故障不应让已完成
         # 的客服回答回滚或重试。短期组件会再次脱敏，形成持久化链路的第二道保护。

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：通过。

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) == []

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="保守型")

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 == []

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 的短期会话依赖与长期画像抽取依赖必须彼此独立。"""