Конечно-конечная архитектура GraphRAG с FalkorDB, эмбеддингами и маршрутизацией LangGraph

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

В этом руководстве создается рабочий процесс обработки данных о фильмах: FalkorDB служит хранилищем графов, вкладки Hugging Face MiniLM используются для семантического поиска, модели, размещенные на Groq, отвечают за маршрутизацию и формирование ответов, инструменты LangChain помогают с получением данных, а LangGraph координирует весь процесс. Цель не в демонстрационном примере, а в предоставлении шаблона, который можно адаптировать к любой области, где связи между объектами имеют такое же значение, как и текстовые абзацы.

Почему графы сочетаются с векторами

Векторный поиск находит похожие фрагменты. Запросы к графу находят определенные пути. Если запросить фильмы, снятые Кристофером Ноланом, графовая связь окажется более точной, чем предположение на основе ближайшего соседа. Если запросить «фильмы о сложных кражах в духе „Инсайта“», тогда векторы покажут себя с лучшей стороны. Настоящим ассистентам нужны и те, и другие инструменты, плюс механизм выбора того, по какому пути действовать — или когда использовать гибридный подход.

MATCH (d:Director {name: 'Christopher Nolan'})-[:DIRECTED]->(m:Movie) RETURN m.title
MATCH (m:Movie)-[r]->(d:Director) RETURN m, r, d

Среда и FalkorDB

Создайте виртуальную среду, установите Jupyter, если хотите использовать ноутбук для исследований, и закрепите версии. Интерфейс LangChain может меняться; закрепление версий — часть обеспечения воспроизводимости.

$ pip install jupyterlab
$ jupyter lab

# 1. SETUP WSL (Run in Windows PowerShell as Administrator)
wsl --install
# Restart your computer if prompted, then create your Linux username/password.
# Inside the WSL/Ubuntu terminal, update packages:
sudo apt update && sudo apt upgrade -y

# 2. SET ENVIRONMENT VARIABLES (Paste your keys from Groq & Hugging Face)
export GROQ_API_KEY="your_groq_api_key_here"
export HUGGINGFACEHUB_API_TOKEN="your_hf_token_here"

# 3. ENVIRONMENT & DEPENDENCY SETUP
# Install Python pip and virtual environment tools
sudo apt install python3-pip python3-venv -y

# Create and activate a project directory
mkdir movie-graphrag && cd movie-graphrag
python3 -m venv venv
source venv/bin/activate


# Verify everything is correctly configured
echo "Groq Key: $GROQ_API_KEY"
echo "HF Token: $HUGGINGFACEHUB_API_TOKEN"

Запустите FalkorDB локально с помощью Docker или воспользуйтесь управляемой инстансой, если это подходит вашей команде. Локальный Docker позволяет дешево проводить эксперименты:

docker run -p 6379:6379 -p 3000:3000 -it --rm -v ./data:/data falkordb/falkordb:edge

После удаления старых установок LangChain установите точный набор библиотек, на котором тестировалась эта рабочая процедура:

pip uninstall langchain langchain-core langchain-community langchain-groq \
              langchain-huggingface langgraph -y

pip install falkordb pandas pydantic sentence-transformers
pip install "langchain==0.3.26"
pip install "langchain-core==0.3.68"
pip install "langchain-community==0.3.23"
pip install "langchain-groq==0.2.5"
pip install "langchain-huggingface==0.1.2"
pip install "langgraph==0.3.34"

Загрузка набора данных в стиле IMDB

Скачайте CSV-файл с данными фильмов в стиле Top-1000, затем загрузите его с использованием импортов и настроек подключения, ожидаемых вашим ноутбуком:

import os, re, sys, logging
import pandas as pd
from typing import List, TypedDict, Literal
from falkordb import FalkorDB
from langchain_community.document_loaders.csv_loader import CSVLoader
from langchain_community.vectorstores.falkordb_vector import FalkorDBVector
from langchain_community.graphs import FalkorDBGraph
from langchain_community.chains.graph_qa.falkordb import FalkorDBQAChain
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain.output_parsers import PydanticToolsParser
from pydantic import BaseModel, Field          # pydantic v2 — not langchain_core.pydantic_v1
from langchain_huggingface import HuggingFaceEmbeddings  # not langchain.embeddings (deprecated)
from langchain_groq import ChatGroq
from langgraph.graph import END, StateGraph

