Semantic Memory and Knowledge Graphs

Reading time: 33 min  |  Level: Intermediate–Advanced  |  Relevance: AI Engineer, ML Engineer, Knowledge Systems Engineer

The Factual Knowledge Problem

A customer support agent should know that the enterprise plan includes SSO. A research agent should know BERT was published in 2018. A coding agent should know that Python's GIL prevents true thread parallelism for CPU-bound tasks.

These are not personal experiences. They are not things anyone observed in a specific interaction. They are facts - decontextualized, structured, generalizable knowledge that the agent should have available regardless of who is asking or what session it is in.

This is semantic memory. And the way you structure it determines how precisely and reliably your agent can reason about it.

Three options exist. Plain text RAG: chunk a bunch of documents, embed them, retrieve the relevant chunks. Works but imprecise - you might get the chunk containing "SSO is included" but not the chunk that says "SSO requires SAML 2.0 configuration on the customer's identity provider." Knowledge graphs: entities connected by typed relationships, traversable, precise - "Enterprise plan INCLUDES SSO REQUIRES SAML_2.0 CONFIGURED_VIA identity_provider." Databases: structured and precise but rigid schema - hard to represent the messy, varied relationships in real domain knowledge.

Production agents increasingly use all three in combination. This lesson covers knowledge graphs in depth: why they matter, how to build them from unstructured text using LLMs, how to query them, and how to combine them with vector search for hybrid retrieval.

---

:::tip 🎮 Interactive Playground Visualize this concept: Try the Graph RAG demo on the EngineersOfAI Playground - no code required. :::

Semantic vs Episodic Memory: The Distinction

Endel Tulving's 1972 distinction remains the clearest way to draw the line:

Semantic memory (this lesson):

• "Python was created by Guido van Rossum"
• "Enterprise plan costs $500/month"
• "BERT uses bidirectional transformers"
• Could appear in a textbook or documentation
• No personal attribution required

Episodic memory (previous lesson):

• "Sarah told me she prefers Python"
• "The deployment failed last Tuesday due to a missing env var"
• "During our last session, we agreed to use Pydantic V2"
• Tied to a specific time, person, and context

The engineering implication: semantic memory is shared across users and sessions. Episodic memory is per-user. Semantic facts are stable - they change infrequently. Episodic memories are volatile - they grow with every interaction.

---

Why Knowledge Graphs Beat Plain RAG for Structured Knowledge

Plain text RAG is fast to build and works well for unstructured document retrieval. But it has fundamental weaknesses for structured factual knowledge:

Chunk boundary problem: "The enterprise plan includes SSO, advanced reporting, and dedicated support. SSO requires SAML 2.0 configuration." If these sentences appear in different chunks, a query about SSO requirements may retrieve only one, missing the dependency.

Relationship blindness: RAG retrieves text chunks, not relationships. It cannot answer "what are all the features included in the enterprise plan?" by traversing an explicit inclusion relationship - it must hope all features appear in a single retrievable chunk.

Multi-hop reasoning: "Which plans support the MFA requirement for SOC 2 compliance?" requires knowing: SOC 2 requires MFA → MFA is a feature → Enterprise plan includes MFA. RAG struggles with this chain. A knowledge graph traverses it directly.

Update imprecision: Updating a fact in a vector store requires finding and re-embedding the relevant chunks - error-prone. Updating a graph node or edge is a precise, targeted operation.

Knowledge graphs solve all four problems by making relationships explicit, first-class objects.

---

Knowledge Graph Fundamentals

A knowledge graph is a directed property graph:

• Nodes (entities): represent real-world things - people, products, concepts, events
• Edges (relationships): typed connections between nodes - "INCLUDES", "REQUIRES", "CREATED_BY"
• Properties: metadata on nodes and edges - timestamps, confidence, source

Example: Product Knowledge Graph

[Enterprise Plan] --INCLUDES--> [SSO Feature]
[Enterprise Plan] --INCLUDES--> [Advanced Reporting]
[Enterprise Plan] --COSTS--> [$500/month]
[SSO Feature] --REQUIRES--> [SAML 2.0]
[SAML 2.0] --CONFIGURED_VIA--> [Identity Provider]
[Identity Provider] --EXAMPLES--> [Okta, Azure AD, Google Workspace]

This structure supports queries impossible with flat text:

• "What does Enterprise include?" → traverse all INCLUDES edges from Enterprise
• "What are SSO requirements?" → traverse REQUIRES edges from SSO
• "Can I use Okta?" → traverse from Okta backward through CONFIGURED_VIA → SAML 2.0 → REQUIRES → SSO → INCLUDES → Enterprise

---

Building a Knowledge Graph with LLMs

