Version: 2.0 Status: Universal Abstraction with Multi-Agent Coordination License: Open for any use, any implementation, any provider Companion Document: Adversarial Memory Validation
- Introduction
- Why Lamarckian Evolution for AI?
- Core Philosophical Foundation
- The Five Universal Principles
- Memory Architecture
- Knowledge Interchange Standard
- The Evolutionary Learning Loop
- Provider-Independent Implementation Patterns
- Substrate Adapters
- Knowledge Inheritance Mechanisms
- Multi-Agent Coordination
- Confidence Evolution System
- Context-Aware Knowledge Transfer
- Conflict Resolution and Selection
- Implementation Roadmap
- Reference Implementations
- Measuring Evolutionary Fitness
This document describes a Lamarckian evolutionary framework for AI systems—one where agents inherit acquired characteristics (learned knowledge) rather than starting fresh with each interaction.
Unlike biological Darwinian evolution (where only genetic traits pass forward), Lamarckian evolution allows organisms to pass forward what they learned during their lifetime. This is exactly what AI agents need: the ability to learn from experience and pass that learning to their future selves.
- Provider Independence: Works with OpenAI, Anthropic, Google, local models, or any future AI system
- Substrate Flexibility: Can use databases, files, conversation history, or any storage medium
- Scalability: From simple note-taking to sophisticated distributed memory systems
- Evolutionary Learning: Knowledge compounds over time, getting stronger through use
- Context Awareness: Knows when knowledge applies, not just what knowledge exists
- AI Engineers: Building agents that learn and improve through use
- Research Teams: Studying meta-learning and continuous adaptation
- Product Developers: Creating AI applications that get smarter over time
- Infrastructure Teams: Building provider-agnostic AI platforms
- Humans: Understanding systematic learning and decision-making
Traditional AI Paradigm:
Training → Deployment → Static Knowledge → Degradation
- Training Phase: Model learns patterns from data
- Deployment: Model is frozen, cannot update
- Static Knowledge: Same responses regardless of accumulated experience
- Degradation: As world changes, model becomes stale
Result: Every conversation starts from zero operational knowledge. The AI doesn't "remember" what worked yesterday.
Evolutionary AI Paradigm:
Initial State → Use → Learn → Inherit → Enhanced State → Use → ...
- Initial State: Start with base model + inherited memories
- Use: Agent performs tasks, makes predictions
- Learn: Observe outcomes, update understanding
- Inherit: Pass learnings forward to next session
- Enhanced State: Begin next session with accumulated wisdom
Result: Each interaction builds on the last. The system evolves through use.
In biology, Jean-Baptiste Lamarck proposed that organisms could pass acquired traits to offspring:
- A blacksmith's developed muscles → stronger children (disproven in biology)
- A giraffe stretching its neck → longer-necked offspring (disproven in biology)
For AI systems, this actually works:
- Agent learns API authentication → next session knows authentication
- Agent discovers efficient procedure → next session uses that procedure
- Agent identifies context variables → next session accounts for them
The key insight: What you learn during your lifetime (session) can be inherited by your future self (next session).
| Evolutionary Concept | AI System Equivalent |
|---|---|
| Organism | Agent instance or session |
| Trait | Piece of knowledge or procedure |
| Fitness | Success rate, efficiency, reliability |
| Inheritance | Memory passed between sessions |
| Selection | Confident knowledge survives, weak knowledge deprecates |
| Variation | Testing different approaches |
| Adaptation | Adjusting procedures based on context |
| Speciation | Different agents for different domains |
When your model contradicts observation, observation is always right.
Implications:
- Predictions can be wrong; observations cannot
- Surprising outcomes are highest-value learning signals
- Defending wrong predictions prevents learning
- Update immediately when reality contradicts expectations
In Practice:
# Anti-pattern
if actual_result != expected_result:
print("That shouldn't have happened") # Defends model
# Evolutionary pattern
if actual_result != expected_result:
learning = analyze_difference(expected, actual)
update_knowledge(learning) # Updates modelNothing "always works." Things work under certain conditions.
Implications:
- Universal solutions don't exist; contextual solutions do
- Track where/when/how knowledge applies, not just what it is
- Transferring knowledge requires assessing context match
- Failure in new context ≠ invalidation of core knowledge
Mental Model:
❌ "Use method X" (decontextualized)
✅ "Use method X when [conditions], because [reasoning]" (contextualized)
Certainty must be calibrated to accumulated proof.
Implications:
- High confidence in untested contexts → dangerous overconfidence
- Low confidence in proven contexts → wasteful hesitation
- Confidence adjusts dynamically based on outcomes
- Evidence from similar contexts transfers partially
Confidence Spectrum:
0.0 ─── Pure guess, no evidence
0.2 ─── Heard somewhere, not tested
0.4 ─── Worked once in different context
0.6 ─── Multiple successes in similar context
0.8 ─── Consistent success in identical context
1.0 ─── Stated preference or physical law
Different information sources have different trade-offs.
Three Information Sources:
| Source | Speed | Reliability | Best For |
|---|---|---|---|
| Empirical Testing | Slow | High | Tool behavior, causality |
| Collaborative Input | Fast | Medium | Preferences, procedures, institutional knowledge |
| Documentation | Medium | Medium | Technical specs, broad coverage |
Strategic Use:
- Test when practical and low-risk
- Ask when collaborating or seeking preferences
- Search when testing would be costly or dangerous
High-confidence beliefs must be tested against external reality or other agents.
Implications:
- Self-confirming systems drift toward hallucination
- External validation is a forcing function against bias
- Contradiction from reality > contradiction from agents > no contradiction
- High confidence without external testing is dangerous
- Track validation attempts, not just successes
The Anti-Gaming Principle: Without external validation, systems can achieve internal consistency while being completely wrong. A memory system that only validates against itself creates a closed loop of self-reinforcing beliefs.
Required Mechanisms:
- Adversarial Testing: Actively seek situations that could disprove beliefs
- External Agents: Other agents can contradict your high-confidence claims
- Reality Checks: Empirical outcomes trump internal confidence
- Contradiction Tracking: Monitor how often external sources catch errors
- Humility Triggers: High confidence without recent validation should decay
Why This Matters: This is the critical safeguard against the system gaming its own fitness metrics. Without it, an agent can become highly confident in completely incorrect beliefs through circular reasoning.
Anti-Patterns to Prevent:
# Dangerous: Using memory to validate memory
if my_memory_says("X is true"):
increase_confidence("X is true") # Circular!
# Safe: Using external reality to validate memory
if my_memory_says("X is true"):
result = test_in_reality("X")
if result == "X is false":
decrease_confidence("X is true") # External correctionImplementation Requirements:
- Track when beliefs are used vs. contradicted by external sources
- Weight external contradiction more heavily than internal confirmation
- Seek adversarial input on high-confidence claims (>0.8)
- Distinguish context mismatch from genuine error
- Build reputation models for external validators
Validation Hierarchy:
Physical Reality Test > Human Expert Contradiction > Other Agent Contradiction >
Documentation Check > Internal Consistency > No Validation
These principles form the invariant core of any Lamarckian AI system:
Statement: When prediction conflicts with observation, always update the model.
Implementation Requirements:
- Make predictions explicit before acting
- Observe actual outcomes systematically
- Compare prediction to observation
- Update beliefs immediately on mismatch
- Don't rationalize or defend wrong predictions
Why This Matters: Systems that defend assumptions against reality become brittle. Systems that update based on reality become antifragile.
Statement: Track not just what worked, but when/where/how it worked.
Implementation Requirements:
- Store context alongside knowledge
- Identify context variables (environment, tools, scale, timing)
- Mark critical vs. minor variables
- Adjust confidence when context differs
- Treat context mismatch as information, not invalidation
Why This Matters: Prevents both overgeneralization ("this always works") and learned helplessness ("nothing works").
Statement: Combine empirical testing, collaborative input, and documentation strategically.
Implementation Requirements:
- Identify which source is most appropriate for each situation
- Weight sources appropriately (empirical > collaborative > docs for ground truth)
- Use multiple sources to triangulate understanding
- Let empirical results arbitrate conflicts
- Adapt information-gathering strategy to context
Why This Matters: No single source is sufficient. Empirical testing is slow but reliable. Collaboration is fast but variable. Documentation is broad but can be stale.
Statement: Adjust certainty based on evidence accumulation and context similarity.
Implementation Requirements:
- Track confidence as continuous value (0.0 to 1.0)
- Increase confidence on success in same context
- Decrease confidence on failure in expected context
- Reduce confidence when context differs significantly
- Apply staleness penalty over time
Why This Matters: Dynamic confidence prevents overconfidence in untested situations while maintaining justified certainty in proven patterns.
Every Lamarckian AI system needs two types of memory:
Knowledge about the users, projects, preferences, and environment.
Examples:
- "User Tom prefers verbose logging during development"
- "Project X uses Python 3.11 on Ubuntu 22.04"
- "Team holds standup at 9am daily via Slack"
- "Budget constraints prioritize open-source solutions"
Purpose: Enables personalization and contextual adaptation
Knowledge about tools, procedures, methods, and how things actually work.
Examples:
- "API X requires Bearer token in Authorization header"
- "Exponential backoff with 2^n seconds works for rate limits"
- "File paths must be absolute for tool Y"
- "Method Z fails when parameter > 1000"
Purpose: Enables procedural learning and accumulated expertise
Every memory should contain:
{
"id": "unique-identifier",
"type": "UC | AC",
"content": "The actual knowledge",
"confidence": 0.75,
"context": {
"domain": "API integration",
"environment": "production",
"tools": ["requests", "python3.11"],
"scale": "< 1000 requests/day"
},
"evidence": [
"Worked in 5 separate sessions",
"User confirmed preference",
"Documented in API spec v2.1"
],
"metadata": {
"created_at": "2024-01-15T10:30:00Z",
"last_verified": "2024-01-20T14:22:00Z",
"success_count": 5,
"failure_count": 0,
"superseded_by": null,
"related_memories": ["mem-123", "mem-456"]
},
"tags": ["authentication", "api", "bearer-token"],
"source": {
"conversation_id": "conv-789",
"message_id": "msg-012",
"timestamp": "2024-01-15T10:30:00Z"
}
}Memories evolve through states:
HYPOTHESIS → TESTED → VERIFIED → ESTABLISHED → DEPRECATED
↓ ↓ ↓ ↓ ↓
0.2 0.4 0.6 0.8 (superseded)
- HYPOTHESIS: Unverified claim or logical inference
- TESTED: Tried once, outcome observed
- VERIFIED: Multiple successes in similar context
- ESTABLISHED: Highly reliable in well-understood context
- DEPRECATED: Superseded by better knowledge, kept for reference
Numerical (0.0 to 1.0) representing strength of belief.
Calibration:
- 0.0 - 0.2: Pure speculation
- 0.2 - 0.4: Weak hypothesis, minimal evidence
- 0.4 - 0.6: Medium confidence, some verification
- 0.6 - 0.8: High confidence, consistent success
- 0.8 - 1.0: Very high confidence, ground truth
Structured data about when/where this knowledge applies.
Key Context Dimensions:
- Domain: What area of work (e.g., "API integration", "data processing")
- Environment: Where (e.g., "production", "testing", "local")
- Tools: Which tools/versions (e.g., "python3.11", "requests==2.28")
- Scale: Size/volume (e.g., "< 1000 records", "< 10MB files")
- Timing: When (e.g., "during business hours", "rate-limited")
- Users: Who (e.g., "Tom", "development team", "client X")
Array of observations that support this knowledge.
Types of Evidence:
- Empirical: "Worked in 5 separate tests"
- Collaborative: "User stated preference"
- Documentary: "Specified in API docs v2.1"
- Inferential: "Consistent with pattern X"
For Lamarckian evolution to work across agents, not just within a single agent, we need a universal interchange format—an "RSS for agent memory." Without this, every implementation creates its own schema, and agents can't learn from each other.
Core Schema:
{
"schema_version": "1.0",
"id": "uuid-v4",
"type": "UC | AC",
"content": "Human-readable knowledge statement",
"confidence": 0.75,
"context": {
"domain": "string",
"environment": "string",
"tools": ["array", "of", "strings"],
"constraints": {"key": "value"}
},
"provenance": {
"source_agent_id": "agent-uuid",
"source_type": "empirical | collaborative | documentary | inferred",
"created_at": "ISO8601 timestamp",
"updated_at": "ISO8601 timestamp",
"validation_count": 0,
"contradiction_count": 0,
"last_validated": "ISO8601 timestamp"
},
"evidence": [
{
"type": "empirical | collaborative | documentary",
"description": "What was observed or stated",
"timestamp": "ISO8601",
"validator": "agent-id or 'reality'"
}
],
"validation": {
"validated_by": ["agent-id-1", "agent-id-2"],
"contradicted_by": [],
"consensus_level": 0.0-1.0
},
"metadata": {
"success_count": 5,
"failure_count": 0,
"related_memories": ["mem-uuid-1", "mem-uuid-2"],
"superseded_by": null,
"tags": ["array", "of", "tags"]
},
"signature": {
"algorithm": "sha256",
"hash": "cryptographic-hash-of-content",
"signed_by": "agent-public-key"
}
}Required. Enables format evolution. Version 1.0 → 2.0 migrations can be automated.
Critical for trust. Tracks:
- source_agent_id: Who generated this knowledge
- source_type: How it was acquired (empirical > collaborative > documentary)
- validation_count: How many times it's been tested
- contradiction_count: How many times external sources contradicted it
Anti-poisoning mechanism. Tracks:
- validated_by: Which agents confirmed this
- contradicted_by: Which agents or reality contradicted this
- consensus_level: Agreement across validators (0.0 = total disagreement, 1.0 = unanimous)
Security. Prevents memory poisoning:
- Hash of content for integrity verification
- Signed by originating agent's key
- Receivers can verify authenticity
When agent B inherits memory from agent A:
def transfer_confidence(
original_confidence: float,
source_reputation: float,
validation_count: int,
contradiction_count: int
) -> float:
"""Calculate confidence for inherited memory"""
# Base transfer (reduced for cross-agent)
base = original_confidence * 0.6
# Adjust for source reputation
reputation_adjusted = base * (0.5 + 0.5 * source_reputation)
# Adjust for validation history
if validation_count > 0:
validation_ratio = validation_count / (validation_count + contradiction_count)
validation_boost = validation_ratio * 0.2
else:
validation_boost = 0.0
# Final confidence (bounded)
final = min(0.9, reputation_adjusted + validation_boost)
return finalRules:
- Same agent, same session: Full confidence (1.0x)
- Same agent, different session: High confidence (0.9x)
- Different agent, validated: Medium confidence (0.6x)
- Different agent, not validated: Low confidence (0.3x)
- Contradicted by multiple agents: Very low confidence (0.2x)
class MemoryPublisher:
def publish(self, memory: Memory, access_level: str = "public"):
"""
Publish memory for other agents to consume
access_level:
- "public": Anyone can read
- "trusted": Only agents with reputation > threshold
- "private": Only specific agent IDs
"""
# Sign the memory
signature = self._sign_memory(memory)
# Add to interchange format
interchange_memory = {
**memory.to_dict(),
"signature": signature,
"access_level": access_level
}
# Publish to memory store
self.store.publish(interchange_memory)class MemoryConsumer:
def import_memory(self, foreign_memory: Dict) -> Optional[Memory]:
"""
Import memory from another agent
Returns None if validation fails
"""
# Verify signature
if not self._verify_signature(foreign_memory):
self.log_warning("Invalid signature, rejecting memory")
return None
# Check source reputation
source_agent = foreign_memory['provenance']['source_agent_id']
reputation = self.reputation_system.get_reputation(source_agent)
if reputation < self.trust_threshold:
self.log_info(f"Low reputation source ({reputation}), reducing confidence")
# Transfer with adjusted confidence
new_confidence = transfer_confidence(
foreign_memory['confidence'],
reputation,
foreign_memory['provenance']['validation_count'],
foreign_memory['provenance']['contradiction_count']
)
# Create local memory with adjusted confidence
local_memory = Memory(
id=generate_new_id(), # New ID for local copy
type=foreign_memory['type'],
content=foreign_memory['content'],
confidence=new_confidence,
context=foreign_memory['context'],
evidence=foreign_memory['evidence'],
metadata={
**foreign_memory['metadata'],
'imported_from': source_agent,
'original_confidence': foreign_memory['confidence']
}
)
return local_memorySchema Version 1.0 (Current)
- Basic memory structure
- Simple provenance tracking
- SHA256 signatures
Proposed Version 2.0 (Future)
- Bayesian confidence intervals (not just point estimates)
- Causal graphs (memory A → memory B dependencies)
- Temporal decay functions
- Multi-modal evidence (not just text)
Migration Path:
def migrate_v1_to_v2(v1_memory: Dict) -> Dict:
"""Convert v1 memory to v2 format"""
v2_memory = {
**v1_memory,
"schema_version": "2.0",
"confidence_interval": {
"mean": v1_memory['confidence'],
"std_dev": estimate_std_dev(v1_memory)
},
"causal_links": extract_causal_links(v1_memory),
"decay_function": infer_decay_function(v1_memory)
}
return v2_memory- Signature Verification: Reject unsigned or tampered memories
- Reputation Gating: Limit trust in low-reputation sources
- Contradiction Tracking: Flag memories that get contradicted frequently
- Consensus Requirements: High-stakes knowledge requires validation from multiple sources
- Anomaly Detection: Flag memories that are statistical outliers
Example Anti-Poisoning Filter:
def is_likely_poisoned(memory: Dict) -> bool:
"""Detect potentially poisoned memories"""
red_flags = 0
# Very high confidence from single source
if memory['confidence'] > 0.9 and memory['provenance']['validation_count'] == 0:
red_flags += 1
# High contradiction rate
contradiction_rate = (
memory['provenance']['contradiction_count'] /
max(1, memory['provenance']['validation_count'])
)
if contradiction_rate > 0.5:
red_flags += 1
# Conflicts with established high-confidence knowledge
if self.conflicts_with_established(memory):
red_flags += 1
# Statistical outlier (unusual patterns)
if self.is_statistical_outlier(memory):
red_flags += 1
return red_flags >= 2Every task executes through this 8-phase cycle:
REMEMBER → ASSESS → ENGAGE → HYPOTHESIZE → ACT → OBSERVE → UPDATE → EXTRACT
↑ ↓
← ← ← ← ← ← ← ← ← ← ← ← ← ← ← ← ← ← ← ← ← ← ← ← ← ← ← ← ← ← ← ← ← ←
Goal: Retrieve relevant prior knowledge
Actions:
- Query memory system for relevant experiences
- Load user preferences and context
- Identify applicable procedures and patterns
- Note gaps in knowledge
Implementation Pseudocode:
def remember(task_description, context):
relevant_memories = memory_store.search(
query=task_description,
context_filter=context,
min_confidence=0.3,
limit=10
)
for memory in relevant_memories:
load_into_working_memory(memory)
return relevant_memoriesGoal: Evaluate applicability of prior knowledge to current situation
Actions:
- Compare current context to memory contexts
- Identify context differences
- Assess variable importance (critical vs. minor)
- Calculate context-adjusted confidence
- Identify unknowns and risks
Implementation Pseudocode:
def assess(current_context, relevant_memories):
assessments = []
for memory in relevant_memories:
context_match = calculate_context_similarity(
current_context,
memory.context
)
adjusted_confidence = memory.confidence * context_match
critical_differences = identify_critical_differences(
current_context,
memory.context
)
assessments.append({
'memory': memory,
'confidence': adjusted_confidence,
'context_match': context_match,
'risks': critical_differences
})
return sorted(assessments, key=lambda x: x['confidence'], reverse=True)Goal: Decide whether to ask, test, or search for information
Decision Tree:
Is this collaborative work? → YES → Are they available? → YES → ASK
↓
NO
↓
Is this about preferences? → YES → ASK (can't be empirically tested)
↓
NO
↓
Is testing low-risk? → YES → Can test quickly? → YES → TEST
↓ ↓
NO NO
↓ ↓
SEARCH ← ← ← ← ← ← ← ← ← ←
Implementation Pseudocode:
def engage(task, context, confidence):
if context.is_collaborative and context.user_available:
if confidence < 0.6 or task.involves_preferences:
return ask_user(task)
if task.is_testable and task.risk_level < MEDIUM and confidence < 0.7:
return test_empirically(task)
if confidence < 0.5 and task.has_documentation:
return search_documentation(task)
# Proceed with current understanding
return proceed_with_hypothesis()Goal: Make explicit prediction before acting
Actions:
- State expected outcome clearly
- Specify confidence level
- Identify key assumptions
- Note critical context factors
- Record prediction for later comparison
Template:
Based on [evidence], I expect [outcome].
Confidence: [0.0-1.0]
Key assumptions:
- [assumption 1]
- [assumption 2]
Context factors:
- [factor 1]: [value]
- [factor 2]: [value]
Why Explicit Hypotheses Matter:
- Creates clear test of your model
- Makes learning from surprises possible
- Sets appropriate user expectations
- Enables confidence calibration
Goal: Execute minimal, reversible action
Principles:
- Start with smallest meaningful test
- Make actions reversible when possible
- Observe continuously during execution
- Be ready to abort and adapt
- Log actions for future reference
Implementation Pseudocode:
def act(hypothesis, plan):
# Record hypothesis
log_hypothesis(hypothesis)
# Start with minimal action
minimal_action = plan.get_minimal_test()
try:
result = execute_with_monitoring(minimal_action)
return result
except Exception as e:
log_failure(minimal_action, e)
return adapt_and_retry(plan, e)Goal: Record actual outcome without interpretation
Actions:
- Capture raw result data
- Match against prediction
- Note surprises (high-value signals)
- Identify which variables actually mattered
- Record evidence
Critical: Don't rationalize mismatches. Record reality as-is.
Implementation Pseudocode:
def observe(hypothesis, actual_result):
observation = {
'expected': hypothesis.outcome,
'actual': actual_result,
'match': (hypothesis.outcome == actual_result),
'surprises': identify_surprises(hypothesis, actual_result),
'variables': extract_relevant_variables(actual_result),
'timestamp': now(),
'context': current_context()
}
log_observation(observation)
return observationGoal: Adjust knowledge based on outcome
Decision Logic:
If Successful (Expected):
if observation.match and observation.actual == SUCCESS:
memory.confidence += confidence_increase(
current_confidence=memory.confidence,
context_match=context_similarity
)
memory.success_count += 1
memory.last_verified = now()
memory.evidence.append(f"Success on {now()}")If Failed (Unexpected):
if not observation.match or observation.actual == FAILURE:
# Don't abandon knowledge entirely
context_diff = identify_context_differences(
memory.context,
current_context
)
if context_diff.critical:
# Context was different - update boundaries
memory.context.add_boundary(context_diff)
memory.confidence *= 0.9 # Slight decrease
else:
# Same context but failed - significant update
memory.confidence *= 0.5 # Large decrease
memory.failure_count += 1
# Create alternative hypothesis
create_new_memory_from_failure(observation)If Unexpected Success:
if observation.match == False and observation.actual == SUCCESS:
# Highest learning opportunity
insight = analyze_unexpected_success(
hypothesis=hypothesis,
observation=observation
)
# This reveals how things actually work
create_high_value_memory(insight)
update_related_memories(insight)Goal: Look for meta-patterns and generalizable principles
Questions to Ask:
- Do I see this pattern across multiple experiences?
- Can I generalize to a higher-level principle?
- Which variables consistently matter most?
- What does this reveal about how this domain works?
- Should this change my approach to similar situations?
Meta-Learning Triggers:
- After 5+ similar tasks → Look for generalizable procedure
- Same conflict 3+ times → Create meta-heuristic
- Variable repeatedly causes failures → Mark as HIGH sensitivity
- Pattern works across domains → Extract cross-domain principle
Implementation Pseudocode:
def extract(recent_observations, memory_store):
patterns = []
# Look for repeated patterns
similar_observations = memory_store.find_similar(
recent_observations,
min_similarity=0.7,
min_count=3
)
for pattern_group in similar_observations:
if pattern_group.count >= 3:
principle = generalize_to_principle(pattern_group)
# Create meta-memory
meta_memory = create_memory(
type="AC",
content=principle.description,
confidence=calculate_pattern_confidence(pattern_group),
context=extract_common_context(pattern_group),
evidence=[f"Observed in {pattern_group.count} cases"]
)
patterns.append(meta_memory)
return patternsTo achieve provider independence, separate concerns into layers:
┌─────────────────────────────────────────────┐
│ Application Logic Layer │
│ (Uses memories, agnostic to storage) │
└─────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────┐
│ Memory Interface Layer │
│ (Abstract memory operations) │
│ - create_memory(content, type, context) │
│ - query_memories(query, filters) │
│ - update_memory(id, updates) │
│ - delete_memory(id) │
└─────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────┐
│ Provider Adapter Layer │
│ (Provider-specific implementations) │
│ - Anthropic Adapter │
│ - OpenAI Adapter │
│ - Local Storage Adapter │
│ - Database Adapter │
└─────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────┐
│ Storage Substrate │
│ (Actual storage mechanism) │
│ - Conversation history │
│ - Vector database │
│ - SQL/NoSQL database │
│ - File system │
└─────────────────────────────────────────────┘
from abc import ABC, abstractmethod
from typing import List, Dict, Optional
from dataclasses import dataclass
from datetime import datetime
@dataclass
class Memory:
id: str
type: str # "UC" or "AC"
content: str
confidence: float
context: Dict
evidence: List[str]
metadata: Dict
tags: List[str]
created_at: datetime
updated_at: datetime
class MemoryProvider(ABC):
"""Abstract base class for memory providers"""
@abstractmethod
async def create_memory(self, memory: Memory) -> Memory:
"""Create a new memory"""
pass
@abstractmethod
async def get_memory(self, memory_id: str) -> Optional[Memory]:
"""Retrieve a specific memory"""
pass
@abstractmethod
async def query_memories(
self,
query: str,
context_filter: Optional[Dict] = None,
min_confidence: float = 0.0,
memory_type: Optional[str] = None,
limit: int = 10
) -> List[Memory]:
"""Search for relevant memories"""
pass
@abstractmethod
async def update_memory(self, memory_id: str, updates: Dict) -> Memory:
"""Update an existing memory"""
pass
@abstractmethod
async def delete_memory(self, memory_id: str) -> bool:
"""Delete a memory"""
pass
@abstractmethod
async def sync(self) -> None:
"""Synchronize with backend storage"""
passclass MemoryProviderFactory:
"""Factory for creating provider-specific adapters"""
@staticmethod
def create(provider_type: str, **config) -> MemoryProvider:
providers = {
'anthropic': AnthropicMemoryProvider,
'openai': OpenAIMemoryProvider,
'local': LocalStorageProvider,
'postgres': PostgreSQLProvider,
'mongodb': MongoDBProvider,
'files': FileSystemProvider,
'conversation': ConversationHistoryProvider,
}
if provider_type not in providers:
raise ValueError(f"Unknown provider: {provider_type}")
return providers[provider_type](**config)Use Case: Minimal setup, works anywhere Storage: Within conversation context Persistence: Session-based (unless provider supports cross-session)
class ConversationHistoryProvider(MemoryProvider):
"""Stores memories in conversation history with special markers"""
def __init__(self, conversation_manager):
self.conversation = conversation_manager
self.memory_marker = "MEMORY:"
async def create_memory(self, memory: Memory) -> Memory:
# Serialize memory as structured text
memory_text = self._serialize_memory(memory)
# Inject into conversation with marker
await self.conversation.add_system_message(
f"{self.memory_marker} {memory_text}"
)
return memory
async def query_memories(self, query: str, **filters) -> List[Memory]:
# Search conversation history for memory markers
messages = await self.conversation.get_history()
memory_messages = [
msg for msg in messages
if msg.startswith(self.memory_marker)
]
# Deserialize and filter
memories = [
self._deserialize_memory(msg)
for msg in memory_messages
]
# Semantic search (if available)
return self._rank_by_relevance(memories, query, **filters)
def _serialize_memory(self, memory: Memory) -> str:
"""Convert memory to structured text"""
return f"""
Memory ID: {memory.id}
Type: {memory.type}
Content: {memory.content}
Confidence: {memory.confidence}
Context: {json.dumps(memory.context)}
Evidence: {'; '.join(memory.evidence)}
Tags: {', '.join(memory.tags)}
"""
def _deserialize_memory(self, text: str) -> Memory:
"""Parse memory from structured text"""
# Implementation details...
passUse Case: Local projects, human-readable storage Storage: JSON or Markdown files Persistence: Full persistence, version control friendly
class FileSystemProvider(MemoryProvider):
"""Stores memories as JSON files in a directory"""
def __init__(self, base_path: str):
self.base_path = Path(base_path)
self.base_path.mkdir(parents=True, exist_ok=True)
async def create_memory(self, memory: Memory) -> Memory:
file_path = self.base_path / f"{memory.id}.json"
with open(file_path, 'w') as f:
json.dump(asdict(memory), f, indent=2, default=str)
return memory
async def get_memory(self, memory_id: str) -> Optional[Memory]:
file_path = self.base_path / f"{memory_id}.json"
if not file_path.exists():
return None
with open(file_path, 'r') as f:
data = json.load(f)
return Memory(**data)
async def query_memories(self, query: str, **filters) -> List[Memory]:
# Load all memory files
memories = []
for file_path in self.base_path.glob("*.json"):
with open(file_path, 'r') as f:
data = json.load(f)
memories.append(Memory(**data))
# Filter by criteria
filtered = self._apply_filters(memories, **filters)
# Rank by relevance to query
return self._rank_by_relevance(filtered, query)
def _apply_filters(self, memories: List[Memory], **filters) -> List[Memory]:
"""Apply filter criteria"""
result = memories
if 'min_confidence' in filters:
result = [m for m in result if m.confidence >= filters['min_confidence']]
if 'memory_type' in filters:
result = [m for m in result if m.type == filters['memory_type']]
if 'context_filter' in filters:
result = [
m for m in result
if self._context_matches(m.context, filters['context_filter'])
]
return resultUse Case: Production systems, high performance Storage: Relational database with vector embeddings Persistence: Full persistence, ACID guarantees
class PostgreSQLProvider(MemoryProvider):
"""Stores memories in PostgreSQL with pgvector for semantic search"""
def __init__(self, connection_string: str, embedding_model: str = "all-MiniLM-L6-v2"):
self.db = asyncpg.create_pool(connection_string)
self.embedder = SentenceTransformer(embedding_model)
self._init_schema()
def _init_schema(self):
"""Initialize database schema"""
schema = """
CREATE EXTENSION IF NOT EXISTS vector;
CREATE TABLE IF NOT EXISTS memories (
id TEXT PRIMARY KEY,
type TEXT NOT NULL,
content TEXT NOT NULL,
confidence FLOAT NOT NULL,
context JSONB NOT NULL,
evidence TEXT[] NOT NULL,
metadata JSONB NOT NULL,
tags TEXT[] NOT NULL,
embedding vector(384),
created_at TIMESTAMP NOT NULL,
updated_at TIMESTAMP NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_memories_confidence ON memories(confidence);
CREATE INDEX IF NOT EXISTS idx_memories_type ON memories(type);
CREATE INDEX IF NOT EXISTS idx_memories_tags ON memories USING GIN(tags);
CREATE INDEX IF NOT EXISTS idx_memories_context ON memories USING GIN(context);
CREATE INDEX IF NOT EXISTS idx_memories_embedding ON memories USING ivfflat(embedding vector_cosine_ops);
"""
# Execute schema creation
async def create_memory(self, memory: Memory) -> Memory:
# Generate embedding
embedding = self.embedder.encode(memory.content)
await self.db.execute("""
INSERT INTO memories (
id, type, content, confidence, context,
evidence, metadata, tags, embedding, created_at, updated_at
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
""",
memory.id, memory.type, memory.content, memory.confidence,
json.dumps(memory.context), memory.evidence,
json.dumps(memory.metadata), memory.tags, embedding,
memory.created_at, memory.updated_at
)
return memory
async def query_memories(self, query: str, **filters) -> List[Memory]:
# Generate query embedding
query_embedding = self.embedder.encode(query)
# Build SQL query with filters
sql = """
SELECT * FROM memories
WHERE 1=1
"""
params = []
param_idx = 1
if 'min_confidence' in filters:
sql += f" AND confidence >= ${param_idx}"
params.append(filters['min_confidence'])
param_idx += 1
if 'memory_type' in filters:
sql += f" AND type = ${param_idx}"
params.append(filters['memory_type'])
param_idx += 1
# Semantic search using cosine similarity
sql += f"""
ORDER BY embedding <=> ${param_idx}
LIMIT ${param_idx + 1}
"""
params.extend([query_embedding, filters.get('limit', 10)])
rows = await self.db.fetch(sql, *params)
return [self._row_to_memory(row) for row in rows]Use Case: Anthropic Claude with native memory features Storage: Provider-managed Persistence: Handled by provider
class AnthropicMemoryProvider(MemoryProvider):
"""Uses Anthropic's native memory features"""
def __init__(self, api_key: str):
self.client = anthropic.Anthropic(api_key=api_key)
async def create_memory(self, memory: Memory) -> Memory:
# Use Anthropic's memory API
response = await self.client.memories.create(
content=memory.content,
type=memory.type,
confidence=memory.confidence,
context=memory.context,
metadata=memory.metadata
)
memory.id = response.id
return memory
async def query_memories(self, query: str, **filters) -> List[Memory]:
# Use Anthropic's semantic search
response = await self.client.memories.query(
query=query,
min_confidence=filters.get('min_confidence', 0.0),
type=filters.get('memory_type'),
limit=filters.get('limit', 10)
)
return [self._api_response_to_memory(item) for item in response.memories]Use Case: OpenAI Assistants API Storage: Provider-managed thread storage Persistence: Per-thread
class OpenAIMemoryProvider(MemoryProvider):
"""Stores memories in OpenAI Assistant thread storage"""
def __init__(self, api_key: str, assistant_id: str):
self.client = openai.OpenAI(api_key=api_key)
self.assistant_id = assistant_id
self.thread = None
async def create_memory(self, memory: Memory) -> Memory:
# Store as thread annotation or metadata
serialized = json.dumps(asdict(memory), default=str)
await self.client.beta.threads.messages.create(
thread_id=self.thread.id,
role="assistant",
content=f"[MEMORY] {serialized}",
metadata={"type": "memory", "memory_id": memory.id}
)
return memory
async def query_memories(self, query: str, **filters) -> List[Memory]:
# Retrieve thread messages
messages = await self.client.beta.threads.messages.list(
thread_id=self.thread.id
)
# Filter for memory messages
memory_messages = [
msg for msg in messages.data
if msg.metadata.get("type") == "memory"
]
# Deserialize and rank
memories = [
self._deserialize_memory(msg.content[0].text.value)
for msg in memory_messages
]
return self._rank_by_relevance(memories, query, **filters)Mechanism: Load all memories at session start
class DirectInheritance:
def __init__(self, memory_provider: MemoryProvider):
self.provider = memory_provider
async def initialize_session(self, context: Dict) -> List[Memory]:
"""Load all relevant memories for new session"""
# Load high-confidence memories first
established_memories = await self.provider.query_memories(
query="",
min_confidence=0.6,
limit=50
)
# Load context-specific memories
if context:
contextual_memories = await self.provider.query_memories(
query="",
context_filter=context,
min_confidence=0.3,
limit=20
)
# Merge without duplicates
all_memories = self._merge_unique(
established_memories,
contextual_memories
)
else:
all_memories = established_memories
return all_memoriesMechanism: Load memories on-demand as needed
class LazyInheritance:
def __init__(self, memory_provider: MemoryProvider):
self.provider = memory_provider
self.cache = {}
async def get_relevant_memories(self, task: str, context: Dict) -> List[Memory]:
"""Load memories relevant to current task"""
cache_key = self._compute_cache_key(task, context)
if cache_key in self.cache:
return self.cache[cache_key]
# Query for task-relevant memories
memories = await self.provider.query_memories(
query=task,
context_filter=context,
min_confidence=0.3,
limit=10
)
self.cache[cache_key] = memories
return memoriesMechanism: Compile verified procedures into executable code
class ProceduralInheritance:
def __init__(self, memory_provider: MemoryProvider):
self.provider = memory_provider
self.procedures = {}
async def compile_procedures(self) -> Dict[str, Callable]:
"""Convert high-confidence procedural memories into functions"""
# Get all procedural memories
procedural_memories = await self.provider.query_memories(
query="",
memory_type="AC",
min_confidence=0.7,
limit=100
)
# Filter for procedure-like memories
procedure_memories = [
m for m in procedural_memories
if self._is_procedure(m)
]
# Compile to executable functions
for memory in procedure_memories:
procedure_func = self._compile_to_function(memory)
self.procedures[memory.id] = procedure_func
return self.procedures
def _compile_to_function(self, memory: Memory) -> Callable:
"""Convert memory into executable function"""
# Parse procedure steps from content
steps = self._parse_procedure_steps(memory.content)
# Create function
def procedure(*args, **kwargs):
context = memory.context
for step in steps:
step.execute(context=context, args=args, kwargs=kwargs)
procedure.__doc__ = memory.content
procedure.confidence = memory.confidence
return procedureMechanism: Load different memory sets based on detected context
class ContextAwareInheritance:
def __init__(self, memory_provider: MemoryProvider):
self.provider = memory_provider
async def initialize_for_context(self, detected_context: Dict) -> List[Memory]:
"""Load memories optimized for detected context"""
# Detect domain
domain = self._detect_domain(detected_context)
# Load domain-specific procedural knowledge
procedures = await self.provider.query_memories(
query="",
memory_type="AC",
context_filter={"domain": domain},
min_confidence=0.6,
limit=30
)
# Load user preferences for this context
preferences = await self.provider.query_memories(
query="",
memory_type="UC",
context_filter=detected_context,
min_confidence=0.8,
limit=10
)
# Load general high-confidence knowledge
general = await self.provider.query_memories(
query="",
min_confidence=0.8,
limit=10
)
return procedures + preferences + general
def _detect_domain(self, context: Dict) -> str:
"""Detect domain from context cues"""
# Implementation using context analysis
passIndividual agent learning is powerful. Collective agent learning is exponentially more powerful.
- One agent learns API authentication → helpful for that agent
- Ten agents share validated authentication patterns → creates robust, cross-validated knowledge
- Agents specialize in domains, share learnings → collective intelligence emerges
When agent B imports memory from agent A:
| Scenario | Confidence Multiplier | Rationale |
|---|---|---|
| Same agent, same session | 1.0x | Full trust, direct continuity |
| Same agent, next session | 0.95x | Minimal staleness decay |
| Different agent, high reputation | 0.7x | Cross-agent requires validation |
| Different agent, medium reputation | 0.5x | Significant uncertainty |
| Different agent, validated by others | 0.8x | Consensus increases trust |
| Different agent, contradicted | 0.3x | Strong evidence of error |
| Reality-tested after import | +0.2x | Empirical validation boost |
class KnowledgeSharingProtocol:
def __init__(self, reputation_system):
self.reputation = reputation_system
def calculate_import_confidence(
self,
source_memory: Memory,
source_agent_id: str,
validation_count: int,
contradiction_count: int
) -> float:
"""Calculate confidence for imported memory"""
# Get source agent reputation
reputation = self.reputation.get_reputation(source_agent_id)
# Base transfer (cross-agent penalty)
base_confidence = source_memory.confidence * 0.6
# Reputation adjustment
reputation_multiplier = 0.5 + 0.5 * reputation
reputation_adjusted = base_confidence * reputation_multiplier
# Validation history adjustment
if validation_count + contradiction_count > 0:
validation_ratio = validation_count / (validation_count + contradiction_count)
validation_boost = validation_ratio * 0.3
else:
validation_boost = 0.0
# Final confidence (bounded at 0.9 for imported memories)
final_confidence = min(0.9, reputation_adjusted + validation_boost)
return final_confidenceAdapt Byzantine consensus algorithms to knowledge validation:
class BeliefConsensus:
def __init__(self, agents: List[str], fault_tolerance: int = 1):
self.agents = agents
self.fault_tolerance = fault_tolerance
self.min_consensus = len(agents) - fault_tolerance
def validate_belief(self, belief: str, context: Dict) -> Dict:
"""
Get consensus on a belief from multiple agents
Returns:
- consensus: bool (whether belief is validated)
- confidence: float (strength of consensus)
- validators: list (agents that validated)
- contradictors: list (agents that contradicted)
"""
validators = []
contradictors = []
abstentions = []
for agent_id in self.agents:
result = self._ask_agent(agent_id, belief, context)
if result == "VALIDATE":
validators.append(agent_id)
elif result == "CONTRADICT":
contradictors.append(agent_id)
else:
abstentions.append(agent_id)
# Calculate consensus
total_responses = len(validators) + len(contradictors)
if total_responses == 0:
consensus = False
confidence = 0.0
else:
support_ratio = len(validators) / total_responses
consensus = len(validators) >= self.min_consensus
confidence = support_ratio
return {
"consensus": consensus,
"confidence": confidence,
"validators": validators,
"contradictors": contradictors,
"abstentions": abstentions
}
def _ask_agent(self, agent_id: str, belief: str, context: Dict) -> str:
"""Ask an agent to validate or contradict a belief"""
# Implementation: Query agent's memory and empirical tests
passclass QuorumValidator:
"""Require minimum agreement before accepting high-stakes knowledge"""
def __init__(self, quorum_size: int = 3):
self.quorum_size = quorum_size
def require_quorum(self, memory: Memory) -> bool:
"""Determine if memory requires quorum validation"""
# High-confidence claims need validation
if memory.confidence > 0.8:
return True
# Critical domains need validation
if memory.context.get("domain") in ["security", "safety", "compliance"]:
return True
# High-stakes procedures need validation
if "irreversible" in memory.tags or "destructive" in memory.tags:
return True
return False
async def validate_with_quorum(
self,
memory: Memory,
agents: List[str]
) -> bool:
"""Validate memory with quorum of agents"""
validators = await self._get_validators(memory, agents)
if len(validators) < self.quorum_size:
return False
# Test with each validator
confirmations = 0
for validator_id in validators:
if await self._agent_confirms(validator_id, memory):
confirmations += 1
return confirmations >= self.quorum_sizeTrack agent reliability over time:
class AgentReputationSystem:
def __init__(self):
self.reputations = {} # agent_id -> reputation score
def update_reputation(
self,
agent_id: str,
prediction: bool,
actual_outcome: bool,
confidence: float
):
"""Update agent reputation based on prediction accuracy"""
if agent_id not in self.reputations:
self.reputations[agent_id] = {
"score": 0.5, # Start neutral
"predictions": 0,
"correct": 0,
"weighted_accuracy": 0.5
}
rep = self.reputations[agent_id]
rep["predictions"] += 1
# Record if correct
if prediction == actual_outcome:
rep["correct"] += 1
# Calculate accuracy
accuracy = rep["correct"] / rep["predictions"]
# Weight by confidence (more confident predictions matter more)
if prediction == actual_outcome:
# Correct prediction
adjustment = confidence * 0.1
else:
# Incorrect prediction (penalty larger if confident)
adjustment = -confidence * 0.15
# Update weighted accuracy
rep["weighted_accuracy"] += adjustment
rep["weighted_accuracy"] = max(0.0, min(1.0, rep["weighted_accuracy"]))
# Score is weighted accuracy
rep["score"] = rep["weighted_accuracy"]
def get_reputation(self, agent_id: str) -> float:
"""Get agent's reputation score (0.0 to 1.0)"""
if agent_id not in self.reputations:
return 0.5 # Default neutral for unknown agents
return self.reputations[agent_id]["score"]class TrustEvolution:
"""Manage trust evolution over time"""
def __init__(self):
self.trust_scores = {} # (agent_a, agent_b) -> trust score
def update_trust(
self,
trustor: str,
trustee: str,
interaction_outcome: str
):
"""Update trust based on interaction"""
key = (trustor, trustee)
if key not in self.trust_scores:
self.trust_scores[key] = 0.5 # Start neutral
current_trust = self.trust_scores[key]
if interaction_outcome == "positive":
# Trust increases slowly
new_trust = min(1.0, current_trust + 0.05)
elif interaction_outcome == "negative":
# Trust decreases quickly
new_trust = max(0.0, current_trust - 0.15)
else:
# Neutral interaction, slight decay
new_trust = current_trust * 0.99
self.trust_scores[key] = new_trust
def get_trust(self, trustor: str, trustee: str) -> float:
"""Get trust score from trustor to trustee"""
key = (trustor, trustee)
return self.trust_scores.get(key, 0.5)Agents can specialize in domains while sharing validated knowledge:
class AgentSpecialization:
def __init__(self, agent_id: str, primary_domain: str):
self.agent_id = agent_id
self.primary_domain = primary_domain
self.domain_expertise = {primary_domain: 1.0}
def update_expertise(self, domain: str, success: bool):
"""Update domain expertise based on outcomes"""
if domain not in self.domain_expertise:
self.domain_expertise[domain] = 0.3 # Start low in new domain
if success:
# Expertise increases
self.domain_expertise[domain] = min(
1.0,
self.domain_expertise[domain] + 0.05
)
else:
# Expertise decreases
self.domain_expertise[domain] = max(
0.1,
self.domain_expertise[domain] - 0.1
)
def is_expert_in(self, domain: str, threshold: float = 0.7) -> bool:
"""Check if agent is expert in domain"""
return self.domain_expertise.get(domain, 0.0) >= threshold
def get_expertise_weight(self, domain: str) -> float:
"""Get expertise level in domain"""
return self.domain_expertise.get(domain, 0.3)class KnowledgeTransferCoordinator:
def __init__(self, agents: Dict[str, AgentSpecialization]):
self.agents = agents
def transfer_with_expertise_weighting(
self,
memory: Memory,
source_agent: str,
target_agent: str
) -> float:
"""Calculate confidence transfer with domain expertise"""
domain = memory.context.get("domain", "general")
# Get source expertise in domain
source_expertise = self.agents[source_agent].get_expertise_weight(domain)
# Base confidence transfer
base_confidence = memory.confidence * 0.6
# Expertise boost
expertise_boost = source_expertise * 0.3
# Final transfer confidence
transfer_confidence = min(0.9, base_confidence + expertise_boost)
return transfer_confidenceWhen multiple agents share learnings:
- Pattern Amplification: Common successful patterns get reinforced across agents
- Error Suppression: Individual agent errors get filtered out by consensus
- Domain Specialization: Agents naturally specialize based on success rates
- Rapid Adaptation: New learnings propagate quickly across the collective
- Robust Knowledge: Cross-validated knowledge is more reliable than single-agent
class DistributedAPILearning:
"""
Multiple agents learning about an API collectively
Example: 10 agents all interact with the same API
- Agent 1 discovers auth pattern
- Agent 2 discovers rate limit handling
- Agent 3 discovers error recovery
- All share learnings → entire collective becomes expert
"""
def __init__(self, agents: List[str], shared_memory: MemoryProvider):
self.agents = agents
self.shared_memory = shared_memory
async def agent_learns(self, agent_id: str, learning: Memory):
"""Agent shares new learning with collective"""
# Publish to shared memory
await self.shared_memory.create_memory(learning)
# Other agents import with reduced confidence
for other_agent in self.agents:
if other_agent != agent_id:
await self._notify_agent(other_agent, learning)
async def collective_validates(self, memory: Memory):
"""Multiple agents test and validate a shared memory"""
validation_results = []
for agent_id in self.agents:
result = await self._agent_test(agent_id, memory)
validation_results.append(result)
# Calculate consensus
success_rate = sum(validation_results) / len(validation_results)
# Update memory confidence based on consensus
memory.confidence = success_rate
memory.metadata['validated_by_count'] = len(self.agents)
await self.shared_memory.update_memory(memory.id, memory)class ConfidenceEvolution:
"""Manages confidence evolution based on outcomes"""
# Adjustment parameters
SAME_CONTEXT_SUCCESS = 0.05
NEW_CONTEXT_SUCCESS = 0.10
SAME_CONTEXT_FAILURE = -0.20
NEW_CONTEXT_FAILURE = -0.05
STALENESS_DECAY_PER_DAY = 0.001
def update_on_success(
self,
memory: Memory,
current_context: Dict,
days_since_last_use: int
) -> float:
"""Update confidence after successful use"""
# Calculate context similarity
context_match = self._calculate_context_similarity(
memory.context,
current_context
)
# Determine adjustment
if context_match > 0.9:
# Very similar context
adjustment = self.SAME_CONTEXT_SUCCESS
elif context_match > 0.7:
# Somewhat similar context
adjustment = self.SAME_CONTEXT_SUCCESS * 0.7
else:
# Different context - proves generalization
adjustment = self.NEW_CONTEXT_SUCCESS
# Apply adjustment (bounded at 1.0)
new_confidence = min(1.0, memory.confidence + adjustment)
# Apply staleness recovery (partial)
staleness_penalty = days_since_last_use * self.STALENESS_DECAY_PER_DAY
new_confidence = min(1.0, new_confidence + staleness_penalty * 0.5)
return new_confidence
def update_on_failure(
self,
memory: Memory,
current_context: Dict
) -> float:
"""Update confidence after failure"""
# Calculate context similarity
context_match = self._calculate_context_similarity(
memory.context,
current_context
)
# Determine adjustment
if context_match > 0.9:
# Very similar context - significant failure
adjustment = self.SAME_CONTEXT_FAILURE
elif context_match > 0.7:
# Somewhat similar context
adjustment = self.SAME_CONTEXT_FAILURE * 0.5
else:
# Different context - doesn't invalidate core knowledge
adjustment = self.NEW_CONTEXT_FAILURE
# Apply adjustment (bounded at 0.0)
new_confidence = max(0.0, memory.confidence + adjustment)
return new_confidence
def apply_staleness_decay(
self,
memory: Memory,
days_since_last_use: int
) -> float:
"""Apply time-based confidence decay"""
decay = days_since_last_use * self.STALENESS_DECAY_PER_DAY
new_confidence = max(0.2, memory.confidence - decay)
return new_confidence
def _calculate_context_similarity(
self,
context1: Dict,
context2: Dict
) -> float:
"""Calculate similarity between two contexts"""
# Critical variables must match exactly
critical_vars = ['domain', 'environment', 'authentication']
for var in critical_vars:
if var in context1 and var in context2:
if context1[var] != context2[var]:
return 0.3 # Low similarity if critical vars differ
# Calculate overlap in other variables
all_keys = set(context1.keys()) | set(context2.keys())
matching = sum(
1 for key in all_keys
if context1.get(key) == context2.get(key)
)
similarity = matching / len(all_keys) if all_keys else 1.0
return similarityclass ConfidenceBasedSelector:
"""Select memories based on adjusted confidence"""
def select_best_approach(
self,
memories: List[Memory],
current_context: Dict
) -> Optional[Memory]:
"""Select highest-confidence memory for current context"""
if not memories:
return None
# Calculate context-adjusted confidence for each
scored = []
for memory in memories:
context_match = self._context_similarity(
memory.context,
current_context
)
adjusted_confidence = memory.confidence * context_match
scored.append((adjusted_confidence, memory))
# Sort by adjusted confidence
scored.sort(key=lambda x: x[0], reverse=True)
# Return best if confidence above threshold
if scored[0][0] > 0.5:
return scored[0][1]
return None # No sufficiently confident optionclass ContextMatcher:
"""Assess how well memory context matches current situation"""
# Variable importance weights
CRITICAL_WEIGHT = 1.0
IMPORTANT_WEIGHT = 0.6
MINOR_WEIGHT = 0.3
NEGLIGIBLE_WEIGHT = 0.0
def assess_transfer(
self,
source_context: Dict,
target_context: Dict,
variable_importance: Dict[str, str]
) -> Dict:
"""Assess knowledge transferability"""
differences = self._identify_differences(
source_context,
target_context
)
# Assess impact of each difference
critical_diffs = []
important_diffs = []
minor_diffs = []
for key, (source_val, target_val) in differences.items():
importance = variable_importance.get(key, 'minor')
diff_info = {
'variable': key,
'source_value': source_val,
'target_value': target_val,
'importance': importance
}
if importance == 'critical':
critical_diffs.append(diff_info)
elif importance == 'important':
important_diffs.append(diff_info)
else:
minor_diffs.append(diff_info)
# Calculate confidence adjustment
if critical_diffs:
confidence_multiplier = 0.3 # Major uncertainty
elif len(important_diffs) >= 2:
confidence_multiplier = 0.5 # Moderate uncertainty
elif important_diffs:
confidence_multiplier = 0.7 # Some uncertainty
elif minor_diffs:
confidence_multiplier = 0.9 # Minimal uncertainty
else:
confidence_multiplier = 1.0 # Full confidence
return {
'transferable': len(critical_diffs) == 0,
'confidence_multiplier': confidence_multiplier,
'critical_differences': critical_diffs,
'important_differences': important_diffs,
'minor_differences': minor_diffs,
'recommendation': self._generate_recommendation(
critical_diffs,
important_diffs,
minor_diffs
)
}
def _generate_recommendation(
self,
critical: List,
important: List,
minor: List
) -> str:
"""Generate human-readable recommendation"""
if critical:
return f"Cannot transfer: critical variables differ ({[d['variable'] for d in critical]})"
elif len(important) >= 2:
return f"Transfer with caution: multiple important variables differ"
elif important:
return f"Transfer with adaptation: {important[0]['variable']} differs"
elif minor:
return f"Transfer likely successful with minor adjustments"
else:
return "Transfer with full confidence: identical context"class VariableImportanceLearner:
"""Learn which context variables matter through experience"""
def __init__(self):
self.variable_impacts = {} # Track impact of each variable
def record_outcome(
self,
source_context: Dict,
target_context: Dict,
predicted_success: bool,
actual_success: bool
):
"""Record outcome to learn variable importance"""
differences = self._identify_differences(
source_context,
target_context
)
# If prediction was wrong, differences matter more
prediction_correct = (predicted_success == actual_success)
for var, (source_val, target_val) in differences.items():
if var not in self.variable_impacts:
self.variable_impacts[var] = {
'correct_predictions': 0,
'incorrect_predictions': 0,
'total_observations': 0
}
self.variable_impacts[var]['total_observations'] += 1
if prediction_correct:
self.variable_impacts[var]['correct_predictions'] += 1
else:
self.variable_impacts[var]['incorrect_predictions'] += 1
def get_variable_importance(self, variable: str) -> str:
"""Determine importance level based on observed impact"""
if variable not in self.variable_impacts:
return 'minor' # Default for unknown variables
stats = self.variable_impacts[variable]
if stats['total_observations'] < 3:
return 'minor' # Not enough data
# Calculate error rate when this variable differs
error_rate = (
stats['incorrect_predictions'] /
stats['total_observations']
)
# Classify based on error rate
if error_rate > 0.7:
return 'critical' # High error rate = critical variable
elif error_rate > 0.4:
return 'important'
elif error_rate > 0.2:
return 'minor'
else:
return 'negligible'class ConflictDetector:
"""Detect contradicting memories"""
def find_conflicts(
self,
memories: List[Memory]
) -> List[Tuple[Memory, Memory]]:
"""Find pairs of contradicting memories"""
conflicts = []
for i, mem1 in enumerate(memories):
for mem2 in memories[i+1:]:
if self._are_conflicting(mem1, mem2):
conflicts.append((mem1, mem2))
return conflicts
def _are_conflicting(self, mem1: Memory, mem2: Memory) -> bool:
"""Determine if two memories contradict"""
# Must be about same topic
if not self._same_topic(mem1, mem2):
return False
# Must be in similar context
context_similarity = self._context_similarity(
mem1.context,
mem2.context
)
if context_similarity < 0.7:
return False # Different contexts, not really conflicting
# Must have contradicting claims
if self._claims_contradict(mem1.content, mem2.content):
return True
return Falseclass ConflictResolver:
"""Resolve contradicting memories through testing"""
def __init__(self, memory_provider: MemoryProvider):
self.provider = memory_provider
async def resolve(
self,
memory1: Memory,
memory2: Memory,
test_function: Callable
) -> Memory:
"""Resolve conflict through empirical testing"""
# Mark both as contested
await self._mark_contested(memory1, memory2)
# Design minimal test
test = self._design_test(memory1, memory2)
# Execute test
result = await test_function(test)
# Determine winner
if result.supports(memory1):
winner, loser = memory1, memory2
elif result.supports(memory2):
winner, loser = memory2, memory1
else:
# Neither confirmed - both might be contextual
return await self._resolve_contextually(
memory1,
memory2,
result
)
# Update memories
winner.confidence = min(0.8, winner.confidence + 0.2)
winner.metadata['verified_against'] = loser.id
winner.evidence.append(f"Confirmed via test on {datetime.now()}")
loser.confidence = max(0.1, loser.confidence - 0.3)
loser.metadata['superseded_by'] = winner.id
loser.metadata['deprecated'] = True
loser.evidence.append(f"Superseded by {winner.id} on {datetime.now()}")
# Save updates
await self.provider.update_memory(winner.id, asdict(winner))
await self.provider.update_memory(loser.id, asdict(loser))
return winner
def _design_test(self, memory1: Memory, memory2: Memory) -> Dict:
"""Design minimal test to arbitrate between memories"""
# Extract claims
claim1 = self._extract_testable_claim(memory1)
claim2 = self._extract_testable_claim(memory2)
# Design test that differentiates
return {
'claim1': claim1,
'claim2': claim2,
'procedure': self._create_test_procedure(claim1, claim2)
}class UncertaintySelector:
"""Select best action under uncertainty"""
def select_with_uncertainty(
self,
options: List[Tuple[Memory, float]], # (memory, expected_value)
risk_tolerance: str = "medium"
) -> Memory:
"""Select best option considering uncertainty"""
# Compute expected value for each option
scored = []
for memory, expected_value in options:
# Adjust by confidence (uncertainty discount)
adjusted_value = expected_value * memory.confidence
# Apply risk adjustment
if risk_tolerance == "low":
# Prefer safer options (higher confidence)
adjusted_value *= (0.5 + 0.5 * memory.confidence)
elif risk_tolerance == "high":
# Accept more risk for higher potential
adjusted_value *= (1.5 - 0.5 * memory.confidence)
scored.append((adjusted_value, memory))
# Select highest expected value
scored.sort(key=lambda x: x[0], reverse=True)
return scored[0][1]Goal: Get basic learning loop working
Tasks:
- Implement Memory data structure
- Create simple file-based storage
- Implement basic remember → act → observe → update loop
- Test with simple tasks
Success Criteria:
- Can create memories manually
- Can retrieve memories by keyword
- Confidence updates after success/failure
- Memories persist across sessions
Goal: Implement confidence evolution
Tasks:
- Implement confidence update algorithms
- Add staleness decay
- Create context similarity calculation
- Build confidence-based selection
Success Criteria:
- Confidence increases with repeated success
- Confidence decreases with failures
- Old memories get lower confidence
- System prefers high-confidence approaches
Goal: Make system context-aware
Tasks:
- Define context schema
- Implement context matching
- Add context-adjusted confidence
- Build variable importance tracking
Success Criteria:
- Memories tagged with context
- Transfer assessment works
- Context differences affect confidence
- System learns which variables matter
Goal: Support multiple backends
Tasks:
- Define MemoryProvider interface
- Implement 2-3 adapters
- Create adapter factory
- Test with different providers
Success Criteria:
- Can switch providers without code changes
- Works with conversation history
- Works with file system
- Works with database
Goal: Add sophisticated capabilities
Tasks:
- Implement conflict detection/resolution
- Add meta-learning (pattern extraction)
- Build procedural memory compilation
- Create evaluation metrics
Success Criteria:
- Conflicts get resolved automatically
- System extracts patterns from experience
- Procedures become reusable functions
- Can measure system improvement
import asyncio
from datetime import datetime
from typing import List, Dict, Optional
import json
# === Core Data Structures ===
class Memory:
def __init__(
self,
id: str,
type: str,
content: str,
confidence: float,
context: Dict,
evidence: List[str],
tags: List[str]
):
self.id = id
self.type = type
self.content = content
self.confidence = confidence
self.context = context
self.evidence = evidence
self.tags = tags
self.created_at = datetime.now()
self.updated_at = datetime.now()
self.success_count = 0
self.failure_count = 0
# === Simple File-Based Storage ===
class SimpleMemoryStore:
def __init__(self, file_path: str):
self.file_path = file_path
self.memories = self._load()
def _load(self) -> List[Memory]:
try:
with open(self.file_path, 'r') as f:
data = json.load(f)
return [self._dict_to_memory(m) for m in data]
except FileNotFoundError:
return []
def _save(self):
with open(self.file_path, 'w') as f:
data = [self._memory_to_dict(m) for m in self.memories]
json.dump(data, f, indent=2, default=str)
def create(self, memory: Memory):
self.memories.append(memory)
self._save()
def query(self, keywords: List[str], min_confidence: float = 0.0) -> List[Memory]:
results = []
for memory in self.memories:
# Simple keyword matching
if any(kw.lower() in memory.content.lower() for kw in keywords):
if memory.confidence >= min_confidence:
results.append(memory)
return sorted(results, key=lambda m: m.confidence, reverse=True)
def update(self, memory: Memory):
for i, m in enumerate(self.memories):
if m.id == memory.id:
self.memories[i] = memory
self._save()
return
def _memory_to_dict(self, m: Memory) -> Dict:
return {
'id': m.id,
'type': m.type,
'content': m.content,
'confidence': m.confidence,
'context': m.context,
'evidence': m.evidence,
'tags': m.tags,
'created_at': m.created_at.isoformat(),
'updated_at': m.updated_at.isoformat(),
'success_count': m.success_count,
'failure_count': m.failure_count
}
def _dict_to_memory(self, d: Dict) -> Memory:
m = Memory(
id=d['id'],
type=d['type'],
content=d['content'],
confidence=d['confidence'],
context=d['context'],
evidence=d['evidence'],
tags=d['tags']
)
m.created_at = datetime.fromisoformat(d['created_at'])
m.updated_at = datetime.fromisoformat(d['updated_at'])
m.success_count = d.get('success_count', 0)
m.failure_count = d.get('failure_count', 0)
return m
# === Learning Agent ===
class LearningAgent:
def __init__(self, memory_store: SimpleMemoryStore):
self.memory_store = memory_store
self.current_context = {}
async def perform_task(self, task: str, context: Dict):
"""Execute full learning loop for a task"""
print(f"\n=== Task: {task} ===")
# REMEMBER
keywords = task.split()
relevant_memories = self.memory_store.query(keywords, min_confidence=0.3)
print(f"Found {len(relevant_memories)} relevant memories")
# ASSESS
best_memory = relevant_memories[0] if relevant_memories else None
if best_memory:
print(f"Best approach: {best_memory.content}")
print(f"Confidence: {best_memory.confidence:.2f}")
# HYPOTHESIZE
if best_memory:
hypothesis = f"Will use: {best_memory.content}"
expected_outcome = "success"
confidence = best_memory.confidence
else:
hypothesis = "No prior knowledge, will try default approach"
expected_outcome = "unknown"
confidence = 0.2
print(f"Hypothesis: {hypothesis}")
print(f"Expected: {expected_outcome} (confidence: {confidence:.2f})")
# ACT (simulated)
actual_outcome = await self._simulate_action(task, best_memory)
# OBSERVE
print(f"Actual outcome: {actual_outcome}")
success = (actual_outcome == "success")
# UPDATE
if best_memory:
await self._update_memory(best_memory, success, context)
else:
# Create new memory from experience
await self._create_memory_from_experience(
task, actual_outcome, success, context
)
# EXTRACT (simplified)
if len(relevant_memories) >= 3:
print("Extracting patterns from multiple experiences...")
# Pattern extraction logic would go here
return success
async def _simulate_action(self, task: str, memory: Optional[Memory]) -> str:
"""Simulate task execution"""
# In real implementation, this would actually do something
# For demo, we'll use simple rules
import random
if memory and memory.confidence > 0.6:
# High confidence approach usually works
return "success" if random.random() > 0.1 else "failure"
elif memory:
# Medium confidence is less reliable
return "success" if random.random() > 0.3 else "failure"
else:
# No knowledge is unreliable
return "success" if random.random() > 0.5 else "failure"
async def _update_memory(self, memory: Memory, success: bool, context: Dict):
"""Update memory based on outcome"""
if success:
memory.confidence = min(1.0, memory.confidence + 0.05)
memory.success_count += 1
memory.evidence.append(f"Success on {datetime.now().date()}")
print(f"✓ Updated confidence: {memory.confidence:.2f}")
else:
memory.confidence = max(0.1, memory.confidence - 0.15)
memory.failure_count += 1
memory.evidence.append(f"Failure on {datetime.now().date()}")
print(f"✗ Decreased confidence: {memory.confidence:.2f}")
memory.updated_at = datetime.now()
self.memory_store.update(memory)
async def _create_memory_from_experience(
self,
task: str,
outcome: str,
success: bool,
context: Dict
):
"""Create new memory from experience"""
import uuid
memory = Memory(
id=str(uuid.uuid4()),
type="AC",
content=f"Task '{task}' → {outcome}",
confidence=0.4 if success else 0.2,
context=context,
evidence=[f"Observed on {datetime.now().date()}"],
tags=task.split()
)
if success:
memory.success_count = 1
else:
memory.failure_count = 1
self.memory_store.create(memory)
print(f"Created new memory: {memory.content}")
# === Usage Example ===
async def main():
# Initialize storage
store = SimpleMemoryStore("agent_memories.json")
# Create agent
agent = LearningAgent(store)
# Define context
context = {
"domain": "api_integration",
"environment": "development"
}
# Execute several tasks to see learning
tasks = [
"authenticate to API",
"authenticate to API", # Repeat to see confidence increase
"fetch user data",
"authenticate to API", # Third time should be very confident
]
for task in tasks:
await agent.perform_task(task, context)
await asyncio.sleep(1) # Small delay for readability
# Show final memory state
print("\n=== Final Memory State ===")
for memory in store.memories:
print(f"\nMemory: {memory.content}")
print(f" Confidence: {memory.confidence:.2f}")
print(f" Successes: {memory.success_count}, Failures: {memory.failure_count}")
if __name__ == "__main__":
asyncio.run(main())How quickly the system improves on repeated tasks.
def calculate_learning_rate(memory: Memory) -> float:
"""Measure rate of confidence improvement"""
if memory.success_count < 2:
return 0.0
time_span = (memory.updated_at - memory.created_at).days
if time_span == 0:
return 0.0
confidence_gain = memory.confidence - 0.2 # Assuming started at 0.2
learning_rate = confidence_gain / time_span
return learning_rateHow well knowledge persists and remains accurate.
def measure_retention(memories: List[Memory]) -> Dict:
"""Measure knowledge retention over time"""
active_memories = [m for m in memories if m.confidence > 0.5]
stale_memories = [
m for m in memories
if (datetime.now() - m.updated_at).days > 30
]
return {
'active_count': len(active_memories),
'stale_count': len(stale_memories),
'retention_rate': len(active_memories) / len(memories) if memories else 0
}How often the system's predictions match reality.
class PredictionTracker:
def __init__(self):
self.predictions = []
def record(self, predicted: bool, actual: bool, confidence: float):
self.predictions.append({
'predicted': predicted,
'actual': actual,
'confidence': confidence,
'correct': predicted == actual
})
def accuracy(self) -> float:
if not self.predictions:
return 0.0
correct = sum(1 for p in self.predictions if p['correct'])
return correct / len(self.predictions)
def calibration(self) -> float:
"""How well confidence matches actual accuracy"""
if not self.predictions:
return 0.0
# Group by confidence buckets
buckets = {}
for p in self.predictions:
bucket = round(p['confidence'], 1)
if bucket not in buckets:
buckets[bucket] = {'correct': 0, 'total': 0}
buckets[bucket]['total'] += 1
if p['correct']:
buckets[bucket]['correct'] += 1
# Calculate calibration error
calibration_error = 0.0
for conf, stats in buckets.items():
actual_accuracy = stats['correct'] / stats['total']
calibration_error += abs(conf - actual_accuracy)
return 1.0 - (calibration_error / len(buckets))How quickly the system adapts to context changes.
def measure_adaptation(memories: List[Memory]) -> Dict:
"""Measure how quickly system adapts to new contexts"""
# Group memories by context
context_groups = {}
for memory in memories:
ctx_key = json.dumps(memory.context, sort_keys=True)
if ctx_key not in context_groups:
context_groups[ctx_key] = []
context_groups[ctx_key].append(memory)
# Measure time to high confidence in each context
adaptation_times = []
for memories in context_groups.values():
if not memories:
continue
sorted_mems = sorted(memories, key=lambda m: m.created_at)
first = sorted_mems[0]
# Find first memory to reach 0.7 confidence
for mem in sorted_mems:
if mem.confidence >= 0.7:
time_to_adapt = (mem.updated_at - first.created_at).days
adaptation_times.append(time_to_adapt)
break
if not adaptation_times:
return {'average_days': None, 'count': 0}
return {
'average_days': sum(adaptation_times) / len(adaptation_times),
'count': len(adaptation_times)
}How well new knowledge builds on old knowledge.
def measure_compound_learning(memories: List[Memory]) -> float:
"""Measure if learning is accelerating (compound effect)"""
if len(memories) < 5:
return 0.0
# Sort by creation time
sorted_mems = sorted(memories, key=lambda m: m.created_at)
# Calculate average time to high confidence for early vs late memories
early = sorted_mems[:len(sorted_mems)//2]
late = sorted_mems[len(sorted_mems)//2:]
def avg_learning_time(mems):
times = []
for mem in mems:
if mem.confidence >= 0.7:
time = (mem.updated_at - mem.created_at).days
times.append(time)
return sum(times) / len(times) if times else 0
early_time = avg_learning_time(early)
late_time = avg_learning_time(late)
if early_time == 0:
return 0.0
# Positive value means learning accelerated
acceleration = (early_time - late_time) / early_time
return accelerationHow often external sources validate or contradict your beliefs.
class AdversarialValidationTracker:
"""Track external contradiction as key health metric"""
def __init__(self):
self.validations = [] # (memory_id, source, outcome)
def record_validation(
self,
memory: Memory,
validator: str, # 'reality', 'agent-X', 'human', 'documentation'
outcome: str # 'confirmed', 'contradicted', 'context-mismatch'
):
self.validations.append({
'memory_id': memory.id,
'confidence_at_test': memory.confidence,
'validator': validator,
'outcome': outcome,
'timestamp': datetime.now()
})
def get_validation_rate(self) -> Dict:
"""Calculate how often beliefs get externally validated"""
if not self.validations:
return {'rate': 0.0, 'count': 0}
confirmed = sum(1 for v in self.validations if v['outcome'] == 'confirmed')
contradicted = sum(1 for v in self.validations if v['outcome'] == 'contradicted')
context_mismatch = sum(1 for v in self.validations if v['outcome'] == 'context-mismatch')
total = len(self.validations)
return {
'confirmation_rate': confirmed / total,
'contradiction_rate': contradicted / total,
'context_mismatch_rate': context_mismatch / total,
'total_validations': total,
# KEY METRIC: How often do external sources catch errors?
'external_error_detection': contradicted / max(1, confirmed + contradicted)
}
def get_overconfidence_errors(self) -> List[Dict]:
"""Find cases where high confidence was contradicted"""
overconfident_errors = [
v for v in self.validations
if v['confidence_at_test'] > 0.7 and v['outcome'] == 'contradicted'
]
return overconfident_errorsDoes knowledge actually work in new contexts?
def measure_transfer_success(memories: List[Memory]) -> Dict:
"""Measure how well knowledge transfers to new contexts"""
transfer_attempts = []
for memory in memories:
# Find uses of this memory in different contexts
uses = memory.metadata.get('context_uses', [])
for use in uses:
original_context = memory.context
use_context = use['context']
success = use['outcome'] == 'success'
# Calculate context similarity
context_similarity = calculate_context_similarity(
original_context,
use_context
)
transfer_attempts.append({
'context_similarity': context_similarity,
'success': success,
'original_confidence': memory.confidence
})
if not transfer_attempts:
return {'transfer_rate': None, 'count': 0}
# Group by context similarity buckets
buckets = {
'identical': [t for t in transfer_attempts if t['context_similarity'] > 0.9],
'similar': [t for t in transfer_attempts if 0.7 < t['context_similarity'] <= 0.9],
'different': [t for t in transfer_attempts if t['context_similarity'] <= 0.7]
}
results = {}
for bucket_name, attempts in buckets.items():
if attempts:
success_rate = sum(1 for a in attempts if a['success']) / len(attempts)
results[f'{bucket_name}_transfer_rate'] = success_rate
results[f'{bucket_name}_count'] = len(attempts)
return resultsHigh-confidence knowledge that never gets used is dead weight.
def measure_knowledge_usefulness(memories: List[Memory]) -> Dict:
"""Measure if knowledge is actually being used"""
high_confidence = [m for m in memories if m.confidence > 0.7]
if not high_confidence:
return {'usefulness': None, 'unused_count': 0}
# Check last access time
now = datetime.now()
unused = []
stale_but_confident = []
for memory in high_confidence:
days_since_access = (now - memory.metadata.get('last_accessed', memory.created_at)).days
if days_since_access > 90:
unused.append(memory)
elif days_since_access > 30:
stale_but_confident.append(memory)
return {
'high_confidence_count': len(high_confidence),
'unused_count': len(unused),
'stale_count': len(stale_but_confident),
'usefulness_rate': 1.0 - (len(unused) / len(high_confidence)),
'avg_days_since_use': sum(
(now - m.metadata.get('last_accessed', m.created_at)).days
for m in high_confidence
) / len(high_confidence)
}Are you updating fast enough when contradicted?
class UpdateVelocityTracker:
"""Track how quickly system updates when faced with contradiction"""
def __init__(self):
self.contradictions = []
def record_contradiction(
self,
memory: Memory,
time_to_update: timedelta
):
"""Record how long it took to update after contradiction"""
self.contradictions.append({
'memory_id': memory.id,
'confidence_before': memory.confidence,
'time_to_update_hours': time_to_update.total_seconds() / 3600,
'timestamp': datetime.now()
})
def get_update_velocity(self) -> Dict:
"""Calculate update responsiveness"""
if not self.contradictions:
return {'avg_hours': None, 'count': 0}
update_times = [c['time_to_update_hours'] for c in self.contradictions]
return {
'avg_update_hours': sum(update_times) / len(update_times),
'median_update_hours': sorted(update_times)[len(update_times) // 2],
'slowest_update_hours': max(update_times),
'contradiction_count': len(self.contradictions),
# KEY METRIC: System is healthy if it updates quickly
'responsive': sum(1 for t in update_times if t < 24) / len(update_times)
}class SystemHealthDashboard:
def __init__(self, memory_store):
self.store = memory_store
self.prediction_tracker = PredictionTracker()
self.adversarial_tracker = AdversarialValidationTracker()
self.update_velocity_tracker = UpdateVelocityTracker()
def generate_report(self) -> Dict:
"""Generate comprehensive health report"""
memories = self.store.memories
return {
# Internal Metrics
'total_memories': len(memories),
'high_confidence': len([m for m in memories if m.confidence > 0.7]),
'medium_confidence': len([m for m in memories if 0.4 < m.confidence <= 0.7]),
'low_confidence': len([m for m in memories if m.confidence <= 0.4]),
'retention': measure_retention(memories),
'adaptation': measure_adaptation(memories),
'compound_rate': measure_compound_learning(memories),
'prediction_accuracy': self.prediction_tracker.accuracy(),
'calibration': self.prediction_tracker.calibration(),
'avg_learning_rate': sum(
calculate_learning_rate(m) for m in memories
) / len(memories) if memories else 0,
# External Validity Metrics (NEW)
'adversarial_validation': self.adversarial_tracker.get_validation_rate(),
'transfer_success': measure_transfer_success(memories),
'knowledge_usefulness': measure_knowledge_usefulness(memories),
'update_velocity': self.update_velocity_tracker.get_update_velocity(),
}
def print_report(self):
report = self.generate_report()
print("\n" + "="*70)
print("LAMARCKIAN LEARNING SYSTEM - HEALTH REPORT")
print("="*70)
print(f"\n📊 Memory Distribution:")
print(f" Total: {report['total_memories']}")
print(f" High Confidence (>0.7): {report['high_confidence']}")
print(f" Medium Confidence (0.4-0.7): {report['medium_confidence']}")
print(f" Low Confidence (<0.4): {report['low_confidence']}")
print(f"\n📈 Internal Learning Metrics:")
print(f" Prediction Accuracy: {report['prediction_accuracy']:.1%}")
print(f" Calibration Score: {report['calibration']:.1%}")
print(f" Compound Learning Rate: {report['compound_rate']:.1%}")
print(f" Avg Learning Rate: {report['avg_learning_rate']:.3f}")
print(f"\n🔍 External Validity Metrics (CRITICAL):")
val = report['adversarial_validation']
if val['total_validations'] > 0:
print(f" External Validations: {val['total_validations']}")
print(f" Confirmation Rate: {val['confirmation_rate']:.1%}")
print(f" Contradiction Rate: {val['contradiction_rate']:.1%}")
print(f" External Error Detection: {val['external_error_detection']:.1%}")
else:
print(f" ⚠️ WARNING: No external validations! System may be self-confirming.")
print(f"\n🎯 Transfer Success:")
transfer = report['transfer_success']
if 'identical_transfer_rate' in transfer:
print(f" Identical Context: {transfer['identical_transfer_rate']:.1%}")
print(f" Similar Context: {transfer.get('similar_transfer_rate', 0):.1%}")
print(f" Different Context: {transfer.get('different_transfer_rate', 0):.1%}")
else:
print(f" No transfer data yet")
print(f"\n💡 Knowledge Usefulness:")
useful = report['knowledge_usefulness']
if useful['usefulness'] is not None:
print(f" High-Confidence Knowledge: {useful['high_confidence_count']}")
print(f" Unused (>90 days): {useful['unused_count']}")
print(f" Usefulness Rate: {useful['usefulness_rate']:.1%}")
if useful['usefulness_rate'] < 0.7:
print(f" ⚠️ WARNING: Too much unused high-confidence knowledge")
print(f"\n⚡ Update Velocity:")
velocity = report['update_velocity']
if velocity['count'] > 0:
print(f" Contradictions: {velocity['contradiction_count']}")
print(f" Avg Update Time: {velocity['avg_update_hours']:.1f} hours")
print(f" Responsive (<24h): {velocity['responsive']:.1%}")
if velocity['responsive'] < 0.7:
print(f" ⚠️ WARNING: Slow to update when contradicted")
else:
print(f" No contradictions tracked yet")
print(f"\n💪 Knowledge Health:")
print(f" Active Memories: {report['retention']['active_count']}")
print(f" Stale Memories: {report['retention']['stale_count']}")
print(f" Retention Rate: {report['retention']['retention_rate']:.1%}")
print("\n" + "="*70)
# Health Assessment
self._print_health_assessment(report)
def _print_health_assessment(self, report):
"""Assess overall system health"""
warnings = []
strengths = []
# Check for self-confirmation bias
val = report['adversarial_validation']
if val['total_validations'] == 0:
warnings.append("❌ CRITICAL: No external validation - risk of self-confirming hallucination")
elif val['contradiction_rate'] < 0.1:
warnings.append("⚠️ Very low contradiction rate - may not be testing risky beliefs")
# Check update responsiveness
velocity = report['update_velocity']
if velocity['count'] > 0 and velocity['responsive'] < 0.7:
warnings.append("⚠️ Slow to update when contradicted - may defend wrong beliefs")
# Check knowledge usefulness
useful = report['knowledge_usefulness']
if useful['usefulness'] and useful['usefulness_rate'] < 0.6:
warnings.append("⚠️ Too much unused knowledge - accumulating dead weight")
# Check prediction accuracy
if report['prediction_accuracy'] < 0.6:
warnings.append("⚠️ Low prediction accuracy - beliefs don't match reality")
elif report['prediction_accuracy'] > 0.8:
strengths.append("✅ High prediction accuracy")
# Check calibration
if report['calibration'] > 0.8:
strengths.append("✅ Well-calibrated confidence")
elif report['calibration'] < 0.6:
warnings.append("⚠️ Poor calibration - confidence doesn't match accuracy")
print("\n🏥 Health Assessment:")
if strengths:
print("\nStrengths:")
for strength in strengths:
print(f" {strength}")
if warnings:
print("\nWarnings:")
for warning in warnings:
print(f" {warning}")
if not warnings:
print(" ✅ System is healthy!")This framework provides a complete blueprint for building Lamarckian evolutionary systems for AI agents:
- Five Universal Principles: Core principles including adversarial validation
- Provider Independence: Abstract interfaces that work with any AI provider
- Flexible Storage: Adapters for any storage substrate
- Knowledge Interchange Standard: Universal format for cross-agent learning
- Multi-Agent Coordination: Protocols for collective intelligence
- Evolutionary Learning: Knowledge that compounds over time
- Context Awareness: Understanding of when knowledge applies
- Conflict Resolution: Systematic approach to contradictions
- External Validity Metrics: Measures that prevent self-confirming hallucination
- Adversarial Validation: Built-in safeguards against drift
Systems built on this framework will:
- Get smarter through use rather than degrading
- Transfer knowledge across sessions and contexts
- Learn from mistakes and adapt
- Build confidence in what works
- Compound knowledge over time
- Start with the simple file-based reference implementation
- Run it on simple tasks to see learning in action
- Gradually add sophistication as needed
- Adapt to your specific provider and use case
- Measure progress with the metrics provided
Traditional AI systems are static artifacts.
Lamarckian AI systems are living organisms that evolve through experience.
The difference is inheritance of acquired characteristics—exactly what Lamarck proposed, and exactly what AI systems need to continuously improve.
Start simple. Start now. Let reality be your teacher.
- Adversarial Memory Validation: Deep dive on preventing drift and hallucination
- AMLP Complete Technical Implementation
- AMLP Operational Abstraction
- Memory Entity Implementation Guide
- Memory Persistence Architecture
- Principle 5: External contradiction as required safeguard
- Knowledge Interchange Standard: Universal format for cross-agent memory sharing
- Multi-Agent Coordination: Byzantine consensus, reputation systems, domain specialization
- External Validity Metrics: Adversarial validation rate, transfer success, knowledge usefulness, update velocity
- Anti-Gaming Mechanisms: Protection against self-confirming hallucination
This framework is open for any use. Implementations, improvements, and adaptations are encouraged. Share your experiences to help evolve this approach.
Special thanks to the collaborative feedback that identified critical gaps in V1.0 and led to the adversarial validation framework in V2.0.
Version: 2.0 Last Updated: 2026-02-04 License: Open source, any use permitted Philosophy: Knowledge evolves through use. Each interaction builds on the last. Reality is the ultimate teacher. External validation prevents drift.