GROQ_API_KEY = "your_groq_api_key"
CSV_PATH     = "imdb_top_1000.csv"
os.environ["HF_TOKEN"] = "your_hf_token"

logging.basicConfig(stream=sys.stdout, level=logging.WARNING)

# Connect to FalkorDB and select the graph
db    = FalkorDB(host='localhost', port=6379)
graph = db.select_graph('imdb')

data = pd.read_csv(CSV_PATH)

Создание графа знаний

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

def clean_column(column_data):
    """Remove special characters that would break Cypher queries."""
    return column_data.apply(
        lambda x: re.sub(r"[^a-zA-Z0-9\s]", "", x) if pd.notnull(x) else 'NA'
    )

for col in ['Series_Title', 'Director', 'Certificate', 'Runtime', 'Genre',
            'Overview', 'Star1', 'Star2', 'Star3', 'Star4']:
    data[col] = clean_column(data[col])

data['IMDB_Rating']   = data['IMDB_Rating'].fillna('NA')
data['Released_Year'] = pd.to_numeric(data['Released_Year'], errors='coerce').fillna(-1)

star_columns = ['Star1', 'Star2', 'Star3', 'Star4']

for index, row in data.iterrows():
    movie_title   = row['Series_Title']
    director_name = row['Director']
    released_year = int(row['Released_Year']) if row['Released_Year'] != -1 else None

    if released_year is None:
        print(f"Skipping '{movie_title}' — invalid year.")
        continue

    movie_node = (
        f"MERGE (m:Movie {{"
        f"title: '{movie_title}', year: {released_year}, "
        f"certificate: '{row['Certificate']}', runtime: '{row['Runtime']}', "
        f"genre: '{row['Genre']}', imdb_rating: {row['IMDB_Rating']}, "
        f"description: '{row['Overview']}'"
        f"}})"
    )
    director_node     = f"MERGE (d:Director {{name: '{director_name}'}})"
    directed_relation = "MERGE (d)-[:DIRECTED]->(m)"

    star_nodes, starred_relations = "", ""
    for s, star_col in enumerate(star_columns, start=1):
        star_name = row[star_col]
        if star_name != 'NA':
            star_nodes        += f"MERGE (s{s}:Star {{name: '{star_name}'}})\n"
            starred_relations += f"MERGE (s{s})-[:STARRED_IN]->(m)\n"

    cypher_query = (
        f"{movie_node} {director_node} {directed_relation} "
        f"{star_nodes.strip()}\n{starred_relations.strip()}"
    )
    graph.query(cypher_query)

print("Knowledge Graph created.")

Эмбеддинги для семантического поиска

Используйте all-MiniLM-L6-v2 через современный пакет интеграции Hugging Face вместо устаревших путей импорта:

embedding_model = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-MiniLM-L6-v2"
)

result = graph.query(
    "MATCH (m:Movie) RETURN m.title AS title, m.description AS description"
)

documents = [
    Document(
        page_content=row[1] if row[1] else "",
        metadata={"title": row[0], "description": row[1]}
    )
    for row in result.result_set if row[1]
]

vector_index = FalkorDBVector(host='localhost', port=6379, embedding=embedding_model)
vector_index.add_documents(documents)
print("Vector index created.")

Храните векторы рядом с графом или согласованно с ним, чтобы гибридные узлы могли получать названия фильмов из семантического поиска, а затем расширять их с помощью отношений.

GraphState для LangGraph

Определите TypedDict (или эквивалентный объект) GraphState, который будет хранить вопрос пользователя, решение о маршрутизации, подзапросы, результаты векторного поиска, ответы графа и окончательный ответ. LangGraph сохраняет этот состояние между узлами, чтобы последующие запросы могли использовать уже имеющийся контекст вместо того, чтобы начинаться с пустого запроса.

