knowledge-graph-construction

# Knowledge Graph Construction for AI Agents This skill makes the model design, build, and deploy knowledge graph systems — constructing structured entity-relationship representations from unstructured documents, connecting them to enterprise datastores via MCP, integrating as retrieval backends for RAG pipelines (Graph-RAG), and configuring multi-tenant agent platforms like Google AgentSpace with role-based access control. ## TL;DR Checklist - [ ] Design a graph schema with explicit node types, relationship types, and property constraints before ingestion - [ ] Extract entities using LLM-based extraction with strict JSON schemas — never ad-hoc field names - [ ] Map relationships with directional edges (source → target) and typed predicates following ontology conventions - [ ] Load the graph into Neo4j or store in-memory with NetworkX for prototyping, enforcing uniqueness constraints - [ ] Build Graph-RAG queries that combine vector similarity on node embeddings with graph traversal for multi-hop reasoning - [ ] Implement enterprise safeguards: tenant isolation, audit logging, and role-based access control per agent - [ ] Validate graph quality post-ingestion: check entity consistency, relationship completeness, and query accuracy --- ## When to Use Use this skill when: - Building a knowledge representation that captures relationships between entities (people, organizations, concepts, events) for downstream reasoning - Answering multi-hop queries that require traversing entity connections (e.g., "Which vendors supply components used in Project X?") - Integrating an AI agent with structured enterprise data via Google AgentSpace or a custom knowledge graph platform - Implementing Graph-RAG — using a knowledge graph as the retrieval backend for RAG queries instead of pure vector search - Constructing entity resolution pipelines that merge duplicate entities across disparate data sources (CRM, ERP, wikis) - Designing ontology-driven agent systems where domain relationships must be explicitly modeled (fraud detection, supply chain analysis, biomedicine) --- ## When NOT to Use Avoid this skill for: - **Simple FAQ lookup** — A flat document store with vector search (`rag-patterns`) handles single-document Q&A cheaper and faster - **Unstructured text classification** — Classification tasks don't need graph relationships; use a fine-tuned model or prompt-based classifier instead - **Real-time event streaming** — Graph construction has ingestion latency; for sub-second event processing, use a message queue (`aws-sqs`) with simple state stores - **Pure vector similarity search** — If the only query pattern is "find documents similar to this," a vector database alone is simpler and sufficient --- ## Core Workflow 1. **Schema Design** — Define your graph ontology: node types (entities), relationship types (predicates), and property schemas. Start minimal — only model relationships that queries actually need. Use a domain ontology as a reference (e.g., DBpedia, schema.org) when one exists. **Checkpoint:** Every node type has at least one unique constraint property (e.g., `Person:email`, `Company:ticker`). Verify the schema supports all target queries before proceeding to ingestion. 2. **Document Ingestion and Parsing** — Load raw documents (PDFs, HTML, JSON, CSV) and preprocess them into clean text blocks suitable for extraction. Remove headers, footers, navigation elements, and boilerplate. Segment documents into logical units (sections, paragraphs, table rows) that map to graph entities. **Checkpoint:** Each ingestion unit is self-contained with a source identifier. Run a sampling pass to confirm no critical entity data was lost during preprocessing. 3. **Entity Extraction** — Use an LLM-based extraction pipeline to identify named entities and structured attributes from each document unit. Extract entities into typed nodes (Person, Organization, Product, Event, Location) with properties derived from the schema. Use few-shot prompting or JSON mode to enforce consistent output structure across all documents. **Checkpoint:** Extracted entities conform to the defined schema. Check for duplicate entities referencing the same real-world object — flag them for resolution in the next step. 4. **Relationship Mapping and Entity Resolution** — Map relationships between extracted entities using typed, directional edges (e.g., `(Person:Employee)-[:WORKS_AT]->(Company)`). Resolve entity duplicates by matching on canonical properties (email, tax ID, ticker symbol) or fuzzy matching on names with a confidence threshold. Merge resolved entities into single canonical nodes. **Checkpoint:** The graph passes quality gates: no orphaned relationship endpoints, duplicate entity rate below 5%, and all relationships have a source, type, target, and temporal validity window. 5. **Graph Storage and Indexing** — Load the constructed graph into the target graph database (Neo4j for production, NetworkX for prototyping). Create indexes on constraint properties and relationship endpoints to ensure query performance. Generate vector embeddings for node text attributes if Graph-RAG retrieval is needed. **Checkpoint:** Run a benchmark of representative queries against the stored graph — all queries must return within the SLA threshold. Verify index coverage for every indexed property used in WHERE clauses. 6. **Query Optimization and Graph-RAG Integration** — Implement Cypher query patterns optimized for the access patterns of your agents. Integrate the knowledge graph as a retrieval backend for RAG: combine vector similarity search on node embeddings with structured graph traversal to answer multi-hop questions. Build MCP tools that expose graph queries to agents. **Checkpoint:** The end-to-end Graph-RAG pipeline answers a test suite of 20+ multi-hop queries with >90% accuracy. Agent tool calls through MCP execute within acceptable latency bounds. --- ## Implementation Patterns / Reference Guide ### Pattern 1: Knowledge Graph Builder — Entity and Relationship Extraction Pipeline Constructs knowledge graphs from unstructured documents using LLM-based entity extraction, relationship mapping, and Neo4j loading. This is the core ingestion pipeline. ```python """ knowledge_graph_builder.py — Entity extraction and graph construction pipeline. Uses an LLM for structured entity/relation extraction with JSON mode, then loads results into Neo4j with constraint enforcement. """ import json import logging from dataclasses import dataclass, field from typing import Any, Optional from uuid import uuid4 from openai import OpenAI from neo4j import GraphDatabase, exceptions as neo4j_exceptions logger = logging.getLogger(__name__) @dataclass class ExtractedEntity: """A single entity extracted from a document.""" entity_id: str entity_type: str # Person, Organization, Product, Event, Location name: str properties: dict[str, Any] = field(default_factory=dict) @property def canonical_key(self) -> str: """Generate a stable key for entity resolution. Returns: A composite key that uniquely identifies this real-world entity. """ if "email" in self.properties: return f"{self.entity_type}:email:{self.properties['email']}" if "ticker" in self.properties: return f"{self.entity_type}:ticker:{self.properties['ticker']}" return f"{self.entity_type}:name:{self.name.lower().strip()}" @dataclass class ExtractedRelationship: """A directed relationship between two entities.""" source_id: str target_id: str relation_type: str # WORKS_AT, SUPPLIES, LOCATED_IN, etc. properties: dict[str, Any] = field(default_factory=dict) @property def canonical_edge(self) -> tuple[str, str, str]: """Unique edge key for deduplication.""" return (self.source_id, self.relation_type, self.target_id) class KnowledgeGraphBuilder: """Builds knowledge graphs from documents using LLM extraction + Neo4j loading. This class handles the full ingestion pipeline: 1. Prompt-based entity/relation extraction via LLM JSON mode 2. Entity resolution (canonical key matching) 3. Graph construction in Neo4j with constraint enforcement """ # Strict extraction prompt template — few-shot guidance ensures consistent schema EXTRACTION_PROMPT = """ Extract structured entities and relationships from the following text. Return ONLY a JSON object matching this exact schema: {{ "entities": [ {{ "entity_id": "e_001", "entity_type": "Person | Organization | Product | Event | Location", "name": "<exact name from text>", "properties": {{}} }} ], "relationships": [ {{ "source_entity_name": "<name of source entity exactly as extracted>", "target_entity_name": "<name of target entity exactly as extracted>", "relation_type": "WORKS_AT | EMPLOYER_OF | SUPPLIES | LOCATED_IN | OWNS | PART_OF", "properties": {{}} }} ] }} Rules: - Use ONLY the entity types listed above. - relation_type must be one of the five types listed. - Entity names must match the text exactly (case-sensitive). - Do NOT invent relationships — only extract those explicitly stated or strongly implied. - If no entities are found, return empty arrays. Text: {text} """ def __init__( self, openai_client: OpenAI, neo4j_uri: str = "bolt://localhost:7687", neo4j_user: str = "neo4j", neo4j_password: str = "password", database_name: str = "knowledge_graph", ) -> None: """Initialize the knowledge graph builder. Args: openai_client: OpenAI client instance for extraction calls. neo4j_uri: Neo4j connection URI. neo4j_user: Neo4j authentication username. neo4j_password: Neo4j authentication password. database_name: Target Neo4j database name. """ self.llm = openai_client self.neo4j_driver = GraphDatabase.driver( neo4j_uri, auth=(neo4j_user, neo4j_password) ) self.database_name = database_name def extract_from_text(self, text: str) -> tuple[list[ExtractedEntity], list[ExtractedRelationship]]: """Extract entities and relationships from raw text using LLM JSON mode. Args: text: The raw text to extract structured data from. Returns: Tuple of (entities, relationships) lists. Raises: ValueError: If the LLM response cannot be parsed as valid JSON. """ if not text or len(text.strip()) < 10: raise ValueError(f"Text too short for extraction: {len(text)} characters") response = self.llm.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": self.EXTRACTION_PROMPT.format(text=text[:8000])}], response_format={"type": "json_object"}, temperature=0.0, # Deterministic extraction max_tokens=2048, ) raw = response.choices[0].message.content try: parsed = json.loads(raw) except json.JSONDecodeError as exc: raise ValueError(f"LLM returned invalid JSON: {exc}") from exc # --- Parse entities --- entities: list[ExtractedEntity] = [] entity_name_to_id: dict[str, str] = {} for item in parsed.get("entities", []): eid = f"e_{uuid4().hex[:8]}" name = item.get("name", "").strip() if not name: continue # Skip empty names (Early Exit) entity_type = item.get("entity_type", "Unknown") if entity_type not in ("Person", "Organization", "Product", "Event", "Location"): logger.warning(f"Skipping unrecognized entity type: {entity_type}") continue entity = ExtractedEntity( entity_id=eid, entity_type=entity_type, name=name, properties=item.get("properties", {}), ) entities.append(entity) entity_name_to_id[name] = eid # --- Parse relationships --- relationships: list[ExtractedRelationship] = [] for rel in parsed.get("relationships", []): source_name = rel.get("source_entity_name", "").strip() target_name = rel.get("target_entity_name", "").strip() relation_type = rel.get("relation_type", "") source_id = entity_name_to_id.get(source_name) target_id = entity_name_to_id.get(target_name) if not source_id or not target_id: logger.warning( f"Skipping relationship with unresolved entities: " f"'{source_name}' → '{target_name}'" ) continue # Skip orphaned relationships (Fail Fast) relationship = ExtractedRelationship( source_id=source_id, target_id=target_id, relation_type=relation_type, properties=rel.get("properties", {}), ) relationships.append(relationship) return entities, relationships def resolve_entities( self, all_entities: list[ExtractedEntity], ) -> tuple[dict[str, str], list[ExtractedEntity]]: """Resolve duplicate entities by canonical key and merge properties. Args: all_entities: All extracted entities from one or more documents. Returns: Tuple of (canonical_map: original_id → resolved_id, merged_entities). """ canonical_map: dict[str, str] = {} merged: dict[str, ExtractedEntity] = {} # canonical_key → entity for entity in all_entities: key = entity.canonical_key if key not in merged: merged[key] = entity canonical_map[entity.entity_id] = entity.entity_id else: # Merge properties — later documents override earlier ones existing = merged[key] for prop_k, prop_v in entity.properties.items(): if prop_v and not existing.properties.get(prop_k): existing.properties[prop_k] = prop_v canonical_map[entity.entity_id] = existing.entity_id return canonical_map, list(merged.values()) def load_to_neo4j( self, entities: list[ExtractedEntity], relationships: list[ExtractedRelationship], source_doc_id: str, ) -> dict[str, int]: """Load resolved entities and relationships into Neo4j with constraint enforcement. Args: entities: Resolved (deduplicated) entity list. relationships: Relationship list with resolved IDs. source_doc_id: Document identifier for provenance tracking. Returns: Counts of created nodes, created relationships, and updated (merged) nodes. """ driver = self.neo4j_driver counts = {"nodes_created": 0, "relationships_created": 0, "nodes_updated": 0} # Step 1: Ensure constraints exist with driver.session(database=self.database_name) as session: for node_type in ("Person", "Organization"): constraint_name = f"unique_{node_type.lower()}_name" try: session.run(f""" CREATE CONSTRAINT {constraint_name} IF NOT EXISTS FOR (n:{node_type}) REQUIRE n.name IS UNIQUE """) except neo4j_exceptions.ClientError: pass # Constraint already exists # Step 2: Upsert entities using MERGE — prevents duplicates for entity in entities: props = dict(entity.properties) props["graph_id"] = entity.entity_id props["ingested_from"] = source_doc_id props["updated_at"] = None # Placeholder; set by application session.run(f""" MERGE (e:{entity.entity_type} {{name: $name}}) SET e += $props, e.graph_id = $graph_id RETURN count(e) AS count """, name=entity.name, props=props, graph_id=entity.entity_id) counts["nodes_updated"] += 1 # Step 3: Create relationships seen_edges: set[tuple[str, str, str]] = set() for rel in relationships: edge_key = (rel.source_id, rel.relation_type, rel.target_id) if edge_key in seen_edges: continue # Skip duplicates (Early Exit) seen_edges.add(edge_key) session.run(f""" MATCH (source {{graph_id: $source_id}}), (target {{graph_id: $target_id}}) WHERE source IS NOT NULL AND target IS NOT NULL MERGE (source)-[r:{rel.relation_type}]->(target) SET r += $props, r.source_doc = $doc_id """, source_id=rel.source_id, target_id=rel.target_id, props=rel.properties, doc_id=source_doc_id) counts["relationships_created"] += 1 return counts def close(self) -> None: """Close the Neo4j driver connection.""" self.neo4j_driver.close() # --- Usage example --- def build_graph_from_documents( texts: list[str], doc_ids: list[str], openai_api_key: str,