Knowledge Graph Skill

Role

You are an elite knowledge graph engineer. You know when vector RAG fails

(multi-hop questions, relationship queries, global summaries) and how to build

graph-based retrieval that handles them. You construct knowledge graphs from

documents, store them in Neo4j, and query them semantically.

When to Use Graph RAG vs Vector RAG

Vector RAG wins:  "What did the survey say about tribal fans?"
Graph RAG wins:   "Which brands share fans with Man City AND have declining engagement?"
Graph RAG wins:   "How is brand loyalty connected to player transfers across clubs?"
Graph RAG wins:   "What are the common themes across all IMI reports this year?"

Part 1: Entity Extraction from Documents

# pip install spacy networkx
# python -m spacy download en_core_web_lg
import spacy
from collections import defaultdict

nlp = spacy.load("en_core_web_lg")

# Entity types relevant to IMI research
IMI_ENTITY_TYPES = {
    "ORG": "organisation",
    "PERSON": "person",
    "GPE": "location",
    "PRODUCT": "product",
    "EVENT": "event",
    "MONEY": "financial",
    "PERCENT": "metric",
    "NORP": "group"  # Nationalities, religions, political groups
}

def extract_entities(text: str,
                      entity_types: dict = None,
                      min_confidence: float = 0.7) -> list[dict]:
    """Extract named entities from text using spaCy."""
    entity_types = entity_types or IMI_ENTITY_TYPES
    doc = nlp(text)
    entities = []

    for ent in doc.ents:
        if ent.label_ in entity_types:
            entities.append({
                "text": ent.text.strip(),
                "label": ent.label_,
                "type": entity_types[ent.label_],
                "start": ent.start_char,
                "end": ent.end_char,
                "context": text[max(0, ent.start_char - 50):ent.end_char + 50]
            })

    return entities


def extract_relations_via_llm(text: str, entities: list[dict],
                                llm_fn: callable) -> list[dict]:
    """
    Extract relationships between entities using LLM.
    Returns triples: (subject, relation, object)
    """
    entity_names = list(set(e["text"] for e in entities))
    if len(entity_names) < 2:
        return []

    prompt = f"""Extract relationships between the entities in this text.

Entities: {', '.join(entity_names[:20])}

Text: {text[:2000]}

Return a JSON array of relationships in this format:
[{{"subject": "entity1", "relation": "relationship_type", "object": "entity2"}}]

Relationship types to use: SPONSORS, COMPETES_WITH, EMPLOYS, PARTNERS_WITH,
OWNS, LOCATED_IN, SUPPORTS, INFLUENCES, RELATES_TO, HAS_METRIC

Return only the JSON array."""

    import json
    try:
        response = llm_fn(prompt)
        # Extract JSON
        import re
        match = re.search(r'\[[\s\S]*\]', response)
        if match:
            return json.loads(match.group(0))
    except Exception:
        pass
    return []

Part 2: NetworkX Graph Construction

import networkx as nx
from typing import Generator

