Skip to main content

Models

All models are defined in memproof.models.core and re-exported from the top-level memproof package. Every model is a Pydantic BaseModel (or str, Enum for enumerations).
from memproof import (
    MemoryScope, OperationContext, MemoryRecord,
    RiskFactor, RiskAssessment, PolicyDecision,
    MemoryOperationResponse, OperationStatusResponse,
    MemorySearchResponse,
    OperationType, OperationStatus, RiskLevel,
    DecisionAction, ActorType,
)

Enumerations

OperationType

Identifies the kind of memory operation.
class OperationType(str, Enum):
    remember = "remember"
    update   = "update"
    forget   = "forget"
    search   = "search"
    get      = "get"

OperationStatus

Tracks the lifecycle state of an operation as it moves through the pipeline.
class OperationStatus(str, Enum):
    received         = "received"
    pending_approval = "pending_approval"
    committed        = "committed"
    blocked          = "blocked"
    quarantined      = "quarantined"
    failed           = "failed"
ValueMeaning
receivedThe request has been accepted but processing has not completed.
pending_approvalThe policy engine returned require_approval. A human reviewer must call approve() or deny().
committedThe operation completed successfully and the memory has been persisted (or deleted).
blockedThe policy engine denied the operation.
quarantinedThe operation was quarantined for later review.
failedAn unexpected error occurred during processing.

RiskLevel

Categorical risk level derived from the numeric risk score.
class RiskLevel(str, Enum):
    low      = "low"
    medium   = "medium"
    high     = "high"
    critical = "critical"

DecisionAction

The action taken by the policy engine after evaluating rules against the risk assessment.
class DecisionAction(str, Enum):
    allow            = "allow"
    deny             = "deny"
    require_approval = "require_approval"
    quarantine       = "quarantine"

ActorType

Identifies the type of actor performing an operation.
class ActorType(str, Enum):
    agent  = "agent"
    user   = "user"
    system = "system"

Value Objects

MemoryScope

Identifies the multi-tenant hierarchy a memory belongs to. Every memory is scoped to at least a tenant, project, and agent.
class MemoryScope(BaseModel):
    tenant_id: str
    project_id: str
    agent_id: str
    session_id: str | None = None
    subject_id: str | None = None
tenant_id
str
required
Top-level tenant identifier.
project_id
str
required
Project within the tenant.
agent_id
str
required
Agent that owns or created the memory.
session_id
str | None
default:"None"
Optional session identifier. Use this to scope memories to a specific conversation session.
subject_id
str | None
default:"None"
Optional subject identifier. Use this to scope memories to a specific end-user or entity the memory is about.

Example

from memproof import MemoryScope

scope = MemoryScope(
    tenant_id="acme",
    project_id="assistant",
    agent_id="agent-1",
    session_id="sess-001",
    subject_id="user-42",
)

OperationContext

Captures the identity and metadata of the actor performing an operation.
class OperationContext(BaseModel):
    actor_type: ActorType
    actor_id: str
    source: str
    request_id: str | None = None
    correlation_id: str | None = None
    timestamp: datetime
    metadata: dict[str, Any] | None = None
actor_type
ActorType
required
The type of actor: "agent", "user", or "system".
actor_id
str
required
Unique identifier of the actor performing the operation.
source
str
required
The originating system or framework (e.g., "langgraph", "web-ui", "api").
request_id
str | None
default:"None"
Optional request-level identifier for tracing.
correlation_id
str | None
default:"None"
Optional correlation identifier for linking related operations across services.
timestamp
datetime
required
ISO 8601 timestamp of when the operation was initiated. Accepts both datetime objects and ISO 8601 strings (Pydantic coerces strings automatically).
metadata
dict[str, Any] | None
default:"None"
Arbitrary key-value metadata associated with the operation context.

Example

from memproof import OperationContext

context = OperationContext(
    actor_type="agent",
    actor_id="agent-1",
    source="langgraph",
    timestamp="2026-01-15T10:30:00Z",
    request_id="req-abc",
    correlation_id="corr-xyz",
)

