Knowledge Graph Builder

Knowledge Graph Builder

Overview

Knowledge graphs make implicit relationships explicit, enabling AI systems to reason about connections, verify facts, and reduce hallucinations. They combine structured entity-relationship modeling with semantic search for powerful knowledge retrieval.

When to use: Complex entity relationships central to the domain, verifying AI-generated facts against structured knowledge, semantic search combined with relationship traversal, recommendation systems, fraud detection, or pattern recognition.

When NOT to use: Simple tabular data (use a relational database), purely document-based search with no relationships (use the rag-implementer skill), read-heavy workloads with no traversal needs, or when the team lacks graph modeling expertise. For KB architecture selection and governance, use the knowledge-base-manager skill.

Quick Reference

Common Mistakes

Delegation

• Extract entities and relationships from unstructured text: Use Task agent to run NER pipelines and build relationship triples
• Evaluate graph database options for project requirements: Use Explore agent to compare Neo4j, Neptune, ArangoDB, and TigerGraph against scale and query needs
• Design ontology and hybrid architecture for a new domain: Use Plan agent to define entity types, relationship schemas, and graph-vector integration strategy
• For hybrid KG+RAG systems, delegate to the rag-implementer skill
• For knowledge-graph-powered agent workflows, delegate to the agent-patterns skill

References

• Ontology Design — Entity types, relationships, properties, RDF schema, validation
• Database Selection — Neo4j, Neptune, ArangoDB, TigerGraph comparison and setup
• Entity Extraction — NER pipeline, relationship extraction, LLM-based extraction
• Hybrid Architecture — Graph + vector integration, hybrid search implementation
• Query Patterns — Cypher queries, API design, common traversal patterns
• AI Integration — KG-RAG, hallucination detection, grounded response generation

AI Integration & Hallucination Prevention

Knowledge Graph RAG

Use the knowledge graph to ground LLM responses with structured context:

class KnowledgeGraphRAG:
    def __init__(self, kg_api, llm_client):
        self.kg = kg_api
        self.llm = llm_client

    def retrieve_context(self, query: str) -> str:
        # Extract entities from query
        entities = self.extract_entities_from_query(query)

        # Retrieve relevant subgraph
        subgraph = self.kg.get_subgraph(
            [e.id for e in entities],
            max_hops=2
        )

        # Format subgraph for LLM
        context = self.format_subgraph_for_llm(subgraph)
        return context

    def generate_with_grounding(self, query: str) -> GroundedResponse:
        context = self.retrieve_context(query)

        prompt = f"""
        Context from knowledge graph:
        {context}

        User query: {query}

        Answer based only on the provided context. Include source entities.
        """

        response = self.llm.generate(prompt)

        return GroundedResponse(
            response=response,
            sources=self.extract_sources(context),
            confidence=self.calculate_confidence(response, context)
        )

Hallucination Detection

Verify LLM claims against the knowledge graph:

class HallucinationDetector:
    def __init__(self, knowledge_graph):
        self.kg = knowledge_graph

    def verify_claim(self, claim: str) -> VerificationResult:
        # Parse claim into (subject, predicate, object)
        parsed_claim = self.parse_claim(claim)

        # Query knowledge graph for evidence
        evidence = self.kg.find_evidence(
            parsed_claim.subject,
            parsed_claim.predicate,
            parsed_claim.object
        )

        if evidence:
            return VerificationResult(
                is_supported=True,
                evidence=evidence,
                confidence=evidence.confidence
            )

        # Check for contradictory evidence
        contradiction = self.kg.find_contradiction(parsed_claim)

        return VerificationResult(
            is_supported=False,
            is_contradicted=bool(contradiction),
            contradiction=contradiction
        )

Database Selection

Neo4j (Recommended for Most Projects)

• Query language: Cypher (native), with CalVer versioning (2025.x+)
• Pros: Most mature ecosystem, rich graph algorithms library, excellent visualization tools, native vector index support, property sharding for horizontal scaling
• Cons: Enterprise licensing costs, Community Edition limited for production scale
• Use when: Complex relationship queries, graph algorithms (centrality, community detection, pathfinding), team can learn Cypher
• Editions: Community (free, open-source) and Enterprise (commercial license)

