MCP-сервер без налад для вектарнага пошуку з фільтрамі ў Neo4j

Уявіце сабэру знаёмасцей, заполненую статыямі пра новыні, і агента, які мусiць выконаць такі запіт: паказаць пасяблівыя атрыбуты статэй пра адновальныя крыніцы энергіі з Reuters за празьледзейшыя тыдні. У гэтым выразе супакоўваны трое аспектаў: першы — сэмантычны пошук па ключоўцу «адновальныя крыніцы энергіі», а іншыя — фільтры метаданых па выкладчыку (Reuters), настрою статэй (пасяблівы) і даце (празьледзейшыя тыдні).

Звычны вектарны пошук размовляе з сэмантычнай часткай запіту. Але без фільтраў ён будзе спакойна вяртаць негатыўныя статыі з 2019 года, паблікованыя рандамнымі блогамі, а большая мовная мадэль не можа з’ясавіць, што яны нерэлевантныя, пакуль не прачытае ўсія. Гэта і ёсць проблема вектарнага пошуку з фільтрацыяй метаданых, і Neo4j давно праставляе два спосабы яе рашэння.

1. Спачатку за дапамогою Cypher адбуваецца прадзвінутая фільтрацыя, каб зменшыць колькасць кандыдатаў, пасля чаго адбываецца вектарны пошук сярод тых, што засталіся. Рэзультаты будуць точнымі, але працэс можа стаць медленным, калі колькасць кандыдатаў застаецца вялікай.

Спачатку запускайце пошук па вектарах у всім індэксе, а потым выкарыстоўвайце дадатковыя фільтраўанні. эта швайна, але рэкалі можа знізіцца, калі фільтры стануць стрэлкавымі, таму што рэлевантныя элементы можаць ніколі не патрапіць у першыя k рэзультатаў.

У обох былі рэальныя перавагі, і ў обох застаўся простор для паследніх удосконаленняў.

Якія змены ў Neo4j 2026.02

Neo4j 2026.02 (дывіцеся прыметкі да выпуску) зробіў пошук па вектарах з фільтрамі загаловым як частку Cypher 25. Цяпер прэдыкаты можна застаўляць унутры вектаровага індэкса пад час запіту, таму алгорытм HNSW прахоўвае вектары, якія не падходзяць, пры ранжаванні. Затрымка застаецца нізкай, рэкалі — высокай, і не трэба загружаць занадто многа даных. Таксама з’явілася новая вбудованая клаза SEARCH, якая заменіла старыяй процедуру db.index.vector.queryNodes() і ператворыла пошук па вектарах на абсалютна некальківы элемент запітовага языка.

Гэта падаець пытанне: чы можа сервер MCP самастаяна адкрыць вектарныя індэксы, пераглядаць ўсія ўласнасці ўжо індэксаў і стварыць засобы пошуку з фільтрамі без якой-лебе ручной налады?

Універсальны сервер для вектарнага пошуку без жадной налады

Кожны, хто ствараў інструменты MCP для вектарнага пошуку, знае стандартны алгорытм: пісаецца код на мове Cypher, задаюцься параметры, насталяваецца мадэль укладання дадзеных, опісуюцься фільтры. Для аднаго вядомага індэкса такое падходзіць. Але ўсё прыемней, калі існуе сервер, які можа працаваць з будзь-яй базай дадзеных Neo4j — проста паказваецца на ёё экземпляр, і сервер сам вырашае рэшту.

Самэ гэта і робіць этот сервер. Пад час запуску ён падключаецца да Neo4j, знаходзіць усе вектарныя індэксы, берае прыклады вузлаў з кожнага, ўбачвае тыпы ўласнасцяў і стварае по аднаму індэксу абоўсюды тыпаваны інструмент MCP з параметрамі фільтраў і докстрангам, які можа прачытаць LLM.

Ёсць одна рыч, яку неможліво віявіты автоматычно: сам мадэль імбеддынгу. 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 парадоўжна пераглядае ліст (вектар), тып bool, int, float, datetime і date, а калі нічога з гэтага не знайдзецца, выкарыстоўвае тып string. Datetime і date лічыцца як разныя тыпы. Атрыбут з часовымі складаванням мае тып datetime і аплікуюцца канвертаванне datetime() з Neo4j, а не date(). Ліст з значэннямі float ёсць embedding, а всі іншыя атрыбуты стаюць полям метаданых, якія можна фільтруваць. Для прыкладу з новынамі рэзультат выглядае так:

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 вырахоўвае умовы, таму рэзультаты апрыксіматыўных найбліжэйшых суседаў вяртаюцца вучораненыя, без падтрымкі рэкаліі пасля дадатковай фільтрацыі.
• Атрыбут embedding у адпаведных рэкордах стаў нулявым ({embed_prop}: null), таму вялікія вектары не вяртаюцца да LLM, дзе яны бы толькі спалілі токены.
• Докстрынг ствараецца дынамічна на адной з выбраных тыпаў і містыць конкрэтны прыклад значэння для кожнага фільтра. Такім чынам агент бачыць, якія фільтры існуюць, якія тыпы яны очакваны і як выглядаюць рэалістычныя значэння.

Стварэнне клазу WHERE

Funkція _build_where перэтворае словнік фільтрыв Python у судове тверджэння Cypher, вібіраючы форму згідно з оголошэнім тыпом кожноі властывосты:

• Для чысел і інтэрцевів вікорыстовуецца логіка дыяпазонів праз {"min": ..., "max": ...}, што створуе параўнання тыпу >= і <=.
• Даты таксама вікорыстовуюць цю ж логіку, але замест неі вікорыстовуецца функція 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

Чаму бы не дазволіць LLM пісаць Cypher?

Сервер Neo4j MCP Cypher могаў бы генераваць запіты за дапамогою перакладу тэксту на Cypher, але вектарны пошук выканаўся з адбівамі, і сёння ёдыны спосаб перадаць адбів между інструментамі — праз кантэкст LLM. Перадача 1536-вымірных вектароў праз мадэль марнуе токены і час.

Якшы ўсю рабочую практыку — генераванне адбіваў, стварэнне і выконанне Cypher — аб’еднаваць у адны інструмент, адбівы застануцца пахацелыя ад кантэксту модэлі. LLM задае запіт на натуральнай мове і неабавязковы фільтры, а пасля получае рэзультаты з оценкамі.

Выкананне

Сервер можна запусціць безпосерэдна з GitHub за дапамогою uvx. Да настройкі кліента MCP для Claude Desktop, VS Code, Cursor або падобных дадзіце такы элемент:

{
  "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. Пошук эмбеддзінга адбываецца праз функцыю init_embeddings() з LangChain, таму будзе працаваць любы прадастчык, які падтрымліваецца 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, разам з сінтаксам SEARCH у Cypher 25, дае просты спосаб для інструментаў вектарнага пошуку з фільтрацыяй. У поўнай злучэнні з MCP і невялікай самапрацэйнай дыягностыкай сервер можа знаходзіць вектарныя індэксы, выявляць атрыбуты, якія можна фільтруваць, і ствараць інструменты пошуку з адпаведным тыпам дадзеных, не пішохчы ні адной лініі коду, прызначанага конкрэтна для інструмента. Як толькі ён працуе з базай дадзеных Neo4j, якая мае вектарныя індэксы, ён сам стварае свой набор інструментаў. У даследжэннях па фільтрацыі прадстаўленаў весь сінтаксіс.