1. Neo4j: установка и настройка

Первый шаг — развернуть графовую базу данных Neo4j. У нас есть два основных пути: локальный Docker-контейнер для разработки или облачный сервис Neo4j AuraDB с бесплатным тиром. Docker-подход даёт полный контроль и не требует интернета для работы графа, а AuraDB избавляет от необходимости управлять инфраструктурой.

База данных Neo4j хранит данные в виде графа: узлы (Nodes) представляют сущности (люди, организации, концепции), а связи (Relationships) — отношения между ними. Для запросов используется язык Cypher — декларативный язык, напоминающий SQL, но оптимизированный для графовых паттернов.

# docker-compose.yml — запуск Neo4j с APOC-плагином
version: '3.8'

services:
  neo4j:
    image: neo4j:5.22-community
    container_name: neo4j-graphrag
    ports:
      - "7474:7474"   # HTTP-интерфейс
      - "7687:7687"   # Bolt-протокол
    environment:
      NEO4J_AUTH: neo4j/graphrag2024
      NEO4J_PLUGINS: '["apoc"]'
    volumes:
      - ./neo4j_data:/data
      - ./neo4j_logs:/logs
      - ./neo4j_import:/var/lib/neo4j/import

Запускаем контейнер и проверяем подключение из Python. Плагин APOC (Awesome Procedures on Cypher) критически важен — он расширяет Cypher вспомогательными процедурами, включая векторные операции для GraphRAG.

# Терминал: запуск Neo4j
docker compose up -d
docker logs neo4j-graphrag --tail 20

# Ожидаемый вывод:
2024-07-03 12:00:00.000+0000 INFO  Started.
2024-07-03 12:00:01.000+0000 INFO  Bolt enabled on 0.0.0.0:7687.
Remote interface available at http://localhost:7474/

# Подключение к Neo4j из Python
from neo4j import GraphDatabase

class Neo4jConnection:
    def __init__(self, uri, user, password):
        self.driver = GraphDatabase.driver(uri, auth=(user, password))

    def verify(self):
        self.driver.verify_connectivity()
        print("✅ Neo4j подключена")

    def close(self):
        self.driver.close()

db = Neo4jConnection(
    uri="bolt://localhost:7687",
    user="neo4j",
    password="graphrag2024"
)
db.verify()

2. Извлечение сущностей через LLM

Ключевой этап GraphRAG — извлечение структурированных знаний из неструктурированного текста. Мы используем LLM (GPT-4o или локальную модель) с чётко сформулированным промптом для задачи Named Entity Recognition (NER) и Relationship Extraction (RE). Модель возвращает не просто список сущностей, а структурированный граф: узлы с типами и свойствами, связи с типами и направлениями.

Для надёжного парсинга ответа LLM используем Pydantic-схему — это гарантирует, что ответ модели всегда будет иметь ожидаемую структуру, а ошибки валидации можно обработать программно.

from pydantic import BaseModel, Field
from typing import List, Optional
from openai import OpenAI

# --- Pydantic-схемы ---
class Entity(BaseModel):
    """Узел графа знаний."""
    id: str = Field(..., description="Уникальный идентификатор (snake_case)")
    type: str = Field(..., description="Тип: Person, Organization, Location, Event, Concept, Product")
    name: str = Field(..., description="Человекочитаемое имя")
    properties: dict = Field(default_factory=dict)

class Relationship(BaseModel):
    """Связь между узлами."""
    source_id: str
    target_id: str
    type: str = Field(..., description="Тип связи: WORKS_FOR, LOCATED_IN, CREATED, OWNS и т.д.")
    description: Optional[str] = None

class KnowledgeGraph(BaseModel):
    """Полный граф, извлечённый из текста."""
    entities: List[Entity]
    relationships: List[Relationship]
    summary: str = Field(..., description="Краткое резюме текста (1-2 предложения)")

Критически важен промпт для LLM. Он должен быть конкретным, с примерами и ограничениями. Ниже — промпт, который даёт стабильные результаты на русскоязычных и англоязычных текстах:

EXTRACTION_PROMPT = """
Ты — система извлечения графа знаний. Проанализируй текст и
выдели все значимые сущности и связи между ними.

ПРАВИЛА:
1. Сущность — всё, что может быть узлом графа: человек,
   организация, локация, событие, продукт, концепция.
2. Каждая сущность получает уникальный snake_case id.
3. Связи должны быть значимыми и именованными (глагол или
   отношение, например: WORKS_FOR, LOCATED_IN, FOUNDED).
4. Не выдумывай сущности, которых нет в тексте.
5. Если текст содержит даты — включай их в properties.

Верни JSON строго по схеме KnowledgeGraph.
"""