The hardest part of knowledge graphs used to be construction - entity extraction, relationship extraction, coreference resolution. LLMs have made this dramatically easier.

---

Full Implementation: Knowledge Graph Memory System

"""
Knowledge Graph Memory System for AI Agents.

Stack:
  - NetworkX: in-memory graph (production: replace with Neo4j or Kuzu)
  - Anthropic Claude: entity/relationship extraction
  - Simple vector index: node descriptions for semantic search

Install: pip install networkx anthropic
"""

from __future__ import annotations
import json
import time
import uuid
from dataclasses import dataclass, field
from typing import Optional, Any
from collections import defaultdict

import networkx as nx
import anthropic


# ─────────────────────────────────────────────
# SCHEMA
# ─────────────────────────────────────────────

@dataclass
class Entity:
    """A node in the knowledge graph."""
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    name: str = ""
    entity_type: str = ""            # person | organization | product | concept | event | location
    description: str = ""
    aliases: list[str] = field(default_factory=list)
    confidence: float = 1.0
    source: str = "manual"
    created_at: float = field(default_factory=time.time)
    last_updated: float = field(default_factory=time.time)
    properties: dict[str, Any] = field(default_factory=dict)

    def canonical_name(self) -> str:
        """Normalized name for deduplication."""
        return self.name.lower().strip()


@dataclass
class Relationship:
    """A typed, directed edge in the knowledge graph."""
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    source_id: str = ""              # Source entity ID
    target_id: str = ""              # Target entity ID
    relation_type: str = ""          # INCLUDES | REQUIRES | CREATED_BY | etc.
    description: str = ""            # Human-readable description
    confidence: float = 1.0
    bidirectional: bool = False      # If True, edge goes both ways
    source: str = "manual"
    created_at: float = field(default_factory=time.time)
    properties: dict[str, Any] = field(default_factory=dict)


@dataclass
class ExtractionResult:
    """Output of LLM-based entity/relationship extraction."""
    entities: list[dict] = field(default_factory=list)
    relationships: list[dict] = field(default_factory=list)
    source_text: str = ""


# ─────────────────────────────────────────────
# LLM EXTRACTOR
# ─────────────────────────────────────────────

class KnowledgeExtractor:
    """
    Uses Claude to extract entities and relationships from text.
    """

    EXTRACTION_PROMPT = """Extract entities and relationships from the text below.

Return a JSON object with exactly this structure:
{
  "entities": [
    {
      "name": "entity name",
      "type": "person|organization|product|concept|feature|location|event",
      "description": "brief description",
      "aliases": ["alternative names if any"]
    }
  ],
  "relationships": [
    {
      "source": "entity name",
      "relation": "RELATION_TYPE",
      "target": "entity name",
      "description": "human readable description of this relationship"
    }
  ]
}

Use UPPERCASE_SNAKE_CASE for relation types. Use specific, meaningful relation types.
Common types: INCLUDES, REQUIRES, CREATED_BY, PART_OF, COSTS, COMPATIBLE_WITH, DEPENDS_ON,
WORKS_AT, PUBLISHED_IN, USES, SUPPORTS, REPLACES, CONFLICTS_WITH.

Return ONLY valid JSON. No explanation."""

    def __init__(self):
        self.client = anthropic.Anthropic()

    def extract(self, text: str) -> ExtractionResult:
        """Extract entities and relationships from text."""
        try:
            response = self.client.messages.create(
                model="claude-opus-4-6",
                max_tokens=1500,
                system=self.EXTRACTION_PROMPT,
                messages=[{"role": "user", "content": text}],
            )
            raw = response.content[0].text.strip()

            # Handle markdown code blocks if model wraps JSON
            if raw.startswith("```"):
                lines = raw.split("\n")
                raw = "\n".join(lines[1:-1])

            data = json.loads(raw)
            return ExtractionResult(
                entities=data.get("entities", []),
                relationships=data.get("relationships", []),
                source_text=text,
            )
        except (json.JSONDecodeError, KeyError, Exception) as e:
            print(f"[KnowledgeExtractor] Extraction failed: {e}")
            return ExtractionResult(source_text=text)


# ─────────────────────────────────────────────
# KNOWLEDGE GRAPH
# ─────────────────────────────────────────────

