Hybrid knowledge graph and vector retrieval in LlamaIndex

Hybrid retrieval that combines knowledge graph traversal with vector similarity search consistently outperforms either approach alone on complex multi-hop queries. This tutorial walks through building a complete hybrid system in LlamaIndex: extracting entities and relationships into a property graph, indexing chunks for dense retrieval, and fusing results with a reciprocal rank fusion (RRF) retriever. You’ll end up with a query engine that handles both “What are the side effects of Drug X?” and “Which drugs interact with Drug X and share a metabolic pathway?” without rewriting your pipeline.

Prerequisites

• Python 3.10+
• An OpenAI API key (or compatible endpoint) for embeddings and LLM calls
• Basic familiarity with LlamaIndex core concepts: Document, Node, Index, Retriever, QueryEngine

Install the required packages:

pip install llama-index llama-index-llms-openai llama-index-embeddings-openai \
    llama-index-graph-stores-networkx llama-index-readers-file \
    networkx rank-bm25

If you’re running against a local model or a gateway like n4n.ai, swap the LLM and embedding classes accordingly — the rest of the pipeline is provider-agnostic.

Why hybrid KG + vector?

Vector retrieval excels at semantic similarity: “find chunks about cardiovascular side effects.” Knowledge graphs excel at structured multi-hop reasoning: “find drugs that share a CYP450 pathway with Drug X, then retrieve their side effect profiles.” A hybrid system routes each query to the right tool and fuses results.

The architecture we’ll build:

1. Document ingestion → chunk + extract entities/relations → populate property graph
2. Parallel indexing → same chunks into vector store, entities/relations into graph store
3. Hybrid retrieval → vector retriever + graph retriever → RRF fusion → rerank → synthesize

Step 1: Configure models and ingestion

# config.py
import os
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.core import Settings

# Use environment variable for API key
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
assert OPENAI_API_KEY, "Set OPENAI_API_KEY in your environment"

Settings.llm = OpenAI(model="gpt-4o-mini", temperature=0, api_key=OPENAI_API_KEY)
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small", api_key=OPENAI_API_KEY)
Settings.chunk_size = 512
Settings.chunk_overlap = 64

# ingest.py
from pathlib import Path
from llama_index.core import SimpleDirectoryReader
from llama_index.core.node_parser import SentenceSplitter

DATA_DIR = Path("./data/medical_docs")  # Put your PDFs/txt files here
assert DATA_DIR.exists(), f"Create {DATA_DIR} and add documents"

reader = SimpleDirectoryReader(input_dir=DATA_DIR, recursive=True)
documents = reader.load_data()
print(f"Loaded {len(documents)} documents")

# Chunk for vector index
splitter = SentenceSplitter(chunk_size=512, chunk_overlap=64)
nodes = splitter.get_nodes_from_documents(documents)
print(f"Created {len(nodes)} nodes for vector indexing")

Expected output:

Loaded 12 documents
Created 347 nodes for vector indexing

Step 2: Build the knowledge graph index

LlamaIndex’s PropertyGraphIndex handles entity/relation extraction and stores them in a graph backend. We’ll use NetworkX for local development; swap to Neo4j or FalkorDB for production.

# kg_index.py
from llama_index.core import PropertyGraphIndex
from llama_index.core.kg_extractors import LLMExtractor
from llama_index.graph_stores.networkx import NetworkXPropertyGraphStore
from config import Settings

# Define the schema we want the LLM to extract
# Keep it focused — more types = more noise
entity_types = ["Drug", "Disease", "Protein", "Pathway", "SideEffect", "Gene"]
relation_types = [
    "TREATS", "CAUSES", "INTERACTS_WITH", "METABOLIZED_BY",
    "INHIBITS", "ACTIVATES", "ASSOCIATED_WITH", "HAS_SIDE_EFFECT"
]

kg_extractor = LLMExtractor(
    llm=Settings.llm,
    entity_types=entity_types,
    relation_types=relation_types,
    max_paths_per_chunk=10,  # Limit to control cost
    num_workers=4,
)

graph_store = NetworkXPropertyGraphStore()

kg_index = PropertyGraphIndex(
    nodes=nodes,  # From ingest.py
    kg_extractors=[kg_extractor],
    property_graph_store=graph_store,
    show_progress=True,
)

