Кінцево-кінцевий GraphRAG з FalkorDB, ембеддингами та маршрутизацією LangGraph

Великі мовні моделі відповідають плавно на основі даних навчання, але все одно стикаються з труднощами, коли запит вимагає нової структури: хто зрежисував який фільм, які актори брали участь у зйомках, які зв’язки поєднують дві сутності. Системи GraphRAG усувають цю прогалину, поєднуючи векторний пошук у неструктурованому тексті з чіткими запитами до графів щодо взаємозв’язків, а потім дозволяючи LLM обробляти інформацію з обох джерел.

У цьому посібнику створюється функціональний робочий процес для обробки даних про фільми: 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}")

Результати, які варто перевірити

Переконайтеся, що запити у стилі Нолана повертають назви фільмів, засновані на графічних даних, що запити типу “vibe” повертають фільми з семантично обґрунтованим вибором, а гібридні запити посилаються на обидва типи доказів. Якщо відповіді відрізняються, перевіряйте дані стану між вузлами, а не лише кінцевий результат.

Підсумок та поради щодо адаптації

GraphRAG — це насамперед проблема маршрутизації, а не лише проблема зберігання. FalkorDB (або будь-яка база даних графів з низькою затримкою) забезпечує структуровані дані. Ембеддинги допомагають у нечіткому пошуку інформації. LangGraph робить потоки керування достатньо прозорими для тестування. Замініть фільми на елементи вашої сфери діяльності — комплектуючі та постачальники, пацієнти та протоколи, послуги та залежності — залишаючи при цьому ті самі ролі вузлів: маршрутизація, отримання даних, розширення інформації, відповідь.

Фіксуйте версії, зберігайте набір „золотих“ запитань для кожного типу маршруту та фіксуйте зміни стану. Саме така дисципліна у роботі має більше значення, ніж те, з якого набору даних для демонстрації ви почали.

Примітки до проектування, які витримують роботу в продакшені

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

Еволюція схеми вимагає міграцій, як і SQL. Додати новий тип взаємозв’язку легко у нотбуці, але складно, коли дашборди залежать від старих країв. Версіонуйте схему графа та зберігайте її поруч із кодом робочого процесу.

Бюджети на затримки мають бути чітко визначені. Виклики LLM-роутера додають час ще до початку отримання даних. Для систем з високою кількістю запитів розгляньте можливість використання правил чи простих класифікаторів для очевидних взаємозв’язків („хто зрежисерував X“) та залиште використання LLM-роутингу для неоднозначного тексту.

Оцінка має враховувати точність маршруту окремо від якості відповіді. Ідеальна відповідь по неправильному маршруту все одно споживає бюджет та призведе до невдачі, якщо одна з підсистем не працюватиме. Перевірте на стійкість, вимкнувши граф чи векторний індекс, та переконайтеся, що робочий процес справді погіршується.

Безпека має велике значення: генерація запитів до графа має бути обмежена. Віддавайте перевагу параметризованим шаблонам замість вільноформатного Cypher, отриманого з моделі. Тут застосовується те саме правило, що й до інструментів SQL.

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

Під час пояснення цього в інтерв’ю спочатку намалюйте об’єкт стану, а потім ребра. Інтерв’юери швидше слідкують за потоком керування, ніж за назвами пакетів. Згадайте FalkorDB через швидкість роботи з графами властивостей, MiniLM для локальних ембеддингів, а Groq — лише як одного з можливих хостів LLM; ваша архітектура не повинна бути прив’язана до одного постачальника послуг інтерпретації.

Нарешті, задокументуйте повідомлення про помилки, які побачать користувачі, коли граф порожній або роутер не функціонує. Саме через такі „тихі галюцинації“ демонстрації GraphRAG втрачають довіру вже через перший тиждень. Зберігайте інструменти на рівні маршрутів у стані „зеленому“, перш ніж розширювати схему графа на нові типи ентитетів. Зберігайте інструменти на рівні маршрутів у стані „зеленому“, перш ніж розширювати схему графа на нові типи ентитетів. Зберігайте інструменти на рівні маршрутів у стані „зеленому“, перш ніж розширювати схему графа на нові типи ентитетів. Зберігайте інструменти на рівні маршрутів у стані „зеленому“, перш ніж розширювати схему графа на нові типи ентитетів. Зберігайте інструменти на рівні маршрутів у стані „зеленому“, перш ніж розширювати схему графа на нові типи ентитетів. Зберігайте інструменти на рівні маршрутів у стані „зеленому“, перш ніж розширювати схему графа на нові типи ентитетів. Зберігайте інструменти на рівні маршрутів у стані „зеленому“, перш ніж розширювати схему графа на нові типи ентитетів. Зберігайте інструменти на рівні маршрутів у стані „зеленому“, перш ніж розширювати схему графа на нові типи ентитетів. Зберігайте інструменти на рівні маршрутів у стані „зеленому“, перш ніж розширювати схему графа на нові типи ентитетів. Зберігайте інструменти на рівні маршрутів у стані „зеленому“, перш ніж розширювати схему графа на нові типи ентитетів.

Зберігайте елементи рівня e у зеленому кольорі перед розширенням схеми графа на нові типи ентитетів. Зберігайте елементи рівня маршруту у зеленому кольорі перед розширенням схеми графа на нові типи ентитетів. Зберігайте елементи рівня маршруту у зеленому кольорі перед розширенням схеми графа на нові типи ентитетів. Зберігайте елементи рівня маршруту у зеленому кольорі перед розширенням схеми графа на нові типи ентитетів. Зберігайте елементи рівня маршруту у зеленому кольорі перед розширенням схеми графа на нові типи ентитетів. Зберігайте елементи рівня маршруту у зеленому кольорі перед розширенням схеми графа на нові типи ентитетів. Зберігайте елементи рівня маршруту у зеленому кольорі перед розширенням схеми графа на нові типи ентитетів. Зберігайте елементи рівня маршруту у зеленому кольорі перед розширенням схеми графа на нові типи ентитетів. Зберігайте елементи рівня маршруту у зеленому кольорі перед розширенням схеми графа на нові типи ентитетів. Зберігайте елементи рівня маршруту у зеленому кольорі перед розширенням схеми графа на нові типи ентитетів. Зберігайте елементи рівня маршруту у зеленому кольорі перед розширенням схеми графа на нові типи ентитетів. Зберігайте елементи рівня маршруту у зеленому кольорі.Зберігайте елементи на рівні маршруту у зеленому кольорі перед розширенням схеми графа на нові типи ентитетів.Перекладіть наведений нижче текст з англійської на українську. Виведіть лише український переклад:

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