适用于 Neo4j 的零配置过滤向量搜索 MCP 服务器

想象一个充满新闻文章的知识图谱，以及一个需要处理如下请求的智能体：展示路透社上周关于可再生能源的正面报道。这个句子中包含了三重含义：其一是对“可再生能源”进行语义搜索；其二是对来源（路透社）、情感倾向（正面）和日期（上周）进行元数据过滤。

普通的向量搜索可以处理语义部分，但如果没有过滤机制，它就会毫无差别地返回2019年由各种博客发布的负面文章，而大型语言模型在读完所有文章之前根本无法判断这些文章是否无关。这就是元数据过滤向量搜索存在的问题，Neo4j早已提供了两种解决方法。

1. 先用Cypher进行预过滤以缩小候选集，然后再对剩余内容进行向量搜索。这样得到的结果更为准确，但当候选集仍然很大时，搜索速度会变慢。

首先对整个索引进行向量搜索，然后再进行后过滤。这样速度很快，但随着过滤条件越来越严格，召回率可能会下降，因为相关项可能根本无法进入最初的 top-k 结果中。

这两种方法都有各自的优点，同时也都有改进空间。

Neo4j 2026.02 的变化

Neo4j 2026.02（参见发布说明）将带过滤条件的向量搜索作为 Cypher 25 的功能全面推出。现在可以在查询时在向量索引内部应用谓词，因此 HNSW 算法在排序前就能跳过不匹配的向量。这样延迟保持较低，召回率保持较高，且无需进行过度查询。此外还新增了一个原生的 SEARCH 子句，取代了旧版的 db.index.vector.queryNodes() 函数，使向量搜索成为查询语言中的核心功能。

这就引出了一个问题：MCP服务器能否自行发现向量索引，查看其元数据属性，并在无需手动配置的情况下生成过滤后的搜索工具？

零配置的通用向量搜索服务器

凡是开发过用于向量搜索的MCP工具的人都知道常规流程：编写Cypher查询语句、定义参数、设置嵌入模型、记录过滤条件。对于某些知名的索引来说，这样的方法已经足够。但若能有一个可直接用于任何Neo4j数据库的服务器，只需指向相应实例即可自动完成其余工作，那就更加理想了。

这正是该服务器的功能。它在启动时会连接到Neo4j，找出所有的向量索引，从每个索引中抽取一个节点以检测属性类型，然后为每个索引创建一个完全类型化的MCP工具，其中包含过滤参数以及大语言模型能够理解的文档字符串。

有一样东西无法自动发现：嵌入模型。Neo4j目前尚未在索引元数据中存储该模型，而且无论如何都需要API密钥才能调用该模型。因此，只需提供嵌入服务提供商及其凭证作为配置即可。

步骤1：查找向量索引

首先，服务器会向Neo4j请求所有的向量索引：

records, _, _ = driver.execute_query(
    """
    SHOW VECTOR INDEXES
    """,
    database_=NEO4J_DATABASE,
)

结果会显示每个索引的名称、其所覆盖的节点标签以及被索引的属性。如果不存在向量索引，服务器会因无内容可提供而报错退出。

步骤2：样本类型与示例值

对于每个索引，服务器需要知道哪个属性存储了嵌入向量，以及其他属性的类型是什么。published_date存储的是普通日期还是完整的时间戳？source是文本吗？情感得分是十进制数吗？专门的_sample_property辅助函数通过检查每个属性对应的单个节点来给出答案：

def _sample_property(driver, label: str, prop: str) -> tuple[str, Any]:
    """Return (type, sample_value) for a single property."""
    cypher = (
        "MATCH (n:$($label)) WHERE n[$prop] IS NOT NULL "
        "WITH n LIMIT 1 RETURN n[$prop] AS val"
    )
    result, _, _ = driver.execute_query(
        cypher, label=label, prop=prop, database_=NEO4J_DATABASE
    )
    if not result:
        return "string", None
    val = result[0]["val"]
    return _detect_type(val), val

它返回检测到的类型以及实际的样本值。主函数会将这些内容分别放入两个字典中，一个用于存储类型，另一个用于存储示例：

def _sample_property_types(
    driver, label: str, properties: list[str]
) -> tuple[str | None, dict[str, str], dict[str, Any]]:
    embedding_prop: str | None = None
    meta_types: dict[str, str] = {}
    meta_examples: dict[str, Any] = {}

    for p in properties:
        t, sample = _sample_property(driver, label, p)
        if t == "vector":
            embedding_prop = p
        else:
            meta_types[p] = t
            meta_examples[p] = sample

    return embedding_prop, meta_types, meta_examples

_detect_type函数会按顺序检查列表（向量）、布尔值、整数、浮点数、datetime类型和date类型，若均不匹配则默认使用字符串类型。datetime与date被视为不同的数据类型。包含时间组件的属性会被标记为datetime类型，此时应使用Neo4j的datetime()转换函数而非date()。浮点数列表即为嵌入向量，其余所有属性则成为可过滤的元数据字段。以新闻示例而言，最终结果如下：