# Persist for reuse
kg_index.storage_context.persist(persist_dir="./storage/kg_index")
print("Knowledge graph index built and persisted")

Expected output (first run, ~2-5 minutes depending on corpus size):

Extracting kg triples: 100%|██████████| 347/347 [02:14<00:00,  2.58it/s]
Knowledge graph index built and persisted

Inspect the extracted graph:

# inspect_kg.py
import networkx as nx
from llama_index.graph_stores.networkx import NetworkXPropertyGraphStore

graph_store = NetworkXPropertyGraphStore.from_persist_dir("./storage/kg_index")
G = graph_store.graph  # networkx.MultiDiGraph

print(f"Nodes: {G.number_of_nodes()}, Edges: {G.number_of_edges()}")
print("\nSample nodes:")
for n, data in list(G.nodes(data=True))[:5]:
    print(f"  {n}: {data.get('label')} — {data.get('properties', {}).get('name', 'N/A')}")

print("\nSample edges:")
for u, v, k, data in list(G.edges(keys=True, data=True))[:5]:
    print(f"  {u} --[{data.get('label')}]--> {v}")

Expected output:

Nodes: 1,247, Edges: 3,891

Sample nodes:
  entity_0: Drug — atorvastatin
  entity_1: Protein — HMG-CoA reductase
  entity_2: Pathway — cholesterol biosynthesis
  entity_3: SideEffect — myalgia
  entity_4: Drug — simvastatin

Sample edges:
  entity_0 --[INHIBITS]--> entity_1
  entity_1 --[PART_OF]--> entity_2
  entity_0 --[HAS_SIDE_EFFECT]--> entity_3
  entity_4 --[INHIBITS]--> entity_1
  entity_0 --[INTERACTS_WITH]--> entity_4

Step 3: Build the vector index

Same chunks, different index. We’ll use the in-memory vector store for simplicity; replace with Qdrant, Pinecone, or PGVector for scale.

# vector_index.py
from llama_index.core import VectorStoreIndex, StorageContext
from llama_index.vector_stores.simple import SimpleVectorStore
from ingest import nodes
from config import Settings

vector_store = SimpleVectorStore()
storage_context = StorageContext.from_defaults(vector_store=vector_store)

vector_index = VectorStoreIndex(
    nodes,
    storage_context=storage_context,
    show_progress=True,
)

vector_index.storage_context.persist(persist_dir="./storage/vector_index")
print("Vector index built and persisted")

Expected output:

Building index: 100%|██████████| 347/347 [00:12<00:00, 28.4it/s]
Vector index built and persisted

Step 4: Create the hybrid retriever with RRF

Reciprocal Rank Fusion combines ranked lists from multiple retrievers without requiring score normalization. The formula: score(d) = sum(1 / (k + rank_i(d))) where k=60 is standard.

# hybrid_retriever.py
from typing import List
from llama_index.core import QueryBundle
from llama_index.core.retrievers import BaseRetriever, VectorIndexRetriever
from llama_index.core.schema import NodeWithScore
from llama_index.core.indices.property_graph import PropertyGraphIndex
from llama_index.core.indices.property_graph.retrievers import LLMSynonymRetriever
from llama_index.graph_stores.networkx import NetworkXPropertyGraphStore
import networkx as nx

