TL;DR
I built IMMUNE because I hit a wall: my AI agent kept giving wrong answers and I had no idea which stored memory was causing it. IMMUNE applies delta debugging — a classical algorithm from software testing — to pinpoint the exact memory responsible. No LLM judge, no guessing. It’s a drop-in for LangGraph and quarantines bad memories at the Redis index level. This post explains how it works, shows the code, and is honest about what it can’t do yet.
The Problem: Memory Attribution in AI Agents
I ran into this while building an agent that accumulated memories across sessions. At some point it started giving answers that were just wrong — confidently wrong. The hard part wasn’t fixing it. The hard part was figuring out which memory caused it. With dozens of stored memories, there’s no obvious way to trace a wrong answer back to its source.
Today's approaches fail in predictable ways:
LLM-as-judge: Non-deterministic, expensive, hallucinates "explanations" that aren't real causal chains
Attention-weight analysis: Requires white-box model access; doesn't generalize across providers
Manual review: Doesn't scale past a few dozen memories
Embedding similarity: Finds topically related memories, not causally responsible ones
What we need is a method that isolates the minimal set of memories that actually causes a specific wrong answer—deterministically, without requiring model internals.
The Approach: Applying Delta Debugging to Memory Attribution
Core Insight
The idea came from reading about delta debugging in software testing. Zeller’s 2002 paper describes ddmin: given a binary predicate (does this input cause the failure?), find the minimal failing subset in O(n log n) calls. I realized the problem maps directly. Instead of minimizing a sequence of code changes, I’m minimizing a set of retrieved memories. The predicate is the same: does this subset reproduce the wrong answer?

Algorithm: ddmin Applied to Memory Attribution
The delta_debug function below implements Zeller's ddmin:
def delta_debug(candidates, question, failing_answer, n=2):
"""
ddmin: find minimal failing subset of candidate memories.
candidates: list of memory dicts [{id, content, embedding...}]
question: the query that produced the wrong answer
failing_answer: the incorrect output we are trying to explain
n: current granularity (starts at 2, doubles each round)
"""
if len(candidates) == 1:
return candidates # base case: single memory is culprit

