PostgreSQL-based structured memory system for OpenClaw agents with pre/post-compaction integration, dual-write capability, and full context preservation. Primary storage with markdown backup. Supports multi-agent deployments.
# When user asks about past work
results = mem.search("Rasa", days=7, min_importance=0.6)
# Full-text search on raw exchanges
exchanges = mem.search_exchanges("Docker error", days=1)
# Recent high-importance observations
obs = mem.get_recent_observations(hours=24, min_importance=0.8)
🗣️ Natural Language Query Builder (v2.4)
Ask questions in plain English - no SQL required!
⚠️ Ollama Required: Natural Language queries require a local Ollama instance. Install with: brew install ollama && ollama serve
# Simple queries
pg-memory query "show me high-importance unresolved projects from last week"
pg-memory query "what did I work on yesterday"
pg-memory query "find all observations tagged with docker and error"
pg-memory query "list active projects from this month"
# With options
pg-memory query "top 10 recent decisions" --sql # Show the SQL
pg-memory query "errors with high importance" --explain # Preview without executing
Python API
python
from pg_memory import ask, query_nl
# Quick query
result = ask("show me high-importance unresolved projects from last week")
print(f"Found {result.result_count} observations")
for obs in result.results:
print(f" - {obs['title']} ({obs['status']})")
# Get explanation
explanation = ask("errors from yesterday", explain=True)
print(explanation) # Shows SQL translation
# Access results
print(result.sql_query) # Generated SQL
print(result.params) # Query parameters
print(result.execution_time_ms) # Performance metric
print(result.interpretation) # What was understood
# Time-based
"show me what I captured yesterday"
"find all observations from last 30 days"
"high-importance items from this week"
# Status-based
"list all unresolved projects"
"show ongoing tasks"
"recently resolved observations"
# Combined
"high-importance errors from last week"
"active projects tagged with urgent"
"top 5 unresolved decisions from this month"
"oldest ongoing observations"
Query Interpretation
The system returns a human-readable interpretation:
text
Query: show me high-importance unresolved projects from last week
Interpretation: Search: from last week; with status: active, ongoing; importance: 70-100%; type: project
Results: 3 (in 12.3ms)
📊 Search Capabilities
Types of Search
Method
Use Case
Speed
search()
Find curated observations
Instant (GIN indexes)
search_exchanges()
Search raw conversation
Fast (tsvector)
get_recent_observations()
Recent important items
Instant
vsearch()
Semantic similarity
Slow (requires pgvector)
Example Queries
sql
-- Find important decisions
SELECT * FROM observations
WHERE obs_type = 'decision'
AND importance_score > 0.8
AND created_at > NOW() - INTERVAL '7 days';
-- Search raw exchanges
SELECT * FROM search_exchanges('Rasa port', 'arty', 7);
-- Recent activity across sessions
SELECT * FROM recent_activity;
-- Session summary with metrics
SELECT * FROM session_summary;
Day 1: Create with status='ongoing', started_at=2 days ago
Day N: Resolve with status='resolved', resolved_at=now
Duration tracked automatically
Creating Multi-Day Work
python
from pg_memory_v2 import AgentMemory
from datetime import datetime, timedelta
mem = AgentMemory()
# Day 1: Start work
obs_id = mem.capture_observation(
session_key="caption_day1",
obs_type="ongoing", # Mark as work-in-progress
title="[WIP] Caption System",
content="Starting Y position tests...",
importance=0.9,
tags=["caption", "video", "wip"],
status="ongoing", # Not yet complete
started_at=datetime.now() - timedelta(days=1)
)
# Returns UUID: '97d3f2ee-319b-44dd-9c84-e8d1f90cac73'
Tracking Progress
python
# Get all active/ongoing work
active = mem.get_active_observations(min_importance=0.5)
# Returns with duration_hours calculated automatically
# Get timeline view
timeline = mem.get_observation_timeline(days=30)
# Returns observations with status and duration
Completing Work
python
# Day N: Mark complete
mem.resolve_observation(
observation_id='97d3f2ee-319b-44dd-9c84-e8d1f90cac73',
final_title="Caption System Complete",
final_content="Final: Y=1360, green #00FF00, 58px font"
)
# Sets status='resolved', resolved_at=NOW(), calculates duration
Status Values
Status
Meaning
When to Use
active
Default, newly created
Single-day tasks
ongoing
Work in progress
Multi-day features
resolved
Work completed
After resolve_observation()
superseded
Replaced by newer
Outdated observations
Database Columns
sql
ALTER TABLE observations ADD COLUMN:
status VARCHAR(20) DEFAULT 'active', -- active, ongoing, resolved, superseded
started_at TIMESTAMP WITH TIME ZONE, -- When work began
resolved_at TIMESTAMP WITH TIME ZONE; -- When work completed (NULL if ongoing)
Temporal Views
sql
-- Active work (daily standup view)
SELECT * FROM active_observations;
-- Completed work with duration (weekly review)
SELECT * FROM completed_observations
WHERE resolved_at > NOW() - INTERVAL '7 days';
🔄 Dual-Write Strategy
Why Dual-Write?
PostgreSQL = Primary (fast, queryable, structured)
Markdown = Safety net (human-readable, works if pgdb down)
Configurable Retention
yaml
# Keep 7 days of markdown backup
retention_days: 7
# After 7 days, markdown files are auto-deleted
# PostgreSQL retention: Indefinite (disk limited)
Fallback Behavior
If PostgreSQL is down:
Write to markdown only (emergency mode)
Set status flag
Recover when pgdb back up
If both fail:
Context is lost (rare edge case)
Agent starts fresh
👥 Multi-Agent Support
sql
-- Each agent has isolated sessions
SELECT * FROM sessions
WHERE agent_id = 'arty'
AND started_at > NOW() - INTERVAL '24 hours';
-- Cross-agent search (if needed)
SELECT * FROM observations
WHERE tags @> ARRAY['shared-project'];
📈 Migration from Markdown
Existing Files
python
from pg_memory_v2 import AgentMemory
mem = AgentMemory()
# Import existing markdown memories
mem.import_legacy_markdown(
glob_pattern="~/.openclaw/workspace/memory/2026-02-*.md",
parse_strategy="auto" # Extract headers, bullet lists
)
# Track imports
SELECT * FROM memory_imports
WHERE import_status = 'success';
Archive After Import
bash
# Archive imported files to safe location
mv ~/.openclaw/workspace/memory/archive/
⚡ Performance Tips
Indexes
sessions(agent_id, started_at DESC) - Fast agent queries
# AgentMemory uses connection pooling by default
mem = AgentMemory(max_connections=10)
# Auto-reconnect if connection drops
if not mem._ensure_connection():
# Retries in background
Priority: ⛔ CRITICAL - REMEMBER FOREVER
Directive: "All new projects or tasks assigned should have an observation created if one does not exist."
Why This Matters
The Problem:
Agent receives new task/project
Doesn't document it immediately
Context gets compacted
Agent forgets the work existed
User thinks agent dropped the ball
The Solution:
Every new assignment gets automatic observation
Check existing → Create if missing → Return
High importance (0.9) by default
Tagged for easy filtering
Never rely on "mental notes"
Auto-Creation Methods
Method 1: ensure_observation_exists()
python
from pg_memory_v2 import AgentMemory
mem = AgentMemory()
# Check if observation exists, create if not
result = mem.ensure_observation_exists(
project_name="Gaming Trends Article",
project_location="content/posts/gaming/",
assigned_by="Skip",
key_details="Write gaming trends for Feb 27",
next_steps="Research Steam, PC Gamer"
)
print(result["was_created"]) # True = created, False = already exists
print(result["message"])
# "Observation created per protocol for 'Gaming Trends Article'"
Template Applied:
markdown
## Gaming Trends Article
**Assigned**: 2026-02-27 18:14 EST
**Status**: Active
**Location**: content/posts/gaming/
**Key Details**: Write gaming trends for Feb 27
**Next Steps**: Research Steam, PC Gamer
*Observation created per protocol*
Method 2: auto_capture_project()
python
from pg_memory_v2 import AgentMemory
mem = AgentMemory()
# Auto-extract project name from task description
result = mem.auto_capture_project(
task_description="""Gaming Trends - Feb 27
Write article about Baldur's Gate patch, Marathon, Space Marine 2.
Make it Gen X tone.""",
project_location="content/posts/gaming/",
assigned_by="Skip"
)
# Automatically creates:
# - Project name: "Gaming Trends - Feb 27"
# - Key details: "Write article about..."
# - Tags: ['project:Gaming Trends', 'observation', 'assigned', 'active']
Method 3: check_observation_exists()
python
# Just check if project already has observation
exists = mem.check_observation_exists("Gaming Trends Article")
# Returns: True or False
Quick Convenience Functions
python
from skills.pg_memory.scripts.pg_memory import (
ensure_observation,
check_observation,
auto_capture
)
# One-liner convenience
result = ensure_observation(
"Gaming Trends Article",
"content/posts/gaming/",
key_details="Write about Baldur's Gate patch",
next_steps="Research PC Gamer, Steam"
)
# Quick check
if check_observation("Gaming Trends Article"):
print("Already documented")
else:
print("Needs observation")
# Auto-detection
result = auto_capture("""New task:
Build YouTube channel @PlotTwist
Create profile, banner, thumbnails
Upload first video""")
Observation Protocol - Rules
Every new assignment gets an observation
Check first - don't create duplicates
High importance (0.9) - assignments are always important
Include metadata - location, assigner, key details, next steps
Tag properly - project:name, observation, assigned, active
Never assume - write it down even if it seems obvious
Database Schema
sql
-- Observations are created with:
-- - content_type = 'observation'
-- - importance_score = 0.9
-- - tags = ARRAY['project:Name', 'observation', 'assigned', 'active']
-- - metadata includes project_location, assigned_by, etc.
SELECT * FROM observations
WHERE tags @> ARRAY['observation', 'assigned']
AND importance_score >= 0.9
AND created_at > NOW() - INTERVAL '30 days';
