Purple8 Graph

Appearance

Graph as Memory ​

Purple8 does not have a separate "memory layer" or "learning engine" component. There is no MemoryLayer class to instantiate and no background process that trains on your data.

What exists — and what makes this pattern powerful — is simpler: every AI decision, every human override, and every stage transition is written into the graph as an immutable edge. The graph is the memory. You query it with Cypher, the same way you query everything else.

This page explains exactly what gets written, how to read it back, and how to feed it forward as context so the AI makes better decisions over time.

What gets written automatically ​

AI_ADVISED edges ​

Every time JourneyAIAdvisor.advise() runs, it appends an AI_ADVISED edge from the journey instance to the recommended stage:

(JourneyInstance) -[:AI_ADVISED]-> (Stage)

Properties written:
  action_type   "advance" | "hold" | "escalate"
  recommended   "contract-review"
  reasoning     "3 of 4 stakeholders signed; legal SLA at 85%"
  confidence    0.91
  model         "gpt-4o"
  timestamp     "2026-03-25T09:22:11Z"

Nothing is overwritten. If the advisor runs ten times on the same instance, you get ten AI_ADVISED edges — a complete history of what the AI said and why.

ADVANCED_TO edges ​

When a stage transition happens (human or automated), an ADVANCED_TO edge records it:

(JourneyInstance) -[:ADVANCED_TO]-> (Stage)

Properties written:
  actor         "alice@acme.com"
  from_stage    "negotiation"
  to_stage      "contract-review"
  notes         "Stakeholder confirmed via email"
  timestamp     "2026-03-25T09:45:00Z"

HITLTask nodes ​

When the AI escalates to a human reviewer, a HITLTask node is created and linked to the instance:

(HITLTask)

Properties written:
  task_id       "hitl:9871-xyz"
  status        "resolved"
  assignee      "bob@acme.com"
  decision      "reject"
  rationale     "Missing regulatory sign-off for EU accounts"
  created_at    "2026-03-25T09:10:00Z"
  resolved_at   "2026-03-25T09:44:00Z"

SLA_BREACHED edges ​

If a stage exceeds its configured SLA, a SLA_BREACHED edge is appended to the instance — never deleted, never modified.

Querying the memory ​

