跳转至

Retriever 接口 (Retrievers)

检索器负责从 LightRAG 知识图谱中获取相关文档。每个检索器封装一种查询模式的检索策略，通过向量相似度搜索和/或图遍历获取实体、关系和文本块，返回 LangChain Document 对象。

基类

lightrag_langchain.retriever.base.LightRAGBaseRetriever

Bases: BaseRetriever

所有 LightRAG 查询模式 Retriever 的抽象基类。

封装共享基础设施（embedding 生成、异步桥接、错误处理），使每个模式特定子类 只需实现 _aget_relevant_documents 及其自身的策略调用和 Document 转换逻辑 (D-06)。

Parameters

vector_store: 用于向量相似度搜索的 PGVectorStore 实例 (D-01)。 embedding_config: 用于延迟创建 embedding 模型的 EmbeddingConfig (D-02)。 graph_store: 用于图查询的 PGGraphStore 实例。可选 — naive 和 bypass 模式不需要（默认 None）。 top_k: 覆盖全局 top_k。当为 None 时，Retriever 使用 Settings 级别的默认值 (D-03)。 chunk_top_k: 覆盖全局 chunk_top_k。当为 None 时，Retriever 使用 Settings 级别的默认值 (D-03)。

Example 
from lightrag_langchain.retriever import NaiveRetriever
from lightrag_langchain.config import settings

retriever = NaiveRetriever(
    vector_store=vector_store,
    embedding_config=settings.embedding,
)
docs = await retriever.ainvoke("your query")
源代码位于： src/lightrag_langchain/retriever/base.py 
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
class LightRAGBaseRetriever(BaseRetriever):
    """所有 LightRAG 查询模式 Retriever 的抽象基类。

    封装共享基础设施（embedding 生成、异步桥接、错误处理），使每个模式特定子类
    只需实现 ``_aget_relevant_documents`` 及其自身的策略调用和 Document 转换逻辑 (D-06)。

    Parameters
    ----------
    vector_store:
        用于向量相似度搜索的 PGVectorStore 实例 (D-01)。
    embedding_config:
        用于延迟创建 embedding 模型的 EmbeddingConfig (D-02)。
    graph_store:
        用于图查询的 PGGraphStore 实例。可选 — naive 和 bypass 模式不需要（默认 ``None``）。
    top_k:
        覆盖全局 top_k。当为 ``None`` 时，Retriever 使用 Settings 级别的默认值 (D-03)。
    chunk_top_k:
        覆盖全局 chunk_top_k。当为 ``None`` 时，Retriever 使用 Settings 级别的默认值 (D-03)。

    Example:
        ```python
        from lightrag_langchain.retriever import NaiveRetriever
        from lightrag_langchain.config import settings

        retriever = NaiveRetriever(
            vector_store=vector_store,
            embedding_config=settings.embedding,
        )
        docs = await retriever.ainvoke("your query")
        ```
    """

    model_config = ConfigDict(arbitrary_types_allowed=True)

    vector_store: PGVectorStore
    """用于向量相似度搜索的 PGVectorStore (D-01 构造函数注入)。"""

    embedding_config: EmbeddingConfig
    """Embedding provider 配置；由延迟 ``embedding`` 属性消费 (D-02)。"""

    graph_store: PGGraphStore | None = None
    """用于图查询的 PGGraphStore；可选（naive/bypass 不需要）。"""

    top_k: int | None = None
    """覆盖全局 top_k；``None`` 表示使用 Settings 默认值 (D-03)。"""

    chunk_top_k: int | None = None
    """覆盖 chunk_top_k；``None`` 表示使用 Settings 默认值 (D-03)。"""

    # ------------------------------------------------------------------
    # Private attributes
    # ------------------------------------------------------------------

    _embedding: OpenAIEmbeddings | None = PrivateAttr(default=None)
    """延迟初始化的 embedding 模型 (D-02)。"""

    _logger: logging.Logger = PrivateAttr(
        default_factory=lambda: logging.getLogger(__name__)
    )
    """每个实例的日志记录器，用于警告和错误。"""

    # ------------------------------------------------------------------
    # Embedding (lazy init, D-02)
    # ------------------------------------------------------------------

    @property
    def embedding(self) -> OpenAIEmbeddings:
        """返回 OpenAIEmbeddings 实例，在首次访问时创建。

        使用 :func:`create_embedding(self.embedding_config)`，它返回一个
        ``_LazyEmbedding`` 代理 — 实际的 ``OpenAIEmbeddings`` 构造延迟到
        首次访问返回对象的属性时。导入时无网络调用 (D-02)。
        """
        if self._embedding is None:
            from lightrag_langchain.llm import create_embedding

            self._embedding = create_embedding(self.embedding_config)
        return self._embedding

    # ------------------------------------------------------------------
    # Sync / async bridge
    # ------------------------------------------------------------------

    def _get_relevant_documents(
        self, query: str, *, run_manager=None, **kwargs
    ) -> list[Document]:
        """同步路径 — 使用 ``asyncio.run`` 桥接到异步实现。

        匹配 :class:`LightRAGReranker.compress_documents` 模式
        (:file:`reranker.py` 第 357 行)。
        """
        return asyncio.run(
            self._aget_relevant_documents(query, run_manager=run_manager, **kwargs)
        )

    # ------------------------------------------------------------------
    # Abstract: subclasses MUST override
    # ------------------------------------------------------------------

    @abc.abstractmethod
    async def _aget_relevant_documents(
        self, query: str, *, run_manager=None, **kwargs
    ) -> list[Document]:
        """异步检索 — 子类通过模式特定逻辑覆盖。

        每个实现应：
        1. 通过 :meth:`_generate_query_embedding` 生成查询 embedding
        2. 调用其模式特定的 Phase 4 策略函数
        3. 将返回的 :class:`QueryResult` 转换为 ``list[Document]``
        """

    # ------------------------------------------------------------------
    # Helper: query embedding
    # ------------------------------------------------------------------

    def _generate_query_embedding(self, query: str) -> list[float]:
        """通过 embedding 模型将 *query* 编码为稠密向量。

        Returns
        -------
        list[float]
            适用于 pgvector ``<=>`` 余弦距离搜索和 Phase 4 策略函数调用的
            embedding 向量。
        """
        return self.embedding.embed_query(query)

