Purple8 Hyper Graph

Appearance

Real-time Augmented AI with Structured Data ​

Most AI retrieval systems are built for unstructured content — PDFs, notes, chat history. But the most valuable data in any organisation is structured: CRM records, ERP transactions, contracts, tickets, financial accounts, HR systems.

Purple8 Hyper Graph bridges that gap. It lets you model structured data as a connected graph and query it with hybrid search — so your AI gets relational, up-to-date context with a single round-trip, not a pipeline of JOINs and vector calls.

The core idea ​

Traditional RAG:

User question → embed → vector search → top-k chunks → LLM

Graph-augmented AI with structured data:

User question → embed → Cypher (vector + relationships + filters) → structured context row → LLM

The difference: the LLM sees not just what a record says, but who owns it, what it's connected to, and what state it's in right now.

Step 1 — Model your structured data as a graph ​

Any structured record becomes a node. Foreign keys become edges.

python
from purple8_graph import GraphEngine

engine = GraphEngine("./data")

# CRM records — add_node(node_id, labels, properties)
engine.add_node("acct-001", ["Account"], {"name": "Acme Corp",   "arr": 240000, "tier": "enterprise", "industry": "manufacturing"})
engine.add_node("ctt-001",  ["Contact"], {"name": "Jane Smith",  "role": "CFO", "email": "jane@acme.com"})
engine.add_node("opp-001", ["Opportunity"], {"name": "Renewal Q3", "value": 120000, "stage": "negotiation", "close_date": "2026-09-30"})
engine.add_node("tkt-001", ["SupportTicket"], {"subject": "API timeout errors", "priority": "high", "status": "open"})
engine.add_node("ctr-001", ["Contract"], {"start": "2025-10-01", "renewal_date": "2026-09-30", "auto_renew": False})

# Relationships encode the structure that SQL can't traverse in one query
# add_edge(edge_id, edge_type, source_id, target_id)
engine.add_edge("opp-001_belongs_to_acct-001", "BELONGS_TO", "opp-001", "acct-001")
engine.add_edge("acct-001_contact_ctt-001", "PRIMARY_CONTACT", "acct-001", "ctt-001")
engine.add_edge("acct-001_ticket_tkt-001", "HAS_TICKET", "acct-001", "tkt-001")
engine.add_edge("acct-001_contract_ctr-001", "HAS_CONTRACT", "acct-001", "ctr-001")

No schema required

You don't need to define a schema before ingesting. Purple8 infers it from the data. Add new record types — a new ERP object, a new CRM field — without any DDL or migration.

Step 2 — Retrieve structured context with one Cypher query ​

At query time, hybrid Cypher combines vector similarity with relationship traversal and property filters — in a single round-trip.

python
import openai

user_question = "What's the renewal risk on Acme Corp?"
query_vec = openai.embeddings.create(input=user_question, model="text-embedding-3-small").data[0].embedding

results = engine.query("""
    CALL db.vector.search('Opportunity', $vec, 5) YIELD node, score
    WHERE score > 0.70
    MATCH (node)-[:BELONGS_TO]->(acct:Account)
    MATCH (acct)<-[:PRIMARY_CONTACT]-(contact:Contact)
    OPTIONAL MATCH (acct)-[:HAS_TICKET]->(t:SupportTicket {status: 'open'})
    OPTIONAL MATCH (acct)-[:HAS_CONTRACT]->(c:Contract)
    RETURN
        node.name          AS opportunity,
        node.stage         AS stage,
        node.value         AS value,
        node.close_date    AS close_date,
        acct.name          AS account,
        acct.arr           AS arr,
        acct.tier          AS tier,
        contact.name       AS owner,
        contact.email      AS owner_email,
        count(t)           AS open_tickets,
        c.renewal_date     AS renewal_date,
        c.auto_renew       AS auto_renew,
        score
    ORDER BY score DESC
    LIMIT 1
""", vec=query_vec)

One query. Zero application-side joins.

Step 3 — Inject structured context into the LLM ​

python
row = results[0]

