managing-knowledge-graphs

name	managing-knowledge-graphs
description	Graph RAG patterns, entity extraction with LLMs, relationship mapping, Use when this capability is needed.
metadata	{"author":"gitwalter"}

Knowledge Graphs

Graph RAG patterns, entity extraction with LLMs, relationship mapping, and Neo4j integration

Build knowledge graphs for RAG systems using entity extraction, relationship mapping, and graph databases like Neo4j.

Process

1. Review the task requirements.
2. Apply the skill's methodology.
3. Validate the output against the defined criteria.

Step 1: Entity Extraction with LLMs

from langchain_google_genai import ChatGoogleGenerativeAI
from langchain.prompts import PromptTemplate
from pydantic import BaseModel, Field
from typing import List, Optional
import json

class Entity(BaseModel):
    """Entity model."""
    name: str = Field(description="Entity name")
    type: str = Field(description="Entity type (Person, Organization, Concept, etc.)")
    description: Optional[str] = Field(default=None, description="Brief description")

class Relationship(BaseModel):
    """Relationship model."""
    source: str = Field(description="Source entity name")
    target: str = Field(description="Target entity name")
    relationship_type: str = Field(description="Type of relationship")
    description: Optional[str] = Field(default=None, description="Relationship description")

class KnowledgeGraph(BaseModel):
    """Knowledge graph structure."""
    entities: List[Entity] = Field(description="List of entities")
    relationships: List[Relationship] = Field(description="List of relationships")

class EntityExtractor:
    """Extract entities and relationships using LLMs."""

    def __init__(self, model: str = "gemini-2.5-flash"):
        self.llm = ChatGoogleGenerativeAI(model=model)

    def extract_from_text(self, text: str) -> KnowledgeGraph:
        """Extract entities and relationships from text."""
        prompt = PromptTemplate(
            template="""Extract entities and relationships from the following text.

Text: {text}

Return a JSON object with:
- "entities": [{{"name": "...", "type": "...", "description": "..."}}]
- "relationships": [{{"source": "...", "target": "...", "relationship_type": "...", "description": "..."}}]

Focus on:
- Named entities (people, organizations, locations, concepts)
- Clear relationships between entities
- Important facts and connections

JSON:""",
            input_variables=["text"]
        )

        chain = prompt | self.llm
        response = chain.invoke({"text": text})

        # Parse JSON
        try:
            data = json.loads(response.content)
            return KnowledgeGraph(**data)
        except:
            # Fallback: try structured output
            return self._extract_structured(text)

    def _extract_structured(self, text: str) -> KnowledgeGraph:
        """Extract using structured output."""
        from langchain_core.output_parsers import PydanticOutputParser

        parser = PydanticOutputParser(pydantic_object=KnowledgeGraph)

        prompt = PromptTemplate(
            template="""Extract entities and relationships from the text.

Text: {text}

{format_instructions}""",
            input_variables=["text"],
            partial_variables={"format_instructions": parser.get_format_instructions()}
        )

        chain = prompt | self.llm | parser
        return chain.invoke({"text": text})

# Usage
extractor = EntityExtractor()
text = "Apple Inc. was founded by Steve Jobs. Tim Cook is the current CEO."
kg = extractor.extract_from_text(text)

### Step 2: Building Knowledge Graph
Prioritize the use of the `memory` MCP server for persistent, local graph storage.

