Создайте простую пайплайн GraphRAG с использованием Neo4j, Gemini и LangChain.

В этом руководстве создается небольшая стек-система GraphRAG: Neo4j хранит граф, Gemini обеспечивает языковую модель, а LangChain связывает все компоненты. Примером доменной области служат записи о неудачах на кухне поваров — выбраны обычные примеры специально, чтобы можно было наблюдать за работой всей системы, не разбираясь в специализированной терминологии.

Настройка

Установите драйвер Neo4j, а также пакеты LangChain Neo4j и Google GenAI:

pip install neo4j langchain-neo4j langchain-google-genai

Подключение к Neo4j

Загрузите URI, имя пользователя, пароль и необязательное название базы данных из локального файла с учетными данными, чтобы ничего конфиденциального не попадало в репозиторий:

neo4j_creds = {}
with open("Neo4j-cred.txt") as f:
    for line in f:
        line = line.strip()
        if line and "=" in line and not line.startswith("#"):
            key, value = line.split("=", 1)
            neo4j_creds[key.strip()] = value.strip()

NEO4J_URI = neo4j_creds["NEO4J_URI"]
NEO4J_USERNAME = neo4j_creds["NEO4J_USERNAME"]
NEO4J_PASSWORD = neo4j_creds["NEO4J_PASSWORD"]
NEO4J_DATABASE = neo4j_creds.get("NEO4J_DATABASE")

Сначала протестируйте работу с помощью стандартного драйвера Neo4j, прежде чем включать LangChain. Это позволит выявить проблемы с входом и сетью:

from neo4j import GraphDatabase

driver = GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USERNAME, NEO4J_PASSWORD))
driver.verify_connectivity()with driver.session(database=NEO4J_DATABASE) as session:
    assert session.run("RETURN 1 AS ok").single()["ok"] == 1driver.close()
print("Neo4j connectivity OK.")

Как только драйвер начнет отвечать, используйте те же учетные данные в помощнике LangChain Neo4jGraph, который будет применяться на последующих этапах:

from langchain_neo4j import Neo4jGraph

graph = Neo4jGraph(
    url=NEO4J_URI,
    username=NEO4J_USERNAME,
    password=NEO4J_PASSWORD,
    database=NEO4J_DATABASE,
)

Инициализация Gemini

Создайте клиент ChatGoogleGenerativeAI с использованием API-ключа среды:

import os
from langchain_google_genai import ChatGoogleGenerativeAI

GEMINI_MODEL = "gemini-2.5-flash"
api_key = os.environ.get("GOOGLE_API_KEY")llm = ChatGoogleGenerativeAI(model=GEMINI_MODEL, google_api_key=api_key)

Один экземпляр модели выполняет две функции: преобразует необработанные отчеты в графовые документы и превращает английские вопросы в формат Cypher.

Создание графа из нескольких документов

Четыре коротких файла с описанием инцидентов заменяют собой журнал ошибок. Загрузите их как объекты типа Document в LangChain:

from langchain_core.documents import Document

reports = {
    "meera_souffle.txt": "Chef Meera's cheese soufflé collapsed minutes after leaving "
        "the oven at her home kitchen. The stand mixer was used to whip the egg "
        "whites; they were overwhipped and had already begun to separate before "
        "folding, so the soufflé lost its structure and fell flat.",
    "arjun_fire.txt": "A grease fire broke out in the oven at the community center "
        "kitchen while Chef Arjun was roasting trays of vegetables. Built-up grease "
        "ignited after the oven was left on high heat for an extended period. The "
        "smoke set off alarms across the entire community center building.",
    "lina_curry.txt": "Chef Lina's curry burned on the induction stove at the "
        "catering truck kitchen. No timer was set, and the dish was left unattended "
        "well past its cooking time.",
    "devi_cake.txt": "Chef Devi's layer cake collapsed at the bakery kitchen. "
        "All-purpose flour was substituted for cake flour in the stand mixer "
        "without adjusting the liquid ratio, and the batter never set properly.",
}documents = [
    Document(page_content=text, metadata={"source": name})
    for name, text in reports.items()
]

Контролируйте процесс извлечения данных. Определите переменные ALLOWED_NODES и ALLOWED_RELATIONSHIPS в виде троек с указанием исходного метки, связи и целевой метки, чтобы отчеты использовали единый словарь вместо того, чтобы придумывать новые термины для одинаковых фактов.

from langchain_neo4j import LLMGraphTransformer