class GraphState(TypedDict):
    """
    Tracks state across the entire LangGraph workflow.
    Every node reads from and writes to this dictionary.
    """
    question:            str
    documents:           object
    movie_titles:        List[str]
    subqueries:          object
    prompt_with_context: object

Маршрутизация запросов

Не каждый вопрос должен обрабатываться каждой подсистемой. Маршрутизатор определяет, является ли вопрос преимущественно реляционным, преимущественно семантическим, гибридным или не относится к сфере действия. Это снижает задержки и уменьшает количество противоречивых данных в контексте.

llm = ChatGroq(groq_api_key=GROQ_API_KEY, model_name="llama-3.3-70b-versatile")

class RouteQuery(BaseModel):
    """Route a user query to the most relevant datasource."""
    datasource: Literal["vector search", "graph query"] = Field(
        ...,
        description="Route to 'vector search' for similarity queries, "
                    "'graph query' for relationship/fact queries.",
    )

structured_llm_router = llm.with_structured_output(RouteQuery)

route_system_prompt = """You are an expert at routing user questions.

Route to VECTOR SEARCH for questions involving:
- Similarity (similar, related, like, closest, relevant)
- Themes or topics (movies about space, about love, etc.)
- "Find movies similar to X"

Route to GRAPH QUERY for questions involving:
- Specific facts (who directed X, what year was X released)
- Specific people (movies by Christopher Nolan)
- Filtering by property (movies from 1990, Action movies)
"""

route_prompt    = ChatPromptTemplate.from_messages([
    ("system", route_system_prompt),
    ("human",  "{question}"),
])
question_router = route_prompt | structured_llm_router

def route_question(state: GraphState):
    source = question_router.invoke({"question": state["question"]})
    if source.datasource == "vector search":
        return "decomposer"
    else:
        return "graph_qa"

Разборщик

При выполнении гибридного пути разборщик разделяет вопрос пользователя на подзапросы, которые могут быть обработаны узлами векторного и графового поиска:

class SubQuery(BaseModel):
    """A single sub-query decomposed from the original question."""
    sub_query: str = Field(
        ..., description="A unique paraphrasing of the original question.",
    )

decompose_system = """You are an expert at breaking down complex questions.
Given a user question, split it into exactly two sub-queries:
  1. A query for vector similarity search (find related/similar items)
  2. A query for a graph database (find specific relationships or facts)

Example:
Question: Find movies directed by Christopher Nolan related to dreams.
sub_query 1: Find movies related to dreams or the subconscious.
sub_query 2: Find movies directed by Christopher Nolan.
"""

decompose_prompt = ChatPromptTemplate.from_messages([
    ("system", decompose_system),
    ("human",  "{question}"),
])

llm_with_tools = llm.bind_tools([SubQuery])
parser         = PydanticToolsParser(tools=[SubQuery])
query_analyzer = decompose_prompt | llm_with_tools | parser

def decomposer(state: GraphState):
    question   = state["question"]
    subqueries = query_analyzer.invoke({"question": question})
    return {"subqueries": subqueries, "question": question}

Узел векторного поиска

Реализуйте поиск по сходству с использованием текущих инструментов получения данных из LangChain:

def vector_search(state: GraphState):
    question  = state["question"]
    retriever = vector_index.as_retriever(search_kwargs={"k": 3})

    qa_prompt = ChatPromptTemplate.from_messages([
        ("system",
         "You are a movie expert. Use the context below to answer the question.\n\n"
         "Context:\n{context}"),
        ("human", "{input}"),
    ])

    combine_docs_chain = create_stuff_documents_chain(llm, qa_prompt)
    retrieval_chain    = create_retrieval_chain(retriever, combine_docs_chain)

    result      = retrieval_chain.invoke({"input": question})
    source_docs = result.get("context", [])

    extracted_data, movie_titles = [], []
    for doc in source_docs:
        title       = doc.metadata.get("title", "Unknown")
        description = doc.metadata.get("description", "")
        extracted_data.append({"title": title, "description": description})
        movie_titles.append(title)

    return {
        "movie_titles": movie_titles,
        "documents":    extracted_data,
        "question":     question,
        "subqueries":   state["subqueries"],
    }

