lzf_0626
7635014d9e
第 2 步（Neo4j）主体，三件事：
1. 新增 app/service/graph_model.py：图模型的**单一来源**。
节点标签与主属性名会被拼进 Cypher（Neo4j 的标签不能用参数占位），因此必须来自受控常量，
绝不能是调用方传入的字符串——否则就是 Cypher 注入；同时读服务与投影侧必须就"客户节点
长什么样"达成一致。此前两边各写各的（投影写 :Entity{entity_id}、读服务查
:Customer{customer_id}），结果写进去的关系永远读不出来。
另用 RELATION_SEMANTICS 补上"谁指向谁"的方向校验：底座已有关系白名单（8 种），
但白名单只约束关系名，不约束方向，这里补上，避免把 TRADED 写成 Customer→Tag。
2. 改 app/worker/graph_projection_worker.py：按 payload 的 source_type/target_type 写具体标签，
投影前校验关系方向。标签与属性名全部取自 graph_model，调用方传不进任意字符串。
3. 新增 app/service/profile_graph_projection_service.py：把画像事实投影成节点与关系，
并补上此前缺失的**投影删除客户端**（销户或记忆失效时 DETACH DELETE，同时清悬挂边）。
只投影 user_facts（与画像同源），不直接读原始记忆——否则同一件事在画像与图里会有两种说法。
图库故障一律返回 degraded 而不抛异常：投顾推荐可以暂时没有图，但不该因为图挂了
导致画像更新失败。
背景（核查发现）：MemorySyncOutbox 与 GraphProjectionWorker 此前是**孤儿代码**——表建了、
worker 也实现了（含幂等、重试、死信、MERGE），但没有任何代码往 outbox 写、也没有任何地方
实例化这个 worker，整条图投影链路从未接上过。这正是图中只有 Neo4j 自带 Person/Movie
示例数据的根因。本次把"画像 → 图"这条链路接通；领域事件驱动的投影仍待接入（worker 已修好可用）。
实测验证：
· 投影客户 9001 → relations=1，随后 neighbors(PREFERS) 直接读到
{'tag_key': 'preference:risk_level=稳健型'} —— 写后读通，标签已对齐；
· delete_customer → 关系归零（degraded=False），重新投影恢复为 1，再次投影仍为 1（幂等）；
· 图中标签为 Customer/Tag、关系为 PREFERS；早先探针残留的 Entity 节点已清理；
· ruff 通过、mypy 112 文件无错。
83 lines
3.8 KiB
Python
83 lines
3.8 KiB
Python
from datetime import UTC, datetime, timedelta
|
|
from typing import Any
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.model.memory import MemorySyncOutbox
|
|
from app.service.graph_model import merge_node_clause, node_id, node_spec, relation_allowed
|
|
from app.service.relationship_service import RelationshipService
|
|
|
|
|
|
class GraphProjectionWorker:
|
|
"""Projects approved domain events to Neo4j; callers provide durable dedup storage."""
|
|
|
|
def __init__(
|
|
self, relationships: RelationshipService, session: AsyncSession | None = None
|
|
) -> None:
|
|
self.relationships = relationships
|
|
self.session = session
|
|
self.processed_event_ids: set[str] = set()
|
|
|
|
async def project(self, event_id: str, payload: dict[str, Any]) -> bool:
|
|
if event_id in self.processed_event_ids:
|
|
return False
|
|
if self.session is not None:
|
|
existing = await self.session.scalar(
|
|
select(MemorySyncOutbox).where(MemorySyncOutbox.event_uuid == event_id)
|
|
)
|
|
if existing is not None and existing.status == "processed":
|
|
return False
|
|
relation = payload.get("relationship")
|
|
if relation not in RelationshipService.ALLOWED_RELATIONSHIPS:
|
|
raise ValueError("relationship is not allowed")
|
|
# 节点按**类型**写标签与主属性：原先统一写 `:Entity {entity_id}`，而读服务查的是
|
|
# `:Customer {customer_id}`，两侧从未对齐，写进去的关系永远读不出来。
|
|
# 标签与属性名来自 `graph_model` 的受控常量，调用方传不进任意字符串。
|
|
source_type = str(payload.get("source_type", "")).strip().lower()
|
|
target_type = str(payload.get("target_type", "")).strip().lower()
|
|
source = node_spec(source_type)
|
|
target = node_spec(target_type)
|
|
if not relation_allowed(relation, source_type, target_type):
|
|
raise ValueError(
|
|
f"relationship {relation} cannot connect {source_type} -> {target_type}"
|
|
)
|
|
query = (
|
|
f"{merge_node_clause('a', source, 'source_id')} "
|
|
f"{merge_node_clause('b', target, 'target_id')} "
|
|
f"MERGE (a)-[r:{relation}]->(b) "
|
|
"SET r.trace_id = $trace_id, r.confidence = $confidence, r.updated_at = $updated_at"
|
|
)
|
|
try:
|
|
await self.relationships.driver.execute_query(
|
|
query,
|
|
source_id=node_id(source, payload["source_id"]),
|
|
target_id=node_id(target, payload["target_id"]),
|
|
trace_id=str(payload.get("trace_id", "")),
|
|
confidence=float(payload.get("confidence", 0.0)),
|
|
updated_at=datetime.now(UTC).replace(tzinfo=None).isoformat(),
|
|
)
|
|
except Exception as exc:
|
|
if self.session is not None:
|
|
record = await self.session.scalar(
|
|
select(MemorySyncOutbox).where(MemorySyncOutbox.event_uuid == event_id)
|
|
)
|
|
if record is not None:
|
|
record.retry_count += 1
|
|
record.last_error = str(exc)[:500]
|
|
record.status = "dead" if record.retry_count >= 5 else "pending"
|
|
record.next_retry_at = datetime.now(UTC).replace(tzinfo=None) + timedelta(
|
|
seconds=min(300, 2 ** record.retry_count)
|
|
)
|
|
await self.session.commit()
|
|
raise
|
|
self.processed_event_ids.add(event_id)
|
|
if self.session is not None:
|
|
record = await self.session.scalar(
|
|
select(MemorySyncOutbox).where(MemorySyncOutbox.event_uuid == event_id)
|
|
)
|
|
if record is not None:
|
|
record.status = "processed"
|
|
await self.session.commit()
|
|
return True
|