vector_store instance-attribute 

vector_store: PGVectorStore

用于向量相似度搜索的 PGVectorStore (D-01 构造函数注入)。

embedding_config instance-attribute 

embedding_config: EmbeddingConfig

Embedding provider 配置；由延迟 embedding 属性消费 (D-02)。

graph_store class-attribute instance-attribute 

graph_store: PGGraphStore | None = None

用于图查询的 PGGraphStore；可选（naive/bypass 不需要）。

top_k class-attribute instance-attribute 

top_k: int | None = None

覆盖全局 top_k；None 表示使用 Settings 默认值 (D-03)。

chunk_top_k class-attribute instance-attribute 

chunk_top_k: int | None = None

覆盖 chunk_top_k；None 表示使用 Settings 默认值 (D-03)。

embedding property 

embedding: OpenAIEmbeddings

返回 OpenAIEmbeddings 实例，在首次访问时创建。

使用 :func:create_embedding(self.embedding_config)，它返回一个 _LazyEmbedding 代理 — 实际的 OpenAIEmbeddings 构造延迟到 首次访问返回对象的属性时。导入时无网络调用 (D-02)。

查询模式 Retriever 子类

lightrag_langchain.retriever.retrievers.NaiveRetriever

Bases: LightRAGBaseRetriever

LightRAG naive 查询模式的 LangChain Retriever。

在 chunks_vdb 表上执行纯向量相似度搜索。 不进行图遍历 — 仅返回 chunk Document。

RETR-01 (naive mode)。graph_store 不使用（始终为 None）。