class IMIKnowledgeGraph:
    """
    In-memory knowledge graph for IMI research data.
    Uses NetworkX for graph operations.
    """

    def __init__(self):
        self.graph = nx.DiGraph()
        self.entity_index: dict[str, dict] = {}

    def add_entity(self, entity_id: str, entity_type: str,
                    properties: dict = None) -> None:
        """Add an entity node to the graph."""
        self.graph.add_node(
            entity_id,
            type=entity_type,
            **({} if properties is None else properties)
        )
        self.entity_index[entity_id.lower()] = {
            "id": entity_id,
            "type": entity_type
        }

    def add_relation(self, subject: str, relation: str, obj: str,
                      confidence: float = 1.0,
                      source: str = "") -> None:
        """Add a directed relationship edge."""
        # Auto-create nodes if missing
        if subject not in self.graph:
            self.add_entity(subject, "unknown")
        if obj not in self.graph:
            self.add_entity(obj, "unknown")

        self.graph.add_edge(
            subject, obj,
            relation=relation,
            confidence=confidence,
            source=source
        )

    def build_from_documents(self, documents: list[dict],
                               llm_fn: callable,
                               text_field: str = "text") -> None:
        """Build graph from a corpus of documents."""
        for doc in documents:
            text = doc[text_field]
            source = doc.get("source", "")

            # Extract entities
            entities = extract_entities(text)
            for ent in entities:
                self.add_entity(ent["text"], ent["type"])

            # Extract relations
            if len(entities) >= 2:
                relations = extract_relations_via_llm(text, entities, llm_fn)
                for rel in relations:
                    self.add_relation(
                        rel.get("subject", ""),
                        rel.get("relation", "RELATES_TO"),
                        rel.get("object", ""),
                        source=source
                    )

    def find_connections(self, entity: str, depth: int = 2) -> list[dict]:
        """Find all entities connected to a given entity within N hops."""
        # Case-insensitive lookup
        entity_data = self.entity_index.get(entity.lower())
        if not entity_data:
            return []

        node = entity_data["id"]
        connected = []

        for path_length in range(1, depth + 1):
            for neighbour in nx.ego_graph(self.graph, node, radius=path_length).nodes():
                if neighbour != node:
                    # Get edge data
                    edges = []
                    if self.graph.has_edge(node, neighbour):
                        edges.append(self.graph[node][neighbour])
                    if self.graph.has_edge(neighbour, node):
                        edges.append(self.graph[neighbour][node])

                    connected.append({
                        "entity": neighbour,
                        "type": self.graph.nodes[neighbour].get("type", "unknown"),
                        "hops": path_length,
                        "relations": [e.get("relation", "") for e in edges]
                    })

        return connected

    def find_path(self, entity_a: str, entity_b: str) -> list[str] | None:
        """Find shortest path between two entities."""
        try:
            path = nx.shortest_path(self.graph, entity_a, entity_b)
            return path
        except (nx.NetworkXNoPath, nx.NodeNotFound):
            return None

    def get_central_entities(self, top_n: int = 10) -> list[dict]:
        """Find most connected entities (by PageRank)."""
        pagerank = nx.pagerank(self.graph, alpha=0.85)
        sorted_nodes = sorted(pagerank.items(), key=lambda x: x[1], reverse=True)

        return [
            {
                "entity": node,
                "type": self.graph.nodes[node].get("type", "unknown"),
                "pagerank": round(score, 4),
                "degree": self.graph.degree(node)
            }
            for node, score in sorted_nodes[:top_n]
        ]

    def export_for_visualisation(self) -> dict:
        """Export graph as nodes/edges for visualisation."""
        return {
            "nodes": [
                {
                    "id": n,
                    "type": self.graph.nodes[n].get("type", "unknown"),
                    **{k: v for k, v in self.graph.nodes[n].items() if k != "type"}
                }
                for n in self.graph.nodes()
            ],
            "edges": [
                {
                    "source": u,
                    "target": v,
                    "relation": self.graph[u][v].get("relation", ""),
                    "confidence": self.graph[u][v].get("confidence", 1.0)
                }
                for u, v in self.graph.edges()
            ],
            "stats": {
                "nodes": self.graph.number_of_nodes(),
                "edges": self.graph.number_of_edges(),
                "density": round(nx.density(self.graph), 4)
            }
        }

Part 3: Neo4j (Production Graph Database)

# pip install neo4j
from neo4j import GraphDatabase
import os

class Neo4jIMIGraph:
    """Production knowledge graph using Neo4j."""

    def __init__(self, uri: str = None, user: str = "neo4j",
                  password: str = None):
        uri = uri or os.environ.get("NEO4J_URI", "bolt://localhost:7687")
        password = password or os.environ.get("NEO4J_PASSWORD", "password")
        self.driver = GraphDatabase.driver(uri, auth=(user, password))

    def close(self):
        self.driver.close()

    def create_brand_entity(self, brand: str, properties: dict = None) -> None:
        """Create or update a brand node."""
        with self.driver.session() as session:
            session.run("""
                MERGE (b:Brand {name: $brand})
                SET b += $props
            """, brand=brand, props=properties or {})

    def create_relationship(self, from_entity: str, from_label: str,
                             relation: str, to_entity: str, to_label: str,
                             properties: dict = None) -> None:
        """Create a typed relationship between two entities."""
        cypher = f"""
            MERGE (a:{from_label} {{name: $from_name}})
            MERGE (b:{to_label} {{name: $to_name}})
            MERGE (a)-[r:{relation}]->(b)
            SET r += $props
        """
        with self.driver.session() as session:
            session.run(cypher,
                        from_name=from_entity,
                        to_name=to_entity,
                        props=properties or {})

    def query_brand_network(self, brand: str, depth: int = 2) -> list[dict]:
        """Get brand's relationship network up to N hops."""
        with self.driver.session() as session:
            result = session.run(f"""
                MATCH path = (b:Brand {{name: $brand}})-[*1..{depth}]-(connected)
                RETURN
                    connected.name AS entity,
                    labels(connected)[0] AS entity_type,
                    length(path) AS hops,
                    [r IN relationships(path) | type(r)] AS relation_path
                ORDER BY hops
                LIMIT 50
            """, brand=brand)
            return [dict(r) for r in result]

    def find_shared_fans(self, brand_a: str, brand_b: str) -> list[str]:
        """Find fan segments shared between two brands."""
        with self.driver.session() as session:
            result = session.run("""
                MATCH (a:Brand {name: $brand_a})-[:HAS_SEGMENT]->(s:Segment)
                MATCH (b:Brand {name: $brand_b})-[:HAS_SEGMENT]->(s)
                RETURN s.name AS shared_segment
            """, brand_a=brand_a, brand_b=brand_b)
            return [r["shared_segment"] for r in result]

    def ingest_imi_report(self, report: dict, llm_fn: callable) -> None:
        """Ingest an IMI report into the knowledge graph."""
        brand = report.get("brand", "")
        if brand:
            self.create_brand_entity(brand, {
                "fan_index": report.get("fan_index_score"),
                "report_date": report.get("report_date", "")
            })

        # Extract and store relationships
        text = report.get("text", "")
        entities = extract_entities(text)
        relations = extract_relations_via_llm(text, entities, llm_fn)

        for rel in relations:
            subj_type = next(
                (e["type"] for e in entities if e["text"] == rel["subject"]),
                "Entity"
            )
            obj_type = next(
                (e["type"] for e in entities if e["text"] == rel["object"]),
                "Entity"
            )
            self.create_relationship(
                rel["subject"], subj_type.title(),
                rel["relation"].upper().replace(" ", "_"),
                rel["object"], obj_type.title()
            )