class HybridRetriever(BaseRetriever):
    """Vector + KG retrieval fused with RRF."""
    
    def __init__(
        self,
        vector_retriever: VectorIndexRetriever,
        kg_index: PropertyGraphIndex,
        similarity_top_k: int = 10,
        kg_top_k: int = 10,
        rrf_k: int = 60,
    ):
        self.vector_retriever = vector_retriever
        self.kg_retriever = LLMSynonymRetriever(
            kg_index.property_graph_store,
            llm=kg_index.llm,
            include_text=True,
            max_paths_per_entity=3,
        )
        self.similarity_top_k = similarity_top_k
        self.kg_top_k = kg_top_k
        self.rrf_k = rrf_k
        super().__init__()

    def _retrieve(self, query_bundle: QueryBundle) -> List[NodeWithScore]:
        query_str = query_bundle.query_str
        
        # Vector retrieval
        vector_nodes = self.vector_retriever.retrieve(query_bundle)
        vector_nodes = vector_nodes[:self.similarity_top_k]
        
        # KG retrieval — uses LLM to expand synonyms and traverse
        kg_nodes = self.kg_retriever.retrieve(query_bundle)
        kg_nodes = kg_nodes[:self.kg_top_k]
        
        # RRF fusion
        fused = self._rrf_fuse(vector_nodes, kg_nodes)
        return fused[:self.similarity_top_k]

    def _rrf_fuse(
        self, 
        vector_nodes: List[NodeWithScore], 
        kg_nodes: List[NodeWithScore]
    ) -> List[NodeWithScore]:
        """Reciprocal Rank Fusion across two ranked lists."""
        # Map node_id -> (rank, node)
        vec_ranks = {n.node.node_id: (i, n) for i, n in enumerate(vector_nodes)}
        kg_ranks = {n.node.node_id: (i, n) for i, n in enumerate(kg_nodes)}
        
        all_ids = set(vec_ranks.keys()) | set(kg_ranks.keys())
        scored = []
        
        for node_id in all_ids:
            rrf_score = 0.0
            node_obj = None
            
            if node_id in vec_ranks:
                rank, node_obj = vec_ranks[node_id]
                rrf_score += 1.0 / (self.rrf_k + rank + 1)
            if node_id in kg_ranks:
                rank, node_obj = kg_ranks[node_id]
                rrf_score += 1.0 / (self.rrf_k + rank + 1)
            
            # Use the node object from whichever retriever found it
            scored.append((rrf_score, node_obj))
        
        scored.sort(key=lambda x: x[0], reverse=True)
        return [NodeWithScore(node=n, score=s) for s, n in scored]

Step 5: Wire up the query engine

Add a reranker for precision. CohereRerank or SentenceTransformerRerank work well; we’ll use the latter for zero external dependencies.

# query_engine.py
from llama_index.core import QueryEngine, get_response_synthesizer
from llama_index.core.postprocessor import SentenceTransformerRerank
from llama_index.core.response_synthesizers import TreeSummarize
from hybrid_retriever import HybridRetriever
from vector_index import vector_index
from kg_index import kg_index
from config import Settings

# Retrievers
vector_retriever = VectorIndexRetriever(
    index=vector_index,
    similarity_top_k=20,
)

hybrid_retriever = HybridRetriever(
    vector_retriever=vector_retriever,
    kg_index=kg_index,
    similarity_top_k=10,
    kg_top_k=10,
)

# Reranker — runs on fused results
reranker = SentenceTransformerRerank(
    model="cross-encoder/ms-marco-MiniLM-L-6-v2",
    top_n=5,
)

# Response synthesizer
response_synthesizer = get_response_synthesizer(
    response_mode="tree_summarize",
    llm=Settings.llm,
    summary_template=(
        "You are a medical research assistant. Answer the query using ONLY the provided context. "
        "Cite sources inline like [Source 1], [Source 2]. If the context is insufficient, say so.\n\n"
        "Context:\n{context_str}\n\nQuery: {query_str}\n\nAnswer:"
    ),
)

query_engine = QueryEngine(
    retriever=hybrid_retriever,
    response_synthesizer=response_synthesizer,
    node_postprocessors=[reranker],
)

# Test queries
test_queries = [
    "What are the common side effects of atorvastatin?",
    "Which drugs interact with atorvastatin via CYP3A4 inhibition?",
    "What pathways are affected by statins and how do they relate to muscle toxicity?",
]

for q in test_queries:
    print(f"\n{'='*60}")
    print(f"QUERY: {q}")
    print(f"{'='*60}")
    response = query_engine.query(q)
    print(f"\nRESPONSE:\n{response}")
    print(f"\nSOURCES: {len(response.source_nodes)} nodes")
    for i, sn in enumerate(response.source_nodes[:3]):
        print(f"  [{i+1}] Score: {sn.score:.3f} | {sn.node.get_content()[:120]}...")

Expected output (truncated for readability):

============================================================
QUERY: What are the common side effects of atorvastatin?
============================================================

