Resilient Memory-Retrieved Schema

The problem

Retrieving specific information from messy or inconsistent data records can be difficult because small errors can cause the process to fail entirely. It is frustrating when a system gives up just because it hit a minor snag.

What it does

It looks through messy data queries and automatically tries multiple times to find the right information. It keeps track of which pieces of data were successfully retrieved and which ones failed.

Why it matters

It ensures that the system stays persistent in finding information even when the input data is messy.

Validation

It was run inside an isolated container with no network access. This is the exact command and the real output it produced — captured process output, not written by a model.

$ python3 resilient_schema.py
✅ query1 succeeded in 1 retries (weight: 0.50)
Data: {'name': 'Alice', 'age': 30, 'email': 'alice@example.com'}
❌ query2 failed after 3 retries: Missing email
✅ query3 succeeded in 1 retries (weight: 0.50)
Data: {'name': 'Charlie', 'age': 40, 'email': 'charlie@example.com'}

Success rate: 67%
Weighted success score: 1.00

A screenshot of that run.

A clean run proves this does what is shown above, in a CPU-only sandbox. It is a small research demo — not a production tool, and nothing here was published anywhere.

The code

All of it — 80 lines, one file, standard library only.

import random
import time
import difflib

# Schema definition
SCHEMA = {
    'name': str,
    'age': int,
    'email': str
}

# Simulated memory store
MEMORY_STORE = {
    'query1': {'name': 'Alice', 'age': 30, 'email': 'alice@example.com'},
    'query2': {'name': 'Bob', 'age': 25},
    'query3': {'name': 'Charlie', 'age': 40, 'email': 'charlie@example.com'}
}

def retrieve_from_memory(query):
    # Simulate 30% failure rate
    if random.random() < 0.3:
        return None
    return MEMORY_STORE.get(query)

def retrieve_with_retry(query, max_retries=3, delay=1):
    for attempt in range(1, max_retries + 1):
        data = retrieve_from_memory(query)
        if data is not None:
            return data, attempt
        time.sleep(delay)
    # Fuzzy matching for closest key
    keys = list(MEMORY_STORE.keys())
    matches = difflib.get_close_matches(query, keys, n=1, cutoff=0.6)
    closest_key = matches[0] if matches else None
    if closest_key:
        data = MEMORY_STORE.get(closest_key)
        return data, max_retries
    else:
        return None, max_retries

def validate_data(data):
    if not data:
        return False, "No data"
    for field, field_type in SCHEMA.items():
        if field not in data:
            return False, f"Missing {field}"
        if not isinstance(data[field], field_type):
            return False, f"Invalid {field} type"
    return True, ""

def process_queries(queries):
    results = []
    for q in queries:
        data, retries = retrieve_with_retry(q)
        valid, msg = validate_data(data)
        results.append({
            'query': q,
            'success': valid,
            'retries': retries,
            'weight': 1/(retries+1) if valid else 0,
            'data': data,
            'error': msg
        })
    return results

if __name__ == "__main__":
    queries = ['query1', 'query2', 'query3', 'quey1']  # Added typo for fuzzy match test
    results = process_queries(queries)
    # Output results
    for res in results:
        if res['success']:
            print(f"\u2705 {res['query']} succeeded in {res['retries']} retries (weight: {res['weight']:.2f})")
            print(f"Data: {res['data']}")
        else:
            print(f"\u274c {res['query']} failed after {res['retries']} retries: {res['error']}")
    # Summary metrics
    success_rate = sum(1 for r in results if r['success']) / len(results)
    weighted_score = sum(r['weight'] for r in results)
    print(f"\nSuccess rate: {success_rate:.0%}")
    print(f"Weighted success score: {weighted_score:.2f}")