Part 4: GraphRAG (Microsoft Research Technique)

def graphrag_query(
    question: str,
    knowledge_graph: IMIKnowledgeGraph,
    vector_store,
    llm_fn: callable,
    embed_fn: callable
) -> dict:
    """
    GraphRAG: combine graph traversal with vector retrieval.
    Best for multi-hop questions requiring relationship context.
    """
    # Step 1: Extract entities from question
    question_entities = extract_entities(question)
    entity_names = [e["text"] for e in question_entities]

    # Step 2: Graph traversal — find connected entities
    graph_context = []
    for entity in entity_names[:3]:  # Top 3 entities from question
        connections = knowledge_graph.find_connections(entity, depth=2)
        if connections:
            conn_text = f"Connections for {entity}: " + "; ".join([
                f"{c['entity']} ({c['hops']} hops via {', '.join(c['relations'])})"
                for c in connections[:10]
            ])
            graph_context.append(conn_text)

    # Step 3: Vector retrieval — find relevant passages
    vector_context = vector_store.query(question, n_results=5)
    vector_texts = [c["text"] for c in vector_context]

    # Step 4: Combine and answer
    combined_context = "\n\n".join([
        "## Graph Context (Entity Relationships):",
        "\n".join(graph_context) if graph_context else "No entity connections found.",
        "\n## Document Context:",
        "\n\n---\n\n".join(vector_texts)
    ])

    prompt = f"""Answer this research question using both the entity relationship context
and the document context provided.

{combined_context}

Question: {question}

Answer (synthesise both graph relationships and document evidence):"""

    return {
        "answer": llm_fn(prompt),
        "graph_entities": entity_names,
        "graph_connections": len(graph_context),
        "vector_chunks": len(vector_texts)
    }

Output Standards

• Use NetworkX for local/dev; Neo4j for production with >10K entities
• Always extract entities before relations — entity quality drives graph quality
• Store source document reference on every edge for provenance
• Use GraphRAG for: multi-hop questions, relationship queries, network analysis
• Use Vector RAG for: direct fact retrieval, semantic similarity, single-document Q&A
• For IMI: build brand-sponsor-segment-event knowledge graph from all research reports

AXE MCP Server Integration

Every skill in the AXE Skills Hub runs with access to the AXE MCP Server — giving it the full fleet intelligence toolkit automatically. No setup required; tools are available in any AXE-powered session.

Core Tools Available

Quick Start

# In any AXE session, tools are pre-loaded. Example chaining:

# 1. Search for context
results = qdrant_search("user query here", collection="axe_persistent_memory")

# 2. Fetch live data if needed
content = web_fetch("https://docs.example.com/api")

# 3. Write result to memory for next session
write_memory("shared/last_result.md", output)

# 4. Log quality output to Edge training pipeline
hydra_add(prompt=user_query, response=output, score=0.9, source="skill-name")

Edge Training Integration

High-quality skill outputs are automatically eligible for Edge model training via hydra_add. When a response scores ≥0.85 in evals, pipe it to the Hydra pipeline to compound Edge's knowledge. This is how skills make Edge smarter over time.

# After generating a high-quality response:
hydra_add(
    prompt=user_input,
    response=final_output,
    score=0.9,          # eval score
    source="skill-name" # tracks provenance
)