feat: 记忆→画像→图全自动触发(含修复记忆内容覆盖失败的无符号列 bug)
一、自动触发 在记忆抽取 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 文件无错。
This commit is contained in:
@@ -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,
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user