Example 
from lightrag_langchain.retriever import NaiveRetriever

retriever = NaiveRetriever(
    vector_store=vector_store,
    embedding_config=settings.embedding,
)
docs = await retriever.ainvoke("your query")
源代码位于： src/lightrag_langchain/retriever/retrievers.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
class NaiveRetriever(LightRAGBaseRetriever):
    """LightRAG **naive** 查询模式的 LangChain Retriever。

    在 chunks_vdb 表上执行纯向量相似度搜索。
    不进行图遍历 — 仅返回 chunk Document。

    RETR-01 (naive mode)。``graph_store`` 不使用（始终为 ``None``）。

    Example:
        ```python
        from lightrag_langchain.retriever import NaiveRetriever

        retriever = NaiveRetriever(
            vector_store=vector_store,
            embedding_config=settings.embedding,
        )
        docs = await retriever.ainvoke("your query")
        ```
    """

    async def _aget_relevant_documents(
        self, query: str, *, run_manager=None, **kwargs
    ) -> list[Document]:
        """异步 naive 检索：embed query → 向量搜索 chunks → 转换。"""
        embedding = self._generate_query_embedding(query)

        from lightrag_langchain.query.strategies import naive_strategy

        result = await naive_strategy(
            embedding,
            vector_store=self.vector_store,
            chunk_top_k=self.chunk_top_k,
        )

        if not result.chunks:
            self._logger.warning(
                "NaiveRetriever: no chunks found for query %r", query[:120]
            )

        return [
            chunk_to_document(c, retrieval_mode="naive") for c in result.chunks
        ]

lightrag_langchain.retriever.retrievers.LocalRetriever

Bases: LightRAGBaseRetriever

LightRAG local 查询模式的 LangChain Retriever。

在 entities_vdb 中搜索 top-K 实体，扩展到 AGE 图中发现边/邻居， 返回 entity + GraphTriple Document。

RETR-01 (local mode)。需要同时 vector_store 和 graph_store。

Example 
from lightrag_langchain.retriever import LocalRetriever

retriever = LocalRetriever(
    vector_store=vector_store,
    embedding_config=settings.embedding,
    graph_store=graph_store,
)
docs = await retriever.ainvoke("your query")
源代码位于： src/lightrag_langchain/retriever/retrievers.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
class LocalRetriever(LightRAGBaseRetriever):
    """LightRAG **local** 查询模式的 LangChain Retriever。

    在 entities_vdb 中搜索 top-K 实体，扩展到 AGE 图中发现边/邻居，
    返回 entity + GraphTriple Document。

    RETR-01 (local mode)。需要同时 ``vector_store`` 和 ``graph_store``。

    Example:
        ```python
        from lightrag_langchain.retriever import LocalRetriever

        retriever = LocalRetriever(
            vector_store=vector_store,
            embedding_config=settings.embedding,
            graph_store=graph_store,
        )
        docs = await retriever.ainvoke("your query")
        ```
    """

    async def _aget_relevant_documents(
        self, query: str, *, run_manager=None, **kwargs
    ) -> list[Document]:
        """异步 local 检索：embed → 实体搜索 → 图扩展 → 转换。"""
        embedding = self._generate_query_embedding(query)

        from lightrag_langchain.query.strategies import local_strategy

        result = await local_strategy(
            embedding,
            vector_store=self.vector_store,
            graph_store=self.graph_store,
            top_k=self.top_k,
        )

        if not result.entities and not result.graph_triples:
            self._logger.warning(
                "LocalRetriever: no results found for query %r", query[:120]
            )

        node_lookup, _edge_lookup = build_graph_lookups(result.graph_triples)

        # Entity Documents — enrich with GraphNode entity_type/description
        entity_docs: list[Document] = []
        for entity in result.entities:
            node = node_lookup.get(entity.entity_name)
            if node is not None:
                doc = entity_to_document(
                    entity,
                    entity_type=node.entity_type,
                    description=node.description,
                    retrieval_mode="local",
                )
            else:
                doc = entity_to_document(entity, retrieval_mode="local")
            entity_docs.append(doc)

        # GraphTriple Documents
        triple_docs = [
            graph_triple_to_document(t, retrieval_mode="local")
            for t in result.graph_triples
        ]

        return entity_docs + triple_docs