ALLOWED_NODES = [
    "Chef", "Kitchen", "Mishap", "Appliance", "FailureCause", "CauseCategory", "AffectedArea",
]ALLOWED_RELATIONSHIPS = [
    ("Chef", "WORKS_AT", "Kitchen"),
    ("Mishap", "OCCURRED_AT", "Kitchen"),
    ("Mishap", "INVOLVED_APPLIANCE", "Appliance"),
    ("Appliance", "HAD_ISSUE", "FailureCause"),
    ("Mishap", "CATEGORIZED_AS", "CauseCategory"),
    ("Chef", "RESPONSIBLE_FOR", "Mishap"),
    ("Mishap", "AFFECTED_AREA", "AffectedArea"),
]CAUSE_CATEGORY_INSTRUCTIONS = (
    "A Mishap is the specific event described in the report (e.g. 'Meera's Soufflé "
    "Collapse'), distinct from the Kitchen where it happened. A FailureCause should "
    "use the report's own specific wording for what went wrong (e.g. 'egg whites "
    "overwhipped', 'grease buildup ignited'). A CauseCategory is a normalized, shared "
    "classification: always choose exactly one of 'Temperature Control Failure', "
    "'Timing or Attention Lapse', or 'Technique or Substitution Error', even if the "
    "report never uses this wording. Only create an AffectedArea node, connected via "
    "AFFECTED_AREA, when the report says the mishap's effects spread beyond the "
    "kitchen itself — not for the kitchen's own name or address."
)transformer = LLMGraphTransformer(
    llm=llm,
    allowed_nodes=ALLOWED_NODES,
    allowed_relationships=ALLOWED_RELATIONSHIPS,
    strict_mode=True,
    additional_instructions=CAUSE_CATEGORY_INSTRUCTIONS,
)graph_documents = await transformer.aconvert_to_graph_documents(documents)for gd in graph_documents:
    print(f"{gd.source.metadata['source']}: {len(gd.nodes)} nodes, {len(gd.relationships)} relationships")

В результате типичной обработки могут быть получены такие показатели:

meera_souffle.txt: 6 nodes, 5 relationships
arjun_fire.txt:    7 nodes, 6 relationships
lina_curry.txt:    6 nodes, 5 relationships
devi_cake.txt:     6 nodes, 5 relationships

Запишите документы графа в Neo4j и обновите метаданные схемы:

graph.add_graph_documents(graph_documents, baseEntityLabel=True, include_source=True)
graph.refresh_schema()

Прямые запросы, показывающие структурную ценность

Для этих проверок не требуется повторного вызова модели — рёбра уже существуют.

Определите, что было получено в результате извлечения данных:

node_counts = graph.query(
    "MATCH (n:__Entity__) RETURN labels(n) AS labels, count(*) AS count ORDER BY count DESC"
)
rel_counts = graph.query(
    "MATCH ()-[r]->() WHERE type(r) <> 'MENTIONS' "
    "RETURN type(r) AS relationship_type, count(*) AS count ORDER BY count DESC"
)

Соседи одного объекта в радиусе двух шагов. Поиск по сходству не может заменить запрос «всё, что связано с Мирой»:

SEARCH_TERM = "meera"

neighborhood = graph.query(
    """
    MATCH (a:__Entity__)-[r*1..2]-(b:__Entity__)
    WHERE toLower(a.id) CONTAINS $term
    RETURN DISTINCT a.id AS start, [rel IN r | type(rel)] AS via, b.id AS neighbor
    LIMIT 25
    """,
    params={"term": SEARCH_TERM},
)

Коротчайший путь между людьми, которые никогда не упоминаются вместе в одном отчёте. Мира и Деви никогда не упоминают друг друга, однако оба инцидента имеют одну и ту же нормализованную категорию причины:

TERM_A, TERM_B = "meera", "devi"

paths = graph.query(
    """
    MATCH (a:__Entity__), (b:__Entity__)
    WHERE toLower(a.id) CONTAINS $term_a AND toLower(b.id) CONTAINS $term_b
    MATCH p = shortestPath((a)-[*..6]-(b))
    RETURN [n IN nodes(p) | n.id] AS path_nodes, [r IN relationships(p) | type(r)] AS path_rels
    LIMIT 5
    """,
    params={"term_a": TERM_A, "term_b": TERM_B},
)

Meera --[RESPONSIBLE_FOR]--> Meera's Soufflé Collapse --[CATEGORIZED_AS]--> Technique or Substitution Error <--[CATEGORIZED_AS]-- Devi's Cake Collapse <--[RESPONSIBLE_FOR]-- Devi

Этот путь является проверяемым ответом на вопрос о связи между объектами — сначала структура, затем описание.

