lzf_0626
0c642133d4
语义召回"恒空"的根因分四层，本提交修掉投递层与读取层（另两层——重试计数
门禁、episode 不投 rebuild 事件——已在前两个提交修复）。
R2 投递层：dispatch_profile_rebuild 只做图投影，没有任何 Milvus 投递，
长期记忆的向量从未被写入过（实测客户 9001 在 memory_sync_outbox 里 0 行）。
新增 _enqueue_memory_vector_projection()，在画像重建后投
MemorySyncOutbox(target_store="milvus")。走事件而不是同步写，是为了拿
Outbox 的重试/退避/死信，且不把 embedding 的网络等待拖进事务。
⚠️ payload 必须带 version（正整数）：适配器 _coerce_profile_version 缺它
直接抛 ValueError（实测踩到，事件立刻 failed）。
R4 读取层：写的集合与读的集合不是同一个 ——
写（MilvusProfileProjection）用 "user_long_term_memory_v1"，
读（bootstrap.get_vector_memory_adapter）与删（projection_cleanup_service）
却用 settings.milvus_collection = "jr_memory"，而该集合从未被创建。
⇒ 召回：MilvusClient 构造不校验集合存在，适配器"构造成功"但每次 search
抛异常被 VectorMemoryAdapter 吞成 degraded → 召回恒
degraded_reasons=('milvus_unavailable',)、向量命中恒 0 条；
⇒ 清理：jr_memory 不在集合列表 → 走 vector_collection_absent 分支 →
报清理成功但一个向量都没删，陈旧向量永久留存。
修法：PROFILE_COLLECTION 成为唯一常量，读/删两侧直接引用它；
并删除 Settings.milvus_collection 配置项、清掉 .env.example 的
MILVUS_COLLECTION —— 写侧从来没读过它，一个只在契约一侧生效的配置项
比没有配置项更危险（Settings 的 extra="ignore" 会让其他环境残留的该
变量被安全忽略）。
顺带：
- 语义检索把客户过滤下推到 Milvus（filter="customer_id == N"）。此前不带
过滤，别家客户的命中会白占 limit 名额，稀释本客户的召回条数。
- upsert 在 sources 为空时先返回，不再无条件 load_collection ——
"本来就没有可写内容"不该被记成投递失败（10001/10002 那两条事件即如此
重试 5 次进死信）。
回归守卫：tests/unit/infrastructure/test_memory_vector_collection_consistency.py
断言读侧与删侧用的都是 PROFILE_COLLECTION，且被删掉的配置项不得回归。
这个缺陷能活下来，正是因为两侧单测全绿而接缝无人守。
验证（走生产装配、进程内调用，未重启你正在跑的 API 窗口）：
读侧集合打印 user_long_term_memory_v1（修前为 jr_memory）；
召回 degraded=False / reasons=() / sources 含 milvus，且排序随 query 语义
变化（投资期限→horizon 0.288 > risk_level 0.201；风险偏好→risk_level 0.287）；
query="进取型" 双通道合并且 vector_score=0.9987；query=None 走 mysql 全量。
pytest tests/unit tests/contract → 1432 passed, 2 skipped, 1 failed
（唯一失败是同事正在改的投顾页面，与记忆链路无关）。
文档：docs/演示用/记忆召回恒空-根因与修复-2026-09-14.md（新增，四层根因+证据）、
docs/37-记忆投影链路实现说明.md（补集合名三侧契约）。
218 lines
10 KiB
Python
218 lines
10 KiB
Python
"""Milvus 长期记忆投影适配器。
|
||
|
||
只写入已经审核的 `memory_sources`，不接受画像快照整体冒充单条记忆。
|
||
|
||
来源：同事 `ZSY_develop` 分支（`app/infrastructure/milvus_profile_projection.py`），
|
||
本文件在其基础上做了**两处契约放宽**，均为"避免整批失败"，不改变写入语义：
|
||
|
||
1. **`customer_id` 接受 int 或数字字符串**。原实现要求 `isinstance(customer_id, int)`，
|
||
而本仓**所有** outbox 生产者写的都是 `str(customer_id)`
|
||
（`profile_generation_service`、投顾线 `profile_governance_service` 与
|
||
`risk_questionnaire_service` 三处皆然）。不放宽则每个事件必然失败、重试 5 次后进死信。
|
||
2. **不可投影的 `memory_key` 跳过而非整批报错**。受控词表
|
||
（`app/service/memory_taxonomy.py`）共 13 个键，其中 `constraint:*`（3 个）与
|
||
`profile:*`（4 个）不以 `preference:`/`goal:` 开头。原实现对第一个不合规的键直接
|
||
`raise`，一条 `constraint:` 记忆就会毒死该客户整批同步；现改为跳过并留痕，
|
||
使"能投影的照常写入"。
|
||
"""
|
||
|
||
import logging
|
||
from collections.abc import Awaitable, Callable
|
||
from datetime import UTC, datetime
|
||
from typing import Any, Protocol
|
||
from uuid import UUID
|
||
|
||
from app.core.conversation_privacy import sanitize_customer_service_message
|
||
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
|
||
|
||
#: 可投影的键前缀。其余受控键（`constraint:*` / `profile:*`）属结构化字段，
|
||
#: 与向量召回不是同一用途，因此不进长期记忆向量集合。
|
||
PROJECTABLE_KEY_PREFIXES: tuple[str, ...] = ("preference:", "goal:")
|
||
|
||
|
||
class MilvusProfileClient(Protocol):
|
||
async def query(self, **kwargs: Any) -> list[dict[str, Any]]: ...
|
||
|
||
async def upsert(self, **kwargs: Any) -> Any: ...
|
||
|
||
|
||
EmbeddingProvider = Callable[[str], Awaitable[list[float]]]
|
||
|
||
|
||
class MilvusProfileProjection:
|
||
"""按记忆 UUID 幂等写入长期记忆向量。"""
|
||
|
||
def __init__(
|
||
self,
|
||
client: MilvusProfileClient,
|
||
embed: EmbeddingProvider,
|
||
*,
|
||
collection: str = PROFILE_COLLECTION,
|
||
) -> None:
|
||
self._client = client
|
||
self._embed = embed
|
||
self._collection = collection
|
||
|
||
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)
|
||
rows: list[dict[str, Any]] = []
|
||
for source in sources:
|
||
vector = await self._embed(source["content"])
|
||
if len(vector) != VECTOR_DIM:
|
||
raise RecoverableAgentError("画像向量维度不一致")
|
||
existing = await self._client.query(
|
||
collection_name=self._collection,
|
||
filter=f'memory_uuid == "{source["memory_uuid"]}"',
|
||
output_fields=["memory_uuid", "version", "customer_id"],
|
||
)
|
||
# 幂等：库里已有更新版本时不回退覆盖。
|
||
if existing and int(existing[0].get("version", 0)) > source["version"]:
|
||
continue
|
||
rows.append({
|
||
"memory_uuid": source["memory_uuid"],
|
||
"customer_id": customer_id,
|
||
"content": source["content"],
|
||
"embedding": vector,
|
||
"memory_type": source["memory_type"],
|
||
"memory_key": source["memory_key"],
|
||
"confidence": source["confidence"],
|
||
"version": source["version"],
|
||
"status": "active",
|
||
"valid_until_ts": source["valid_until_ts"],
|
||
"updated_at_ts": source["updated_at_ts"],
|
||
})
|
||
if rows:
|
||
await self._client.upsert(collection_name=self._collection, data=rows)
|
||
# 全部被跳过时也要留一条记录：否则"零写入"与"没跑"在日志上无法区分。
|
||
logger.info(
|
||
"milvus profile projection: customer_id=%s profile_version=%s written=%s",
|
||
customer_id,
|
||
profile_version,
|
||
len(rows),
|
||
)
|
||
|
||
@staticmethod
|
||
def _coerce_customer_id(raw: Any) -> int:
|
||
"""接受 int 或数字字符串。
|
||
|
||
生产端统一写 `str(customer_id)`（本仓三处生产者皆然），若严格要求 int，
|
||
所有事件都会失败。仅接受**纯数字**字符串，非数字一律拒绝，
|
||
避免把 uuid 之类当成客户号写进向量库。
|
||
"""
|
||
if isinstance(raw, bool):
|
||
raise ValueError("customer_id is invalid")
|
||
if isinstance(raw, int):
|
||
value = raw
|
||
elif isinstance(raw, str) and raw.strip().isdigit():
|
||
value = int(raw.strip())
|
||
else:
|
||
raise ValueError("customer_id is invalid")
|
||
if value <= 0:
|
||
raise ValueError("customer_id is invalid")
|
||
return value
|
||
|
||
@staticmethod
|
||
def _coerce_profile_version(payload: dict[str, Any]) -> int:
|
||
"""兼容两种键名：`profile_version`（本适配器契约）与 `version`（本仓生产端）。"""
|
||
raw = payload.get("profile_version", payload.get("version"))
|
||
if isinstance(raw, bool) or not isinstance(raw, int) or raw <= 0:
|
||
raise ValueError("profile_version is invalid")
|
||
return raw
|
||
|
||
@classmethod
|
||
def _normalize(
|
||
cls,
|
||
payload: dict[str, Any],
|
||
) -> tuple[int, int, list[dict[str, Any]]]:
|
||
customer_id = cls._coerce_customer_id(payload.get("customer_id"))
|
||
profile_version = cls._coerce_profile_version(payload)
|
||
sources = payload.get("memory_sources")
|
||
if not isinstance(sources, list):
|
||
raise ValueError("memory_sources is invalid")
|
||
normalized: list[dict[str, Any]] = []
|
||
skipped: list[str] = []
|
||
now = int(datetime.now(UTC).timestamp())
|
||
for source in sources:
|
||
if not isinstance(source, dict):
|
||
raise ValueError("memory source is invalid")
|
||
required = [
|
||
source.get(name)
|
||
for name in ("memory_uuid", "memory_key", "content", "memory_type")
|
||
]
|
||
if not all(isinstance(value, str) and value.strip() for value in required):
|
||
raise ValueError("memory source fields are invalid")
|
||
try:
|
||
memory_uuid = str(UUID(str(source["memory_uuid"])))
|
||
except ValueError as exc:
|
||
raise ValueError("memory_uuid is invalid") from exc
|
||
memory_key = str(source["memory_key"]).strip()
|
||
if not memory_key.startswith(PROJECTABLE_KEY_PREFIXES):
|
||
# 跳过而非整批失败：`constraint:*` / `profile:*` 是结构化事实，
|
||
# 不进向量召回。一条不可投影的键不得毒死同一客户其余记忆。
|
||
skipped.append(memory_key)
|
||
continue
|
||
confidence = source.get("confidence")
|
||
version = source.get("version")
|
||
if not isinstance(confidence, (int, float)) or not 0 <= confidence <= 1:
|
||
raise ValueError("memory confidence is invalid")
|
||
if not isinstance(version, int) or version <= 0:
|
||
raise ValueError("memory version is invalid")
|
||
valid_until = source.get("valid_until")
|
||
valid_until_ts = None
|
||
if isinstance(valid_until, str) and valid_until:
|
||
try:
|
||
valid_until_ts = int(datetime.fromisoformat(valid_until).timestamp())
|
||
except ValueError as exc:
|
||
raise ValueError("memory valid_until is invalid") from exc
|
||
normalized.append({
|
||
"memory_uuid": memory_uuid,
|
||
"memory_key": memory_key,
|
||
"content": sanitize_customer_service_message(str(source["content"])).strip(),
|
||
"memory_type": str(source["memory_type"]).strip(),
|
||
"confidence": float(confidence),
|
||
"version": version,
|
||
"valid_until_ts": valid_until_ts,
|
||
"updated_at_ts": now,
|
||
})
|
||
if skipped:
|
||
logger.info(
|
||
"milvus profile projection: skipped %s non-projectable memory keys %s",
|
||
len(skipped),
|
||
sorted(set(skipped)),
|
||
)
|
||
return customer_id, profile_version, normalized
|