From 8a0cbab636dd661cdd8a534f8c6c863abb93d016 Mon Sep 17 00:00:00 2001 From: Windows <19353512109@163.com> Date: Sat, 12 Sep 2026 10:45:40 +0800 Subject: [PATCH] =?UTF-8?q?fix(memory-projection):=20=E8=AE=A2=E6=AD=A3=20?= =?UTF-8?q?outbox=20=E5=8F=96=E5=80=BC=E5=8F=A3=E5=BE=84=E5=B9=B6=E6=8E=A5?= =?UTF-8?q?=E9=80=9A=E7=94=BB=E5=83=8F=E6=8A=95=E5=BD=B1=E9=93=BE=E8=B7=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 背景：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，会被消费至死信，待架构师确认是否投影。 --- AGENTS.md | 37 ++- app/core/conversation_privacy.py | 32 +++ .../milvus_profile_projection.py | 191 +++++++++++++++ .../milvus_profile_vector_client.py | 65 +++++ app/repository/profile_repository.py | 18 ++ app/service/agent/bootstrap.py | 21 ++ app/service/profile_generation_service.py | 54 ++++- app/worker/memory_sync_outbox_worker.py | 100 ++++++++ app/worker/runtime.py | 101 ++++++++ docs/32-记忆投影链路实现说明.md | 229 ++++++++++++++++++ .../test_milvus_profile_projection.py | 201 +++++++++++++++ .../test_profile_generation_service.py | 68 +++++- .../worker/test_memory_sync_outbox_worker.py | 137 +++++++++++ tools/normalize_memory_sync_outbox.py | 91 +++++++ tools/setup_milvus_profile_collection.py | 144 +++++++++++ 15 files changed, 1477 insertions(+), 12 deletions(-) create mode 100644 app/core/conversation_privacy.py create mode 100644 app/infrastructure/milvus_profile_projection.py create mode 100644 app/infrastructure/milvus_profile_vector_client.py create mode 100644 app/worker/memory_sync_outbox_worker.py create mode 100644 docs/32-记忆投影链路实现说明.md create mode 100644 tests/unit/infrastructure/test_milvus_profile_projection.py create mode 100644 tests/unit/worker/test_memory_sync_outbox_worker.py create mode 100644 tools/normalize_memory_sync_outbox.py create mode 100644 tools/setup_milvus_profile_collection.py diff --git a/AGENTS.md b/AGENTS.md index 2fb0b42..1c14b00 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -72,6 +72,24 @@ - 解释器：本机用 **`.\.venv\Scripts\python.exe`**；架构师环境用 `D:\conda\envs\jr_py313\python.exe`。 两者等价，**各用本机可用的那个**（`.venv` 被 `.gitignore` 忽略、不进仓库，不存在"需要统一"的问题）。 +- ⚠️ **Windows 控制台跑测试请加 `-X utf8`，否则中文输出是乱码**： + + ```powershell + .\.venv\Scripts\python.exe -X utf8 -m pytest tests -q + ``` + + 原因：本机控制台是代码页 **936（GBK）**，Python 的 `sys.stdout.encoding` 随之为 `gbk`， + 而终端按 UTF-8 解码 ⇒ 中文（断言消息、`docs/` 中文文件名、测试内 `print`）全部乱码。 + **乱码只影响"显示"，不影响测试结果**（`passed`/`failed` 计数是英文，照常可信）。 + + 为什么必须用 `-X utf8` 而不是别的办法： + - `chcp 65001` **单独无效**（Python 仍以 GBK 输出，终端按 UTF-8 解码，反而更乱）； + - 在 `tests/conftest.py` 里 `sys.stdout.reconfigure(encoding="utf-8")` **只能修测试内部的 `print`**， + **修不了 pytest 自己的输出**（失败摘要、短测试摘要）—— pytest 的终端写入器在启动时就绑定了编码， + 早于 conftest 被加载。**实测两组对照：摘要照样乱码。** + - `-X utf8` 在解释器启动时生效，早于 pytest 的一切，且只影响当次进程、无副作用。 + - 等价替代：环境变量 `PYTHONUTF8=1` 或 `PYTHONIOENCODING=utf-8`（PyCharm 用户可写进 Run Configuration）。 + 同一原因，**跑 `tools/*.py` 等脚本时也要加 `-X utf8`**（例如 `python -X utf8 tools\check_authoritative_docs.py`）。 - 数据库现为 **69 张表**（含 `alembic_version`）＝ **68 张业务表** = **场内 51 + 场外/推广 17**。 后 17 张（`offsite_*` / `promotion_*`）**不进 `docs/00` 基线**（规则 8：场外基金运营流程独立）， 逐表登记见 `docs/28-场外与推广域数据表登记.md`。核验命令：`python tools/audit_schema.py`。 @@ -86,13 +104,22 @@ （`app/core/knowledge_schema.py`）——**不要在任何地方硬编码字段名**，那会把另一套环境打挂。 - ⚠️ **Docker Desktop 不会常驻**：它没运行时 Milvus 不可用（`docker` CLI 报连不上守护进程）。 跑真机验证前先确认 Docker Desktop 在运行。 -- 测试基线：`3 failed, 1219 passed, 2 skipped`（2026-09-11 合并主干后实测）。 - 三个失败**都不是代码缺陷**，接手时不要"修"它们： - ① `tests/unit/repository/test_fund_readonly_contract.py`（**底座既有缺陷，不要修也不要报**）； - ② ③ `tests/unit/service/test_offsite_document_recognition_adapter.py` 的 2 个用例 —— **环境相关**： +- ⚠️ **`memory_sync_outbox` 的取值必须是小写英文**（`milvus`/`neo4j`、`upsert`、 + `pending`/`failed`/`processed`/`dead`）。`docs/00` §6.4.6 那一栏曾写作大写 + `MILVUS`/`NEO4J`、`UPSERT` + 中文 `待处理`，**与全仓实现从未对齐，照它写会静默失效**： + 消费端按 `handlers.get(target_store)` 分派、且只领 `status in {"pending","failed"}`， + 大写 + 中文两个条件都不满足 ⇒ **事件任何消费者都领不到、永久滞留且不报错** + （唯一键 `(event_uuid, target_store)` 对大小写无约束，MySQL 也不报错）。 + 取值口径以**主干既有读取方**为准（`projection_reconciliation_service.py`、 + `graph_projection_worker.py`），不是文档。详见 `docs/32-记忆投影链路实现说明.md`。 +- 测试基线：`2 failed, 1307 passed, 2 skipped`（2026-09-12 记忆投影链路落地后实测； + 此前为 `3 failed, 1219 passed`——第 3 个失败 + `tests/unit/repository/test_fund_readonly_contract.py` 已由投顾线合入主干时修复，**不要再当既有缺陷引用**）。 + 剩下 2 个失败**都不是代码缺陷**，接手时不要"修"它们： + `tests/unit/service/test_offsite_document_recognition_adapter.py` 的 2 个用例 —— **环境相关**： 它们断言请求体里是中文原文，而 httpx 会把中文序列化成 `\uXXXX`，字节序列自然不匹配。 功能无影响；若要修，正确做法是断言 `json.loads(body)` 后的字段值（字节级断言不该用来测 JSON）。 -- **mypy：`mypy app` → `Success: no issues found in 180 source files`（0 错）。** +- **mypy：`mypy app` → `Success: no issues found in 227 source files`（0 错）。** ⚠️ 曾在本机报 184 个错，**已查明是环境版本旧**，与代码质量无关 —— 复现矩阵： | SQLAlchemy | mypy | 报错数 | diff --git a/app/core/conversation_privacy.py b/app/core/conversation_privacy.py new file mode 100644 index 0000000..abb97b8 --- /dev/null +++ b/app/core/conversation_privacy.py @@ -0,0 +1,32 @@ +"""客服会话落库前的敏感凭据最小化处理。 + +来源：同事 `ZSY_develop` 分支（`app/core/conversation_privacy.py`），整文件移植，未改语义。 +纯正则、无外部依赖，供记忆投影（`MilvusProfileProjection`）在写入向量前做最后一道脱敏。 + +为什么要单独抽一层：记忆内容最终会进入 Milvus 与 Neo4j，一旦写入就脱离了会话事务的 +管控范围。把脱敏放在**写入适配器内部**（而不是依赖调用方记得做），是为了让"未经脱敏的 +文本不得落外部存储"成为代码保证，而不是流程约定。 +""" + +import re + +# 替换顺序从带业务语义的凭据开始，避免通用数字规则先破坏上下文。 +_SENSITIVE_PATTERNS: tuple[tuple[re.Pattern[str], str], ...] = ( + ( + re.compile(r"(?i)((?:登录|交易)?密码)\s*(?:[:：=]|是)\s*[^\s，。；,;]{1,64}"), + r"\1[已隐藏]", + ), + (re.compile(r"(?i)((?:登录|交易)?密码)\s*\d{4,32}"), r"\1[已隐藏]"), + (re.compile(r"(?i)(验证码|短信码|校验码)\s*(?:[:：=]|是)?\s*\d{4,8}"), r"\1[已隐藏]"), + (re.compile(r"(? str: + """保留风险关键词，移除不应进入会话、Outbox 或后续 Redis 的凭据值。""" + sanitized = message + for pattern, replacement in _SENSITIVE_PATTERNS: + sanitized = pattern.sub(replacement, sanitized) + return sanitized diff --git a/app/infrastructure/milvus_profile_projection.py b/app/infrastructure/milvus_profile_projection.py new file mode 100644 index 0000000..fa00efb --- /dev/null +++ b/app/infrastructure/milvus_profile_projection.py @@ -0,0 +1,191 @@ +"""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 diff --git a/app/infrastructure/milvus_profile_vector_client.py b/app/infrastructure/milvus_profile_vector_client.py new file mode 100644 index 0000000..941d756 --- /dev/null +++ b/app/infrastructure/milvus_profile_vector_client.py @@ -0,0 +1,65 @@ +"""长期记忆画像向量的 Milvus 客户端。 + +与 `MilvusKnowledgeWriter` 分开：那个适配器的 `upsert` 签名绑定知识集合的 +`knowledge_id` 主键与字段表，而画像投影的主键是 `memory_uuid`、且需要先 `query` +按版本判重。两者共用一套连接口径（惰性连接 + 失败一律 `RecoverableAgentError`）， +但不硬把两种 schema 塞进同一个类。 + +与召回侧的客户端也分开：写路径不与检索进程共用连接（读写物理隔离，向量库故障 +不能从写路径传染到问答主链路），与 `get_milvus_knowledge_writer` 的取向一致。 +""" + +from typing import Any + +from app.core.errors import RecoverableAgentError + + +class MilvusProfileVectorClient: + """满足 `MilvusProfileProjection` 所需的 `query` / `upsert` / `load_collection`。""" + + def __init__(self, uri: str, token: str = "") -> None: + self._uri = uri + self._token = token + self._client: Any = None + + async def _ensure(self) -> Any: + if self._client is None: + try: + from pymilvus import AsyncMilvusClient # type: ignore[import-untyped] + except ImportError as exc: # pragma: no cover - 依赖已声明，缺装是环境问题 + raise RecoverableAgentError("pymilvus 未安装，无法写入画像向量") from exc + try: + self._client = AsyncMilvusClient(uri=self._uri, token=self._token or None) + except Exception as exc: + raise RecoverableAgentError("Milvus 画像写客户端初始化失败") from exc + return self._client + + async def load_collection(self, *, collection_name: str) -> None: + """把集合载入内存后再查/写。 + + 集合不存在时抛 `RecoverableAgentError`：由 outbox 退避重试并最终判死信， + 而不是静默跳过——"集合没建"是装配问题，必须可见。 + """ + client = await self._ensure() + try: + await client.load_collection(collection_name=collection_name) + except Exception as exc: + raise RecoverableAgentError("画像向量集合不可用") from exc + + async def query(self, **kwargs: Any) -> list[dict[str, Any]]: + """按 filter 查询；返回空列表表示无匹配（不是错误）。""" + client = await self._ensure() + try: + raw = await client.query(**kwargs) + except Exception as exc: + raise RecoverableAgentError("画像向量查询失败") from exc + return list(raw or []) + + async def upsert(self, **kwargs: Any) -> Any: + client = await self._ensure() + try: + return await client.upsert(**kwargs) + except RecoverableAgentError: + raise + except Exception as exc: + raise RecoverableAgentError("画像向量写入失败") from exc diff --git a/app/repository/profile_repository.py b/app/repository/profile_repository.py index 04488d8..f5666f8 100644 --- a/app/repository/profile_repository.py +++ b/app/repository/profile_repository.py @@ -50,6 +50,17 @@ _SQL_NEXT_VERSION = text(""" FROM profile_snapshots WHERE customer_id = :customer_id """) +#: 该客户**有效**的长期记忆（投影到 Milvus 长期记忆向量集合的数据源）。 +#: +#: 只取 `status='active'`：失效/被取代的记忆不应再进入向量召回，否则投顾会召回 +#: 已过期偏好。字段与 `MilvusProfileProjection` 的 `memory_sources` 契约一一对应。 +_SQL_ACTIVE_MEMORIES = text(""" + SELECT memory_uuid, memory_key, content, memory_type, confidence, version, valid_until + FROM memory_unit + WHERE customer_id = :customer_id AND status = 'active' + ORDER BY id +""") + #: 旧画像置非当前（`current_customer_id` 有唯一键，必须先置 0 才能插新的当前版本）。 _SQL_CLEAR_CURRENT = text(""" UPDATE profile_snapshots @@ -88,6 +99,13 @@ class ProfileRepository: row = result.mappings().first() return dict(row) if row is not None else None + async def active_memories(self, customer_id: int) -> list[dict[str, Any]]: + """该客户所有 `status='active'` 的长期记忆，按 id 稳定排序。""" + result = await self._session.execute( + _SQL_ACTIVE_MEMORIES, {"customer_id": customer_id} + ) + return [dict(row) for row in result.mappings().all()] + async def next_version(self, customer_id: int) -> int: return int(await self._session.scalar(_SQL_NEXT_VERSION, {"customer_id": customer_id})) diff --git a/app/service/agent/bootstrap.py b/app/service/agent/bootstrap.py index b71c2c0..85fe555 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_vector_client import MilvusProfileVectorClient from app.infrastructure.vector_memory import VectorMemoryAdapter from app.service.agent.factory import AgentFactory from app.service.agent.governance import PlatformGovernance @@ -159,6 +160,26 @@ def get_milvus_knowledge_writer() -> MilvusKnowledgeWriter | None: return MilvusKnowledgeWriter(uri, settings.milvus_token or "") +@lru_cache(maxsize=1) +def get_milvus_profile_vector_client() -> MilvusProfileVectorClient | None: + """长期记忆画像向量**写**适配器（装配入口）；`milvus_uri` 缺失时返回 None。 + + 与 `get_milvus_knowledge_writer` 同一取向：读写隔离、构造惰性、缺配置显式降级。 + 返回 None 的语义是**显式降级**——`WorkerRuntime` 会因此不注册画像投影 handler， + `memory_sync_outbox` 的事件在库里保持 pending（可观测、可重放），并留一条 warning， + 绝不静默，也绝不伪造同步成功。 + """ + settings = get_settings() + uri = (settings.milvus_uri or "").strip() + if not uri: + logging.getLogger(__name__).warning( + "milvus_uri not configured; profile vector projection disabled and " + "memory_sync_outbox events will stay pending" + ) + return None + return MilvusProfileVectorClient(uri, settings.milvus_token or "") + + async def _embed_text(text: str) -> list[float]: """把文本向量化；端点来自发布配置（task_type=embedding），无端点时失败关闭。""" endpoints = await DatabaseModelEndpointResolver().resolve( diff --git a/app/service/profile_generation_service.py b/app/service/profile_generation_service.py index 17cf61a..6705d3b 100644 --- a/app/service/profile_generation_service.py +++ b/app/service/profile_generation_service.py @@ -44,14 +44,24 @@ from app.core.errors import ValidationAgentError from app.core.profile_projection import PROFILE_FIELD_POLICY from app.repository.profile_repository import ProfileRepository -#: `memory_sync_outbox` 的两个目标存储（`docs/00` §6.4.6 的 `target_store` 取值）。 -TARGET_MILVUS = "MILVUS" -TARGET_NEO4J = "NEO4J" +#: `memory_sync_outbox` 的两个目标存储。 +#: +#: ⚠️ **一律小写**：消费端按 `target_store` 的值分派 handler +#: （`MemorySyncOutboxWorker.handlers.get(event.target_store)`），且 +#: `graph_projection_worker` / `projection_reconciliation_service` 的领取与重放 +#: 都以 `status in {"pending","failed"}` + `"processed"` 为准。全仓（包含投顾线两处生产者） +#: 统一使用小写 `milvus`/`neo4j`、`upsert`、`pending`。 +#: +#: 历史说明：`docs/00` §6.4.6 该栏曾写作大写 `MILVUS`/`NEO4J`、`UPSERT` 与中文 `待处理`， +#: 与上述实现从未对齐；本模块原先照文档写，是**全仓唯一的异类**，导致自己写的事件 +#: 任何消费者都领不到。现统一为小写，并已把历史 2 行就地改齐（主键/唯一键不变）。 +TARGET_MILVUS = "milvus" +TARGET_NEO4J = "neo4j" SYNC_TARGETS: tuple[str, ...] = (TARGET_MILVUS, TARGET_NEO4J) -#: 同步操作与状态取值（与 `memory_sync_outbox` DDL 的语义一致）。 -SYNC_OPERATION_UPSERT = "UPSERT" -SYNC_STATUS_PENDING = "待处理" +#: 同步操作与状态取值。 +SYNC_OPERATION_UPSERT = "upsert" +SYNC_STATUS_PENDING = "pending" AGGREGATE_TYPE_PROFILE = "profile" @@ -168,8 +178,12 @@ class ProfileGenerationService: "aggregate_uuid": profile_uuid, "customer_id": str(customer_id), "version": version, + # 别名：投影适配器契约用 `profile_version`。两个键都写，避免消费端 + # 因生产者用词不同而取不到值（本仓三种 payload 形状的历史遗留）。 + "profile_version": version, "snapshot_hash": snapshot_hash, "snapshot": snapshot, + "memory_sources": await self._memory_sources(customer_id), } for target in SYNC_TARGETS: self._repo.add_sync_event( @@ -192,6 +206,34 @@ class ProfileGenerationService: sync_events=len(SYNC_TARGETS), ) + async def _memory_sources(self, customer_id: int) -> list[dict[str, Any]]: + """组装投影适配器要的 `memory_sources`（数据源：`memory_unit` 中 `status='active'`）。 + + 为什么由生产端组装而不是消费端回查：消费端到时那条记忆可能已改版本， + 让它自己去查会把"投递的是哪一版"变成不确定。事件里带上当时的确定快照， + 投递语义才与 `aggregate_version` 一致（幂等判据也才有意义）。 + + 为什么是 `memory_unit` 而不是 `user_facts`：两者用途不同——`user_facts` 是 + **已确认的结构化事实**，喂画像与图投影；`memory_unit` 是**长期记忆条目**， + 正是长期记忆向量集合要存的东西。这不与"图投影只读 user_facts"的不变式冲突， + 因为 Milvus 存的就是记忆本身的向量，不是画像事实的另一种说法。 + """ + rows = await self._repo.active_memories(customer_id) + return [ + { + "memory_uuid": str(row["memory_uuid"]), + "memory_key": str(row["memory_key"]), + "content": str(row["content"]), + "memory_type": str(row["memory_type"]), + "confidence": float(row["confidence"]), + "version": int(row["version"]), + "valid_until": ( + row["valid_until"].isoformat() if row["valid_until"] else None + ), + } + for row in rows + ] + @staticmethod def _generation_basis(assessment_row: Mapping[str, Any] | None) -> dict[str, object]: """记录这次画像**依据了什么**（`docs/00`：使用的测评版本、交易窗口和记忆版本列表）。 diff --git a/app/worker/memory_sync_outbox_worker.py b/app/worker/memory_sync_outbox_worker.py new file mode 100644 index 0000000..461fddd --- /dev/null +++ b/app/worker/memory_sync_outbox_worker.py @@ -0,0 +1,100 @@ +"""画像投影 Outbox 消费器。 + +外部存储通过 handler 注入；本模块只负责领取、重试、死信和 MySQL 状态更新。 + +来源：同事 `ZSY_develop` 分支（`app/worker/memory_sync_outbox_worker.py`）。 +本文件保留其**领取/重试/退避/死信骨架原样未改**——这部分（`skip_locked` 并发领取、 +指数退避、5 次转死信）是正确的，也是这条链上唯一把"投递可靠性"做对的地方。 + +分派口径：`runtime.py` 是本消费器的装配方，它会把 handler 键注册为**全仓统一的小写** +`milvus` / `neo4j`（与生产端写入的 `target_store` 取值一致）。 +""" + +from collections.abc import Awaitable, Callable +from datetime import UTC, datetime, timedelta +from typing import Any + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.infrastructure.db import SessionFactory +from app.model.memory import MemorySyncOutbox + +ProjectionHandler = Callable[[dict[str, Any]], Awaitable[Any]] +MAX_RETRY_COUNT = 5 + + +class MemorySyncOutboxWorker: + """按目标存储独立消费画像投影事件。""" + + def __init__( + self, + handlers: dict[str, ProjectionHandler], + *, + session_factory: Callable[[], AsyncSession] = SessionFactory, + ) -> None: + self.handlers = handlers + self.session_factory = session_factory + + async def run_once( + self, *, target_store: str | None = None, event_uuid: str | None = None + ) -> bool: + """领取并处理一条到期事件；没有可处理事件时返回 False。""" + if not self.handlers: + return False + async with self.session_factory() as session: + now = datetime.now(UTC).replace(tzinfo=None) + conditions: list[Any] = [ + MemorySyncOutbox.status.in_({"pending", "failed"}), + MemorySyncOutbox.next_retry_at.is_(None) + | (MemorySyncOutbox.next_retry_at <= now), + ] + if target_store is not None: + conditions.append(MemorySyncOutbox.target_store == target_store) + if event_uuid is not None: + conditions.append(MemorySyncOutbox.event_uuid == event_uuid) + event = await session.scalar( + select(MemorySyncOutbox) + .where(*conditions) + .order_by(MemorySyncOutbox.id) + .limit(1) + .with_for_update(skip_locked=True) + ) + if event is None: + await session.rollback() + return False + handler = self.handlers.get(event.target_store) + if handler is None: + # 没有对应 handler 时判死信（不留 pending 空转），并写明原因。 + self._fail(event, "target_handler_not_configured", now, dead=True) + await session.commit() + return True + try: + await handler(event.payload) + except Exception as exc: + self._fail(event, type(exc).__name__, now) + else: + event.status = "processed" + event.processed_at = datetime.now(UTC).replace(tzinfo=None) + event.last_error = None + event.next_retry_at = None + await session.commit() + return True + + @staticmethod + def _fail( + event: MemorySyncOutbox, + reason: str, + now: datetime, + *, + dead: bool = False, + ) -> None: + """写入可重试失败或死信状态，不吞掉失败事实。""" + event.retry_count = int(event.retry_count) + 1 + event.last_error = reason[:500] + event.status = "dead" if dead or event.retry_count >= MAX_RETRY_COUNT else "failed" + event.next_retry_at = ( + None + if event.status == "dead" + else now + timedelta(seconds=min(300, 2 ** int(event.retry_count))) + ) diff --git a/app/worker/runtime.py b/app/worker/runtime.py index c125ed3..abb8394 100644 --- a/app/worker/runtime.py +++ b/app/worker/runtime.py @@ -22,6 +22,7 @@ from app.service.agent.bootstrap import ( get_memory_cache_adapter, get_memory_embedding_service, get_milvus_knowledge_writer, + get_milvus_profile_vector_client, get_model_service, ) from app.service.agent.executor import AgentExecutor @@ -91,6 +92,7 @@ class WorkerRuntime: knowledge_writer: Any = _UNSET, knowledge_embedder: Any = _UNSET, knowledge_endpoint_resolver: Any = _UNSET, + profile_vector_client: Any = _UNSET, ) -> None: self.factory = factory if factory is not None else get_agent_factory() self.settings = settings or get_settings() @@ -138,6 +140,18 @@ class WorkerRuntime: ) # 降级告警只打一次：dispatch 是轮询热路径，每轮一条 warning 会把日志淹掉。 self._knowledge_degraded_logged = False + # 画像投影（`memory_sync_outbox`）消费装配。 + # + # 与 `knowledge_writer` 同一取向：客户端构造**惰性**（不连 Milvus）， + # `milvus_uri` 未配置时显式降级为不注册 handler（事件留 pending、可观测、可重放）， + # 绝不伪造同步成功。 + self.profile_vector_client = ( + get_milvus_profile_vector_client() + if profile_vector_client is _UNSET + else profile_vector_client + ) + self.profile_endpoint_resolver = self.knowledge_endpoint_resolver + self._profile_degraded_logged = False # episode 聚合是低频批处理，按轮次节流而不是每轮都查。 self._episode_rounds = 0 @@ -408,6 +422,14 @@ class WorkerRuntime: async def run_once(self) -> bool: dispatched = await self.dispatch_batch() > 0 + # 画像投影消费：与领域事件同一轮次内处理。失败只告警，不影响 run 的处理与 + # 轮询节奏——事件仍在库里，下一轮照常重试（退避由 worker 自己记在 next_retry_at）。 + try: + projected = await self.consume_profile_projections() > 0 + except Exception: + logger.warning("profile projection consumption failed", exc_info=True) + projected = False + dispatched = dispatched or projected self._episode_rounds += 1 if self._episode_rounds % EPISODE_INTERVAL_ROUNDS == 0: # 会话片段聚合：内部幂等（content_hash 唯一键），失败只告警， @@ -431,6 +453,85 @@ class WorkerRuntime: return dispatched return await self.execute(run_id) or dispatched + async def consume_profile_projections(self, *, limit: int = 20) -> int: + """消费 `memory_sync_outbox` 的画像投影事件，返回本次处理条数。 + + 两个目标存储的分工（**方案 A**：以架构师主干为主线，不引入第二套 Neo4j 投影）： + + - `milvus` → 写入长期记忆向量集合 `user_long_term_memory_v1` + （此前**完全没有消费者**，事件永久滞留）； + - `neo4j` → 复用主干 `ProfileGraphProjectionService`。主干已由 + `profile.rebuild_requested` 事件驱动同一条链，图投影是 `MERGE` 幂等的， + 因此这里再投一次不产生重复节点/关系，只用于把 outbox 行的投递状态收敛掉。 + + 为什么不让 handler 自己 commit：事务边界与 `dispatch_batch` 一致， + 由本方法按条提交；单条失败由 `MemorySyncOutboxWorker` 内部转成 + `failed`+退避或死信，不冒泡打断本轮其余事件。 + """ + if self.profile_vector_client is None: + # 显式降级：不注册 handler 就交给 worker 判死信是**错的**（那是把配置缺失 + # 伪装成投递失败）。这里直接不消费，事件保持 pending，由启动日志提示。 + if not self._profile_degraded_logged: + self._profile_degraded_logged = True + logger.warning( + "profile projection disabled: milvus profile vector client unavailable; " + "memory_sync_outbox events stay pending" + ) + return 0 + # 收窄到局部变量：闭包内访问 self 属性时 mypy 无法保留上面的 None 判定。 + vector_client = self.profile_vector_client + + from app.infrastructure.milvus_profile_projection import MilvusProfileProjection + from app.worker.memory_sync_outbox_worker import MemorySyncOutboxWorker + + async def project_milvus(payload: dict[str, Any]) -> None: + projection = MilvusProfileProjection(vector_client, self._profile_embed) + await projection.upsert(payload) + + async def project_neo4j(payload: dict[str, Any]) -> None: + raw_customer_id = payload.get("customer_id") + # 显式 isinstance 而不是 `in (None, "")`：后者不做类型收窄，mypy 无法确认 + # int() 的入参类型；同时也把"客户号必须是数字"这一契约写在类型检查里。 + if not isinstance(raw_customer_id, (int, str)) or raw_customer_id == "": + raise RecoverableAgentError("profile projection payload has no customer_id") + customer_id = int(raw_customer_id) + # 延迟导入：与 dispatch_profile_rebuild 同一理由，避免模块级循环依赖。 + from app.service.profile_graph_projection_service import ( + ProfileGraphProjectionService, + ) + + async with SessionFactory() as session: + outcome = await ProfileGraphProjectionService( + session, self.relationships + ).project_customer(customer_id) + if outcome.degraded: + # 图库不可用：如实抛出，让 worker 走失败/退避，而不是记成已投递。 + raise RecoverableAgentError(f"graph projection degraded: {outcome.reason}") + + worker = MemorySyncOutboxWorker( + {"milvus": project_milvus, "neo4j": project_neo4j} + ) + handled = 0 + for _ in range(max(1, limit)): + if not await worker.run_once(): + break + handled += 1 + return handled + + async def _profile_embed(self, text: str) -> list[float]: + """向量化一条记忆正文；端点走与知识向量化同一套已批准端点解析。 + + 不复用 `bootstrap._embed_text`：那是模块私有函数，跨模块引用私有名会把 + 两处的耦合藏起来。这里用同一组公开装配（端点解析器 + embedding 服务）。 + """ + endpoints = await self.profile_endpoint_resolver.resolve( + agent_type="memory_recall", task_type="embedding" + ) + if not endpoints: + raise RecoverableAgentError("没有可用的 embedding 端点，无法投影长期记忆") + execution = await self.knowledge_embedder.embed(endpoints, text) + return list(execution.vector) + async def aggregate_episodes(self, *, customer_limit: int = 50) -> int: """把已静默的会话片段聚合为 episode，返回新写入的片段数。 diff --git a/docs/32-记忆投影链路实现说明.md b/docs/32-记忆投影链路实现说明.md new file mode 100644 index 0000000..9356ac8 --- /dev/null +++ b/docs/32-记忆投影链路实现说明.md @@ -0,0 +1,229 @@ +# 记忆投影链路（Outbox → Milvus 长期记忆 / Neo4j）实现说明 + +**适用分支**：`NL_develop`（用户端线）｜**日期**：2026-09-12｜**状态**：已实现并通过真机验证 + +--- + +## 1. 这条链路是干什么的 + +记忆（对话里被抽取出来的长期事实）要能被后续召回，必须从 MySQL 同步到两处外部存储： + +``` +对话消息 + └─→ memory.extraction_requested（domain_event_outbox） + └─→ MemoryExtractionWorker → MemoryService.upsert() → memory_unit（MySQL，唯一真相） + └─→ profile.rebuild_requested（domain_event_outbox） + ├─→ ProfileAssemblyService.rebuild() → profile_snapshots + ├─→ ProfileGraphProjectionService → Neo4j（图） + └─→ ProfileGenerationService.generate() → memory_sync_outbox + ├─ milvus → MilvusProfileProjection → Milvus 长期记忆向量 + └─ neo4j → ProfileGraphProjectionService（幂等复投） +``` + +`memory_sync_outbox` 是"画像版本 → 外部存储"的投递队列，唯一键 +`uk_memory_sync_event (event_uuid, target_store)`：**同一 `event_uuid` 对两个目标库各写一条**。 + +--- + +## 2. ⚠️ 最容易踩的坑：枚举取值必须全仓统一（小写 + 英文） + +这条链曾**完全失效但不报错**，根因就是取值口径不统一。 + +### 唯一正确口径 + +| 字段 | 取值 | 大小写 | +|---|---|---| +| `target_store` | `milvus` / `neo4j` | **小写** | +| `operation` | `upsert` / `archive` / `delete` | **小写** | +| `status` | `pending` / `failed` / `processed` / `dead` | **小写英文** | + +（`aggregate_type` 用 `memory` / `profile` / `relationship` / `deletion`。） + +### 为什么照 `docs/00` §6.4.6 写会坏 + +`docs/00` §6.4.6 那一栏曾写作大写 `MILVUS`/`NEO4J`、`UPSERT` 与中文 `待处理`， +与**全仓实现从未对齐**。按那份文档写会造成： + +1. `MemorySyncOutboxWorker` 按 `handlers.get(event.target_store)` 分派 handler + —— 大写值找不到 handler； +2. 领取条件是 `status.in_({"pending","failed"})` —— 中文 `待处理` 不满足。 + +⇒ **两个条件都不满足，事件任何消费者都领不到，永久滞留且不报错。** +唯一键 `(event_uuid, target_store)` 对大小写没有约束，MySQL 也不会报错，所以是**静默失效**。 + +判断依据应以**主干既有读取方**为准，不是文档： +`projection_reconciliation_service.py:20`（`{"pending","failed"}`）、 +`graph_projection_worker.py`（`pending`/`processed`/`dead`）。 + +### 现状 + +- 生产端取值集中在 `app/service/profile_generation_service.py` 的常量 + （`TARGET_MILVUS`/`TARGET_NEO4J`/`SYNC_OPERATION_UPSERT`/`SYNC_STATUS_PENDING`）； +- **测试不再硬编码字面量**，并断言 `SYNC_STATUS_PENDING == "pending"`、 + `set(SYNC_TARGETS) == {"milvus","neo4j"}` + （硬编码正是当初跑偏的直接原因）； +- `tests/unit/worker/test_memory_sync_outbox_worker.py` 有一条契约回归测试， + 断言大写 `MILVUS` 分派不到 handler、会进死信 —— 谁改回大写，测试立刻红； +- 历史 2 行已就地把取值改齐（只改值，主键/唯一键/payload 未动）， + 正式订正工具 `tools/normalize_memory_sync_outbox.py` + （**默认 dry-run，加 `--apply` 才写库**，幂等可复跑；换环境若也有同批旧值可直接用）。 + +--- + +## 3. Milvus 长期记忆向量集合 + +| 项 | 值 | +|---|---| +| 集合名 | `user_long_term_memory_v1` | +| 主键 | `memory_uuid`（VARCHAR 64，UUID 字符串，**按 UUID 幂等 upsert**） | +| 向量 | `embedding`，`FLOAT_VECTOR` dim **1024**，索引 `AUTOINDEX` + `COSINE` | +| 其他字段 | `customer_id`(INT64) / `version`(INT64,可空) / `valid_until_ts`(INT64,可空) / `updated_at_ts`(INT64,可空) / `confidence`(DOUBLE) / `content`(VARCHAR 2048) / `memory_type`(VARCHAR 32) / `memory_key`(VARCHAR 64) / `status`(VARCHAR 16) | + +建集合：`python tools/setup_milvus_profile_collection.py` +——**幂等**，集合已存在时只做结构比对报告、不覆盖不删重建（共享 Milvus 实例里还有别的项目的集合）。 + +### `memory_sources` 契约（适配器的输入） + +```json +{ + "customer_id": 9102, + "profile_version": 2, + "memory_sources": [{ + "memory_uuid": "uuid 字符串", + "memory_key": "preference:risk_level", + "content": "正文", + "memory_type": "preference", + "confidence": 0.9, + "version": 1, + "valid_until": null + }] +} +``` + +数据源：`memory_unit` 中 `status='active'` 的行 +（`ProfileRepository.active_memories()`）。**由生产端组装**而不是消费端回查： +消费端到时记忆可能已改版本，事件里带确定快照才能与 `aggregate_version` 语义一致。 + +> 为什么用 `memory_unit` 而不是 `user_facts`：两者用途不同 —— `user_facts` 是 +> **已确认的结构化事实**（喂画像与图投影），`memory_unit` 是**长期记忆条目** +> （正是长期记忆向量集合要存的东西）。这不与"图投影只读 `user_facts`"的不变式冲突。 + +--- + +## 4. 移植自 `ZSY_develop` 的改动（含两处契约放宽） + +来源：同事 `ZSY_develop` 分支。整文件移植、骨架未改的部分： +`app/core/conversation_privacy.py`、`app/worker/memory_sync_outbox_worker.py` +（领取/`skip_locked`/指数退避/5 次转死信的可靠性骨架原样保留）、 +`app/infrastructure/milvus_profile_projection.py`。 + +对适配器做了**两处契约放宽**，都是"避免整批失败"，不改变写入语义： + +1. **`customer_id` 接受 int 或数字字符串** + 原实现要求 `isinstance(customer_id, int)`，而本仓**所有**生产者写的都是 + `str(customer_id)`。不放宽则**每个事件必然失败**、重试 5 次后进死信。 + 放宽仅限**纯数字**串，非数字（如 uuid）仍拒绝。 +2. **不可投影的 `memory_key` 跳过而非整批报错** + 受控词表 `app/service/memory_taxonomy.py` 共 13 个键，其中 `constraint:*`（3 个）与 + `profile:*`（4 个）不以 `preference:`/`goal:` 开头。原实现对第一个不合规的键直接 + `raise` —— 一条 `constraint:` 记忆就会毒死该客户整批同步。现改为**跳过并留痕**。 + +**收窄而非扩容的理由**：`preference:*`/`goal:*` 是偏好与目标，适合语义召回； +`constraint:*`/`profile:*` 是结构化约束与属性，应由结构化通道查询（`user_facts` → 图）， +放进向量集合只会造成召回噪声。若日后要给它们做语义召回，应扩容而非靠现在的跳过。 + +--- + +## 5. Neo4j 分支：方案 A（不引入第二套投影） + +同事分支另有一套 `neo4j_profile_projection.py`（按客户各建私有 +`Preference`/`Goal` 节点、数据源 `memory_unit`）。**未采用**，理由： + +主干 `ProfileGraphProjectionService` 已由 `profile.rebuild_requested` 驱动同一条链， +其数据源是 `user_facts`（**已确认事实**）、`MERGE` **共享 tag 节点**，且文件头写明不变式： + +> "只投影'已确认'的事实……保证图里的偏好标签与画像口径一致； +> 否则同一件事在画像和图里会有两种说法。" + +两套并存 = 同一事实在图中两种表示，正好违反这条不变式。 +因此 `memory_sync_outbox` 的 `neo4j` 分支**复用主干服务**： +图投影是 `MERGE` 幂等的，再投一次不产生重复节点/关系，只用于收敛 outbox 的投递状态。 + +--- + +## 6. 消费端装配 + +`WorkerRuntime.consume_profile_projections()`（`app/worker/runtime.py`），在 +`run_once()` 每轮执行，按条提交、单条失败不冒泡。 + +- `milvus` handler → `MilvusProfileProjection`（注入 `MilvusProfileVectorClient` + 向量化）； +- `neo4j` handler → `ProfileGraphProjectionService`；其 `degraded` 会**如实抛错** + 走失败/退避，不记成已投递； +- `milvus_uri` 未配置时**显式降级**：不消费、事件留 `pending`（可观测、可重放）， + 启动路径留一条 warning，绝不伪造同步成功（与 `knowledge_writer` 同一取向）； +- 目标存储没有对应 handler 时判**死信**并记 `target_handler_not_configured` + —— 这正是当初大写值事件的下场（有回归测试守着）。 + +向量化复用与知识向量同一套已批准端点解析（`agent_type="memory_recall"`, +`task_type="embedding"`）。 + +--- + +## 7. 真机验证证据（2026-09-12） + +| 验证项 | 结果 | +|---|---| +| 集合创建幂等 | 首次 `created`，复跑 `exists` | +| 真实 embedding 维度 | **1024**（与集合定义一致） | +| 真实写入 + 回读 | 2 条可投影键写入成功并可回读（内容/版本正确） | +| 不可投影键 | `constraint:liquidity` **未写入**（跳过生效，未毒死整批） | +| 测试数据清理 | 已按 `customer_id=999999` 删除，集合残留 **0** 条 | +| 单元测试 | 新增 17 个（适配器 10 + worker 5 + 生产端 2），全过 | +| 全量回归 | `2 failed, 1307 passed, 2 skipped` —— 与基线一致，**无新增失败** | +| mypy | `Success: no issues found in 227 source files` | + +> 全量的 2 个失败是既有环境相关项（`test_offsite_document_recognition_adapter.py` +> 断言请求体里的中文原文，而 httpx 序列化成 `\uXXXX`），与本次改动无关。 + +> 真机注意：Milvus 写入后**短时间内可能查不到**（索引尚未可见）， +> 验证脚本按重试处理；同理删除后立即查询可能仍返回旧行，需重查确认。 + +--- + +## 8. 尚未完成 / 依赖他人 + +1. **常驻 Worker 未运行**：核对时数据库里 `memory_sync_outbox` 2 行、 + `domain_event_outbox` 积压 + （`agent.run_requested` 428、`profile.rebuild_requested` 216、 + `agent.run_completed` 61、`memory.extraction_requested` 58）。 + **`memory_unit` 与 `user_facts` 目前都是 0 行** —— 代码链路是通的， + 但没有 Worker 在跑，所以记忆永远不会被抽取出来。起 + `python -m app.worker --once`（或常驻）即会开始消费。 +2. **投顾线两处生产者的取值**：`profile_governance_service.py` 与 + `risk_questionnaire_service.py` 已写小写 `milvus`/`neo4j` + `pending`（与本文口径一致）， + 但它们的 `payload` 是 `{customer_id, profile_uuid, version, profile}`， + **没有 `memory_sources`** → 投影时会因 `memory_sources is invalid` 失败重试至死信。 + 需投顾线补 `memory_sources` 或明确这两个来源是否也要投影长期记忆。 + （属架构师线的改动，本线未动。） +3. `docs/00` §6.4.6 的取值栏与实现不一致：按评审要求**未改基线文档**， + 实际口径以本文第 2 节为准。 + +--- + +## 9. 相关文件 + +**新增** +- `app/core/conversation_privacy.py`（移植） +- `app/infrastructure/milvus_profile_projection.py`（移植 + 2 处放宽） +- `app/infrastructure/milvus_profile_vector_client.py` +- `app/worker/memory_sync_outbox_worker.py`（移植） +- `tools/setup_milvus_profile_collection.py` +- `tests/unit/infrastructure/test_milvus_profile_projection.py` +- `tests/unit/worker/test_memory_sync_outbox_worker.py` + +**修改** +- `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()`） +- `app/worker/runtime.py`（`consume_profile_projections()` + `run_once` 接线） +- `tests/unit/service/test_profile_generation_service.py`（+2 用例、断言改引用常量） diff --git a/tests/unit/infrastructure/test_milvus_profile_projection.py b/tests/unit/infrastructure/test_milvus_profile_projection.py new file mode 100644 index 0000000..836e46e --- /dev/null +++ b/tests/unit/infrastructure/test_milvus_profile_projection.py @@ -0,0 +1,201 @@ +"""`MilvusProfileProjection` 的定向测试。 + +前 4 个用例移植自同事 `ZSY_develop` 的 +`tests/unit/infrastructure/test_milvus_profile_projection.py`； +后 4 个覆盖本仓对其做的**两处契约放宽**（`customer_id` 兼容字符串、 +不可投影键跳过而非整批失败）与脱敏，这些是移植时必须钉住的差异点。 +""" + +from uuid import uuid4 + +import pytest + +from app.core.errors import RecoverableAgentError +from app.infrastructure.milvus_profile_projection import MilvusProfileProjection + + +class FakeMilvus: + def __init__(self, existing: list[dict[str, object]] | None = None) -> None: + self.existing = existing or [] + self.queries: list[dict[str, object]] = [] + self.upserts: list[dict[str, object]] = [] + + async def query(self, **kwargs: object) -> list[dict[str, object]]: + self.queries.append(kwargs) + return self.existing + + async def upsert(self, **kwargs: object) -> None: + self.upserts.append(kwargs) + + +def payload() -> dict[str, object]: + return { + "customer_id": 7, + "profile_version": 1, + "memory_sources": [{ + "memory_uuid": str(uuid4()), + "memory_key": "preference:risk_level", + "content": "稳健型", + "memory_type": "preference", + "confidence": 0.9, + "version": 2, + "valid_until": None, + }], + } + + +def _source(data: dict[str, object]) -> dict[str, object]: + sources = data["memory_sources"] + assert isinstance(sources, list) + source = sources[0] + assert isinstance(source, dict) + return source + + +@pytest.mark.asyncio +async def test_upsert_writes_schema_fields_and_vector() -> None: + client = FakeMilvus() + projection = MilvusProfileProjection(client, _embed) + + await projection.upsert(payload()) + + assert len(client.upserts) == 1 + row = client.upserts[0]["data"][0] + assert row["customer_id"] == 7 + assert row["status"] == "active" + assert len(row["embedding"]) == 1024 + + +@pytest.mark.asyncio +async def test_lower_memory_version_is_not_overwritten() -> None: + data = payload() + memory_uuid = _source(data)["memory_uuid"] + client = FakeMilvus(existing=[{ + "memory_uuid": memory_uuid, "customer_id": 7, "version": 3, + }]) + + await MilvusProfileProjection(client, _embed).upsert(data) + + assert client.upserts == [] + + +@pytest.mark.asyncio +async def test_embedding_dimension_is_enforced() -> None: + with pytest.raises(RecoverableAgentError, match="维度"): + await MilvusProfileProjection(client=FakeMilvus(), embed=_embed_short).upsert( + payload() + ) + + +@pytest.mark.asyncio +async def test_non_uuid_memory_id_is_rejected() -> None: + data = payload() + _source(data)["memory_uuid"] = "unsafe\" or true" + + with pytest.raises(ValueError, match="memory_uuid"): + await MilvusProfileProjection(FakeMilvus(), _embed).upsert(data) + + +# --- 本仓放宽的契约（移植差异点） ------------------------------------------- + + +@pytest.mark.asyncio +async def test_string_customer_id_is_accepted() -> None: + """本仓三处生产者写的都是 `str(customer_id)`；不接受字符串则事件必然全部失败。""" + data = payload() + data["customer_id"] = "9102" + + client = FakeMilvus() + await MilvusProfileProjection(client, _embed).upsert(data) + + assert client.upserts[0]["data"][0]["customer_id"] == 9102 + + +@pytest.mark.asyncio +async def test_non_numeric_customer_id_is_rejected() -> None: + """放宽不等于不校验：uuid 之类的非数字串必须拒绝，不能当成客户号写进向量库。""" + data = payload() + data["customer_id"] = "957c0552-7fa2-4f2a-924d-d2d2e133b245" + + with pytest.raises(ValueError, match="customer_id"): + await MilvusProfileProjection(FakeMilvus(), _embed).upsert(data) + + +@pytest.mark.asyncio +async def test_version_key_fallback_is_supported() -> None: + """本仓生产端 payload 用 `version`；适配器契约用 `profile_version`。两个都要认。""" + data = payload() + del data["profile_version"] + data["version"] = 5 + + client = FakeMilvus() + await MilvusProfileProjection(client, _embed).upsert(data) + + assert len(client.upserts) == 1 + + +@pytest.mark.asyncio +async def test_non_projectable_memory_key_is_skipped_not_fatal() -> None: + """`constraint:*` / `profile:*` 不在可投影前缀内。 + + 关键：一条不可投影的键**不得**毒死同一客户其余可投影记忆。 + """ + data = payload() + good = _source(data) + data["memory_sources"] = [ + { + "memory_uuid": str(uuid4()), + "memory_key": "constraint:liquidity", + "content": "半年内需要流动性", + "memory_type": "constraint", + "confidence": 0.8, + "version": 1, + "valid_until": None, + }, + good, + ] + + client = FakeMilvus() + await MilvusProfileProjection(client, _embed).upsert(data) + + assert len(client.upserts) == 1 + rows = client.upserts[0]["data"] + assert [row["memory_key"] for row in rows] == ["preference:risk_level"] + + +@pytest.mark.asyncio +async def test_all_keys_non_projectable_writes_nothing() -> None: + """全部不可投影时不写 Milvus，但也不报错（不是失败，是无需投影）。""" + data = payload() + _source(data)["memory_key"] = "profile:occupation" + + client = FakeMilvus() + await MilvusProfileProjection(client, _embed).upsert(data) + + assert client.upserts == [] + + +@pytest.mark.asyncio +async def test_sensitive_credentials_are_sanitized_before_write() -> None: + """落外部存储前必须脱敏：手机号不得原样写进向量库。""" + data = payload() + _source(data)["content"] = "我的手机号是 15936583816，稳健型" + + client = FakeMilvus() + await MilvusProfileProjection(client, _embed).upsert(data) + + content = client.upserts[0]["data"][0]["content"] + assert "15936583816" not in content + assert "[手机号已隐藏]" in content + + +def _vector(size: int = 1024) -> list[float]: + return [0.0] * size + + +async def _embed(_: str) -> list[float]: + return _vector() + + +async def _embed_short(_: str) -> list[float]: + return _vector(3) diff --git a/tests/unit/service/test_profile_generation_service.py b/tests/unit/service/test_profile_generation_service.py index 8c04a63..65fab79 100644 --- a/tests/unit/service/test_profile_generation_service.py +++ b/tests/unit/service/test_profile_generation_service.py @@ -21,6 +21,8 @@ from app.core.errors import ValidationAgentError from app.core.profile_projection import project_profile from app.service.profile_generation_service import ( REQUIRED_SNAPSHOT_FIELDS, + SYNC_OPERATION_UPSERT, + SYNC_STATUS_PENDING, SYNC_TARGETS, ProfileGenerationService, build_snapshot, @@ -45,11 +47,13 @@ class FakeRepository: assessment: dict[str, Any] | None = None, current: dict[str, Any] | None = None, next_version: int = 2, + memories: list[dict[str, Any]] | None = None, ) -> None: self._profile = profile self._assessment = assessment self._current = current self._next_version = next_version + self._memories = memories or [] self.executed: list[str] = [] self.inserted_snapshot: dict[str, Any] | None = None self.sync_events: list[dict[str, Any]] = [] @@ -66,6 +70,10 @@ class FakeRepository: self.executed.append("current_snapshot") return self._current + async def active_memories(self, _cid: int) -> list[dict[str, Any]]: + self.executed.append("active_memories") + return self._memories + async def next_version(self, _cid: int) -> int: self.executed.append("next_version") return self._next_version @@ -179,7 +187,13 @@ async def test_first_generation_writes_two_events_and_clears_old_current( # `aggregate_type='profile'` 由仓储层固定写入（不在 kwargs 里），此处断言服务传入的实体标识 assert len({e["aggregate_uuid"] for e in repo.sync_events}) == 1 assert all(e["aggregate_uuid"] == result.profile_uuid for e in repo.sync_events) - assert all(e["operation"] == "UPSERT" for e in repo.sync_events) + # 断言取值本身，且**与消费端领取条件对齐**：outbox worker 只领 `pending`/`failed`， + # 写成别的取值事件就永远没人消费。这里不再硬编码字面量（硬编码正是当初跑偏的原因）。 + assert all(e["operation"] == SYNC_OPERATION_UPSERT for e in repo.sync_events) + assert all(e["status"] == SYNC_STATUS_PENDING for e in repo.sync_events) + assert SYNC_OPERATION_UPSERT == "upsert" + assert SYNC_STATUS_PENDING == "pending" + assert set(SYNC_TARGETS) == {"milvus", "neo4j"} # 顺序：清旧当前标记必须在插新版本之前（否则撞唯一键） assert repo.executed.index("clear_current") < repo.executed.index("insert_snapshot") # 新版本标为当前 @@ -187,6 +201,58 @@ async def test_first_generation_writes_two_events_and_clears_old_current( assert repo.inserted_snapshot["version"] == 2 +@pytest.mark.asyncio +async def test_payload_carries_memory_sources_for_projection( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """payload 必须带 `memory_sources` 与 `profile_version`。 + + 这是 Milvus 长期记忆投影的输入契约：适配器要 `memory_sources` 才知道往向量库 + 写什么，要 `profile_version`（或 `version`）才认得出这一批属于哪个画像版本。 + 缺了它，事件能被领取、却什么也投影不出来——属于"静默空转"，必须由测试挡住。 + """ + memories = [{ + "memory_uuid": "11111111-2222-3333-4444-555555555555", + "memory_key": "preference:risk_level", + "content": "稳健型", + "memory_type": "preference", + "confidence": 0.9, + "version": 1, + "valid_until": None, + }] + repo = FakeRepository( + profile=profile_row(), assessment=assessment_row(), current=None, memories=memories + ) + + result = await service(monkeypatch, repo).generate(9102, now=NOW) + + for sync_event in repo.sync_events: + payload = sync_event["payload"] + assert payload["profile_version"] == result.version + assert payload["version"] == result.version + sources = payload["memory_sources"] + assert len(sources) == 1 + assert sources[0]["memory_uuid"] == memories[0]["memory_uuid"] + assert sources[0]["memory_key"] == "preference:risk_level" + + +@pytest.mark.asyncio +async def test_payload_memory_sources_is_empty_without_active_memories( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """没有有效记忆时给**空列表**（而不是省略该键）。 + + 省略键会让适配器的 `memory_sources is invalid` 报错、事件反复重试直至死信； + 空列表是"确实没有要投影的记忆"，语义不同。这里把这个区别钉住。 + """ + repo = FakeRepository(profile=profile_row(), assessment=assessment_row(), current=None) + + await service(monkeypatch, repo).generate(9102, now=NOW) + + for sync_event in repo.sync_events: + assert sync_event["payload"]["memory_sources"] == [] + + @pytest.mark.asyncio async def test_each_version_gets_a_fresh_profile_uuid(monkeypatch: pytest.MonkeyPatch) -> None: """`profile_uuid` 有唯一键，**每个版本必须用新 uuid**（实测撞过 Duplicate entry）。""" diff --git a/tests/unit/worker/test_memory_sync_outbox_worker.py b/tests/unit/worker/test_memory_sync_outbox_worker.py new file mode 100644 index 0000000..40c18fa --- /dev/null +++ b/tests/unit/worker/test_memory_sync_outbox_worker.py @@ -0,0 +1,137 @@ +"""`MemorySyncOutboxWorker` 的定向测试。 + +4 个用例移植自同事 `ZSY_develop` 的 +`tests/unit/worker/test_memory_sync_outbox_worker.py`；最后一个用例是本仓新增的 +**契约回归测试**——它是这次整条链故障的根因所在，必须有人守着。 +""" + +from datetime import datetime + +import pytest + +from app.model.memory import MemorySyncOutbox +from app.worker.memory_sync_outbox_worker import MemorySyncOutboxWorker + + +class FakeSession: + def __init__(self, event: MemorySyncOutbox | None) -> None: + self.event = event + self.commits = 0 + self.rollbacks = 0 + + async def __aenter__(self) -> "FakeSession": + return self + + async def __aexit__(self, *args: object) -> None: + return None + + async def scalar(self, statement: object) -> MemorySyncOutbox | None: + del statement + return self.event + + async def commit(self) -> None: + self.commits += 1 + + async def rollback(self) -> None: + self.rollbacks += 1 + + +def event(*, target: str = "neo4j", retry_count: int = 0) -> MemorySyncOutbox: + return MemorySyncOutbox( + id=1, event_uuid="event-1", aggregate_type="profile", + aggregate_uuid="profile-1", aggregate_version=1, target_store=target, + operation="upsert", payload={"customer_id": 7}, status="pending", + retry_count=retry_count, next_retry_at=None, last_error=None, + created_at=datetime(2026, 1, 1), processed_at=None, + ) + + +@pytest.mark.asyncio +async def test_success_marks_event_processed() -> None: + item = event() + session = FakeSession(item) + seen: list[dict[str, object]] = [] + + async def handler(payload: dict[str, object]) -> None: + seen.append(payload) + + worker = MemorySyncOutboxWorker({"neo4j": handler}, session_factory=lambda: session) + assert await worker.run_once() is True + assert seen == [{"customer_id": 7}] + assert item.status == "processed" + assert item.processed_at is not None + assert session.commits == 1 + + +@pytest.mark.asyncio +async def test_handler_failure_uses_backoff_and_keeps_event() -> None: + item = event() + session = FakeSession(item) + + async def handler(payload: dict[str, object]) -> None: + del payload + raise TimeoutError + + worker = MemorySyncOutboxWorker({"neo4j": handler}, session_factory=lambda: session) + assert await worker.run_once() is True + assert item.status == "failed" + assert item.retry_count == 1 + assert item.next_retry_at is not None + assert item.last_error == "TimeoutError" + + +@pytest.mark.asyncio +async def test_fifth_failure_enters_dead_state() -> None: + item = event(retry_count=4) + session = FakeSession(item) + + async def handler(payload: dict[str, object]) -> None: + del payload + raise RuntimeError + + worker = MemorySyncOutboxWorker({"neo4j": handler}, session_factory=lambda: session) + await worker.run_once() + assert item.status == "dead" + assert item.retry_count == 5 + assert item.next_retry_at is None + + +@pytest.mark.asyncio +async def test_missing_handler_enters_dead_state_without_external_call() -> None: + item = event(target="milvus") + session = FakeSession(item) + worker = MemorySyncOutboxWorker({"neo4j": lambda _: None}, session_factory=lambda: session) + + assert await worker.run_once() is True + assert item.status == "dead" + assert item.last_error == "target_handler_not_configured" + + +# --- 本仓新增：契约回归 ------------------------------------------------------ + + +@pytest.mark.asyncio +async def test_no_handler_is_registered_for_uppercase_target_store() -> None: + """契约回归：`target_store` 只能是小写 `milvus`/`neo4j`。 + + 背景（这条链真实故障的根因）：`profile_generation_service` 曾照 `docs/00` §6.4.6 + 写作大写 `MILVUS`/`NEO4J` + 中文状态 `待处理`，而本 worker 按 `target_store` 的**值** + 分派 handler、且只领 `{"pending","failed"}`。结果该事件两个条件都不满足， + **任何消费者都领不到，永久滞留且不报错**（唯一键 `(event_uuid, target_store)` + 对大小写没有约束，所以静默）。 + + 本用例把"大写分派不到"这一事实钉住：将来谁把取值改回大写，这里会立刻红。 + """ + item = event(target="MILVUS", retry_count=0) + session = FakeSession(item) + called: list[object] = [] + + async def handler(payload: dict[str, object]) -> None: + called.append(payload) + + worker = MemorySyncOutboxWorker({"milvus": handler}, session_factory=lambda: session) + assert await worker.run_once() is True + + assert called == [] # 大写键分派不到 milvus handler + assert item.status == "dead" + assert item.last_error == "target_handler_not_configured" diff --git a/tools/normalize_memory_sync_outbox.py b/tools/normalize_memory_sync_outbox.py new file mode 100644 index 0000000..94cd8e1 --- /dev/null +++ b/tools/normalize_memory_sync_outbox.py @@ -0,0 +1,91 @@ +"""数据订正：把 `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))) diff --git a/tools/setup_milvus_profile_collection.py b/tools/setup_milvus_profile_collection.py new file mode 100644 index 0000000..c18eba7 --- /dev/null +++ b/tools/setup_milvus_profile_collection.py @@ -0,0 +1,144 @@ +"""幂等创建长期记忆画像向量集合 `user_long_term_memory_v1`。 + +写入侧是 `app/infrastructure/milvus_profile_projection.py`，字段必须与之一致。 + +用法： + .\\.venv\\Scripts\\python.exe tools\\setup_milvus_profile_collection.py + +安全口径（与 `setup_milvus_knowledge_collections.py` 相同：这是**共享** Milvus 实例， +里面还有别的项目在用的集合）： +- 只碰 `user_long_term_memory_v1` 这一个集合，绝不 list 后批量删除； +- 集合已存在时**只做结构比对并报告**，不覆盖、不删重建； +- 结构不一致时明确报错退出，由人决定怎么处理，避免静默丢数据。 +""" + +import asyncio +import sys +from typing import Any + +COLLECTION = "user_long_term_memory_v1" +VECTOR_FIELD = "embedding" +VECTOR_DIM = 1024 +PRIMARY_FIELD = "memory_uuid" + +#: (字段名, 最大长度)。`memory_uuid` 是主键（UUID 字符串）。 +VARCHAR_FIELDS: tuple[tuple[str, int], ...] = ( + ("memory_uuid", 64), + ("content", 2048), + ("memory_type", 32), + ("memory_key", 64), + ("status", 16), +) +#: INT64 字段；可空，因为 `valid_until_ts` 对永久记忆为空。 +INT64_FIELDS: tuple[str, ...] = ( + "customer_id", + "version", + "valid_until_ts", + "updated_at_ts", +) +FLOAT_FIELDS: tuple[str, ...] = ("confidence",) + + +def _build_schema() -> Any: + from pymilvus import DataType, MilvusClient # type: ignore[import-untyped] + + schema = MilvusClient.create_schema(auto_id=False, enable_dynamic_field=False) + for name, max_length in VARCHAR_FIELDS: + schema.add_field( + field_name=name, + datatype=DataType.VARCHAR, + max_length=max_length, + is_primary=(name == PRIMARY_FIELD), + nullable=False, + ) + for name in INT64_FIELDS: + # 可空：`valid_until_ts` 对永久记忆必须能不写。 + schema.add_field( + field_name=name, datatype=DataType.INT64, nullable=name != "customer_id" + ) + for name in FLOAT_FIELDS: + schema.add_field(field_name=name, datatype=DataType.DOUBLE, nullable=False) + schema.add_field( + field_name=VECTOR_FIELD, datatype=DataType.FLOAT_VECTOR, dim=VECTOR_DIM + ) + return schema + + +def _build_index_params() -> Any: + from pymilvus import MilvusClient # type: ignore[import-untyped] + + index_params = MilvusClient.prepare_index_params() + index_params.add_index( + field_name=VECTOR_FIELD, + index_name="profile_vector_index", + index_type="AUTOINDEX", + metric_type="COSINE", + ) + return index_params + + +def describe_mismatch(described: dict[str, Any]) -> list[str]: + """比对已存在集合与期望结构，返回差异列表（一致时为空）。""" + problems: list[str] = [] + actual = {field["name"]: field for field in described.get("fields", [])} + expected_names = ( + [name for name, _ in VARCHAR_FIELDS] + list(INT64_FIELDS) + + list(FLOAT_FIELDS) + [VECTOR_FIELD] + ) + for name in expected_names: + if name not in actual: + problems.append(f"缺少字段 {name}") + for name, _ in VARCHAR_FIELDS: + if name in actual and not actual[name].get("is_primary") and name == PRIMARY_FIELD: + problems.append(f"{name} 不是主键") + vector = actual.get(VECTOR_FIELD) + if vector is not None: + params = vector.get("params") or {} + dim = params.get("dim") + if dim is not None and int(dim) != VECTOR_DIM: + problems.append(f"{VECTOR_FIELD} 维度是 {dim}，期望 {VECTOR_DIM}") + return problems + + +async def ensure_collection(uri: str, token: str = "") -> str: + """返回 `created` / `exists` / `conflict:<原因>`。""" + from pymilvus import AsyncMilvusClient # type: ignore[import-untyped] + + client = AsyncMilvusClient(uri=uri, token=token or None) + try: + if await client.has_collection(COLLECTION): + described = await client.describe_collection(COLLECTION) + problems = describe_mismatch(described) + if problems: + return "conflict:" + "; ".join(problems) + return "exists" + await client.create_collection( + collection_name=COLLECTION, + schema=_build_schema(), + index_params=_build_index_params(), + ) + return "created" + finally: + close = getattr(client, "close", None) + if close is not None: + await close() + + +async def main() -> int: + from app.core.config import get_settings + + settings = get_settings() + uri = settings.milvus_uri + if not uri: + print("未配置 milvus_uri，无法创建集合") + return 1 + outcome = await ensure_collection(uri, settings.milvus_token or "") + print(f"{COLLECTION}: {outcome}") + if outcome.startswith("conflict:"): + print("结构不一致，未做任何修改。请人工确认后再处理。") + return 2 + return 0 + + +if __name__ == "__main__": + sys.exit(asyncio.run(main()))