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; agents can also call memory.history and memory.search in the memory.* namespace, and feedback.open, feedback.record, feedback.resolve, feedback.summary in the feedback.* namespace).

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(
    edge_id="outcome:9871-abc",
    edge_type="OUTCOME",
    source_id="ji:9871-abc",
    target_id="ji:9871-abc",   # self-edge on the instance node
    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 optional few_shot_patterns argument. The advisor already pulls the current instance's own AI_ADVISED / ADVANCED_TO / HITLTask audit history into its context window automatically — few_shot_patterns is for injecting other instances' outcomes (e.g. closed-won patterns) alongside it. 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(journey_engine=je, provider=provider)

# 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"})

result = await advisor.advise(
    instance_id="ji:9871-abc",
    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 AIGraphBuilder class (in purple8_graph.genai) combines extraction, node/edge creation, and embedding generation — it extracts entities and relationships from raw text and writes them as nodes and edges:

python

from purple8_graph.genai import AIGraphBuilder
from purple8_graph import GraphEngine

engine  = GraphEngine(data_dir="./data")
builder = AIGraphBuilder(engine, provider)   # builds its own KnowledgeExtractor internally

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

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

AIGraphBuilder.query() translates plain-English questions to a graph operation (vector search or traversal) at query time:

python

result = await builder.query(
    "Which suppliers have active contracts with more than two of our top accounts?"
)
# Returns a dict with "operation" and "results" keys — 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.data.get("edge_type") == "AI_ADVISED":
                await post_to_slack(event.data)

Or stream over WebSocket — no Python process needed:

ws://localhost:8100/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",
  "entity_kind": "edge",
  "tenant_id":  "acme",
  "timestamp":  1743016931.42,
  "data": {
    "edge_type":   "AI_ADVISED",
    "properties": {
      "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
}

Parameter	What it does
`retrieval_k`	Nodes retrieved per query. Higher = broader context; lower = higher precision.
`min_similarity`	Cosine similarity floor. Raise to filter noise; lower to catch edge-case matches.
`embedding_model`	Swap per tenant without reindexing.
`chunk_size`	How 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.

Graph as Memory ​

What gets written automatically ​

AI_ADVISED edges ​

ADVANCED_TO edges ​

HITLTask nodes ​

SLA_BREACHED edges ​

Querying the memory ​

What did the AI recommend for a specific instance? ​

Where did humans override the AI? ​

Override rate by journey type ​

What precedes SLA breaches? ​

Full chronological audit trail for one instance ​

Building a learning loop ​

Step 1 — write the outcome ​

Step 2 — query patterns from closed instances ​

Step 3 — pass patterns as context to the advisor ​

Getting new knowledge into the graph ​

Subscribing to memory changes in real time ​

RAG retrieval tuning ​

What this is not ​

See Also ​

Graph as Memory

What gets written automatically

`AI_ADVISED` edges

`ADVANCED_TO` edges

`HITLTask` nodes

`SLA_BREACHED` edges

Querying the memory

What did the AI recommend for a specific instance?

Where did humans override the AI?

Override rate by journey type

What precedes SLA breaches?

Full chronological audit trail for one instance

Building a learning loop

Step 1 — write the outcome

Step 2 — query patterns from closed instances

Step 3 — pass patterns as context to the advisor

Getting new knowledge into the graph

Subscribing to memory changes in real time

RAG retrieval tuning

What this is not

See Also