# Функция извлечения графа через LLM
client = OpenAI(api_key="sk-...")

def extract_graph(text: str) -> KnowledgeGraph:
    """Извлекает граф знаний из текста через GPT-4o."""
    response = client.beta.chat.completions.parse(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": EXTRACTION_PROMPT},
            {"role": "user", "content": f"Текст для анализа:\n\n{text}"}
        ],
        response_format=KnowledgeGraph,
        temperature=0.1  # Низкая температура для стабильности
    )
    return response.parsed

# Пример использования:
text = """Илон Маск основал SpaceX в 2002 году в Хоторне,
Калифорния. Компания разработала ракету Falcon 9 и
корабль Dragon. SpaceX выполняет контракты для NASA."""

kg = extract_graph(text)
print(f"Извлечено {len(kg.entities)} сущностей и {len(kg.relationships)} связей")
for e in kg.entities:
    print(f"  [{e.type}] {e.name}")

3. Построение графа знаний в Neo4j

Получив структурированные данные от LLM, переносим их в Neo4j с помощью Cypher-запросов. Для каждой сущности создаём узел с соответствующим лейблом, а для каждой связи — ребро между узлами. Используем MERGE вместо CREATE — это предотвращает дублирование узлов при повторном запуске пайплайна на обновлённых данных.

Важный архитектурный момент: для каждого узла мы генерируем свойство embedding_text — конкатенацию типа, имени и свойств. Оно понадобится на следующем шаге для построения векторного индекса.

def populate_graph(db: Neo4jConnection, kg: KnowledgeGraph):
    """Заполняет Neo4j извлечённым графом знаний."""
    with db.driver.session() as session:

        # 1. Создаём узлы (MERGE для идемпотентности)
        for entity in kg.entities:
            # Формируем текст для эмбеддинга
            props_str = ", ".join(f"{k}: {v}" for k, v in entity.properties.items())
            embedding_text = f"[{entity.type}] {entity.name}"
            if props_str:
                embedding_text += f" ({props_str})"

            session.run("""
                MERGE (n:`{entity.type}` {id: $id})
                SET n.name = $name,
                    n += $properties,
                    n.embedding_text = $embedding_text,
                    n.source = $source
            """,
                id=entity.id,
                name=entity.name,
                properties=entity.properties,
                embedding_text=embedding_text,
                source="llm_extraction"
            )

        # 2. Создаём связи
        for rel in kg.relationships:
            session.run("""
                MATCH (a {id: $source_id})
                MATCH (b {id: $target_id})
                MERGE (a)-[r:`{rel.type}`]->(b)
                SET r.description = $desc
            """,
                source_id=rel.source_id,
                target_id=rel.target_id,
                desc=rel.description or ""
            )

    print(f"✅ Граф загружен: {len(kg.entities)} узлов")

# Проверочный Cypher-запрос в Neo4j Browser:
# Показывает граф SpaceX с соседями
MATCH (n {name: 'SpaceX'})-[r]-(m)
RETURN n, r, m LIMIT 25

4. Векторный индекс и эмбеддинги узлов

Классический RAG полагается только на векторный поиск по чанкам документов. GraphRAG идёт дальше: мы индексируем не только документы, но и узлы графа. Каждый узел получает embedding-вектор, вычисленный по его текстовому описанию (embedding_text). Это позволяет находить семантически близкие сущности даже если они не связаны напрямую в графе.

Neo4j 5.x поддерживает векторный индекс нативно через синтаксис CREATE VECTOR INDEX. Эмбеддинги генерируем через OpenAI text-embedding-3-small — дёшево и качественно, 1536 измерений.

# Шаг 1: Создание векторного индекса в Neo4j
# Выполнить в Neo4j Browser или через Python:

def create_vector_index(db: Neo4jConnection):
    """Создаёт векторный индекс для embedding-поля узлов."""
    with db.driver.session() as session:
        session.run("""
            CREATE VECTOR INDEX entity_embeddings IF NOT EXISTS
            FOR (n:__ALL__)
            ON (n.embedding)
            OPTIONS {
                indexConfig: {
                    `vector.dimensions`: 1536,
                    `vector.similarity_function`: 'cosine'
                }
            }
        """)
    print("✅ Векторный индекс создан")

