Version: 1.0 Companion to: Lamarckian Evolution for AI Systems V2.0 Status: Architectural Specification License: Open for any use, any implementation, any provider
- The Problem: Self-Confirming Hallucination
- Why Internal Metrics Aren't Enough
- Adversarial Validation as Forcing Function
- Byzantine Fault Tolerance for Beliefs
- Trust and Reputation Models
- Cross-Agent Validation Protocols
- Poisoning Prevention
- Attack Vectors and Defenses
- Implementation Patterns
- Validation Cadence and Triggers
Lamarckian evolutionary systems are powerful because they learn from experience and inherit knowledge. But this same mechanism creates a critical vulnerability:
Without external validation, the system can drift into high-confidence hallucination.
1. Agent creates memory from experience (confidence: 0.4)
2. Agent uses memory successfully (confidence: 0.5)
3. Agent uses memory again successfully (confidence: 0.6)
4. Agent's success reinforces belief (confidence: 0.7)
5. High confidence → use more often (confidence: 0.8)
6. Increased use → more reinforcement (confidence: 0.9)
7. System is now highly confident in potentially wrong belief
Day 1: Agent tries `/api/v1/users` and gets 200 OK
→ Creates memory: "Users endpoint is /api/v1/users" (confidence: 0.4)
Day 2: Agent uses `/api/v1/users` successfully again
→ Updates memory: confidence → 0.5
Day 3-10: Agent uses endpoint repeatedly, always gets 200
→ Memory confidence → 0.9
Reality: The endpoint actually is `/api/v2/users`.
The v1 endpoint is deprecated and returns cached data.
Agent is operating on stale information with high confidence.
The agent has created a self-confirming loop without ever validating against external reality.
- Confirmation Bias: Successful outcomes reinforce beliefs
- Lack of Adversarial Testing: System doesn't seek contradictions
- Internal Consistency: High confidence + repeated success = self-validation
- No External Ground Truth: System validates memories against other memories
- Gaming Fitness Metrics: High prediction accuracy within closed system
# The agent's internal view (all consistent)
memories = [
Memory("API endpoint is /v1/users", confidence=0.9),
Memory("Always returns user data", confidence=0.8),
Memory("No errors in 100 uses", confidence=0.95)
]
# Reality
actual_endpoint = "/v2/users"
v1_returns_cached_data = True
agent_operating_on_stale_data = True
# Agent's self-assessment
internal_consistency = 1.0 # Everything agrees!
prediction_accuracy = 0.95 # Predictions match observations!
confidence_calibration = 0.90 # Well-calibrated!
# Actual correctness
external_validity = 0.0 # Completely wrong about realityThe system looks healthy by internal metrics while being completely wrong.
| Metric | What It Measures | What It Doesn't Measure |
|---|---|---|
| Prediction Accuracy | Do predictions match observations? | Are observations reflecting reality? |
| Confidence Calibration | Does confidence match accuracy? | Is the system accurate about the right things? |
| Knowledge Retention | Do memories persist? | Are memories actually correct? |
| Learning Rate | How fast confidence increases? | Is increasing confidence justified? |
| Compound Learning | Does learning accelerate? | Is learning in the right direction? |
An agent can have:
- Perfect internal consistency
- High prediction accuracy
- Well-calibrated confidence
- Accelerating learning
While being completely disconnected from external reality.
This is analogous to the Chinese Room: perfect symbol manipulation with no semantic grounding.
External validation is the semantic grounding that connects internal symbols to external reality.
- Reality Checks: Does belief match observable fact?
- Expert Contradiction: Do domain experts disagree?
- Cross-Agent Consensus: Do independent agents concur?
- Adversarial Testing: Can we find counterexamples?
- Documentation Verification: Do authoritative sources confirm?
High-confidence beliefs MUST be tested against external sources that can contradict them.
External contradiction is not optional—it's a required forcing function that prevents drift.
Strength of validation (highest to lowest):
1. Physical Reality Test
- Empirical observation of actual outcome
- Cannot be gamed or misinterpreted
- Ultimate arbiter
2. Multiple Independent Agents (Consensus)
- 3+ agents independently validate
- Byzantine fault tolerance
- Guards against individual agent error
3. Domain Expert Human
- Human with verified expertise
- Authoritative but fallible
- Higher trust than general human
4. Authoritative Documentation
- Official specs, verified sources
- Can be outdated
- Stronger than casual sources
5. Single Agent (Cross-Validation)
- Different agent from originator
- Reduces individual bias
- Weaker than consensus
6. Internal Consistency
- Memory validates memory
- Necessary but insufficient
- Lowest validation strength
class ValidationPolicy:
"""Define validation requirements based on confidence"""
def get_validation_requirements(self, memory: Memory) -> Dict:
"""Return validation requirements for memory"""
confidence = memory.confidence
if confidence >= 0.9:
# Very high confidence requires strongest validation
return {
'min_validators': 3,
'validator_types': ['reality', 'expert', 'agent'],
'consensus_threshold': 0.8,
'required_frequency_days': 30,
'adversarial_test_required': True
}
elif confidence >= 0.7:
# High confidence requires external validation
return {
'min_validators': 2,
'validator_types': ['reality', 'agent', 'documentation'],
'consensus_threshold': 0.7,
'required_frequency_days': 60,
'adversarial_test_required': True
}
elif confidence >= 0.5:
# Medium confidence should be validated
return {
'min_validators': 1,
'validator_types': ['agent', 'documentation'],
'consensus_threshold': 0.6,
'required_frequency_days': 90,
'adversarial_test_required': False
}
else:
# Low confidence doesn't require validation yet
return {
'min_validators': 0,
'validator_types': [],
'consensus_threshold': 0.0,
'required_frequency_days': None,
'adversarial_test_required': False
}Definition: Actively seeking situations that could disprove a belief.
class AdversarialTester:
"""Actively seek contradictions to beliefs"""
def design_adversarial_test(self, memory: Memory) -> Test:
"""
Design a test specifically intended to potentially
disprove this memory if it's wrong
"""
# Extract testable claim
claim = self._extract_claim(memory)
# Identify conditions where claim should hold
expected_context = memory.context
# Design test that varies one critical variable
test_context = self._vary_critical_variable(expected_context)
# Predict outcome if claim is true
expected_if_true = self._predict_if_true(claim, test_context)
# Predict outcome if claim is false
expected_if_false = self._predict_if_false(claim, test_context)
return Test(
claim=claim,
context=test_context,
expected_if_true=expected_if_true,
expected_if_false=expected_if_false,
confidence_if_false=memory.confidence * 0.3, # Heavy penalty
confidence_if_true=min(1.0, memory.confidence + 0.1) # Small boost
)
def _vary_critical_variable(self, context: Dict) -> Dict:
"""
Change a critical variable to see if belief still holds
If belief is actually contextual knowledge, it should
fail in different context. If it's universal, it should succeed.
"""
test_context = context.copy()
# Change environment
if 'environment' in context:
test_context['environment'] = 'adversarial-test-env'
# Change scale
if 'scale' in context:
original_scale = context['scale']
test_context['scale'] = f"10x_{original_scale}"
return test_contextIn distributed systems, Byzantine Fault Tolerance handles:
- Malicious actors sending false information
- Faulty nodes providing incorrect data
- Network issues causing inconsistent state
For AI agent memories, we need the same guarantees:
- Agents might have incorrect beliefs (fault)
- Agents might be poisoned (malicious)
- Beliefs might be context-dependent (inconsistent)
class SimpleMajorityConsensus:
"""Require simple majority agreement"""
def __init__(self, fault_tolerance: int = 1):
"""
fault_tolerance: Number of faulty agents to tolerate
Requires: 2 * fault_tolerance + 1 total agents
"""
self.fault_tolerance = fault_tolerance
self.min_agents = 2 * fault_tolerance + 1
def validate(self, belief: str, agents: List[str]) -> Dict:
"""Validate belief with Byzantine fault tolerance"""
if len(agents) < self.min_agents:
raise ValueError(
f"Need at least {self.min_agents} agents for "
f"{self.fault_tolerance}-fault tolerance"
)
# Collect votes
votes = {'confirm': [], 'contradict': [], 'abstain': []}
for agent_id in agents:
vote = self._get_agent_vote(agent_id, belief)
votes[vote].append(agent_id)
# Simple majority
total_votes = len(votes['confirm']) + len(votes['contradict'])
if total_votes == 0:
return {'consensus': False, 'reason': 'no votes'}
confirm_ratio = len(votes['confirm']) / total_votes
# Require supermajority for consensus
consensus_threshold = 0.66 # 2/3 majority
consensus = confirm_ratio >= consensus_threshold
return {
'consensus': consensus,
'confirms': len(votes['confirm']),
'contradicts': len(votes['contradict']),
'abstains': len(votes['abstain']),
'confidence': confirm_ratio
}class PBFTBeliefValidator:
"""
Practical Byzantine Fault Tolerance for critical knowledge
Phases:
1. Pre-prepare: Leader proposes belief
2. Prepare: Agents validate and vote
3. Commit: Final consensus
"""
def __init__(self, agents: List[str], fault_tolerance: int = 1):
self.agents = agents
self.f = fault_tolerance
self.min_replicas = 3 * fault_tolerance + 1
if len(agents) < self.min_replicas:
raise ValueError(
f"PBFT requires {self.min_replicas} agents "
f"for {fault_tolerance}-fault tolerance"
)
async def validate_critical_belief(
self,
belief: Memory,
leader_id: str
) -> Dict:
"""
Validate critical belief using PBFT
Returns consensus only if 2f + 1 agents agree
"""
# Phase 1: Pre-prepare
proposal = {
'belief': belief,
'leader': leader_id,
'sequence': self._get_sequence_number(),
'phase': 'pre-prepare'
}
# Phase 2: Prepare - collect votes
prepare_votes = []
for agent_id in self.agents:
if agent_id == leader_id:
continue
vote = await self._agent_validate(agent_id, belief)
prepare_votes.append((agent_id, vote))
# Need 2f + 1 confirmations (including leader)
confirms = sum(1 for _, vote in prepare_votes if vote == 'confirm')
required = 2 * self.f + 1
if confirms + 1 < required: # +1 for leader
return {
'consensus': False,
'phase': 'prepare',
'confirms': confirms,
'required': required
}
# Phase 3: Commit - final validation
commit_votes = []
for agent_id in self.agents:
# Agents re-validate knowing others confirmed
final_vote = await self._agent_commit(agent_id, belief, prepare_votes)
commit_votes.append((agent_id, final_vote))
final_confirms = sum(1 for _, vote in commit_votes if vote == 'commit')
if final_confirms >= required:
return {
'consensus': True,
'phase': 'commit',
'confirms': final_confirms,
'validators': [aid for aid, vote in commit_votes if vote == 'commit']
}
return {
'consensus': False,
'phase': 'commit',
'confirms': final_confirms,
'required': required
}class QuorumValidator:
"""
Require different quorum sizes based on stakes
Low stakes: Simple majority
Medium stakes: 2/3 supermajority
High stakes: 3/4 supermajority
Critical stakes: Unanimous
"""
def get_required_quorum(self, memory: Memory) -> float:
"""Determine required quorum based on stakes"""
# Assess stakes
stakes = self._assess_stakes(memory)
if stakes == 'critical':
return 1.0 # Unanimous
elif stakes == 'high':
return 0.75 # 3/4 supermajority
elif stakes == 'medium':
return 0.66 # 2/3 supermajority
else:
return 0.51 # Simple majority
def _assess_stakes(self, memory: Memory) -> str:
"""Assess stakes of this belief"""
# Check tags
if any(tag in memory.tags for tag in ['security', 'safety', 'compliance']):
return 'critical'
# Check domain
if memory.context.get('domain') in ['medical', 'legal', 'financial']:
return 'high'
# Check reversibility
if 'irreversible' in memory.tags or 'destructive' in memory.tags:
return 'high'
# Check confidence
if memory.confidence > 0.8:
return 'medium'
return 'low'Track agent reliability over time to weight their validations appropriately.
class AgentReputationSystem:
"""
Track agent reputation based on:
1. Prediction accuracy over time
2. Calibration (confidence matches reality)
3. Validation success (when validating others)
4. Domain expertise
5. Contribution quality
"""
def __init__(self):
self.agents = {} # agent_id -> reputation profile
def initialize_agent(self, agent_id: str):
"""Initialize reputation for new agent"""
self.agents[agent_id] = {
'reputation_score': 0.5, # Start neutral
'total_predictions': 0,
'correct_predictions': 0,
'validation_confirms': 0,
'validation_contradicts': 0,
'calibration_history': [],
'domain_expertise': {},
'trust_ratings': {}, # from other agents
'last_update': datetime.now()
}
def update_on_prediction(
self,
agent_id: str,
predicted: bool,
actual: bool,
confidence: float,
domain: str
):
"""Update reputation after prediction"""
profile = self.agents[agent_id]
# Update counts
profile['total_predictions'] += 1
if predicted == actual:
profile['correct_predictions'] += 1
# Update calibration
calibration_error = abs(confidence - (1.0 if predicted == actual else 0.0))
profile['calibration_history'].append(calibration_error)
# Update domain expertise
if domain not in profile['domain_expertise']:
profile['domain_expertise'][domain] = {'attempts': 0, 'correct': 0}
profile['domain_expertise'][domain]['attempts'] += 1
if predicted == actual:
profile['domain_expertise'][domain]['correct'] += 1
# Recalculate reputation
self._recalculate_reputation(agent_id)
def update_on_validation(
self,
validator_id: str,
validation_outcome: str, # 'correct' or 'incorrect'
stakes: str # 'low', 'medium', 'high', 'critical'
):
"""Update reputation after serving as validator"""
profile = self.agents[validator_id]
if validation_outcome == 'correct':
# Validation was correct - boost reputation
profile['validation_confirms'] += 1
boost = {'low': 0.01, 'medium': 0.02, 'high': 0.04, 'critical': 0.08}[stakes]
profile['reputation_score'] = min(1.0, profile['reputation_score'] + boost)
else:
# Validation was incorrect - penalty
profile['validation_contradicts'] += 1
penalty = {'low': 0.02, 'medium': 0.04, 'high': 0.08, 'critical': 0.16}[stakes]
profile['reputation_score'] = max(0.0, profile['reputation_score'] - penalty)
def _recalculate_reputation(self, agent_id: str):
"""Recalculate overall reputation score"""
profile = self.agents[agent_id]
# Component 1: Prediction accuracy
if profile['total_predictions'] > 0:
accuracy = profile['correct_predictions'] / profile['total_predictions']
else:
accuracy = 0.5
# Component 2: Calibration
if profile['calibration_history']:
avg_calibration_error = sum(profile['calibration_history']) / len(profile['calibration_history'])
calibration_score = 1.0 - avg_calibration_error
else:
calibration_score = 0.5
# Component 3: Validation reliability
total_validations = profile['validation_confirms'] + profile['validation_contradicts']
if total_validations > 0:
validation_reliability = profile['validation_confirms'] / total_validations
else:
validation_reliability = 0.5
# Weighted combination
reputation = (
0.4 * accuracy +
0.3 * calibration_score +
0.3 * validation_reliability
)
profile['reputation_score'] = reputation
profile['last_update'] = datetime.now()
def get_reputation(self, agent_id: str) -> float:
"""Get agent's current reputation score"""
if agent_id not in self.agents:
return 0.5 # Default neutral for unknown agents
return self.agents[agent_id]['reputation_score']
def get_domain_expertise(self, agent_id: str, domain: str) -> float:
"""Get agent's expertise in specific domain"""
if agent_id not in self.agents:
return 0.3 # Default low for unknown
expertise = self.agents[agent_id]['domain_expertise'].get(domain)
if not expertise or expertise['attempts'] == 0:
return 0.3 # Default low for unproven
return expertise['correct'] / expertise['attempts']
def get_weighted_vote(
self,
agent_id: str,
domain: str,
stakes: str
) -> float:
"""Get weighted vote value for this agent"""
# Base weight from reputation
reputation = self.get_reputation(agent_id)
# Boost for domain expertise
domain_expertise = self.get_domain_expertise(agent_id, domain)
# Higher stakes require higher expertise/reputation
stakes_multiplier = {
'low': 1.0,
'medium': domain_expertise,
'high': (reputation + domain_expertise) / 2,
'critical': min(reputation, domain_expertise)
}[stakes]
return reputation * stakes_multiplierTrust should decay over time without interaction and recover through positive interactions.
class TrustEvolution:
"""Manage trust evolution between agents"""
DECAY_HALF_LIFE_DAYS = 90 # Trust decays to 50% in 90 days
def __init__(self):
self.trust_edges = {} # (agent_a, agent_b) -> trust_state
def initialize_trust(self, trustor: str, trustee: str):
"""Initialize trust relationship"""
key = (trustor, trustee)
self.trust_edges[key] = {
'score': 0.5, # Start neutral
'last_interaction': datetime.now(),
'positive_interactions': 0,
'negative_interactions': 0,
'history': []
}
def update_trust(
self,
trustor: str,
trustee: str,
interaction: str, # 'validate-correct', 'validate-incorrect', etc.
stakes: str = 'medium'
):
"""Update trust based on interaction"""
key = (trustor, trustee)
if key not in self.trust_edges:
self.initialize_trust(trustor, trustee)
state = self.trust_edges[key]
current_trust = state['score']
# Apply decay first
days_since = (datetime.now() - state['last_interaction']).days
if days_since > 0:
decay_factor = 0.5 ** (days_since / self.DECAY_HALF_LIFE_DAYS)
# Decay toward neutral (0.5)
current_trust = 0.5 + (current_trust - 0.5) * decay_factor
# Apply interaction
if interaction in ['validate-correct', 'share-good-knowledge', 'helpful']:
# Positive interaction
state['positive_interactions'] += 1
adjustment = self._get_positive_adjustment(current_trust, stakes)
new_trust = min(1.0, current_trust + adjustment)
elif interaction in ['validate-incorrect', 'share-bad-knowledge', 'unhelpful']:
# Negative interaction
state['negative_interactions'] += 1
adjustment = self._get_negative_adjustment(current_trust, stakes)
new_trust = max(0.0, current_trust - adjustment)
else:
# Neutral interaction
new_trust = current_trust
# Update state
state['score'] = new_trust
state['last_interaction'] = datetime.now()
state['history'].append({
'timestamp': datetime.now(),
'interaction': interaction,
'stakes': stakes,
'trust_before': current_trust,
'trust_after': new_trust
})
def _get_positive_adjustment(self, current_trust: float, stakes: str) -> float:
"""Calculate trust increase"""
# Trust increases slowly, especially at high levels
base_increase = {'low': 0.02, 'medium': 0.04, 'high': 0.06, 'critical': 0.08}[stakes]
# Diminishing returns at high trust
diminish = 1.0 - (current_trust - 0.5) # Max at 0.5, min at 1.0
return base_increase * max(0.5, diminish)
def _get_negative_adjustment(self, current_trust: float, stakes: str) -> float:
"""Calculate trust decrease"""
# Trust decreases quickly, especially for high stakes
base_decrease = {'low': 0.04, 'medium': 0.08, 'high': 0.16, 'critical': 0.32}[stakes]
# Larger penalty at high trust (more to lose)
amplify = 0.5 + (current_trust - 0.5) # Min at 0.5, max at 1.0
return base_decrease * amplify
def get_trust(self, trustor: str, trustee: str) -> float:
"""Get current trust score with decay applied"""
key = (trustor, trustee)
if key not in self.trust_edges:
return 0.5 # Default neutral
state = self.trust_edges[key]
current_trust = state['score']
# Apply decay
days_since = (datetime.now() - state['last_interaction']).days
if days_since > 0:
decay_factor = 0.5 ** (days_since / self.DECAY_HALF_LIFE_DAYS)
current_trust = 0.5 + (current_trust - 0.5) * decay_factor
return current_trustclass ValidationRequest:
"""Request another agent to validate a belief"""
def __init__(
self,
belief: Memory,
requestor_id: str,
stakes: str,
context: Dict
):
self.belief = belief
self.requestor_id = requestor_id
self.stakes = stakes
self.context = context
self.id = str(uuid.uuid4())
self.created_at = datetime.now()
class ValidationResponse:
"""Response to validation request"""
def __init__(
self,
request_id: str,
validator_id: str,
outcome: str, # 'confirm', 'contradict', 'context-mismatch', 'unable'
confidence: float,
evidence: str,
reasoning: str
):
self.request_id = request_id
self.validator_id = validator_id
self.outcome = outcome
self.confidence = confidence
self.evidence = evidence
self.reasoning = reasoning
self.timestamp = datetime.now()
class CrossAgentValidator:
"""Coordinate validation across multiple agents"""
def __init__(self, reputation_system: AgentReputationSystem):
self.reputation = reputation_system
self.pending_requests = {}
async def request_validation(
self,
belief: Memory,
requestor_id: str,
validators: List[str],
stakes: str = 'medium'
) -> List[ValidationResponse]:
"""Request validation from multiple agents"""
request = ValidationRequest(
belief=belief,
requestor_id=requestor_id,
stakes=stakes,
context=belief.context
)
self.pending_requests[request.id] = request
# Send to validators
responses = []
for validator_id in validators:
response = await self._send_validation_request(validator_id, request)
if response:
responses.append(response)
return responses
async def _send_validation_request(
self,
validator_id: str,
request: ValidationRequest
) -> Optional[ValidationResponse]:
"""Send validation request to specific agent"""
# Agent should:
# 1. Check their own memories
# 2. Optionally test empirically
# 3. Return their assessment
# This is agent-specific implementation
# Here we show the interface
response = await self._agent_validate(validator_id, request)
return response
def aggregate_validation_responses(
self,
responses: List[ValidationResponse],
requestor_id: str
) -> Dict:
"""Aggregate validation responses with reputation weighting"""
if not responses:
return {
'consensus': False,
'confidence': 0.0,
'reason': 'no responses'
}
# Weight by validator reputation
weighted_confirms = 0.0
weighted_contradicts = 0.0
total_weight = 0.0
for response in responses:
validator_rep = self.reputation.get_reputation(response.validator_id)
weight = validator_rep * response.confidence
if response.outcome == 'confirm':
weighted_confirms += weight
elif response.outcome == 'contradict':
weighted_contradicts += weight
# Ignore 'unable' and 'context-mismatch'
total_weight += weight
if total_weight == 0:
return {
'consensus': False,
'confidence': 0.0,
'reason': 'no valid responses'
}
# Calculate consensus
confirm_ratio = weighted_confirms / total_weight
consensus = confirm_ratio >= 0.6 # 60% threshold
return {
'consensus': consensus,
'confidence': confirm_ratio,
'confirms': sum(1 for r in responses if r.outcome == 'confirm'),
'contradicts': sum(1 for r in responses if r.outcome == 'contradict'),
'context_mismatches': sum(1 for r in responses if r.outcome == 'context-mismatch'),
'responses': responses
}Attacker injects false memories with high confidence.
Defense:
- Signature verification
- Source reputation checking
- Consensus validation before accepting
- Quarantine period for new memories
Attacker creates memories with artificially high confidence.
Defense:
- Cap imported confidence (e.g., max 0.7 for cross-agent)
- Require validation proportional to confidence
- Track confidence-outcome mismatch rate
Attacker creates multiple fake agent identities to dominate consensus.
Defense:
- Reputation-weighted voting
- Cost of identity creation
- Social proof requirements
- Established agent prioritization
Attacker creates memories with misleading context tags.
Defense:
- Context verification against reality
- Anomaly detection in context patterns
- Cross-reference context with other memories
Attacker creates fake evidence to support false beliefs.
Defense:
- Evidence source verification
- Evidence type hierarchy (empirical > documentary)
- Cross-validation of evidence claims
class PoisoningDetector:
"""Detect potentially poisoned memories"""
def __init__(
self,
reputation_system: AgentReputationSystem,
memory_store
):
self.reputation = reputation_system
self.store = memory_store
def assess_memory_risk(self, memory: Memory) -> Dict:
"""Assess risk that memory is poisoned"""
risk_factors = []
risk_score = 0.0
# Factor 1: Source reputation
source_rep = self.reputation.get_reputation(
memory.metadata.get('source_agent_id', 'unknown')
)
if source_rep < 0.4:
risk_factors.append('low-reputation-source')
risk_score += 0.3
# Factor 2: Confidence without validation
if memory.confidence > 0.8 and memory.metadata.get('validation_count', 0) == 0:
risk_factors.append('high-confidence-no-validation')
risk_score += 0.4
# Factor 3: Conflict with established knowledge
conflicts = self._find_conflicting_memories(memory)
if conflicts:
high_rep_conflicts = [
c for c in conflicts
if self.reputation.get_reputation(c.metadata.get('source_agent_id')) > 0.7
]
if high_rep_conflicts:
risk_factors.append('conflicts-with-trusted-knowledge')
risk_score += 0.5
# Factor 4: Statistical anomaly
if self._is_statistical_outlier(memory):
risk_factors.append('statistical-outlier')
risk_score += 0.2
# Factor 5: Rapid confidence increase
if self._has_suspicious_confidence_trajectory(memory):
risk_factors.append('suspicious-confidence-trajectory')
risk_score += 0.3
# Factor 6: Context manipulation indicators
if self._has_context_manipulation_signs(memory):
risk_factors.append('context-manipulation')
risk_score += 0.4
return {
'risk_score': min(1.0, risk_score),
'risk_level': self._categorize_risk(risk_score),
'risk_factors': risk_factors,
'recommendation': self._get_recommendation(risk_score)
}
def _categorize_risk(self, score: float) -> str:
if score >= 0.8:
return 'critical'
elif score >= 0.6:
return 'high'
elif score >= 0.4:
return 'medium'
elif score >= 0.2:
return 'low'
else:
return 'minimal'
def _get_recommendation(self, risk_score: float) -> str:
if risk_score >= 0.8:
return 'REJECT: High risk of poisoning'
elif risk_score >= 0.6:
return 'QUARANTINE: Require strong validation before use'
elif risk_score >= 0.4:
return 'VERIFY: Seek external validation'
elif risk_score >= 0.2:
return 'MONITOR: Watch for anomalies'
else:
return 'ACCEPT: Normal risk level'
def _find_conflicting_memories(self, memory: Memory) -> List[Memory]:
"""Find memories that contradict this one"""
# Search for memories about same topic in similar context
similar = self.store.query_memories(
query=memory.content[:100],
context_filter=memory.context,
min_confidence=0.5
)
# Check for contradictions
conflicts = []
for candidate in similar:
if self._contradicts(memory, candidate):
conflicts.append(candidate)
return conflicts
def _is_statistical_outlier(self, memory: Memory) -> bool:
"""Check if memory is statistical outlier"""
# Compare against distribution of similar memories
similar_domain_memories = self.store.query_memories(
query="",
context_filter={'domain': memory.context.get('domain')},
min_confidence=0.0
)
if len(similar_domain_memories) < 10:
return False # Not enough data
# Check confidence distribution
confidences = [m.confidence for m in similar_domain_memories]
mean_conf = statistics.mean(confidences)
std_conf = statistics.stdev(confidences)
# Is this memory more than 2 standard deviations from mean?
z_score = abs(memory.confidence - mean_conf) / max(0.1, std_conf)
return z_score > 2.0
def _has_suspicious_confidence_trajectory(self, memory: Memory) -> bool:
"""Check if confidence increased too quickly"""
history = memory.metadata.get('confidence_history', [])
if len(history) < 2:
return False
# Check for rapid increases
for i in range(len(history) - 1):
increase = history[i+1]['confidence'] - history[i]['confidence']
time_delta = (history[i+1]['timestamp'] - history[i]['timestamp']).days
# Suspicious if confidence increased >0.3 in <7 days
if increase > 0.3 and time_delta < 7:
return True
return False
def _has_context_manipulation_signs(self, memory: Memory) -> bool:
"""Check for context manipulation"""
context = memory.context
# Overly broad context (trying to apply everywhere)
if len(context) <= 2 and memory.confidence > 0.7:
return True
# Contradictory context values
if context.get('environment') == 'production' and context.get('scale') == 'test':
return True
# Context claims high stakes but has low validation
critical_domains = ['security', 'safety', 'medical', 'financial']
if (context.get('domain') in critical_domains and
memory.metadata.get('validation_count', 0) == 0):
return True
return Falseclass SecurityModel:
"""Comprehensive security for memory systems"""
def __init__(self):
self.defenses = {
'injection': InjectionDefense(),
'sybil': SybilDefense(),
'poisoning': PoisoningDetector(),
'replay': ReplayDefense(),
'timing': TimingAttackDefense()
}
def validate_memory_security(self, memory: Memory) -> Dict:
"""Run all security checks"""
results = {}
threat_level = 0.0
for defense_name, defense in self.defenses.items():
result = defense.check(memory)
results[defense_name] = result
if result['threat_detected']:
threat_level = max(threat_level, result['threat_level'])
return {
'secure': threat_level < 0.5,
'threat_level': threat_level,
'checks': results,
'recommendation': self._get_security_recommendation(threat_level)
}
def _get_security_recommendation(self, threat_level: float) -> str:
if threat_level >= 0.8:
return 'BLOCK: High security threat'
elif threat_level >= 0.6:
return 'QUARANTINE: Isolate and investigate'
elif threat_level >= 0.4:
return 'VALIDATE: Require strong proof'
else:
return 'MONITOR: Normal security posture'import hashlib
import hmac
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.primitives.asymmetric import rsa
class SignatureDefense:
"""Verify cryptographic signatures on memories"""
def __init__(self):
self.public_keys = {} # agent_id -> public_key
def sign_memory(self, memory: Memory, private_key) -> str:
"""Sign memory with agent's private key"""
# Create canonical representation
canonical = self._canonicalize(memory)
# Sign with private key
signature = private_key.sign(
canonical.encode('utf-8'),
padding.PSS(
mgf=padding.MGF1(hashes.SHA256()),
salt_length=padding.PSS.MAX_LENGTH
),
hashes.SHA256()
)
return signature.hex()
def verify_signature(
self,
memory: Memory,
signature: str,
agent_id: str
) -> bool:
"""Verify memory signature"""
# Get agent's public key
public_key = self.public_keys.get(agent_id)
if not public_key:
return False
# Create canonical representation
canonical = self._canonicalize(memory)
# Verify signature
try:
public_key.verify(
bytes.fromhex(signature),
canonical.encode('utf-8'),
padding.PSS(
mgf=padding.MGF1(hashes.SHA256()),
salt_length=padding.PSS.MAX_LENGTH
),
hashes.SHA256()
)
return True
except:
return False
def _canonicalize(self, memory: Memory) -> str:
"""Create canonical string representation for signing"""
# Deterministic ordering
return json.dumps({
'id': memory.id,
'content': memory.content,
'confidence': memory.confidence,
'context': memory.context,
'created_at': memory.created_at.isoformat()
}, sort_keys=True)class AdversarialValidationSystem:
"""
Complete implementation of adversarial validation
Combines:
- Reputation tracking
- Cross-agent validation
- Byzantine consensus
- Poisoning detection
- Trust evolution
"""
def __init__(self, agent_id: str, memory_store):
self.agent_id = agent_id
self.memory_store = memory_store
# Components
self.reputation = AgentReputationSystem()
self.trust = TrustEvolution()
self.poisoning_detector = PoisoningDetector(self.reputation, memory_store)
self.cross_validator = CrossAgentValidator(self.reputation)
self.consensus = SimpleMajorityConsensus(fault_tolerance=1)
async def validate_high_confidence_belief(
self,
memory: Memory,
available_agents: List[str]
) -> Dict:
"""
Full validation pipeline for high-confidence belief
Returns validation result with confidence adjustment
"""
# Step 1: Check for poisoning
poisoning_risk = self.poisoning_detector.assess_memory_risk(memory)
if poisoning_risk['risk_level'] in ['critical', 'high']:
return {
'validated': False,
'confidence_multiplier': 0.3,
'reason': f"Poisoning risk: {poisoning_risk['risk_level']}"
}
# Step 2: Request cross-agent validation
stakes = self._assess_stakes(memory)
num_validators = self._get_required_validators(memory.confidence, stakes)
# Select validators by reputation
validators = self._select_validators(
available_agents,
num_validators,
memory.context.get('domain')
)
responses = await self.cross_validator.request_validation(
memory,
self.agent_id,
validators,
stakes
)
# Step 3: Aggregate with reputation weighting
aggregated = self.cross_validator.aggregate_validation_responses(
responses,
self.agent_id
)
# Step 4: Update trust based on validation
for response in responses:
self._update_validator_trust(response, memory)
# Step 5: Return validation result
if aggregated['consensus']:
# Validated - boost confidence slightly
return {
'validated': True,
'confidence_multiplier': 1.1,
'validators': [r.validator_id for r in responses],
'consensus_confidence': aggregated['confidence']
}
else:
# Not validated - reduce confidence
return {
'validated': False,
'confidence_multiplier': 0.6,
'validators': [r.validator_id for r in responses],
'reason': 'Failed consensus validation'
}
def _select_validators(
self,
available_agents: List[str],
num_needed: int,
domain: str
) -> List[str]:
"""Select best validators by reputation and domain expertise"""
# Score each agent
scored = []
for agent_id in available_agents:
if agent_id == self.agent_id:
continue # Don't validate own memories
reputation = self.reputation.get_reputation(agent_id)
expertise = self.reputation.get_domain_expertise(agent_id, domain)
trust = self.trust.get_trust(self.agent_id, agent_id)
# Combined score
score = (reputation * 0.4 + expertise * 0.4 + trust * 0.2)
scored.append((agent_id, score))
# Sort by score and take top N
scored.sort(key=lambda x: x[1], reverse=True)
return [agent_id for agent_id, _ in scored[:num_needed]]
def _get_required_validators(self, confidence: float, stakes: str) -> int:
"""Determine how many validators needed"""
if stakes == 'critical':
return max(5, int(confidence * 7)) # 5-7 validators
elif stakes == 'high':
return max(3, int(confidence * 5)) # 3-5 validators
elif stakes == 'medium':
return max(2, int(confidence * 3)) # 2-3 validators
else:
return 1 # Low stakes, one validator sufficientclass ValidationTriggers:
"""Determine when validation is required"""
def should_validate(self, memory: Memory) -> Tuple[bool, str]:
"""
Determine if memory needs validation
Returns: (should_validate, reason)
"""
# Trigger 1: High confidence without recent validation
if memory.confidence > 0.7:
days_since_validation = self._days_since_last_validation(memory)
if days_since_validation > 30:
return (True, "high-confidence-stale")
# Trigger 2: About to be used in high-stakes context
if self._will_be_used_in_critical_context(memory):
return (True, "critical-context-imminent")
# Trigger 3: Conflict with new information
if self._conflicts_with_recent_memory(memory):
return (True, "conflict-detected")
# Trigger 4: Periodic audit
if self._due_for_periodic_audit(memory):
return (True, "periodic-audit")
# Trigger 5: Low confidence but frequently used
if memory.confidence < 0.6 and memory.metadata.get('usage_count', 0) > 10:
return (True, "high-usage-low-confidence")
# Trigger 6: External request
if memory.metadata.get('validation_requested'):
return (True, "external-request")
return (False, "no-trigger")
def get_validation_priority(self, memory: Memory) -> int:
"""
Get validation priority (1-10, 10 = highest)
"""
priority = 5 # Default medium
# Higher priority for high-confidence
if memory.confidence > 0.8:
priority += 3
elif memory.confidence > 0.6:
priority += 1
# Higher priority for critical domains
if memory.context.get('domain') in ['security', 'safety', 'medical']:
priority += 2
# Higher priority if never validated
if memory.metadata.get('validation_count', 0) == 0:
priority += 2
# Higher priority if frequently used
if memory.metadata.get('usage_count', 0) > 20:
priority += 1
return min(10, priority)class ValidationScheduler:
"""Schedule and manage validation queue"""
def __init__(self, validation_system: AdversarialValidationSystem):
self.validation_system = validation_system
self.validation_queue = [] # Priority queue
self.validations_in_progress = {}
def schedule_validation(self, memory: Memory, priority: int):
"""Add memory to validation queue"""
item = {
'memory': memory,
'priority': priority,
'scheduled_at': datetime.now()
}
heapq.heappush(self.validation_queue, (-priority, item))
async def process_validation_queue(
self,
available_agents: List[str],
max_concurrent: int = 3
):
"""Process validation queue"""
while len(self.validations_in_progress) < max_concurrent and self.validation_queue:
# Get highest priority item
priority, item = heapq.heappop(self.validation_queue)
memory = item['memory']
# Start validation
task = asyncio.create_task(
self.validation_system.validate_high_confidence_belief(
memory,
available_agents
)
)
self.validations_in_progress[memory.id] = {
'task': task,
'started_at': datetime.now()
}
# Clean up completed validations
completed = []
for memory_id, info in self.validations_in_progress.items():
if info['task'].done():
completed.append(memory_id)
for memory_id in completed:
del self.validations_in_progress[memory_id]Adversarial validation is not optional—it's the critical safeguard that prevents Lamarckian AI systems from drifting into self-confirming hallucination.
-
Internal metrics aren't enough: Accuracy, calibration, and consistency can all look good while the system is completely wrong about reality.
-
External contradiction is required: High-confidence beliefs MUST be tested against external sources that can contradict them.
-
Byzantine consensus prevents drift: Multi-agent validation with fault tolerance ensures robust knowledge even with some incorrect agents.
-
Reputation systems enable trust: Tracking agent reliability over time allows proper weighting of validations.
-
Poisoning prevention is essential: Cryptographic signatures, anomaly detection, and security protocols protect against malicious knowledge injection.
-
Validation must be continuous: Periodic audits, trigger-based validation, and adversarial testing keep knowledge grounded in reality.
Lamarckian Evolution (V2.0) =
Individual Learning (V1.0)
+ External Validation (this document)
+ Multi-Agent Coordination
+ Anti-Gaming Mechanisms
Phase 1 (Minimum Viable):
- Basic external validation tracking
- Simple consensus voting
- Poisoning detection
Phase 2 (Production-Ready):
- Reputation system
- Byzantine consensus
- Trust evolution
- Signature verification
Phase 3 (Advanced):
- PBFT for critical beliefs
- Domain specialization
- Automated validation scheduling
- Cross-provider validation networks
Start with external validation. It's the difference between a system that learns and a system that hallucinates.
Version: 1.0 Last Updated: 2026-02-04 License: Open source, any use permitted Philosophy: Trust, but verify. Reality is the ultimate validator. External contradiction prevents drift.