Record Models

MemoryRecord

The persisted memory object as returned by the adapter.
class MemoryRecord(BaseModel):
    memory_id: str
    content: str
    tags: list[str] = Field(default_factory=list)
    scope: MemoryScope
    metadata: dict[str, Any] | None = None
    ttl_seconds: int | None = None
    created_at: datetime
    updated_at: datetime
memory_id
str
Unique identifier for the memory record.
content
str
The stored memory content.
tags
list[str]
default:"[]"
Tags associated with the memory.
scope
MemoryScope
The scope hierarchy this memory belongs to.
metadata
dict[str, Any] | None
default:"None"
Arbitrary metadata attached to the memory.
ttl_seconds
int | None
default:"None"
Time-to-live in seconds, if set.
created_at
datetime
Timestamp when the memory was first created.
updated_at
datetime
Timestamp of the most recent update.

Risk and Policy Models

RiskFactor

A single contributing factor to the overall risk score.
class RiskFactor(BaseModel):
    name: str
    contribution: float = Field(ge=0, le=1)
    description: str
    evidence: str | None = None
name
str
Machine-readable name of the risk factor (e.g., "pii_detected", "cross_tenant").
contribution
float
Numeric contribution to the overall risk score, between 0.0 and 1.0.
description
str
Human-readable explanation of why this factor was triggered.
evidence
str | None
default:"None"
Optional supporting evidence or details.

RiskAssessment

The output of the risk engine, combining individual factors into an overall score and level.
class RiskAssessment(BaseModel):
    score: float = Field(ge=0, le=1)
    level: RiskLevel
    factors: list[RiskFactor]
    scorer: str
score
float
Overall risk score between 0.0 (no risk) and 1.0 (maximum risk).
level
RiskLevel
Categorical risk level: "low", "medium", "high", or "critical".
factors
list[RiskFactor]
Individual risk factors that contributed to the score.
scorer
str
Identifier of the risk scoring algorithm or engine version that produced this assessment.

PolicyDecision

The output of the policy engine after evaluating rules.
class PolicyDecision(BaseModel):
    action: DecisionAction
    reason_codes: list[str]
    matched_rule_ids: list[str] | None = None
    policy_version: str
action
DecisionAction
The decided action: "allow", "deny", "require_approval", or "quarantine".
reason_codes
list[str]
Machine-readable codes explaining why this decision was made (e.g., ["pii_detected", "high_risk"]).
matched_rule_ids
list[str] | None
default:"None"
IDs of the policy rules that matched and triggered this decision.
policy_version
str
Version string of the policy configuration that was evaluated.

Response Models

MemoryOperationResponse

Returned by remember() and update(). Contains the full result of a create or update operation.
class MemoryOperationResponse(BaseModel):
    operation_id: str
    status: OperationStatus
    memory: MemoryRecord | None = None
    risk_assessment: RiskAssessment | None = None
    decision: PolicyDecision
operation_id
str
Unique identifier for this operation. Use this to track status via get_operation_status(), approve(), or deny().
status
OperationStatus
Current status of the operation.
memory
MemoryRecord | None
The created or updated memory record. This is None when the operation was blocked, quarantined, or is pending approval.
risk_assessment
RiskAssessment | None
The risk assessment produced during pipeline execution.
decision
PolicyDecision
The policy engine’s decision for this operation.

Example

result = await mp.remember(
    content="user email is alice@example.com",
    scope={"tenant_id": "acme", "project_id": "p1", "agent_id": "a1"},
    context={
        "actor_type": "agent", "actor_id": "a1",
        "source": "langgraph", "timestamp": "2026-01-15T10:30:00Z",
    },
)

if result.status == OperationStatus.committed:
    print(f"Stored: {result.memory.memory_id}")
elif result.status == OperationStatus.pending_approval:
    print(f"Awaiting approval: {result.operation_id}")