class KnowledgeGraph:
    """
    In-memory knowledge graph using NetworkX.

    Production replacement: Neo4j (pip install neo4j) or Kuzu (pip install kuzu)
    Both support Cypher query language and scale to billions of nodes/edges.
    """

    def __init__(self):
        # Directed multi-graph (allows multiple edges between same node pair)
        self.graph = nx.MultiDiGraph()
        self.entities: dict[str, Entity] = {}           # id → Entity
        self.name_index: dict[str, str] = {}            # canonical_name → id
        self.alias_index: dict[str, str] = {}           # alias → id
        self.relationships: dict[str, Relationship] = {}

        self.extractor = KnowledgeExtractor()

    # ─── ENTITY MANAGEMENT ──────────────────────────────────────

    def add_entity(self, entity: Entity) -> Entity:
        """Add or update an entity. Returns the stored entity (may be existing)."""
        canonical = entity.canonical_name()

        # Check if entity already exists (by name or alias)
        existing_id = self.name_index.get(canonical)
        if not existing_id:
            for alias in entity.aliases:
                existing_id = self.alias_index.get(alias.lower())
                if existing_id:
                    break

        if existing_id:
            # Update existing entity
            existing = self.entities[existing_id]
            existing.description = entity.description or existing.description
            existing.last_updated = time.time()
            # Merge aliases
            for alias in entity.aliases:
                if alias not in existing.aliases:
                    existing.aliases.append(alias)
                    self.alias_index[alias.lower()] = existing_id
            existing.confidence = max(existing.confidence, entity.confidence)
            return existing

        # New entity
        self.entities[entity.id] = entity
        self.name_index[canonical] = entity.id
        for alias in entity.aliases:
            self.alias_index[alias.lower()] = entity.id

        # Add to NetworkX graph
        self.graph.add_node(
            entity.id,
            name=entity.name,
            entity_type=entity.entity_type,
            description=entity.description,
        )

        return entity

    def find_entity(self, name: str) -> Optional[Entity]:
        """Look up an entity by name or alias."""
        canonical = name.lower().strip()
        entity_id = self.name_index.get(canonical) or self.alias_index.get(canonical)
        if entity_id:
            return self.entities[entity_id]

        # Fuzzy fallback: check if query is contained in any entity name
        for stored_name, eid in self.name_index.items():
            if canonical in stored_name or stored_name in canonical:
                return self.entities[eid]

        return None

    def add_entity_simple(
        self,
        name: str,
        entity_type: str = "concept",
        description: str = "",
        source: str = "manual",
    ) -> Entity:
        """Convenience method: create and add entity from basic params."""
        entity = Entity(
            name=name,
            entity_type=entity_type,
            description=description,
            source=source,
        )
        return self.add_entity(entity)

    # ─── RELATIONSHIP MANAGEMENT ──────────────────────────────────

    def add_relationship(
        self,
        source_name: str,
        relation_type: str,
        target_name: str,
        description: str = "",
        confidence: float = 1.0,
        bidirectional: bool = False,
        source: str = "manual",
    ) -> Optional[Relationship]:
        """
        Add a relationship between two named entities.
        Auto-creates entities if they do not exist.
        """
        # Ensure both entities exist
        source_entity = self.find_entity(source_name)
        if not source_entity:
            source_entity = self.add_entity_simple(source_name)

        target_entity = self.find_entity(target_name)
        if not target_entity:
            target_entity = self.add_entity_simple(target_name)

        rel = Relationship(
            source_id=source_entity.id,
            target_id=target_entity.id,
            relation_type=relation_type,
            description=description or f"{source_name} {relation_type} {target_name}",
            confidence=confidence,
            bidirectional=bidirectional,
            source=source,
        )
        self.relationships[rel.id] = rel

        # Add to NetworkX graph
        self.graph.add_edge(
            source_entity.id,
            target_entity.id,
            key=rel.id,
            relation_type=relation_type,
            description=rel.description,
            confidence=confidence,
        )

        if bidirectional:
            self.graph.add_edge(
                target_entity.id,
                source_entity.id,
                key=f"{rel.id}_rev",
                relation_type=f"INVERSE_{relation_type}",
                description=f"{target_name} {relation_type} {source_name}",
                confidence=confidence,
            )

        return rel

    # ─── QUERY PATTERNS ───────────────────────────────────────────

    def get_entity_facts(
        self,
        entity_name: str,
        relation_types: list[str] | None = None,
        depth: int = 1,
    ) -> list[str]:
        """
        Get all facts about an entity: its direct relationships and their targets.

        Args:
            entity_name: Name of the entity to query
            relation_types: If specified, filter to only these relationship types
            depth: How many hops to traverse (1 = direct neighbors only)
        """
        entity = self.find_entity(entity_name)
        if not entity:
            return []

        facts = []
        visited = {entity.id}
        queue = [(entity.id, 0)]

        while queue:
            current_id, current_depth = queue.pop(0)
            if current_depth >= depth:
                continue

            # Get all outgoing edges from current entity
            for _, target_id, edge_data in self.graph.out_edges(current_id, data=True):
                rel_type = edge_data.get("relation_type", "RELATED_TO")

                # Apply relation type filter
                if relation_types and rel_type not in relation_types:
                    continue

                target_entity = self.entities.get(target_id)
                if not target_entity:
                    continue

                source_entity = self.entities.get(current_id)
                fact = f"{source_entity.name} {rel_type} {target_entity.name}"
                if edge_data.get("description"):
                    fact = edge_data["description"]
                facts.append(fact)

                # Queue for deeper traversal if needed
                if target_id not in visited and current_depth + 1 < depth:
                    visited.add(target_id)
                    queue.append((target_id, current_depth + 1))

        return facts

    def find_path(
        self,
        source_name: str,
        target_name: str,
    ) -> Optional[list[str]]:
        """
        Find relationship path between two entities (multi-hop reasoning).
        Returns list of edge descriptions along the path.
        """
        source = self.find_entity(source_name)
        target = self.find_entity(target_name)
        if not source or not target:
            return None

        try:
            path_nodes = nx.shortest_path(
                self.graph,
                source.id,
                target.id,
            )
        except nx.NetworkXNoPath:
            return None

        # Reconstruct path as readable facts
        path_facts = []
        for i in range(len(path_nodes) - 1):
            src_id = path_nodes[i]
            tgt_id = path_nodes[i + 1]
            edge_data_dict = self.graph.get_edge_data(src_id, tgt_id)
            if edge_data_dict:
                # MultiDiGraph returns dict of {key: edge_data}
                first_edge = next(iter(edge_data_dict.values()))
                desc = first_edge.get(
                    "description",
                    f"{self.entities[src_id].name} → {self.entities[tgt_id].name}"
                )
                path_facts.append(desc)

        return path_facts

    def search_entities(self, query: str, top_k: int = 5) -> list[Entity]:
        """
        Keyword search over entity names and descriptions.
        Production: replace with vector similarity search over entity embeddings.
        """
        query_words = set(query.lower().split())
        scored = []

        for entity in self.entities.values():
            # Score by keyword overlap with name and description
            entity_words = set(
                (entity.name + " " + entity.description).lower().split()
            )
            overlap = len(query_words & entity_words)
            if overlap > 0:
                # Boost for name matches over description matches
                name_words = set(entity.name.lower().split())
                name_overlap = len(query_words & name_words)
                score = overlap + name_overlap * 2  # Name matches count double
                scored.append((score, entity))

        scored.sort(key=lambda x: x[0], reverse=True)
        return [e for _, e in scored[:top_k]]

    def get_neighbors(
        self,
        entity_name: str,
        direction: str = "out",   # "out" | "in" | "both"
    ) -> list[tuple[str, str, str]]:
        """
        Get neighbors of an entity.
        Returns list of (relation_type, neighbor_name, description).
        """
        entity = self.find_entity(entity_name)
        if not entity:
            return []

        results = []

        if direction in ("out", "both"):
            for _, target_id, edge_data in self.graph.out_edges(entity.id, data=True):
                target = self.entities.get(target_id)
                if target:
                    results.append((
                        edge_data.get("relation_type", "RELATED_TO"),
                        target.name,
                        edge_data.get("description", ""),
                    ))

        if direction in ("in", "both"):
            for source_id, _, edge_data in self.graph.in_edges(entity.id, data=True):
                source = self.entities.get(source_id)
                if source:
                    results.append((
                        f"←{edge_data.get('relation_type', 'RELATED_TO')}",
                        source.name,
                        edge_data.get("description", ""),
                    ))

        return results

    # ─── INGESTION FROM TEXT ──────────────────────────────────────

    def ingest_text(self, text: str, source: str = "document") -> tuple[int, int]:
        """
        Extract entities and relationships from text and add to graph.
        Returns (entities_added, relationships_added).
        """
        result = self.extractor.extract(text)

        entities_added = 0
        name_to_entity: dict[str, Entity] = {}

        # Add entities first
        for ent_data in result.entities:
            entity = Entity(
                name=ent_data.get("name", ""),
                entity_type=ent_data.get("type", "concept"),
                description=ent_data.get("description", ""),
                aliases=ent_data.get("aliases", []),
                source=source,
                confidence=0.85,  # LLM-extracted, not human-verified
            )
            if entity.name:
                stored = self.add_entity(entity)
                name_to_entity[entity.name.lower()] = stored
                entities_added += 1

        # Add relationships
        relationships_added = 0
        for rel_data in result.relationships:
            src_name = rel_data.get("source", "")
            tgt_name = rel_data.get("target", "")
            rel_type = rel_data.get("relation", "RELATED_TO")
            desc = rel_data.get("description", "")

            if src_name and tgt_name and rel_type:
                rel = self.add_relationship(
                    source_name=src_name,
                    relation_type=rel_type,
                    target_name=tgt_name,
                    description=desc,
                    confidence=0.80,
                    source=source,
                )
                if rel:
                    relationships_added += 1

        return entities_added, relationships_added

    # ─── CONTEXT FORMATTING ───────────────────────────────────────

    def format_for_context(
        self,
        query: str,
        max_chars: int = 2500,
    ) -> str:
        """
        Format relevant knowledge graph facts for injection into agent context.
        """
        # Find relevant entities
        relevant_entities = self.search_entities(query, top_k=4)
        if not relevant_entities:
            return ""

        parts = ["## Relevant Knowledge Graph Facts"]
        char_budget = max_chars - len(parts[0])

        for entity in relevant_entities:
            # Get this entity's direct facts
            entity_facts = self.get_entity_facts(entity.name, depth=2)

            if not entity_facts:
                continue

            block = f"\n{entity.name}:\n" + "\n".join(f"  - {f}" for f in entity_facts[:5])
            if char_budget - len(block) < 0:
                break
            parts.append(block)
            char_budget -= len(block)

        return "\n".join(parts)

    def stats(self) -> dict:
        return {
            "entities": len(self.entities),
            "relationships": len(self.relationships),
            "graph_nodes": self.graph.number_of_nodes(),
            "graph_edges": self.graph.number_of_edges(),
        }