lightrag_langchain.retriever.retrievers.GlobalRetriever

Bases: LightRAGBaseRetriever

LightRAG global 查询模式的 LangChain Retriever。

在 relationships_vdb 中搜索 top-K 关系，批量从 AGE 图中检索边数据， 返回 relation + GraphTriple Document。

RETR-01 (global mode)。需要同时 vector_store 和 graph_store。

Example 
from lightrag_langchain.retriever import GlobalRetriever

retriever = GlobalRetriever(
    vector_store=vector_store,
    embedding_config=settings.embedding,
    graph_store=graph_store,
)
docs = await retriever.ainvoke("your query")
源代码位于： src/lightrag_langchain/retriever/retrievers.py 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
class GlobalRetriever(LightRAGBaseRetriever):
    """LightRAG **global** 查询模式的 LangChain Retriever。

    在 relationships_vdb 中搜索 top-K 关系，批量从 AGE 图中检索边数据，
    返回 relation + GraphTriple Document。

    RETR-01 (global mode)。需要同时 ``vector_store`` 和 ``graph_store``。

    Example:
        ```python
        from lightrag_langchain.retriever import GlobalRetriever

        retriever = GlobalRetriever(
            vector_store=vector_store,
            embedding_config=settings.embedding,
            graph_store=graph_store,
        )
        docs = await retriever.ainvoke("your query")
        ```
    """

    async def _aget_relevant_documents(
        self, query: str, *, run_manager=None, **kwargs
    ) -> list[Document]:
        """异步 global 检索：embed → 关系搜索 → 丰富 → 转换。"""
        embedding = self._generate_query_embedding(query)

        from lightrag_langchain.query.strategies import global_strategy

        result = await global_strategy(
            embedding,
            vector_store=self.vector_store,
            graph_store=self.graph_store,
            top_k=self.top_k,
        )

        if not result.relations and not result.graph_triples:
            self._logger.warning(
                "GlobalRetriever: no results found for query %r", query[:120]
            )

        _node_lookup, edge_lookup = build_graph_lookups(result.graph_triples)

        # Relation Documents — enrich with GraphEdge keywords/weight/source_id
        relation_docs: list[Document] = []
        for relation in result.relations:
            edge = edge_lookup.get((relation.src_id, relation.tgt_id))
            if edge is not None:
                doc = relation_to_document(
                    relation,
                    keywords=edge.keywords or "",
                    weight=edge.weight,
                    source_id=edge.source_id or "",
                    retrieval_mode="global",
                )
            else:
                doc = relation_to_document(relation, retrieval_mode="global")
            relation_docs.append(doc)

        # GraphTriple Documents
        triple_docs = [
            graph_triple_to_document(t, retrieval_mode="global")
            for t in result.graph_triples
        ]

        return relation_docs + triple_docs

lightrag_langchain.retriever.retrievers.HybridRetriever

Bases: LightRAGBaseRetriever

LightRAG hybrid 查询模式的 LangChain Retriever。

并行运行 local 和 global 策略，然后轮询合并 entities 和 relations。 返回 entity + relation + GraphTriple Document。

RETR-01 (hybrid mode)。需要同时 vector_store 和 graph_store。

Example 
from lightrag_langchain.retriever import HybridRetriever