CREATE CONSTRAINT person_id IF NOT EXISTS
FOR (p:Person) REQUIRE p.id IS UNIQUE;

CREATE CONSTRAINT org_name IF NOT EXISTS
FOR (o:Organization) REQUIRE o.name IS UNIQUE;

CREATE INDEX entity_search IF NOT EXISTS
FOR (e:Entity) ON (e.name, e.type);

CREATE INDEX relationship_type IF NOT EXISTS
FOR ()-[r:RELATED_TO]-() ON (r.type, r.confidence);

Amazon Neptune

• Query languages: openCypher, Apache TinkerPop Gremlin, SPARQL (RDF)
• Pros: Fully managed, serverless scaling, Multi-AZ high availability, GraphRAG integration with Amazon Bedrock, GraphStorm for graph ML
• Cons: AWS vendor lock-in, higher cost than self-hosted, limited graph algorithm library compared to Neo4j
• Use when: AWS infrastructure, need managed service, compliance requirements, want GraphRAG with Bedrock
• Products: Neptune Database (OLTP, up to 100K queries/sec) and Neptune Analytics (in-memory analytics on large datasets)

Neptune supports three query languages:
- openCypher: Property graph queries (similar to Neo4j Cypher)
- Gremlin: Apache TinkerPop traversal language
- SPARQL: W3C standard for RDF graphs

ArangoDB

• Query language: AQL (ArangoDB Query Language, SQL-like)
• Pros: True multi-model (graph + document + key-value) in one engine, single query language for all models, flexible schema
• Cons: Smaller community than Neo4j, fewer graph-specific algorithms, BUSL-1.1 license from v3.12+ (free Community Edition includes all features but has commercial use restrictions above 100 GiB)
• Use when: Need document store and graph in one system, want to avoid managing multiple databases

AQL supports graph, document, and key-value operations:
- Graph: TRAVERSAL, SHORTEST_PATH, K_SHORTEST_PATHS
- Document: INSERT, UPDATE, REPLACE, REMOVE, UPSERT
- Full-text: ANALYZER, SEARCH with ArangoSearch

TigerGraph

• Query languages: GSQL (native), OpenCypher, ISO GQL
• Pros: Fastest for deep traversals and massive graphs, parallel processing, free Community Edition (16 CPUs, 200 GB graph + 100 GB vector storage), native hybrid vector+graph search
• Cons: Steeper learning curve, smaller ecosystem, company has undergone significant organizational changes
• Use when: Billions of edges, real-time analytics, deep multi-hop traversals, fraud detection at scale

TigerGraph query language support:
- GSQL: Native, most powerful for TigerGraph-specific features
- OpenCypher: Compatible with Neo4j-style queries
- ISO GQL: Emerging standard for graph query languages

Technology Stack by Scale

Selection Decision Matrix

Setup Checklist

1. Define scale requirements (entity count, query volume, traversal depth) 2. Evaluate infrastructure constraints (cloud provider, managed vs self-hosted) 3. Assess team expertise (Cypher, Gremlin, SQL, GSQL) 4. Test with representative queries on sample data 5. Configure constraints and indexes before bulk loading 6. Set up monitoring for query performance and memory usage

Entity Extraction & Relationship Building

Data Sources

• Structured: Databases, APIs, CSV files
• Unstructured: Documents, web content, text files
• Semi-structured: JSON, XML, knowledge bases

Entity Extraction Pipeline

class EntityExtractionPipeline:
    def __init__(self):
        self.ner_model = load_ner_model()  # spaCy, Hugging Face
        self.entity_linker = EntityLinker()
        self.deduplicator = EntityDeduplicator()

    def process_text(self, text: str) -> List[Entity]:
        # 1. Extract named entities
        entities = self.ner_model.extract(text)

        # 2. Link to existing entities (entity resolution)
        linked_entities = self.entity_linker.link(entities)

        # 3. Deduplicate and resolve conflicts
        resolved_entities = self.deduplicator.resolve(linked_entities)

        return resolved_entities

