diff --git a/aperag/indexing/graph.py b/aperag/indexing/graph.py index 121329af9..0742195c2 100644 --- a/aperag/indexing/graph.py +++ b/aperag/indexing/graph.py @@ -508,25 +508,29 @@ async def get_relation(self, source: str, target: str, type: str) -> RelationWit """Read-path helper for relations.""" # ------------------------------------------------------------------ - # Wave 6 #33 chunk 1 — LightRAG-style retrieval query layer (Protocol - # stubs; cross-backend implementations land in chunk 2 per - # §K.11.11). The retrieval pipeline (§G.5) and graph-curation flows - # consume these to compose a graph-recall context for a user query - # without going through the legacy ``GraphIndexService.query_context``. + # Wave 6 #33 — LightRAG-style retrieval query layer. The retrieval + # pipeline (§G.5) and graph-curation flows consume these to compose + # a graph-recall context for a user query without going through the + # legacy ``GraphIndexService.query_context``. # # Design notes: # * Returns are the canonical :class:`EntityWithLineage` / # :class:`RelationWithLineage` shapes already used by ``get_*`` — # keeps the read surface uniform; callers compose context text # themselves rather than relying on a backend-formatted string. - # * ``query_by_vector`` takes a pre-computed embedding so backends - # stay agnostic of the embedding model; the retrieval pipeline - # already owns the embedder. # * ``expand_neighbors_n_hops`` returns BOTH neighbour entities and # the relation edges that connect them (1-hop or multi-hop). Per # simple-stable directive the multi-hop frontier is bounded by # ``hops`` (caller picks; default 1) so backends don't need a # query planner. + # * Vector-recall (``query_entities_by_vector``) was originally + # declared in chunk 1 but dropped per architect ruling + # msg=54eac595 (Option A): the Wave 4 lineage schema does not + # carry an entity-vector column, so a Protocol method on this + # store would either invite a Qdrant cross-concern coupling or + # silently no-op. Wave 7+ may add it via a separate + # ``LightRAGQueryService`` if real evidence surfaces; until then + # no dead code lives here. See §K.11 amend. # ------------------------------------------------------------------ async def query_entities_by_keyword( @@ -541,30 +545,8 @@ async def query_entities_by_keyword( Match semantics are backend-defined (case-insensitive substring on Postgres / Cypher CONTAINS on Neo4j / etc.); the contract is "best-effort lexical recall, not strict equality". An empty - ``query`` returns ``[]`` (no spurious recall). - - Wave 6 #33 chunk 1 — Protocol stub; backend implementations - land chunk 2. - """ - - async def query_entities_by_vector( - self, - *, - embedding: list[float], - top_k: int, - ) -> list[EntityWithLineage]: - """Return up to ``top_k`` entities ranked by similarity of - ``embedding`` against an entity-vector index the backend - maintains. Backends without a native vector index may compose - this against the collection's existing Qdrant collection - (entity-name-keyed points) — chunk 2 picks the per-backend - approach. - - Empty ``embedding`` is invalid and raises ``ValueError`` (the - retrieval pipeline always passes a real vector — the embedder - has already validated the query upstream). - - Wave 6 #33 chunk 1 — Protocol stub. + or whitespace-only ``query`` returns ``[]`` (no spurious recall). + ``top_k <= 0`` returns ``[]``. """ async def expand_neighbors_n_hops( @@ -585,9 +567,9 @@ async def expand_neighbors_n_hops( Implementations MUST bound the result by ``hops`` to keep the traversal cost predictable; per simple-stable directive operators don't manage tunable per-collection traversal - budgets — this is the only knob. - - Wave 6 #33 chunk 1 — Protocol stub. + budgets — this is the only knob. ``hops <= 0`` returns just + the requested seed entities (no relations). Empty + ``entity_names`` returns ``([], [])``. """ @@ -789,13 +771,7 @@ async def get_relation(self, source: str, target: str, type: str) -> RelationWit ) # ------------------------------------------------------------------ - # Wave 6 #33 chunk 1 — LightRAG-style query layer stubs. - # Real implementations land chunk 2 (per §K.11.11 3-chunk - # decomposition). The in-memory store also defers to chunk 2 so - # tests that assert the Protocol surface can pin the - # NotImplementedError contract today and switch to the real - # behaviour when chunk 2 ships — without rewriting the test - # scaffolding. + # Wave 6 #33 chunk 2 — LightRAG-style query layer real impls. # ------------------------------------------------------------------ async def query_entities_by_keyword( @@ -804,15 +780,25 @@ async def query_entities_by_keyword( query: str, top_k: int, ) -> list[EntityWithLineage]: - raise NotImplementedError("Wave 6 #33 chunk 2 — query_entities_by_keyword pending implementation") - - async def query_entities_by_vector( - self, - *, - embedding: list[float], - top_k: int, - ) -> list[EntityWithLineage]: - raise NotImplementedError("Wave 6 #33 chunk 2 — query_entities_by_vector pending implementation") + if not query or not query.strip() or top_k <= 0: + return [] + needle = query.strip().lower() + matches: list[EntityWithLineage] = [] + async with self._guard: + for name, row in self._entities.items(): + if needle not in name.lower(): + continue + matches.append( + EntityWithLineage( + name=row.name, + type=row.type, + source_lineage=tuple(_sorted_lineage(row.source_lineage.values())), + description_parts=tuple(_sorted_description_parts(row.description_parts.values())), + ) + ) + # Deterministic ordering for stable test output. + matches.sort(key=lambda e: e.name) + return matches[:top_k] async def expand_neighbors_n_hops( self, @@ -820,7 +806,61 @@ async def expand_neighbors_n_hops( entity_names: list[str], hops: int = 1, ) -> tuple[list[EntityWithLineage], list[RelationWithLineage]]: - raise NotImplementedError("Wave 6 #33 chunk 2 — expand_neighbors_n_hops pending implementation") + if not entity_names: + return ([], []) + + seen_entities: dict[str, EntityWithLineage] = {} + seen_relations: dict[tuple[str, str, str], RelationWithLineage] = {} + frontier: set[str] = {n for n in entity_names if n} + + async with self._guard: + # Always include the seed entities (whether or not they + # exist in the store). + for name in frontier: + row = self._entities.get(name) + if row is None: + continue + seen_entities[name] = EntityWithLineage( + name=row.name, + type=row.type, + source_lineage=tuple(_sorted_lineage(row.source_lineage.values())), + description_parts=tuple(_sorted_description_parts(row.description_parts.values())), + ) + + current = set(frontier) + for _ in range(max(hops, 0)): + next_frontier: set[str] = set() + for rel_key, rel_row in self._relations.items(): + src, tgt, rtype = rel_key + if src in current or tgt in current: + if rel_key not in seen_relations: + seen_relations[rel_key] = RelationWithLineage( + source=rel_row.source, + target=rel_row.target, + type=rel_row.type, + evidence_lineage=tuple(_sorted_lineage(rel_row.evidence_lineage.values())), + description_parts=tuple(_sorted_description_parts(rel_row.description_parts.values())), + ) + for neighbour in (src, tgt): + if neighbour in seen_entities: + continue + row = self._entities.get(neighbour) + if row is None: + continue + seen_entities[neighbour] = EntityWithLineage( + name=row.name, + type=row.type, + source_lineage=tuple(_sorted_lineage(row.source_lineage.values())), + description_parts=tuple(_sorted_description_parts(row.description_parts.values())), + ) + next_frontier.add(neighbour) + if not next_frontier: + break + current = next_frontier + + entities = sorted(seen_entities.values(), key=lambda e: e.name) + relations = sorted(seen_relations.values(), key=lambda r: (r.source, r.target, r.type)) + return (entities, relations) def _sorted_lineage(members: Iterable[LineageMember]) -> list[LineageMember]: diff --git a/aperag/indexing/graph_storage/nebula.py b/aperag/indexing/graph_storage/nebula.py index f26e7df6e..d8871f5db 100644 --- a/aperag/indexing/graph_storage/nebula.py +++ b/aperag/indexing/graph_storage/nebula.py @@ -788,6 +788,119 @@ def _read() -> RelationWithLineage | None: return await asyncio.to_thread(_read) + # -- LightRAG-style query layer (Wave 6 #33 chunk 2) -------------- + + async def query_entities_by_keyword( + self, + *, + query: str, + top_k: int, + ) -> list[EntityWithLineage]: + if not query or not query.strip() or top_k <= 0: + return [] + await self.ensure_schema() + + needle = query.strip().lower() + + def _scan() -> list[EntityWithLineage]: + # Nebula has no JSON / text-search index on tag string + # properties; we enumerate all entity VIDs via LOOKUP and + # filter substring-match in Python. For collections with + # very large entity counts this is slower than the + # SQL/Cypher equivalents — acceptable per simple-stable + # directive (no per-collection text-index infra to manage). + matches: list[EntityWithLineage] = [] + for vid in self._list_all_entity_vids(): + row = self._read_entity_lineage_by_vid(vid) + if row is None: + continue + name, type_value, members, parts = row + if needle not in name.lower(): + continue + matches.append( + EntityWithLineage( + name=name, + type=type_value, + source_lineage=tuple(members), + description_parts=tuple(parts), + ) + ) + matches.sort(key=lambda e: e.name) + return matches[:top_k] + + return await asyncio.to_thread(_scan) + + async def expand_neighbors_n_hops( + self, + *, + entity_names: list[str], + hops: int = 1, + ) -> tuple[list[EntityWithLineage], list[RelationWithLineage]]: + if not entity_names: + return ([], []) + await self.ensure_schema() + + def _walk() -> tuple[list[EntityWithLineage], list[RelationWithLineage]]: + seen_entities: dict[str, EntityWithLineage] = {} + seen_relations: dict[tuple[str, str, str], RelationWithLineage] = {} + + def _add_entity(name: str) -> None: + if name in seen_entities: + return + row = self._read_entity_lineage(name) + if row is None: + return + type_value, members, parts = row + seen_entities[name] = EntityWithLineage( + name=name, + type=type_value, + source_lineage=tuple(members), + description_parts=tuple(parts), + ) + + current = {n for n in entity_names if n} + for name in current: + _add_entity(name) + + # Nebula has no edge type for relations (relations are + # tag-vertices). Walk by enumerating all relation VIDs and + # filtering by source/target ∈ current — same approach as + # ``find_relation_keys_with_lineage``. + for _ in range(max(hops, 0)): + next_frontier: set[str] = set() + if not current: + break + for vid in self._list_all_relation_vids(): + row = self._read_relation_lineage_by_vid(vid) + if row is None: + continue + src, tgt, rtype, members, parts = row + if src not in current and tgt not in current: + continue + key = (src, tgt, rtype) + if key not in seen_relations: + seen_relations[key] = RelationWithLineage( + source=src, + target=tgt, + type=rtype, + evidence_lineage=tuple(members), + description_parts=tuple(parts), + ) + for endpoint in (src, tgt): + if endpoint not in seen_entities and endpoint not in next_frontier: + next_frontier.add(endpoint) + if not next_frontier: + break + for name in next_frontier: + _add_entity(name) + current = next_frontier + + entities = sorted(seen_entities.values(), key=lambda e: e.name) + relations = sorted(seen_relations.values(), key=lambda r: (r.source, r.target, r.type)) + return (entities, relations) + + return await asyncio.to_thread(_walk) + __all__ = [ "NebulaLineageGraphStore", diff --git a/aperag/indexing/graph_storage/neo4j.py b/aperag/indexing/graph_storage/neo4j.py index bcc432fe6..31e256c06 100644 --- a/aperag/indexing/graph_storage/neo4j.py +++ b/aperag/indexing/graph_storage/neo4j.py @@ -495,6 +495,130 @@ async def get_relation(self, source: str, target: str, type: str) -> RelationWit ), ) + # -- LightRAG-style query layer (Wave 6 #33 chunk 2) -------------- + + async def query_entities_by_keyword( + self, + *, + query: str, + top_k: int, + ) -> list[EntityWithLineage]: + if not query or not query.strip() or top_k <= 0: + return [] + # Cypher's CONTAINS is case-sensitive; we lowercase both sides + # to honour the Protocol contract of "case-insensitive lexical + # recall". Cypher 5.x has no built-in case-insensitive CONTAINS, + # so this is the canonical idiom. + cypher = ( + f"MATCH (n:{_ENTITY_LABEL} {{collection_id: $collection_id}}) " + f"WHERE toLower(n.name) CONTAINS toLower($query) " + f"RETURN n.name AS name, n.type AS type, " + f" n.source_lineage AS source_lineage, " + f" n.description_parts AS description_parts " + f"ORDER BY n.name " + f"LIMIT $top_k" + ) + async with self._session() as session: + result = await session.run( + cypher, + collection_id=self._collection_id, + query=query.strip(), + top_k=int(top_k), + ) + return [ + EntityWithLineage( + name=rec["name"], + type=rec["type"] or "", + source_lineage=tuple(LineageMember.from_dict(json.loads(s)) for s in (rec["source_lineage"] or [])), + description_parts=tuple( + DescriptionPart.from_dict(json.loads(s)) for s in (rec["description_parts"] or []) + ), + ) + async for rec in result + ] + + async def expand_neighbors_n_hops( + self, + *, + entity_names: list[str], + hops: int = 1, + ) -> tuple[list[EntityWithLineage], list[RelationWithLineage]]: + if not entity_names: + return ([], []) + + seen_entities: dict[str, EntityWithLineage] = {} + seen_relations: dict[tuple[str, str, str], RelationWithLineage] = {} + + async def _fetch_entities(session, names: list[str]) -> None: + if not names: + return + cypher = ( + f"MATCH (n:{_ENTITY_LABEL} {{collection_id: $collection_id}}) " + f"WHERE n.name IN $names " + f"RETURN n.name AS name, n.type AS type, " + f" n.source_lineage AS source_lineage, " + f" n.description_parts AS description_parts" + ) + result = await session.run(cypher, collection_id=self._collection_id, names=names) + async for rec in result: + if rec["name"] in seen_entities: + continue + seen_entities[rec["name"]] = EntityWithLineage( + name=rec["name"], + type=rec["type"] or "", + source_lineage=tuple(LineageMember.from_dict(json.loads(s)) for s in (rec["source_lineage"] or [])), + description_parts=tuple( + DescriptionPart.from_dict(json.loads(s)) for s in (rec["description_parts"] or []) + ), + ) + + async def _fetch_relations_touching(session, names: list[str]) -> set[str]: + if not names: + return set() + cypher = ( + f"MATCH (r:{_RELATION_LABEL} {{collection_id: $collection_id}}) " + f"WHERE r.source IN $names OR r.target IN $names " + f"RETURN r.source AS source, r.target AS target, r.type AS type, " + f" r.evidence_lineage AS evidence_lineage, " + f" r.description_parts AS description_parts" + ) + result = await session.run(cypher, collection_id=self._collection_id, names=names) + next_frontier: set[str] = set() + async for rec in result: + key = (rec["source"], rec["target"], rec["type"]) + if key not in seen_relations: + seen_relations[key] = RelationWithLineage( + source=rec["source"], + target=rec["target"], + type=rec["type"], + evidence_lineage=tuple( + LineageMember.from_dict(json.loads(s)) for s in (rec["evidence_lineage"] or []) + ), + description_parts=tuple( + DescriptionPart.from_dict(json.loads(s)) for s in (rec["description_parts"] or []) + ), + ) + for endpoint in (rec["source"], rec["target"]): + if endpoint not in seen_entities and endpoint not in next_frontier: + next_frontier.add(endpoint) + return next_frontier + + async with self._session() as session: + current = [n for n in entity_names if n] + await _fetch_entities(session, current) + + for _ in range(max(hops, 0)): + next_frontier = await _fetch_relations_touching(session, current) + if not next_frontier: + break + next_list = list(next_frontier) + await _fetch_entities(session, next_list) + current = next_list + + entities = sorted(seen_entities.values(), key=lambda e: e.name) + relations = sorted(seen_relations.values(), key=lambda r: (r.source, r.target, r.type)) + return (entities, relations) + __all__ = [ "Neo4jLineageGraphStore", diff --git a/aperag/indexing/graph_storage/postgres.py b/aperag/indexing/graph_storage/postgres.py index c7ecf7efb..00267dd66 100644 --- a/aperag/indexing/graph_storage/postgres.py +++ b/aperag/indexing/graph_storage/postgres.py @@ -528,6 +528,133 @@ async def get_relation(self, source: str, target: str, type: str) -> RelationWit description_parts=tuple(DescriptionPart.from_dict(part) for part in (row.description_parts or [])), ) + # -- LightRAG-style query layer (Wave 6 #33 chunk 2) -------------- + + async def query_entities_by_keyword( + self, + *, + query: str, + top_k: int, + ) -> list[EntityWithLineage]: + if not query or not query.strip() or top_k <= 0: + return [] + # ILIKE substring match — case-insensitive lexical recall. + # Backend-defined match semantics per Protocol contract. + like = f"%{query.strip()}%" + async with self._engine.connect() as conn: + result = await conn.execute( + select( + _LineageEntityRow.name, + _LineageEntityRow.type, + _LineageEntityRow.source_lineage, + _LineageEntityRow.description_parts, + ) + .where( + _LineageEntityRow.collection_id == self._collection_id, + _LineageEntityRow.name.ilike(like), + ) + .order_by(_LineageEntityRow.name) + .limit(top_k) + ) + return [ + EntityWithLineage( + name=row.name, + type=row.type, + source_lineage=tuple(LineageMember.from_dict(elem) for elem in (row.source_lineage or [])), + description_parts=tuple(DescriptionPart.from_dict(part) for part in (row.description_parts or [])), + ) + for row in result + ] + + async def expand_neighbors_n_hops( + self, + *, + entity_names: list[str], + hops: int = 1, + ) -> tuple[list[EntityWithLineage], list[RelationWithLineage]]: + if not entity_names: + return ([], []) + + seen_entities: dict[str, EntityWithLineage] = {} + seen_relations: dict[tuple[str, str, str], RelationWithLineage] = {} + + async def _fetch_entities(names: set[str]) -> None: + if not names: + return + result = await conn.execute( + select( + _LineageEntityRow.name, + _LineageEntityRow.type, + _LineageEntityRow.source_lineage, + _LineageEntityRow.description_parts, + ).where( + _LineageEntityRow.collection_id == self._collection_id, + _LineageEntityRow.name.in_(list(names)), + ) + ) + for row in result: + if row.name in seen_entities: + continue + seen_entities[row.name] = EntityWithLineage( + name=row.name, + type=row.type, + source_lineage=tuple(LineageMember.from_dict(elem) for elem in (row.source_lineage or [])), + description_parts=tuple(DescriptionPart.from_dict(part) for part in (row.description_parts or [])), + ) + + async def _fetch_relations_touching(names: set[str]) -> set[str]: + """Fetch relations where source or target is in ``names``. + Returns the set of neighbour names (the OTHER endpoint not + already in ``names``) for the next-hop frontier.""" + if not names: + return set() + result = await conn.execute( + select( + _LineageRelationRow.source, + _LineageRelationRow.target, + _LineageRelationRow.type, + _LineageRelationRow.evidence_lineage, + _LineageRelationRow.description_parts, + ).where( + _LineageRelationRow.collection_id == self._collection_id, + (_LineageRelationRow.source.in_(list(names))) | (_LineageRelationRow.target.in_(list(names))), + ) + ) + next_frontier: set[str] = set() + for row in result: + key = (row.source, row.target, row.type) + if key not in seen_relations: + seen_relations[key] = RelationWithLineage( + source=row.source, + target=row.target, + type=row.type, + evidence_lineage=tuple(LineageMember.from_dict(elem) for elem in (row.evidence_lineage or [])), + description_parts=tuple( + DescriptionPart.from_dict(part) for part in (row.description_parts or []) + ), + ) + for endpoint in (row.source, row.target): + if endpoint not in seen_entities and endpoint not in next_frontier: + next_frontier.add(endpoint) + return next_frontier + + async with self._engine.connect() as conn: + current = {n for n in entity_names if n} + await _fetch_entities(current) + + for _ in range(max(hops, 0)): + next_frontier = await _fetch_relations_touching(current) + if not next_frontier: + break + await _fetch_entities(next_frontier) + # next iteration's seeds are the new neighbours so we + # don't re-walk relations from the previous frontier. + current = next_frontier + + entities = sorted(seen_entities.values(), key=lambda e: e.name) + relations = sorted(seen_relations.values(), key=lambda r: (r.source, r.target, r.type)) + return (entities, relations) + __all__ = [ "ENTITY_TABLE", diff --git a/docs/modularization/indexing-redesign-design-pack.md b/docs/modularization/indexing-redesign-design-pack.md index 6f38f4994..0abf01ce5 100644 --- a/docs/modularization/indexing-redesign-design-pack.md +++ b/docs/modularization/indexing-redesign-design-pack.md @@ -1885,7 +1885,7 @@ Phase C (PR-B): └── #39 provider format variations (明书, 2-4 days, p | # | must-be-real | may-be-gated | fully-resolves | |---|---|---|---| -| 33 | LightRAG-style query layer 真实实现 (by-keyword + vector-recall + traversal API) + cascade-migrate 全 callers，行为等价 (existing retrieval search results within ε tolerance) | 无 (full hard-cut per §K.10) | §K.8 Wave 5 backlog item 1 deferred + §K.10 sub-spec | +| 33 | LightRAG-style query layer 真实实现 (by-keyword + traversal API) + cascade-migrate 全 callers，行为等价 within ε tolerance for keyword + n-hop graph recall paths。**Graph entity vector recall NOT in Wave 6 scope** — deferred to Wave 7+ if real evidence surfaces (per architect ruling msg=54eac595 chunk 2 narrowed scope; lineage schema 没承担 entity vector storage，inline coupling 违反 simple-stable #3) | 无 (full hard-cut per §K.10) | §K.8 Wave 5 backlog item 1 deferred + §K.10 sub-spec | | 34 | parser layer chunk_id 字段 schema 一致 + remaining utc_now → CURRENT_TIMESTAMP unify | parser legacy alt-shape transitional decoder OK to keep if needed for legacy fixtures | huangheng T1 obs B + Wave 5 P5B chunk_id 5th item 推迟项 | | 35 | Postgres / Nebula / Neo4j 全 lineage SET storage 改 parallel-list O(N) encoding + alembic migration | 无 — hard-cut policy per earayu2 msg=30c81478 (无生产数据 → schema 直改) | §K.10 item 6 (`feedback_simple_stable_zero_maintenance.md` directive #4 "私有化部署免维护" align — operator deploy 后不管 schema migration since 自动 alembic upgrade) | | 36 | `EntityRecord.type` → `entity_type` Protocol surface rename + Postgres column rename + Cypher property rewrite + Nebula tag-prop rename | 无 — hard-cut | §K.10 item 7 + §D.3 Protocol stability | @@ -1893,21 +1893,27 @@ Phase C (PR-B): └── #39 provider format variations (明书, 2-4 days, p | 38 | `application_runtime.get_application_cache()` 不再静默返 Noop on loop switch；real lazy-rebuild OR explicit metrics counter `cache_noop_due_to_loop_switch_total` + WARN log | 第一次 loop switch 后 rebuild 仍可能小 cost (latency observation OK，但不静默 zero-cache) | huangheng PR #1734 sync point 7 / `feedback_announce_equals_landed.md` (silent zero-cache 是 narrative-truth violation) | | 39 | Voyage / Jina v3 / OpenAI multimodal SDK input format 各自实现 (real provider tests OR documented capability matrix) | LiteLLM canonical shape 是 fallback；provider-specific format 是 polish item，可分 sub-batch incremental ship | §G.2.5.1 spec provider variations footnote | -#### K.11.5. Pre-check pattern lock (per `feedback_spec_lock_grep_verify_caller.md` 双 pattern) +#### K.11.5. Pre-check pattern lock (per `feedback_spec_lock_grep_verify_caller.md` 三 pattern) **Pattern 1 (caller cascade)** 强制触发条件: - **#33**: pre-implementation 必跑 `grep -rn "from aperag.domains.knowledge_graph.graphindex" aperag/` 列全 caller graph，分类 (test-only / new-code-self-ref / legacy-production)；legacy production caller (retrieval/pipeline.py / knowledge_graph/service.py / graph_curation/*) 要全列出迁移到新 query layer 的 mapping - **#36**: pre-implementation 必跑 `grep -rn "EntityRecord.type\|RelationRecord.type\|\.type=" aperag/indexing/ aperag/domains/` 列 Protocol surface caller graph + Cypher template caller graph + Postgres column caller graph，避免 rename 漏 site -**Pattern 2 (state binding)** 强制触发条件: +**Pattern 2 (state binding — runtime config)** 强制触发条件: - **#34**: pre-implementation 必跑 `grep -rn "chunk_id" aperag/indexing/parser*.py aperag/indexing/dispatcher.py aperag/indexing/cleanup.py` 验证 chunk_id 现 binding sites 是否全 align 新 schema - **#35**: pre-implementation 必跑 `grep -rn "_lineage\|set_lineage\|sets_for" aperag/indexing/graph_storage/` 列 lineage SET binding sites + alembic head verify - **#37**: pre-implementation 必跑 `grep -rn "embed_image\|get_or_compute" aperag/llm/embed/ aperag/cache/` 验证 cache infra (`NAMESPACE_EMBEDDING` / `get_or_compute`) 真实存在 (PR #1734 merged ✅) + `embed_image` API 真实存在 (Wave 5 P2 chunk 1 merged ✅) - **#38**: pre-implementation 必跑 `grep -rn "get_application_cache\|Noop" aperag/cache/` 验证 Noop fallback site + loop-switch trigger condition -未通过 pre-check 不允许 push — architect spec lock 时 implementer push 前自跑此 grep + 验证 caller graph 与 spec 一致。 +**Pattern 3 (state binding — Protocol method reads from state X)** 强制触发条件 (added 2026-04-27 per Wave 6 #33 chunk 2 architect ruling msg=54eac595 lesson): + +- 任何 spec acceptance lock 形式 "Protocol method that reads from state X (column / table / index)" — **架构师 必先 grep-verify state X 现 schema 中是否存在/可加**。否则 implementer 实施时 surface state-binding gap，触发 architect ruling round-trip + chunk amend +- **#33 chunk 1 retrospective miss**: §K.11.11 chunk 1 sub-spec 锁 `query_entities_by_vector` Protocol method 时未 grep-verify entity vector storage state binding — Postgres `aperag_lineage_entity` 没 vector column (Wave 4 lineage schema 是 write-side only)。chunk 2 实施时 surface gap → architect Option A ruling msg=54eac595 → narrowed scope (delete `query_entities_by_vector` Protocol method + stub + 2 tests, defer vector recall to Wave 7+ if real evidence surfaces)。Lesson 沉淀到 `feedback_spec_lock_grep_verify_caller.md` Pattern 3 +- **forward**: 未来 Wave/任 architect spec lock for "Protocol method reads state X" 必跑 `grep -rn "state_x_column\|state_x_index" alembic/migrations/ aperag/db/models/` 验证 schema 存在；如不存在 → 4 sub-options (in-chunk schema add / prior chunk add / separate service taking state dep / defer to next Wave) + +未通过 pre-check 不允许 push — architect spec lock 时 implementer push 前自跑此 grep + 验证 caller graph / state binding / Protocol-method-state-binding 与 spec 一致。 #### K.11.6. simple-stable 4 guardrail lock (per `feedback_simple_stable_zero_maintenance.md`) @@ -1983,17 +1989,22 @@ PR-A "Wave 6 graphindex" 是 cross-cutting refactor，独立 architect direct ra **3-chunk decomposition**: -| Chunk | Scope | Owner | Estimated LOC | -|---|---|---|---| -| **chunk 1** | LightRAG-style read Protocol stubs on `LineageGraphStore` (`query_by_keyword` / `query_by_vector` / `expand_neighbors_n_hops`) — NO implementation, NO caller migration; spec ↔ Protocol surface align verify only | Bryce | ~200 LOC docs+protocol | -| **chunk 2** | Protocol implementations across Postgres / Neo4j / Nebula backends + unit tests — backend-specific real query implementations | Bryce | ~600 LOC + tests | -| **chunk 3** | Caller migration (5 files) + legacy graphindex package hard-cut delete + legacy tests delete + grep-zero verify | Bryce | ~400 LOC | +| Chunk | Scope | Owner | Estimated LOC | Status | +|---|---|---|---|---| +| **chunk 1** | LightRAG-style read Protocol stubs on `LineageGraphStore` — NO implementation, NO caller migration; spec ↔ Protocol surface align verify only | Bryce | ~200 LOC docs+protocol | ✅ MERGED PR #1737 commit `20b9071b` 2026-04-27 | +| **chunk 2** | Protocol implementations (`query_entities_by_keyword` + `expand_neighbors_n_hops`) across Postgres / Neo4j / Nebula backends + unit tests + chunk 1 amend (drop `query_entities_by_vector` per architect Option A ruling msg=54eac595) | Bryce | ~600 LOC + tests | 🔄 PR #1741 in_review | +| **chunk 3** | Caller migration (5 files) + legacy graphindex package hard-cut delete + legacy tests delete + grep-zero verify | Bryce | ~400 LOC | ⏳ post-chunk-2 | + +**Chunk 1 → chunk 2 amend note (architect ruling msg=54eac595 — Option A narrowed scope)**: +- chunk 1 originally declared 3 Protocol methods (keyword + vector + traversal). chunk 2 backend impl surface state-binding gap: lineage schema 没 entity vector column (Wave 4 design choice, write-side only) +- Architect ruling Option A (msg=54eac595): chunk 2 amends chunk 1 — drop `query_entities_by_vector` Protocol method declaration + InMemoryLineageGraphStore stub + 2 contract tests. Vector recall deferred to Wave 7+ if real evidence。 +- Lesson: chunk 1 spec lock 时 architect 应先 grep-verify entity vector storage schema → Pattern 3 (Protocol method state binding) added to §K.11.5 + sediment 到 `feedback_spec_lock_grep_verify_caller.md` **Each chunk lands independently within PR-A** (chunked-rotation per `feedback_no_refresh_complete_all_tasks.md` Layer 1 same-session continuation directive)。Each chunk has own architect direct ratify gate per §K.11.8 lane lock。 **Why 3 chunks (not 1 monolith)**: (a) chunk 1 spec/Protocol-only allows architect ratify to lock query layer surface before backend implementations; (b) chunk 2 backend impl can independently verify Protocol semantics on real engines (per `feedback_dataflow_review.md` dataflow trace each backend); (c) chunk 3 caller migration is pure mechanical refactor once Protocol stable — chunk 4d Option C precedent (msg=b26f64b2 narrowed scope) avoids "spec drift mid-cascade" anti-pattern。 -PR-A 的 acceptance items 即 §K.10 (1-5)：query layer + cascade migrate + delete legacy package + delete legacy tests + grep-zero verify。 +PR-A 的 acceptance items 即 §K.10 (1-5) + amended #33 must-be-real per §K.11.4 (graph entity vector recall not in scope; keyword + traversal sufficient for retrieval/curation existing behaviors)。 #### K.11.12. Wave 6 close-out gate diff --git a/tests/unit_test/indexing/test_lineage_query_protocol.py b/tests/unit_test/indexing/test_lineage_query_protocol.py index e75ec2ec3..ee3e06425 100644 --- a/tests/unit_test/indexing/test_lineage_query_protocol.py +++ b/tests/unit_test/indexing/test_lineage_query_protocol.py @@ -12,18 +12,21 @@ # See the License for the specific language governing permissions and # limitations under the License. -"""Wave 6 #33 chunk 1 — Protocol surface for the LightRAG-style query -layer on :class:`aperag.indexing.graph.LineageGraphStore`. +"""Wave 6 #33 — Protocol surface + in-memory implementation for the +LightRAG-style query layer on :class:`aperag.indexing.graph.LineageGraphStore`. -Pin the Protocol method signatures so chunk 2 (cross-backend impl) can -be reviewed against a stable surface, and so callers (chunk 3 caller -migration) can rely on the methods existing even before the per-backend -implementations land. +Pin the Protocol method signatures so chunk 3 caller migration can +rely on a stable surface, and pin the in-memory query/traversal +semantics that production backends must match. -Per §K.11.11 sub-spec the read API is: +Per architect ruling msg=54eac595 (Option A) the read API is: * ``query_entities_by_keyword(query, top_k)`` — lexical recall -* ``query_entities_by_vector(embedding, top_k)`` — vector recall * ``expand_neighbors_n_hops(entity_names, hops)`` — graph traversal + +``query_entities_by_vector`` was originally declared in chunk 1 but +dropped in chunk 2 because the Wave 4 lineage schema does not carry an +entity-vector column. Wave 7 may re-introduce vector recall via a +separate ``LightRAGQueryService`` if real evidence surfaces. """ from __future__ import annotations @@ -31,23 +34,25 @@ import asyncio import inspect -import pytest - from aperag.indexing.graph import ( + EntityRecord, InMemoryLineageGraphStore, LineageGraphStore, + LineageMember, + RelationRecord, ) -def test_lineage_query_protocol_exposes_three_read_methods(): - """The new query layer surface must declare exactly the three - LightRAG-style read methods agreed in §K.11.11. Adding a method - here implies a chunk 2 backend implementation contract.""" +def test_lineage_query_protocol_exposes_two_read_methods(): + """The new query layer surface must declare exactly the two + LightRAG-style read methods agreed in §K.11.11 amend.""" members = set(LineageGraphStore.__dict__) assert "query_entities_by_keyword" in members - assert "query_entities_by_vector" in members assert "expand_neighbors_n_hops" in members + # Vector recall was removed by architect ruling msg=54eac595 — + # no entity-vector column in Wave 4 lineage schema. + assert "query_entities_by_vector" not in members def test_query_by_keyword_signature_takes_query_and_top_k(): @@ -60,15 +65,6 @@ def test_query_by_keyword_signature_takes_query_and_top_k(): assert params["top_k"].kind is inspect.Parameter.KEYWORD_ONLY -def test_query_by_vector_signature_takes_embedding_and_top_k(): - sig = inspect.signature(LineageGraphStore.query_entities_by_vector) - params = sig.parameters - assert "embedding" in params - assert "top_k" in params - assert params["embedding"].kind is inspect.Parameter.KEYWORD_ONLY - assert params["top_k"].kind is inspect.Parameter.KEYWORD_ONLY - - def test_expand_neighbors_signature_takes_entity_names_and_hops(): sig = inspect.signature(LineageGraphStore.expand_neighbors_n_hops) params = sig.parameters @@ -81,25 +77,138 @@ def test_expand_neighbors_signature_takes_entity_names_and_hops(): assert params["hops"].default == 1 -def test_in_memory_store_query_methods_raise_not_implemented_in_chunk_1(): - """Chunk 1 ships Protocol stubs only — the in-memory test store - must surface a clear NotImplementedError so callers (chunk 3 - migration) can rely on the gate firing if they accidentally - invoke the API before chunk 2 lands.""" +def _seed_minimal_graph(store: InMemoryLineageGraphStore) -> None: + """Three entities (Alice / Bob / Carol) connected by two + relations: Alice -knows-> Bob, Bob -manages-> Carol. Used by the + in-memory contract tests below.""" + async def _run() -> None: + members = LineageMember( + document_id="doc-1", + parse_version="v1", + tenant_scope_key="user:test", + chunk_ids=("chunk-1",), + ) + for name in ("Alice", "Bob", "Carol"): + await store.upsert_entity_with_lineage( + record=EntityRecord( + name=name, + type="person", + description=f"{name} description", + source_chunk_ids=("chunk-1",), + ), + lineage=members, + ) + await store.upsert_relation_with_lineage( + record=RelationRecord( + source="Alice", + target="Bob", + type="knows", + description="knows", + source_chunk_ids=("chunk-1",), + ), + lineage=members, + ) + await store.upsert_relation_with_lineage( + record=RelationRecord( + source="Bob", + target="Carol", + type="manages", + description="manages", + source_chunk_ids=("chunk-1",), + ), + lineage=members, + ) + + asyncio.run(_run()) + + +def test_in_memory_query_by_keyword_finds_substring_matches(): store = InMemoryLineageGraphStore() + _seed_minimal_graph(store) async def _run() -> None: - with pytest.raises(NotImplementedError) as exc: - await store.query_entities_by_keyword(query="anything", top_k=5) - assert "chunk 2" in str(exc.value) + hits = await store.query_entities_by_keyword(query="ali", top_k=5) + assert [e.name for e in hits] == ["Alice"] + + hits = await store.query_entities_by_keyword(query="A", top_k=5) + # Substring "A" matches "Alice" + "Carol"; sorted by name. + assert [e.name for e in hits] == ["Alice", "Carol"] - with pytest.raises(NotImplementedError) as exc: - await store.query_entities_by_vector(embedding=[0.1, 0.2, 0.3], top_k=5) - assert "chunk 2" in str(exc.value) + asyncio.run(_run()) - with pytest.raises(NotImplementedError) as exc: - await store.expand_neighbors_n_hops(entity_names=["X"], hops=1) - assert "chunk 2" in str(exc.value) + +def test_in_memory_query_by_keyword_empty_or_zero_top_k_returns_empty(): + store = InMemoryLineageGraphStore() + _seed_minimal_graph(store) + + async def _run() -> None: + assert await store.query_entities_by_keyword(query="", top_k=5) == [] + assert await store.query_entities_by_keyword(query=" ", top_k=5) == [] + assert await store.query_entities_by_keyword(query="A", top_k=0) == [] + + asyncio.run(_run()) + + +def test_in_memory_expand_neighbors_one_hop_includes_seed_and_neighbours(): + store = InMemoryLineageGraphStore() + _seed_minimal_graph(store) + + async def _run() -> None: + entities, relations = await store.expand_neighbors_n_hops(entity_names=["Alice"], hops=1) + names = {e.name for e in entities} + assert names == {"Alice", "Bob"} + rel_keys = {(r.source, r.target, r.type) for r in relations} + assert rel_keys == {("Alice", "Bob", "knows")} + + asyncio.run(_run()) + + +def test_in_memory_expand_neighbors_two_hops_walks_further(): + store = InMemoryLineageGraphStore() + _seed_minimal_graph(store) + + async def _run() -> None: + entities, relations = await store.expand_neighbors_n_hops(entity_names=["Alice"], hops=2) + names = {e.name for e in entities} + # 2-hop reaches Carol via Alice→Bob→Carol. + assert names == {"Alice", "Bob", "Carol"} + rel_keys = {(r.source, r.target, r.type) for r in relations} + assert rel_keys == {("Alice", "Bob", "knows"), ("Bob", "Carol", "manages")} + + asyncio.run(_run()) + + +def test_in_memory_expand_neighbors_zero_hops_returns_seed_only(): + store = InMemoryLineageGraphStore() + _seed_minimal_graph(store) + + async def _run() -> None: + entities, relations = await store.expand_neighbors_n_hops(entity_names=["Bob"], hops=0) + assert {e.name for e in entities} == {"Bob"} + assert relations == [] + + asyncio.run(_run()) + + +def test_in_memory_expand_neighbors_empty_input_returns_empty_pair(): + store = InMemoryLineageGraphStore() + _seed_minimal_graph(store) + + async def _run() -> None: + result = await store.expand_neighbors_n_hops(entity_names=[], hops=1) + assert result == ([], []) + + asyncio.run(_run()) + + +def test_in_memory_expand_neighbors_unknown_seed_returns_empty_entities(): + store = InMemoryLineageGraphStore() + _seed_minimal_graph(store) + + async def _run() -> None: + entities, relations = await store.expand_neighbors_n_hops(entity_names=["Unknown"], hops=1) + assert entities == [] + assert relations == [] asyncio.run(_run())