wangjianlong_0626
67ba1b8eee
该适配器与主干 `ProfileGraphProjectionService` 是同一件事的两套实现，对图的建模不同： 本模块按客户各建**私有** `Preference`/`goal` 节点、数据源是 `memory_unit`； 主干服务写**共享** tag 节点、数据源是 `user_facts` 且只投影已确认事实。 ## 它不是"本来就没被装配" | 提交 | 事件 | |---|---| | `f167390`（ZSY） | 新建该适配器 | | `5e848f5`（ZSY） | `feat: wire neo4j projection into worker` —— 在 `__main__.py` 装配，此后一直是**活的** | | `4d8edb4`（主干） | PR #7 合并后接线仍在，**仍是活的** | | `57677f6`（本次合并） | 主动摘掉那段装配 ⇒ 失去生产引用 | `__main__.py` 在本次合并中并没有冲突（git 自动取的是带接线的主干版本）， 是本线解决完冲突后**主动手工删除**的。 ## 但根本原因是它与方案 A 互斥 只要落实方案 A，它就必然失去引用 —— "删 `__main__.py` 接线、保留 runtime 那套"与 "保留 `__main__.py` 骨架、把它的 neo4j handler 换成主干服务"两种做法结果相同。 所以这不是方案 A 的副作用，而是"两套图投影本来就只能活一套"。 ## 改动 - `app/infrastructure/neo4j_profile_projection.py` 文件头加 `.. warning::`： 写明当前未被生产装配、为什么、其单测保护的是**模块自身契约**而非"已装配"， 以及**启用前提** —— 必须先决定"图的节点模型以谁为准"，只加回 `__main__.py` 装配 会重新变成两套图投影并存。 - `docs/39-主干合并对策记录.md` §3.3 补完整时间线与上述论证；§6 第 1 条改为准确表述。 - **保留文件**（实现本身完整：`MERGE` 幂等、按 `profile_version` 判重不被旧版本覆盖、 写入前经 `sanitize_customer_service_message` 脱敏），去留待架构师定： 删除 / 保留为参考实现（当前取此）/ 反过来改用它（则方案 A 需重议）。 验证：`mypy app` → 245 文件 0 错；该模块 4 个单测通过；文档守卫 53 份无编号冲突。
180 lines
8.0 KiB
Python
180 lines
8.0 KiB
Python
"""Neo4j 客户画像最小投影适配器。
|
||
|
||
该模块只接受已审核画像快照的结构化来源，不接受模型生成的 Cypher 或关系名称。
|
||
|
||
.. warning::
|
||
|
||
**本模块当前未被生产代码装配**（2026-09-12 起）。
|
||
|
||
它与主干 `app/service/profile_graph_projection_service.py` 是**同一件事的两套实现**，
|
||
而两者对图的建模不同：
|
||
|
||
- **本模块**：`MERGE (c:Customer)-[:PREFERS/HAS_GOAL]->(p:Preference/goal)`，
|
||
按客户各建**私有**节点，数据源是 `memory_unit`（原始记忆）；
|
||
- **主干服务**：`MERGE (a:Customer)-[:PREFERS]->(b:tag)`，写**共享** tag 节点，
|
||
数据源是 `user_facts`（**已确认**事实），并显式承诺
|
||
"只投影已确认的事实……否则同一件事在画像和图里会有两种说法"。
|
||
|
||
两套同时上线 ⇒ 同一事实在图中两种表示。2026-09-12 合并主干 PR #7 时据此取舍为
|
||
**方案 A：只保留主干服务**，`memory_sync_outbox` 的 `neo4j` 分支改由
|
||
`WorkerRuntime.consume_profile_projections()` 调用 `ProfileGraphProjectionService`；
|
||
原先在 `app/worker/__main__.py` 里对本模块的装配（`memory_sync_handlers["neo4j"]`）
|
||
已删除。取舍的完整理由见 `docs/39-主干合并对策记录.md` §3.3。
|
||
|
||
**保留本文件**是因为实现本身是完整的（`MERGE` 幂等、按 `profile_version` 判重不被旧版本覆盖、
|
||
写入前经 `sanitize_customer_service_message` 脱敏），对后续讨论仍有参考价值；
|
||
其单测 `tests/unit/infrastructure/test_neo4j_profile_projection.py` 仍在跑，
|
||
保护的是模块自身的契约，**不代表它已被装配**。
|
||
|
||
**若要启用**：不要只加回 `__main__.py` 的装配 —— 那会重新变成两套图投影并存。
|
||
正确顺序是先决定"图的节点模型以谁为准"，再改主干服务或本模块使二者一致。
|
||
"""
|
||
|
||
from dataclasses import dataclass
|
||
from datetime import UTC, datetime
|
||
from typing import Any, Protocol
|
||
|
||
from app.core.conversation_privacy import sanitize_customer_service_message
|
||
|
||
|
||
class Neo4jQueryDriver(Protocol):
|
||
async def execute_query(self, *args: Any, **kwargs: Any) -> Any: ...
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ProjectionResult:
|
||
"""一次画像投影结果；`applied=False` 表示版本已被更新版本覆盖。"""
|
||
|
||
applied: bool
|
||
reason: str = ""
|
||
|
||
|
||
_CUSTOMER_QUERY = """
|
||
MERGE (c:Customer {customer_id: $customer_id})
|
||
WITH c, coalesce(c.profile_version, 0) AS current_version
|
||
WHERE current_version < $profile_version
|
||
SET c.profile_version = $profile_version, c.updated_at = $updated_at
|
||
RETURN true AS applied
|
||
"""
|
||
|
||
_PREFERENCE_QUERY = """
|
||
UNWIND $items AS item
|
||
MERGE (p:Preference {customer_id: $customer_id, key: item.memory_key})
|
||
WITH p, item
|
||
WHERE coalesce(p.version, 0) <= $profile_version
|
||
SET p.value = item.content, p.memory_uuid = item.memory_uuid,
|
||
p.version = item.version, p.confidence = item.confidence
|
||
WITH p, item
|
||
MATCH (c:Customer {customer_id: $customer_id})
|
||
MERGE (c)-[r:PREFERS {memory_uuid: item.memory_uuid}]->(p)
|
||
SET r.confidence = item.confidence, r.version = item.version,
|
||
r.valid_from = item.valid_from, r.valid_until = item.valid_until
|
||
RETURN count(p) AS projected
|
||
"""
|
||
|
||
_GOAL_QUERY = """
|
||
UNWIND $items AS item
|
||
MERGE (g:Goal {customer_id: $customer_id, key: item.memory_key})
|
||
WITH g, item
|
||
WHERE coalesce(g.version, 0) <= $profile_version
|
||
SET g.value = item.content, g.memory_uuid = item.memory_uuid,
|
||
g.version = item.version, g.confidence = item.confidence
|
||
WITH g, item
|
||
MATCH (c:Customer {customer_id: $customer_id})
|
||
MERGE (c)-[r:HAS_GOAL {memory_uuid: item.memory_uuid}]->(g)
|
||
SET r.confidence = item.confidence, r.version = item.version,
|
||
r.valid_from = item.valid_from, r.valid_until = item.valid_until
|
||
RETURN count(g) AS projected
|
||
"""
|
||
|
||
|
||
class Neo4jProfileProjection:
|
||
"""把已审核画像来源投影为受控 Neo4j 节点和关系。"""
|
||
|
||
def __init__(self, driver: Neo4jQueryDriver) -> None:
|
||
self._driver = driver
|
||
|
||
async def upsert(self, payload: dict[str, Any]) -> ProjectionResult:
|
||
customer_id, profile_version, updated_at, sources = self._normalize(payload)
|
||
customer_result = await self._driver.execute_query(
|
||
_CUSTOMER_QUERY,
|
||
customer_id=customer_id,
|
||
profile_version=profile_version,
|
||
updated_at=updated_at,
|
||
)
|
||
if not getattr(customer_result, "records", None):
|
||
return ProjectionResult(False, "newer_profile_version_exists")
|
||
grouped = {
|
||
"preference": [item for item in sources if item["kind"] == "preference"],
|
||
"goal": [item for item in sources if item["kind"] == "goal"],
|
||
}
|
||
for kind, items in grouped.items():
|
||
if not items:
|
||
continue
|
||
query = _PREFERENCE_QUERY if kind == "preference" else _GOAL_QUERY
|
||
await self._driver.execute_query(
|
||
query,
|
||
customer_id=customer_id,
|
||
profile_version=profile_version,
|
||
items=items,
|
||
)
|
||
return ProjectionResult(True, "applied")
|
||
|
||
@staticmethod
|
||
def _normalize(
|
||
payload: dict[str, Any],
|
||
) -> tuple[int, int, str, list[dict[str, Any]]]:
|
||
customer_id = payload.get("customer_id")
|
||
profile_version = payload.get("profile_version")
|
||
profile_uuid = payload.get("profile_uuid")
|
||
sources = payload.get("memory_sources")
|
||
if not isinstance(customer_id, int) or customer_id <= 0:
|
||
raise ValueError("customer_id is invalid")
|
||
if not isinstance(profile_version, int) or profile_version <= 0:
|
||
raise ValueError("profile_version is invalid")
|
||
if not isinstance(profile_uuid, str) or not profile_uuid.strip():
|
||
raise ValueError("profile_uuid is invalid")
|
||
if not isinstance(sources, list):
|
||
raise ValueError("memory_sources is invalid")
|
||
normalized: list[dict[str, Any]] = []
|
||
for source in sources:
|
||
if not isinstance(source, dict):
|
||
raise ValueError("memory source is invalid")
|
||
memory_uuid = source.get("memory_uuid")
|
||
memory_key = source.get("memory_key")
|
||
content = source.get("content")
|
||
memory_type = source.get("memory_type")
|
||
if not isinstance(memory_uuid, str) or not memory_uuid.strip():
|
||
raise ValueError("memory source fields are invalid")
|
||
if not isinstance(memory_key, str) or not memory_key.strip():
|
||
raise ValueError("memory source fields are invalid")
|
||
if not isinstance(content, str) or not content.strip():
|
||
raise ValueError("memory source fields are invalid")
|
||
if not isinstance(memory_type, str) or not memory_type.strip():
|
||
raise ValueError("memory source fields are invalid")
|
||
if memory_key.startswith("preference:"):
|
||
kind = "preference"
|
||
elif memory_key.startswith("goal:"):
|
||
kind = "goal"
|
||
else:
|
||
raise ValueError("memory key is not projectable")
|
||
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")
|
||
normalized.append({
|
||
"kind": kind,
|
||
"memory_uuid": memory_uuid.strip(),
|
||
"memory_key": memory_key.strip(),
|
||
"content": sanitize_customer_service_message(content).strip(),
|
||
"memory_type": memory_type.strip(),
|
||
"confidence": float(confidence),
|
||
"version": version,
|
||
"valid_until": source.get("valid_until"),
|
||
"valid_from": source.get("valid_from"),
|
||
})
|
||
updated_at = str(payload.get("updated_at") or datetime.now(UTC).isoformat())
|
||
return customer_id, profile_version, updated_at, normalized
|