Knowledge Graphs

Structured knowledge representation for agents: construction, querying, and Graph RAG.

Why Knowledge Graphs?

Best of both: Hybrid Graph RAG (vector for retrieval + graph for reasoning)

Graph Construction

From Documents

from llama_index.core import KnowledgeGraphIndex
from llama_index.core import SimpleDirectoryReader
from llama_index.graph_stores import Neo4jGraphStore

graph_store = Neo4jGraphStore(
    username="neo4j",
    password="...",
    url="bolt://localhost:7687"
)

documents = SimpleDirectoryReader("./data").load_data()

kg_index = KnowledgeGraphIndex.from_documents(
    documents,
    graph_store=graph_store,
    max_triplets_per_chunk=10,
    include_embeddings=True,  # Hybrid: graph + vector
    kg_extract_prompt=CUSTOM_KG_PROMPT  # Domain-specific
)

Custom Extraction Prompt

CUSTOM_KG_PROMPT = """
Extract entities and relationships from the text.
Focus on: {domain_entities}
Relationship types: {relationship_types}

Output as triplets: (head, relation, tail)
Include confidence scores.

Text: {text}
"""

Incremental Construction

class IncrementalKGBuilder:
    def __init__(self, graph_store):
        self.store = graph_store

    async def add_document(self, doc: Document):
        # 1. Extract triplets
        triplets = await self.extract_triplets(doc)

        # 2. Resolve entities (entity linking)
        resolved = await self.resolve_entities(triplets)

        # 3. Merge with existing
        await self.merge_graph(resolved)

        # 4. Update embeddings
        await self.update_embeddings(resolved.new_entities)

    async def merge_graph(self, triplets):
        for h, r, t in triplets:
            # Upsert nodes
            await self.store.upsert_node(h)
            await self.store.upsert_node(t)
            # Upsert edge
            await self.store.upsert_edge(h, r, t)

Graph Querying

Cypher (Neo4j)

// Find all proteins interacting with BRCA1
MATCH (p1:Protein {name: "BRCA1"})-[:INTERACTS_WITH]->(p2:Protein)
RETURN p2.name, p2.function

// Find path between two entities
MATCH path = shortestPath((a:Entity {name: $start})-[:RELATES_TO*]-(b:Entity {name: $end}))
RETURN path

// Community detection
CALL gds.louvain.stream('entity-graph')
YIELD nodeId, communityId
RETURN gds.util.asNode(nodeId).name, communityId

Natural Language to Cypher

from llama_index.core.query_engine import GraphRAGQueryEngine

query_engine = GraphRAGQueryEngine(
    graph_store=graph_store,
    llm=llm,
    # LLM generates Cypher from natural language
    text_to_cypher_prompt=CUSTOM_CYPHER_PROMPT
)

response = query_engine.query(
    "What proteins interact with BRCA1 in breast cancer?"
)

Hybrid Query (Graph + Vector)

class HybridGraphVectorQuery:
    def __init__(self, graph_store, vector_store):
        self.graph = graph_store
        self.vector = vector_store

    async def query(self, query: str):
        # 1. Vector search for relevant entities
        entities = await self.vector.search(query, k=10)

        # 2. Graph expansion from entities
        subgraph = await self.graph.get_subgraph(entities, depth=2)

        # 3. Vector search within subgraph
        chunks = await self.vector.search_in_subgraph(query, subgraph, k=5)

        return chunks

Graph RAG Patterns

1. Entity-Centric Retrieval

class EntityCentricRAG:
    async def retrieve(self, query: str):
        # 1. Identify entities in query
        entities = await self.extract_entities(query)

        # 2. Get entity neighborhoods
        neighborhoods = await self.graph.get_neighborhoods(entities, depth=2)

        # 3. Retrieve chunks for entities
        chunks = await self.vector.search_for_entities(neighborhoods)

        return chunks

2. Path-Based Reasoning

class PathBasedRAG:
    async def query(self, query: str):
        # Identify start/end entities
        start, end = await self.identify_endpoints(query)

        # Find paths
        paths = await self.graph.find_paths(start, end, max_depth=4)

        # Retrieve evidence for each path
        evidence = []
        for path in paths:
            chunks = await self.get_evidence(path)
            evidence.append({"path": path, "evidence": chunks})

        # Synthesize
        return await self.synthesize(query, evidence)

