diff --git a/app/service/graph_model.py b/app/service/graph_model.py new file mode 100644 index 0000000..e96eb13 --- /dev/null +++ b/app/service/graph_model.py @@ -0,0 +1,80 @@ +"""图模型定义：节点类型、主属性与关系语义（**单一来源**）。 + +为什么必须集中定义： + +1. **标签与属性名要拼进 Cypher**（Neo4j 的标签不能用查询参数占位），因此它们**必须**来自 + 受控常量，绝不能接受调用方传入的任意字符串，否则就是 Cypher 注入。 +2. 读服务（`RelationshipService`）与投影侧需要就"客户节点长什么样"达成一致。此前投影写 + `:Entity {entity_id}`、读服务查 `:Customer {customer_id}`，两边各写各的，结果是**写进去的 + 关系永远读不出来**。把节点规格放在一处，两边都从这里取，才不会再次漂移。 +3. `RELATION_SEMANTICS` 记录每种关系连接哪两类节点，用于在投影时拒绝无意义的边 + （例如把 `TRADED` 写成 Customer→Tag）。底座已有关系白名单（8 种），这里补的是 + "谁指向谁"的语义约束。 +""" + +from dataclasses import dataclass +from typing import Any + + +@dataclass(frozen=True) +class NodeSpec: + """一个节点类型的规格。`cast` 决定主键类型：客户 id 是整数（与读服务查询一致）。""" + + type_key: str + label: str + property: str + cast: type + + +# payload 里使用的类型名 → 节点规格。未登记的类型一律拒绝。 +NODE_SPECS: dict[str, NodeSpec] = { + "customer": NodeSpec("customer", "Customer", "customer_id", int), + "product": NodeSpec("product", "Product", "product_code", str), + "tag": NodeSpec("tag", "Tag", "tag_key", str), + "industry": NodeSpec("industry", "Industry", "category", str), + "event": NodeSpec("event", "Event", "event_id", str), +} + +# 关系 → (允许的源节点类型, 允许的目标节点类型) +RELATION_SEMANTICS: dict[str, tuple[tuple[str, ...], tuple[str, ...]]] = { + "PREFERS": (("customer",), ("tag",)), + "HAS_GOAL": (("customer",), ("tag",)), + "INTERESTED_IN": (("customer",), ("product", "industry")), + "TRADED": (("customer",), ("product",)), + "HOLDS": (("customer",), ("product",)), + "TRIGGERED_RISK": (("customer",), ("event",)), + "BELONGS_TO_CATEGORY": (("product",), ("tag",)), + "EXPOSED_TO_INDUSTRY": (("product", "customer"), ("industry",)), +} + + +def node_spec(type_key: object) -> NodeSpec: + """按类型名取节点规格；未登记的类型抛 `ValueError`（宁可失败也不拼出任意标签）。""" + key = str(type_key or "").strip().lower() + spec = NODE_SPECS.get(key) + if spec is None: + raise ValueError(f"node type is not allowed: {type_key!r}") + return spec + + +def node_id(spec: NodeSpec, value: Any) -> Any: + """把主键值转成该节点应有的类型。""" + return value if spec.cast is str else spec.cast(value) + + +def relation_allowed(relation: str, source_type: str, target_type: str) -> bool: + """该关系是否允许连接这两类节点。""" + expected = RELATION_SEMANTICS.get(relation) + if expected is None: + return False + sources, targets = expected + return source_type in sources and target_type in targets + + +def merge_node_clause(variable: str, spec: NodeSpec, id_parameter: str) -> str: + """生成 `MERGE (a:Label {prop: $param})` 子句。 + + 只有 `spec` 里的标签与属性名会进入查询文本（它们来自本模块常量），`$param` 走参数绑定， + 因此调用方无法通过数据影响 Cypher 结构。 + """ + return f"MERGE ({variable}:{spec.label} {{{spec.property}: ${id_parameter}}})" diff --git a/app/service/profile_graph_projection_service.py b/app/service/profile_graph_projection_service.py new file mode 100644 index 0000000..fc9a12e --- /dev/null +++ b/app/service/profile_graph_projection_service.py @@ -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() diff --git a/app/worker/graph_projection_worker.py b/app/worker/graph_projection_worker.py index d381589..890a67a 100644 --- a/app/worker/graph_projection_worker.py +++ b/app/worker/graph_projection_worker.py @@ -5,6 +5,7 @@ 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 @@ -30,19 +31,31 @@ class GraphProjectionWorker: 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 = ( - "MERGE (a:Entity {entity_id: $source_id}) " - "MERGE (b:Entity {entity_id: $target_id}) " + 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" + "SET r.trace_id = $trace_id, r.confidence = $confidence, r.updated_at = $updated_at" ) try: await self.relationships.driver.execute_query( query, - source_id=str(payload["source_id"]), - target_id=str(payload["target_id"]), + 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: