wangjianlong_0626
8a0cbab636
背景：memory_sync_outbox 这条链此前**完全没有消费者**，且生产端照 docs/00 §6.4.6
写成大写 MILVUS/NEO4J + 中文「待处理」，而消费端按 target_store 的**值**分派 handler、
且只领 status in {pending, failed} —— 两个条件都不满足，事件任何消费者都领不到、
永久滞留且不报错（唯一键 (event_uuid, target_store) 对大小写无约束，MySQL 也不报错）。
根因是代码与测试都硬编码字面量，所以测试跟着一起错、谁也没拦住。
订正
- profile_generation_service：取值改为全仓一致的小写（milvus/neo4j/upsert/pending）
- 测试改为引用常量并断言消费端契约，不再硬编码（硬编码是本次跑偏的直接原因）
- 新增契约回归测试：断言大写值分派不到 handler、会进死信，谁改回大写立刻红
- 新增 tools/normalize_memory_sync_outbox.py：订正历史脏行（默认 dry-run、幂等）
接通投影链路（此前零消费者）
- 新增 Milvus 集合 user_long_term_memory_v1 及建集合工具（幂等、不覆盖已有集合）
- 新增 MilvusProfileProjection / MilvusProfileVectorClient，并修掉移植带来的两处必炸点：
customer_id 由「必须 int」放宽为接受数字字符串（本仓所有生产者都写 str，
不放宽则每个事件必然失败）；不可投影的 memory_key 由「整批 raise」改为跳过留痕
（否则一条 constraint: 记忆毒死该客户整批，而受控词表 13 个键里有 7 个不满足前缀）
- 新增 MemorySyncOutboxWorker（领取/指数退避/死信骨架保留原样）并接入 WorkerRuntime
- milvus → 向量投影；neo4j → 复用主干 ProfileGraphProjectionService（方案 A，
不引入第二套投影，避免同一事实在图中两种说法、违反主干既有的只投影已确认事实的不变式）
- 生产端从 memory_unit(status=active) 组装 memory_sources，随事件带上确定快照
- 前置移植 conversation_privacy：写外部存储前脱敏手机号/证件号/银行卡等
验证
- 新增 17 个单测；全量 2 failed, 1307 passed, 2 skipped
（2 个失败为既有环境项：断言请求体中文原文而 httpx 序列化成 \uXXXX，非本次引入）
- mypy app → 0 错（227 文件）；audit_schema → 89 张业务表无缺失/意外，未改动表结构
- 真机：真实 embedding(1024 维) + 真实 Milvus 写入并回读通过
- 整合链路（测试记忆 → 生产端组装 → outbox → 消费端投递 → Milvus 回读）通过，
且 MySQL 已回滚、Milvus 无残留
文档
- 新增 docs/32-记忆投影链路实现说明.md：真实口径、根因、契约与验证证据（供接手）
- AGENTS.md：新增该易错点；新增 Windows 中文输出乱码的正确命令（-X utf8）；
校正测试基线与 mypy 文件数
未做：未改 docs/00 基线、未动数据库迁移、未改投顾线代码、未启动常驻 Worker。
遗留：投顾线两处生产者的 payload 缺 memory_sources，会被消费至死信，待架构师确认是否投影。
92 lines
3.2 KiB
Python
92 lines
3.2 KiB
Python
"""数据订正：把 `memory_sync_outbox` 的历史取值改齐为全仓小写口径。
|
||
|
||
**默认 dry-run**（只报告将影响几行），加 `--apply` 才真正提交。幂等，可重复运行。
|
||
|
||
### 背景
|
||
|
||
`docs/00` §6.4.6 的取值栏曾写作大写 `MILVUS`/`NEO4J`、`UPSERT` 与中文 `待处理`，
|
||
与全仓实现从未对齐。按那份文档写入的事件会**静默失效**：
|
||
|
||
- 消费端 `MemorySyncOutboxWorker` 按 `handlers.get(event.target_store)` 分派 handler；
|
||
- 领取条件是 `status.in_({"pending","failed"})`。
|
||
|
||
大写 + 中文两个条件都不满足 ⇒ 事件任何消费者都领不到、永久滞留且不报错
|
||
（唯一键 `(event_uuid, target_store)` 对大小写没有约束，MySQL 也不会报错）。
|
||
|
||
因此凡是在本仓写入过 `memory_sync_outbox` 的环境，都可能有这类脏行。
|
||
|
||
### 安全口径
|
||
|
||
- 只改 `target_store` / `operation` / `status` 的**值**；
|
||
- **不触碰**主键、唯一键 `uk_memory_sync_event (event_uuid, target_store)`、
|
||
`payload`、时间戳、`retry_count`；
|
||
- 就地改值不会新增行（两组取值大小写不同，本会各自成行，这里把它们并到正确的一组）；
|
||
- 中文字符不受 `LOWER()` 影响（非 ASCII 字节不变），故 `LOWER()` 是安全的。
|
||
|
||
用法：
|
||
.\\.venv\\Scripts\\python.exe tools\\normalize_memory_sync_outbox.py # 先看
|
||
.\\.venv\\Scripts\\python.exe tools\\normalize_memory_sync_outbox.py --apply # 再改
|
||
"""
|
||
|
||
import asyncio
|
||
import sys
|
||
|
||
from sqlalchemy import text
|
||
|
||
from app.infrastructure.db import engine
|
||
|
||
_STATUS_CASE = """
|
||
CASE status
|
||
WHEN '待处理' THEN 'pending'
|
||
WHEN '处理中' THEN 'processing'
|
||
WHEN '已完成' THEN 'processed'
|
||
WHEN '失败' THEN 'failed'
|
||
ELSE LOWER(status)
|
||
END
|
||
"""
|
||
|
||
_UPDATE = text(f"""
|
||
UPDATE memory_sync_outbox
|
||
SET target_store = LOWER(target_store),
|
||
operation = LOWER(operation),
|
||
status = {_STATUS_CASE}
|
||
WHERE target_store <> LOWER(target_store)
|
||
OR operation <> LOWER(operation)
|
||
OR status <> {_STATUS_CASE}
|
||
""")
|
||
|
||
_GROUP = text(
|
||
"SELECT target_store, operation, status, COUNT(*) n "
|
||
"FROM memory_sync_outbox GROUP BY target_store, operation, status"
|
||
)
|
||
|
||
_DETAIL = text(
|
||
"SELECT id, event_uuid, target_store, operation, status, retry_count "
|
||
"FROM memory_sync_outbox ORDER BY id"
|
||
)
|
||
|
||
|
||
async def main(apply: bool) -> int:
|
||
async with engine.connect() as conn:
|
||
before = (await conn.execute(_GROUP)).mappings().all()
|
||
print("改前:", [dict(row) for row in before] or "（空表）")
|
||
|
||
affected = (await conn.execute(_UPDATE)).rowcount
|
||
print("将订正行数:", affected)
|
||
|
||
if not apply:
|
||
print("dry-run：未提交。确认无误后加 --apply 再运行。")
|
||
await conn.rollback()
|
||
return 0
|
||
|
||
await conn.commit()
|
||
after = (await conn.execute(_GROUP)).mappings().all()
|
||
print("改后:", [dict(row) for row in after] or "（空表）")
|
||
for row in (await conn.execute(_DETAIL)).mappings().all():
|
||
print(" ", dict(row))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(asyncio.run(main("--apply" in sys.argv)))
|