retriever = HybridRetriever(
    vector_store=vector_store,
    embedding_config=settings.embedding,
    graph_store=graph_store,
)
docs = await retriever.ainvoke("your query")
源代码位于： src/lightrag_langchain/retriever/retrievers.py 
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
class HybridRetriever(LightRAGBaseRetriever):
    """LightRAG **hybrid** 查询模式的 LangChain Retriever。

    并行运行 local 和 global 策略，然后轮询合并 entities 和 relations。
    返回 entity + relation + GraphTriple Document。

    RETR-01 (hybrid mode)。需要同时 ``vector_store`` 和 ``graph_store``。

    Example:
        ```python
        from lightrag_langchain.retriever import HybridRetriever

        retriever = HybridRetriever(
            vector_store=vector_store,
            embedding_config=settings.embedding,
            graph_store=graph_store,
        )
        docs = await retriever.ainvoke("your query")
        ```
    """

    async def _aget_relevant_documents(
        self, query: str, *, run_manager=None, **kwargs
    ) -> list[Document]:
        """异步 hybrid 检索：embed → 并行 local+global → 合并 → 转换。"""
        embedding = self._generate_query_embedding(query)

        from lightrag_langchain.query.strategies import hybrid_strategy

        result = await hybrid_strategy(
            embedding,
            vector_store=self.vector_store,
            graph_store=self.graph_store,
            top_k=self.top_k,
        )

        if (
            not result.entities
            and not result.relations
            and not result.graph_triples
        ):
            self._logger.warning(
                "HybridRetriever: no results found for query %r", query[:120]
            )

        node_lookup, edge_lookup = build_graph_lookups(result.graph_triples)

        # Entity Documents — enrich with GraphNode entity_type/description
        entity_docs: list[Document] = []
        for entity in result.entities:
            node = node_lookup.get(entity.entity_name)
            if node is not None:
                doc = entity_to_document(
                    entity,
                    entity_type=node.entity_type,
                    description=node.description,
                    retrieval_mode="hybrid",
                )
            else:
                doc = entity_to_document(entity, retrieval_mode="hybrid")
            entity_docs.append(doc)

        # Relation Documents — enrich with GraphEdge keywords/weight/source_id
        relation_docs: list[Document] = []
        for relation in result.relations:
            edge = edge_lookup.get((relation.src_id, relation.tgt_id))
            if edge is not None:
                doc = relation_to_document(
                    relation,
                    keywords=edge.keywords or "",
                    weight=edge.weight,
                    source_id=edge.source_id or "",
                    retrieval_mode="hybrid",
                )
            else:
                doc = relation_to_document(relation, retrieval_mode="hybrid")
            relation_docs.append(doc)

        # GraphTriple Documents
        triple_docs = [
            graph_triple_to_document(t, retrieval_mode="hybrid")
            for t in result.graph_triples
        ]

        return entity_docs + relation_docs + triple_docs

lightrag_langchain.retriever.retrievers.MixRetriever

Bases: LightRAGBaseRetriever

LightRAG mix 查询模式的 LangChain Retriever。

并行运行 hybrid 策略和 chunk 向量搜索，通过轮询将实体伪块与文本块合并。 返回 entity + relation + chunk + GraphTriple Document。

RETR-01 (mix mode)。需要同时 vector_store 和 graph_store。

Example 
from lightrag_langchain.retriever import MixRetriever