All of this is standard Cypher against POST /query (or the graph_query tool if you're using an MCP agent).

What did the AI recommend for a specific instance? ​

cypher
MATCH (ji:JourneyInstance {id: "ji:9871-abc"})-[r:AI_ADVISED]->(s:Stage)
RETURN r.timestamp, r.action_type, r.recommended, r.reasoning, r.confidence
ORDER BY r.timestamp ASC

Where did humans override the AI? ​

cypher
MATCH (ji:JourneyInstance)-[:AI_ADVISED]->(s:Stage)
MATCH (ji)-[:HITL_TASK]->(t:HITLTask {decision: "reject"})
RETURN ji.id, s.name AS ai_recommended, t.rationale, t.assignee
LIMIT 50

Override rate by journey type ​

cypher
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 DESC

Run this monthly. A journey type with a high override rate is one where the AI's reasoning isn't matching human judgement — a signal to review what context the advisor is receiving.

What precedes SLA breaches? ​

cypher
MATCH (ji:JourneyInstance)-[:SLA_BREACHED]->(s:Stage)
MATCH (ji)-[ai:AI_ADVISED]->(prev)
WHERE ai.timestamp < s.breached_at
RETURN ai.action_type, ai.reasoning, count(*) AS frequency
ORDER BY frequency DESC
LIMIT 10

Full chronological audit trail for one instance ​

cypher
MATCH (ji:JourneyInstance {id: "ji:9871-abc"})
OPTIONAL MATCH (ji)-[r]->()
RETURN type(r) AS event_type, r.timestamp AS ts, properties(r) AS detail
ORDER BY ts ASC

Building a learning loop ​

The graph accumulates signal automatically. To close the loop — so the AI makes better decisions on future instances — you add one thing: an OUTCOME edge when a journey closes.

Step 1 — write the outcome ​

python
from purple8_graph import GraphEngine

engine = GraphEngine(data_dir="./data")

engine.add_edge(
    src_id="ji:9871-abc",
    dst_id="ji:9871-abc",   # self-edge on the instance node
    edge_type="OUTCOME",
    properties={
        "result":     "won",         # "won" | "lost" | "expired" | "abandoned"
        "closed_at":  "2026-03-25T16:00:00Z",
        "revenue":    2_000_000,
        "cycle_days": 47,
    },
)

Step 2 — query patterns from closed instances ​

cypher
MATCH (ji:JourneyInstance {journey_type: "sales-cycle"})-[o:OUTCOME {result: "won"}]->()
MATCH (ji)-[ai:AI_ADVISED]->(s)
RETURN ai.action_type, ai.recommended, avg(o.cycle_days) AS avg_cycle
ORDER BY avg_cycle ASC
LIMIT 5

This tells you: among closed-won deals, which AI actions correlated with the shortest cycles.

Step 3 — pass patterns as context to the advisor ​

JourneyAIAdvisor.advise() takes an audit_history argument — the list of AI_ADVISED, ADVANCED_TO, and HITLTask events for the current instance. The model has no internal weights to update; the learning is entirely in what context you give it.

python
from purple8_graph.journey import JourneyAIAdvisor

advisor = JourneyAIAdvisor(llm_provider=provider, journey_config=cfg)

# Pull the closed-won patterns you queried above
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_cycle
    ORDER BY avg_cycle ASC LIMIT 5
""", {"jtype": "sales-cycle"})

recommendation = await advisor.advise(
    instance=current_instance,
    audit_history=history,       # current instance's edge history
    few_shot_patterns=patterns,  # closed-won patterns injected into system prompt
)

This is the entire learning loop. No retraining, no embeddings to update, no vector index rebuild. The graph stores the signal; you decide which signal to surface as context.

Getting new knowledge into the graph ​

The KnowledgeExtractor and AIGraphBuilder classes (in purple8_graph.genai) extract entities and relationships from raw text and write them as nodes and edges:

python
from purple8_graph.genai import KnowledgeExtractor, AIGraphBuilder
from purple8_graph import GraphEngine

engine    = GraphEngine(data_dir="./data")
extractor = KnowledgeExtractor(llm_provider=provider)
builder   = AIGraphBuilder(engine=engine, extractor=extractor)

await builder.ingest(
    text="""
    GlobalParts signed a 12-month supply agreement with Acme Corp.
    The deal was brokered by Sarah Chen and approved by the Acme procurement board.
    """,
    source="contracts-2026-q1",
)

# Written to the graph:
#   (Company:GlobalParts) -[:SUPPLIES]->     (Company:AcmeCorp)
#   (Deal)                -[:BROKERED_BY]->  (Person:SarahChen)
#   (Deal)                -[:APPROVED_BY]->  (Board:AcmeProcurement)

NaturalLanguageQuery translates plain-English questions to Cypher at query time:

python
from purple8_graph.genai import NaturalLanguageQuery

nlq = NaturalLanguageQuery(engine=engine, llm_provider=provider)

result = await nlq.query(
    "Which suppliers have active contracts with more than two of our top accounts?"
)
# Returns Cypher results as structured data — no schema knowledge required

The REST equivalent is POST /rag/query with a {"question": "…"} body.

Subscribing to memory changes in real time ​

The EventBus (from purple8_graph.cdc) fires a ChangeEvent on every graph write. Use it to react the moment an AI_ADVISED edge lands — alert a Slack channel, update a dashboard, trigger a downstream workflow:

python
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)

Or stream over WebSocket — no Python process needed:

ws://localhost:8010/ws/changes?tenant_id=acme&event_type=edge_added

Each message is a JSON-serialised ChangeEvent:

json
{
  "event_id":   "a3f9c1…",
  "event_type": "edge_added",
  "entity_id":  "ai_advised:9871-xyz",
  "tenant_id":  "acme",
  "timestamp":  1743016931.42,
  "properties": {
    "edge_type":   "AI_ADVISED",
    "action_type": "advance",
    "recommended": "contract-review",
    "confidence":  0.91
  }
}

RAG retrieval tuning ​

When the AI generates answers via POST /rag/query, it retrieves graph nodes as context first. The retrieval behaviour is controlled per tenant by rag_config.json in the tenant data directory:

json
{
  "chunk_size":      512,
  "retrieval_k":     8,
  "embedding_model": "text-embedding-3-large",
  "min_similarity":  0.72
}
ParameterWhat it does
retrieval_kNodes retrieved per query. Higher = broader context; lower = higher precision.
min_similarityCosine similarity floor. Raise to filter noise; lower to catch edge-case matches.
embedding_modelSwap per tenant without reindexing.
chunk_sizeHow text properties are split before embedding.

What this is not ​

To be explicit about what does not exist today:

  • There is no background process that automatically re-ranks or updates embeddings based on feedback
  • There is no LearningEngine or MemoryLayer class to import
  • The AI advisor does not update its own weights — it is a stateless LLM call that receives whatever context you pass it

The pattern this page describes — write outcomes, query patterns, pass patterns as context — is the learning loop. It is simple on purpose: the graph handles persistence and querying; you decide what signal is worth surfacing.

See Also ​

Purple8 Graph is proprietary software. All rights reserved.

© 2026 Purple8