Relationship Extraction

class RelationshipExtractor:
    def extract_relationships(self, entities: List[Entity],
                            text: str) -> List[Relationship]:
        relationships = []

        # Use dependency parsing or LLM for extraction
        doc = self.nlp(text)
        for sent in doc.sents:
            rels = self.extract_from_sentence(sent, entities)
            relationships.extend(rels)

        # Validate against ontology
        valid_relationships = self.validate_relationships(relationships)
        return valid_relationships

LLM-Based Extraction (Complex Relationships)

def extract_with_llm(text: str) -> List[Relationship]:
    prompt = f"""
    Extract entities and relationships from this text:
    {text}

    Format: (Entity1, Relationship, Entity2, Confidence)
    Only extract factual relationships.
    """

    response = llm.generate(prompt)
    relationships = parse_llm_response(response)
    return relationships

Validation

• Entity extraction accuracy > 85%
• Entity deduplication working
• Relationships validated against ontology
• Confidence scores assigned

Hybrid Knowledge-Vector Architecture

Why Hybrid

Neither graph traversal nor vector search alone covers all knowledge retrieval needs:

• Vector search: Finds semantically similar content even with different wording, handles fuzzy queries
• Graph traversal: Follows explicit relationships, provides structured reasoning paths, explains connections
• Hybrid: Vector search finds entry points, graph traversal expands context through relationships

Architecture Overview

                    ┌─────────────┐
                    │   Query     │
                    └──────┬──────┘
                           │
                    ┌──────▼──────┐
                    │   Router    │
                    └──┬──────┬───┘
                       │      │
              ┌────────▼─┐  ┌─▼────────┐
              │  Vector  │  │   Graph   │
              │  Search  │  │ Traversal │
              └────┬─────┘  └─────┬─────┘
                   │              │
              ┌────▼──────────────▼────┐
              │    Merge & Re-rank     │
              └───────────┬────────────┘
                          │
                   ┌──────▼──────┐
                   │   Results   │
                   └─────────────┘

Dual Storage Implementation

class HybridKnowledgeSystem:
    def __init__(self):
        self.graph_db = Neo4jConnection()
        self.vector_db = PineconeClient()
        self.embedding_model = OpenAIEmbeddings()

    def store_entity(self, entity: Entity):
        self.graph_db.create_node(entity)

        embedding = self.embedding_model.embed(entity.description)
        self.vector_db.upsert(
            id=entity.id,
            values=embedding,
            metadata={
                "type": entity.type,
                "name": entity.name,
                "graph_id": entity.id
            }
        )

    def store_relationship(self, source_id: str, target_id: str,
                          rel_type: str, properties: dict):
        self.graph_db.create_relationship(
            source_id, target_id, rel_type, properties
        )

        rel_text = f"{properties.get('evidence', '')} {rel_type}"
        embedding = self.embedding_model.embed(rel_text)
        self.vector_db.upsert(
            id=f"rel_{source_id}_{target_id}",
            values=embedding,
            metadata={
                "type": "relationship",
                "rel_type": rel_type,
                "source_id": source_id,
                "target_id": target_id
            }
        )

Hybrid Search