# ─────────────────────────────────────────────
# KNOWLEDGE GRAPH AGENT
# ─────────────────────────────────────────────

class KnowledgeGraphAgent:
    """
    Agent with semantic memory backed by a knowledge graph.
    Answers questions by querying the KG and injecting facts into context.
    """

    MODEL = "claude-opus-4-6"

    def __init__(self):
        self.client = anthropic.Anthropic()
        self.kg = KnowledgeGraph()
        self.conversation: list[dict] = []
        self._seed_domain_knowledge()

    def _seed_domain_knowledge(self) -> None:
        """Pre-populate the knowledge graph with domain facts."""

        # ── Product knowledge ─────────────────────────────────
        self.kg.add_entity_simple("Starter Plan", "product", "Basic plan for small teams")
        self.kg.add_entity_simple("Professional Plan", "product", "Mid-tier plan with advanced features")
        self.kg.add_entity_simple("Enterprise Plan", "product", "Full-featured plan for large organizations")
        self.kg.add_entity_simple("SSO", "feature", "Single Sign-On authentication")
        self.kg.add_entity_simple("SAML 2.0", "concept", "Security Assertion Markup Language protocol version 2")
        self.kg.add_entity_simple("Advanced Analytics", "feature", "In-depth data analytics and reporting")
        self.kg.add_entity_simple("API Access", "feature", "Programmatic access via REST API")
        self.kg.add_entity_simple("Dedicated Support", "feature", "24/7 dedicated account manager")
        self.kg.add_entity_simple("Okta", "organization", "Enterprise identity provider")
        self.kg.add_entity_simple("Azure AD", "organization", "Microsoft's enterprise identity service")

        # Plan → feature relationships
        self.kg.add_relationship("Starter Plan", "INCLUDES", "API Access", "Starter plan provides REST API access")
        self.kg.add_relationship("Professional Plan", "INCLUDES", "API Access")
        self.kg.add_relationship("Professional Plan", "INCLUDES", "Advanced Analytics")
        self.kg.add_relationship("Enterprise Plan", "INCLUDES", "API Access")
        self.kg.add_relationship("Enterprise Plan", "INCLUDES", "Advanced Analytics")
        self.kg.add_relationship("Enterprise Plan", "INCLUDES", "SSO", "Enterprise plan includes SSO at no extra cost")
        self.kg.add_relationship("Enterprise Plan", "INCLUDES", "Dedicated Support")
        self.kg.add_relationship("SSO", "USES_PROTOCOL", "SAML 2.0", "SSO authentication uses SAML 2.0 protocol")
        self.kg.add_relationship("SSO", "COMPATIBLE_WITH", "Okta", "SSO integrates with Okta IdP")
        self.kg.add_relationship("SSO", "COMPATIBLE_WITH", "Azure AD", "SSO integrates with Azure Active Directory")

        # Pricing
        self.kg.add_relationship("Starter Plan", "COSTS", "$49/month")
        self.kg.add_relationship("Professional Plan", "COSTS", "$199/month")
        self.kg.add_relationship("Enterprise Plan", "COSTS", "Custom pricing - contact sales")

        # ── Technology knowledge ──────────────────────────────
        self.kg.add_entity_simple("Python", "concept", "High-level programming language")
        self.kg.add_entity_simple("GIL", "concept", "Global Interpreter Lock - Python's thread safety mechanism")
        self.kg.add_entity_simple("asyncio", "concept", "Python's built-in async I/O library")
        self.kg.add_entity_simple("FastAPI", "concept", "Modern Python web framework for APIs")
        self.kg.add_entity_simple("Pydantic", "concept", "Python data validation library")

        self.kg.add_relationship("Python", "HAS_LIMITATION", "GIL", "Python's GIL prevents true thread parallelism for CPU-bound tasks")
        self.kg.add_relationship("GIL", "AFFECTS", "CPU-bound threads", "CPU-bound threads are serialized by the GIL")
        self.kg.add_relationship("asyncio", "BYPASSES", "GIL", "asyncio uses cooperative multitasking, not threads - unaffected by GIL")
        self.kg.add_relationship("FastAPI", "BUILT_ON", "Pydantic", "FastAPI uses Pydantic for request/response validation")
        self.kg.add_relationship("FastAPI", "USES", "asyncio", "FastAPI is async-first using asyncio")

        # ── AI/ML research knowledge ──────────────────────────
        self.kg.add_entity_simple("BERT", "concept", "Bidirectional Encoder Representations from Transformers")
        self.kg.add_entity_simple("GPT-3", "concept", "Generative Pre-trained Transformer 3 by OpenAI")
        self.kg.add_entity_simple("Transformer", "concept", "Attention-based neural architecture")
        self.kg.add_entity_simple("Google", "organization", "Technology company")
        self.kg.add_entity_simple("OpenAI", "organization", "AI research company")
        self.kg.add_entity_simple("NeurIPS", "event", "Conference on Neural Information Processing Systems")

        self.kg.add_relationship("BERT", "CREATED_BY", "Google", "BERT was developed by Google Research")
        self.kg.add_relationship("BERT", "PUBLISHED_IN", "2018", "BERT paper published at NAACL 2019, arXiv 2018")
        self.kg.add_relationship("BERT", "USES", "Transformer", "BERT uses bidirectional transformer encoder")
        self.kg.add_relationship("GPT-3", "CREATED_BY", "OpenAI")
        self.kg.add_relationship("GPT-3", "PUBLISHED_IN", "2020", "GPT-3 paper: Brown et al., 2020")
        self.kg.add_relationship("GPT-3", "USES", "Transformer", "GPT-3 uses unidirectional (autoregressive) transformer")
        self.kg.add_relationship("BERT", "DIFFERS_FROM", "GPT-3", "BERT is bidirectional encoder; GPT-3 is autoregressive decoder")

    def ingest_document(self, text: str, source: str = "document") -> None:
        """Add knowledge from a document to the graph."""
        entities_added, rels_added = self.kg.ingest_text(text, source=source)
        print(f"[KG] Ingested: +{entities_added} entities, +{rels_added} relationships")

    def answer(self, question: str) -> str:
        """
        Answer a question using knowledge graph retrieval + LLM reasoning.
        """
        # 1. Search for relevant entities
        relevant_entities = self.kg.search_entities(question, top_k=3)

        # 2. Get facts for each relevant entity
        all_facts = []
        for entity in relevant_entities:
            facts = self.kg.get_entity_facts(entity.name, depth=2)
            all_facts.extend(facts[:8])  # Cap per entity

        # 3. Try multi-hop path finding for comparison/relationship questions
        entities_mentioned = [e.name for e in relevant_entities]
        if len(entities_mentioned) >= 2:
            path = self.kg.find_path(entities_mentioned[0], entities_mentioned[1])
            if path:
                all_facts.extend(path)

        # 4. Build system prompt with KG facts
        system = "You are a precise AI assistant with access to a knowledge graph.\n"
        if all_facts:
            facts_text = "\n".join(f"- {f}" for f in all_facts[:20])
            system += f"\n## Knowledge Graph Facts\n{facts_text}\n\n"
            system += "Use these facts to answer the question. Cite specific facts when possible."
        else:
            system += "Answer based on your training knowledge. The knowledge graph has no relevant facts for this query."

        self.conversation.append({"role": "user", "content": question})
        response = self.client.messages.create(
            model=self.MODEL,
            max_tokens=1024,
            system=system,
            messages=self.conversation,
        )
        answer_text = response.content[0].text
        self.conversation.append({"role": "assistant", "content": answer_text})
        return answer_text

    def explain_entity(self, entity_name: str) -> str:
        """Get a structured explanation of an entity and its relationships."""
        entity = self.kg.find_entity(entity_name)
        if not entity:
            return f"No entity found for '{entity_name}'"

        outgoing = self.kg.get_neighbors(entity_name, direction="out")
        incoming = self.kg.get_neighbors(entity_name, direction="in")

        lines = [
            f"Entity: {entity.name}",
            f"Type: {entity.entity_type}",
            f"Description: {entity.description or 'N/A'}",
        ]
        if outgoing:
            lines.append(f"\nOutgoing relationships ({len(outgoing)}):")
            for rel_type, neighbor, desc in outgoing[:8]:
                lines.append(f"  [{rel_type}] → {neighbor}")
        if incoming:
            lines.append(f"\nIncoming relationships ({len(incoming)}):")
            for rel_type, neighbor, desc in incoming[:8]:
                lines.append(f"  {neighbor} [{rel_type}] →")

        return "\n".join(lines)


