Сервер MCP с нулевой настройкой для фильтрованного векторного поиска в Neo4j

Автоматически обнаруживаются векторные индексы Neo4j, приводятся примеры типов свойств, и генерируются инструменты поиска типа MCP, использующие фильтрацию в режиме in-index с Cypher 25, без необходимости вручную настраивать что-либо.

Издатель: REACTAPP.TOPДобавлено на сайт 28 окт. 2026 г.9 мин чтения1974 слов

Источники: Automagic Metadata Filtering MCP in Neo4j

Представьте себе граф знаний, заполненный новостными статьями, и агента, который получил задание: показать положительные материалы о возобновляемой энергетике от Reuters за прошлую неделю. В этом предложении содержатся три составляющих: семантический поиск по запросу «возобновляемая энергетика», фильтрация по метаданным источника (Reuters), тональности (положительная) и даты (прошлая неделя).

Обычный векторный поиск обрабатывает семантическую часть. Однако без фильтрации он с удовольствием вернет негативные статьи 2019 года с различных блогов, и большая языковая модель не сможет понять их бесполезность, пока не прочитает все они. Это и есть проблема векторного поиска с использованием фильтрации метаданных, и Neo4j уже давно предлагает два способа её решения.

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

Сначала выполните векторный поиск по всему индексу, а затем примените дополнительные фильтры. Это быстро, однако точность может снижаться по мере ужесточения фильтров, поскольку релевантные элементы могут так и не попасть в начальный список топ-к.

У обоих подходов были сильные стороны, и у обоих оставался простор для улучшений.

Что изменилось в 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 поочередно проверяет наличие списка (вектора), значения типа bool, int, float, datetime и date; в случае отсутствия подходящего типа используется строка. datetime и date считаются разными типами. Свойство, содержащее временные компоненты, имеет тип datetime и подвергается преобразованию с использованием функции datetime() Neo4j, а не date(). Список значений типа float представляет собой вектор признаков, а все остальные свойства становятся полями метаданных, подлежащими фильтрации. В примере с новостями результат будет выглядеть следующим образом:

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

Создание оператора 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?

Сервер Cypher MCP от Neo4j может генерировать запросы с помощью функции преобразования текста в Cypher, но для векторного поиска требуются эмбеддинги, и в настоящее время единственный способ передачи эмбеддингов между инструментами — это через контекст ЯИИ. Передача 1536-мерных векторов через модель тратит токены и время.

Если объединить весь процесс — генерацию эмбеддингов, создание и выполнение запросов на Cypher — в один инструмент, эмбеддинги останутся вне контекста модели. ЯИИ получает запрос на естественном языке с необязательными фильтрами и возвращает результаты с оценкой.

Запуск

Сервер можно запустить непосредственно с 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 с векторными индексами — сервер сам соберёт необходимый набор инструментов. В документации по фильтрации приведён полный синтаксис.

← Все статьи Прислать бриф