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-hyper-graph[mcp]"
# Start the server — Claude Desktop, Cursor, and custom agents connect via stdio
purple8-hyper-graph mcp-server \
--url http://localhost:8100 \
--api-key YOUR_API_KEYClaude Desktop config (~/.claude/claude_desktop_config.json):
{
"mcpServers": {
"purple8": {
"command": "purple8-hyper-graph",
"args": ["mcp-server", "--url", "http://localhost:8100", "--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-hyper-graph",
args=["mcp-server", "--url", "http://localhost:8100", "--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)→ Memory & Learning guide → Journey Engine guide
🏦 Customer Onboarding Across Departments and Systems
A customer signs up. What follows touches six departments and four systems — CRM, KYC vendor, core banking mainframe, product provisioning service. Until now, no single system knew the full story.
Purple8's Journey Engine becomes the single source of truth. Every system — from a legacy mainframe to a modern microservice — makes one call when its step completes. Every handoff, every human decision, every AI recommendation, every SLA breach is an immutable graph edge, queryable forever.
from purple8_graph.journey import JourneyEngine, StageSpec, SLAPolicy
je = JourneyEngine(engine)
je.define_journey(
journey_type="customer_onboarding",
stages=[
StageSpec("application_submitted", owner_system="CRMService"),
StageSpec("kyc_screening", owner_system="KYCVendorAPI",
sla=SLAPolicy(warn_after_seconds=3600, breach_after_seconds=86400)),
StageSpec("manual_review", owner_system="ComplianceTeam",
requires_human=True),
StageSpec("account_provisioning", owner_system="CoreBankingMainframe"),
StageSpec("products_assigned", owner_system="ProductCatalogue"),
StageSpec("welcome_sent", owner_system="NotificationsEngine"),
StageSpec("onboarding_complete"),
],
)Modern systems advance the journey over REST. Legacy systems that can't make outbound calls are polled by Purple8:
from purple8_graph.connectors import RESTPoller, PollerConfig
# Purple8 polls the mainframe — the mainframe never knows Purple8 exists
poller = RESTPoller(je, PollerConfig(
name="CoreBankingMainframe",
url="http://mainframe-adapter.internal/onboarding/status",
poll_interval_seconds=30,
))
poller.start()Every question about any customer becomes a Cypher query:
-- Full interaction timeline
MATCH (i:JourneyInstance {entity_id: "C-8812"})-[t:ADVANCED_TO]->()
RETURN t.from_stage, t.to_stage, t.actor, t.timestamp, t.notes
ORDER BY t.timestamp
-- All customers breaching KYC SLA right now
MATCH (i:JourneyInstance {journey_type: "customer_onboarding",
current_stage: "kyc_screening",
sla_status: "BREACHED"})
RETURN i.entity_id, i.entered_at
ORDER BY i.entered_at→ Customer Onboarding guide → Journey Engine guide
🏥 Healthcare — Patient Journey & Care Coordination
A patient is referred by their GP. What follows touches a referral management system, a scheduling platform, a clinical EMR, a lab system, a radiology PACS, an insurance prior-auth portal, a pharmacy system, and a discharge planner — often across different hospitals, different vendors, different EHR instances.
Each system has a fragment of the patient's story. None has the whole picture. Delays compound invisibly. A prior-auth sits pending for 11 days. Nobody flags it. The patient deteriorates waiting for a procedure that was technically approved on day 3.
Purple8 makes the full care journey a graph. Every clinical event, every handoff, every pending approval, every SLA breach is an immutable edge — visible in real time, queryable by any clinician or care coordinator with access.
je.define_journey(
journey_type="patient_care_pathway",
stages=[
StageSpec("referral_received", owner_system="ReferralManagement"),
StageSpec("triage_assessed", owner_system="ClinicalTeam",
sla=SLAPolicy(warn_after_seconds=3600, breach_after_seconds=86400)),
StageSpec("prior_auth_submitted", owner_system="InsurancePortal",
sla=SLAPolicy(warn_after_seconds=86400, breach_after_seconds=345600)), # 4 days
StageSpec("prior_auth_approved", owner_system="InsurancePortal"),
StageSpec("procedure_scheduled", owner_system="SchedulingSystem"),
StageSpec("procedure_completed", owner_system="ClinicalEMR"),
StageSpec("results_available", owner_system="LaboratorySystem"),
StageSpec("care_plan_updated", owner_system="ClinicalEMR",
requires_human=True), # clinician must review before discharge
StageSpec("discharged"),
StageSpec("follow_up_booked"),
],
)The prior-auth portal can't make outbound HTTP calls. Purple8 polls it:
poller = RESTPoller(je, PollerConfig(
name="InsurancePortal",
url="https://payer-api.healthplan.internal/authorizations/status",
poll_interval_seconds=300, # check every 5 minutes
headers={"Authorization": "Bearer $PAYER_API_TOKEN"},
))
poller.start()Every blindspot that currently kills patients or wastes billions is now a Cypher query:
-- All patients with prior auth pending for more than 4 days
MATCH (i:JourneyInstance {journey_type: "patient_care_pathway",
current_stage: "prior_auth_submitted",
sla_status: "BREACHED"})
RETURN i.entity_id AS patient_id, i.entered_at, i.sla_status
ORDER BY i.entered_at
-- Average time from referral to procedure by hospital department
MATCH (i:JourneyInstance {journey_type: "patient_care_pathway"})
-[t:ADVANCED_TO {to_stage: "procedure_completed"}]->()
RETURN t.actor AS department,
avg(duration.inSeconds(i.started_at, t.timestamp)) / 86400 AS avg_days,
count(*) AS volume
ORDER BY avg_days DESC
-- Patients who fell through the cracks — referral received, no movement in 72 hours
MATCH (i:JourneyInstance {journey_type: "patient_care_pathway",
current_stage: "referral_received"})
WHERE i.updated_at < datetime() - duration({hours: 72})
RETURN i.entity_id, i.started_at
ORDER BY i.started_at
-- Full care timeline for a patient (complete audit for clinician handoff)
MATCH (i:JourneyInstance {entity_id: "P-44821"})-[t:ADVANCED_TO]->()
RETURN t.from_stage, t.to_stage, t.actor, t.timestamp, t.notes
ORDER BY t.timestampNo ETL. No data warehouse job running overnight. The graph is the care record.
🏭 Manufacturing — Production Order & Quality Traceability
A production order flows from planning through procurement, shop floor machining, sub-assembly, quality inspection, final assembly, packaging, and despatch. Each step touches a different system — ERP, MES, SCADA, QMS, WMS, supplier portals. A defect is found in the field. The question is: which batch number, which machine, which operator, which raw material lot, which supplier?
Without a connected graph, the answer takes days of manual triage across five systems. With Purple8, the answer is a Cypher query that runs in milliseconds.
je.define_journey(
journey_type="production_order",
stages=[
StageSpec("order_released", owner_system="ERP"),
StageSpec("materials_kitted", owner_system="WarehouseSystem",
sla=SLAPolicy(warn_after_seconds=7200, breach_after_seconds=28800)),
StageSpec("machining_complete", owner_system="MES"),
StageSpec("in_process_inspection", owner_system="QMS",
requires_human=True,
sla=SLAPolicy(warn_after_seconds=3600, breach_after_seconds=14400)),
StageSpec("sub_assembly_complete", owner_system="MES"),
StageSpec("final_inspection_passed", owner_system="QMS",
requires_human=True),
StageSpec("serialised_and_labelled", owner_system="WMS"),
StageSpec("despatched", owner_system="LogisticsSystem"),
StageSpec("on_hold_quality"), # rejected at any inspection point
StageSpec("scrapped"),
],
)Legacy SCADA and MES systems that cannot make outbound calls are wired via the poller. Modern ERP webhooks fire directly:
# ERP fires a webhook when an order is released
receiver = WebhookReceiver(je, engine)
receiver.register_connector("ERP", secret="erp-webhook-secret")
# SCADA cannot push — Purple8 polls the MES adapter
poller = RESTPoller(je, PollerConfig(
name="MES",
url="http://mes-api.plant.internal/orders/status",
poll_interval_seconds=60,
))
poller.start()Traceability that used to take days of manual investigation:
-- Full production history for a serial number (field defect investigation)
MATCH (i:JourneyInstance {entity_id: "SN-998821", journey_type: "production_order"})
-[t:ADVANCED_TO]->()
RETURN t.from_stage, t.to_stage, t.actor, t.timestamp, t.notes
ORDER BY t.timestamp
-- Which machine operator performed the last in-process inspection on this batch?
MATCH (i:JourneyInstance {entity_id: "BATCH-44A"})
-[t:ADVANCED_TO {to_stage: "in_process_inspection"}]->()
RETURN t.actor AS inspector, t.timestamp, t.notes
-- All orders currently on hold for quality — with time in hold
MATCH (i:JourneyInstance {journey_type: "production_order",
current_stage: "on_hold_quality"})
RETURN i.entity_id AS order_id,
i.updated_at AS held_since,
duration.inSeconds(i.updated_at, datetime()) / 3600 AS hours_in_hold
ORDER BY hours_in_hold DESC
-- Scrap rate by machine (which asset is causing the most quality failures?)
MATCH (i:JourneyInstance {journey_type: "production_order"})
-[t:ADVANCED_TO {to_stage: "scrapped"}]->()
MATCH (i)-[m:ADVANCED_TO {from_stage: "machining_complete"}]->()
RETURN m.actor AS machine_id,
count(*) AS scrap_count
ORDER BY scrap_count DESC
-- SLA breach leaderboard — which production stage is the biggest bottleneck?
MATCH ()-[b:SLA_BREACHED]->(s)
WHERE b.journey_type = "production_order"
RETURN s.stage_name AS stage,
count(*) AS breach_count
ORDER BY breach_count DESCWhen a regulator asks for a complete manufacturing record — materials, operators, machines, inspections, decisions — the answer is a single graph query, not a three-week audit exercise.
🏛 Insurance — Claims Lifecycle & Fraud Network Detection
A claim is filed. It touches a FNOL system, a reserve management platform, a field adjuster app, a repair estimator, a legal case management system, a payment system, and a fraud detection service — often across outsourced partners. Adjusters change. Supervisors escalate. SLAs are breached silently.
Meanwhile, the fraud team is working with a completely separate dataset, unable to see that three claims in this month all share the same body shop, the same repairer, and the same claimant solicitor.
Purple8 solves both: the operational blindspot and the fraud network problem in one graph.
je.define_journey(
journey_type="insurance_claim",
stages=[
StageSpec("fnol_received", owner_system="FNOLSystem"),
StageSpec("reserve_set", owner_system="ReserveManagement",
sla=SLAPolicy(warn_after_seconds=3600, breach_after_seconds=86400)),
StageSpec("adjuster_assigned", owner_system="ClaimsWorkbench"),
StageSpec("field_inspection", owner_system="FieldAdjusterApp",
requires_human=True,
sla=SLAPolicy(warn_after_seconds=172800, breach_after_seconds=432000)),
StageSpec("estimate_approved", owner_system="EstimatingSystem"),
StageSpec("legal_review", owner_system="LegalCaseManagement",
requires_human=True),
StageSpec("payment_authorised", owner_system="PaymentSystem"),
StageSpec("closed_paid"),
StageSpec("closed_declined"),
StageSpec("referred_to_siu"), # Special Investigations Unit
],
)Fraud network detection — the problem no other database solves cleanly:
-- Claims sharing the same repairer and solicitor this month (ring indicator)
MATCH (c1:JourneyInstance {journey_type: "insurance_claim"})
-[t1:ADVANCED_TO {to_stage: "estimate_approved"}]->(n1)
MATCH (c2:JourneyInstance {journey_type: "insurance_claim"})
-[t2:ADVANCED_TO {to_stage: "estimate_approved"}]->(n2)
WHERE t1.actor = t2.actor -- same repairer
AND c1.entity_id <> c2.entity_id
AND t1.timestamp >= "2026-03-01T00:00:00Z"
MATCH (c1)-[l1:ADVANCED_TO {to_stage: "legal_review"}]->()
MATCH (c2)-[l2:ADVANCED_TO {to_stage: "legal_review"}]->()
WHERE l1.actor = l2.actor -- same solicitor
RETURN c1.entity_id, c2.entity_id,
t1.actor AS repairer, l1.actor AS solicitor
ORDER BY repairer
-- Average days from FNOL to payment by adjuster (performance view)
MATCH (i:JourneyInstance {journey_type: "insurance_claim",
current_stage: "closed_paid"})
-[a:ADVANCED_TO {to_stage: "adjuster_assigned"}]->()
RETURN a.actor AS adjuster,
avg(duration.inSeconds(i.started_at, i.completed_at)) / 86400 AS avg_days,
count(*) AS claims_handled
ORDER BY avg_days
-- Claims where adjuster changed mid-lifecycle (handoff risk)
MATCH (i:JourneyInstance {journey_type: "insurance_claim"})
-[t:ADVANCED_TO {to_stage: "adjuster_assigned"}]->()
WITH i, collect({adjuster: t.actor, at: t.timestamp}) AS assignments
WHERE size(assignments) > 1
RETURN i.entity_id, assignments
-- All claims breaching SLA right now with adjuster identity
MATCH (i:JourneyInstance {journey_type: "insurance_claim",
sla_status: "BREACHED"})
-[a:ADVANCED_TO {to_stage: "adjuster_assigned"}]->()
RETURN i.entity_id, i.current_stage, a.actor AS adjuster, i.updated_at
ORDER BY i.updated_at🏛 Government & Public Services — Citizen Service Delivery
A citizen applies for a building permit. It touches a planning department, an environmental review team, a highways authority, a utilities sign-off process, a legal records team, and a councillor approval step — often across different agencies, different legacy systems, different paper-based workflows still being digitised.
The same problem exists for benefits processing, social care referrals, business license applications, and grant disbursements. Every department knows its fragment. The citizen has no visibility. Delays are invisible to anyone not in that specific queue.
je.define_journey(
journey_type="building_permit",
stages=[
StageSpec("application_received", owner_system="PlanningPortal"),
StageSpec("validation_check", owner_system="PlanningDept",
sla=SLAPolicy(warn_after_seconds=86400, breach_after_seconds=259200)),
StageSpec("environmental_review", owner_system="EnvironmentalAgency",
sla=SLAPolicy(warn_after_seconds=604800, breach_after_seconds=1209600)), # 2 weeks
StageSpec("highways_assessment", owner_system="HighwaysAuthority",
requires_human=True,
sla=SLAPolicy(warn_after_seconds=604800, breach_after_seconds=1209600)),
StageSpec("legal_review", owner_system="LegalRecords",
requires_human=True),
StageSpec("committee_decision", owner_system="PlanningCommittee",
requires_human=True),
StageSpec("approved_conditions_set", owner_system="PlanningDept"),
StageSpec("permit_issued", owner_system="PlanningPortal"),
StageSpec("refused"),
StageSpec("appealed"),
],
)Legacy agency systems that can't push events are polled. The citizen-facing portal fires webhooks:
# Planning portal fires webhook on new application
receiver.register_connector("PlanningPortal", secret="portal-hmac-secret")
# Highways Authority runs on a 15-year-old system with a read API
poller = RESTPoller(je, PollerConfig(
name="HighwaysAuthority",
url="http://highways-legacy.council.gov.uk/api/reviews/status",
poll_interval_seconds=600, # every 10 minutes
))
poller.start()The accountability and transparency queries that governments are mandated to answer — but currently can't without weeks of manual triage:
-- All applications breaching statutory determination period (legal obligation)
MATCH (i:JourneyInstance {journey_type: "building_permit",
sla_status: "BREACHED"})
RETURN i.entity_id AS application_ref,
i.current_stage,
i.updated_at AS last_activity,
duration.inSeconds(i.started_at, datetime()) / 86400 AS days_elapsed
ORDER BY days_elapsed DESC
-- Which agency is the biggest bottleneck in permit processing?
MATCH ()-[b:SLA_BREACHED]->(s)
WHERE b.journey_type = "building_permit"
RETURN s.owner_system AS agency,
count(*) AS breaches
ORDER BY breaches DESC
-- Full audit trail for a specific application (FOI response)
MATCH (i:JourneyInstance {entity_id: "APP-2026-8821"})-[t:ADVANCED_TO]->()
RETURN t.from_stage, t.to_stage, t.actor, t.timestamp, t.notes
ORDER BY t.timestamp
-- Applications with no activity in any stage for more than 30 days
MATCH (i:JourneyInstance {journey_type: "building_permit"})
WHERE i.updated_at < datetime() - duration({days: 30})
AND i.current_stage NOT IN ["permit_issued", "refused", "appealed"]
RETURN i.entity_id, i.current_stage, i.updated_at
ORDER BY i.updated_at
-- Average processing time per agency department
MATCH (i:JourneyInstance {journey_type: "building_permit",
current_stage: "permit_issued"})
-[t:ADVANCED_TO]->()
RETURN t.actor AS department,
avg(duration.inSeconds(t.entered_at, t.timestamp)) / 86400 AS avg_days,
count(*) AS volume
ORDER BY avg_days DESCEvery Freedom of Information request about a specific application. Every statutory reporting deadline. Every ministerial inquiry about processing times. All of it is a Cypher query — not a six-week manual extraction exercise.
🔬 Pharmaceutical & Clinical Trials — Patient Recruitment & Regulatory Traceability
A clinical trial recruits patients across 40 sites, 12 countries, and 3 CROs. A patient is screened, consented, randomised, dosed, and followed up over 24 months. Every protocol deviation, every adverse event, every lab result, every dosing change, and every site staff change that touched the patient record must be traceable for regulatory submission — with a complete audit trail showing who did what, when, and why.
Without a connected graph, the eTMF, EDC, CTMS, safety database, and site management systems each hold a fragment. Reconciling them for an FDA submission takes months.
je.define_journey(
journey_type="clinical_trial_patient",
stages=[
StageSpec("pre_screened", owner_system="CTMS"),
StageSpec("informed_consent_signed", owner_system="EDC",
requires_human=True), # site coordinator must confirm
StageSpec("screening_complete", owner_system="EDC",
sla=SLAPolicy(warn_after_seconds=604800, breach_after_seconds=1209600)),
StageSpec("randomised", owner_system="RandomisationSystem"),
StageSpec("first_dose_administered", owner_system="EDC",
requires_human=True),
StageSpec("on_treatment", owner_system="SafetyDatabase"),
StageSpec("adverse_event_reported", owner_system="SafetyDatabase",
sla=SLAPolicy(warn_after_seconds=86400, breach_after_seconds=259200)), # 72hr SAE reporting
StageSpec("completed_protocol", owner_system="EDC"),
StageSpec("early_termination"),
StageSpec("lost_to_followup"),
],
)-- Complete audit trail for a patient (regulatory submission)
MATCH (i:JourneyInstance {entity_id: "SUBJ-044", journey_type: "clinical_trial_patient"})
-[t:ADVANCED_TO|AI_ADVISED|HITL_RESOLVED]->()
RETURN t.from_stage, t.to_stage, t.actor, t.timestamp, t.notes
ORDER BY t.timestamp
-- Protocol deviations by site (site performance for regulatory report)
MATCH (i:JourneyInstance {journey_type: "clinical_trial_patient"})
WHERE i.metadata.site_id IS NOT NULL
MATCH (i)-[t:ADVANCED_TO]->()
WHERE t.notes CONTAINS "protocol deviation"
RETURN i.metadata.site_id AS site,
count(*) AS deviations
ORDER BY deviations DESC
-- Serious adverse events not escalated within 72 hours (regulatory breach)
MATCH (i:JourneyInstance {journey_type: "clinical_trial_patient",
current_stage: "adverse_event_reported",
sla_status: "BREACHED"})
RETURN i.entity_id AS subject_id,
i.metadata.site_id AS site,
i.updated_at AS sae_reported_at
ORDER BY i.updated_at
-- All staff who ever touched a specific patient's record (access audit)
MATCH (i:JourneyInstance {entity_id: "SUBJ-044"})-[t:ADVANCED_TO|HITL_RESOLVED]->()
RETURN DISTINCT t.actor AS staff_member, collect(t.to_stage) AS actionsThe same pattern applies to device vigilance, post-market surveillance, and pharmacovigilance signal detection across product lineage graphs.
Every industry has the same underlying problem: operational data scattered across systems, assembled manually only when something goes wrong. Purple8's Journey Engine makes the graph the operational record — connected, live, and queryable. The domain changes. The graph structure does not.