```python
# Create entities and relations in the managing-memory-bank
create_entities(entities=[...])
create_relations(relations=[...])

from typing import Dict, List, Set
from dataclasses import dataclass

@dataclass
class Entity:
    """Entity in knowledge graph."""
    id: str
    name: str
    type: str
    properties: Dict = None

    def __post_init__(self):
        if self.properties is None:
            self.properties = {}

@dataclass
class Relationship:
    """Relationship in knowledge graph."""
    source_id: str
    target_id: str
    relationship_type: str
    properties: Dict = None

    def __post_init__(self):
        if self.properties is None:
            self.properties = {}

class KnowledgeGraphBuilder:
    """Build and manage knowledge graph."""

    def __init__(self):
        self.entities: [, Entity] = {}
        .relationships: [Relationship] = []
        .entity_index: [, ] = {}  

     () -> :
        
        
         name  .entity_index:
            entity_id = .entity_index[name]
            
             properties:
                .entities[entity_id].properties.update(properties)
             entity_id

        
        entity_id = 
        entity = Entity(
            =entity_id,
            name=name,
            =entity_type,
            properties=properties  {}
        )

        .entities[entity_id] = entity
        .entity_index[name] = entity_id

         entity_id

     ():
        
        
        source_id = .entity_index.get(source_name)
          source_id:
            source_id = .add_entity(source_name, )

        target_id = .entity_index.get(target_name)
          target_id:
            target_id = .add_entity(target_name, )

        
        rel = Relationship(
            source_id=source_id,
            target_id=target_id,
            relationship_type=relationship_type,
            properties=properties  {}
        )

        .relationships.append(rel)

     ():
        
        
         entity  kg.entities:
            .add_entity(
                name=entity.name,
                entity_type=entity.,
                properties={: entity.description}  entity.description  {}
            )

        
         rel  kg.relationships:
            .add_relationship(
                source_name=rel.source,
                target_name=rel.target,
                relationship_type=rel.relationship_type,
                properties={: rel.description}  rel.description  {}
            )

     () -> [Entity]:
        
        entity_id = .entity_index.get(entity_name)
          entity_id:
             []

        neighbor_ids = ()
         rel  .relationships:
             rel.source_id == entity_id:
                neighbor_ids.add(rel.target_id)
             rel.target_id == entity_id:
                neighbor_ids.add(rel.source_id)

         [.entities[eid]  eid  neighbor_ids  eid  .entities]

     () -> :
        
         {
            : [
                {
                    : e.,
                    : e.name,
                    : e.,
                    : e.properties
                }
                 e  .entities.values()
            ],
            : [
                {
                    : r.source_id,
                    : r.target_id,
                    : r.relationship_type,
                    : r.properties
                }
                 r  .relationships
            ]
        }

Step 3: Neo4j Integration

from neo4j import GraphDatabase
from typing import List, Dict

class Neo4jKnowledgeGraph:
    """Knowledge graph stored in Neo4j."""

    def __init__(self, uri: str, user: str, password: str):
        self.driver = GraphDatabase.driver(uri, auth=(user, password))

    def close(self):
        """Close database connection."""
        self.driver.close()

    def create_entity(self, name: str, entity_type: str, properties: Dict = None):
        """Create entity node."""
        with self.driver.session() as session:
            props = properties or {}
            props["name"] = name
            props["type"] = entity_type

            query = """
            MERGE (e:Entity {name: $name})
            SET e.type = $type
            SET e += $properties
            RETURN e
            """

            session.run(query, name=name, type=entity_type, properties=props)

    def create_relationship(self, source_name: str, target_name: str,
                           relationship_type: str, properties: Dict = ):
        
         .driver.session()  session:
            props = properties  {}

            query = 

            session.run(query,
                       source_name=source_name,
                       target_name=target_name,
                       properties=props)

     () -> []:
        
         .driver.session()  session:
             entity_type:
                query = 
                result = session.run(query, =entity_type, limit=limit)
            :
                query = 
                result = session.run(query, limit=limit)

             [record.data()  record  result]

     () -> [[]]:
        
         .driver.session()  session:
            query = 

            result = session.run(query,
                                source_name=source_name,
                                target_name=target_name,
                                max_depth=max_depth)

             [record[]  record  result]

     () -> :
        
         .driver.session()  session:
            query = 

            result = session.run(query, name=entity_name)
            record = result.single()

             record:
                 {
                    : (record[]),
                    : [(c)  c  record[]],
                    : [(r)  rels  record[]  r  rels]
                }
             {}


neo4j_kg = Neo4jKnowledgeGraph(, , )
neo4j_kg.create_entity(, , {: })
neo4j_kg.create_entity(, )
neo4j_kg.create_relationship(, , )

Step 4: Graph RAG Patterns

from langchain_google_genai import ChatGoogleGenerativeAI
from langchain.prompts import PromptTemplate
from typing import List, Dict

class GraphRAG:
    """RAG system using knowledge graph."""

    def __init__(self, neo4j_kg: Neo4jKnowledgeGraph):
        self.kg = neo4j_kg
        self.llm = ChatGoogleGenerativeAI(model="gemini-2.5-flash")

    def extract_query_entities(self, query: str) -> List[str]:
        """Extract entity names from query."""
        prompt = PromptTemplate(
            template="""Extract entity names from this query. Return only the entity names, comma-separated.

Query: {query}

Entities:""",
            input_variables=["query"]
        )

        chain = prompt | self.llm
        response = chain.invoke({"query": query})

        entities = [e.strip() for e in response.content.split(",")]
        return entities

    def retrieve_subgraph(self, entity_names: List[str], depth: int = 2) -> Dict:
        """Retrieve subgraph around entities."""
        all_entities = set()
        all_relationships = []

        for entity_name  entity_names:
            context = .kg.get_entity_context(entity_name, depth=depth)
             context:
                all_entities.add(context[][])
                 conn  context[]:
                    all_entities.add(conn[])
                all_relationships.extend(context[])

         {
            : (all_entities),
            : all_relationships
        }

     () -> :
        
        
        entities = .extract_query_entities(query)

          entities:
             {: , : []}

        
        subgraph = .retrieve_subgraph(entities)

        
        entities_str = .join(subgraph[])
        relationships_str = .join([
            
             r  subgraph[]
        ])

        context = 

        
        prompt = PromptTemplate(
            template=,
            input_variables=[, ]
        )

        chain = prompt | .llm
        answer = chain.invoke({: context, : query})

         {
            : answer.content  (answer, )  (answer),
            : entities,
            : subgraph
        }

Step 5: Entity Resolution and Merging

class EntityResolver:
    """Resolve and merge duplicate entities."""

    def __init__(self, llm):
        self.llm = llm

    def find_duplicates(self, entities: List[Entity], threshold: float = 0.8) -> List[List[str]]:
        """Find potential duplicate entities."""
        from sentence_transformers import SentenceTransformer

        model = SentenceTransformer("all-MiniLM-L6-v2")

        # Embed entity names
        names = [e.name for e in entities]
        embeddings = model.encode(names)

        # Find similar pairs
        duplicates = []
        seen = set()

        for i, entity1 in enumerate(entities):
            if entity1.name in seen:
                continue

            group = [entity1.name]

            for j, entity2 in enumerate(entities[i+1:], i+1):
                similarity = self._cosine_similarity(embeddings[i], embeddings[j])
                if similarity >= threshold:
                    group.append(entity2.name)
                    seen.add(entity2.name)

            if len(group) > 1:
                duplicates.append(group)
                seen.add(entity1.name)

        return duplicates

     ():
        
         numpy  np
         np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))

     ():
        
         group  entity_groups:
             (group) < :
                

            
            canonical = group[]
            duplicates = group[:]

            
            canonical_id = kg_builder.entity_index.get(canonical)

            
             duplicate_name  duplicates:
                duplicate_id = kg_builder.entity_index.get(duplicate_name)
                  duplicate_id:
                    

                
                 rel  kg_builder.relationships:
                     rel.source_id == duplicate_id:
                        rel.source_id = canonical_id
                     rel.target_id == duplicate_id:
                        rel.target_id = canonical_id

                
                 duplicate_id  kg_builder.entities:
                     kg_builder.entities[duplicate_id]
                 duplicate_name  kg_builder.entity_index:
                     kg_builder.entity_index[duplicate_name]

Step 6: Complete Graph RAG System

class CompleteGraphRAG:
    """Complete graph RAG system."""

    def __init__(self, neo4j_uri: str, neo4j_user: str, neo4j_password: str):
        self.neo4j_kg = Neo4jKnowledgeGraph(neo4j_uri, neo4j_user, neo4j_password)
        self.extractor = EntityExtractor()
        self.graph_rag = GraphRAG(self.neo4j_kg)
        self.builder = KnowledgeGraphBuilder()

    def ingest_documents(self, documents: List[str]):
        """Ingest documents and build knowledge graph."""
        for doc_text in documents:
            # Extract entities and relationships
            kg = self.extractor.extract_from_text(doc_text)

            # Merge into builder
            self.builder.merge_kg(kg)

        # Resolve duplicates
        resolver = EntityResolver(self.graph_rag.llm)
        duplicates = resolver.find_duplicates(list(self.builder.entities.values()))
        resolver.merge_entities(duplicates, self.builder)

        # Store in Neo4j
        for entity in self.builder.entities.values():
            self.neo4j_kg.create_entity(
                entity.name,
                entity.type,
                entity.properties
            )

        for rel in .builder.relationships:
            source_name = .builder.entities[rel.source_id].name
            target_name = .builder.entities[rel.target_id].name
            .neo4j_kg.create_relationship(
                source_name,
                target_name,
                rel.relationship_type,
                rel.properties
            )

     () -> :
        
         .graph_rag.answer_with_graph(question)

     ():
        
        .neo4j_kg.close()

Knowledge Graph Patterns

| Pattern | Use Case | Pros | Cons | ||-||| | LLM Extraction | Unstructured text | High quality | Slower, costs | | Rule-based | Structured data | Fast, precise | Limited coverage | | Hybrid | Mixed sources | Best of both | More complex | | Graph RAG | Entity queries | Structured answers | Requires graph DB |

Best Practices

• Use LLMs for entity extraction from unstructured text
• Store entities with rich metadata (type, properties)
• Resolve duplicate entities before storing
• Use graph databases (Neo4j) for complex queries
• Extract relationships explicitly, not just entities
• Use subgraph retrieval for focused context
• Maintain entity canonicalization
• Index entities for fast lookup

Anti-Patterns

Anti-Pattern	Fix
No entity resolution	Merge duplicates before storage
Ignoring relationships	Extract and store relationships
Flat entity storage	Use graph database
No metadata	Store entity types and properties
Single extraction pass	Iteratively refine graph
No canonicalization	Resolve entity variants
Ignoring graph structure	Use graph queries for retrieval

Related

• Skill: applying-rag-patterns
• Skill: retrieving-advanced
• Skill: vision-agents

When to Use

This skill should be used when strict adherence to the defined process is required.

Prerequisites

• Basic understanding of the agent factory context.
• Access to the necessary tools and resources.

---

Converted and distributed by TomeVault — claim your Tome and manage your conversions.