# ─────────────────────────────────────────────
# DEMONSTRATION
# ─────────────────────────────────────────────

def demo():
    print("=" * 60)
    print("KNOWLEDGE GRAPH AGENT - DEMONSTRATION")
    print("=" * 60)

    agent = KnowledgeGraphAgent()

    print(f"\nKnowledge graph initialized: {agent.kg.stats()}")

    # ── Ingest additional knowledge from text ─────────────────────
    print("\n[Ingesting document...]")
    agent.ingest_document(
        """
        Our new Professional Plus plan launches in Q2 2024. It sits between Professional
        and Enterprise, priced at $349/month. It includes API Access, Advanced Analytics,
        SSO with up to 50 users, and priority support (but not dedicated support).
        The Professional Plus plan requires a minimum 12-month commitment.
        It is compatible with Okta, Azure AD, and Google Workspace for SSO configuration.
        """,
        source="product_announcement_2024",
    )

    print(f"Knowledge graph after ingestion: {agent.kg.stats()}")

    # ── Query 1: Direct entity lookup ─────────────────────────────
    print("\n\n=== QUERY 1: Enterprise Plan features ===")
    answer1 = agent.answer("What features does the Enterprise Plan include?")
    print(f"Answer:\n{answer1}")

    # ── Query 2: Multi-hop reasoning ─────────────────────────────
    print("\n\n=== QUERY 2: SSO and Okta compatibility ===")
    answer2 = agent.answer(
        "Our team uses Okta for identity management. "
        "Which of your plans would support our SSO setup?"
    )
    print(f"Answer:\n{answer2}")

    # ── Query 3: Comparison question ─────────────────────────────
    print("\n\n=== QUERY 3: BERT vs GPT-3 ===")
    answer3 = agent.answer(
        "What is the key architectural difference between BERT and GPT-3?"
    )
    print(f"Answer:\n{answer3}")

    # ── Query 4: Python question (tests tech KG) ─────────────────
    print("\n\n=== QUERY 4: Python concurrency ===")
    answer4 = agent.answer(
        "Why does Python have a GIL and how does asyncio relate to it?"
    )
    print(f"Answer:\n{answer4}")

    # ── Entity exploration ─────────────────────────────────────
    print("\n\n=== ENTITY INSPECTION: Enterprise Plan ===")
    print(agent.explain_entity("Enterprise Plan"))

    print("\n\n=== ENTITY INSPECTION: SSO ===")
    print(agent.explain_entity("SSO"))

    # ── Path finding ───────────────────────────────────────────
    print("\n\n=== PATH: Okta → Enterprise Plan ===")
    path = agent.kg.find_path("Okta", "Enterprise Plan")
    if path:
        print("Relationship path:")
        for step in path:
            print(f"  → {step}")
    else:
        path_rev = agent.kg.find_path("Enterprise Plan", "Okta")
        print("Path (Enterprise → Okta):")
        for step in (path_rev or []):
            print(f"  → {step}")