# Шаг 2: Генерация эмбеддингов для всех узлов
from openai import OpenAI

client = OpenAI()
EMBEDDING_MODEL = "text-embedding-3-small"

def embed_nodes(db: Neo4jConnection, batch_size=50):
    """Генерирует эмбеддинги для всех узлов без embedding."""
    with db.driver.session() as session:
        # Получаем узлы без эмбеддинга
        result = session.run("""
            MATCH (n)
            WHERE n.embedding_text IS NOT NULL
              AND n.embedding IS NULL
            RETURN n.id AS id, n.embedding_text AS text
        """)

        nodes = [(r["id"], r["text"]) for r in result]

        # Генерируем эмбеддинги батчами
        for i in range(0, len(nodes), batch_size):
            batch = nodes[i:i+batch_size]
            texts = [t for _, t in batch]
            ids = [nid for nid, _ in batch]

            response = client.embeddings.create(
                model=EMBEDDING_MODEL,
                input=texts
            )
            embeddings = [d.embedding for d in response.data]

            # Записываем обратно в Neo4j
            for nid, vec in zip(ids, embeddings):
                session.run("""
                    MATCH (n {id: $id})
                    SET n.embedding = $embedding
                """, id=nid, embedding=vec)

    print(f"✅ Эмбеддинги сгенерированы для {len(nodes)} узлов")

5. Гибридный поиск (GraphRAG): графовые паттерны + семантика

Магия GraphRAG — в гибридном поиске. Вместо того чтобы искать ответ только в изолированных чанках текста, мы комбинируем два подхода: векторный поиск по эмбеддингам узлов (семантическая близость) и графовый поиск по связям (структурная близость). Результат — ответы, которые учитывают не только похожие формулировки, но и реальные связи между фактами.

Вот четыре ключевых поисковых паттерна GraphRAG, реализованных на Cypher:

# ПАТТЕРН 1: Векторный поиск узлов (семантическая близость)
def vector_search(db, query_text, top_k=5):
    query_embedding = client.embeddings.create(
        model=EMBEDDING_MODEL,
        input=query_text
    ).data[0].embedding

    with db.driver.session() as session:
        result = session.run("""
            CALL db.index.vector.queryNodes(
                'entity_embeddings', $k, $embedding
            )
            YIELD node, score
            RETURN node.name AS name, node.embedding_text AS text, score
            ORDER BY score DESC
        """, k=top_k, embedding=query_embedding)
        return [dict(r) for r in result]

# ПАТТЕРН 2: Поиск соседей первого уровня
def neighbor_search(db, entity_id, depth=1):
    """Возвращает всех соседей узла на заданную глубину."""
    with db.driver.session() as session:
        result = session.run("""
            MATCH (n {id: $id})-[r*1..{depth}]-(neighbor)
            RETURN DISTINCT neighbor.name AS name,
                   type(r[0]) AS relation,
                   labels(neighbor)[0] AS type
            LIMIT 20
        """, id=entity_id, depth=depth)
        return [dict(r) for r in result]

# ПАТТЕРН 3: Поиск кратчайшего пути
def path_search(db, entity_a_id, entity_b_id):
    """Находит все кратчайшие пути между двумя сущностями."""
    with db.driver.session() as session:
        result = session.run("""
            MATCH p = shortestPath((a {id: $id_a})-[*]-(b {id: $id_b}))
            RETURN [node IN nodes(p) | node.name] AS path,
                   length(p) AS hops
        """, id_a=entity_a_id, id_b=entity_b_id)
        return [dict(r) for r in result]

# ПАТТЕРН 4: Гибридный — векторный поиск → графовое расширение
def hybrid_search(db, query_text, top_k=3, neighbor_depth=1):
    """
    1. Находит top-k семантически близких узлов
    2. Для каждого собирает соседей-первого уровня
    3. Возвращает объединённый контекст
    """
    # Шаг 1: векторный поиск
    top_nodes = vector_search(db, query_text, top_k)

    # Шаг 2: графовое расширение для каждого найденного узла
    context_chunks = []
    for node in top_nodes:
        # Получаем соседей через Cypher
        with db.driver.session() as session:
            neighbors = session.run("""
                MATCH (n {name: $name})-[r]-(m)
                RETURN n.name AS source,
                       type(r) AS relation,
                       m.name AS target,
                       labels(m)[0] AS target_type
                LIMIT 10
            """, name=node["name"])

            for n in neighbors:
                context_chunks.append(
                    f"[{n['source']}] --{n['relation']}--> [{n['target']}] (тип: {n['target_type']})"
                )

    return context_chunks

