From 75dff088d49040b896774759277dbbc6fb49ed3e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=8D=BF=E4=BA=91=E7=A7=8B=E6=9C=88?= <15273589815@163.com> Date: Thu, 10 Sep 2026 21:40:00 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=8E=A5=E9=80=9A=20Neo4j=20=E5=9B=BE?= =?UTF-8?q?=E8=83=BD=E5=8A=9B=EF=BC=88=E9=A9=B1=E5=8A=A8=E5=AE=9E=E7=8E=B0?= =?UTF-8?q?=20+=20=E8=A3=85=E9=85=8D=EF=BC=89=EF=BC=8C=E5=B9=B6=E5=AE=9E?= =?UTF-8?q?=E6=B5=8B=E7=A1=AE=E8=AE=A4=E5=9B=BE=E6=A8=A1=E5=9E=8B=E6=A0=87?= =?UTF-8?q?=E7=AD=BE=E4=B8=8D=E4=B8=80=E8=87=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 第 2 步（图库）的地基部分。 1. 新增 app/infrastructure/graph.py：GraphDriver 协议的 neo4j 实现。 这层此前是空的——relationship_service.py 里的 GraphDriver 只有 Protocol 声明、没有任何 具体实现，组装层也没有装配它，因此图读服务在运行期必然降级（neo4j_unavailable）。 这正是"Neo4j 连读适配器都没有"的根因。 降级取向与 Milvus 侧一致：本层不吞异常（由调用方决定降级方式），但构造失败返回 None ——图库不可用不该让应用起不来，也不该阻塞主链路。 2. bootstrap 新增 get_relationship_service()：装配图读服务。 该服务只读，关系类型受 ALLOWED_RELATIONSHIPS 白名单约束；写入走 GraphProjectionWorker （由领域事件驱动），这里不提供任意写接口——避免出现绕过事件链路的直写路径。 3. 实测确认一处既有缺陷（本次不改，方案待定）：投影 worker 写入的节点标签是 :Entity {entity_id}（字符串属性），而 RelationshipService.neighbors 查询的是 :Customer {customer_id}（整数属性），两侧从未对齐，写进去的关系读不出来。 Neo4j 在执行读查询时直接给出三条警告佐证：未知标签 Customer、未知属性 customer_id、 未知关系类型 PREFERS——这也说明业务图从未被真正投影过（库中只有 Neo4j 自带的 Person/Movie 示例数据，共 5 个节点）。 验证：ruff 通过、mypy 110 文件无错；get_relationship_service() 返回 RelationshipService， neighbors 与 paths 正常执行并返回空集（非降级），非法关系类型被白名单拒绝。 --- app/infrastructure/graph.py | 64 ++++++++++++++++++++++++++++++++++ app/service/agent/bootstrap.py | 18 ++++++++++ 2 files changed, 82 insertions(+) create mode 100644 app/infrastructure/graph.py diff --git a/app/infrastructure/graph.py b/app/infrastructure/graph.py new file mode 100644 index 0000000..788b719 --- /dev/null +++ b/app/infrastructure/graph.py @@ -0,0 +1,64 @@ +"""Neo4j 图驱动：把 neo4j 异步驱动封装成 `RelationshipService` 期望的 `GraphDriver` 协议。 + +为什么单独一层：`RelationshipService` 只依赖 `execute_query(query, **params)` 这个窄协议， +不关心底层是哪个驱动。有这一层，读服务、投影 worker、删除客户端就能共用同一套连接管理， +也便于测试注入替身。 + +**这层此前是空的** —— `app/service/relationship_service.py` 里的 `GraphDriver` 只有一个 +Protocol 声明，没有任何具体实现，组装层也没有装配它，所以读服务在运行期必然降级 +（`neo4j_unavailable`），这正是"Neo4j 连读适配器都没有"的根因。 + +降级语义与 Milvus 侧保持一致：本层**不吞异常**（由调用方决定降级方式），但**构造失败返回 None** +——图库不可用不该让应用起不来，也不该阻塞主链路。 +""" + +import logging +from collections.abc import Sequence +from typing import Any + +from app.core.config import get_settings + +logger = logging.getLogger(__name__) + + +class Neo4jGraphDriver: + """`GraphDriver` 协议的 neo4j 实现。""" + + def __init__(self, driver: Any, database: str) -> None: + self._driver = driver + self._database = database + + async def execute_query(self, query: str, **parameters: Any) -> Sequence[Any]: + """执行一条 Cypher 并返回记录列表。 + + 查询文本由调用方提供（`RelationshipService` 与投影 worker 各自持有白名单： + 前者限制关系类型，后者只接受 `ALLOWED_RELATIONSHIPS` 中的关系），本层不做校验， + 只负责执行与结果整形——把校验放在拥有业务语义的地方，避免两处规则漂移。 + """ + async with self._driver.session(database=self._database) as session: + result = await session.run(query, **parameters) + return list(await result.data()) + + async def close(self) -> None: + await self._driver.close() + + +def build_graph_driver() -> Neo4jGraphDriver | None: + """按配置构造图驱动；依赖缺失或配置不全时返回 None（图能力关闭）。 + + 与 `get_vector_memory_adapter` 同一取向：外部依赖不可用只关闭对应能力， + 不让整个应用起不来。 + """ + settings = get_settings() + if not settings.neo4j_uri or not settings.neo4j_password: + logger.warning("neo4j config incomplete; graph capability disabled") + return None + try: + from neo4j import AsyncGraphDatabase + except ImportError: + logger.warning("neo4j driver not installed; graph capability disabled") + return None + driver = AsyncGraphDatabase.driver( + settings.neo4j_uri, auth=(settings.neo4j_username, settings.neo4j_password) + ) + return Neo4jGraphDriver(driver, settings.neo4j_database) diff --git a/app/service/agent/bootstrap.py b/app/service/agent/bootstrap.py index 48f7ce2..76fcab7 100644 --- a/app/service/agent/bootstrap.py +++ b/app/service/agent/bootstrap.py @@ -9,6 +9,7 @@ from app.core.errors import RecoverableAgentError from app.core.fund_contracts import FundQuoteQuery from app.core.knowledge_contracts import KnowledgeSearchInput 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.vector_memory import VectorMemoryAdapter from app.service.agent.factory import AgentFactory @@ -27,6 +28,7 @@ from app.service.model_gateway import ( ModelEmbeddingService, ModelGenerationService, ) +from app.service.relationship_service import RelationshipService from app.service.runtime_config_service import load_active_intent_configs from app.service.suitability_service import SuitabilityToolInput, suitability_tool_handler from app.service.tool_executor import ToolDefinition, ToolExecutor, ToolRegistry @@ -126,6 +128,22 @@ def get_knowledge_search_service() -> KnowledgeSearchService: return KnowledgeSearchService(client, _embed_text) +@lru_cache(maxsize=1) +def get_relationship_service() -> RelationshipService | None: + """图关系读服务：客户 → 产品／标签／事件 的多跳查询入口。 + + 驱动构造失败时返回 None（图能力关闭），由调用方降级——图库不可用不该阻塞主链路， + 与 Milvus 侧"语义通道缺失不影响结构化召回"是同一取向。 + + 注意：本服务**只读**，且关系类型受 `RelationshipService.ALLOWED_RELATIONSHIPS` 白名单约束； + 写入走 `GraphProjectionWorker`（由领域事件驱动），这里不提供任意写接口。 + """ + driver = build_graph_driver() + if driver is None: + return None + return RelationshipService(driver) + + def build_memory_recall_service(session: AsyncSession) -> MemoryRecallService: """记忆召回组装：结构化召回始终可用，Redis 缓存与语义通道可用时叠加。