class HybridSearch:
    def __init__(self, graph_db, vector_db, embedding_model):
        self.graph = graph_db
        self.vectors = vector_db
        self.embedder = embedding_model

    def search(self, query: str, top_k: int = 10,
               graph_hops: int = 2) -> list[SearchResult]:
        query_embedding = self.embedder.embed(query)
        vector_results = self.vectors.query(
            vector=query_embedding,
            top_k=top_k * 5
        )

        entity_ids = [r.metadata["graph_id"] for r in vector_results.matches
                      if r.metadata.get("type") != "relationship"]

        graph_context = self.graph.get_neighborhood(
            entity_ids[:20],
            max_hops=graph_hops
        )

        return self._merge_and_rank(
            vector_results, graph_context, query, top_k
        )

    def _merge_and_rank(self, vector_results, graph_context,
                        query: str, top_k: int) -> list[SearchResult]:
        scored_results = {}

        for result in vector_results.matches:
            scored_results[result.id] = SearchResult(
                entity_id=result.id,
                vector_score=result.score,
                graph_score=0.0,
                metadata=result.metadata
            )

        for node in graph_context.nodes:
            if node.id in scored_results:
                scored_results[node.id].graph_score = node.centrality
            else:
                scored_results[node.id] = SearchResult(
                    entity_id=node.id,
                    vector_score=0.0,
                    graph_score=node.centrality,
                    metadata=node.properties
                )

        for result in scored_results.values():
            result.combined_score = (
                0.6 * result.vector_score +
                0.4 * result.graph_score
            )

        ranked = sorted(
            scored_results.values(),
            key=lambda r: r.combined_score,
            reverse=True
        )
        return ranked[:top_k]

Ingestion Pipeline

class HybridIngestionPipeline:
    def __init__(self, hybrid_system: HybridKnowledgeSystem,
                 entity_extractor, relationship_extractor):
        self.system = hybrid_system
        self.entity_extractor = entity_extractor
        self.relationship_extractor = relationship_extractor

    def ingest_document(self, document: Document):
        entities = self.entity_extractor.extract(document.text)

        resolved_entities = self._resolve_entities(entities)

        for entity in resolved_entities:
            self.system.store_entity(entity)

        relationships = self.relationship_extractor.extract(
            resolved_entities, document.text
        )

        for rel in relationships:
            self.system.store_relationship(
                source_id=rel.source_id,
                target_id=rel.target_id,
                rel_type=rel.type,
                properties={
                    "confidence": rel.confidence,
                    "source": document.id,
                    "evidence": rel.evidence
                }
            )

    def _resolve_entities(self, entities: list[Entity]) -> list[Entity]:
        resolved = []
        for entity in entities:
            existing = self.system.graph_db.find_by_alias(
                entity.name, entity.type
            )
            if existing:
                existing.aliases.append(entity.name)
                resolved.append(existing)
            else:
                resolved.append(entity)
        return resolved

Query Routing Strategy

Route queries to the appropriate search mode based on query characteristics:

class QueryRouter:
    def route(self, query: str) -> str:
        if self._has_relationship_pattern(query):
            return "graph_first"

        if self._is_semantic_search(query):
            return "vector_first"

        return "hybrid"

    def _has_relationship_pattern(self, query: str) -> bool:
        patterns = [
            "how are .* related",
            "connected to",
            "path between",
            "reports to",
            "works for"
        ]
        return any(re.search(p, query, re.IGNORECASE) for p in patterns)

    def _is_semantic_search(self, query: str) -> bool:
        return len(query.split()) > 10 or "similar to" in query.lower()

Consistency Considerations

• Keep graph and vector stores in sync during writes (store to both atomically or use eventual consistency with reconciliation)
• Embed entity descriptions, not just names, for better semantic matching
• Include relationship evidence text in vector embeddings for relationship search
• Re-embed entities when descriptions change significantly
• Monitor vector-graph ID mapping for orphaned entries

Ontology Design

Core Concepts

An ontology defines the vocabulary and structure of a knowledge graph: what types of entities exist, how they relate, and what properties they carry. Design the ontology before ingesting data -- changing it later requires re-processing all entities and relationships.

Entity Types (Nodes)

Common entity types and their typical properties:

Person       — id, name, title, email, organization
Organization — id, name, type (company/nonprofit/government), industry, founded
Location     — id, name, coordinates, type (city/country/building)
Product      — id, name, version, category, status
Concept      — id, name, definition, domain
Event        — id, name, date, location, participants
Document     — id, title, author, date, content_hash, source_url

Define entity types based on domain requirements. Start with 3-5 core types and expand as needed.