system_prompt = "You are a revenue intelligence assistant. Answer based only on the structured context provided."

user_prompt = f"""
Structured context:
- Opportunity: {row['opportunity']} (stage: {row['stage']}, value: ${row['value']:,}, closes: {row['close_date']})
- Account: {row['account']} ({row['tier']} tier, ARR: ${row['arr']:,})
- Primary contact: {row['owner']} <{row['owner_email']}>
- Open support tickets: {row['open_tickets']}
- Contract renewal: {row['renewal_date']} (auto-renew: {'yes' if row['auto_renew'] else 'no'})

Question: {user_question}
"""

response = openai.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": system_prompt},
        {"role": "user",   "content": user_prompt},
    ]
)
print(response.choices[0].message.content)

Step 4 — Keep context fresh with real-time graph updates ​

The graph is your source of truth. Update records as they change in your source system — the next AI query automatically gets the latest state.

python
# A deal just moved stage — update the graph (properties is the 3rd positional arg)
engine.update_node("opp-001", properties={"stage": "closed-won", "close_date": "2026-07-15"})

# A new ticket opened
engine.add_node("tkt-002", ["SupportTicket"], {"subject": "Data export failure", "priority": "critical", "status": "open"})
engine.add_edge("acct-001_ticket_tkt-002", "HAS_TICKET", "acct-001", "tkt-002")

No cache invalidation. No pipeline re-run. The next vector search returns the updated context.

Step 5 — Trigger AI workflows automatically on SLA breach or stage entry ​

Use the Journey Engine to fire AI analysis whenever a journey instance enters a stage or breaches its SLA — without polling. Journeys track discrete instances (e.g. one Opportunity per renewal cycle) through a defined stage list; the AI advisor and HITL gates are configured per stage, not as arbitrary property-change conditions on graph nodes.

python
from purple8_graph import GraphEngine
from purple8_graph.journey import (
    JourneyEngine, JourneyAIAdvisor, StageSpec, SLAPolicy, AIStepConfig, AIStepTrigger,
)
from purple8_graph.genai import LiteLLMProvider

je = JourneyEngine(engine)

je.define_journey("renewal-risk-check", stages=[
    StageSpec(
        name="negotiation",
        sla=SLAPolicy(warn_after_seconds=3600 * 24 * 25, breach_after_seconds=3600 * 24 * 30),
        ai_step=AIStepConfig(trigger=AIStepTrigger.ON_SLA_WARN),
    ),
    StageSpec(
        name="human-review",
        hitl={"required": True, "assigned_role": "csm", "title": "Renewal risk review"},
    ),
])

provider = LiteLLMProvider(model="gpt-4o")
je.register_ai_advisor(JourneyAIAdvisor(je, provider))

# Start an instance tied to the Opportunity node
instance = je.start("renewal-risk-check", entity_id="opp-001")

When the negotiation stage's SLA warning fires (25 days in, per SLAPolicy.warn_after_seconds), SLAMonitor automatically invokes the registered JourneyAIAdvisor, which writes an AI_ADVISED edge with its recommendation. In v0.72.0, every AI-triggered transition or SLA event uses the durable Journey AI outbox (ai_dispatched flag on the audit edge + je.replay_pending_ai_dispatches() on crash recovery), ensuring advisor triggers survive abrupt process crashes (SIGKILL). Advancing the instance into human-review automatically creates a HITLTask for a CSM — all with a full audit trail.

→ Journey Engine guide

SQL JOIN + vectorPurple8 Hyper Graph
Round-trips per query2–41
Arbitrary relationship depth❌✅
Schema migrations for new objectsRequiredNot required
Real-time updates in contextCache invalidationDirect node update
AI workflow triggers on data changeExternal job/eventBuilt-in Journey Engine
Hybrid keyword + semantic + graph❌✅

More use cases ​

See the Use Cases page for working Cypher examples across:

Further reading ​

Purple8 Hyper Graph v0.72.0 · AI-Native Execution Infrastructure · NVIDIA Inception Member · Red Hat Technology Partner.

© 2026 Purple8, Inc. All rights reserved.