Custom Schemas Deep Dive
Use custom schemas to enforce domain-specific entities and relationships in your memory graph.
What You Will Build
- A domain schema with node and relationship types
- Memory ingestion constrained by schema defaults
- Retrieval and analytics on schema-specific entities
Step 1: Create a Schema
from papr_memory import Papr
import os
client = Papr(x_api_key=os.environ.get("PAPR_MEMORY_API_KEY"))
schema = client.schemas.create(
name="customer_support_schema",
description="Schema for support conversations and issue tracking",
node_types={
"Customer": {"properties": {"name": "string", "tier": "string"}},
"Issue": {"properties": {"severity": "string", "status": "string"}},
"FeatureRequest": {"properties": {"priority": "string"}}
},
relationship_types={
"REPORTED": {"allowed_source_types": ["Customer"], "allowed_target_types": ["Issue"]},
"REQUESTED": {"allowed_source_types": ["Customer"], "allowed_target_types": ["FeatureRequest"]}
}
)
schema_id = schema.data.idStep 2: Ingest with Schema-Aware Policy
client.memory.add(
content="Enterprise customer Acme reports SSO timeout issue and requests SCIM provisioning",
external_user_id="support_agent_001",
memory_policy={
"mode": "auto",
"schema_id": schema_id,
"node_constraints": [
{"node_type": "Issue", "set": {"status": {"mode": "exact", "value": "open"}}}
]
}
)Step 3: Retrieve by Schema Context
results = client.memory.search(
query="Open enterprise customer issues related to SSO",
schema_id=schema_id,
enable_agentic_graph=True,
max_memories=20,
max_nodes=15
)Step 4: Analyze with GraphQL
analytics = client.graphql.query(
query="""
query {
nodes(type: "Issue") {
id
properties
}
}
"""
)Production Notes
- Keep schema versions explicit (
support_schema_v1,support_schema_v2). - Store migration notes whenever node or relationship contracts change.
- Prefer schema defaults in
memory_policyand override only where necessary.