Узел графового поиска

Переводите вопросы, связанные с отношениями между элементами, в запросы к графу. Для вопросов о фильмографии режиссера граф должен возвращать названия фильмов, соединённые ребрами типа «снял», а не парафразированные предположения:

def graph_qa(state: GraphState):
    question      = state["question"]
    graph_wrapper = FalkorDBGraph(database="imdb", host="localhost", port=6379)

    chain  = FalkorDBQAChain.from_llm(
        llm,
        graph=graph_wrapper,
        verbose=True,
        allow_dangerous_requests=True,
    )
    result = chain.invoke({"query": question})
    answer = result.get("result", str(result))

    return {"documents": answer, "question": question}

Вопросы к графу с векторным контекстом

Гибридный узел берёт названия, полученные на основе семантического анализа, и расширяет их с помощью структуры графа — актёрский состав, съёмочная группа, связанные фильмы — прежде чем большая языковая модель сформирует окончательный ответ:

def graph_qa_with_context(state: GraphState):
    titles      = state["movie_titles"]
    question    = state["question"]
    input_query = (
        f"Return the title, genre, director, and IMDB rating of movies "
        f"with these titles: {str(titles)}"
    )

    graph_wrapper = FalkorDBGraph(database="imdb", host="localhost", port=6379)
    chain  = FalkorDBQAChain.from_llm(
        llm,
        graph=graph_wrapper,
        verbose=True,
        allow_dangerous_requests=True,
        return_intermediate_steps=True,
    )
    result  = chain.invoke({"query": input_query})
    context = ""
    steps   = result.get("intermediate_steps", [])
    if len(steps) > 1:
        context = str(steps[1].get("context", ""))

    return {
        "documents": context,
        "subqueries": state["subqueries"],
        "question":   question,
    }

Узел ответа

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

def respond(state: GraphState):
    question = state["question"]
    context  = state["documents"]

    respond_prompt = ChatPromptTemplate.from_messages([
        ("system",
         "You are a helpful movie expert assistant.\n"
         "Use the context below to answer the user's question clearly and concisely.\n\n"
         "Context:\n{context}"),
        ("human", "{question}"),
    ])

    answer = (respond_prompt | llm).invoke({"context": context, "question": question})
    return {"documents": answer.content}

Связь узлов в рабочем процессе LangGraph

Соедините узлы маршрутизатора, декомпозитора, векторного модуля, графа, гибридного узла и узла ответа с помощью условных рёбер, соответствующих меткам маршрута:

workflow = StateGraph(GraphState)

# Register all nodes
workflow.add_node("graph_qa",              graph_qa)
workflow.add_node("decomposer",            decomposer)
workflow.add_node("vector_search",         vector_search)
workflow.add_node("graph_qa_with_context", graph_qa_with_context)
workflow.add_node("respond",               respond)

# Conditional entry point — route_question decides the first node
workflow.set_conditional_entry_point(
    route_question,
    {
        "decomposer": "decomposer",
        "graph_qa":   "graph_qa",
    },
)

# Hybrid path
workflow.add_edge("decomposer",            "vector_search")
workflow.add_edge("vector_search",         "graph_qa_with_context")
workflow.add_edge("graph_qa_with_context", "respond")
workflow.add_edge("respond",               END)

# Direct graph path
workflow.add_edge("graph_qa", END)

app = workflow.compile()

Визуализируйте граф для документации и обучения пользователей:

from IPython.display import Image, display

try:
    display(Image(app.get_graph().draw_mermaid_png()))
except Exception:
    pass

Тестирование и интерактивная консоль

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

# Graph query — specific fact question
result1 = app.invoke({"question": "Which movies did Christopher Nolan direct?"})
print(result1["documents"])

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

print("\n" + "=" * 60)
print("INTERACTIVE MODE — type a question (or 'quit' to exit)")
print("=" * 60)

