feat: 画像→图投影打通(按类型写标签,读写对齐;补删除客户端)
第 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 文件无错。
This commit is contained in:
@@ -0,0 +1,155 @@
|
||||
"""画像 → 图投影:把 MySQL 里权威的画像与事实投影成 Neo4j 的节点与关系。
|
||||
|
||||
**为什么要有图**(这是它存在的理由,不是技术偏好):投顾要靠"买过 A 的客户还买过什么"
|
||||
这类多跳关系做组合推荐,风控要靠关系网络发现关联账户与资金链路;这些查询用关系库写
|
||||
既别扭又慢,用图是顺手的事。
|
||||
|
||||
设计约束:
|
||||
|
||||
1. **单向投影**:MySQL 是唯一真相,图库只是派生视图。图数据损坏或模型变更时可以直接重建
|
||||
(拿同一份画像重投影即可),因此这里**不需要与 MySQL 保持事务一致**。
|
||||
2. **幂等**:全部用 MERGE,重复投影不会产生重复节点或关系。
|
||||
3. **降级**:图库不可用时返回 `degraded`,**绝不把异常抛给画像重建这类主链路**
|
||||
—— 投顾推荐可以暂时没有图,但不能因为图挂了就让画像更新失败。
|
||||
4. **只投影"已确认"的事实**:与画像同源(`user_facts`),不直接读原始记忆,
|
||||
保证图里的偏好标签与画像口径一致;否则同一件事在画像和图里会有两种说法。
|
||||
"""
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import select, text
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.model.profile import UserFact
|
||||
from app.service.graph_model import merge_node_clause, node_spec, relation_allowed
|
||||
from app.service.relationship_service import RelationshipService
|
||||
|
||||
# 事实键 → (关系类型, 目标节点类型)
|
||||
FACT_PROJECTION: dict[str, tuple[str, str]] = {
|
||||
"preference:risk_level": ("PREFERS", "tag"),
|
||||
"preference:asset_class": ("PREFERS", "tag"),
|
||||
"preference:horizon": ("HAS_GOAL", "tag"),
|
||||
"profile:family": ("PREFERS", "tag"),
|
||||
}
|
||||
|
||||
MAX_EDGES_PER_CUSTOMER = 200
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ProjectionOutcome:
|
||||
relations: int = 0
|
||||
degraded: bool = False
|
||||
reason: str = ""
|
||||
|
||||
|
||||
class ProfileGraphProjectionService:
|
||||
def __init__(
|
||||
self, session: AsyncSession, relationships: RelationshipService | None
|
||||
) -> None:
|
||||
self.session = session
|
||||
self.relationships = relationships
|
||||
|
||||
async def project_customer(self, customer_id: int) -> ProjectionOutcome:
|
||||
"""把该客户的画像事实投影到图中。"""
|
||||
if self.relationships is None:
|
||||
return ProjectionOutcome(degraded=True, reason="graph_client_unavailable")
|
||||
|
||||
edges = await self._edges(customer_id)
|
||||
written = 0
|
||||
try:
|
||||
for relation, target_type, target_id in edges[:MAX_EDGES_PER_CUSTOMER]:
|
||||
await self._merge_edge(customer_id, relation, target_type, target_id)
|
||||
written += 1
|
||||
except Exception as exc:
|
||||
# 图库故障:如实标记降级,不伪装成功(与 Milvus/Neo4j 投影清理同一取向)
|
||||
return ProjectionOutcome(relations=written, degraded=True,
|
||||
reason=f"graph_write_failed:{type(exc).__name__}")
|
||||
return ProjectionOutcome(relations=written)
|
||||
|
||||
async def delete_customer(self, customer_id: int) -> ProjectionOutcome:
|
||||
"""删除该客户在图中的节点及其全部关系。
|
||||
|
||||
这是此前缺失的**投影删除客户端**:记忆失效或客户销户后,若图里仍留着旧关系,
|
||||
投顾会拿过期偏好做推荐、风控会看到已不存在的关系网络。
|
||||
`DETACH DELETE` 同时清理关系,避免留下悬挂边。
|
||||
"""
|
||||
if self.relationships is None:
|
||||
return ProjectionOutcome(degraded=True, reason="graph_client_unavailable")
|
||||
try:
|
||||
await self.relationships.driver.execute_query(
|
||||
"MATCH (c:Customer {customer_id: $customer_id}) DETACH DELETE c",
|
||||
customer_id=int(customer_id),
|
||||
)
|
||||
except Exception as exc:
|
||||
return ProjectionOutcome(degraded=True,
|
||||
reason=f"graph_delete_failed:{type(exc).__name__}")
|
||||
return ProjectionOutcome()
|
||||
|
||||
# ---------- 内部 ----------
|
||||
|
||||
async def _edges(self, customer_id: int) -> list[tuple[str, str, str]]:
|
||||
"""收集要投影的边:(关系, 目标节点类型, 目标节点标识)。"""
|
||||
edges: list[tuple[str, str, str]] = []
|
||||
facts = list(await self.session.scalars(
|
||||
select(UserFact).where(UserFact.customer_id == customer_id)
|
||||
))
|
||||
for fact in facts:
|
||||
key = str(fact.fact_key)
|
||||
projection = FACT_PROJECTION.get(key)
|
||||
if projection is None:
|
||||
continue
|
||||
relation, target_type = projection
|
||||
edges.append((relation, target_type, f"{key}={self._text(fact.fact_value)}"))
|
||||
for product_code in await self._holdings(customer_id):
|
||||
edges.append(("HOLDS", "product", product_code))
|
||||
return edges
|
||||
|
||||
async def _holdings(self, customer_id: int) -> list[str]:
|
||||
"""客户当前持仓的产品代码。
|
||||
|
||||
交易模块尚未产生数据,或表结构与预期不符时返回空列表——持仓投影自动跳过,
|
||||
而不是让整次投影失败。等交易侧就绪后无需改这里的代码。
|
||||
"""
|
||||
try:
|
||||
rows = await self.session.execute(text(
|
||||
"SELECT p.product_code FROM fin_holding h "
|
||||
"JOIN fin_product p ON p.id = h.product_id "
|
||||
"WHERE h.customer_id = :customer_id LIMIT 50"
|
||||
), {"customer_id": customer_id})
|
||||
except Exception:
|
||||
return []
|
||||
return [str(row[0]) for row in rows if row[0]]
|
||||
|
||||
async def _merge_edge(
|
||||
self, customer_id: int, relation: str, target_type: str, target_id: str
|
||||
) -> None:
|
||||
if not relation_allowed(relation, "customer", target_type):
|
||||
raise ValueError(f"relationship {relation} cannot connect customer -> {target_type}")
|
||||
source = node_spec("customer")
|
||||
target = node_spec(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.source = $source, r.updated_at = $updated_at"
|
||||
)
|
||||
await self.relationships.driver.execute_query( # type: ignore[union-attr]
|
||||
query,
|
||||
source_id=node_spec("customer").cast(customer_id),
|
||||
target_id=target_id,
|
||||
source="user_facts",
|
||||
updated_at=_now_iso(),
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _text(value: Any) -> str:
|
||||
if isinstance(value, str):
|
||||
return value.strip().strip('"')
|
||||
return str(value)
|
||||
|
||||
|
||||
def _now_iso() -> str:
|
||||
from datetime import UTC, datetime
|
||||
|
||||
return datetime.now(UTC).replace(tzinfo=None).isoformat()
|
||||
Reference in New Issue
Block a user