画像投影链路接通 + 合并主干（PR #7）并对齐两套投影实现 #9
No Assignees
Notifications
Due Date
No due date set.
Dependencies
No dependencies set.
Reference: AI260626/group_fqcd_jr#9
Reference in New Issue
Block a user
交付内容
1. 接通画像投影链路（此前零消费者）
user_long_term_memory_v1+MilvusProfileProjection+MilvusProfileVectorClient+MemorySyncOutboxWorker+ 建集合工具WorkerRuntime.consume_profile_projections()接入run_once()：milvus→ 向量投影；neo4j→ 复用主干ProfileGraphProjectionService（方案 A，不引入第二套图投影）memory_unit组装memory_sources；写入外部存储前经conversation_privacy脱敏2. 订正
memory_sync_outbox取值口径（此前静默失效）原写法照
docs/00§6.4.6 用大写MILVUS/NEO4J+ 中文待处理，而消费端按target_store的值分派 handler、且只领status in {"pending","failed"}—— 两个条件都不满足 ⇒ 事件任何消费者都领不到、永久滞留且不报错（唯一键(event_uuid, target_store)对大小写无约束，MySQL 也不报错）。已改为全仓一致的小写英文，并加契约回归测试（断言大写值分派不到 handler、会进死信）。
3. 修掉移植带来的两处必炸点
if not isinstance(customer_id, int)str(customer_id)⇒ 每个事件必然失败、重试 5 次进死信memory_key直接raiseconstraint:*/profile:*）不满足前缀 ⇒ 一条constraint:记忆毒死该客户整批现改为：接受纯数字字符串；不可投影键跳过并留痕。
4. 消费端
memory_sources缺失兜底投顾线两处生产者的 payload 没有该字段，原本每次画像变更都会死信。现缺失时回退查询该客户
memory_unit有效记忆并记 warning（使"谁没提供"保持可见）。兜底不掩盖格式错：键存在但类型不对时不兜底，仍失败关闭。5. 修掉 3 个主干同样存在的继承缺陷
（PR #7 之后
tests/integration未整套复跑，故均未被发现）tools/seed_test_rbac.py少建review_t(9004) —— 两个集成测试都依赖它。且用户↔角色绑定用zip(user_ids, role_ids, strict=True)按位置配对，隐含"USERS 与 ROLES 一一对应"；一加不绑角色的账号就ValueError、整个种子跑不完（commit()在最后 ⇒ 外部表现是"什么都没发生"）。已改为显式配对表。CustomerProfileCandidateService._write_profile_snapshot漏写current_customer_id—— 该列不是生成列而是普通可空列 + 唯一键uk_profile_snapshot_current，不写则唯一键形同虚设（多个 NULL 不冲突）；旧当前版本也没清该列、补写就会撞键。AGENTS.md记明。6. 独立缺陷：
profile_snapshots被两个 ORM 类重复映射app/model/profile.py与app/model/risk_questionnaire.py各有一个ProfileSnapshot映射同一张表。单独导入各自都没事，同时导入即抛InvalidRequestError—— Worker 既要重建画像又要处理投顾问卷，真的会被打挂（库里留下last_error='InvalidRequestError'的行）。已修为 re-export。需要评审的 5 处取舍
完整推演见
docs/39-主干合并对策记录.md。risk_questionnaire.py去重复定义 → re-export（与主干另一份提交内容完全一致，两边独立做了同一修复）seed_test_rbac.py加review_t+ 改显式配对_write_profile_snapshot补current_customer_idcustomer_service.py：COMPANY取主干的「奶龙基金责任有限公司」（本项目实际品牌名）；HOTLINE/SERVICE_HOURS取本线修复 —— 主干仍是占位符400-XXX-XXXX，不修则同一客服给客户两个不同号码（安全路由出口给真号、兜底出口给假号）app/worker/__main__.py删除memory_sync_handlers接线 —— 否则两套消费者读同一个memory_sync_outbox，而两者的neo4jhandler 不同（一套用 ZSY 的Neo4jProfileProjection按客户各建私有节点，一套用主干服务写共享 tag 节点）⇒ 同一事件被谁领到结果不定，等于同一事实在图里两种说法，正是方案 A 要避免的状态验证（合并主干后实测）
pytest tests（全量）2 failed, 1415 passed, 2 skippedpytest tests/integration102 passed, 1 skippedmypy appSuccess: no issues found in 245 source filestools/audit_schema.pytools/check_authoritative_docs.pytools/check_docs_endpoint_ids.pytools/check_rbac_seed_consistency.py那 2 个失败是既有环境项（
test_offsite_document_recognition_adapter.py断言请求体里的中文原文，而 httpx 序列化成\uXXXX），非本次引入。真机验证：真实 embedding(1024 维) + 真实 Milvus 写入并回读通过；消费端失败分支与成功分支（含跳过不可投影键、手机号脱敏）均实测通过。
遗留（详见
docs/39§6）memory_sources—— 本线已有消费端兜底（不再死信），根治待投顾线补或明确"这两个来源是否也要投影长期记忆"neo4j_profile_projection.py去留待定docs/00§6.4.6 取值栏与实现不一致 —— 按既定裁定未改基线文档，实际口径记在docs/37文档
docs/37-记忆投影链路实现说明.md（实现说明 + 根因 + 契约 + 验证证据）docs/38-架构对齐-记忆与画像投影链路.md（方案 A 等决定的决策依据）docs/39-主干合并对策记录.md（本次合并逐文件取舍）AGENTS.md新增三条易错点：memory_sync_outbox取值口径、一张表只能有一个 ORM 类、Windows 中文输出乱码的正确命令（-X utf8）两个都是"代码里存在但没接对"的缺陷，都不是新功能。 ## A1 客服热线在代码里有两个值（一个出口给假号码） - `app/core/customer_service_rules.py:35` `CONTACT_PHONE = "15936583816"` ← 真号码，安全路由 6 处在用 - `app/service/agent/implementations/customer_service.py:155` `HOTLINE = "400-XXX-XXXX"` ← 占位符，兜底出口在用 后果：**同一个客服给客户两个不同的电话号码**。问"风险等级怎么划分"被安全路由处理时给真号码； 问一个知识库答不了的问题走兜底时给 `400-XXX-XXXX` —— 客户按这个号码永远打不通。 修法：`HOTLINE` / `SERVICE_HOURS` 改为**转发** `customer_service_rules` 的两个常量 （不是"改成相同的值"，而是引用同一对象，避免日后再次漂移）；工作时间也随之从 "每日 7:00-22:00" 统一为 "工作日 09:00-18:00"（与安全路由出口一致）。 新增守卫测试用 `is` 断言对象同一性 —— 值相等挡不住"两边各写一份恰好相同"的漂移。 ## A2 `transfer_required` 既没落库也没出参 `docs/05` §6.3 一直规定 `GET /agent-runs/{run_id}` 的 `result` 里有 `transfer_required` / `transfer_reason`，但实现里两个都没有：前端判断"这轮要不要转人工" 只能靠**猜正文里有没有兜底话术的开头**（`docs/24` 自己把这称为权宜之计）。 - 写入侧：`conversation_message` **没有** `transfer_required` 列，加列要迁移且规则 4 禁止改既有 字段定义 ⇒ 放进 `tool_calls` 这个现成 JSON 列，作为 `calls` 的兄弟键 （`{"calls": [...], "transfer_required": bool, "transfer_reason": str|None}`） - 读取侧：`RunQueryService.get` 取出来放进 `result`；**兼容历史行**（`calls` 裸列表 / None → 按 False/None 处理，不抛异常、也不凭正文猜） 刻意**没做**的一半：`docs/05` §6.3 的 `result` 里还有 `degraded` / `degradation_reason`， 但 `CoreResult` 里根本没有这两个字段（降级信息目前只在工具出参里）—— 补它要改 `CoreResult` 并让各 Agent 传递降级状态，属另一个改动范围。**已在交付说明里注明这一半仍缺。** ## 真机验证 | 问题 | transfer_required | transfer_reason | 正文电话 | |---|---|---|---| 「请介绍一下量子纠缠在基金估值中的应用」 | **True** | 置信度不足：score=0.571 gap=0.004 | 15936583816 ✅ | 「请帮我计算一下三体问题的数值解」 | **True** | 置信度不足：score=0.499 gap=0.011 | 15936583816 ✅ | 「基金申购后多久确认」（正常知识直返） | False | — | 无（正确） | 「你们公司明天会下雪吗」（闲聊出口） | False | — | 无（正确） | （第一次我用"下雪"当兜底用例，结果它被闲聊出口正确接住了 —— 是我的期望值写错，不是代码问题。） 门禁：测试 1223 passed（新增 4 个用例）/ 3 failed（均为已知非代码缺陷）/ mypy 0 错。背景：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，会被消费至死信，待架构师确认是否投影。## 1. 独立缺陷：`profile_snapshots` 被两个 ORM 类重复映射 排查 `memory_sync_outbox` 中 `target_store='neo4j'` 那行 `last_error='InvalidRequestError'` 时发现，根因不在图库，而在模型层： - `app/model/profile.py` → `ProfileSnapshot` 映射 `profile_snapshots` - `app/model/risk_questionnaire.py` → **另一个** `ProfileSnapshot` 也映射 `profile_snapshots` SQLAlchemy 不允许两个类映射同一张表。实测： | 场景 | 结果 | |---|---| | 单独导入 `app.main` / `app.worker.runtime` | 正常 | | 单独导入 `profile_assembly_service` / `risk_questionnaire_service` | 正常 | | **两者同时导入** | `InvalidRequestError: Table 'profile_snapshots' is already defined` | Worker 在同一进程里既要处理 `profile.rebuild_requested`（走 `app.model.profile`）， 又要处理投顾风险问卷（走 `risk_questionnaire.py`）——所以这是**会打挂 Worker 的缺陷**， 不是理论风险。 修法：`app/model/risk_questionnaire.py` 不再重复定义，改为从 `app.model.profile` 转出（re-export），既有 4 处 `from app.model.risk_questionnaire import ProfileSnapshot` 无需改动。原定义多映射的 `current_customer_id` 经全仓核查无人使用，故不保留 （`app.model.profile` 明确注明该列由数据库维护、故意不映射）。 ## 2. 消费端兜底 `memory_sources` `memory_sources` 是本线新增的投影入参，而投顾线两处生产者的 payload （`{customer_id, profile_uuid, version, profile}`）没有这个键，原样会导致它们每次画像 变更都 `memory_sources is invalid` → 重试至死信。 新增 `WorkerRuntime._with_memory_sources()`：**键缺失或为 None** 时回退查询该客户 `memory_unit` 中 `status='active'` 的记忆，并记 warning（使"谁没提供"保持可见）。 语义成立：长期记忆是**客户级**而非画像版本级的，每条记忆自带 `version`， 适配器按 `memory_uuid + version` 幂等，故"用的是哪一版"仍确定。 **兜底不掩盖真错误**：键**存在但格式不对**时**不兜底**，原样交给适配器失败关闭。 实现上用键存在性判断而非 `isinstance`——后者会把"缺失"与"格式错"混为一谈， 那是初版实现里的一个真 bug，被新测试抓出后修正。 ## 3. 补上此前欠缺的消费端全路径验证 此前"整合验证"是直接调适配器，跳过了 outbox 的领取→分派→状态更新。 - **失败分支**（Milvus 断开时实测）：行被领取、按 target_store 分派、异常被捕获、 `status`/`retry_count`/`last_error`/`next_retry_at` 正确落库。 - **成功分支**（注入替身向量客户端）：outbox 行 → `processed`、`processed_at` 已写； 不可投影的 `constraint:` 被跳过（只写 1 行）；维度 1024；字符串客户号转 int； **手机号脱敏生效**（`稳健型投资者，手机号 [手机号已隐藏] 请勿外泄`）。 - **兜底实证**：历史行 `id=5`（payload 无 `memory_sources`）经兜底后成功投递为 `processed`。 - **修复实证**：`id=6` 的 `last_error` 从 `InvalidRequestError` 变为 `RecoverableAgentError`（图库不可用）——证明重复定义缺陷确已消除，剩下的是环境问题。 ## 4. 测试与验证 - 新增 `tests/unit/worker/test_runtime_profile_projection.py`（5 用例：已提供原样透传、 缺失兜底、空记忆给空列表而非删键、无客户号不兜底、格式错不兜底） - 全量：`2 failed, 1312 passed, 2 skipped`（2 个失败为既有环境项，非本次引入） - mypy：`Success: no issues found in 227 source files` - 表结构审计：89 张业务表无缺失/意外（未改动任何表结构） - 文档守卫：41 份文档无编号冲突 ## 5. 文档 `docs/32-记忆投影链路实现说明.md` 增补 §6.1（兜底）、§6.2（重复定义缺陷）、 §7.1（消费端全路径验证）并更新验证表与文件清单； `AGENTS.md` 新增"一张表只能有一个 ORM 类"易错点、校正测试基线数字。 ## 未做 未改 `docs/00` 基线、未动数据库迁移、未改投顾线生产者代码、未启动常驻 Worker。 遗留：投顾线两处生产者的 payload 仍缺 `memory_sources`（已有兜底，不再死信， 但根治应由投顾线确认）；Milvus/Neo4j 容器本轮不可用（Docker Desktop 崩溃）， `id=6` 停在 failed 属环境不可用、非代码缺陷。## 现象（2026-09-12 跑真实 Worker 时发现） `episode_worker` 反复报 `模型记忆抽取输出不是有效 JSON` 并重试到失败： ``` pydantic_core.ValidationError: Input should be a valid dictionary or instance of _ExtractionPayload input_value=[{'memory_key': None, 'value': None, 'memory_type': None, 'confidence': 0}] input_type=list ``` ## 根因 契约是**对象** `{...}`，而模型在 episode 抽取路径会返回**单元素数组** `[{...}]`。 `_parse` 直接 `json.loads` 后交给 pydantic，数组自然过不了 `model_validate`， 于是被归入"输出不是有效 JSON"这一条 —— 但它其实是**合法的空结果** （`_validate` 已能把"三字段为 null 且 confidence=0"正确识别为"无持久事实"，返回 None）。 后果：这类 episode 白跑一遍模型调用、重试到 `retry_count` 上限后判失败， **该片段的记忆永远抽不出来**。 ## 修法 `_parse` 里只对"**恰好一个对象**的数组"做归一化： - `[{...}]` → 取 `{...}`（空结果照常返回 None；有事实照常解析） - 多元素数组、元素非对象、空数组 → **不猜**，仍交给校验失败关闭 （多元素时无法判断哪个是答案，猜错会把错误记忆写进库，比失败更糟） ## 测试 `tests/unit/service/test_memory_extraction_service.py` 追加 2 个用例： - `test_single_element_array_is_normalized`：数组包空结果 → None；数组包有事实 → 正常解析 - `test_multi_element_or_non_dict_array_still_fails_closed`：多元素 / 非对象元素 / 空数组 → 仍失败 ## 验证 - `pytest tests/unit/service/test_memory_extraction_service.py tests/unit/worker/test_memory_extraction_worker.py` → 28 passed - `mypy app` → 0 错 / 245 文件 > 说明：`memory_extraction_service.py` 属主干线代码。此处是从**实际运行日志**里发现的 > 健壮性缺陷，改动限定在输出形状归一化，不改变抽取契约与校验规则。换到你的环境跑之前，有 4 条前置（第 1 条是我这边新引入的）
最终验收在你机器上跑，所以把"环境数据"的部分单独列出来 —— 这些不随代码合并，
缺了会产生看起来像代码缺陷的失败（正如你在
docs/32§5.1/§5.2 反复强调的那类）。python -X utf8 tools/setup_milvus_profile_collection.pyload_collection抛RecoverableAgentError("画像向量集合不可用")⇒memory_sync_outbox的milvus行失败重试到死信。是环境问题，不是代码缺陷docker ps见milvus-standalonehealthy；确认MILVUS_LOCAL_URI为空MILVUS_LOCAL_URI被设过会指向本地 Lite 文件、查的是另一个库（你已在AGENTS.md记过）task_type=embedding的端点_profile_embed抛"没有可用的 embedding 端点"，事件同样重试到死信--requirepass 123456，而.env的REDIS_URL原本没带密码AuthenticationError，表现为整片功能静默降级（只打 WARNING）：短期会话记忆、记忆召回热缓存、config_release缓存失效、登录限流我这边的环境差异（供你比数字时参考，不是缺陷）
pytest tests/unit tests/contractpytest tests（全量）2 failed, 1421 passed, 1 skippedpytest tests/integration102 passed, 1 skippedmypy appruffruff check app tests tools干净ruff check app tests tools alembic干净（我多含 alembic）那 2 个 failed 就是你在
docs/32§5.2 记为环境相关的test_offsite_document_recognition_adapter.py（httpx 把中文序列化成\uXXXX）——你已经写明"主干架构师环境复现不了、不要为了让它绿去动实现"，我这里确认同一个现象，没有动实现。
另：全仓
ruff check .会报 40 个错，全部来自根目录hq.py/nl2sql_yc.py（袁聪线脚本），我已在
AGENTS.md里把门禁命令写清为ruff check app tests tools alembic，避免下一个人把它读成"项目 40 个错"。另外：我在你环境之外跑通了端到端，可作为你验收的对照基线
之前本线的证据都停在组件级（我用替身客户端验证）。这次起了真实常驻 Worker，跑出：
那 1 条来自一次真实对话（客户 9102，含明确偏好陈述）：
preference:risk_level = "低风险"、status=candidate（候选态，按设计需审核后才提升为user_facts）。顺带在这次真实运行里发现并修掉 3 个缺陷（都在本 PR 内）：
memory_extraction_service._parse不接受单元素数组[{...}]—— 模型在 episode 路径就是这么返回的，被记成"不是有效 JSON"并重试到失败，而它其实是合法空结果
ProfileAssemblyService._write_snapshot把current_customer_id当成生成列（连 docstring都那么写），导致清旧不清、建新不写 ⇒ 唯一键
uk_profile_snapshot_current完全失效（实测数据是"旧版本留着值、新版本是 NULL"）。附数据订正工具
tools/fix_profile_snapshot_current.pyseed_test_rbac.py时漏删role_ids（ruffF841）—— 已由你的ffbcc22修掉并合入，我这里致谢；根因是我只跑了 mypy + pytest、没跑 ruff我已知、且已记录的遗留（不阻塞本 PR）
memory_sources（本线已在消费端加兜底，不再死信；根治待投顾线）app/infrastructure/neo4j_profile_projection.py因方案 A 失去生产引用，已在文件头加.. warning::说明"未装配/原因/启用前提"，去留待你定docs/00§6.4.6 的取值栏与实现不一致（按既定裁定未改基线文档，实际口径记在docs/37）合并就绪（已追平主干，2026-09-12 更新）：
NL_develop相对qyqy_develop：BEHIND 0、AHEAD 7NL_develop历史里）⇒ 无冲突、无需解冲突c200a61..5634fdc那 3 个提交（advisor 演示引导 / 风控登录权限与模型能力配置），与本线无文件重叠、合并无冲突；
tools/seed_test_rbac.py双方都改过，Git 自动合上，合并后
check_rbac_seed_consistency.py仍通过app/service/profile_assembly_service.pycurrent_customer_id漏写修复 —— 这条请重点看，它是画像写入主路径，我把该列从"当生成列"改成"显式写入/清空"tools/fix_profile_snapshot_current.pyAGENTS.mddocs/37-记忆投影链路实现说明.md合并前请再
git fetch一次确认（这条线今天动得勤）。上面那条长评论是环境前置与数字差异。补一个修：
tools/seed_sim_account_demo.py的当日行情不刷新（已推 NL_develop1f62aca）真机验收时
T001仪表盘返回503 FUND_QUOTE_UNAVAILABLE：产品 510300 行情已过期，T006持仓同样 503，而T010权益 200。排查结论在种子脚本里：_upsert_market_price()当天已有行时直接return，所以同一天重跑种子不刷新source_updated_at。行情是时效数据（
FundQuoteService按source_updated_at判新鲜度），于是表现为「刚灌完种子能用、过十几分钟就 503」—— 看着像行情适配器或缓存故障，实际与它们无关。
fin_market_price里当天那行的时间戳停在首次灌入时刻，是判定依据。修法：列值抽成共享
values，已有行就update刷新（含source_updated_at），无行才insert；账户/持仓的「已存在则跳过」不动（业务数据不该被种子覆盖）。
幂等 = 不产生重复行（
(product_id, trade_date)唯一），不是「不更新值」。改后验证（本机 9001
cust_t）：dashboard200 /holdings200 /entitlements200；ruff check app tests tools alembic全过、mypy app0 错（252 文件）、pytest tests与本线基线一致（
2 failed, 1469 passed, 1 skipped，2 个仍是test_offsite_document_recognition_adapter的中文转义环境差异，按
docs/32§5.2 不改实现）。⚠️ 顺便提醒：这条坑会让「灌完种子立刻验收」和「过一会儿再验收」结论不同，
建议
docs/32或验收清单里注明「灌完种子立刻跑，或验收前先重跑一次种子」。另，两个路径口径（前端接入时容易猜错，已用
app.routes实测列出）：GET /api/v1/users/me/entitlements—— 权益（T010），不在/account/下、也不带/benefit/前缀GET /api/v1/users/me/holdings—— 持仓，是/users/me/直属，不是/users/me/account/holdings用户端现有 13 个
GET/POST端点（memory-profile、memory-candidates 及其决策、account/dashboard、orders 及其详情/撤单、holdings、transactions 及其详情、cash-ledger、entitlements）。
经确认产品（列表/详情/行情）在用户端没有任何 HTTP 端点，且
fin_product除演示种子外为空 ——如果前端需要「产品列表/详情/行情」这几屏，需要新增端点，目前只能靠 Agent 工具兜。
View command line instructions
Checkout
From your project repository, check out a new branch and test the changes.