while True:
    user_question = input("\nYour question: ").strip()
    if user_question.lower() in ("quit", "exit", "q"):
        print("Goodbye!")
        break
    if not user_question:
        continue
    try:
        result = app.invoke({"question": user_question})
        print("\nANSWER:", result["documents"])
    except Exception as e:
        print(f"Error: {e}")

Результаты, которые стоит проверить

Убедитесь, что запросы к режиссерам в стиле Нолана возвращают названия, основанные на графе, что запросы к фильмам по атмосфере возвращают семантически обоснованные картины, и что гибридные запросы приводят примеры обоих типов доказательств. Если ответы отклоняются, изучайте данные о состоянии между узлами, а не только окончательную строку.

Краткое резюме и советы по адаптации

GraphRAG — это проблема маршрутизации в той же степени, что и проблема хранения. FalkorDB (или любая база данных графов с низкой задержкой) обеспечивает структурированные данные. Эмбеддинги позволяют достигать более высокой точности воспроизведения информации. LangGraph делает поток управления достаточно явным для тестирования. Замените примеры фильмов на элементы вашей области деятельности — комплектующие и поставщики, пациенты и протоколы лечения, услуги и зависимости — сохраняя при этом те же роли узлов: маршрутизация, получение данных, расширение информации, ответ.

Фиксируйте версии, храните набор идеальных вопросов для каждого типа маршрута и ведите журнал изменений состояния. Такая дисциплина в работе важнее, чем то, с какого набора данных для демонстрации вы начинали.

Заметки по проектированию, выдерживающие нагрузку в реальных условиях

Разрешение сущностей будет играть ключевую роль в реальных развертываниях. Названия фильмов относительно просты в обработке; названия компаний и уникальные идентификаторы продуктов — нет. Планируйте этап нормализации перед созданием узлов и ведите справочник соответствия между исходными строками и каноническими ID. Без этого запросы к графу становятся фрагментированными, а результаты поиска указывают на дубликаты.

Для эволюции схемы требуются миграции, как и в SQL. Добавление нового типа связи просто в ноутбуке, но затруднено, когда дашборды зависят от старых связей. Версионируйте схему графа и храните её рядом с кодом рабочего процесса.

Бюджеты на задержку должны быть чётко определены. Вызовы LLM-маршрутизатора добавляют время ещё до начала получения данных. Для систем с высокой частотой запросов рассмотрите использование правил или простых классификаторов для очевидных реляционных паттернов («кто снял X») и оставьте маршрутизацию с помощью LLM для неоднозначного текста.

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

Безопасность имеет большое значение: генерация запросов к графу должна быть ограничена. Предпочитайте параметризованные шаблоны вместо свободноформатного Cypher, сгенерированного моделью. Здесь применима та же правило, что и для инструментов SQL.

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

При объяснении этого в интервью сначала нарисуйте объект состояния, затем рёбра. Интервьюеры быстрее понимают поток управления, чем названия пакетов. Упомяните FalkorDB для высокой скорости обработки свойств-графов, MiniLM для локальных эмбеддингов, а Groq — лишь как один из возможных хостов больших языковых моделей; ваша архитектура не должна быть привязана к одному поставщику инференса.

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

Сохраняйте инструменты уровня e в зеленом цвете перед расширением схемы графа до новых типов сущностей.Должны оставаться зелеными до расширения схемы графа новыми типами сущностей.Переведите приведённый ниже текст с английского на русский. Выведите ТОЛЬКО русскый перевод:

• Графы финансовых преступлений с участием нескольких агентов с использованием LangChain, LangGraph и LangSmith — маршрутизация операций надзирателем, параллельные проверки, подграфы, прерывания, оценка и механизмы контроля для рабочих процессов уровня расследований.
• Временный семантический поиск с использованием LangGraph в случаях, когда не следует хранить векторы — вкладки, действующие в рамках сессии и исчезающие при её завершении, помогают четко определять границы использования данных; это также упрощает последующую замену хранилища на Redis или Azure AI Search путем изменения адаптера хранения.