基于FalkorDB、嵌入模型与LangGraph路由的端到端GraphRAG系统

大型语言模型能够基于训练数据流畅作答，但当问题需要新的结构化分析时就会遇到困难——比如哪位导演执导了哪部电影、哪些演员共同出演、两个实体之间通过什么关联相连。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"

可以使用 Docker 在本地运行 FalkorDB，或者根据团队需求选择托管实例。本地 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 风格的数据集

下载一份Top-1000风格电影数据的CSV文件，然后使用笔记本要求的导入方式和连接设置将其加载：

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.")

用于语义检索的嵌入模型

请通过现代的Hugging Face集成包来使用all-MiniLM-L6-v2，而非已过时的导入路径：

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.")

将向量数据与图谱一同存储或协同存储，这样混合节点就能从语义搜索中获取电影标题，再通过关系链进一步扩展这些信息。

LangGraph的GraphState功能

定义一个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能让控制流程足够清晰以便测试。只需将示例中的电影替换为你的领域相关元素——如零部件与供应商、患者与治疗方案、服务与依赖关系——同时保持相同的节点功能：路由、检索、扩展、响应。

固定版本号，为每种路由类型保留一套标准问题集，并记录状态变化。这种操作规范比初始使用的演示数据集更为重要。

经实际应用验证的设计要点

在实际应用中，实体解析将起到关键作用。电影标题相对规整，而公司名称和产品SKU则不然。在创建节点之前应先进行规范化处理，并建立从原始字符串到标准ID的映射关系。否则，图查询结果会支离破碎，向量搜索也会指向重复项。

与SQL类似，模式演进也需要相应的迁移机制。在笔记本环境中添加新的关系类型较为容易，但一旦仪表板依赖于旧的关系结构，就会变得十分复杂。应对图模式进行版本控制，并将其与工作流代码一起存储。

延迟预算应当明确设定。在开始检索之前，路由器的LLM调用就已经耗费了时间。对于高QPS的场景，可针对明显的关系模式（如“谁导演了X”）设计规则或小型分类器，将LLM路由功能保留给含义模糊的文本处理。

评估时应将路线准确性与答案质量分开评分。即使答案完美但路线错误，仍会消耗预算；且当某个子系统出现故障时，整个系统也会失效。可通过禁用图结构或向量索引来开展混沌测试，以确认工作流程确实会随之恶化。

安全性至关重要：必须对图查询的生成过程加以限制。应优先使用参数化模板，而非从模型中自动生成的自由格式Cypher语句。此处适用的规则与SQL工具中的相同。

成本控制方面：对于内容稳定的文档，应缓存其嵌入向量；对于相同的问题，在较短的时间阈值内缓存路由决策结果；当纯图结构已能返回高置信度的实体时，应避免采用混合扩展方式。

在面试中解释这一点时，先画出状态对象，再画出连接边。面试官更易理解控制流而非包名。可提及FalkorDB在属性图处理方面的速度优势、MiniLM在本地嵌入方面的表现，而Groq仅作为可选的LLM托管平台之一——你的架构不应过度依赖某一家推理服务提供商。

最后，要记录用户在图表为空或路由器拒绝响应时所看到的错误信息。正是这些隐性的问题导致GraphRAG演示在最初一周后就失去了用户的信任。 在将图表架构扩展为新的实体类型之前，务必确保路由级别的配置保持正常状态。 在将图表架构扩展为新的实体类型之前，务必确保路由级别的配置保持正常状态。 在将图表架构扩展为新的实体类型之前，务必确保路由级别的配置保持正常状态。 在将图表架构扩展为新的实体类型之前，务必确保路由级别的配置保持正常状态。 在将图表架构扩展为新的实体类型之前，务必确保路由级别的配置保持正常状态。 在将图表架构扩展为新的实体类型之前，务必确保路由级别的配置保持正常状态。 在将图表架构扩展为新的实体类型之前，务必确保路由级别的配置保持正常状态。 在将图表架构扩展为新的实体类型之前，务必确保路由级别的配置保持正常状态。 在将图表架构扩展为新的实体类型之前，务必确保路由级别的配置保持正常状态。

在将图表模式扩展为新的实体类型之前，需保持e级固定项为绿色状态。 在将图表模式扩展为新的实体类型之前，需保持路由级固定项为绿色状态。 在将图表模式扩展为新的实体类型之前，需保持路由级固定项为绿色状态。 在将图表模式扩展为新的实体类型之前，需保持路由级固定项为绿色状态。 在将图表模式扩展为新的实体类型之前，需保持路由级固定项为绿色状态。 在将图表模式扩展为新的实体类型之前，需保持路由级固定项为绿色状态。 在将图表模式扩展为新的实体类型之前，需保持路由级固定项为绿色状态。 在将图表模式扩展为新的实体类型之前，需保持路由级固定项为绿色状态。 在将图表模式扩展为新的实体类型之前，需保持路由级固定项为绿色状态。 在将图表模式扩展为新的实体类型之前，需保持路由级固定项为绿色状态。 在将图表模式扩展为新的实体类型之前，需保持路由级固定项为绿色状态。在将图表架构扩展为新的实体类型之前，需保持纹理显示为绿色。 在将图表架构扩展为新的实体类型之前，需保持路由级别的固定项显示为绿色。 在将图表架构扩展为新的实体类型之前，需保持路由级别的固定项显示为绿色。 在将图表架构扩展为新的实体类型之前，需保持路由级别的固定项显示为绿色。 在将图表架构扩展为新的实体类型之前，需保持路由级别的固定项显示为绿色。 在将图表架构扩展为新的实体类型之前，需保持路由级别的固定项显示为绿色。 在将图表架构扩展为新的实体类型之前，需保持路由级别的固定项显示为绿色。 在将图表架构扩展为新的实体类型之前，需保持路由级别的固定项显示为绿色。 在将图表架构扩展为新的实体类型之前，需保持路由级别的固定项显示为绿色。 在将图表架构扩展为新的实体类型之前，需保持路由级别的固定项显示为绿色。 在将图表架构扩展为新的实体类型之前，需保持路由级别的固定项显示为绿色。在将图结构扩展为新的实体类型之前，需确保“en”状态为正常。 在将图结构扩展为新的实体类型之前，需保持路由级别的固定配置处于正常状态。 在将图结构扩展为新的实体类型之前，需保持路由级别的固定配置处于正常状态。 在将图结构扩展为新的实体类型之前，需保持路由级别的固定配置处于正常状态。 在将图结构扩展为新的实体类型之前，需保持路由级别的固定配置处于正常状态。 在将图结构扩展为新的实体类型之前，需保持路由级别的固定配置处于正常状态。 在将图结构扩展为新的实体类型之前，需保持路由级别的固定配置处于正常状态。