elif result.status == OperationStatus.blocked:
    print(f"Blocked: {result.decision.reason_codes}")

OperationStatusResponse

Returned by forget(), approve(), deny(), and get_operation_status().
class OperationStatusResponse(BaseModel):
    operation_id: str
    operation_type: OperationType
    status: OperationStatus
    memory_id: str | None = None
    decision: PolicyDecision
    risk_assessment: RiskAssessment | None = None
    created_at: datetime | None = None
    updated_at: datetime | None = None
operation_id
str
Unique identifier of the operation.
operation_type
OperationType
The type of operation: "remember", "update", "forget", "search", or "get".
status
OperationStatus
Current lifecycle status.
memory_id
str | None
The ID of the affected memory, if applicable.
decision
PolicyDecision
The policy decision that governed this operation.
risk_assessment
RiskAssessment | None
The risk assessment, if one was performed.
created_at
datetime | None
When the operation was first received.
updated_at
datetime | None
When the operation status was last updated.

MemorySearchResponse

Returned by search(). Contains ranked search results.
class MemorySearchResponse(BaseModel):
    hits: list[MemorySearchHit]
hits
list[MemorySearchHit]
Ordered list of search results, ranked by relevance score (highest first).

MemorySearchHit

A single search result pairing a memory record with its relevance score.
class MemorySearchHit(BaseModel):
    memory: MemoryRecord
    score: float = Field(ge=0, le=1)
memory
MemoryRecord
The matching memory record.
score
float
Relevance score between 0.0 and 1.0.

Example

results = await mp.search(
    query="dark mode",
    scope={"tenant_id": "acme", "project_id": "p1", "agent_id": "a1"},
    context={
        "actor_type": "agent", "actor_id": "a1",
        "source": "langgraph", "timestamp": "2026-01-15T10:31:00Z",
    },
)

for hit in results.hits:
    print(f"[{hit.score:.2f}] {hit.memory.content}")

Audit Trail Models

TrailEvent

An immutable audit event emitted at each stage of the orchestration pipeline. Events are stored in the Trailproof audit trail with SHA-256 hash chains and optional HMAC signing for tamper evidence.
class TrailEvent:
    event_id: str
    event_type: str
    timestamp: datetime
    actor: str | None
    metadata: dict[str, Any]
    hash: str
    prev_hash: str | None
    signature: str | None
event_id
str
Unique identifier for this event.
event_type
str
Namespaced event type (e.g., "memproof.pipeline.received", "memproof.pipeline.committed").
timestamp
datetime
When the event was created.
actor
str | None
The actor who triggered the event, if available.
metadata
dict[str, Any]
Event-specific data including operation_id, tenant_id, project_id, and stage-specific payload (e.g., risk assessment result, policy decision).
hash
str
SHA-256 hash of this event for chain integrity.
prev_hash
str | None
Hash of the previous event in the chain. None for the first event.
signature
str | None
HMAC-SHA256 signature if signing is enabled. None otherwise.

Request Models

These models are used internally by the Memproof class to structure requests before passing them to the orchestrator. You generally do not need to construct them directly — the Memproof methods accept raw dictionaries and build these internally.

MemoryCreateRequest

class MemoryCreateRequest(BaseModel):
    content: str = Field(min_length=1)
    tags: list[str] | None = None
    scope: MemoryScope
    context: OperationContext
    metadata: dict[str, Any] | None = None
    ttl_seconds: int | None = Field(default=None, gt=0)

MemoryUpdateRequest

class MemoryUpdateRequest(BaseModel):
    content: str | None = None
    tags: list[str] | None = None
    metadata_patch: dict[str, Any] | None = None
    ttl_seconds: int | None = Field(default=None, gt=0)
    context: OperationContext

MemorySearchRequest

class MemorySearchRequest(BaseModel):
    query: str = Field(min_length=1)
    scope: MemoryScope
    context: OperationContext
    limit: int = Field(default=20, ge=1, le=100)
    filters: dict[str, Any] | None = None