embedding -> vector
published_date -> date (or datetime, if timestamps are present)
source -> string
sentiment -> float

meta_examples 字典保存了真实的样本值，例如来源为“Reuters”，情感值为0.72，这些值会用于丰富下一步生成的文档字符串。

第3步：生成搜索工具

现在所有部分都已整合到位。对于每个索引，一个闭包会捕获该索引的元数据并生成相应的搜索函数：

def _make_search_fn(idx: dict[str, Any]):
    """Factory that closes over a single index's metadata."""

    index_name = idx["name"]
    label = idx["label"]
    embed_prop = idx["embed_prop"]
    prop_types = idx["prop_types"]
    prop_examples = idx["prop_examples"]

    # Build the filter-parameter docstring dynamically
    filter_lines = []
    for prop, t in prop_types.items():
        safe = prop.replace(" ", "_").replace("-", "_")
        ex = _format_example(prop_examples.get(prop), t)
        if t in ("float", "int"):
            filter_lines.append(
                f"  {safe}: Optional dict with 'min' and/or 'max' (numeric range). Example value: {ex}"
            )
        elif t == "datetime":
            filter_lines.append(
                f"  {safe}: Optional dict with 'min' and/or 'max' (ISO datetime strings, uses datetime()). Example value: {ex}"
            )
        elif t == "date":
            filter_lines.append(
                f"  {safe}: Optional dict with 'min' and/or 'max' (ISO date strings, uses date()). Example value: {ex}"
            )
        elif t == "bool":
            filter_lines.append(
                f"  {safe}: Optional bool. Example value: {ex}"
            )
        else:
            filter_lines.append(
                f"  {safe}: Optional str (exact match). Example value: {ex}"
            )

    filters_doc = "\n".join(filter_lines) if filter_lines else "  (none)"

    async def search(
        query: str,
        top_k: int = 10,
        filters: Optional[dict[str, Any]] = None,
    ) -> str:
        query_vec = embedder.embed_query(query)
        where_clause, cypher_params = _build_where(prop_types, filters or {})
        where_part = f"\n    WHERE {where_clause}" if where_clause else ""

        cypher = (
            f"CYPHER 25\n"
            f"MATCH (n:$($label))\n"
            f"  SEARCH n IN (\n"
            f"    VECTOR INDEX {index_name}\n"
            f"    FOR $query_vec{where_part}\n"
            f"    LIMIT $top_k\n"
            f"  ) SCORE AS score\n"
            f"RETURN n {{ .*, `{embed_prop}`: null }} AS doc, score\n"
            f"ORDER BY score DESC"
        )

        cypher_params["query_vec"] = query_vec
        cypher_params["top_k"] = top_k

        rows, _, _ = driver.execute_query(
            cypher, cypher_params, database_=NEO4J_DATABASE
        )
        results = [{"doc": dict(r["doc"]), "score": r["score"]} for r in rows]
        return json.dumps(results, indent=2, default=str)

    search.__name__ = f"search_{index_name}"
    search.__qualname__ = f"search_{index_name}"
    search.__doc__ = (
        f"Semantic vector search over :{label} nodes using the "
        f"'{index_name}' index.\n\n"
        f"Args:\n"
        f"  query: Natural-language search text (will be embedded).\n"
        f"  top_k: Number of results to return (default 10).\n"
        f"  filters: Optional dict of metadata filters:\n"
        f"{filters_doc}\n"
    )
    return search

有一些细节值得注意：

• CYPHER 25前缀启用了SEARCH子句及索引内过滤所需的新型语法。
• WHERE子句位于SEARCH块内部，这就是2026.02版本中的索引内过滤功能。HNSW本身会处理这些谓词，因此近似最近邻搜索结果在返回时已经过过滤，无需后续过滤即可保持较高的召回率。
• 返回的行中嵌入属性被设置为null（{embed_prop}: null），这样大型向量就不会被传回大语言模型，避免浪费计算资源。
• 文档字符串会根据抽样类型动态生成，并为每种过滤条件提供具体的示例值。这样代理就能知道存在哪些过滤条件、期望的类型以及合理的数值范围。

构建WHERE子句

_build_where函数可将Python过滤字典转换为Cypher谓词，具体形式根据每个属性声明的类型来确定：

• 浮点数和整数通过{"min": ..., "max": ...}生成范围过滤，从而形成>=和<=比较。
• 日期时间也使用相同的范围逻辑，但需通过datetime()函数进行转换。
• 日期同样适用该逻辑，但需通过date()函数进行转换。
• 布尔值则使用完全相等比较。
• 默认类型的字符串也使用完全相等比较。

所有谓词均通过AND连接，这与Neo4j在向量索引过滤中支持的简单属性比较方式一致。

