From d7f6ef7ddc4579e6554e46f48010bdbcdf35f58f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=8D=BF=E4=BA=91=E7=A7=8B=E6=9C=88?= <15273589815@163.com> Date: Thu, 10 Sep 2026 21:52:20 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E8=AE=B0=E5=BF=86=E2=86=92=E7=94=BB?= =?UTF-8?q?=E5=83=8F=E2=86=92=E5=9B=BE=E5=85=A8=E8=87=AA=E5=8A=A8=E8=A7=A6?= =?UTF-8?q?=E5=8F=91=EF=BC=88=E5=90=AB=E4=BF=AE=E5=A4=8D=E8=AE=B0=E5=BF=86?= =?UTF-8?q?=E5=86=85=E5=AE=B9=E8=A6=86=E7=9B=96=E5=A4=B1=E8=B4=A5=E7=9A=84?= =?UTF-8?q?=E6=97=A0=E7=AC=A6=E5=8F=B7=E5=88=97=20bug=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 一、自动触发 在记忆抽取 worker 里，记忆写入成功后**写一条 profile.rebuild_requested 事件**，由 Worker 的事件循环在下一轮消费，完成画像重建与图投影。 为什么不直接在抽取处调用：抽取时那条记忆还在**未提交**的事务里，另开 session 去重建画像 看不到它——实测踩到过：画像重建确实执行了、快照也多了一条，但新事实没进 user_facts、 画像字段没更新、图里也没多出关系。改走事件后，它只可能在本事务提交之后被消费，届时数据 一定可见，且与记忆写入共享事务边界（要么都留痕、要么都不留）。 链路因此变成：客户说话 → 记忆抽取 → 画像更新 → 图投影，全程无需手工介入。 runtime 侧新增 dispatch_profile_rebuild handler 与 relationships 注入点（与既有 projection_cleaner 同一模式）；未注入或图库不可用时投影如实降级，不影响画像更新。 二、顺带修复：记忆内容变化会导致整条更新失败 实测触发：同一 memory_key 的内容从"约三年"改成"长期（5年以上）"时， INSERT INTO memory_conflict 报 `1264 Out of range for column 'right_memory_id'`。 根因：memory_service._conflict_right_id 在"同一行原地更新、没有独立新值行"时返回 `-memory.id` 作为合成标识（注释写明了意图是与恒为正的自增主键不冲突），但库中 right_memory_id 是 `BIGINT UNSIGNED NOT NULL`，写负数被 MySQL 直接拒绝。 后果不是丢一条冲突记录，而是**记忆内容一旦变化、整条更新就失败**， Worker 反复重试直至事件进入死信。 因基线字段不可变更（AGENTS.md 第 4 条禁止改动已有字段的类型），改为在无符号范围内的 高位取值 `2**63 + memory.id`：真实自增主键从 1 开始且远小于 2^63，因此该值必为正、 且必然不等于任何真实记忆行主键，原设计"左右不相等且不混淆"的意图完整保留。 三、实测结果（全程未运行任何手工脚本） 客户两条消息（"投资期限约三年" → 改口"长期，五年以上"）之后： · memory_unit 2 行，horizon 记忆 version=2、conflict_count=1； · memory_conflict 1 行，合成标识 9223372036854776037（= 2^63+229）合法写入； · user_facts 2 行（事实自动提升，置信 0.95 过门槛）； · fin_customer_profile.investment_horizon 自动更新为"长期（5年以上）"； · profile_snapshots 4 个版本； · Neo4j 自动出现 HAS_GOAL 关系；profile.rebuild_requested 事件为 published； · ruff 通过、mypy 112 文件无错。 --- app/service/memory_service.py | 15 +++++++--- app/worker/memory_extraction_worker.py | 21 ++++++++++++++ app/worker/runtime.py | 40 ++++++++++++++++++++++++++ 3 files changed, 72 insertions(+), 4 deletions(-) diff --git a/app/service/memory_service.py b/app/service/memory_service.py index f71f389..76c44e8 100644 --- a/app/service/memory_service.py +++ b/app/service/memory_service.py @@ -217,9 +217,16 @@ class MemoryService: async def _conflict_right_id(self, memory: MemoryUnit) -> int: """冲突右侧标识：优先取同键历史版本行，否则取新版本的合成标识。 - 同一行原地更新时旧值与新值落在同一行，库中没有"新值行"的主键可用； - 这里用左侧 id 的相反数作为新版本的稳定标识——它必然不等于左侧 id， - 且与任何自增主键（恒为正）不冲突，因此不会与真实记忆行混淆。 + 同一行原地更新时旧值与新值落在同一行，库中没有"新值行"的主键可用， + 因此需要一个不会与真实记忆行混淆的合成标识。 + + **已修正的缺陷**：原实现返回 `-memory.id`，但库中 `right_memory_id` 是 + `BIGINT UNSIGNED NOT NULL`，写入负数在 MySQL 上直接报 1264 Out of range， + 后果是**记忆内容一旦发生变化，整条更新就失败**（实测触发：同一 key 的 + 投资期限从"约三年"改成"长期（5年以上）"）。因为基线字段不可变更 + （AGENTS.md 第 4 条禁止改动已有字段的类型），这里改为在无符号范围内的高位 + 取值：真实自增主键从 1 开始且远小于 2^63，因此合成标识既为正、又必然 + 不等于任何真实记忆行的主键，原设计"左右不相等且不混淆"的意图得以保留。 """ historical = await self.session.scalar( select(MemoryUnit.id) @@ -233,7 +240,7 @@ class MemoryService: ) if historical is not None: return int(historical) - return -int(memory.id) + return 2**63 + int(memory.id) async def record_evidence( self, diff --git a/app/worker/memory_extraction_worker.py b/app/worker/memory_extraction_worker.py index 921029c..73aac8e 100644 --- a/app/worker/memory_extraction_worker.py +++ b/app/worker/memory_extraction_worker.py @@ -1,6 +1,7 @@ import logging from datetime import UTC, datetime from typing import Any +from uuid import uuid4 from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession @@ -121,6 +122,26 @@ class MemoryExtractionWorker: source_record_id=str(source_message_id), occurred_at=now, ) + if recorded: + # 画像重建**不能在这里直接调用**：本方法的记忆写入还在当前事务里、尚未提交， + # 另开 session 去重建画像看不到这条新记忆——实测踩到过：画像重建确实执行了、 + # 快照也多了一条，但新事实没进 user_facts、画像字段没更新、图里也没多出关系。 + # 改为写一条事件：它只可能在本事务**提交之后**被消费，届时数据一定可见， + # 而且与记忆写入共享事务边界（要么都成功，要么都不留痕）。 + self.session.add(DomainEventOutbox( + id=0, + event_id=str(uuid4()), + event_type="profile.rebuild_requested", + aggregate_type="customer_profile", + aggregate_id=str(customer_id), + trace_id=run_id, + payload={"customer_id": customer_id, "trigger": "memory_extraction"}, + status="pending", + retry_count=0, + occurred_at=now, + created_at=now, + updated_at=now, + )) return recorded async def _event_id(self, run_id: str, result_message_id: int) -> str | None: diff --git a/app/worker/runtime.py b/app/worker/runtime.py index 40eb0d0..a0770df 100644 --- a/app/worker/runtime.py +++ b/app/worker/runtime.py @@ -79,6 +79,7 @@ class WorkerRuntime: endpoint_resolver: ExtractionEndpointResolver | None = None, memory_cache: CacheDeleteAdapter | None = None, projection_cleaner: ProjectionCleaner | None = None, + relationships: Any | None = None, ) -> None: self.factory = factory if factory is not None else get_agent_factory() self.settings = settings or get_settings() @@ -99,6 +100,15 @@ class WorkerRuntime: # Milvus/Neo4j 删除客户端：当前组装层没有提供（bootstrap 只装配召回用的读适配器）， # 因此默认 None = 投影清理显式降级并留痕，绝不写成"删除成功"。 self.projection_cleaner = projection_cleaner + # 图关系服务：画像投影用它写入节点与关系（投顾的多跳推荐、风控的关系网络都读它）。 + # 默认取生产装配；图库不可用时该值为 None，投影如实降级而不是失败。 + if relationships is not None: + self.relationships = relationships + else: + # 延迟导入：bootstrap 会间接导入本模块，模块级导入会形成循环依赖 + from app.service.agent.bootstrap import get_relationship_service + + self.relationships = get_relationship_service() # episode 聚合是低频批处理，按轮次节流而不是每轮都查。 self._episode_rounds = 0 @@ -129,6 +139,34 @@ class WorkerRuntime: async def dispatch_cache_invalidate(payload: dict[str, Any]) -> None: await self._invalidate_config_cache(payload) + async def dispatch_profile_rebuild(payload: dict[str, Any]) -> None: + """画像重建 + 图投影，由记忆写入后发出的事件驱动。 + + 为什么绕一层事件而不在记忆抽取处直接调用：抽取时那条记忆还在**未提交**的 + 事务里，另开 session 去重建画像看不到它（实测：快照加了、事实没进、图里 + 也没多出关系）。事件只可能在本事务提交之后被消费，届时数据一定可见。 + """ + customer_id = payload.get("customer_id") + if not customer_id: + raise ValueError("profile.rebuild_requested payload is incomplete") + # 延迟导入：bootstrap 会间接导入本模块，模块级导入会形成循环依赖 + from app.service.profile_assembly_service import ProfileAssemblyService + from app.service.profile_graph_projection_service import ( + ProfileGraphProjectionService, + ) + + outcome = await ProfileAssemblyService(session).rebuild(int(customer_id)) + if outcome.get("profile") is None: + # 客户尚未开户：画像行由开户流程创建（trade_account 等字段属注册侧所有）， + # 这里不代建。事实已提升进 user_facts，开户后再重建即可。 + logger.info("profile rebuild skipped (not opened) customer_id=%s", customer_id) + return + projection = ProfileGraphProjectionService(session, self.relationships) + result = await projection.project_customer(int(customer_id)) + if result.degraded: + logger.warning("graph projection degraded customer_id=%s reason=%s", + customer_id, result.reason) + async def dispatch_projection_cleanup(payload: dict[str, Any]) -> None: # memory.invalidated / memory.deleted 由 MemoryLifecycleService 按 # memory_uuid 写入，这里做幂等的投影清理（Milvus 向量、Neo4j 关系）。 @@ -159,6 +197,8 @@ class WorkerRuntime: # 投影清理：这两类事件此前没有消费者，永久 pending。 "memory.invalidated": dispatch_projection_cleanup, "memory.deleted": dispatch_projection_cleanup, + # 画像重建：记忆写入后自动触发，使「记忆 → 画像 → 图」全链路无需手工介入 + "profile.rebuild_requested": dispatch_profile_rebuild, } return await OutboxWorker(session, handlers).publish_one(aggregate_id=run_id)