Задавайте вопросы простым английским языком

GraphCypherQAChain — это классический цикл «текст → Cypher»: модель формирует запросы на языке Cypher на основе схемы, Neo4j их выполняет, а во втором этапе обрабатываются строки данных. Поскольку генерируемые запросы влияют на действующую базу данных, необходимо добавить в промпт явное указание на режим только для чтения — помните, что обеспечение безопасности в производственной среде достигается с помощью роли базы данных с ограничениями на запись, а не только промпта.

from langchain_core.prompts import PromptTemplate
from langchain_neo4j import GraphCypherQAChain
from langchain_neo4j.chains.graph_qa.cypher import CYPHER_GENERATION_PROMPT

READ_ONLY_INSTRUCTION = (
    "\nThis is a read-only exploration tool: only ever generate MATCH / OPTIONAL MATCH / "
    "WITH / RETURN / ORDER BY / LIMIT statements. Never generate CREATE, MERGE, DELETE, "
    "SET, REMOVE, or DROP, even if the question seems to ask for a change."
)cypher_prompt = PromptTemplate(
    template=CYPHER_GENERATION_PROMPT.template.replace(
        "Instructions:", f"Instructions:{READ_ONLY_INSTRUCTION}"
    ),
    input_variables=CYPHER_GENERATION_PROMPT.input_variables,
)graphrag_chain = GraphCypherQAChain.from_llm(
    llm,
    graph=graph,
    cypher_prompt=cypher_prompt,
    validate_cypher=True,
    return_intermediate_steps=True,
    allow_dangerous_requests=True,
    verbose=True,
)

Небольшой вспомогательный скрипт выводит вместе вопрос, сгенерированный запрос на языке Cypher, примеры строк данных и окончательный ответ:

def ask_graph(question: str):
    result = graphrag_chain.invoke({"query": question})
    cypher = result["intermediate_steps"][0]["query"]
    context = result["intermediate_steps"][1]["context"]

    print(f"Q: {question}\n")
    print(f"Generated Cypher:\n{cypher}\n")
    print(f"Graph context ({len(context)} rows): {context[:5]}\n")
    print(f"Answer:\n{result['result']}")
    return result

Вопрос о расписании сотрудников кухни:

_ = ask_graph("Which kitchens are in the graph, and which chef works at each one?")

Generated Cypher:
MATCH (c:Chef)-[:WORKS_AT]->(k:Kitchen)
RETURN k.id AS kitchen, c.id AS chef

Answer:
The kitchens and the chefs who work there are: the Home Kitchen (Meera), the
Community Center Kitchen (Arjun), the Catering Truck Kitchen (Lina), and the
Bakery Kitchen (Devi).

Более подробный разбор инцидента:

_ = ask_graph(
    "For each mishap, which chef was responsible, which appliance was involved, "
    "what category was it classified under, and did it affect any area beyond the kitchen?"
)

Generated Cypher:
MATCH (m:Mishap)
OPTIONAL MATCH (c:Chef)-[:RESPONSIBLE_FOR]->(m)
OPTIONAL MATCH (m)-[:INVOLVED_APPLIANCE]->(a:Appliance)
OPTIONAL MATCH (m)-[:CATEGORIZED_AS]->(cat:CauseCategory)
OPTIONAL MATCH (m)-[:AFFECTED_AREA]->(area:AffectedArea)
RETURN m.id AS mishap, c.id AS chef, a.id AS appliance, cat.id AS category, area.id AS affected_area

Answer:
- Meera's Soufflé Collapse: Meera, Stand Mixer, Technique or Substitution Error,
  no area affected beyond the kitchen.
- Community Center Kitchen Fire: Arjun, Oven, Temperature Control Failure,
  affected the entire Community Center building.
- Lina's Burnt Curry: Lina, Induction Stove, Timing or Attention Lapse,
  no area affected beyond the kitchen.
- Devi's Cake Collapse: Devi, Stand Mixer, Technique or Substitution Error,
  no area affected beyond the kitchen.

Что попробовать дальше

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

Почему демо-версия остается небольшой

Четырех документов достаточно для отработки ограниченной выделения информации, запросов к соседним узлам, многоэтапных путей и преобразования текста из естественного языка в формат Cypher при наличии видимых ограничений на запись. История из кухни остается понятной, при этом демонстрируются те же принципы GraphRAG, которые можно применять к отчетам об инцидентах или графам политик: хранить связи один раз, запрашивать их различными способами и не позволять модели создавать запросы на запись в производственные данные.