if __name__ == "__main__":
    demo()

---

Combining Knowledge Graphs with Vector Search

Knowledge graphs and vector search are complementary, not competing:

Capability	Vector RAG	Knowledge Graph
Unstructured text retrieval	Excellent	Poor
Structured relationship queries	Poor	Excellent
Multi-hop reasoning	Poor	Good
Fuzzy/semantic matching	Excellent	Poor (without hybrid)
Setup complexity	Low	High
Update precision	Low (re-embed chunks)	High (update specific nodes)

Hybrid retrieval pattern:

1. Vector search finds relevant entities and documents by semantic similarity
2. Knowledge graph traversal expands from those seed entities to related facts
3. Both results are injected into context together

This pattern, sometimes called Graph-RAG (introduced by Microsoft Research, 2024), significantly outperforms pure RAG on complex multi-hop questions.

---

:::danger Knowledge Quality Degradation LLM-extracted knowledge has an accuracy of approximately 85–90% for well-structured text. The 10–15% error rate compounds over time: wrong relationships get traversed, incorrect facts get injected into agent context, and the agent produces confidently-wrong answers. Always: (1) validate extracted facts against the source before adding to the graph, (2) store confidence scores with every LLM-extracted fact, (3) filter by confidence at query time (only include facts with confidence > 0.7 in context), (4) implement a review queue for low-confidence facts before they affect production. :::