retriever = MixRetriever(
    vector_store=vector_store,
    embedding_config=settings.embedding,
    graph_store=graph_store,
)
docs = await retriever.ainvoke("your query")
源代码位于： src/lightrag_langchain/retriever/retrievers.py 
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
class MixRetriever(LightRAGBaseRetriever):
    """LightRAG **mix** 查询模式的 LangChain Retriever。

    并行运行 hybrid 策略和 chunk 向量搜索，通过轮询将实体伪块与文本块合并。
    返回 entity + relation + chunk + GraphTriple Document。

    RETR-01 (mix mode)。需要同时 ``vector_store`` 和 ``graph_store``。

    Example:
        ```python
        from lightrag_langchain.retriever import MixRetriever

        retriever = MixRetriever(
            vector_store=vector_store,
            embedding_config=settings.embedding,
            graph_store=graph_store,
        )
        docs = await retriever.ainvoke("your query")
        ```
    """

    async def _aget_relevant_documents(
        self, query: str, *, run_manager=None, **kwargs
    ) -> list[Document]:
        """异步 mix 检索：embed → mix 策略 → 转换全部四种类型。"""
        embedding = self._generate_query_embedding(query)

        from lightrag_langchain.query.strategies import mix_strategy

        result = await mix_strategy(
            embedding,
            vector_store=self.vector_store,
            graph_store=self.graph_store,
            top_k=self.top_k,
            chunk_top_k=self.chunk_top_k,
        )

        if (
            not result.entities
            and not result.relations
            and not result.chunks
            and not result.graph_triples
        ):
            self._logger.warning(
                "MixRetriever: no results found for query %r", query[:120]
            )

        node_lookup, edge_lookup = build_graph_lookups(result.graph_triples)

        # Entity Documents — enrich with GraphNode entity_type/description
        entity_docs: list[Document] = []
        for entity in result.entities:
            node = node_lookup.get(entity.entity_name)
            if node is not None:
                doc = entity_to_document(
                    entity,
                    entity_type=node.entity_type,
                    description=node.description,
                    retrieval_mode="mix",
                )
            else:
                doc = entity_to_document(entity, retrieval_mode="mix")
            entity_docs.append(doc)

        # Relation Documents — enrich with GraphEdge keywords/weight/source_id
        relation_docs: list[Document] = []
        for relation in result.relations:
            edge = edge_lookup.get((relation.src_id, relation.tgt_id))
            if edge is not None:
                doc = relation_to_document(
                    relation,
                    keywords=edge.keywords or "",
                    weight=edge.weight,
                    source_id=edge.source_id or "",
                    retrieval_mode="mix",
                )
            else:
                doc = relation_to_document(relation, retrieval_mode="mix")
            relation_docs.append(doc)

        # Chunk Documents
        chunk_docs = [
            chunk_to_document(c, retrieval_mode="mix") for c in result.chunks
        ]

        # GraphTriple Documents
        triple_docs = [
            graph_triple_to_document(t, retrieval_mode="mix")
            for t in result.graph_triples
        ]

        return entity_docs + relation_docs + chunk_docs + triple_docs

lightrag_langchain.retriever.retrievers.BypassRetriever

Bases: LightRAGBaseRetriever

LightRAG bypass 查询模式的 LangChain Retriever。

不进行检索 — 返回空的 list[Document]。无 embedding 生成、无策略调用、 无数据库 I/O。

RETR-01 (bypass mode)。vector_store 和 graph_store 不使用。

Example 
from lightrag_langchain.retriever import BypassRetriever

retriever = BypassRetriever(
    vector_store=vector_store,
    embedding_config=settings.embedding,
)
docs = await retriever.ainvoke("your query")  # 始终返回 []
源代码位于： src/lightrag_langchain/retriever/retrievers.py 
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
class BypassRetriever(LightRAGBaseRetriever):
    """LightRAG **bypass** 查询模式的 LangChain Retriever。

    不进行检索 — 返回空的 ``list[Document]``。无 embedding 生成、无策略调用、
    无数据库 I/O。

    RETR-01 (bypass mode)。``vector_store`` 和 ``graph_store`` 不使用。

    Example:
        ```python
        from lightrag_langchain.retriever import BypassRetriever

        retriever = BypassRetriever(
            vector_store=vector_store,
            embedding_config=settings.embedding,
        )
        docs = await retriever.ainvoke("your query")  # 始终返回 []
        ```
    """

    async def _aget_relevant_documents(
        self, query: str, *, run_manager=None, **kwargs
    ) -> list[Document]:
        """异步 bypass — 返回空列表。无 I/O。"""
        return []

    def _get_relevant_documents(
        self, query: str, *, run_manager=None, **kwargs
    ) -> list[Document]:
        """同步 bypass — 返回空列表。跳过 ``asyncio.run`` 开销。"""
        return []