Use Cases
🗂 Collection Hierarchy Management
Model deeply nested catalogues, taxonomies, and content libraries as a native graph. Traverse any depth of hierarchy — category → subcategory → asset — in a single Cypher query, with no JOINs and no schema migrations when the structure evolves.
MATCH path = (root:Category {name: "Products"})-[:CONTAINS*]->(asset:Asset)
WHERE asset.status = 'active'
RETURN path🔍 Search Filtering
Augment vector similarity search with graph context. Filter by relationship, ownership, tag, or region inside the same query — so "find documents similar to X that Alice's team owns in APAC" is a single round-trip, not a pipeline.
CALL db.vector.search('Document', $vec, 10) YIELD node, score
WHERE node.region = 'APAC' AND score > 0.80
MATCH (node)-[:OWNED_BY]->(:Team)-[:MEMBER]->(:Person {name: 'Alice'})
RETURN node.title, score
ORDER BY score DESC🔐 Access Management
Replace a separate OPA policy engine and Active Directory integration with a graph that is your permission model. Model Collections, features, roles, and data entitlements as nodes and edges. Evaluate complex permission paths — "can this user access this feature on this data object?" — with a traversal, not a rules engine.
MATCH (u:User {id: $user_id})-[:HAS_ROLE]->(:Role)-[:GRANTS]->(f:Feature {id: $feature_id})
MATCH (f)-[:SCOPED_TO]->(c:Collection)-[:CONTAINS]->(d:DataObject {id: $data_id})
RETURN count(*) > 0 AS has_access🔁 Duplicate Knowledge Artifact Detection
Surface redundant documents, snippets, or knowledge items by computing similarity edges across embeddings and metadata. Cluster near-duplicates with community detection and present deduplication candidates to knowledge managers — no external ML pipeline needed.
MATCH (a:Document)-[r:SIMILAR_TO]->(b:Document)
WHERE r.score > 0.95
RETURN a.title, b.title, r.score
ORDER BY r.score DESC♻️ Process Similarity & Waste Reduction
Map business processes as graph workflows. Run betweenness and similarity traversals to find redundant paths, bottleneck steps, or parallel processes that could be consolidated.
MATCH (p:Process)-[:STEP*]->(s:Step)
WITH p, collect(s.name) AS steps
MATCH (q:Process)-[:STEP*]->(t:Step)
WITH p, q, steps, collect(t.name) AS other_steps
WHERE p.id < q.id AND size(apoc.coll.intersection(steps, other_steps)) > 3
RETURN p.name, q.name🎛 Feature Enablement
Manage feature flags as a first-class graph object. Enable a feature globally, then selectively activate or suppress it per tenant, role, or data collection — all connected to the same access management graph. No separate feature-flag service required.
MATCH (f:Feature {id: $feature_id})
OPTIONAL MATCH (f)-[:OVERRIDDEN_FOR]->(t:Tenant {id: $tenant_id})
RETURN coalesce(t.enabled, f.enabled_globally) AS is_enabled🏛 Regulatory Compliance Graph
Encode HIPAA, SOC 2, ISO 27001, or internal control frameworks as policy nodes linked to data assets and processes. Query "which data objects touch this control?" or "what controls cover this process?" in real time — audit-ready, always current.
MATCH (ctrl:Control {framework: 'SOC2'})-[:GOVERNS]->(asset:DataAsset)
WHERE (asset)-[:PROCESSED_BY]->(:Process {id: $process_id})
RETURN ctrl.id, ctrl.description, asset.name🕵️ Fraud & Anomaly Detection
Stream transaction or event data into the graph and run pattern-matching queries to detect rings, cycles, and anomalous paths in real time.
MATCH (a:Account)-[:SENT]->(t:Transaction)-[:RECEIVED_BY]->(b:Account)
WHERE t.amount > 10000
AND (b)-[:SENT]->(:Transaction)-[:RECEIVED_BY]->(a)
RETURN a.id, b.id, count(t) AS suspicious_transfers
ORDER BY suspicious_transfers DESC🤖 LLM / RAG Grounding with Relationship Context
Go beyond chunk retrieval. When answering a question, traverse the knowledge graph to include related documents, authors, and prior decisions — giving the LLM richer, structured context that flat vector search cannot provide.
CALL db.vector.search('Document', $query_vec, 5) YIELD node, score
MATCH (node)-[:REFERENCES]->(ref:Document)
MATCH (node)-[:AUTHORED_BY]->(author:Person)
OPTIONAL MATCH (node)-[:SUPERSEDES]->(old:Document)
RETURN node.title, ref.title, author.name, old.title, score🏢 Supplier & Vendor Risk Graph
Model your supply chain as a graph: vendors → contracts → products → dependencies. Instantly answer "which of our critical products depend on suppliers in high-risk regions?"
MATCH (p:Product {criticality: 'high'})<-[:SUPPLIES]-(v:Vendor)
WHERE v.risk_region IN ['CN', 'RU', 'IR']
RETURN p.name, v.name, v.risk_region
ORDER BY p.name📊 Knowledge Graph Analytics & Health Monitoring
Run the built-in OLAP analytics engine to understand graph structure: label distribution, edge type counts, degree histograms, density, and connected components.
from purple8_graph import AnalyticsEngine
ax = AnalyticsEngine()
print(ax.label_counts(engine)) # node counts by label
print(ax.density(engine)) # graph completeness ratio
print(ax.connected_components(engine)) # isolated subgraph count🏆 Influence & Authority Scoring
Identify the most influential nodes in any domain — most-cited documents, most-trusted sources, most-connected entities. PageRank runs natively; surface top-K authority nodes in a single call.
result = engine.pagerank(damping=0.85, top_k=10)
for node in result.top_nodes:
print(node.node_id, node.score)🕸 Cluster & Community Discovery
Automatically partition the graph into natural clusters — business units, topic clusters, customer segments, knowledge domains — without manual labelling. The modularity score measures how well-defined the clusters are.
result = engine.detect_communities(seed=42)
print(f"{result.num_communities} communities, modularity Q={result.modularity:.3f}")💼 CRM Deal Intelligence & Renewal Risk
Connect deals, accounts, contacts, open tickets, and contract terms as a live graph. Ask "what are the highest-risk renewals this quarter?" in one query — no SQL joins, no BI dashboard export.
MATCH (opp:Opportunity {stage: 'negotiation'})-[:BELONGS_TO]->(acct:Account)
MATCH (acct)<-[:PRIMARY_CONTACT]-(csm:Contact)
OPTIONAL MATCH (acct)-[:HAS_TICKET]->(t:SupportTicket {status: 'open', priority: 'high'})
OPTIONAL MATCH (acct)-[:HAS_CONTRACT]->(c:Contract {auto_renew: false})
WITH opp, acct, csm, count(t) AS open_critical_tickets, c
WHERE open_critical_tickets > 0 OR c IS NOT NULL
RETURN
opp.name AS opportunity,
opp.value AS value,
opp.close_date AS close_date,
acct.name AS account,
csm.name AS owner,
open_critical_tickets,
c.renewal_date AS renewal_date
ORDER BY opp.value DESCFeed each row into an LLM to auto-generate churn risk summaries and recommended next actions, with no external data pipeline.
→ Real-time Augmented AI guide
🏦 Financial Risk Graph — Counterparty Exposure
Map accounts, transactions, counterparties, and jurisdictions as a graph. Traverse exposure chains at arbitrary depth to answer "how much indirect exposure do we have to any entity linked to Counterparty X?"
MATCH path = (our:Institution {id: 'us'})-[:EXPOSED_TO*1..4]->(risky:Counterparty {risk_flag: true})
WITH path, relationships(path) AS chain
UNWIND chain AS edge
WITH path, sum(edge.notional_usd) AS total_exposure, length(path) AS hops
WHERE total_exposure > 1000000
RETURN
[n IN nodes(path) | n.name] AS exposure_chain,
total_exposure,
hops
ORDER BY total_exposure DESC
LIMIT 20No fixed-depth JOIN can do this. Purple8 traverses the full chain regardless of depth.
👥 HR Org-Chart AI Assistant
Model your org chart — employees, managers, teams, roles, projects — as a graph. Let employees ask natural-language questions about the organisation and get answers grounded in the live HR system.
# User asks: "Who on Alice's team has Python skills and is available this quarter?"
query_vec = embed("Python skills available Q3")
results = engine.query("""
CALL db.vector.search('Employee', $vec, 20) YIELD node, score
MATCH (node)-[:REPORTS_TO*1..3]->(manager:Employee {name: 'Alice'})
WHERE node.availability = 'Q3-2026'
AND 'Python' IN node.skills
RETURN
node.name AS employee,
node.role AS role,
node.team AS team,
score
ORDER BY score DESC
LIMIT 5
""", vec=query_vec)Combine with the Journey Engine to auto-surface headcount gaps, flight risks, or promotion readiness when employee records change.
🚢 Supply Chain Disruption Alerts
Model suppliers, products, contracts, and risk regions as a graph. When a geopolitical event flags a region, instantly traverse the supply chain to find every affected product, contract, and customer — then trigger an AI-generated briefing.
-- Flag a supplier as disrupted
MATCH (s:Supplier {region: 'conflict-zone'})
SET s.disrupted = true
-- Find all downstream impact
MATCH (s:Supplier {disrupted: true})-[:SUPPLIES]->(p:Product)
MATCH (p)<-[:DEPENDS_ON]-(c:Contract)
MATCH (c)-[:HELD_BY]->(customer:Account)
RETURN
s.name AS disrupted_supplier,
p.name AS product,
p.criticality AS criticality,
c.value AS contract_value,
customer.name AS affected_customer
ORDER BY p.criticality DESC, c.value DESCThen pass each row into an LLM to generate impact summaries for procurement and account teams — all in real time.
📋 Regulatory Compliance Q&A
Connect controls, data assets, processes, systems, and owners as a graph. Let your compliance team ask questions in plain English and get answers grounded in the live control framework — not a static spreadsheet.
# Auditor asks: "Which unencrypted data assets fall under SOC 2 CC6?"
query_vec = embed("unencrypted data assets SOC 2 CC6 access control")
results = engine.query("""
CALL db.vector.search('DataAsset', $vec, 10) YIELD node, score
MATCH (ctrl:Control {framework: 'SOC2', id: 'CC6'})-[:GOVERNS]->(node)
WHERE node.encrypted = false
MATCH (node)-[:OWNED_BY]->(owner:Team)
OPTIONAL MATCH (node)-[:PROCESSED_BY]->(proc:Process)
RETURN
node.name AS data_asset,
owner.name AS owner_team,
proc.name AS process,
node.last_audit AS last_audit,
score
ORDER BY score DESC
""", vec=query_vec)The LLM gets a structured, citation-ready answer — not a hallucination from training data. The graph is always current because your compliance team updates it directly.
→ Regulatory Compliance Graph use case → Real-time Augmented AI guide
🤝 AI Agent Tool Access via MCP
Give any MCP-compatible agent — Claude, Cursor, or a custom LLM loop — direct, authenticated access to your knowledge graph as a structured toolset. No custom API wrappers, no prompt hacks. The agent calls hybrid_search, rag_query, traverse, ingest_text, and journey_start as first-class tools, exactly like a developer would via the REST API.
# Install the MCP adapter
pip install "purple8-graph[mcp]"
# Start the server — Claude Desktop, Cursor, and custom agents connect via stdio
purple8-graph mcp-server \
--url http://localhost:8010 \
--api-key YOUR_API_KEYClaude Desktop config (~/.claude/claude_desktop_config.json):
{
"mcpServers": {
"purple8": {
"command": "purple8-graph",
"args": ["mcp-server", "--url", "http://localhost:8010", "--api-key", "YOUR_API_KEY"]
}
}
}Once connected, Claude can answer questions like "who are the most connected suppliers in our risk graph?" by running pagerank directly, or ingest a new PDF by calling ingest_text — with zero prompt engineering on your side.
# Custom agent using the MCP Python SDK
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
server_params = StdioServerParameters(
command="purple8-graph",
args=["mcp-server", "--url", "http://localhost:8010", "--api-key", "YOUR_KEY"],
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# Agent finds at-risk accounts, then checks their journey status
accounts = await session.call_tool("hybrid_search", {
"query_text": "enterprise accounts with declining usage",
"label": "Account", "k": 5,
})
for acct in accounts.content:
status = await session.call_tool("journey_status", {
"instance_id": acct["journey_instance_id"],
})
print(status.content)Every tool call is authenticated and logged. The agent never gets raw database access — it goes through the same JWT/API-key layer as your application code.
🧠 Graph Memory & Learning Loop
Every AI recommendation, every human override, and every journey outcome is written as an immutable edge in the graph. Query this accumulated history to understand where the AI is right, where it's overridden, and what patterns lead to success — then feed those patterns back as context on the next decision.
-- What did the AI recommend for this opportunity, and what actually happened?
MATCH (ji:JourneyInstance {id: "ji:9871-abc"})-[ai:AI_ADVISED]->(s:Stage)
OPTIONAL MATCH (ji)-[ht:HITL_TASK]->(t:HITLTask)
RETURN
ai.timestamp AS ai_decision_time,
ai.action_type AS ai_recommended,
ai.confidence AS confidence,
ai.reasoning AS reasoning,
t.decision AS human_decision,
t.rationale AS override_reason
ORDER BY ai.timestamp-- Which journey types have the AI overridden most often?
MATCH (ji:JourneyInstance)-[:AI_ADVISED]->(s)
OPTIONAL MATCH (ji)-[:HITL_TASK]->(t:HITLTask {decision: "reject"})
WITH ji.journey_type AS jtype, count(s) AS ai_calls, count(t) AS overrides
RETURN jtype,
ai_calls,
overrides,
round(100.0 * overrides / ai_calls, 1) AS override_pct
ORDER BY override_pct DESCWrite outcomes back when journeys close and the graph becomes a continuously improving signal store:
from purple8_graph import GraphEngine
engine = GraphEngine(data_dir="./data")
# Record the outcome when a deal closes
engine.add_edge(
src_id="ji:9871-abc",
dst_id="ji:9871-abc",
edge_type="OUTCOME",
properties={
"result": "won",
"closed_at": "2026-03-25T16:00:00Z",
"revenue": 2_000_000,
"cycle_days": 47,
},
)
# Query closed-won patterns and feed them as few-shot context
patterns = engine.query("""
MATCH (ji:JourneyInstance {journey_type: $jtype})-[o:OUTCOME {result: 'won'}]->()
MATCH (ji)-[ai:AI_ADVISED]->(s)
RETURN ai.action_type, ai.recommended, avg(o.cycle_days) AS avg_days
ORDER BY avg_days ASC LIMIT 5
""", {"jtype": "sales-cycle"})
recommendation = await advisor.advise(
instance=current_instance,
audit_history=history,
few_shot_patterns=patterns, # <-- graph memory as LLM context
)Subscribe to the CDC EventBus to react to every AI decision in real time — trigger Slack alerts, update dashboards, or kick off downstream workflows the moment an AI_ADVISED edge lands:
from purple8_graph.cdc import CDCEmitter, EventBus, EventType
bus = EventBus()
emitter = CDCEmitter(engine, bus, tenant_id="acme")
async with bus.subscribe(tenant_id="acme") as queue:
while True:
event = await queue.get()
if event.event_type == EventType.EDGE_ADDED:
if event.properties.get("edge_type") == "AI_ADVISED":
await post_to_slack(event.properties)