:::warning Graph Staleness Knowledge graphs require active maintenance. Product pricing changes, features are added and removed, research findings are superseded. A graph that was accurate six months ago may contain outdated facts today. Track the last_updated timestamp on every node and edge. Set up periodic staleness alerts for high-importance entities. For rapidly-changing domains (prices, availability, API endpoints), prefer database or document storage over knowledge graphs - the update overhead is lower. :::

---

Interview Questions and Answers

Q: When would you choose a knowledge graph over plain RAG for semantic memory?

A: Knowledge graphs are superior when: (1) the domain has important structured relationships between entities - not just facts, but typed connections like INCLUDES, REQUIRES, CONFLICTS_WITH; (2) questions require multi-hop reasoning - "which plans support the authentication requirements for SOC 2?" requires traversing feature → compliance → plan relationships; (3) precise entity updates matter - changing a price or adding a feature should be a surgical update to one node, not re-embedding a dozen document chunks. Plain RAG is better when: the knowledge is primarily unstructured text (long documents, policies, documentation), the question is more about semantic similarity than structured relationships, and the setup time of graph construction is not justified by query complexity.

Q: How do you handle entity resolution in a knowledge graph built from multiple sources?

A: Entity resolution (also called entity linking or coreference resolution) is the hardest part of building a knowledge graph. "OpenAI," "Open AI," and "OpenAI LLC" are the same entity. Strategies: (1) Canonical name normalization: lowercase, strip punctuation, normalize common variations before lookup. (2) Alias tracking: maintain a separate alias index that maps all known names to the canonical entity ID. (3) LLM-based resolution: for ambiguous cases, use a model to decide if two entity mentions refer to the same real-world entity. (4) Embedding similarity: entities with high embedding similarity of their descriptions are candidates for merging - use a threshold like 0.95 cosine similarity. (5) Human review queue: flag potential duplicates for human confirmation before merging. In production, dedicate significant engineering time to entity resolution - it determines the quality of everything downstream.