3. Community Summarization

class CommunityRAG:
    async def build_communities(self):
        # Detect communities
        communities = await self.graph.detect_communities(algorithm="louvain")

        # Summarize each
        for comm in communities:
            entities = comm.entities
            chunks = await self.get_chunks(entities)
            comm.summary = await llm.summarize(chunks)
            comm.key_entities = entities[:10]

        return communities

    async def query(self, query: str):
        # Find relevant communities
        relevant = await self.match_communities(query)
        # Retrieve from community chunks
        return await self.retrieve_from_communities(relevant)

Graph Schema Design

Entity Types

ENTITY_TYPES = {
    "Person": ["name", "role", "organization"],
    "Organization": ["name", "type", "location"],
    "Concept": ["name", "definition", "category"],
    "Document": ["title", "source", "date"],
    "Event": ["name", "date", "location", "participants"],
    "Product": ["name", "version", "features"],
    "Technical": ["name", "specification", "dependencies"]
}

Relationship Types

RELATIONSHIP_TYPES = {
    "WORKS_FOR": ("Person", "Organization"),
    "LOCATED_IN": ("Entity", "Location"),
    "PART_OF": ("Component", "System"),
    "DEPENDS_ON": ("Module", "Module"),
    "CAUSES": ("Event", "Event"),
    "TREATS": ("Drug", "Disease"),
    "INTERACTS_WITH": ("Protein", "Protein"),
    "MENTIONS": ("Document", "Entity"),
    "DERIVED_FROM": ("Data", "Source"),
    "VERSION_OF": ("Artifact", "Artifact")
}

Schema Validation

class GraphSchemaValidator:
    def validate_triplet(self, head, relation, tail):
        if relation not in RELATIONSHIP_TYPES:
            raise ValueError(f"Unknown relation: {relation}")

        valid_head, valid_tail = RELATIONSHIP_TYPES[relation]
        if not isinstance(head, valid_head):
            raise TypeError(f"Head must be {valid_head}")
        if not isinstance(tail, valid_tail):
            raise TypeError(f"Tail must be {valid_tail}")

Graph Maintenance

Entity Resolution

class EntityResolver:
    async def resolve(self, entities: list[Entity]) -> list[Entity]:
        # 1. Blocking (candidate pairs)
        candidates = self.block(entities)

        # 2. Scoring
        scored = []
        for e1, e2 in candidates:
            score = self.similarity(e1, e2)
            if score > 0.85:
                scored.append((e1, e2, score))

        # 3. Clustering
        clusters = self.cluster(scored)

        # 4. Merge
        merged = []
        for cluster in clusters:
            merged.append(self.merge_cluster(cluster))

        return merged

Graph Pruning

class GraphPruner:
    def prune(self, graph, min_degree=1, max_age_days=365):
        # Remove isolated nodes
        isolated = [n for n in graph.nodes if graph.degree(n) < min_degree]
        graph.remove_nodes_from(isolated)

        # Remove stale edges
        stale = [
            (u, v) for u, v, d in graph.edges(data=True)
            if d.get("timestamp", 0) < time.time() - max_age_days * 86400
        ]
        graph.remove_edges_from(stale)

        return graph

Evaluation

Graph Quality Metrics

RAG Quality with Graph

Tools & Platforms

Related Notes

• RAG Fundamentals — Basic RAG
• Advanced RAG — Graph RAG, Hybrid
• Memory Systems — Graph as memory
• Context Window Management — Graph for context
• LlamaIndex Agents — LlamaIndex KG integration
• Scientific Agents — KG for research
• Coding Agents — Code knowledge graphs
• Neurosymbolic Integration — Graph + neural

Sources: "Graph RAG" (Edge et al., Microsoft, 2024), "Knowledge Graphs for RAG" surveys, Neo4j docs, LlamaIndex KG docs, "Knowledge Graph Construction" surveys, biomedical KG literature (BioBERT, PubMed KG), code KG literature (CodeBERT, GraphCodeBERT).