6. Интеграция с AI-агентом: LangChain GraphCypherQAChain

Финальный этап — сборка полноценного AI-агента. Используем LangChain с модулем GraphCypherQAChain. Этот компонент принимает вопрос на естественном языке, генерирует Cypher-запрос через LLM, выполняет его в Neo4j, получает результаты и формулирует ответ. Всё это происходит в одном пайплайне, без ручного написания запросов.

Дополнительно реализуем кастомный агент, который комбинирует GraphCypherQAChain с гибридным поиском из секции 5 — так агент сможет выбирать оптимальную стратегию поиска в зависимости от типа вопроса.

# Установка зависимостей
pip install langchain langchain-openai langchain-neo4j neo4j

from langchain_neo4j import Neo4jGraph, GraphCypherQAChain
from langchain_openai import ChatOpenAI

# 1. Подключение к графу
graph = Neo4jGraph(
    url="bolt://localhost:7687",
    username="neo4j",
    password="graphrag2024",
    enhanced_schema=True  # Авто-обогащение схемы
)

# 2. LLM для генерации Cypher и ответов
llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0
)

# 3. GraphCypherQAChain — автоматический Text-to-Cypher
chain = GraphCypherQAChain.from_llm(
    llm=llm,
    graph=graph,
    verbose=True,
    validate_cypher=True,  # Валидация синтаксиса Cypher
    top_k=10  # Максимум результатов
)

# 4. Задаём вопрос — агент сам напишет Cypher!
response = chain.invoke({
    "query": "Какие компании основал Илон Маск и где они расположены?"
})

print(response["result"])

# Вывод (пример):
> Entering new GraphCypherQAChain chain...
> Generated Cypher:
> MATCH (p:Person {name: 'Илон Маск'})-[r:FOUNDED]->(c:Organization)
> OPTIONAL MATCH (c)-[l:LOCATED_IN]->(loc:Location)
> RETURN c.name AS company, loc.name AS location
>
> Full Context:
> [{'company': 'SpaceX', 'location': 'Хоторн, Калифорния'},
>  {'company': 'Tesla', 'location': 'Остин, Техас'},
>  {'company': 'Neuralink', 'location': 'Фримонт, Калифорния'}]
>
> Илон Маск основал несколько компаний: SpaceX находится в
> Хоторне (Калифорния), Tesla — в Остине (Техас), а Neuralink
> базируется во Фримонте (Калифорния).

# Кастомный GraphRAG-агент с гибридной стратегией
def graphrag_agent_ask(db, graph, llm, question):
    """
    Гибридный агент:
    - Простые фактические вопросы → GraphCypherQAChain
    - Сложные / открытые вопросы → гибридный поиск + LLM
    """

    # Определяем тип вопроса (упрощённая классификация)
    fact_keywords = ["кто", "где", "когда", "сколько",
                     "какая компания", "перечисли"]
    is_factual = any(kw in question.lower() for kw in fact_keywords)

    if is_factual:
        # Стратегия 1: Text-to-Cypher
        chain = GraphCypherQAChain.from_llm(
            llm=llm, graph=graph, verbose=False,
            validate_cypher=True
        )
        return chain.invoke({"query": question})["result"]
    else:
        # Стратегия 2: Гибридный поиск + LLM-ответ
        chunks = hybrid_search(db, question, top_k=5)
        context = "\n".join(chunks)

        response = llm.invoke(f"""На основе графа знаний ответь на вопрос.
Контекст (тройки графа):
{context}

Вопрос: {question}

Ответ:""")
        return response.content

# Тестируем:
print(graphrag_agent_ask(
    db, graph, llm,
    "Как связаны SpaceX и NASA?"
))

🎯 Итог: что мы построили

Мы прошли полный путь создания GraphRAG-системы: от установки Neo4j до работающего AI-агента, который понимает не только текст, но и связи между фактами. Ключевое преимущество такого подхода — структурное понимание предметной области, недоступное классическому RAG.

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