Relationship Types (Edges)

Organize relationships by semantic category:

• Hierarchical: IS_A, PART_OF, REPORTS_TO, BELONGS_TO, SUBCATEGORY_OF
• Associative: WORKS_FOR, LOCATED_IN, AUTHORED_BY, RELATED_TO, COLLABORATES_WITH
• Temporal: CREATED_ON, OCCURRED_BEFORE, OCCURRED_AFTER, STARTED_AT, ENDED_AT
• Causal: CAUSED_BY, LEADS_TO, DEPENDS_ON, ENABLES

Every relationship should have a clear semantic meaning. Avoid generic "RELATED_TO" when a more specific type exists.

Properties (Attributes)

Node Properties

class EntityProperties:
    id: str              # Unique identifier (UUID or domain-specific)
    name: str            # Human-readable label
    type: str            # Entity type from ontology
    aliases: list[str]   # Alternative names for entity resolution
    created_at: datetime # When entity was added to graph
    source: str          # Where entity was extracted from
    confidence: float    # Extraction confidence (0.0-1.0)
    metadata: dict       # Domain-specific additional properties

Edge Properties

class RelationshipProperties:
    type: str            # Relationship type from ontology
    confidence: float    # Confidence score (0.0-1.0)
    source: str          # Document or system that produced this relationship
    extracted_at: datetime
    evidence: str        # Text snippet supporting the relationship
    weight: float        # Strength or importance (optional)

RDF/Turtle Schema Example

@prefix : <http://example.org/ontology#> .
@prefix owl: <http://www.w3.org/2002/07/owl#> .
@prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#> .
@prefix xsd: <http://www.w3.org/2001/XMLSchema#> .

# Entity classes
:Person a owl:Class ;
    rdfs:label "Person" ;
    rdfs:comment "A human individual" .

:Organization a owl:Class ;
    rdfs:label "Organization" ;
    rdfs:comment "A company, institution, or group" .

:Location a owl:Class ;
    rdfs:label "Location" ;
    rdfs:comment "A geographic place or address" .

# Relationship properties
:worksFor a owl:ObjectProperty ;
    rdfs:domain :Person ;
    rdfs:range :Organization ;
    rdfs:label "works for" .

:locatedIn a owl:ObjectProperty ;
    rdfs:domain :Organization ;
    rdfs:range :Location ;
    rdfs:label "located in" .

# Data properties
:hasName a owl:DatatypeProperty ;
    rdfs:domain owl:Thing ;
    rdfs:range xsd:string .

:hasConfidence a owl:DatatypeProperty ;
    rdfs:range xsd:float ;
    rdfs:comment "Confidence score between 0.0 and 1.0" .

Neo4j Property Graph Schema

// Define constraints
CREATE CONSTRAINT entity_id IF NOT EXISTS
FOR (e:Entity) REQUIRE e.id IS UNIQUE;

CREATE CONSTRAINT person_name IF NOT EXISTS
FOR (p:Person) REQUIRE p.name IS NOT NULL;

// Define indexes for common queries
CREATE INDEX entity_type IF NOT EXISTS
FOR (e:Entity) ON (e.type);

CREATE INDEX entity_name_search IF NOT EXISTS
FOR (e:Entity) ON (e.name);

// Example node creation with properties
CREATE (p:Person:Entity {
  id: randomUUID(),
  name: 'Jane Smith',
  type: 'Person',
  aliases: ['J. Smith', 'Jane R. Smith'],
  confidence: 0.95,
  source: 'hr_database'
})

// Example relationship with properties
MATCH (p:Person {name: 'Jane Smith'})
MATCH (o:Organization {name: 'Acme Corp'})
CREATE (p)-[:WORKS_FOR {
  confidence: 0.92,
  source: 'linkedin_scrape',
  since: date('2022-03-15'),
  evidence: 'Senior Engineer at Acme Corp'
}]->(o)

Design Patterns

Reification (Relationships with Rich Metadata)

