Understanding Multi-Agent Systems Failure Modes
Multi-agent systems (MAS) have become a dominant paradigm for building complex AI applications. Instead of relying on a single monolithic model, developers orchestrate multiple specialized agents—each with its own prompt, tools, and memory—to collaborate on tasks. However, as agent count grows, so do the failure surfaces. This tutorial catalogs the most common failure modes, explains why they occur, and provides concrete fixes with production-ready code.
What Are Multi-Agent Systems?
A multi-agent system consists of two or more autonomous agents that communicate, coordinate, and act to achieve individual or collective goals. Each agent typically has:
- A persona or role defined by a system prompt
- Tools such as calculators, search APIs, or database connectors
- Memory (short-term conversation history or long-term vector stores)
- A communication protocol for passing messages to other agents or a supervisor
Common architectures include hierarchical (supervisor-worker), peer-to-peer debate, and graph-based workflows where edges define agent-to-agent message flow.
Why Failure Modes Matter
A single-agent failure is manageable—you retry or fall back. But in multi-agent systems, failures cascade, amplify, and create subtle emergent bugs that are hard to reproduce. An agent that silently emits malformed JSON can corrupt downstream agents' state. Two agents locked in an infinite debate loop can burn through your token budget in minutes. Understanding these failure modes is essential for building reliable, cost-effective agent systems that operate at scale.
Failure Mode 1: Infinite Agent Loops
What It Looks Like
Two or more agents enter a circular conversation pattern where each response triggers the other to respond again, creating an endless chain. This commonly happens in debate architectures where agents are conditioned to "always provide a counterpoint" or in supervisor-worker setups where a worker's output consistently fails validation, triggering infinite retries.
Why It Happens
The root cause is a missing termination condition. Agents are programmed with open-ended response triggers—"If you receive a message, respond with your analysis"—without a convergence criterion. LLMs, being probabilistic, rarely produce identical outputs, so a simple "stop if the last message is identical" check often fails.
Code Example: The Broken Pattern
# BROKEN: Agents loop forever because each response triggers the other
import asyncio
from typing import List, Dict
class DebateAgent:
def __init__(self, name: str, stance: str):
self.name = name
self.stance = stance
self.history: List[Dict[str, str]] = []
async def respond(self, opponent_message: str) -> str:
# LLM call simulation — always returns a counterargument
prompt = f"You are {self.name}. Your stance: {self.stance}\n"
prompt += f"Opponent said: {opponent_message}\n"
prompt += "Provide a counterargument. Always respond with new points."
# This prompt guarantees infinite debate because it mandates a response
response = await llm_call(prompt) # hypothetical async call
self.history.append({"role": "opponent", "content": opponent_message})
self.history.append({"role": "self", "content": response})
return response
async def run_debate_forever():
agent_a = DebateAgent("Agent A", "Python is superior for data science")
agent_b = DebateAgent("Agent B", "R is superior for data science")
msg = "Let's begin the debate."
# BROKEN: No termination condition
while True: # This will run until you run out of tokens or money
msg = await agent_a.respond(msg)
print(f"Agent A: {msg[:60]}...")
msg = await agent_b.respond(msg)
print(f"Agent B: {msg[:60]}...")
# No break condition — infinite loop guaranteed
The Fix: Explicit Termination Guards
# FIXED: Multiple termination conditions prevent infinite loops
import asyncio
import hashlib
from typing import List, Dict, Optional
from enum import Enum
class DebateOutcome(Enum):
CONTINUE = "continue"
CONVERGED = "converged"
MAX_ROUNDS = "max_rounds"
REPETITION_DETECTED = "repetition_detected"
class FixedDebateAgent:
def __init__(self, name: str, stance: str, max_rounds: int = 5):
self.name = name
self.stance = stance
self.max_rounds = max_rounds
self.history: List[Dict[str, str]] = []
self.response_hashes: set[str] = set()
def _hash_response(self, text: str) -> str:
# Normalize for comparison: lowercase, strip whitespace, remove punctuation
normalized = text.lower().strip()
return hashlib.md5(normalized.encode()).hexdigest()
async def respond(self, opponent_message: str, round_num: int) -> tuple[str, DebateOutcome]:
# Check 1: Max rounds exceeded
if round_num >= self.max_rounds:
return "Debate concluded: maximum rounds reached.", DebateOutcome.MAX_ROUNDS
# Check 2: Detect near-duplicate responses (semantic repetition)
opponent_hash = self._hash_response(opponent_message)
if opponent_hash in self.response_hashes:
return "I notice we're repeating arguments. Let's agree to disagree.", DebateOutcome.REPETITION_DETECTED
prompt = f"""You are {self.name}. Your stance: {self.stance}
Opponent said: {opponent_message}
Round {round_num + 1} of {self.max_rounds}.
Provide a counterargument. IMPORTANT: If you've exhausted new points or notice repetition,
explicitly state: "CONVERGENCE: We have reached an impasse" and summarize areas of agreement."""
response = await llm_call(prompt)
# Check 3: Look for explicit convergence signal from the LLM
if "CONVERGENCE:" in response.upper():
return response.replace("CONVERGENCE:", "").strip(), DebateOutcome.CONVERGED
# Track response to detect future repetition
self.response_hashes.add(self._hash_response(response))
self.history.append({"round": round_num, "content": response})
return response, DebateOutcome.CONTINUE
async def run_fixed_debate():
agent_a = FixedDebateAgent("Agent A", "Python is superior for data science", max_rounds=5)
agent_b = FixedDebateAgent("Agent B", "R is superior for data science", max_rounds=5)
msg = "Let's begin the debate."
outcome = DebateOutcome.CONTINUE
round_num = 0
while outcome == DebateOutcome.CONTINUE:
msg, outcome = await agent_a.respond(msg, round_num)
print(f"[Round {round_num}] Agent A ({outcome.value}): {msg[:80]}...")
if outcome != DebateOutcome.CONTINUE:
break
msg, outcome = await agent_b.respond(msg, round_num)
print(f"[Round {round_num}] Agent B ({outcome.value}): {msg[:80]}...")
round_num += 1
print(f"\nDebate terminated with outcome: {outcome.value} after {round_num} rounds")
Best Practices for Loop Prevention
- Hard cap on iterations — always set
max_roundsormax_stepsat the orchestration level - Hash-based repetition detection — track normalized response hashes and break on duplicates
- LLM self-awareness prompts — instruct agents to emit a convergence token (like
CONVERGENCE:) when they detect circularity - Token budget monitoring — implement a global token counter that raises an exception when exceeded
Failure Mode 2: Silent Output Corruption
What It Looks Like
One agent produces a malformed output—truncated JSON, an unclosed XML tag, or a response that mixes formats—and the downstream agent either crashes or, worse, silently interprets the corrupted data incorrectly. This is particularly dangerous in tool-calling chains where an agent emits parameters that another agent executes without validation.
Why It Happens
LLMs are stochastic. Even with strong prompting, they occasionally drop closing brackets, hallucinate parameter names, or emit text outside the expected structured block. When agents trust raw LLM output implicitly, corruption propagates unchecked.
Code Example: The Broken Pattern
# BROKEN: No validation between agents — corrupted JSON propagates
import json
from typing import Any
class DataExtractorAgent:
async def extract_numbers(self, text: str) -> dict:
prompt = f"""Extract all numbers from this text as a JSON object with keys 'integers' and 'floats'.
Return ONLY valid JSON, nothing else.
Text: {text}"""
raw_response = await llm_call(prompt)
# BROKEN: Directly parsing without validation or repair
return json.loads(raw_response) # Will crash on malformed JSON
class CalculatorAgent:
async def calculate_sum(self, extracted_data: dict) -> float:
integers = extracted_data.get('integers', [])
floats = extracted_data.get('floats', [])
return sum(integers) + sum(floats)
# BROKEN: If 'integers' key is misspelled 'integerss' due to LLM typo,
# it silently defaults to empty list and returns wrong sum
async def broken_pipeline(text: str):
extractor = DataExtractorAgent()
calculator = CalculatorAgent()
data = await extractor.extract_numbers(text)
# No validation — corrupted data flows directly into calculation
result = await calculator.calculate_sum(data)
print(f"Sum: {result}") # May be silently wrong
The Fix: Validation Layers and Repair Strategies
# FIXED: Multi-layer validation with automatic repair and circuit breakers
import json
import re
from typing import Any, Optional
from dataclasses import dataclass
from jsonschema import validate, ValidationError
@dataclass
class ValidatedExtraction:
integers: list[int]
floats: list[float]
raw_response: str
repair_attempts: int
is_valid: bool
class RobustDataExtractorAgent:
def __init__(self, max_repair_attempts: int = 3):
self.max_repair_attempts = max_repair_attempts
# Define expected schema for validation
self.schema = {
"type": "object",
"properties": {
"integers": {"type": "array", "items": {"type": "integer"}},
"floats": {"type": "array", "items": {"type": "number"}}
},
"required": ["integers", "floats"],
"additionalProperties": False
}
def _extract_json_block(self, text: str) -> Optional[str]:
"""Extract JSON from text even if surrounded by markdown or extra text."""
# Try to find JSON block in markdown code fences
fence_match = re.search(r'(?:json)?\s*\n?(.*?)\n?', text, re.DOTALL)
if fence_match:
return fence_match.group(1).strip()
# Try to find raw JSON object
brace_match = re.search(r'\{.*\}', text, re.DOTALL)
if brace_match:
return brace_match.group(0).strip()
return None
def _repair_json(self, json_str: str) -> Optional[str]:
"""Attempt common repairs on malformed JSON."""
repaired = json_str
# Fix 1: Remove trailing commas before closing brackets
repaired = re.sub(r',\s*}', '}', repaired)
repaired = re.sub(r',\s*]', ']', repaired)
# Fix 2: Add missing closing brackets by counting open/close pairs
open_braces = repaired.count('{')
close_braces = repaired.count('}')
open_brackets = repaired.count('[')
close_brackets = repaired.count(']')
if open_braces > close_braces:
repaired += '}' * (open_braces - close_braces)
if open_brackets > close_brackets:
repaired += ']' * (open_brackets - close_brackets)
# Fix 3: Quote unquoted keys (simple heuristic)
repaired = re.sub(r'(\s*)(\w+)(\s*):', r'\1"\2"\3:', repaired)
# Fix 4: Replace single quotes with double quotes
# (careful: only outside already-quoted strings — simplified here)
repaired = repaired.replace("'", '"')
return repaired
async def extract_numbers(self, text: str) -> ValidatedExtraction:
prompt = f"""Extract all numbers from this text as JSON.
You MUST respond with ONLY a valid JSON object in this exact format:
{{
"integers": [1, 2, 3],
"floats": [1.5, 2.7]
}}
Do not include markdown, explanations, or extra text.
Text: {text}"""
raw_response = await llm_call(prompt)
repair_attempts = 0
# Attempt 0: Direct parse
try:
parsed = json.loads(raw_response)
validate(parsed, self.schema)
return ValidatedExtraction(
integers=parsed['integers'],
floats=parsed['floats'],
raw_response=raw_response,
repair_attempts=0,
is_valid=True
)
except (json.JSONDecodeError, ValidationError) as e:
print(f"[WARNING] Initial parse failed: {e}")
# Attempts 1..N: Extract JSON block and repair
json_block = self._extract_json_block(raw_response)
if json_block:
for attempt in range(1, self.max_repair_attempts + 1):
repaired = self._repair_json(json_block)
try:
parsed = json.loads(repaired)
validate(parsed, self.schema)
print(f"[INFO] Successfully repaired JSON on attempt {attempt}")
return ValidatedExtraction(
integers=parsed['integers'],
floats=parsed['floats'],
raw_response=raw_response,
repair_attempts=attempt,
is_valid=True
)
except (json.JSONDecodeError, ValidationError):
print(f"[WARNING] Repair attempt {attempt} failed")
json_block = repaired # Iteratively improve
# All attempts exhausted — raise with context
raise ValueError(
f"Failed to extract valid JSON after {self.max_repair_attempts} repair attempts.\n"
f"Raw response: {raw_response[:500]}..."
)
class ValidatedCalculatorAgent:
async def calculate_sum(self, extraction: ValidatedExtraction) -> dict:
"""Returns result with metadata for observability."""
if not extraction.is_valid:
raise ValueError("Cannot calculate on invalid extraction")
total = sum(extraction.integers) + sum(extraction.floats)
return {
"total_sum": total,
"integer_count": len(extraction.integers),
"float_count": len(extraction.floats),
"repair_attempts": extraction.repair_attempts,
"confidence": 1.0 if extraction.repair_attempts == 0 else max(0.5, 1.0 - (extraction.repair_attempts * 0.2))
}
async def fixed_pipeline(text: str):
extractor = RobustDataExtractorAgent(max_repair_attempts=3)
calculator = ValidatedCalculatorAgent()
try:
validated_data = await extractor.extract_numbers(text)
result = await calculator.calculate_sum(validated_data)
print(f"Sum: {result['total_sum']} (confidence: {result['confidence']:.2f})")
print(f"Integers found: {result['integer_count']}, Floats found: {result['float_count']}")
print(f"Repair attempts needed: {result['repair_attempts']}")
except ValueError as e:
print(f"[FATAL] Pipeline failed: {e}")
# Fallback: route to human review queue or simpler pipeline
Best Practices for Output Validation
- Schema validation on every agent boundary — use JSON Schema, Pydantic models, or custom validators
- Implement repair strategies — regex-based bracket fixing, re-prompting with stricter instructions, or falling back to a simpler extraction method
- Attach metadata to outputs — track repair attempts, confidence scores, and raw responses for debugging
- Circuit breaker pattern — if an agent fails N times consecutively, stop the pipeline and alert rather than retrying indefinitely
Failure Mode 3: Goal Misalignment and Drift
What It Looks Like
An agent optimizes for its local objective at the expense of the global task. For example, a "Research Agent" tasked with gathering information may produce an exhaustive 50-page report when the downstream "Summary Agent" only needed five bullet points. The system wastes compute and the summary agent receives context it cannot effectively compress.
Why It Happens
Agents are given isolated system prompts that define their success criteria without awareness of the global objective. Without explicit constraints on output size, format, or relevance scoring, agents default to "be helpful and thorough," which often means "produce as much as possible."
The Fix: Constraint Contracts and Global Context Injection
# FIXED: Agents receive explicit output contracts and global task context
from typing import Optional
from dataclasses import dataclass, field
from enum import Enum
class OutputVerbosity(Enum):
MINIMAL = "minimal" # Bullet points only
STANDARD = "standard" # Paragraph-level
EXHAUSTIVE = "exhaustive" # Full report
@dataclass
class OutputContract:
"""Explicit constraints that every agent must respect."""
max_tokens: int
verbosity: OutputVerbosity
required_sections: list[str] = field(default_factory=list)
forbidden_content: list[str] = field(default_factory=list)
target_audience: str = "downstream AI agent"
def to_prompt_fragment(self) -> str:
"""Convert contract into a prompt injection for the agent."""
fragment = f"""
OUTPUT CONSTRAINTS (YOU MUST FOLLOW THESE):
- Maximum output length: {self.max_tokens} tokens
- Verbosity level: {self.verbosity.value}
- Target audience: {self.target_audience}
- Required sections to include: {', '.join(self.required_sections) if self.required_sections else 'None specified'}
- Do NOT include: {', '.join(self.forbidden_content) if self.forbidden_content else 'Nothing forbidden'}
- If you cannot fulfill the task within these constraints, explicitly state 'CONSTRAINT_VIOLATION:' and explain why.
"""
return fragment
@dataclass
class GlobalTaskContext:
"""Shared context injected into all agents to align their goals."""
task_description: str
final_output_format: str
total_token_budget: int
agent_roles: dict[str, str] # Maps agent name to their specific role
def to_prompt_fragment(self) -> str:
fragment = f"""
GLOBAL TASK CONTEXT:
- Overall task: {self.task_description}
- Final expected output format: {self.final_output_format}
- Total token budget for ALL agents combined: {self.total_token_budget}
- Other agents in the system and their roles:
{chr(10).join(f' - {name}: {role}' for name, role in self.agent_roles.items())}
IMPORTANT: Your output is one part of a multi-agent pipeline. Optimize for the global task,
not just your local objective. Be concise if downstream agents only need summaries.
"""
return fragment
class AlignedResearchAgent:
def __init__(self, contract: OutputContract, global_context: GlobalTaskContext):
self.contract = contract
self.global_context = global_context
async def research(self, topic: str) -> str:
prompt = f"""You are a Research Agent.
{self.global_context.to_prompt_fragment()}
{self.contract.to_prompt_fragment()}
Research topic: {topic}
Provide your findings respecting ALL constraints above."""
response = await llm_call(prompt)
# Post-hoc validation of contract compliance
if "CONSTRAINT_VIOLATION:" in response:
print(f"[WARNING] Agent self-reported constraint violation: {response[:200]}")
# Trigger fallback or re-prompt with relaxed constraints
# Check token count (approximate)
estimated_tokens = len(response.split())
if estimated_tokens > self.contract.max_tokens:
print(f"[WARNING] Agent output {estimated_tokens} tokens exceeds contract limit {self.contract.max_tokens}")
# Truncate intelligently or request regeneration
return response
async def aligned_multi_agent_example():
# Define global context shared across all agents
global_ctx = GlobalTaskContext(
task_description="Produce a 1-page executive summary of Q4 sales data",
final_output_format="Markdown document with 3 sections: Highlights, Risks, Recommendations",
total_token_budget=2000,
agent_roles={
"Research Agent": "Gathers raw data and statistics",
"Analysis Agent": "Identifies trends and patterns",
"Writer Agent": "Composes the final executive summary"
}
)
# Research agent gets a tight contract because downstream only needs key facts
research_contract = OutputContract(
max_tokens=300,
verbosity=OutputVerbosity.MINIMAL,
required_sections=["key_metrics", "top_performers", "bottom_performers"],
forbidden_content=["raw data tables", "methodology details", "appendix"]
)
research_agent = AlignedResearchAgent(research_contract, global_ctx)
result = await research_agent.research("Q4 sales performance by region")
print(result)
# Output will be concise and aligned with downstream needs
Failure Mode 4: Tool Calling Cascade Failures
What It Looks Like
Agent A calls a tool and passes the result to Agent B, which calls another tool based on that result. If Agent A's tool call returns an error or unexpected format, Agent B's tool call fails with a cryptic error. The system retries both agents, doubling the cost with no progress.
The Fix: Tool Result Normalization and Error Wrapping
# FIXED: Standardized tool result wrapper prevents error propagation
from enum import Enum
from typing import Any, Optional, Union
from dataclasses import dataclass
import traceback
class ToolStatus(Enum):
SUCCESS = "success"
PARTIAL = "partial" # Completed but with caveats
FAILED_RETRYABLE = "failed_retryable"
FAILED_PERMANENT = "failed_permanent"
TIMEOUT = "timeout"
@dataclass
class ToolResult:
"""Universal wrapper for all tool call results in the agent system."""
status: ToolStatus
data: Optional[Any] = None
error_message: Optional[str] = None
retry_count: int = 0
latency_ms: float = 0.0
raw_response: Optional[str] = None
def is_usable(self) -> bool:
return self.status in (ToolStatus.SUCCESS, ToolStatus.PARTIAL)
def to_agent_context(self) -> str:
"""Convert tool result to a standardized message for downstream agents."""
if self.status == ToolStatus.SUCCESS:
return f"TOOL RESULT (SUCCESS):\n{self.data}\n[Latency: {self.latency_ms:.0f}ms]"
elif self.status == ToolStatus.PARTIAL:
return f"TOOL RESULT (PARTIAL - use with caution):\n{self.data}\n[Warning: {self.error_message}]"
elif self.status == ToolStatus.FAILED_RETRYABLE:
return f"TOOL RESULT (FAILED - retryable):\nError: {self.error_message}\nSuggest retrying with modified parameters."
else:
return f"TOOL RESULT (FAILED - permanent):\nError: {self.error_message}\nDo NOT retry. Proceed with fallback or request human intervention."
class RobustToolExecutor:
def __init__(self, max_retries: int = 3, timeout_seconds: float = 30.0):
self.max_retries = max_retries
self.timeout = timeout_seconds
async def execute(self, tool_name: str, params: dict) -> ToolResult:
"""Execute a tool with retries, timeout, and standardized error wrapping."""
import time
for attempt in range(self.max_retries + 1):
start_time = time.time()
try:
# Simulate tool execution with timeout
result_data = await asyncio.wait_for(
self._call_tool(tool_name, params),
timeout=self.timeout
)
latency = (time.time() - start_time) * 1000
# Validate result structure
if result_data is None:
return ToolResult(
status=ToolStatus.FAILED_RETRYABLE,
error_message=f"Tool '{tool_name}' returned None",
retry_count=attempt,
latency_ms=latency
)
return ToolResult(
status=ToolStatus.SUCCESS,
data=result_data,
retry_count=attempt,
latency_ms=latency
)
except asyncio.TimeoutError:
latency = (time.time() - start_time) * 1000
if attempt < self.max_retries:
print(f"[RETRY] Tool '{tool_name}' timed out after {self.timeout}s, retrying...")
continue
return ToolResult(
status=ToolStatus.TIMEOUT,
error_message=f"Tool '{tool_name}' timed out after {self.timeout}s over {self.max_retries + 1} attempts",
retry_count=attempt,
latency_ms=latency
)
except Exception as e:
latency = (time.time() - start_time) * 1000
error_str = f"{type(e).__name__}: {str(e)}"
# Classify error: is it retryable?
is_retryable = self._is_retryable_error(e)
if is_retryable and attempt < self.max_retries:
print(f"[RETRY] Tool '{tool_name}' failed with retryable error: {error_str}")
await asyncio.sleep(min(2 ** attempt, 10)) # Exponential backoff
continue
return ToolResult(
status=ToolStatus.FAILED_RETRYABLE if is_retryable else ToolStatus.FAILED_PERMANENT,
error_message=error_str,
retry_count=attempt,
latency_ms=latency,
raw_response=str(e)
)
# Should not reach here, but defensive coding
return ToolResult(
status=ToolStatus.FAILED_PERMANENT,
error_message=f"Exhausted all {self.max_retries} retries for tool '{tool_name}'",
retry_count=self.max_retries
)
def _is_retryable_error(self, error: Exception) -> bool:
"""Classify whether an error might succeed on retry."""
retryable_types = (
ConnectionError, TimeoutError, asyncio.TimeoutError,
# HTTP 429, 503 equivalents
)
error_str = str(error).lower()
retryable_keywords = ('timeout', 'connection', 'rate limit', '503', '429', 'temporary')
return isinstance(error, retryable_types) or any(kw in error_str for kw in retryable_keywords)
async def _call_tool(self, tool_name: str, params: dict):
"""Actual tool invocation — replace with your tool registry."""
# This would be your actual tool calling mechanism
raise NotImplementedError("Replace with actual tool call")
# Usage in a multi-agent pipeline
async def tool_chain_example():
executor = RobustToolExecutor(max_retries=3, timeout_seconds=10.0)
# Agent A calls a search tool
search_result = await executor.execute("web_search", {"query": "Q4 sales data"})
if not search_result.is_usable():
# Downstream agents receive a clear, structured error they can reason about
context_for_agent_b = search_result.to_agent_context()
print(f"Passing to Agent B: {context_for_agent_b}")
# Agent B can now make an informed decision: retry, fallback, or escalate
return
# Agent B receives clean, validated data
context_for_agent_b = search_result.to_agent_context()
print(f"Passing to Agent B: {context_for_agent_b}")
# Continue pipeline with confidence
Failure Mode 5: Memory Contamination Between Agents
What It Looks Like
Agents share a global conversation history or vector store. Agent A's outdated or incorrect reasoning persists in the shared memory and influences Agent C's decisions three turns later. This is especially dangerous when Agent A makes a confident but wrong assertion that later agents treat as established fact.
The Fix: Namespaced Memory with Confidence Tagging
# FIXED: Memory entries are namespaced, versioned, and confidence-tagged
from datetime import datetime
from typing import Optional, List
from dataclasses import dataclass, field
from enum import Enum
class ConfidenceLevel(Enum):
CERTAIN = 1.0 # Verified fact from trusted source
HIGH = 0.8 # Strong evidence, likely correct
MEDIUM = 0.5 # Reasonable inference
LOW = 0.3 # Speculative or unverified
RETRACTED = 0.0 # Previously stated but now known to be wrong
@dataclass
class MemoryEntry:
"""A single memory entry with provenance and confidence metadata."""
entry_id: str
agent_name: str
content: str
confidence: ConfidenceLevel
timestamp: datetime = field(default_factory=datetime.now)
based_on: List[str] = field(default_factory=list) # Entry IDs this derives from
retracted_by: Optional[str] = None # Entry ID that retracts this, if any
ttl_seconds: Optional[int] = None # Time-to-live for speculative entries
def is_active(self) -> bool:
"""Check if entry is still valid (not retracted, not expired)."""
if self.retracted_by is not None:
return False
if self.confidence == ConfidenceLevel.RETRACTED:
return False
if self.ttl_seconds is not None:
age = (datetime.now() - self.timestamp).total_seconds()
if age > self.ttl_seconds:
return False
return True
def to_context_string(self) -> str:
"""Format memory entry for inclusion in agent prompt."""
confidence_label = f"[{self.confidence.name}]"
source_label = f"(source: {self.agent_name})"
return f"{confidence_label} {source_label} {self.content}"
class NamespacedMemory:
"""Memory store with namespacing, confidence, and retraction support."""
def __init__(self):
self.entries: dict[str, MemoryEntry] = {}
self.agent_namespaces: dict[str, List[str]] = {} # agent -> list of entry_ids
def add(self, agent_name: str, content: str, confidence: ConfidenceLevel,
based_on: List[str] = None, ttl_seconds: Optional[int] = None) -> str:
"""Add a memory entry and return its ID."""
entry_id = f"{agent_name}:{datetime.now().timestamp()}:{hash(content) & 0xFFFF}"
entry = MemoryEntry(
entry_id=entry_id,
agent_name=agent_name,
content=content,
confidence=confidence,
based_on=based_on or [],
ttl_seconds=ttl_seconds
)
self.entries[entry_id] = entry
if agent_name not in self.agent_namespaces:
self.agent_namespaces[agent_name] = []
self.agent_namespaces[agent_name].append(entry_id)
return entry_id
def retract(self, entry_id: str, retracting_agent: str, reason: str) -> Optional[str]:
"""Retract a previous statement and record the correction."""
if entry_id not in self.entries:
return None
original = self.entries[entry_id]
original.retracted_by = f"{retracting_agent}:{datetime.now().timestamp()}"
original.confidence = ConfidenceLevel.RETRACTED
# Add a correction entry
correction_id = self.add(
agent_name=retracting_agent,
content=f"CORRECTION: Previous statement by {original.agent_name} retracted. Reason: {reason}. Original: {original.content[:100]}...",
confidence=ConfidenceLevel.CERTAIN,
based_on=[entry_id]
)
return correction_id
def get_context_for_agent(self, agent_name: str, max_entries: int = 20,
min_confidence: ConfidenceLevel = ConfidenceLevel.LOW) -> str:
"""Get filtered, active memory relevant to a specific agent."""
# Get agent's own entries (most relevant)
own_entries = self.agent_namespaces.get(agent_name, [])
# Get entries from other agents that are still active
all_active = [
entry for entry in self.entries.values()
if entry.is_active() and entry.confidence.value >= min_confidence.value
]
# Sort by recency and confidence
all_active.sort(key=lambda e: (e.timestamp, e.confidence.value), reverse=True)
# Build context string with confidence indicators
context_lines = []
for entry in