def _build_where(
    prop_types: dict[str, str], filters: dict[str, Any]
) -> tuple[str, dict[str, Any]]:
    clauses: list[str] = []
    params: dict[str, Any] = {}

    for prop, t in prop_types.items():
        safe = prop.replace(" ", "_").replace("-", "_")
        val = filters.get(safe)
        if val is None:
            continue

        if t in ("float", "int"):
            if isinstance(val, dict):
                if val.get("min") is not None:
                    k = f"filt_{safe}_min"
                    clauses.append(f"n.`{prop}` >= ${k}")
                    params[k] = val["min"]
                if val.get("max") is not None:
                    k = f"filt_{safe}_max"
                    clauses.append(f"n.`{prop}` <= ${k}")
                    params[k] = val["max"]
        elif t == "datetime":
            if isinstance(val, dict):
                if val.get("min") is not None:
                    k = f"filt_{safe}_min"
                    clauses.append(f"n.`{prop}` >= datetime(${k})")
                    params[k] = val["min"]
                if val.get("max") is not None:
                    k = f"filt_{safe}_max"
                    clauses.append(f"n.`{prop}` <= datetime(${k})")
                    params[k] = val["max"]
        elif t == "date":
            if isinstance(val, dict):
                if val.get("min") is not None:
                    k = f"filt_{safe}_min"
                    clauses.append(f"n.`{prop}` >= date(${k})")
                    params[k] = val["min"]
                if val.get("max") is not None:
                    k = f"filt_{safe}_max"
                    clauses.append(f"n.`{prop}` <= date(${k})")
                    params[k] = val["max"]
        elif t == "bool":
            k = f"filt_{safe}"
            clauses.append(f"n.`{prop}` = ${k}")
            params[k] = val
        else:  # string exact match
            k = f"filt_{safe}"
            clauses.append(f"n.`{prop}` = ${k}")
            params[k] = val

    where = " AND ".join(clauses)
    return where, params

为何不让大语言模型来编写Cypher语句？

Neo4j MCP的Cypher服务器可以通过文本转Cypher的功能生成查询语句，但向量搜索需要嵌入向量，而目前在不同工具之间传输嵌入向量的唯一方式就是通过大语言模型的上下文。将1536维的向量传递给模型会浪费计算资源与时间。

将整个工作流程——包括嵌入向量的生成、Cypher语句的构建与执行——整合在同一个工具中，就能避免嵌入向量进入模型的上下文。大语言模型只需输入自然语言查询及可选的过滤条件，即可获得按得分排序的结果。

如何运行

可以使用uvx直接从GitHub上运行该服务器。对于Claude Desktop、VS Code、Cursor等工具，可在其MCP客户端配置中添加如下内容：

{
  "mcpServers": {
    "neo4j-vector": {
      "command": "uvx",
      "args": [
        "--from",
        "git+https://github.com/tomasonjo/neo4j-prefiltering-mcp",
        "--with",
        "langchain-openai",
        "neo4j-prefiltering-mcp"
      ],
      "env": {
        "NEO4J_URI": "neo4j+s://demo.neo4jlabs.com:7687",
        "NEO4J_USER": "companies2",
        "NEO4J_PASSWORD": "companies2",
        "NEO4J_DATABASE": "companies2",
        "EMBEDDING_MODEL": "openai:text-embedding-3-small",
        "OPENAI_API_KEY": "<your-key>"
      }
    }
  }
}

--with langchain-openai 参数用于安装 OpenAI 嵌入提供器。若要使用其他提供器，如 Cohere 或本地模型，则需替换该依赖项并更新 EMBEDDING_MODEL。嵌入查找是通过 LangChain 的 init_embeddings() 函数实现的，因此任何 LangChain 支持的提供器均可使用。

在 Claude Desktop 中使用示例问题进行测试并扩展工具调用后，会显示如下参数：

{
  "query": "positive developments in artificial intelligence AI breakthroughs progress",
  "top_k": 10,
  "filters": {
    "language": "en",
    "date": {
      "min": "2026-01-01",
      "max": "2026-12-31"
    },
    "sentiment": {
      "min": 0.5
    }
  }
}

无需手动编写任何内容：没有工具定义、没有查询语句、也没有过滤模式。只需创建另一个向量索引，重启服务器，新的工具就会出现。

核心要点

Neo4j 2026.02版本中的带过滤功能的向量搜索，结合Cypher 25的SEARCH语法，为实现带过滤功能的向量搜索工具提供了简洁的路径。再结合MCP以及简单的启动时自检机制，就能让服务器自动查找向量索引、识别可过滤的属性，并生成类型化的搜索工具，而无需编写任何特定于工具的代码。只需针对包含向量索引的Neo4j数据库使用该功能，它就会自动构建相应的工具集。过滤功能的文档中包含了完整的语法说明。