When a relationship needs many properties, model it as an intermediate node:

// Instead of a simple edge with many properties:
// (person)-[:EMPLOYED_AT {title, start, end, salary}]->(org)

// Reify into an Employment node:
CREATE (e:Employment {
  title: 'Senior Engineer',
  start_date: date('2022-03-15'),
  end_date: null,
  department: 'Engineering'
})

MATCH (p:Person {name: 'Jane Smith'})
MATCH (o:Organization {name: 'Acme Corp'})
CREATE (p)-[:HAS_EMPLOYMENT]->(e)-[:AT_ORGANIZATION]->(o)

Temporal Versioning

Track how relationships change over time:

// Add valid_from and valid_to on relationships
MATCH (p:Person {name: 'Jane Smith'})
MATCH (o1:Organization {name: 'OldCorp'})
MATCH (o2:Organization {name: 'NewCorp'})
CREATE (p)-[:WORKS_FOR {valid_from: date('2019-01-01'), valid_to: date('2022-03-01')}]->(o1)
CREATE (p)-[:WORKS_FOR {valid_from: date('2022-03-15'), valid_to: null}]->(o2)

Validation Checklist

• Entity types cover all domain concepts identified in requirements
• Relationship types capture key connections with clear semantics
• Every relationship has a defined domain (source type) and range (target type)
• Properties include confidence scores and source provenance
• Ontology reviewed and validated with domain experts
• Classification hierarchy defined (IS_A relationships)
• No generic "RELATED_TO" where a specific relationship type exists
• Entity resolution aliases defined for common name variations

Query Patterns & API Design

Common Cypher Patterns

Find Entity

MATCH (e:Entity {id: $entity_id})
RETURN e

Find Relationships

MATCH (source:Entity {id: $entity_id})-[r]-(target)
RETURN source, r, target
LIMIT 20

Shortest Path Between Entities

MATCH path = shortestPath(
  (source:Person {id: $source_id})-[*..5]-(target:Person {id: $target_id})
)
RETURN path

Multi-Hop Traversal

MATCH (p:Person {name: $name})-[:WORKS_FOR]->(o:Organization)-[:LOCATED_IN]->(l:Location)
RETURN p.name, o.name, l.city

Recommendation Query

// Find people similar to this person based on shared organizations
MATCH (p1:Person {id: $person_id})-[:WORKS_FOR]->(o:Organization)<-[:WORKS_FOR]-(p2:Person)
WHERE p1 <> p2
RETURN p2, COUNT(o) AS shared_orgs
ORDER BY shared_orgs DESC
LIMIT 10

Knowledge Graph API

class KnowledgeGraphAPI:
    def __init__(self, graph_db):
        self.graph = graph_db

    def find_entity(self, entity_name: str) -> Entity:
        """Find entity by name with fuzzy matching"""
        query = """
        MATCH (e:Entity)
        WHERE e.name CONTAINS $name
        RETURN e
        ORDER BY apoc.text.levenshtein(e.name, $name)
        LIMIT 1
        """
        return self.graph.run(query, name=entity_name).single()

    def find_relationships(self, entity_id: str,
                         relationship_type: str = None,
                         max_hops: int = 2) -> List[Relationship]:
        """Find relationships within specified hops"""
        query = f"""
        MATCH (source:Entity {{id: $entity_id}})
        MATCH path = (source)-[r*1..{max_hops}]-(target)
        RETURN path, relationships(path) AS rels
        LIMIT 100
        """
        return self.graph.run(query, entity_id=entity_id).data()

    def get_subgraph(self, entity_ids: List[str],
                    max_hops: int = 2) -> Subgraph:
        """Get connected subgraph for multiple entities"""
        query = f"""
        MATCH (e:Entity)
        WHERE e.id IN $entity_ids
        CALL apoc.path.subgraphAll(e, {{maxLevel: {max_hops}}})
        YIELD nodes, relationships
        RETURN nodes, relationships
        """
        return self.graph.run(query, entity_ids=entity_ids).data()