diff --git a/.env.example b/.env.example index 46c7ecd..9d605e7 100644 --- a/.env.example +++ b/.env.example @@ -29,7 +29,6 @@ MESSAGE_BROKER_DLQ_TOPIC=agent.dlq MILVUS_URI=http://127.0.0.1:19530 MILVUS_TOKEN= -MILVUS_COLLECTION=jr_memory NEO4J_URI=bolt://127.0.0.1:7687 NEO4J_DATABASE=neo4j diff --git a/app/core/config.py b/app/core/config.py index bb47db4..21680b7 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -50,7 +50,12 @@ class Settings(BaseSettings): milvus_uri: str milvus_local_uri: str = "" milvus_token: str = "" - milvus_collection: str = "jr_memory" + # 长期记忆向量集合名**刻意不做成配置项**：它是写（投影）、读（召回）、删（清理） + # 三侧共用的代码级契约，集合 schema 与向量维度（1024）也由代码定义 + # （`app/infrastructure/milvus_profile_projection.py` 的 `PROFILE_COLLECTION`）。 + # 这里原本有个 `milvus_collection`（`jr_memory`），写路径却从没读过它 —— + # 只有读/删两侧读，于是"写进 A、从 B 查、从 B 删"，语义召回恒空、清理恒报成功却不删。 + # 一个只在契约一侧生效的配置项比没有配置项更危险，故直接删除。 neo4j_uri: str neo4j_database: str = "neo4j" neo4j_username: str = "neo4j" diff --git a/app/infrastructure/milvus_profile_projection.py b/app/infrastructure/milvus_profile_projection.py index fa00efb..ad9f332 100644 --- a/app/infrastructure/milvus_profile_projection.py +++ b/app/infrastructure/milvus_profile_projection.py @@ -27,6 +27,15 @@ from app.core.errors import RecoverableAgentError logger = logging.getLogger(__name__) +#: 长期记忆向量集合名的**唯一真相**（写、读、删三侧共用，见下方说明），不是"某处默认值"。 +#: +#: 2026-09-14 故障复盘：本常量与召回/清理两侧读的 `settings.milvus_collection` +#: （`jr_memory`，**该集合从未被创建**）不一致，于是 +#: ① 语义召回每次 search 都抛异常 → 降级 `milvus_unavailable` → 召回恒空； +#: ② 记忆删除走 `vector_collection_absent` → 报清理成功但一个向量都没删。 +#: 两侧各自单测全绿，接缝没人守。现约定：**这里改，就三侧一起改** +#: （`app/service/agent/bootstrap.py`、`app/service/projection_cleanup_service.py`）， +#: 并由 `tests/unit/infrastructure/test_memory_vector_collection_consistency.py` 守住。 PROFILE_COLLECTION = "user_long_term_memory_v1" VECTOR_DIM = 1024 @@ -60,6 +69,23 @@ class MilvusProfileProjection: async def upsert(self, payload: dict[str, Any]) -> None: customer_id, profile_version, sources = self._normalize(payload) + if not sources: + # ⚠️ 零可投影记忆时**直接返回**，不要先去 `load_collection`。 + # + # `load_collection` 在集合不存在或未加载时会抛错，客户端把它包装成 + # `RecoverableAgentError`（`milvus_profile_vector_client.py:40-47` + # 「画像向量集合不可用」）。而 `upsert` 此前**无条件**在 embed 循环之前调它， + # 于是"本来就没有可写内容"的事件也被记成投递失败、重试 5 次进死信。 + # + # 实测正是这样：10001 / 10002 开户风险测评投的两条画像事件， + # `memory_unit` 里没有它们的记忆（`_with_memory_sources` 兜底查出 0 条）， + # 在 load_collection 那一步就死了 —— 而 `user_long_term_memory_v1` + # 当时根本不存在。**"没东西可写"不该等同于"写失败"**。 + logger.info( + "profile projection skipped: no projectable memory customer_id=%s", + customer_id, + ) + return load_collection = getattr(self._client, "load_collection", None) if load_collection is not None: await load_collection(collection_name=self._collection) diff --git a/app/infrastructure/vector_memory.py b/app/infrastructure/vector_memory.py index 5c1ee07..53f30f6 100644 --- a/app/infrastructure/vector_memory.py +++ b/app/infrastructure/vector_memory.py @@ -3,7 +3,10 @@ from typing import Any, Protocol class VectorClient(Protocol): - def search(self, collection_name: str, data: list[list[float]], limit: int) -> Any: ... + def search( + self, collection_name: str, data: list[list[float]], limit: int, + filter: str | None = None, + ) -> Any: ... class VectorSearchResult: @@ -31,12 +34,30 @@ class VectorMemoryAdapter: self.client = client self.collection = collection - def search(self, embedding: list[float], limit: int = 10) -> VectorSearchResult: + def search( + self, + embedding: list[float], + limit: int = 10, + *, + customer_id: int | None = None, + ) -> VectorSearchResult: + """按向量检索；给了 `customer_id` 就**在集合内按客户过滤**。 + + 过滤必须下推到 Milvus：召回服务回表时虽然也会按客户筛一次，但不过滤的话 + 别家客户的命中会白占 `limit` 名额，本客户能拿到的条数被稀释甚至清零—— + 这在多客户数据集上会表现成"语义召回时有时无"。 + + 过滤表达式在 `try` **之外**求值：`customer_id` 不是整数属调用方编码错误， + 应当直接抛出，不能伪装成"Milvus 不可用"而被降级吞掉。 + """ + expr = None if customer_id is None else f"customer_id == {int(customer_id)}" try: + kwargs: dict[str, Any] = {} if expr is None else {"filter": expr} hits = self.client.search( collection_name=self.collection, data=[embedding], limit=max(1, min(limit, 100)), + **kwargs, ) return VectorSearchResult(hits) except Exception: diff --git a/app/service/agent/bootstrap.py b/app/service/agent/bootstrap.py index 9041f51..71c925a 100644 --- a/app/service/agent/bootstrap.py +++ b/app/service/agent/bootstrap.py @@ -19,6 +19,7 @@ from app.infrastructure.fund_quote_cache import FundQuoteCache from app.infrastructure.graph import build_graph_driver from app.infrastructure.memory_cache import MemoryCacheAdapter from app.infrastructure.milvus_knowledge_writer import MilvusKnowledgeWriter +from app.infrastructure.milvus_profile_projection import PROFILE_COLLECTION from app.infrastructure.milvus_profile_vector_client import MilvusProfileVectorClient from app.infrastructure.vector_memory import VectorMemoryAdapter from app.service.agent.factory import AgentFactory @@ -117,13 +118,20 @@ def get_fund_quote_cache() -> FundQuoteCache | None: @lru_cache(maxsize=1) def get_vector_memory_adapter() -> VectorMemoryAdapter | None: - """Milvus 适配器；构造失败返回 None（语义通道关闭），不影响结构化召回。""" + """Milvus 适配器；构造失败返回 None（语义通道关闭），不影响结构化召回。 + + 集合名取 `PROFILE_COLLECTION`，与投影写入路径（`MilvusProfileProjection`） + **同一个常量**。此前这里读的是 `settings.milvus_collection`（`jr_memory`， + 该集合从未创建，现已删除该配置项），于是适配器构造成功、每次 `search` 都抛异常 + 并被降级成 `milvus_unavailable` —— 表现为"语义召回永远没数据"， + 而投影那条链路看起来一切正常。 + """ try: from pymilvus import MilvusClient # type: ignore[import-untyped] settings = get_settings() client = MilvusClient(uri=settings.milvus_uri, token=settings.milvus_token or None) - return VectorMemoryAdapter(client, settings.milvus_collection) + return VectorMemoryAdapter(client, PROFILE_COLLECTION) except Exception: logger.warning("vector memory adapter unavailable; semantic recall disabled", exc_info=True) diff --git a/app/service/memory_recall_service.py b/app/service/memory_recall_service.py index 508d22f..d657231 100644 --- a/app/service/memory_recall_service.py +++ b/app/service/memory_recall_service.py @@ -172,7 +172,7 @@ class MemoryRecallService: # 纵深防御：适配器契约上不抛异常，但召回服务自己对外承诺"外部依赖故障只降级、 # 不阻塞主流程"，这条承诺不能寄托在适配器实现上。 try: - found = self.vector.search(embedding, limit=limit) + found = self.vector.search(embedding, limit=limit, customer_id=customer_id) if found.degraded: logger.warning("vector recall degraded: milvus unavailable collection=%s", self.vector.collection) diff --git a/app/service/projection_cleanup_service.py b/app/service/projection_cleanup_service.py index f6aa8c2..ba3816e 100644 --- a/app/service/projection_cleanup_service.py +++ b/app/service/projection_cleanup_service.py @@ -16,6 +16,7 @@ from sqlalchemy import delete, select from app.core.config import get_settings from app.infrastructure.db import SessionFactory +from app.infrastructure.milvus_profile_projection import PROFILE_COLLECTION from app.model.memory import MemoryUnit from app.model.profile import UserFact from app.service.profile_assembly_service import ProfileAssemblyService @@ -120,12 +121,16 @@ class ProjectionCleanupService: details.append("vector_client_unavailable") return False try: - if settings.milvus_collection not in set(client.list_collections()): + # 集合名与投影写入路径**同一个常量**（`PROFILE_COLLECTION`）。此前读的是 + # 已删除的 `settings.milvus_collection`（`jr_memory`），该集合不存在 + # ⇒ 这里每次都走下面的 `vector_collection_absent` 分支、 + # **报清理成功但一个向量都没删**，陈旧向量永久留存。 + if PROFILE_COLLECTION not in set(client.list_collections()): # 记忆向量集合尚未启用：没有需要清理的派生数据 details.append("vector_collection_absent") return True client.delete( - collection_name=settings.milvus_collection, + collection_name=PROFILE_COLLECTION, filter=f'memory_uuid == "{memory_uuid}"', ) details.append("vector_cleaned") diff --git a/app/worker/runtime.py b/app/worker/runtime.py index 3126dcf..a430e88 100644 --- a/app/worker/runtime.py +++ b/app/worker/runtime.py @@ -8,6 +8,7 @@ from typing import Any, Protocol, cast from uuid import uuid4 from sqlalchemy import select, update +from sqlalchemy.ext.asyncio import AsyncSession from app.core.config import Settings, get_settings from app.core.contracts import AgentRequest, AgentRequestMetadata, AgentResult, RequestContext @@ -15,6 +16,7 @@ from app.core.errors import AgentError, RecoverableAgentError, RunLeaseLostError from app.infrastructure.db import SessionFactory from app.model.audit import InteractionAudit from app.model.conversation import ConversationMessage +from app.model.memory import MemorySyncOutbox from app.model.platform import ( AgentRun, DomainEventOutbox, @@ -292,6 +294,11 @@ class WorkerRuntime: if result.degraded: logger.warning("graph projection degraded customer_id=%s reason=%s", customer_id, result.reason) + # ★ 向量投影：图投影在上面同步做完了，**向量侧此前完全没做** —— + # 这个方法里只有图投影，没有任何 milvus 投递，后果是长期记忆的向量 + # 从未写入过，`memory_recall_service._vector()` 永远搜不到东西 + # （实测：`memory_sync_outbox` 里客户 9001 有 0 行）。 + self._enqueue_memory_vector_projection(session, int(customer_id)) async def dispatch_handover_queue_ready(payload: dict[str, Any]) -> None: """记录转人工队列已就绪；不向客户承诺已接单或处理时限。""" @@ -631,6 +638,52 @@ class WorkerRuntime: handled += 1 return handled + @staticmethod + def _enqueue_memory_vector_projection(session: AsyncSession, customer_id: int) -> None: + """把该客户的长期记忆投递到向量集合 `user_long_term_memory_v1`。 + + **为什么需要它**：`dispatch_profile_rebuild` 此前只做**图**投影 + （`ProfileGraphProjectionService`），**向量侧完全没做** —— 于是长期记忆的向量 + 从未被写入过，`memory_recall_service._vector()` 永远搜不到任何东西。 + 实测：客户 9001 的记忆在 09-13、09-14 都更新过，而 `memory_sync_outbox` + 里它**一行都没有**（表里仅有的 4 行是 10001/10002 开户测评的画像投影）。 + + **为什么走事件而不是在这里直接投影**：写向量要跑一次 embedding 网络调用。 + 放进 outbox 才有重试/退避/死信，也不会把网络等待拖进本事务。 + + **为什么不调 `ProfileGenerationService.generate()`**：上面 + `ProfileAssemblyService.rebuild()` 已经写过 `profile_snapshots`（自带 + version / is_current / hash）。两套生成器并行会让每次重建写两条快照、 + 互相清 `is_current`、版本跳号。这里**只投事件**，不碰快照。 + + **payload 只带 `customer_id`**：消费侧 `_with_memory_sources` 会回退查 + `memory_unit` 里该客户的 active 记忆。语义上成立 —— 长期记忆是**客户级**的， + 不是画像版本级的；每条记忆自带 `version`，适配器按 `memory_uuid + version` + 做幂等，所以"用的是哪一版"仍然确定。 + """ + now = datetime.now(UTC).replace(tzinfo=None) + session.add(MemorySyncOutbox( + event_uuid=str(uuid4()), + aggregate_type="profile", + aggregate_uuid=str(customer_id), + aggregate_version=1, + target_store="milvus", + operation="upsert", + # ⚠️ `version` 不能省：适配器的 `_coerce_profile_version` + # （`milvus_profile_projection.py:140-145`）要求 payload 里 + # `profile_version` 或 `version` 是**正整数**，缺失直接抛 `ValueError` + # （实测踩到：只带 customer_id 时第 43 行投影立刻 failed/ValueError）。 + # 它只用于日志 —— 真正的幂等靠每条记忆自己的 `version` + # （适配器按 `memory_uuid + version` 去重），所以固定 1 是安全的。 + payload={"customer_id": str(customer_id), "version": 1}, + status="pending", + retry_count=0, + next_retry_at=None, + last_error=None, + created_at=now, + processed_at=None, + )) + async def _with_memory_sources(self, payload: dict[str, Any]) -> dict[str, Any]: """保证 payload 带 `memory_sources`；缺失时回退为查询当前有效记忆。 diff --git a/docs/37-记忆投影链路实现说明.md b/docs/37-记忆投影链路实现说明.md index f42a340..f8e9188 100644 --- a/docs/37-记忆投影链路实现说明.md +++ b/docs/37-记忆投影链路实现说明.md @@ -82,6 +82,51 @@ 建集合：`python tools/setup_milvus_profile_collection.py` ——**幂等**，集合已存在时只做结构比对报告、不覆盖不删重建（共享 Milvus 实例里还有别的项目的集合）。 +### ⚠️ 集合名是"写、读、删"**三侧共用**的代码级契约（2026-09-14 修） + +集合名 `user_long_term_memory_v1` **不做成配置项**，唯一真相是 +`app/infrastructure/milvus_profile_projection.py` 的 `PROFILE_COLLECTION` 常量。 + +**踩过的坑（本次语义召回"恒空"的 4 个根因里的第 4 个，也是最隐蔽的一个）**： +三侧各写各的集合名 —— + +| 侧 | 位置 | 用的名字 | +|---|---|---| +| **写** | `MilvusProfileProjection`（`PROFILE_COLLECTION`） | `user_long_term_memory_v1` ✅ | +| **读** | `bootstrap.get_vector_memory_adapter()` | `settings.milvus_collection` = **`jr_memory`** ❌ | +| **删** | `ProjectionCleanupService._cleanup_vector()` | 同上 ❌ | + +`.env` 里 `MILVUS_COLLECTION=jr_memory`，而 **`jr_memory` 这个集合在本仓从未被创建过** +（实测 `list_collections()` 只有 `fin_faq_collection` / `fin_policy_collection` / +`fin_product_collection` / `user_long_term_memory_v1`）。于是： + +- **读**：适配器构造成功（`MilvusClient` 不校验集合存在），每次 `search` 抛异常并被 + `VectorMemoryAdapter.search` 吞掉 → `degraded=True` → 召回结果恒 + `degraded_reasons=('milvus_unavailable',)`、**向量通道条数恒为 0**。 + 写侧的投影日志一切正常，所以"写入成功、召回没数据"看似矛盾，实际是**查错了集合**。 +- **删**：`jr_memory not in list_collections()` 为真 → 走 `vector_collection_absent` + 分支 → **报"清理成功"，但一个向量都没删**，陈旧向量永久留存。 + +两侧各自的单测都是绿的 —— 因为**没人守"两侧必须是同一个名字"这条接缝**。 + +**修法（三条一起改）**： + +1. `app/infrastructure/milvus_profile_projection.py`：`PROFILE_COLLECTION` 成为唯一常量， + 并注明"这里改就三侧一起改"。 +2. `app/service/agent/bootstrap.py` 与 `app/service/projection_cleanup_service.py`： + 改为直接引用该常量。 +3. **删掉 `Settings.milvus_collection` 配置项**（`app/core/config.py`）并清掉 + `.env` / `.env.example` 里的 `MILVUS_COLLECTION`：写侧从来没读过它， + 一个**只在契约一侧生效**的配置项比没有配置项更危险。`Settings` 的 `extra="ignore"` + 保证其他环境残留的 `MILVUS_COLLECTION` 被安全忽略（不会报错、也不会再生效）。 + +回归守卫：`tests/unit/infrastructure/test_memory_vector_collection_consistency.py` +（断言读侧适配器与删侧用的都是 `PROFILE_COLLECTION`，且被删掉的配置项不得回归）。 + +> 另注：同批次还给召回加了**客户过滤下推** —— `VectorMemoryAdapter.search(..., customer_id=N)` +> 现在会带 `filter="customer_id == N"` 去查。此前不带过滤，别家客户的命中会白占 `limit` +> 名额，本客户能拿到的条数被稀释（回表时虽按 customer_id 再筛一次，但名额已经没了）。 + ### `memory_sources` 契约（适配器的输入） ```json @@ -306,10 +351,15 @@ handler 异常被捕获 ✓、`status`/`retry_count`/`last_error`/`next_retry_at - `tools/setup_milvus_profile_collection.py` - `tools/normalize_memory_sync_outbox.py`（历史取值订正，默认 dry-run、幂等） - `tests/unit/infrastructure/test_milvus_profile_projection.py` +- `tests/unit/infrastructure/test_memory_vector_collection_consistency.py`（集合名三侧同源守卫） - `tests/unit/worker/test_memory_sync_outbox_worker.py` - `tests/unit/worker/test_runtime_profile_projection.py`（消费端兜底 5 用例） **修改** +- `app/core/config.py`（**删除 `milvus_collection` 配置项**，见 §3 的集合名契约） +- `app/infrastructure/vector_memory.py`（`search()` 支持按 `customer_id` 下推过滤） +- `app/service/memory_recall_service.py`（语义检索传 `customer_id`） +- `app/service/projection_cleanup_service.py`（清理改引用 `PROFILE_COLLECTION`） - `app/service/profile_generation_service.py`（取值改小写、payload 加 `memory_sources` 与 `profile_version`） - `app/repository/profile_repository.py`（`active_memories()`） - `app/service/agent/bootstrap.py`（`get_milvus_profile_vector_client()`） diff --git a/docs/演示用/记忆召回恒空-根因与修复-2026-09-14.md b/docs/演示用/记忆召回恒空-根因与修复-2026-09-14.md new file mode 100644 index 0000000..4980857 --- /dev/null +++ b/docs/演示用/记忆召回恒空-根因与修复-2026-09-14.md @@ -0,0 +1,204 @@ +# 记忆召回"恒空"根因与修复（第二轮） + +> 承接 `记忆系统排查报告-2026-09-14.md`（第一轮：断头路 / 身份语义 / 引用校验）与 +> `记忆系统修复文档-2026-09-14.md`（F1/F3/F2/F5）。 +> 本轮目标是回答"**为什么语义召回一条数据都没有**"，结论是**四个根因叠在一起**， +> 全部已修并留下可复现证据。技术细节见 `docs/37-记忆投影链路实现说明.md`。 + +--- + +## 〇、一句话结论 + +**"没数据"不是一件事，是四件事串在一起。** 四个根因分层：**投递层**（事件根本没产生）→ +**门禁层**（产生了却被重试计数挡住）→ **写入层**（集合不存在 + 空内容被误判为失败）→ +**读取层**（写进了 A 集合、从 B 集合查）。前三层修好后召回依然是空的，因为第四层 +把读和写指向了两个不同的集合 —— **这是最隐蔽的一层，两侧单测全绿**。 + +| 你提的现象 | 根因 | 状态 | +|---|---|---| +| ① 语义召回（Milvus）没数据 | R1+R2+R3+R4（四层叠加） | ✅ **已修，端到端验证通过** | +| ② 员工身份召回恒空 | 授权范围语义未定义 | ❌ **未修 —— 需你拍板**（见 §4.1） | +| ③ 记忆更新传不到画像 | R1（重试计数把事件挡住） | ✅ **已修，有版本证据** | + +--- + +## 一、四个根因 + +### R1（门禁层）重复聚合把 `retry_count` 刷到 1405，事件被永久过滤 + +- **现象**：`memory_sync_outbox` 里 40 条 `profile.rebuild_requested` 全部 `retry_count=1405`。 +- **机制**：`episode_worker._persist()` 每次都调 `_touch_retry()`，**重复聚合也照样 +1**。 + 消费端只领 `retry_count < worker_retry_limit`(=3) 的事件 ⇒ 一旦超过 3 就**永远领不到， + 且不会报错、不会进死信**（它连被领取的资格都没有）。 +- **修法**：删掉 `_touch_retry` 及其调用点；补回归用例 + `test_repeated_aggregation_never_bumps_retry_count`。存量数据已订正（40 行归零）。 +- **证据**：修前 pending 40 → 修后 **0**。 + +### R2（投递层）`dispatch_profile_rebuild` **只有图投影，没有任何向量投递** + +- **现象**：客户 9001 的记忆在 09-13、09-14 都更新过，而 `memory_sync_outbox` 里 + **它一行都没有**（全表仅 4 行，是 10001/10002 开户测评的画像投影）。 +- **机制**：该处置函数只调 `ProfileGraphProjectionService`（写 Neo4j）， + **Milvus 侧完全没有投递代码** ⇒ 长期记忆的向量从未被写入过。 +- **修法**：新增 `WorkerRuntime._enqueue_memory_vector_projection()`，在画像重建后投一条 + `MemorySyncOutbox(target_store="milvus", aggregate_type="profile")`。走事件而非同步写， + 是为了拿 Outbox 的重试/退避/死信，且不把 embedding 的网络等待拖进事务。 + ⚠️ payload 必须带 `version`（正整数）：适配器的 `_coerce_profile_version` 缺它直接 + 抛 `ValueError`（实测踩到，事件立刻 failed）。 + +### R3（投递层）`episode_worker` 从不投 `profile.rebuild_requested` + +- **机制**：抽取出记忆后没有触发画像重建 ⇒ 记忆进了 `memory_unit`，画像永远不动。 +- **修法**：`record_evidence` 返回非空时投 `DomainEventOutbox(event_type="profile.rebuild_requested", + trigger="episode_extraction")`。 +- **证据**：伪造一个片段跑消费，`extracted=[191]`，rebuild 事件 **3 → 4**。 + +### R4（读取层）**写的集合和读的集合不是同一个** ← 最隐蔽，也是最后一根稻草 + +| 侧 | 位置 | 集合名 | +|---|---|---| +| **写** | `MilvusProfileProjection.PROFILE_COLLECTION` | `user_long_term_memory_v1` ✅ | +| **读** | `bootstrap.get_vector_memory_adapter()` | `settings.milvus_collection` = **`jr_memory`** ❌ | +| **删** | `ProjectionCleanupService._cleanup_vector()` | 同上 ❌ | + +`.env` 里 `MILVUS_COLLECTION=jr_memory`，而 **`jr_memory` 从未被创建** +（实测 `list_collections()` 只有 `fin_faq_collection` / `fin_policy_collection` / +`fin_product_collection` / `user_long_term_memory_v1`）。于是： + +- **读**：`MilvusClient` 构造不校验集合存在 ⇒ 适配器"构造成功"，每次 `search` 抛异常被 + `VectorMemoryAdapter.search` 吞掉 → `degraded=True` → 召回**恒** + `degraded_reasons=('milvus_unavailable',)`、向量命中恒 0 条。 + 写侧日志一切正常，所以"写入成功却说没数据"看似矛盾，**实际是查错了集合**。 +- **删**：`jr_memory` 不在集合列表 ⇒ 走 `vector_collection_absent` 分支 ⇒ + **报"清理成功"但一个向量都没删**，陈旧向量永久留存。 + +**修法（三侧锁死成一个常量，把配置项删掉）**： + +1. `PROFILE_COLLECTION` 成为唯一常量，注释写明"这里改就三侧一起改"； +2. 读侧（`bootstrap.py`）、删侧（`projection_cleanup_service.py`）改为直接引用它； +3. **删除 `Settings.milvus_collection` 配置项**，并清掉 `.env` / `.env.example` 里的 + `MILVUS_COLLECTION`：写侧从来没读过它，一个**只在契约一侧生效**的配置项比没有配置项 + 更危险。`Settings` 的 `extra="ignore"` 保证其他环境残留的该变量被安全忽略。 +4. 回归守卫：`tests/unit/infrastructure/test_memory_vector_collection_consistency.py`。 + +**同批附带**：语义检索此前**不带客户过滤**，别家客户的命中会白占 `limit` 名额； +现在 `VectorMemoryAdapter.search(..., customer_id=N)` 会带 +`filter="customer_id == N"` 下推到 Milvus。 + +### 附：两个"助推"因素（同批修掉） + +- **向量集合本身不存在**：`user_long_term_memory_v1` 此前从未创建过，用 + `python tools/setup_milvus_profile_collection.py` 建好（幂等，已存在只做结构比对）。 +- **`upsert` 无条件先 `load_collection`**：集合不存在时抛错 ⇒ "**本来就没有可写内容**" + 的事件被记成**投递失败**、重试 5 次进死信（实测 10001/10002 那两条就是这样死的）。 + 现改为 `sources` 为空时**先返回**，留给"零写入"与"没跑"可区分的日志。 + +--- + +## 二、端到端验证证据（2026-09-14） + +### 2.1 记忆 → 画像（现象 ③） + +`profile_snapshots`**版本号真的动了**： +**v7 @ 2026-09-10 13:57 → v10 @ 2026-09-14 13:06:57**， +`user_facts.preference:risk_level = "进取型"`（来自伪造的验证片段）。不是"跑了没报错"， +是**库里的值变了**。 + +### 2.2 记忆 → 向量库（现象 ① 的写入侧） + +集合 `user_long_term_memory_v1` 现有 2 行，都是客户 9001： + +| memory_uuid | memory_key | content | version | +|---|---|---|---| +| `187e3b7c…` | `preference:risk_level` | 进取型 | 5 | +| `fdae7288…` | `preference:horizon` | 约三年 | 4 | + +`memory_sync_outbox` 该客户的行**全部 `processed`**（清理掉 1 行验证残留后： +`[('milvus','processed',2), ('neo4j','processed',2)]`）。 + +### 2.3 语义召回（现象 ① 的读取侧）—— 走**生产装配**，不是测试替身 + +```powershell +& D:\conda\envs\jr_py313\python.exe -X utf8 - <脚本：build_memory_recall_service → recall()> +``` + +读侧集合打印为 `user_long_term_memory_v1`（修前是 `jr_memory`），召回结果： + +| query | degraded | sources | 命中与排序 | +|---|---|---|---| +| `投资期限` | **False** | `['milvus']` | `horizon`(0.288) > `risk_level`(0.201) | +| `风险偏好` | **False** | `['milvus']` | `risk_level`(0.287) > `horizon`(0.154) | +| `三年` | **False** | `['milvus','mysql']` | `horizon` 双通道，**vector_score 0.810** | +| `进取型` | **False** | `['milvus','mysql']` | `risk_level` 双通道，**vector_score 0.9987** | +| `货币基金` | **False** | `['milvus']` | 正文不含该词 ⇒ 只有弱相关向量命中(0.255) | +| `None` | **False** | `['mysql']` | 结构化全量 2 条（conf 1.0 / 0.95） | + +**判定**：`degraded=False`、`degraded_reasons=()`、**排序随 query 语义变化**、 +双通道按 `memory_uuid` 正确合并 —— 这是真实语义信号，不是"接口不报错"。 + +### 2.4 回归 + +- `pytest tests/unit tests/contract` → **1432 passed, 2 skipped, 1 failed** + （唯一失败是同事正在改的投顾页面 `test_advisor_workspace_registers_documented_operation_endpoints`， + 与记忆链路无关）。 +- 新增/更新用例：集合名三侧同源守卫 4 条、召回客户过滤断言 1 条、 + episode worker 重试计数回归 1 条（共 +4 通过）。 + +--- + +## 三、演示前必做（否则前端看到的仍是旧行为） + +1. **重启 API 进程**：`get_vector_memory_adapter()` 是 `lru_cache` 且旧进程仍持有 + `jr_memory` 那版代码，**不重启则接口依旧返回 `milvus_unavailable`**。 + （我在验证时刻意用生产装配在进程内调用，没有动你正在跑的窗口。） +2. **确保常驻 Worker 在跑**：`python -X utf8 -m app.worker`。记忆抽取是 + `EPISODE_INTERVAL_ROUNDS = 30` 轮一次的节奏，`--once` **碰不到** episode 链路。 +3. 行情有效期只有 15 分钟，演示前先 `python tools/sync_market_prices.py`。 + +--- + +## 四、未修 / 待你决策 + +### 4.1 ⚠️ 员工身份召回恒空 —— **卡在你的一个决策上** + +`governance.recall` 场景下员工查客户记忆恒空。**技术上已定位**：召回入口按"身份"取 +客户范围，而**员工的可见范围没有权威定义**，所以要么查不到、要么可能越权。 +修它需要你先定**授权口径**，二选一： + +- **(A) 按 `sys_customer_assignment` 归属**：员工只能召回"分配给我"的客户。 + 语义最严（谁负责谁可见），但依赖分配表数据完整 —— 数据没维护就会什么都查不到。 +- **(B) 按角色 `data_scope`**：如投顾可见其服务范围内全部客户。 + 更贴近真实组织，但需要 `data_scope` 有明确定义，越权面更大。 + +**我倾向 (A)**：金融场景"最小可见"优先，且它天然可审计。**你拍板后我再动手改。** + +### 4.2 其他（不阻塞演示） + +| 项 | 说明 | +|---|---| +| `EpisodeExtractionConsumer` 幂等哈希不覆盖记忆内容 | `snapshot_hash` 只算 `fin_customer_profile`+`fin_risk_assessment`，**"只有记忆变了"会被判定 `changed=False` 短路**，`_memory_sources()` 根本不执行。建议把 `memory_uuid+version` 纳入哈希，或拆成 `snapshot_changed`/`memory_changed` 两个判据。 | +| `ProjectionReconciliationService.mark_replay` 无调用方 | 有方法没入口（无 CLI / 无 Worker 接线），对账重放实际跑不起来。 | +| 投顾线两个生产者不发 `memory_sources` | `profile_governance_service` / `risk_questionnaire_service` 的 payload 仍缺该键（`docs/37` §8.2 已登记）。消费端有兜底所以不死信，但根治在投顾线。 | +| `today_profit_loss` 硬编码 0 | 业务决策项，见 `软件需求文档-2026-09-14.md` **Q22**。 | + +--- + +## 五、本轮改动文件 + +**新增** +- `tests/unit/infrastructure/test_memory_vector_collection_consistency.py`（集合名三侧同源守卫） +- 本文件 + +**修改** +- `app/core/config.py`（**删除 `milvus_collection` 配置项**） +- `app/infrastructure/milvus_profile_projection.py`（集合名唯一常量 + 空内容早返回） +- `app/infrastructure/vector_memory.py`（`search()` 支持 `customer_id` 下推过滤） +- `app/service/memory_recall_service.py`（语义检索传 `customer_id`） +- `app/service/agent/bootstrap.py`（读侧改用 `PROFILE_COLLECTION`） +- `app/service/projection_cleanup_service.py`（删侧改用 `PROFILE_COLLECTION`） +- `app/worker/runtime.py`（`_enqueue_memory_vector_projection`） +- `app/worker/episode_worker.py`（去掉 `_touch_retry`；投 `profile.rebuild_requested`） +- `tests/unit/service/test_memory_recall_service.py`（断言客户过滤） +- `tests/unit/worker/test_episode_worker.py`（重试计数回归用例） +- `.env.example`（移除 `MILVUS_COLLECTION`） +- `docs/37-记忆投影链路实现说明.md`（补集合名契约与根因） diff --git a/tests/unit/infrastructure/test_memory_vector_collection_consistency.py b/tests/unit/infrastructure/test_memory_vector_collection_consistency.py new file mode 100644 index 0000000..bc3c91f --- /dev/null +++ b/tests/unit/infrastructure/test_memory_vector_collection_consistency.py @@ -0,0 +1,122 @@ +"""长期记忆向量集合名必须"写、读、删"三侧同源。 + +2026-09-14 实测故障（三侧各写各的，接缝无人守）： + +- **写**（`MilvusProfileProjection`）用模块常量 `PROFILE_COLLECTION` + = `user_long_term_memory_v1`； +- **读**（`bootstrap.get_vector_memory_adapter`）用 `settings.milvus_collection` + = `jr_memory` —— **该集合在本仓从未被创建**； +- **删**（`ProjectionCleanupService._cleanup_vector`）同读侧。 + +后果是两条静默故障：语义召回每次 `search` 抛异常 → 降级 `milvus_unavailable` +→ 召回恒空；记忆删除读到集合不存在 → 走 `vector_collection_absent` 分支 → +**报清理成功但一个向量都没删**。而三侧各自的单测全绿。 + +本文件守的就是这条接缝：任何一侧换集合名，这里必须红。 +""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from app.core.config import Settings +from app.infrastructure.milvus_profile_projection import PROFILE_COLLECTION +from app.service import projection_cleanup_service +from app.service.agent import bootstrap +from app.service.projection_cleanup_service import ProjectionCleanupService + +MEMORY_UUID = "187e3b7c-0000-4000-8000-000000000001" + + +class FakeMilvusClient: + """记录构造参数与删除调用的 Milvus 客户端替身；不发网络请求。""" + + instances: list[FakeMilvusClient] = [] + collections: list[str] = [] + + def __init__(self, *, uri: str, token: str | None = None) -> None: + self.uri = uri + self.token = token + self.deletes: list[dict[str, Any]] = [] + FakeMilvusClient.instances.append(self) + + def list_collections(self) -> list[str]: + return list(FakeMilvusClient.collections) + + def delete(self, *, collection_name: str, filter: str) -> None: # noqa: A002 + self.deletes.append({"collection_name": collection_name, "filter": filter}) + + +@pytest.fixture(autouse=True) +def _reset_fake_client() -> None: + FakeMilvusClient.instances = [] + FakeMilvusClient.collections = [PROFILE_COLLECTION] + + +class _FakeSettings: + milvus_uri = "http://milvus:19530" + milvus_token = "" + + +def test_retired_collection_setting_is_not_reintroduced() -> None: + """`milvus_collection` 配置项必须保持删除状态。 + + 它只在契约的读/删两侧生效、写侧从不读它 —— 一个"看着权威但只有一半生效"的 + 配置项，正是本次故障的成因。重新加回来会让同一个坑再出现一次。 + """ + assert "milvus_collection" not in Settings.model_fields + + +def test_recall_adapter_reads_the_projection_collection( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """读侧集合名 == 写侧常量，而不是任意配置项。""" + monkeypatch.setattr(bootstrap, "get_settings", lambda: _FakeSettings()) + monkeypatch.setattr("pymilvus.MilvusClient", FakeMilvusClient) + bootstrap.get_vector_memory_adapter.cache_clear() + try: + adapter = bootstrap.get_vector_memory_adapter() + finally: + # 必须清缓存：否则被 monkeypatch 污染的实例会泄漏给同会话后续测试。 + bootstrap.get_vector_memory_adapter.cache_clear() + + assert adapter is not None + assert adapter.collection == PROFILE_COLLECTION + + +@pytest.mark.asyncio +async def test_cleanup_deletes_from_the_projection_collection( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """删侧必须真的去同一集合删，而不是因为查了另一个集合名就"报成功"。""" + monkeypatch.setattr(projection_cleanup_service, "get_settings", lambda: _FakeSettings()) + monkeypatch.setattr("pymilvus.MilvusClient", FakeMilvusClient) + + details: list[str] = [] + cleaned = await ProjectionCleanupService()._cleanup_vector(MEMORY_UUID, details) + + assert cleaned is True + assert details == ["vector_cleaned"] + assert FakeMilvusClient.instances[-1].deletes == [{ + "collection_name": PROFILE_COLLECTION, + "filter": f'memory_uuid == "{MEMORY_UUID}"', + }] + + +@pytest.mark.asyncio +async def test_cleanup_reports_honestly_when_collection_is_absent( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """集合确实不存在时如实报 `vector_collection_absent`（合法降级，不是伪造成功）。""" + FakeMilvusClient.collections = [] + monkeypatch.setattr(projection_cleanup_service, "get_settings", lambda: _FakeSettings()) + monkeypatch.setattr("pymilvus.MilvusClient", FakeMilvusClient) + + details: list[str] = [] + cleaned = await ProjectionCleanupService()._cleanup_vector(MEMORY_UUID, details) + + assert cleaned is True + assert details == ["vector_collection_absent"] + assert FakeMilvusClient.instances[-1].deletes == [] diff --git a/tests/unit/service/test_memory_recall_service.py b/tests/unit/service/test_memory_recall_service.py index 9621162..babe842 100644 --- a/tests/unit/service/test_memory_recall_service.py +++ b/tests/unit/service/test_memory_recall_service.py @@ -31,15 +31,24 @@ QUERY = "基金" class StubVectorClient: - """Milvus 客户端替身：返回嵌套命中或抛连接错误，不发网络请求。""" + """Milvus 客户端替身：返回嵌套命中或抛连接错误，不发网络请求。 + + 记录 `filter` 是为了守住"语义检索必须按客户下推过滤"这条约束——不过滤的话 + 别家客户的命中会白占 `limit` 名额，本客户的召回条数被稀释。 + """ def __init__(self, rows: list[Any] | None = None, error: Exception | None = None) -> None: self.rows: list[Any] = list(rows or []) self.error = error self.calls: list[int] = [] + self.filters: list[str | None] = [] - def search(self, collection_name: str, data: list[list[float]], limit: int) -> Any: + def search( + self, collection_name: str, data: list[list[float]], limit: int, + filter: str | None = None, + ) -> Any: self.calls.append(limit) + self.filters.append(filter) if self.error is not None: raise self.error return [self.rows] @@ -193,6 +202,8 @@ async def test_recall_merges_structured_and_vector_channels() -> None: assert vector.confidence == pytest.approx(0.63) assert vector.evidence["vector_score"] == pytest.approx(0.9) assert client.calls == [10] + # 语义检索必须把客户过滤下推到向量库，否则别家客户的命中会白占 limit 名额。 + assert client.filters == [f"customer_id == {CUSTOMER_ID}"] async def test_vector_hit_content_is_reloaded_from_mysql_authority() -> None: