10.4 · SearchFilters 与时序查询

SearchFilters 与时序查询

搜索过滤器与时序查询

相关源文件

本章引用的主要源码文件：

• examples/podcast/podcast_runner.py
• graphiti_core/search/search.py
• graphiti_core/search/search_config.py
• graphiti_core/search/search_config_recipes.py
• graphiti_core/search/search_filters.py
• graphiti_core/search/search_utils.py
• tests/test_graphiti_int.py

本文档记录了 SearchFilters Pydantic 模型及其所有关联的过滤器类型、运算符和 graphiti_core/search/search_filters.py 中的查询构造工具。它解释了如何通过节点标签、边类型、特定边 UUID、属性值和时序范围（事实有效时间、创建时间和过期时间）来限定搜索范围。

有关搜索架构的整体概述（包括搜索方法、重排序器和 SearchConfig），请参见第 4.3 页。有关调整搜索配置选项，请参见第 10.2 页。有关作为时序过滤基础的双时序数据模型，请参见第 3.2 页。

概述

Graphiti 中的每个搜索调用都接受一个 SearchFilters 实例，该实例会贯穿所有底层搜索方法（BM25 全文搜索、余弦相似度、BFS 遍历）。过滤器会在数据库查询执行前转换为 Cypher WHERE 子句。不同的图数据库提供商（Neo4j、FalkorDB、Kuzu、Neptune）会收到略有不同的语法。

数据模型图：SearchFilters 及其组件

来源：graphiti_core/search/search_filters.py:55-67

ComparisonOperator

ComparisonOperator 是一个 Python Enum，其值为 Cypher 运算符字符串。

当使用 is_null 或 is_not_null 时，DateFilter 上的 date 字段（或 PropertyFilter 上的 property_value 字段）会被忽略——不会向查询中发出参数。

来源：graphiti_core/search/search_filters.py:27-35

DateFilter

DateFilter 将一个 datetime 值与一个 ComparisonOperator 配对。它始终在 SearchFilters 的 list[list[DateFilter]] 字段内使用——从不单独使用。

class DateFilter(BaseModel):
    date: datetime | None = Field(default=None, description='用于过滤的日期时间')
    comparison_operator: ComparisonOperator = Field(
        description='日期过滤器的比较运算符'
    )

来源：graphiti_core/search/search_filters.py:38-42

PropertyFilter

PropertyFilter 允许对任意边或节点属性值进行过滤。

class PropertyFilter(BaseModel):
    property_name: str = Field(description='属性名称')
    property_value: str | int | float | None = Field(
        default=None, description='要匹配的属性值'
    )
    comparison_operator: ComparisonOperator = Field(
        description='属性的比较运算符'
    )

来源：graphiti_core/search/search_filters.py:45-52

SearchFilters

SearchFilters 是一个 Pydantic BaseModel，它将所有过滤条件聚合到一个单一对象中，并传递给每个搜索函数。它包含一个针对 node_labels 的验证器，以防止 Cypher 注入。

来源：graphiti_core/search/search_filters.py:55-73

时序过滤逻辑：AND / 或嵌套

所有四个时序字段（valid_at、invalid_at、created_at、expired_at）都接受 list[list[DateFilter]]。这种嵌套编码了一种析取范式（DNF）：

• 内部列表（list[DateFilter]）——条件通过 AND 组合。
• 外部列表（list[list[DateFilter]]）——分组通过 OR 组合。

图：时序过滤器字段的 DNF 结构

来自集成测试 tests/test_graphiti_int.py:63-70 的示例生成的 Cypher 子句等效于： (e.created_at IS NULL) OR (e.created_at < $created_at_0) OR (e.created_at IS NOT NULL)

要表达一个范围（例如，在两个时间戳之间创建的边），请将两个 DateFilter 实例放在同一个内部列表中：

created_at=[
    [
        DateFilter(date=start_dt, comparison_operator=ComparisonOperator.greater_than_equal),
        DateFilter(date=end_dt,   comparison_operator=ComparisonOperator.less_than),
    ]
]
# 生成：(e.created_at >= $created_at_0 AND e.created_at < $created_at_1)

来源：graphiti_core/search/search_filters.py:149-262，tests/test_graphiti_int.py:63-70

查询构造函数

图：过滤器构造和搜索管线

来源：graphiti_core/search/search_filters.py:86-262，graphiti_core/search/search_utils.py:185-294，graphiti_core/search/search_utils.py:440-573

node_search_filter_query_constructor

graphiti_core/search/search_filters.py:86-104

处理 node_labels。返回一个 (list[str], dict) 元组，包含 Cypher WHERE 片段及其绑定的参数。它对标签执行深度防御验证 graphiti_core/search/search_filters.py:95。

edge_search_filter_query_constructor

graphiti_core/search/search_filters.py:120-262

处理所有适用于边的过滤器字段。处理顺序：

1. edge_types → e.name in $edge_types
2. edge_uuids → e.uuid in $edge_uuids
3. node_labels → n:<labels> AND m:<labels>（两个端点）
4. valid_at、invalid_at、created_at、expired_at → DNF 子句。

所有生成的子句被收集到一个列表中，并在 search_utils.py 的调用点通过 AND 连接。

date_filter_query_constructor

graphiti_core/search/search_filters.py:107-117

构建一个单一的日期比较字符串，例如 (e.valid_at < $valid_at_0)。对于 is_null 和 is_not_null 运算符，不会发出参数占位符。

SearchFilters 如何与搜索集成

图：SearchFilters 从 Graphiti.search_ 到数据库的流程

来源：graphiti_core/search/search.py:98-108，graphiti_core/search/search_utils.py:185-294，graphiti_core/search/search_utils.py:440-573

提供商特定行为

cypher_to_opensearch_operator 函数 graphiti_core/search/search_filters.py:76-83 将 ComparisonOperator 值映射到 OpenSearch 范围查询关键字（gt、lt、gte、lte），用于 Neptune 后端。对于 Neo4j，_build_neo4j_fulltext_query graphiti_core/driver/neo4j/operations/search_ops.py:54-73 处理全文查询的构建，包括分组 ID 过滤。

实际示例

仅过滤活动（未过期）的边

from graphiti_core.search.search_filters import SearchFilters, DateFilter, ComparisonOperator

search_filter = SearchFilters(
    expired_at=[
        [DateFilter(date=None, comparison_operator=ComparisonOperator.is_null)]
    ]
)

过滤当前有效的边（时间点查询）

from graphiti_core.utils.datetime_utils import utc_now
now = utc_now()

search_filter = SearchFilters(
    valid_at=[
        [DateFilter(date=now, comparison_operator=ComparisonOperator.less_than_equal)]
    ],
    invalid_at=[
        [DateFilter(date=None, comparison_operator=ComparisonOperator.is_null)],
        [DateFilter(date=now, comparison_operator=ComparisonOperator.greater_than)]
    ]
)

将 SearchFilters 传递给搜索

SearchFilters 直接传递给 Graphiti.search_()：

results = await graphiti.search_(
    query='Who is Tania',
    search_filter=search_filter,
)

search_filter 参数通过 search.py 中的 search() 函数 graphiti_core/search/search.py:103 向下传递，并通过 edge_search_filter_query_constructor graphiti_core/search/search_filters.py:120-123 等工具用于构建最终的数据库查询。

来源：tests/test_graphiti_int.py:63-75，graphiti_core/search/search.py:98-108，graphiti_core/search/search_utils.py:185-195