Skip to content

Experience Loop

Experience Loop — Dyna-style experience recording with Beta competence tracking.

Cluster: Experience & Autonomy | Type: component | MCP Tools: 20

Overview

Dyna-style experience recording system that stores (state, action, reward, next-state) tuples and derives competence estimates using a Beta distribution updated by binary outcome signals. Provides operations for similarity-based episode retrieval with sorted similarity scores, tenant- and task-scoped action recommendations, escalation thresholds, experience replay, stale-entry pruning, compaction, privacy-aware export/delete, data-consumption reporting, capabilities discovery, and learning_layer dataset sync. By default it persists to SQLite for cross-session learning and shared local processes; .jsonl paths remain supported for lightweight compatibility, and db_path="" gives ephemeral in-memory use.

Decision-support only

experience_loop recommendations are experience heuristics: they suggest what has worked before and expose competence trends, but they are not a calibrated safety model or proof of correctness. Elevated and high-stakes autonomy requires outcome provenance fields (outcome_verified_by and anti_gaming_policy) and still needs domain-specific specs, representative evaluations, safety gates, audit logging, and human review before acting on recommendations or making compliance claims.

Persistence security posture

Tenant filtering and redaction are implemented, but built-in auth, RBAC, audit logging, encryption-at-rest, and db-path confinement are unsupported unless the deployment environment supplies them. Production-grade persistence security remains blocked-escalated pending product/security policy.

When to use:

  • Building an agent that improves decision-making by replaying and querying past experiences
  • Tracking per-task-type competence (Beta posterior) to decide when to escalate to a human or higher-agency component
  • Implementing a lightweight Dyna architecture where a world model is approximated from stored transitions
  • Exposing read-oriented experience data through the experience_loop_mcp adapter (capabilities, stats, competence, recommendations, transition table, data consumption, similarity search, and escalation decisions)

Example:

from mvp.experience_loop import ExperienceLoopBlock, ExperienceLoopInput

block = ExperienceLoopBlock(name="exp_loop")
block.infer(ExperienceLoopInput(
    op="record", task_type="math", outcome="success",
    reward=1.0, state_before={"q": "2+2"}, state_after={"a": "4"},
))
result = block.infer(ExperienceLoopInput(op="get_competence", task_type="math"))
# result.ok -> True; result.value.competence -> float in [0, 1]

Works well with: autonomy_governor, cog_arch_soar, evoskill

Public API

ExperienceLoopBlock(AIBlock[ExperienceLoopInput, ExperienceLoopOutput, dict])

Field Type Default
name str 'experience_loop'
state dict field(default_factory=dict)
resource_bounds ResourceBounds field(default_factory=ResourceBounds)
usage ResourceUsage field(default_factory=ResourceUsage)
db_path str \| None None
storage_backend str 'auto'
default_tenant_id str 'default'
redact_sensitive bool True
learning_layer Any \| None None

Methods:

infer(data: ExperienceLoopInput) -> Result[ExperienceLoopOutput]

list_patterns() -> dict[str, Any]

Surface the deterministic applied-pattern + skill catalog (not an op).

to_learning_dataset(tenant_id: str | None = None, task_type: str | None = None, dataset_name: str = '')

Convert recorded experiences into a learning_layer Dataset.

ExperienceLoopInput(BaseModel)

Field Type Default
op Literal['record', 'query_similar', 'get_recommendation', 'get_competence', 'list_task_types', 'get_transition_table', 'should_escalate', 'replay_experience', 'prune_old', 'get_stats', 'get_data_consumption', 'reset', 'compact', 'export_data', 'delete_data', 'sync_learning_layer', 'capabilities'] 'record'
state_before dict Field(default_factory=dict)
action_taken dict Field(default_factory=dict)
outcome str ''
reward float 0.0
state_after dict Field(default_factory=dict)
task_type str ''
timestamp float 0.0
state_features dict Field(default_factory=dict)
action_type str ''
k int 5
threshold float 0.8
min_episodes int 10
domain_risk_class Literal['low_stakes', 'elevated', 'high_stakes'] 'low_stakes'
max_age_seconds float 0.0
compact_keep_records int 0
bin_count int 10
tenant_id str 'default'
redact_sensitive bool True
redact_keys list[str] Field(default_factory=lambda: ['api_key', 'authorization', 'email', 'name', 'password', 'phone', 'secret', 'ssn', 'token'])
include_raw_records bool False
dataset_name str ''
metadata dict Field(default_factory=dict)
outcome_source str ''
outcome_verified_by str ''
anti_gaming_policy str ''
request_id str \| None None
task_id str \| None None
run_id str \| None None

ExperienceLoopOutput(BaseModel)

Field Type Default
op str required
recorded bool False
similar_experiences list[dict] Field(default_factory=list)
recommendation dict Field(default_factory=dict)
competence float 0.0
confidence_interval list[float] Field(default_factory=list)
episode_count int 0
should_escalate bool False
task_types list[str] Field(default_factory=list)
transition_table dict Field(default_factory=dict)
stats dict Field(default_factory=dict)
data_consumption dict Field(default_factory=dict)
exported_records list[dict] Field(default_factory=list)
pruned_count int 0
deleted_count int 0
metadata dict Field(default_factory=dict)
degraded bool False
degradation_reason str \| None None
persistence_status Literal['ok', 'missing', 'degraded_fresh_start', 'write_degraded'] 'ok'
escalation_block_reason str ''
completion_state Literal['verified', 'qualified-draft', 'blocked-escalated'] 'qualified-draft'
warning_card dict \| None None
evidence list[str] Field(default_factory=list)
request_id str \| None None
task_id str \| None None
run_id str \| None None

MCP Tools

Operation Source
get_competence experience_loop_mcp
get_recommendation experience_loop_mcp
query_similar experience_loop_mcp
get_stats experience_loop_mcp
should_escalate experience_loop_mcp
list_task_types experience_loop_mcp
get_transition_table experience_loop_mcp
get_data_consumption experience_loop_mcp
capabilities experience_loop_mcp
record experience_loop_mcp
replay_experience experience_loop_mcp
prune_old experience_loop_mcp
reset experience_loop_mcp
compact experience_loop_mcp
export_data experience_loop_mcp
delete_data experience_loop_mcp
sync_learning_layer experience_loop_mcp
low_stakes experience_loop_mcp
elevated experience_loop_mcp
high_stakes experience_loop_mcp