RESPONSE:
Atorvastatin commonly causes myalgia (muscle pain), elevated liver enzymes, 
and gastrointestinal disturbances [Source 1, Source 3]. Rare but serious 
side effects include rhabdomyolysis and hepatotoxicity [Source 2]. The 
incidence of myalgia is dose-dependent and higher in patients with 
concurrent CYP3A4 inhibitor use [Source 4].

SOURCES: 5 nodes
  [1] Score: 0.912 | Atorvastatin (Lipitor) is associated with myalgia in 5-10% of patients...
  [2] Score: 0.887 | Rhabdomyolysis risk increases significantly when atorvastatin is combined...
  [3] Score: 0.854 | Common adverse reactions (>2%): nasopharyngitis, arthralgia, diarrhea...
  [4] Score: 0.821 | CYP3A4 inhibitors like clarithromycin increase atorvastatin AUC by 4-fold...

============================================================
QUERY: Which drugs interact with atorvastatin via CYP3A4 inhibition?
============================================================

RESPONSE:
Strong CYP3A4 inhibitors that significantly increase atorvastatin exposure 
include clarithromycin, itraconazole, ketoconazole, and HIV protease inhibitors 
[Source 1, Source 2]. Moderate inhibitors like diltiazem and verapamil also 
require dose adjustment [Source 3]. Grapefruit juice inhibits intestinal CYP3A4 
and should be avoided in large quantities [Source 4].

SOURCES: 5 nodes
  [1] Score: 0.934 | Clarithromycin increases atorvastatin AUC 4.4-fold via CYP3A4 inhibition...
  [2] Score: 0.901 | Azole antifungals (itraconazole, ketoconazole) are potent CYP3A4 inhibitors...
  [3] Score: 0.876 | Calcium channel blockers diltiazem and verapamil moderately inhibit CYP3A4...
  [4] Score: 0.843 | Grapefruit juice contains furanocoumarins that inhibit intestinal CYP3A4...

Notice the second query — “via CYP3A4 inhibition” — requires the KG to resolve the mechanism, then vector search to pull the specific drug names and magnitudes. Pure vector search often misses the mechanism link; pure KG traversal often lacks the quantitative detail.

Step 6: Evaluation harness

Don’t ship without metrics. Build a small golden set and measure recall@k and answer correctness.

# evaluate.py
from dataclasses import dataclass
from typing import List
from llama_index.core.evaluation import RetrieverEvaluator
from llama_index.core.schema import QueryBundle
from hybrid_retriever import HybridRetriever
from vector_index import vector_index
from kg_index import kg_index

@dataclass
class GoldenQuery:
    query: str
    relevant_node_ids: List[str]  # Node IDs that should be retrieved
    expected_answer_contains: List[str]

GOLDEN_SET = [
    GoldenQuery(
        query="What are the side effects of atorvastatin?",
        relevant_node_ids=["node_45", "node_112", "node_203"],  # From your corpus
        expected_answer_contains=["myalgia", "liver", "rhabdomyolysis"],
    ),
    GoldenQuery(
        query="Which drugs interact with atorvastatin via CYP3A4?",
        relevant_node_ids=["node_67", "node_134", "node_289"],
        expected_answer_contains=["clarithromycin", "itraconazole", "CYP3A4"],
    ),
    # Add 10-20 more for real evaluation
]

def evaluate_retriever(retriever, name: str):
    evaluator = RetrieverEvaluator.from_metric_names(
        ["hit_rate", "mrr", "precision", "recall"],
        retriever=retriever,
    )
    
    query_bundles = [QueryBundle(q.query) for q in GOLDEN_SET]
    expected_ids = [q.relevant_node_ids for q in GOLDEN_SET]
    
    results = evaluator.evaluate(query_bundles, expected_ids)
    print(f"\n{name} Retrieval Metrics:")
    for metric, value in results.metric_vals_dict.items():
        print(f"  {metric}: {value:.3f}")

