Agentic Architecture Patterns on Purple8 Hyper Graph
Purple8 Hyper Graph is not a database with an AI wrapper. It is an AI-native backend where the graph is the coordination substrate — agents read and write a shared persistent graph instead of passing messages to each other. This inverts the architecture of every mainstream agentic framework.
This document describes the three architectural patterns this substrate enables, how to implement each, and where they apply in the real world.
The core inversion
Every mainstream agentic framework (LangGraph, AutoGen, CrewAI) treats the graph as a metaphor — agents as nodes, messages as edges, orchestrated by a runtime process. Purple8 inverts this: the graph is the substrate.
Mainstream approach Purple8 approach
────────────────── ────────────────
Agents exchange messages ──────→ Agents read/write a shared graph
Orchestrator holds state ──────→ Graph IS the state
Trace logged separately ──────→ Every mutation IS the trace (WAL + edges)
Memory per-agent ──────→ Shared HNSW index across all agents
HITL = optional wrapper ──────→ HITL = structural constraint in JourneyDefThe consequence: agents are stateless processes; all state lives in Purple8. Any agent can be killed and restarted; a different agent (or a different model) can continue the same task by querying the graph. This is only possible because Purple8 persists every mutation via BrickWAL fdatasync before acknowledging it.
The patterns below all follow from this single property.
Pattern 1 — Graph-Grounded Attention
What it is
Standard RAG fans out to the full vector index. Graph-grounded attention uses the graph's topology to pre-select which part of the index to search before the vector search runs. The agent reads structural signals — community membership, PageRank centrality — to focus retrieval on semantically relevant subgraphs.
Each inference then writes back to the graph: the output becomes a node, and edges record what it was derived from with a confidence weight. Over time, the graph gets denser around high-value concepts without any explicit training. That is Hebbian reinforcement at the application layer.
How to implement
from purple8_graph.core import GraphEngine
engine = GraphEngine("./data")
def graph_grounded_query(engine: GraphEngine, question: str) -> str:
# 1. Find which community of nodes is topically relevant
communities = engine.detect_communities()
relevant_community = route_to_community(question, communities)
# 2. Rank nodes within that community by structural importance
pagerank = engine.pagerank(
nodes=[n for n in relevant_community.nodes],
damping=0.85
)
seed_nodes = sorted(pagerank.scores, key=lambda x: x[1], reverse=True)[:5]
# 3. Vector search anchored to that subgraph — not the full index
hits = engine.graph_guided_search(
seed_node_id=seed_nodes[0][0],
query_embedding=embed(question),
hops=2,
k=10,
)
# 4. LLM generates answer from subgraph hits
top_score = hits[0].score if hits else 0.0
answer = llm_generate(question, [h.node for h in hits])
# 5. Write back — output becomes a durable graph node
output_node = engine.add_node(
node_id=generate_id(),
labels=["Inference"],
properties={
"question": question,
"answer": answer,
"confidence": top_score,
}
)
# 6. Edge back to the evidence — confidence-weighted
for seed_id, rank_score in seed_nodes[:3]:
engine.add_edge(
edge_id=f"{output_node.node_id}->derived->{seed_id}",
edge_type="DERIVED_FROM",
source_id=output_node.node_id,
target_id=seed_id,
properties={"weight": rank_score * top_score},
)
return answerVia MCP (no code required)
1. rag.profile_collection ← get topology-aware retrieval params
2. graph.communities ← find relevant cluster
3. graph.pagerank ← rank by structural importance
4. rag.hybrid_query (seed_nodes=…) ← retrieve from subgraph
5. graph.add_node + graph.add_edge ← write inference back to graphReal-world applications
| Domain | How it applies | Outcome |
|---|---|---|
| Legal due diligence | 10,000 contracts ingested. Agent routes a "change of control clause" query to the M&A community, ranks clauses by how many contracts reference them (PageRank), retrieves from that cluster only. | Retrieval precision increases because 9,800 irrelevant contracts are never searched. Each inference adds a CLAUSE_PRECEDENT edge — next query benefits from prior reasoning. |
| Clinical decision support | Patient history, drug interactions, and treatment protocols in one graph. Agent finds the patient's diagnosis community, ranks treatments by outcome frequency edges, retrieves evidence. | Drug interaction risks surface through graph traversal before the LLM reasons about them — the graph structure catches structural contradictions the vector index cannot. |
| Financial research | Earnings calls, analyst reports, SEC filings. Agent routes "supply chain risk" to the semiconductor community by PageRank, not keyword. | Retrieves structurally central documents (widely cross-referenced) rather than most-similar documents. Finds risks that appear peripherally across many filings rather than prominently in one. |
| AEC / BIM | Building model as graph. Agent routes structural query to load-bearing member community, ranks by betweenness centrality (bridge nodes = critical members). | Structural analysis focuses on members whose failure has the highest cascade effect — identified by graph topology, not document similarity. |
Pattern 2 — The Relay
What it is
Multi-agent coordination without message passing, shared memory objects, or an orchestrator process. Each agent writes its output to the graph as a node, then terminates. The next agent finds that node by querying graph state and continues.
The handoff mechanism is the shape of the node, not a function call.
Agent A Graph Agent B
─────── ───── ───────
receives task →
processes →
add_node(status="complete", ←──────────────────────────────
output={...})
terminates →
──────────────────────────→ starts
query(status="complete",
unprocessed=True)
reads A's output node
continues task
add_node(status="complete")Agents can run sequentially (dev), in parallel (prod fan-out), on different machines, with different models — the graph contract between them is identical in all cases. Switching from one agent to N parallel agents is a deployment decision, not a re-architecture.
How to implement
# --- Agent A: Extractor ---
def extractor_agent(engine: GraphEngine, document_id: str):
doc = engine.get_node(document_id)
entities = extract_entities(doc.properties["content"]) # LLM call
# Write output as a graph node — this is the handoff
engine.add_node(
node_id=f"extraction:{document_id}",
node_type="ExtractionResult",
properties={
"source_doc": document_id,
"entities": entities,
"status": "complete",
"processed_by": "extractor",
"handoff_to": "enricher", # declarative routing hint
}
)
engine.add_edge(
source_id=f"extraction:{document_id}",
target_id=document_id,
edge_type="EXTRACTED_FROM",
)
# Agent terminates here — no callback, no message, no return value
# --- Agent B: Enricher (can start any time after A writes) ---
def enricher_agent(engine: GraphEngine):
# Find work by querying graph state — no coordination needed
pending = engine.query(
filters={
"node_type": "ExtractionResult",
"status": "complete",
"handoff_to": "enricher",
}
)
for result_node in pending.nodes:
entities = result_node.properties["entities"]
relationships = infer_relationships(entities) # LLM call
for rel in relationships:
engine.add_edge(
source_id=rel.source,
target_id=rel.target,
edge_type=rel.type,
properties={"confidence": rel.confidence},
)
# Mark as processed — removes from pending queue
engine.update_node(
result_node.node_id,
properties={"status": "enriched", "handoff_to": None}
)Via MCP
Agent A:
data.ingest_file → graph.add_node(status="complete", handoff_to="enricher")
Agent B (polls on schedule or triggered by CDC):
graph.query(filter={status="complete", handoff_to="enricher"})
→ for each result: graph.add_edge × N
→ graph.update_node(status="enriched", handoff_to=None)Real-world applications
| Domain | How it applies | Outcome |
|---|---|---|
| Document intelligence pipeline | DocIntel extracts entities (Agent A). Enricher resolves entity aliases and adds SAME_AS edges (Agent B). Graph analyst runs PageRank to find most-cited entities (Agent C). Each runs independently, triggered by CDC events. | Adding a new agent to the pipeline does not change A or B. The pipeline is extensible without re-architecture. If Agent C crashes mid-run, it resumes from the last status="enriched" node — no work is lost. |
| Software vulnerability triage | Scanner produces CVE nodes (Agent A). Enricher maps CVEs to affected code paths via graph traversal (Agent B). Prioritiser ranks by blast radius (betweenness centrality of affected nodes) (Agent C). | The relay discovers that a low-CVSS vulnerability in a central library node has higher blast radius than a high-CVSS vulnerability in a leaf node — a ranking that keyword-based triage cannot produce. |
| Supply chain due diligence | News ingestion agent writes supplier mention nodes (Agent A). Risk classifier adds RISK_SIGNAL edges (Agent B). Aggregator traverses supplier graph to find second-order exposure (Agent C). | A sanctions event on Supplier X propagates through SUPPLIES_TO edges to surface all customers with indirect exposure — discovered by graph traversal, not document search. |
| Research synthesis | Paper ingestor writes paper nodes + citation edges (Agent A). Claim extractor adds CLAIMS nodes (Agent B). Contradiction detector finds CLAIMS nodes with conflicting polarity connected to the same concept (Agent C). | Contradictions are found by graph pattern matching (two nodes with opposing supports/refutes edges to the same concept), not by LLM reading all papers simultaneously. Scales to 100,000 papers. |
Pattern 3 — The Parliament
What it is
Multiple independent agents assess the same entity from different perspectives and write weighted assessment edges to the graph. An aggregator reads all assessments, computes a combined view, and either proceeds autonomously or blocks on a HITL gate — depending on the degree of disagreement between agents, not a fixed threshold.
The HITL gate is a structural constraint in the JourneyDefinition, not a conditional in agent code. An agent cannot bypass it regardless of confidence. This is what makes the pattern appropriate for regulated industries: human oversight is constitutional, not advisory.
Entity node
│
┌───────────────┼───────────────┐
▼ ▼ ▼
Legal Agent Financial Agent Technical Agent
(reads contract (reads invoice (reads spec nodes)
nodes) nodes)
│ │ │
▼ ▼ ▼
RISK_ASSESSMENT RISK_ASSESSMENT RISK_ASSESSMENT
edge (weight=0.8) edge (weight=0.6) edge (weight=0.9)
│ │ │
└───────────────┼───────────────┘
▼
Aggregator Agent
traverse(RISK_ASSESSMENT edges)
→ weighted mean + max disagreement
→ if disagreement > threshold:
journey.advance(stage="human_review") ← HITL gate
→ else:
journey.advance(stage="approved")How to implement
from purple8_graph.journey import JourneyEngine, StageSpec, SLAPolicy
# --- Define the journey — HITL gate is structural ---
def build_assessment_journey(je: JourneyEngine):
je.define_journey(
journey_type="supplier_assessment",
stages=[
StageSpec(
name="parallel_assessment",
allowed_transitions=["aggregation"],
sla=SLAPolicy(hours=2),
),
StageSpec(
name="aggregation",
allowed_transitions=["approved", "human_review"],
sla=SLAPolicy(hours=1),
),
StageSpec(
name="human_review",
requires_hitl=True, # ← constitutional constraint
allowed_transitions=["approved", "rejected"],
sla=SLAPolicy(hours=24),
),
StageSpec(name="approved"),
StageSpec(name="rejected"),
]
)
# --- Parliament agents (run independently, in parallel) ---
def legal_agent(engine: GraphEngine, entity_id: str):
contracts = engine.traverse(entity_id, edge_type="HAS_CONTRACT", max_hops=2)
risk = assess_legal_risk(contracts) # LLM call
engine.add_edge(
source_id=entity_id,
target_id=entity_id, # self-referential assessment edge
edge_type="RISK_ASSESSMENT",
properties={
"assessor": "legal",
"score": risk.score,
"factors": risk.factors,
"confidence": risk.confidence,
}
)
# financial_agent and technical_agent follow the same pattern
# --- Aggregator agent ---
def aggregator_agent(
engine: GraphEngine,
je: JourneyEngine,
entity_id: str,
instance_id: str,
):
assessments = engine.traverse(
entity_id,
edge_type="RISK_ASSESSMENT",
direction="outbound",
)
scores = [a.properties["score"] for a in assessments.edges]
mean_score = sum(scores) / len(scores)
max_disagreement = max(scores) - min(scores)
# Disagreement > 0.3 means agents diverged — human must decide
if max_disagreement > 0.3:
je.advance(
instance_id=instance_id,
to_stage="human_review",
context={
"mean_score": mean_score,
"disagreement": max_disagreement,
"assessments": scores,
}
)
elif mean_score > 0.7:
je.advance(instance_id=instance_id, to_stage="approved")
else:
je.advance(instance_id=instance_id, to_stage="rejected")Via MCP
Parliament agents (run in parallel, each with a scoped API key):
graph.traverse(entity, "HAS_CONTRACT") ← legal agent reads
graph.add_edge(RISK_ASSESSMENT, weight=score)
Aggregator:
graph.traverse(entity, "RISK_ASSESSMENT")
→ if disagreement > threshold: journey.advance("human_review")
→ else: journey.advance("approved" or "rejected")
Human officer (in portal or via MCP):
journey.list_hitl ← see pending decisions
journey.resolve_hitl(...) ← unblocks the journeyReal-world applications
| Domain | Parliament composition | HITL trigger | Outcome |
|---|---|---|---|
| Loan underwriting | Credit risk agent (financial history graph), Fraud detection agent (transaction pattern graph), Compliance agent (sanctions/PEP graph) | Agents disagree by > 0.25 on risk score, OR any agent flags a hard constraint | Reduces false positives: automated approval when all three agents agree. Human review only on genuine ambiguity. Audit trail is every RISK_ASSESSMENT edge — queryable, not buried in logs. |
| AEC permit approval | Structural safety agent (load-path graph), Fire safety agent (egress graph), Environmental agent (zoning/impact graph) | Structural agent flags load-path failure OR agents disagree on compliance | A design change that passes 2 of 3 agents still reaches human review. The RISK_ASSESSMENT edges show which structural members triggered the flag — not just a yes/no. |
| Clinical trial eligibility | Inclusion criteria agent (patient history graph), Exclusion criteria agent (contraindication graph), Drug interaction agent (pharmacology graph) | Any exclusion agent flags a hard stop, OR inclusion scores diverge > 0.2 | Eligibility decisions are graph-verifiable: the DISQUALIFIED_BY edge points to the exact contraindication node. Audit is complete without a separate logging system. |
| M&A target screening | Financial health agent (balance sheet graph), Regulatory risk agent (litigation/compliance graph), Strategic fit agent (market overlap graph) | Financial and strategic agents disagree by > 0.35 | Pre-qualifies 80% of targets automatically. Human bandwidth concentrates on genuinely ambiguous cases. The dissent record (disagreeing RISK_ASSESSMENT edges) is preserved for post-deal review. |
Combining patterns
The three patterns compose. A typical production system uses all three simultaneously across different layers:
Ingest layer: Relay pattern
(extract → enrich → classify → link)
Query layer: Graph-grounded attention
(community → pagerank → guided RAG → write-back)
Decision layer: Parliament pattern
(parallel assessment → aggregation → HITL gate)The Journey Engine is the skeleton that connects them: the output of the Relay (enriched, linked graph) becomes the input to the Parliament (entities ready for assessment), and the Parliament's approved/rejected verdict advances the Journey to its final stage.
Why agents do not call each other
This is the most important design constraint, and the most frequently violated by developers coming from LangGraph or AutoGen.
Every agent output is a graph write. The graph is the only handoff mechanism.
If you find yourself building a results object to pass between agents, that result should be a graph node instead. If you find yourself building a message queue between agents, that queue should be a CDC event stream from the graph instead.
The reason is not stylistic. It is operational:
| Passing objects between agents | Writing to the graph |
|---|---|
| State lost if agent crashes | State persisted before the agent acknowledges |
| Requires both agents to be alive simultaneously | Agent B can start days after Agent A finishes |
| No audit trail without separate logging | Every write is an immutable edge in the graph |
| Scaling requires re-architecture | Scaling is a deployment decision — same graph contract |
Key API surface for agentic workloads
| Need | P8G tool / API |
|---|---|
| Community-based retrieval routing | engine.detect_communities() + rag.hybrid_query(seed_nodes=…) |
| PageRank-weighted seed selection | engine.pagerank() or graph.pagerank MCP tool |
| Write inference node back to graph | engine.add_node() or graph.add_node MCP tool |
| Write inference edge back to graph | engine.add_edge() or graph.add_edge MCP tool — primary agentic handoff mechanism |
| Advance relay agent state | engine.update_node() or graph.update_node MCP tool — patch status / handoff_to properties |
| Find work left by another agent | engine.query(filters={status="complete", handoff_to="…"}) or graph.query MCP tool |
| Poll for new graph events (CDC) | graph.poll_changes MCP tool |
| Define HITL-gated workflow | JourneyEngine.define_journey(stages=[StageSpec(requires_hitl=True)]) |
| Surface pending HITL tasks | journey.list_hitl MCP tool |
| Resolve a HITL decision | journey.resolve_hitl MCP tool |
| Scope agent to read-only | Issue API key with node:read edge:read query:* permissions only |
| Scope agent to write its domain | Issue API key with node:create edge:create + specific labels |
Related documents
- MCP Integration — full reference for all 125 MCP tools across 16 namespaces
- Hybrid Graph-Vector RAG — RAG retrieval internals
- Journey Engine —
JourneyEngine,StageSpec,SLAPolicy, and durable AI outbox (replay_pending_ai_dispatches()) - Graph Analytics —
pagerank(),detect_communities(),betweenness_centrality()