10.4 · SearchFilters 与时序 Queries（SearchFilters and Temporal Queries）

SearchFilters 与时序 Queries（中文译文）

原始 DeepWiki 页面：https://deepwiki.com/getzep/graphiti/10.4-searchfilters-and-temporal-queries

翻译时间：2026-05-27T08:44:58.786Z

翻译模型：deepseek-chat

原文字符数：13160

项目：Graphiti (graphiti)

---

搜索过滤器与时序查询

相关源文件

以下文件被用作生成此维基页面的上下文：

• 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 及其组件

classDiagram
    class "SearchFilters" {
        +node_labels: list[str] | None
        +edge_types: list[str] | None
        +edge_uuids: list[str] | None
        +valid_at: list[list[DateFilter]] | None
        +invalid_at: list[list[DateFilter]] | None
        +created_at: list[list[DateFilter]] | None
        +expired_at: list[list[DateFilter]] | None
        +property_filters: list[PropertyFilter] | None
    }

    class "DateFilter" {
        +date: datetime | None
        +comparison_operator: ComparisonOperator
    }

    class "PropertyFilter" {
        +property_name: str
        +property_value: str | int | float | None
        +comparison_operator: ComparisonOperator
    }

    class "ComparisonOperator" {
        <<枚举>>
        equals
        not_equals
        greater_than
        less_than
        greater_than_equal
        less_than_equal
        is_null
        is_not_null
    }

    "SearchFilters" "1" --> "0..*" "DateFilter" : "valid_at / invalid_at / created_at / expired_at"
    "SearchFilters" "1" --> "0..*" "PropertyFilter" : "property_filters"
    "DateFilter" --> "ComparisonOperator"
    "PropertyFilter" --> "ComparisonOperator"

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

---

ComparisonOperator

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

成员	Cypher 值	说明
equals	=	精确匹配
not_equals	<>	不相等
greater_than	>	严格晚于/大于
less_than	<	严格早于/小于
greater_than_equal	>=	在或晚于/大于等于
less_than_equal	<=	在或早于/小于等于
is_null	IS NULL	属性必须不存在/为空
is_not_null	IS NOT NULL	属性必须存在且非空

当使用 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 注入。

字段	类型	适用范围	描述
node_labels	`list[str] \	None`	节点和边	限制为带有这些标签的节点。对于边搜索，源节点（n）和目标节点（m）都必须具有该标签。
edge_types	`list[str] \	None`	仅边	匹配 e.name——关系名称/类型。
edge_uuids	`list[str] \	None`	仅边	限制为特定的边 UUID。
valid_at	`list[list[DateFilter]] \	None`	仅边	过滤 e.valid_at——事实在现实世界中为真的时间。
invalid_at	`list[list[DateFilter]] \	None`	仅边	过滤 e.invalid_at——事实变为假的时间。
created_at	`list[list[DateFilter]] \	None`	仅边	过滤 e.created_at——边被入库的时间。
expired_at	`list[list[DateFilter]] \	None`	仅边	过滤 e.expired_at——边被取代的时间。
property_filters	`list[PropertyFilter] \	None`	节点和边	自定义属性过滤器。

来源：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 结构

flowchart TD
    F["SearchFilters.created_at\n(list[list[DateFilter]])"]
    OR1["OR 分组 0\n(内部列表 0)"]
    OR2["OR 分组 1\n(内部列表 1)"]
    OR3["OR 分组 2\n(内部列表 2)"]
    A1["DateFilter\nis_null"]
    A2["DateFilter\nless_than, utc_now"]
    A3["DateFilter\nis_not_null"]
    F -->|"OR"| OR1
    F -->|"OR"| OR2
    F -->|"OR"| OR3
    OR1 -->|"AND"| A1
    OR2 -->|"AND"| A2
    OR3 -->|"AND"| A3

来自集成测试 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

---

查询构造函数

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

flowchart TD
    SF["SearchFilters"]

    subgraph "search_filters.py"
        NSFC["node_search_filter_query_constructor()"]
        ESFC["edge_search_filter_query_constructor()"]
        DFC["date_filter_query_constructor()"]
    end

    subgraph "search_utils.py"
        EFS["edge_fulltext_search()"]
        ESS["edge_similarity_search()"]
        EBS["edge_bfs_search()"]
        NFS["node_fulltext_search()"]
        NSS["node_similarity_search()"]
        NBS["node_bfs_search()"]
    end

    SF --> NSFC
    SF --> ESFC
    ESFC --> DFC

    NSFC --> NFS
    NSFC --> NSS
    NSFC --> NBS
    ESFC --> EFS
    ESFC --> ESS
    ESFC --> EBS

来源：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。

提供商	生成的子句
Neo4j / FalkorDB / Neptune	`n:Label1\	Label2`
Kuzu	list_has_all(n.labels, $labels)

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_ 到数据库的流程

flowchart TD
    GS["Graphiti.search_()"]
    S["search()\n(search.py)"]
    ES["edge_search()"]
    NS["node_search()"]
    EPS["episode_search()"]
    EFS2["edge_fulltext_search()"]
    ESS2["edge_similarity_search()"]
    EBS2["edge_bfs_search()"]
    NFS2["node_fulltext_search()"]
    NSS2["node_similarity_search()"]
    ESFC2["edge_search_filter_query_constructor()"]
    NSFC2["node_search_filter_query_constructor()"]
    DB["GraphDriver.execute_query()"]

    GS -->|"SearchFilters"| S
    S --> ES
    S --> NS
    S --> EPS
    ES --> EFS2
    ES --> ESS2
    ES --> EBS2
    NS --> NFS2
    NS --> NSS2
    EFS2 --> ESFC2
    ESS2 --> ESFC2
    EBS2 --> ESFC2
    NFS2 --> NSFC2
    NSS2 --> NSFC2
    ESFC2 -->|"WHERE 子句 + 参数"| DB
    NSFC2 -->|"WHERE 子句 + 参数"| DB

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

---

提供商特定行为

特性	Neo4j / FalkorDB	Kuzu	Neptune
节点标签过滤	`n:A\	B`	list_has_all(n.labels, $labels)	`n:A\	B`
时序过滤	Cypher WHERE 子句	相同的 Cypher 子句	Cypher WHERE 子句
OpenSearch 运算符映射	不适用	不适用	全文搜索使用 cypher_to_opensearch_operator()

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