Q: How do knowledge graphs handle contradictory facts from different sources?

A: Contradictions are inevitable at scale. Best practices: (1) Store confidence and source on every fact. "Enterprise Plan COSTS $500/month" from the official pricing page (confidence 1.0) should override "$499/month" from a blog post (confidence 0.6). (2) Keep all versions with timestamps - do not overwrite, archive the old fact. This creates an audit trail. (3) Conflict detection: when adding a fact with the same subject-predicate as an existing fact but a different object, trigger a conflict review workflow. (4) Source hierarchy: define a source priority order - official documentation > marketing materials > user reports > blog posts. Higher-priority sources win conflicts automatically. (5) Versioned facts: for rapidly-changing facts (pricing, availability), model them as time-bound: "Enterprise Plan COSTS $500/month VALID_FROM 2024-01 VALID_TO present."

Q: What is Graph-RAG and when should you use it?

A: Graph-RAG (Microsoft Research, 2024) is a retrieval pattern that combines knowledge graph traversal with vector search. Standard RAG retrieves document chunks by embedding similarity to the query. Graph-RAG instead: (1) uses the knowledge graph to identify a community of related entities, (2) generates community summaries that capture cross-document relationships, and (3) retrieves at the community level rather than the chunk level. This produces dramatically better answers for questions that require synthesizing information across many documents - "What are all the factors affecting X across the literature?" - because the graph structure captures cross-document relationships that chunk-level embedding similarity misses. Use Graph-RAG when: questions require global understanding across a corpus (not just local document lookup), the corpus has rich cross-reference relationships, and you can afford the higher setup and query complexity. Pure RAG is sufficient for local lookup ("what does document X say about Y?").

Q: How do you scale a knowledge graph to millions of nodes in production?

A: NetworkX is an in-memory graph limited to roughly 10–50 million edges on a large server. For production scale: (1) Neo4j: the de facto standard graph database, supports billions of nodes and edges, native Cypher query language, excellent Python driver. Runs on a single server or in a cluster. (2) Kuzu: embedded graph database (like SQLite for graphs), no server needed, excellent for single-agent deployments. (3) Amazon Neptune or Google Spanner: managed graph databases for cloud deployments. (4) TigerGraph: purpose-built for very large-scale graph analytics. For most agent use cases, the knowledge graph is domain-specific (a company's product catalog, a research literature graph) and stays well within Neo4j's single-server capacity. The real scaling challenge is usually not the graph itself but entity resolution at ingestion time - running LLM extraction over millions of documents is expensive and slow. Use batching, parallelism, and cheap models for extraction; only use expensive models for validation.