192 lines
8.4 KiB
Python
192 lines
8.4 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__)
|
|||
|
|
|
|||
|
|
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)
|
|||
|
|
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
|