Сервер MCP без налаштувань для фільтрованого пошуку векторів у Neo4j
Автоматично виявляти векторні індекси Neo4j, демонструвати типи властивостей та створювати інструменти пошуку типу MCP, які використовують фільтрацію в момент пошуку за допомогою Cypher 25, без будь-якої ручної налаштування.
Уявіть собі граф знань, наповнений новинними статтями, та агента, який отримує запит: покажіть позитивні матеріали про відновлювану енергетику від 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 на кожен індекс, який містить параметри фільтрації та документацію, яку може прочитати 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 -> 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 є векторні індекси, цей сервер створює власний набір інструментів. У документації щодо фільтрації наведений повний синтаксис.