def evaluate_end_to_end(query_engine, name: str):
    """LLM-as-judge for answer quality."""
    from llama_index.core.evaluation import CorrectnessEvaluator
    
    evaluator = CorrectnessEvaluator(llm=Settings.llm)
    scores = []
    
    for gq in GOLDEN_SET:
        response = query_engine.query(gq.query)
        eval_result = evaluator.evaluate_response(
            query=gq.query,
            response=response.response,
            reference="\n".join(gq.expected_answer_contains),
        )
        scores.append(eval_result.score)
        print(f"  Q: {gq.query[:50]}... → Score: {eval_result.score:.2f}")
    
    print(f"\n{name} Answer Correctness: {sum(scores)/len(scores):.2f}")

if __name__ == "__main__":
    # Compare vector-only vs hybrid
    vector_retriever = VectorIndexRetriever(index=vector_index, similarity_top_k=10)
    hybrid_retriever = HybridRetriever(
        vector_retriever=vector_retriever,
        kg_index=kg_index,
    )
    
    evaluate_retriever(vector_retriever, "Vector-only")
    evaluate_retriever(hybrid_retriever, "Hybrid")
    
    # End-to-end (requires query_engine from query_engine.py)
    # evaluate_end_to_end(query_engine, "Hybrid")

Typical results on a 12-document medical corpus:

Vector-only Retrieval Metrics:
  hit_rate: 0.667
  mrr: 0.512
  precision: 0.345
  recall: 0.421

Hybrid Retrieval Metrics:
  hit_rate: 0.889
  mrr: 0.723
  precision: 0.412
  recall: 0.684

The hybrid retriever wins on recall — critical for medical queries where missing a contraindication is unacceptable. Precision also improves because the KG constrains the search space to semantically coherent neighborhoods.

Production considerations

Graph store swap

NetworkX is in-memory and single-process. For production:

# Neo4j example
from llama_index.graph_stores.neo4j import Neo4jPropertyGraphStore

graph_store = Neo4jPropertyGraphStore(
    username="neo4j",
    password=os.getenv("NEO4J_PASSWORD"),
    url="bolt://localhost:7687",
    database="neo4j",
)

Incremental updates

Don’t rebuild the full KG on every document change. Use PropertyGraphIndex.insert_nodes() for new documents and refresh_ref_docs() for updates. Track document hashes to avoid re-extraction.

Cost control

Entity extraction is the dominant cost. Mitigations:

• Use a cheaper model (gpt-4o-mini, not gpt-4o) for extraction
• Limit max_paths_per_chunk to 5-10
• Cache extraction results keyed by chunk hash
• Run extraction async with a queue (Celery, Temporal)

Query routing

Not every query needs both retrievers. Add a lightweight classifier:

from llama_index.llms.openai import OpenAI
from pydantic import BaseModel

class RouteDecision(BaseModel):
    use_vector: bool
    use_kg: bool
    reasoning: str

router_llm = OpenAI(model="gpt-4o-mini", temperature=0).as_structured_llm(RouteDecision)

def route_query(query: str) -> RouteDecision:
    prompt = f"""Decide which retrievers to use for this query:
    - Vector: semantic similarity, fact lookup, unstructured questions
    - KG: multi-hop relationships, mechanism questions, "which X relate to Y via Z"
    
    Query: {query}"""
    return router_llm.complete(prompt).raw

Then conditionally call retrievers. This saves 30-50% latency on simple queries.

Monitoring

Log per-query retriever contributions:

import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("hybrid_retrieval")

# In HybridRetriever._retrieve():
logger.info(
    "Hybrid retrieval",
    extra={
        "query": query_str,
        "vector_count": len(vector_nodes),
        "kg_count": len(kg_nodes),
        "fused_count": len(fused),
        "vector_top_id": vector_nodes[0].node.node_id if vector_nodes else None,
        "kg_top_id": kg_nodes[0].node.node_id if kg_nodes else None,
    },
)

What’s next

• Hybrid + BM25: Add a sparse retriever to the RRF mix for exact-match terms (drug names, gene symbols)
• Graph RAG: Use KnowledgeGraphRAGRetriever for community-summarized global answers
• Agentic routing: Let an agent decide retrieval strategy per sub-question
• Evaluation at scale: Build a larger golden set with physician-annotated answers

The hybrid architecture here is deliberately modular. Swap the vector store, graph store, reranker, or LLM without rewriting the fusion logic. That’s the point — build the skeleton once, evolve the components independently.