chunk_size = max(1, len(candidates) // n)
chunks = [candidates[i*chunk_size:(i+1)*chunk_size]
for i in range(n)]
# add remainder if n doesn't divide evenly
if len(candidates) % n:
chunks[-1] += candidates[n*chunk_size:]

# 1. Try subsets — one chunk is the culprit
for i, chunk in enumerate(chunks):
if still_fails(chunk, question, failing_answer):
return delta_debug(chunk, question, failing_answer, 2)

# 2. Try complements — culprit is in what's left over
for i, chunk in enumerate(chunks):
complement = [m for m in candidates if m not in chunk]
if still_fails(complement, question, failing_answer):
return delta_debug(complement, question, failing_answer,
max(2, n - 1))

# 3. Increase granularity
if n < len(candidates):
return delta_debug(candidates, question, failing_answer,
min(2*n, len(candidates)))

return candidates # can't reduce further

Predicate: still_fails()
The oracle predicate replays the agent's answer using only the candidate subset and then checks for semantic contradiction against the ground-truth answer.

def still_fails(memory_subset, question, failing_answer):
"""
Returns True if this memory_subset reproduces the failing answer.
Uses exact string match for now; semantic match via embedding
similarity is the planned upgrade (see Limitations).
"""
context = build_context(memory_subset)
candidate_answer = replay_query(context, question)
return is_contradiction(candidate_answer, failing_answer)

def is_contradiction(answer_a, answer_b):
"""
Deterministic contradiction check. No LLM judge.
Uses ContradictionDetector: a rule-based + embedding similarity
approach that flags semantic opposites.
"""
detector = ContradictionDetector()
return detector.contradicts(answer_a, answer_b)

class ContradictionDetector:
NEGATION_PAIRS = [
("is", "is not"), ("can", "cannot"), ("will", "will not"),
("should", "should not"), ("does", "does not"),
("allergic", "not allergic"), ("safe", "unsafe"),
]

def contradicts(self, a, b):
a_lower, b_lower = a.lower(), b.lower()
for pos, neg in self.NEGATION_PAIRS:
if pos in a_lower and neg in b_lower:
return True
if neg in a_lower and pos in b_lower:
return True
# Embedding cosine similarity fallback
sim = cosine_similarity(embed(a), embed(b))
return sim < -0.3 # strong semantic opposition
System Architecture
Component Overview
IMMUNE wraps LangGraph's BaseStore interface as a drop-in replacement:

Before IMMUNE (standard LangGraph)

from langgraph.store.memory import InMemoryStore
store = InMemoryStore()

After IMMUNE (drop-in)

from immune import ImmuneStore
store = ImmuneStore(redis_url='redis://localhost:6379')

All other LangGraph code unchanged

Redis Quarantine Enforcement
Bad memories are quarantined at the Redis index level—they are never returned by search, even if the application code forgets to filter them:

Active memories index (what agents see)

FT.CREATE memory_idx ON HASH PREFIX 1 mem:
SCHEMA
content TEXT WEIGHT 2.0
status TAG # ← 'active' | 'quarantined'
embedding VECTOR HNSW 6 TYPE FLOAT32 DIM 1536 DISTANCE_METRIC COSINE

Search query — quarantined memories are structurally excluded

FT.SEARCH memory_idx
'(@status:{active})=>[KNN 10 @embedding $vec AS score]'
PARAMS 2 vec $query_vector
SORTBY score ASC
DIALECT 2

This is the key architectural decision: quarantine is enforced at the database layer, not the application layer. Even if a future developer removes the filter from the application code, the index schema guarantees that quarantined memories are excluded.

Observability: Arize Phoenix + OpenTelemetry

Every attribution trace is recorded

from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

tracer = trace.get_tracer('immune')

with tracer.start_as_current_span('memory_attribution') as span:
span.set_attribute('total_candidates', len(candidates))
span.set_attribute('oracle_calls', oracle_call_count)
span.set_attribute('minimal_culprits', len(result))
span.set_attribute('quarantine_triggered', bool(result))
# Arize Phoenix receives this span for visualization

MCP Tool Integration

FastMCP server exposes IMMUNE tools to any MCP-compatible host

from fastmcp import FastMCP
mcp = FastMCP('IMMUNE Memory Safety')

@mcp.tool()
def quarantine_memory(memory_id: str, reason: str) -> dict:
"""Quarantine a memory by ID. Reason is logged for audit."""
redis_client.hset(f'mem:{memory_id}', 'status', 'quarantined')
redis_client.hset(f'mem:{memory_id}', 'quarantine_reason', reason)
return {'quarantined': memory_id, 'status': 'success'}

@mcp.tool()
def attribute_answer(question: str, answer: str) -> dict:
"""Run delta debugging attribution for a given question/answer pair."""
candidates = fetch_relevant_memories(question)
culprits = delta_debug(candidates, question, answer)
return {'culprits': [m['id'] for m in culprits],
'count': len(culprits)}
***Results*
Benchmarks (Synthetic Test Suite)**
Tests run on a MacBook Pro M2, Redis 7.2, and Python 3.11. Memory pool size varied from 10 to 200 entries. Each scenario injected 1–3 poisoned memories and measured whether delta debugging isolated exactly those memories.

Scenario	Pool Size	Poisoned	Oracle Calls	Detected?	False Positives
Single poison, 10 memories	10	1	8	✓ Yes	0
Single poison, 50 memories	50	1	22	✓ Yes	0
Single poison, 100 memories	100	1	39	✓ Yes	0
Dual poison, 50 memories	50	2	41	✓ Yes	0
Dual poison, 100 memories	100	2	74	✓ Yes	1*
Triple poison, 50 memories	50	3	89	✓ Yes	0
No poison (baseline)	50	0	0	— (no flag)	0
Semantic contradiction	50	1	28	✓ Yes	0

One healthy memory falsely included in culprit set; investigation showed it was semantically adjacent to the poisoned memory, not causally related. See Limitations.

Performance Characteristics
Oracle call count grows as O(n log n)—consistent with the ddmin theory
Each oracle call: ~200–400 ms on a local LLM (Ollama llama3). On GPT-4o-mini: ~800ms
Attribution of 100 memories with 1 culprit: ~8 seconds end-to-end
Quarantine write: <5 ms (Redis HSET)
Memory retrieval overhead vs. standard InMemoryStore: +12ms average (RediSearch KNN)
Tech Stack
Python 3.11 — core runtime
LangGraph—agent framework (ImmuneStore is a BaseStore drop-in)
Redis 7.2 + RediSearch—vector store with schema-level quarantine
OpenAI text-embedding-3-small (or local Ollama)—memory embeddings
Arize Phoenix + OpenTelemetry—attribution trace observability
FastMCP—MCP server for tool integration
pytest + pytest-asyncio — test suite (42 tests)
Docker Compose — local dev environment (Redis + Phoenix)
How to Run IMMUNE
Prerequisites
Python 3.11+
Docker (for Redis + Phoenix)
OpenAI API key (or Ollama running locally)

Setup

1. Clone and install

git clone https://github.com/Sixxhi/IMMUNE
cd IMMUNE
pip install -r requirements.txt

2. Start Redis and Arize Phoenix

docker-compose up -d

3. Configure environment

cp .env.example .env

Edit .env: set OPENAI_API_KEY or OLLAMA_BASE_URL

4. Run the demo

python examples/basic_attribution.py

5. Run tests

pytest tests/ -v

Basic Usage
from immune import ImmuneStore, AttributionEngine

Initialize store

store = ImmuneStore(
redis_url='redis://localhost:6379',
embedding_model='text-embedding-3-small'
)

Write memories normally

await store.aput(('user', 'alice'), 'pref_001', {
'content': 'User prefers metric units',
'source': 'conversation_2024_10_01'
})

When you detect a wrong answer, attribute it

engine = AttributionEngine(store)
result = await engine.attribute(
question='What is the user preferred unit system?',
wrong_answer='User prefers imperial units'
)
print(result.culprits) # [{'id': 'mem_xyz', 'content': '...'}]
await engine.quarantine(result.culprits) # auto-quarantine
Honest Limitations
I want to be direct about what IMMUNE cannot do yet:
Semantic predicate gap: The current ContradictionDetector uses rule-based negation pairs and embedding similarity. It misses subtle domain-specific contradictions (e.g., medical dosing errors where both sentences sound positive).
Predicate quality is the bottleneck: ddmin only isolates the minimum failing set relative to the predicate. If the predicate cannot distinguish 'wrong answer caused by memory X' from 'wrong answer caused by combination of X and Y', ddmin will return both as culprits.
Single-turn attribution only: Multi-turn conversation poisoning (where the attack spans several memories) is not yet addressed.
No production load testing: The current benchmark is synthetic. Real production memory systems may have distribution shifts that affect both retrieval and the contradiction check.
Embedding model dependency: Quarantine enforcement is strong at the Redis layer, but the embedding model used for retrieval is a soft dependency. Switching models mid-deployment could affect which memories surface as candidates.
No adversarial robustness testing: A sophisticated attacker who knows the ddmin algorithm could craft memories designed to evade the predicate. This aspect has not been studied.
Future Work
Semantic predicate upgrade: Replace rule-based contradiction Detector with a small fine-tuned classifier trained on (question, answer-a, answer-b, contradicts?) triples. Target: <10ms inference, no API call.
Multi-turn attribution: Extend ddmin to operate over conversation-turn × memory-id pairs, isolating poisoning that requires multiple memories to activate.
Adversarial benchmark: Design a red-team evaluation where an attacker who knows the DDIM structure attempts to craft poison memories that evade detection.
Formal verification: Prove soundness and completeness properties for specific predicate classes (e.g., monotone contradiction predicates).
Production pilot: Deploy IMMUNE in a real agent system and measure wall-clock attribution latency, false positive rate, and operator usability.
Related Work
Zeller, A. (2002). The ddmin algorithm isolates cause-effect chains from computer programs and is the theoretical foundation of IMMUNE. SIGSOFT FSE—ddmin algorithm, the theoretical foundation of IMMUNE.
Zou et al. (2023). Universal and Transferable Adversarial Attacks on Aligned Language Models. — shows LLMs can be manipulated through input modification; IMMUNE addresses the memory-layer variant.
Shi et al. (2023). Large Language Models Can Be Easily Distracted by Irrelevant Context. — retrieval interference in RAG systems directly motivates the attribution problem.
Lewis et al. (2020). Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks—a foundational RAG architecture that IMMUNE extends with safety guarantees.
Perez & Ribeiro (2022). Ignore Previous Prompt: Attack Techniques for Language Models.—Prompt injection as a class; memory poisoning is the persistent-storage variant.
LangGraph documentation (2024). BaseStore interface—IMMUNE's drop-in integration point.
Redis documentation (2024). RediSearch vector similarity. — the storage and quarantine enforcement layer.
Conclusion
Memory poisoning is a real problem in deployed agent systems and I don’t think it gets enough attention. Most safety work focuses on training time. But agents in production accumulate memories that can go bad — that’s a runtime problem.
IMMUNE is my attempt to address it concretely. It’s not production-ready and I say so clearly in the limitations. But the core approach — applying ddmin to memory attribution — works for the cases I tested, and I think it’s worth building on.